[spark] Add spark.paimon.search.residual-filter to select search TVF residual behavior

docs/generated/spark_connector_configuration.html

@@ -92,6 +92,12 @@
             <td>Boolean</td>
             <td>Whether to verify SparkSession is initialized with required configurations.</td>
         </tr>
+        <tr>
+            <td><h5>search.residual-filter</h5></td>
+            <td style="word-wrap: break-word;">post-filter</td>
+            <td><p>Enum</p></td>
+            <td>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.<br /><br />Possible values:<ul><li>"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.</li><li>"fail": Reject the query with a clear error, matching the Flink vector_search procedure, so a silently short result never ships.</li></ul></td>
+        </tr>
         <tr>
             <td><h5>source.split.target-size-with-column-pruning</h5></td>
             <td style="word-wrap: break-word;">false</td>

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<SearchResidualFilterMode> 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);
+        }
+    }
 }

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
+  }
+}

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 {

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

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))

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
   }

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<FLOAT>")
+      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<FLOAT>")
+      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<FLOAT>")
+      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()