diff --git a/docs/generated/spark_connector_configuration.html b/docs/generated/spark_connector_configuration.html index d80d14f258ec..fa8a86cd68f2 100644 --- a/docs/generated/spark_connector_configuration.html +++ b/docs/generated/spark_connector_configuration.html @@ -92,6 +92,12 @@ Boolean Whether to verify SparkSession is initialized with required configurations. + +
search.residual-filter
+ post-filter +

Enum

+ How a vector / hybrid / full-text search handles a WHERE conjunct on searched-table columns that cannot be pushed into Paimon. Such a residual is applied by Spark above the already-truncated top-K, so it can only drop rows and may return fewer than K.

Possible values: +
source.split.target-size-with-column-pruning
false diff --git a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java index 4dd9329d1c4c..461bfba820fd 100644 --- a/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java +++ b/paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkConnectorOptions.java @@ -19,8 +19,11 @@ package org.apache.paimon.spark; import org.apache.paimon.options.ConfigOption; +import org.apache.paimon.options.description.DescribedEnum; +import org.apache.paimon.options.description.InlineElement; import static org.apache.paimon.options.ConfigOptions.key; +import static org.apache.paimon.options.description.TextElement.text; /** Options for spark connector. */ public class SparkConnectorOptions { @@ -180,4 +183,45 @@ public class SparkConnectorOptions { .withDescription( "Whether to adjust the target split size based on pruned (projected) columns. " + "If enabled, split size estimation uses only the columns actually being read."); + + public static final ConfigOption SEARCH_RESIDUAL_FILTER = + key("search.residual-filter") + .enumType(SearchResidualFilterMode.class) + .defaultValue(SearchResidualFilterMode.POST_FILTER) + .withDescription( + "How a vector / hybrid / full-text search handles a WHERE conjunct on " + + "searched-table columns that cannot be pushed into Paimon. Such a " + + "residual is applied by Spark above the already-truncated top-K, so it " + + "can only drop rows and may return fewer than K."); + + /** How a search TVF handles a residual filter it cannot push into Paimon. */ + public enum SearchResidualFilterMode implements DescribedEnum { + POST_FILTER( + "post-filter", + "Apply the residual above the top-K result. The result is a subset of the top-K " + + "and may be shorter than K."), + + FAIL( + "fail", + "Reject the query with a clear error, matching the Flink vector_search procedure, " + + "so a silently short result never ships."); + + private final String value; + private final String description; + + SearchResidualFilterMode(String value, String description) { + this.value = value; + this.description = description; + } + + @Override + public String toString() { + return value; + } + + @Override + public InlineElement getDescription() { + return text(description); + } + } } diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/CheckUnpushableSearchFilter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/CheckUnpushableSearchFilter.scala new file mode 100644 index 000000000000..4855b5924439 --- /dev/null +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/CheckUnpushableSearchFilter.scala @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.paimon.spark.catalyst.optimizer + +import org.apache.paimon.spark.{SparkTable, SparkV2FilterConverter} +import org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions +import org.apache.paimon.spark.schema.PaimonMetadataColumn +import org.apache.paimon.spark.util.OptionUtils +import org.apache.paimon.table.{FullTextSearchTable, HybridSearchTable, InnerTable, VectorSearchTable} + +import org.apache.spark.sql.PaimonUtils.translateFilterV2 +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeSet, PredicateHelper} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan} +import org.apache.spark.sql.catalyst.rules.Rule +import org.apache.spark.sql.connector.catalog.Table +import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation} + +/** + * Fails a static vector / hybrid / full-text search whose WHERE carries a residual on + * searched-table columns that cannot be pushed into Paimon. Such a residual stays a Spark filter + * above the search, whose result is already truncated to the top-K, so post-filtering it can only + * drop rows and never refill the ones displaced out of the top-K. Only active when + * `spark.paimon.search.residual-filter` is `fail`; the default `post-filter` keeps the residual and + * accepts a possibly short result. The lateral form is handled by + * [[PushDownLateralVectorSearchFilter]]. + */ +object CheckUnpushableSearchFilter extends Rule[LogicalPlan] with PredicateHelper { + + override def apply(plan: LogicalPlan): LogicalPlan = plan.transformDown { + case filter @ Filter(condition, child) if condition.resolved => + relationTableAndOutput(child).foreach { + case (table, output) => + searchInnerTable(table).foreach { + innerTable => + val converter = SparkV2FilterConverter(innerTable.rowType()) + val dataColumns = AttributeSet( + output.filterNot( + attr => PaimonMetadataColumn.VECTOR_SEARCH_META_COLUMN_NAMES.contains(attr.name))) + val residuals = splitConjunctivePredicates(condition).filter { + predicate => + predicate.references.nonEmpty && + predicate.references.intersect(dataColumns).nonEmpty && + translateFilterV2(predicate).flatMap(converter.convert(_)).isEmpty + } + // Read the option only once a search TVF actually carries a non-pushable residual, so + // a misconfigured value fails just that query rather than every query in the session. + if (residuals.nonEmpty && OptionUtils.searchResidualFilterFailEnabled()) { + PaimonTableValuedFunctions.failUnpushableSearchFilter(residuals.map(_.sql)) + } + } + } + filter + } + + private def relationTableAndOutput(plan: LogicalPlan): Option[(Table, Seq[Attribute])] = + plan match { + case relation: DataSourceV2Relation => Some((relation.table, relation.output)) + case scan: DataSourceV2ScanRelation => Some((scan.relation.table, scan.output)) + case _ => None + } + + private def searchInnerTable(table: Table): Option[InnerTable] = table match { + case st: SparkTable => + st.table match { + case vst: VectorSearchTable => Some(vst.origin()) + case hst: HybridSearchTable => Some(hst.origin()) + case ftst: FullTextSearchTable => Some(ftst.origin()) + case _ => None + } + case _ => None + } +} diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownLateralVectorSearchFilter.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownLateralVectorSearchFilter.scala index de5baf0b3564..593368d4ecc1 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownLateralVectorSearchFilter.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/PushDownLateralVectorSearchFilter.scala @@ -19,6 +19,7 @@ package org.apache.paimon.spark.catalyst.optimizer import org.apache.paimon.spark.catalyst.plans.logical.{LateralVectorSearch, PaimonTableValuedFunctions} +import org.apache.paimon.spark.util.OptionUtils import org.apache.spark.sql.catalyst.expressions.{And, PredicateHelper} import org.apache.spark.sql.catalyst.plans.logical.{Filter, LogicalPlan} @@ -48,6 +49,16 @@ object PushDownLateralVectorSearchFilter extends Rule[LogicalPlan] with Predicat .isDefined } + // A residual on searched-table columns that cannot be pushed into the search is applied above + // its top-K result and can silently drop matching rows. Under `fail` reject it; under the + // default `post-filter` keep it as a filter above the search, which may return fewer than K. + if (OptionUtils.searchResidualFilterFailEnabled()) { + val unpushable = stayUp.filter(_.references.intersect(lvs.searchFilterOutputSet).nonEmpty) + if (unpushable.nonEmpty) { + PaimonTableValuedFunctions.failUnpushableSearchFilter(unpushable.map(_.sql)) + } + } + if (pushDownToLeft.isEmpty && pushDownToSearch.isEmpty) { filter } else { diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala index 4c0123824f95..14ff6a26ad51 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala @@ -69,6 +69,24 @@ object PaimonTableValuedFunctions { HYBRID_SEARCH, FULL_TEXT_SEARCH) + /** + * Rejects a vector / hybrid / full-text search that carries a residual predicate Spark cannot + * push into Paimon, when `spark.paimon.search.residual-filter` is `fail`. Such a residual is + * applied by Spark above the search, whose result is already truncated to the top-K, so a row + * that satisfies the residual but ranks just outside the returned K is lost and the result comes + * back short. Failing is consistent with the Flink `vector_search` procedure, which rejects an + * inexpressible predicate rather than silently dropping rows. + */ + def failUnpushableSearchFilter(residuals: Seq[String]): Nothing = { + throw new UnsupportedOperationException( + "Vector, hybrid or full-text search does not support a filter that cannot be pushed down " + + "to Paimon, because it would be applied above the already-truncated top-K result and can " + + "silently drop matching rows: " + residuals.mkString("[", ", ", "]") + + ". Rewrite it into a pushable predicate on table columns (avoid UDFs, column-to-column " + + "comparisons and unresolvable casts), or set spark.paimon.search.residual-filter to " + + "post-filter to apply it above the search and accept a possibly short result.") + } + def parsePositiveLimit(value: Any): Int = { val limit = value match { case i: Int => i diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala index b216454ba4f8..0bcd1b4ce59f 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/extensions/PaimonSparkSessionExtensions.scala @@ -19,7 +19,7 @@ package org.apache.paimon.spark.extensions import org.apache.paimon.spark.catalyst.analysis.{PaimonAnalysis, PaimonDeleteTable, PaimonFunctionResolver, PaimonIncompatibleResolutionRules, PaimonMergeInto, PaimonPostHocResolutionRules, PaimonProcedureResolver, PaimonUpdateTable, PaimonViewResolver, ReplacePaimonFunctions} -import org.apache.paimon.spark.catalyst.optimizer.{MergePaimonScalarSubqueries, OptimizeMetadataOnlyDeleteFromPaimonTable, PushDownArrayPredicates, PushDownLateralVectorSearchFilter, RepartitionLateralVectorSearchInput} +import org.apache.paimon.spark.catalyst.optimizer.{CheckUnpushableSearchFilter, MergePaimonScalarSubqueries, OptimizeMetadataOnlyDeleteFromPaimonTable, PushDownArrayPredicates, PushDownLateralVectorSearchFilter, RepartitionLateralVectorSearchInput} import org.apache.paimon.spark.catalyst.plans.logical.PaimonTableValuedFunctions import org.apache.paimon.spark.commands.BucketExpression import org.apache.paimon.spark.execution.{OldCompatibleStrategy, PaimonStrategy} @@ -107,6 +107,7 @@ class PaimonSparkSessionExtensions extends (SparkSessionExtensions => Unit) { } extensions.injectOptimizerRule(_ => RepartitionLateralVectorSearchInput) extensions.injectOptimizerRule(_ => PushDownLateralVectorSearchFilter) + extensions.injectOptimizerRule(_ => CheckUnpushableSearchFilter) // planner extensions extensions.injectPlannerStrategy(spark => PaimonStrategy(spark)) diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala index c357256b7cc0..ff3e7943e560 100644 --- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala +++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/util/OptionUtils.scala @@ -146,6 +146,19 @@ object OptionUtils extends SQLConfHelper with Logging { getOptionString(SparkConnectorOptions.SOURCE_SPLIT_TARGET_SIZE_WITH_COLUMN_PRUNING).toBoolean } + def searchResidualFilterFailEnabled(): Boolean = { + val option = SparkConnectorOptions.SEARCH_RESIDUAL_FILTER + val value = getOptionString(option) + val mode = SparkConnectorOptions.SearchResidualFilterMode + .values() + .find(_.toString.equalsIgnoreCase(value)) + .getOrElse( + throw new IllegalArgumentException( + s"Invalid value '$value' for spark.paimon.${option.key()}. Valid values: " + + SparkConnectorOptions.SearchResidualFilterMode.values().mkString(", ") + ".")) + mode == SparkConnectorOptions.SearchResidualFilterMode.FAIL + } + def formatTableRepairCollectStatistics(): Boolean = { getOptionString(SparkConnectorOptions.FORMAT_TABLE_REPAIR_COLLECT_STATISTICS).toBoolean } diff --git a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala index 80806858289c..1c1e9b9fa605 100644 --- a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala +++ b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/PrimaryKeyVectorSearchTest.scala @@ -352,6 +352,106 @@ class PrimaryKeyVectorSearchTest extends PaimonSparkTestBase { } } + test("vector search fail mode rejects a non-convertible residual filter") { + withTable("T") { + createVectorTable(columns = "id INT, threshold INT, embedding ARRAY") + spark.sql(""" + |INSERT INTO T VALUES + | (1, 100, array(1.0f, 0.0f)), + | (2, 100, array(2.0f, 0.0f)), + | (3, 100, array(3.0f, 0.0f)), + | (4, 100, array(4.0f, 0.0f)), + | (5, 1, array(5.0f, 0.0f)), + | (6, 1, array(6.0f, 0.0f)) + |""".stripMargin) + + // `id > threshold` is a column-to-column comparison, which SparkV2FilterConverter cannot + // convert to a Paimon predicate, so it stays a Spark residual applied above the vector + // search. The two nearest rows (1, 2) fail it; the two nearest rows that satisfy it are + // (5, 6), but they rank outside the returned top-2, so post-filtering the top-2 returns + // nothing. Fail mode rejects the query rather than returning a silently short result. + withSparkSQLConf("spark.paimon.search.residual-filter" -> "fail") { + val error = intercept[Exception] { + spark + .sql(""" + |SELECT id + |FROM vector_search('T', 'embedding', array(0.0f, 0.0f), 2) + |WHERE id > threshold + |""".stripMargin) + .collect() + } + assert(error.getMessage.contains("cannot be pushed down"), error.getMessage) + } + } + } + + test("vector search post-filters a non-convertible residual filter by default") { + withTable("T") { + createVectorTable(columns = "id INT, threshold INT, embedding ARRAY") + spark.sql(""" + |INSERT INTO T VALUES + | (1, 100, array(1.0f, 0.0f)), + | (2, 100, array(2.0f, 0.0f)), + | (3, 100, array(3.0f, 0.0f)), + | (4, 100, array(4.0f, 0.0f)), + | (5, 1, array(5.0f, 0.0f)), + | (6, 1, array(6.0f, 0.0f)) + |""".stripMargin) + + // The default post-filter mode applies the same `id > threshold` residual above the top-2 + // (1, 2), which both fail it, so the result is a subset of the top-2 and comes back empty + // rather than the query being rejected. + val unfiltered = spark + .sql("SELECT id FROM vector_search('T', 'embedding', array(0.0f, 0.0f), 2)") + .collect() + .map(_.getInt(0)) + .toSet + assert(unfiltered == Set(1, 2)) + val filtered = spark + .sql(""" + |SELECT id + |FROM vector_search('T', 'embedding', array(0.0f, 0.0f), 2) + |WHERE id > threshold + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSet + assert(filtered.subsetOf(unfiltered)) + assert(filtered.isEmpty) + } + } + + test("vector search fail mode keeps a pushable filter") { + withTable("T") { + createVectorTable(columns = "id INT, threshold INT, embedding ARRAY") + spark.sql(""" + |INSERT INTO T VALUES + | (1, 100, array(1.0f, 0.0f)), + | (2, 100, array(2.0f, 0.0f)), + | (3, 100, array(3.0f, 0.0f)), + | (4, 100, array(4.0f, 0.0f)), + | (5, 1, array(5.0f, 0.0f)), + | (6, 1, array(6.0f, 0.0f)) + |""".stripMargin) + + // `threshold = 100` is convertible, so it is pushed into Paimon and never a genuine residual. + // Even a Spark-side recheck copy of it converts, so fail mode must not reject it. The two + // nearest rows (1, 2) both satisfy it, so the result is (1, 2) either way. + withSparkSQLConf("spark.paimon.search.residual-filter" -> "fail") { + val result = spark + .sql(""" + |SELECT id + |FROM vector_search('T', 'embedding', array(0.0f, 0.0f), 2) + |WHERE threshold = 100 + |""".stripMargin) + .collect() + .map(_.getInt(0)) + .toSet + assert(result == Set(1, 2)) + } + } + } + test("deduplicate updates and deletes primary-key vector results") { withTable("T") { createVectorTable()