You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/09/17 10:44:27 UTC
(spark) branch branch-4.2 updated: [SPARK-59572][SQL] PartitionPredicate evaluation failure must not be reported as a partition match
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 5ef96075f26f [SPARK-59572][SQL] PartitionPredicate evaluation failure must not be reported as a partition match
5ef96075f26f is described below
commit 5ef96075f26f55da613ece03e1a53ba81c92328e
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Thu Sep 17 18:29:34 2026 +0800
[SPARK-59572][SQL] PartitionPredicate evaluation failure must not be reported as a partition match
### What changes were proposed in this pull request?
`PartitionPredicateImpl.eval` reported a partition as matching whenever it could not evaluate
the predicate, either because evaluating the expression threw or because the partition key width
did not match the schema. It now propagates that failure.
The old behavior is kept for runtime filters only, which merely prune: their rows are filtered
anyway, by the post-scan `FilterExec` for a scalar subquery filter, and by the join it was
derived from for a dynamic partition pruning filter, which has no post-scan `FilterExec`.
`createRuntimePartitionPredicates` is the only caller that asks for it.
This also documents the `eval` contract for callers, and corrects a comment in the in-memory
test table that stated the old justification as a general rule.
### Why are the changes needed?
Failing open is sound only while something else still evaluates the filter. Two paths have not
met that condition since `PartitionPredicate` was introduced in 4.2.0:
- the static second pass, where a `PartitionPredicate` the connector accepts is removed from the
post-scan filters;
- the metadata-only DELETE rewrite, which has no post-scan filter at all.
On both, the predicate is Spark's only evaluator, so reporting a match on failure is a wrong
answer. On a table partitioned by `dep STRING` holding `'hr'` and `'1'`, with a filter on `dep`
whose evaluation fails for `'hr'`, an ANSI cast error for example:
- `SELECT ... WHERE <filter>` returns the `'hr'` row, for which the filter is never true, and
reports no error;
- `DELETE FROM ... WHERE <filter>` deletes the `'hr'` partition, for which the condition is never
true.
The second is silent data loss.
The invariant was identified during the SPARK-58523 review, where `fullyPushedFilterAttributes`
was described as the first thing in Spark to remove the post-scan `FilterExec`. The two paths
above already removed it.
Keeping the pushed filter instead of propagating is not available here: by the time `eval` fails,
the connector has accepted the predicate and Spark has already dropped the filter, and Spark
never learns that an evaluation failed.
### Does this PR introduce _any_ user-facing change?
Yes, recorded in `docs/sql-migration-guide.md` for 4.3.0 and 4.2.1.
A query whose partition predicate the source cannot evaluate now fails with the error it would
raise without pushdown, instead of silently returning rows the filter rejects or deleting
partitions its condition never matched. Which queries fail depends on the order the source
evaluates the predicates it accepted; the error can surface while the scan is built, so `EXPLAIN`
may report it in place of a plan, and a partition with no rows can raise it.
A partition key whose width does not match `Table.partitioning()` now raises an internal error.
Runtime filtering is unchanged.
### How was this patch tested?
New tests, all of which fail without the fix:
- a scan whose partition predicate fails to evaluate must surface the error rather than return
the partition it could not evaluate;
- a metadata-only DELETE in the same situation must surface the error and leave both partitions
in place;
- a unit test over `PartitionPredicateImpl` pinning that the same predicate propagates by default
and keeps the partition when constructed for a runtime filter;
- a `PushDownUtilsSuite` test pinning the wiring, so that only the runtime factory produces a
predicate that keeps a key it cannot evaluate.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #58851 from pan3793/partition-predicate-fail-open.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 8974c2ce653c79e828ed827e5667904b745254d6)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
docs/sql-migration-guide.md | 4 ++
.../expressions/filter/PartitionPredicate.java | 5 ++
.../connector/PartitionPredicateImpl.scala | 39 ++++++++++++----
.../connector/PartitionPredicateImplSuite.scala | 54 ++++++++++++++++++++--
.../execution/datasources/v2/PushDownUtils.scala | 14 ++++--
.../DataSourceV2EnhancedDeleteFilterSuite.scala | 50 +++++++++++++++++---
.../DataSourceV2EnhancedPartitionFilterSuite.scala | 23 +++++++++
.../datasources/v2/PushDownUtilsSuite.scala | 47 +++++++++++++++++++
8 files changed, 215 insertions(+), 21 deletions(-)
diff --git a/docs/sql-migration-guide.md b/docs/sql-migration-guide.md
index bfa3189e07c1..949c5a0c1c80 100644
--- a/docs/sql-migration-guide.md
+++ b/docs/sql-migration-guide.md
@@ -22,6 +22,10 @@ license: |
* Table of contents
{:toc}
+## Upgrading from Spark SQL 4.2.0 to 4.2.1
+
+- Since Spark 4.2.1, when a data source cannot evaluate a partition predicate that Spark pushed down, for example because an ANSI cast of a partition value fails, the error is raised instead of the partition being reported as a match. Previously the failure was ignored, so a scan could return rows the filter does not accept and a metadata-only `DELETE` could remove a partition its condition never matched. Queries that appeared to succeed may now fail with the same error they raise withou [...]
+
## Upgrading from Spark SQL 4.1 to 4.2
- Since Spark 4.2, Spark enables order-independent checksums for shuffle outputs by default to detect data inconsistencies during indeterminate shuffle stage retries. If a checksum mismatch is detected, Spark rolls back and re-executes all succeeding stages that depend on the shuffle output. If rolling back is not possible for some succeeding stages, the job will fail. To restore the previous behavior, set `spark.sql.shuffle.orderIndependentChecksum.enabled` and `spark.sql.shuffle.orderI [...]
diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/PartitionPredicate.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/PartitionPredicate.java
index 211a90505abf..91cfa1f9fd86 100644
--- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/PartitionPredicate.java
+++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/expressions/filter/PartitionPredicate.java
@@ -81,6 +81,11 @@ public abstract class PartitionPredicate extends Predicate {
* The caller must pass the <b>full</b> partition key: one value per partition transform in
* {@link Table#partitioning()}, in order. A key for only a subset of referenced fields is not
* supported.
+ * <p>
+ * This method may throw when it cannot evaluate the predicate for the given key, for example on
+ * an ANSI cast failure. Accepting a predicate lets Spark drop the filter it pushed, which leaves
+ * this predicate as the only evaluator, so a caller must let such a failure propagate rather
+ * than treat the partition as matching, and must finish evaluating before it mutates any data.
*
* @param partitionKey the full partition key for one partition, ordered according to
* {@link Table#partitioning()}.
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImpl.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImpl.scala
index 0550344fe4e8..187e86dbb7d4 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImpl.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImpl.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.internal.connector
+import org.apache.spark.SparkException
import org.apache.spark.internal.{Logging, LogKeys}
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{BindReferences, Expression => CatalystExpression, ExprId, Predicate => CatalystPredicate}
@@ -26,10 +27,24 @@ import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
/**
* An implementation for [[PartitionPredicate]] that wraps a Catalyst Expression representing a
* partition filter.
+ *
+ * Reporting a partition [[eval]] cannot evaluate as matching only prunes less, which is safe for a
+ * runtime filter, whose rows are filtered anyway: by the post-scan `FilterExec` for a scalar
+ * subquery filter, by the join it was derived from for a dynamic partition pruning filter, and by
+ * the rewrite re-applying its own condition for a row-level operation's group filter. Everywhere
+ * else Spark drops a filter the connector accepts, leaving this predicate as the only evaluator,
+ * so the failure is propagated instead: reporting a match would return or write rows the filter
+ * does not accept.
+ *
+ * @param catalystExpr the partition filter this predicate evaluates.
+ * @param partitionFields one entry per transform of `Table.partitioning()`, in that order, so a
+ * bound ordinal matches the partition key a connector passes to [[eval]].
+ * @param keepOnEvalFailure whether [[eval]] reports a partition it cannot evaluate as matching.
*/
class PartitionPredicateImpl private (
private val catalystExpr: CatalystExpression,
- private val partitionFields: Seq[PartitionPredicateField])
+ private val partitionFields: Seq[PartitionPredicateField],
+ private val keepOnEvalFailure: Boolean)
extends PartitionPredicate with Logging {
@transient private lazy val exprIdToIndex: Map[ExprId, Int] =
@@ -47,18 +62,21 @@ class PartitionPredicateImpl private (
override def eval(partitionValues: InternalRow): Boolean = {
if (partitionValues.numFields != partitionFields.length) {
- logWarning(
+ val message =
log"Cannot evaluate partition predicate ${MDC(LogKeys.EXPR, catalystExpr.sql)}: " +
log"partition value field count (${MDC(LogKeys.COUNT, partitionValues.numFields)}) " +
- log"does not match schema (${MDC(LogKeys.NUM_PARTITIONS, partitionFields.length)}). " +
- log"Including partition in scan result to avoid incorrect filtering.")
+ log"does not match schema (${MDC(LogKeys.NUM_PARTITIONS, partitionFields.length)})."
+ if (!keepOnEvalFailure) {
+ throw SparkException.internalError(message.message)
+ }
+ logWarning(message + log" Including partition in scan result.")
return true
}
try {
boundPredicate(partitionValues)
} catch {
- case e: Exception =>
+ case e: Exception if keepOnEvalFailure =>
logWarning(
log"Failed to evaluate partition predicate ${MDC(LogKeys.EXPR, catalystExpr.sql)}. " +
log"Including partition in scan result to avoid incorrect filtering.",
@@ -79,12 +97,14 @@ class PartitionPredicateImpl private (
override def equals(obj: Any): Boolean = obj match {
case other: PartitionPredicateImpl =>
catalystExpr.semanticEquals(other.catalystExpr) &&
- partitionFields == other.partitionFields
+ partitionFields == other.partitionFields &&
+ keepOnEvalFailure == other.keepOnEvalFailure
case _ => false
}
override def hashCode(): Int = {
- 31 * catalystExpr.semanticHash() + partitionFields.hashCode()
+ 31 * (31 * catalystExpr.semanticHash() + partitionFields.hashCode()) +
+ keepOnEvalFailure.hashCode()
}
override def toString(): String = s"PartitionPredicate(${catalystExpr.sql})"
@@ -93,7 +113,8 @@ class PartitionPredicateImpl private (
object PartitionPredicateImpl extends Logging {
def apply(catalystExpr: CatalystExpression,
- partitionFields: Seq[PartitionPredicateField])
+ partitionFields: Seq[PartitionPredicateField],
+ keepOnEvalFailure: Boolean = false)
: Option[PartitionPredicateImpl] = {
if (partitionFields.isEmpty) {
logWarning(
@@ -116,6 +137,6 @@ object PartitionPredicateImpl extends Logging {
return None
}
- Some(new PartitionPredicateImpl(catalystExpr, partitionFields))
+ Some(new PartitionPredicateImpl(catalystExpr, partitionFields, keepOnEvalFailure))
}
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImplSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImplSuite.scala
index 7dd6ae8717f0..197ee710a212 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImplSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/internal/connector/PartitionPredicateImplSuite.scala
@@ -17,16 +17,18 @@
package org.apache.spark.sql.internal.connector
-import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.{SparkConf, SparkException, SparkFunSuite, SparkNumberFormatException}
import org.apache.spark.serializer.{JavaSerializer, KryoSerializer, SerializerInstance}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{GreaterThan, Literal}
+import org.apache.spark.sql.catalyst.expressions.{Cast, EqualTo, EvalMode, GreaterThan, Literal}
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.connector.expressions.PartitionFieldReference
+import org.apache.spark.sql.errors.QueryErrorsBase
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{IntegerType, StringType, StructField}
import org.apache.spark.unsafe.types.UTF8String
-class PartitionPredicateImplSuite extends SparkFunSuite {
+class PartitionPredicateImplSuite extends SparkFunSuite with QueryErrorsBase {
test("Kryo serialization: PartitionPredicateImpl works after round-trip") {
val conf = new SparkConf()
@@ -52,6 +54,52 @@ class PartitionPredicateImplSuite extends SparkFunSuite {
checkNestedPartitionPathReferencesAfterSerialization(serializer)
}
+ test("SPARK-59572: eval propagates a failure unless the predicate may keep the partition") {
+ val ref = DataTypeUtils.toAttribute(StructField("p", StringType, nullable = true))
+ val fields = Seq(PartitionPredicateField(Seq("p"), ref))
+ // An ANSI cast of a non-numeric string throws when evaluated.
+ val expr = EqualTo(Cast(ref, IntegerType, None, EvalMode.ANSI), Literal(1))
+ val failing = InternalRow(UTF8String.fromString("hr"))
+ val matching = InternalRow(UTF8String.fromString("1"))
+ val notMatching = InternalRow(UTF8String.fromString("2"))
+
+ // Default: the predicate is Spark's only evaluator, so the failure must surface.
+ val strict = PartitionPredicateImpl(expr, fields).get
+ checkError(
+ exception = intercept[SparkNumberFormatException](strict.eval(failing)),
+ condition = "CAST_INVALID_INPUT",
+ sqlState = Some("22018"),
+ parameters = Map(
+ "expression" -> toSQLValue("hr"),
+ "sourceType" -> toSQLType(StringType),
+ "targetType" -> toSQLType(IntegerType),
+ "ansiConfig" -> toSQLConf(SQLConf.ANSI_ENABLED.key)),
+ queryContext = Array(ExpectedContext("", -1, -1)))
+ assert(strict.eval(matching) === true)
+ assert(strict.eval(notMatching) === false)
+
+ // Runtime filters are re-evaluated after the scan, so failing open only prunes less.
+ val lenient = PartitionPredicateImpl(expr, fields, keepOnEvalFailure = true).get
+ assert(lenient.eval(failing) === true)
+ assert(lenient.eval(matching) === true)
+ // Keeping on failure must not degrade into always matching: an evaluable key still decides.
+ assert(lenient.eval(notMatching) === false)
+
+ // The flag is part of the predicate's identity.
+ assert(strict !== lenient)
+ assert(strict.hashCode() !== lenient.hashCode())
+
+ // The same split applies to a partition key that does not match the schema.
+ val wrongWidth = InternalRow(UTF8String.fromString("1"), UTF8String.fromString("extra"))
+ checkError(
+ exception = intercept[SparkException](strict.eval(wrongWidth)),
+ condition = "INTERNAL_ERROR",
+ sqlState = Some("XX000"),
+ parameters = Map("message" -> ("Cannot evaluate partition predicate " +
+ "(CAST(p AS INT) = 1): partition value field count (2) does not match schema (1).")))
+ assert(lenient.eval(wrongWidth) === true)
+ }
+
private def checkPartitionPredicateImplAfterSerialization(
serializer: SerializerInstance): Unit = {
val ref = DataTypeUtils.toAttribute(StructField("p", IntegerType, nullable = true))
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
index 8c5396d7eb12..deb683cd4b61 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala
@@ -272,18 +272,22 @@ object PushDownUtils extends Logging {
* @param flattenedFilters Catalyst filter expressions with partition field references
* already flattened.
* @param partitionFields Partition field metadata.
+ * @param keepOnEvalFailure whether a created predicate reports a partition as matching when it
+ * cannot be evaluated, instead of propagating the failure.
+ * See [[PartitionPredicateImpl]].
* @return a pair of (created partition predicates, remaining filters not converted).
*/
private[v2] def createPartitionPredicates(
flattenedFilters: Seq[Expression],
- partitionFields: Seq[PartitionPredicateField])
+ partitionFields: Seq[PartitionPredicateField],
+ keepOnEvalFailure: Boolean = false)
: (Seq[PartitionPredicateImpl], Seq[Expression]) = {
val partitionAttributes = partitionFields.map(_.attrRef)
val (partFilters, nonPartitionFilters) =
DataSourceUtils.getPartitionFiltersAndDataFilters(partitionAttributes, flattenedFilters)
val (pushable, nonPushable) = partFilters.partition(isPushablePartitionFilter)
val (partitionPredicates, errorPartitionPredicates) = pushable.partitionMap { e =>
- PartitionPredicateImpl(e, partitionFields).toLeft(e)
+ PartitionPredicateImpl(e, partitionFields, keepOnEvalFailure).toLeft(e)
}
(partitionPredicates, nonPartitionFilters ++ nonPushable ++ errorPartitionPredicates)
}
@@ -322,7 +326,11 @@ object PushDownUtils extends Logging {
case f => Some(f.transform { case s: ExecScalarSubquery => s.toLiteral })
}
val flattened = flattenNestedPartitionFilters(catalystExprs, partitionFields).keys
- createPartitionPredicates(flattened.toSeq, partitionFields)._1
+ // A runtime filter only prunes: its rows are filtered anyway, by the post-scan `FilterExec`
+ // for a scalar subquery filter, by the join it was derived from for a DPP filter, and by the
+ // rewrite re-applying its own condition for a row-level operation's group filter. So a
+ // partition the source cannot evaluate can be kept rather than failing the query.
+ createPartitionPredicates(flattened.toSeq, partitionFields, keepOnEvalFailure = true)._1
}
private def isPushablePartitionFilter(f: Expression) =
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala
index ee2a95febeb8..d055871e9a3e 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.connector
-import org.apache.spark.SparkConf
+import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.sql.Row
import org.apache.spark.sql.connector.catalog.InMemoryPartitionPredicateDeleteCatalog
import org.apache.spark.sql.connector.expressions.PartitionFieldReference
@@ -241,7 +241,42 @@ class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession {
}
}
- private def executeAndKeepPlan(func: => Unit): SparkPlan = {
+ // A metadata-only DELETE has no post-scan filter, so the PartitionPredicate is the only
+ // evaluator. `to_int('hr')` throws, so the condition is never true for that partition, and
+ // reporting the failed evaluation as a match would delete it.
+ test("SPARK-59572: metadata-only DELETE must not drop a partition whose predicate failed") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep)")
+ sql(s"INSERT INTO $deleteTableName VALUES (1, 'hr', 100), (2, '1', 200)")
+
+ spark.udf.register("to_int", (s: String) => s.toInt)
+
+ val (e, plan) = keepingPlan {
+ intercept[SparkException] {
+ sql(s"DELETE FROM $deleteTableName WHERE to_int(dep) = 1").collect()
+ }
+ }
+ assert(e.getCondition == "FAILED_EXECUTE_UDF")
+ assert(e.getCause.isInstanceOf[NumberFormatException])
+ // The fallback row-level DELETE raises the same error, so pin the metadata-only path.
+ assert(plan.isInstanceOf[DeleteFromTableExec],
+ s"Expected a metadata-only DELETE but got ${plan.getClass.getSimpleName}")
+
+ // The partition whose predicate could not be evaluated must still be there.
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "hr", 100), Row(2, "1", 200)))
+ }
+ }
+
+ private def executeAndKeepPlan(func: => Unit): SparkPlan = keepingPlan(func)._2
+
+ /**
+ * Runs `func` and returns its result with the executed plan. The plan is captured for a failed
+ * query too, so a test that intercepts the failure can still assert which plan produced it.
+ */
+ private def keepingPlan[T](func: => T): (T, SparkPlan) = {
var executedPlan: SparkPlan = null
val listener = new QueryExecutionListener {
@@ -250,20 +285,23 @@ class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession {
executedPlan = qe.executedPlan
}
override def onFailure(
- funcName: String, qe: QueryExecution, exception: Exception): Unit = {}
+ funcName: String, qe: QueryExecution, exception: Exception): Unit = {
+ executedPlan = qe.executedPlan
+ }
}
spark.listenerManager.register(listener)
- try {
- func
+ val result = try {
+ val r = func
sparkContext.listenerBus.waitUntilEmpty()
+ r
} finally {
spark.listenerManager.unregister(listener)
}
assert(executedPlan != null,
"QueryExecutionListener did not capture the executed plan")
- executedPlan
+ (result, executedPlan)
}
/**
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedPartitionFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedPartitionFilterSuite.scala
index 3fe928daf9e0..8fc01533b252 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedPartitionFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedPartitionFilterSuite.scala
@@ -21,6 +21,7 @@ import java.util.Locale
import org.scalatest.BeforeAndAfter
+import org.apache.spark.SparkException
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.catalyst.expressions.{Expression, In, PredicateHelper, ScalaUDF}
import org.apache.spark.sql.connector.catalog.BufferedRows
@@ -403,6 +404,28 @@ class DataSourceV2EnhancedPartitionFilterSuite
}
}
+ test("SPARK-59572: a failed second-pass PartitionPredicate must not keep the partition") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (part_col string, data string) USING $v2Source " +
+ "PARTITIONED BY (part_col)")
+ sql(s"INSERT INTO $partFilterTableName VALUES ('hr', 'x'), ('1', 'y')")
+
+ spark.udf.register("to_int", (s: String) => s.toInt)
+
+ // `to_int('hr')` throws, so the filter is never true for that partition. The filter is
+ // untranslatable, so it is pushed as a PartitionPredicate and accepted, which removes it
+ // from the post-scan filters. The source is then its only evaluator, so reporting a failed
+ // evaluation as a match returns the 'hr' row. Optimizing is enough to raise it, which the
+ // fallback path cannot do: there the filter stays post-scan and only execution fails.
+ val e = intercept[SparkException] {
+ sql(s"SELECT * FROM $partFilterTableName WHERE to_int(part_col) = 1")
+ .queryExecution.optimizedPlan
+ }
+ assert(e.getCondition == "FAILED_EXECUTE_UDF")
+ assert(e.getCause.isInstanceOf[NumberFormatException])
+ }
+ }
+
test("non-deterministic partition filter not pushed as PartitionPredicate") {
// Same checks as FileSourceStrategy/PruneFileSourcePartitions: non-deterministic
// partition filters must not be pushed as PartitionPredicate; they are applied after scan.
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtilsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtilsSuite.scala
new file mode 100644
index 000000000000..1165c7c35f31
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtilsSuite.scala
@@ -0,0 +1,47 @@
+/*
+ * 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.spark.sql.execution.datasources.v2
+
+import org.apache.spark.{SparkFunSuite, SparkNumberFormatException}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Cast, EqualTo, EvalMode, Literal}
+import org.apache.spark.sql.catalyst.types.DataTypeUtils
+import org.apache.spark.sql.internal.connector.PartitionPredicateField
+import org.apache.spark.sql.types.{IntegerType, StringType, StructField}
+import org.apache.spark.unsafe.types.UTF8String
+
+class PushDownUtilsSuite extends SparkFunSuite {
+
+ test("SPARK-59572: only a runtime partition predicate keeps a key it cannot evaluate") {
+ val ref = DataTypeUtils.toAttribute(StructField("p", StringType, nullable = true))
+ val fields = Seq(PartitionPredicateField(Seq("p"), ref))
+ // An ANSI cast of a non-numeric string throws when evaluated.
+ val filter = EqualTo(Cast(ref, IntegerType, None, EvalMode.ANSI), Literal(1))
+ val failing = InternalRow(UTF8String.fromString("hr"))
+
+ // A runtime filter only prunes, so a partition it cannot evaluate is kept.
+ val runtime = PushDownUtils.createRuntimePartitionPredicates(Seq(filter), fields)
+ assert(runtime.size === 1)
+ assert(runtime.head.eval(failing) === true)
+
+ // Everywhere else Spark drops the filter the connector accepts, so the failure must surface.
+ val (pushed, _) = PushDownUtils.createPartitionPredicates(Seq(filter), fields)
+ assert(pushed.size === 1)
+ intercept[SparkNumberFormatException](pushed.head.eval(failing))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org