You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/09/18 17:48:47 UTC
(spark) branch master updated: [SPARK-59410][SQL] Derive PartitionPredicate from identity fields of a mixed partitioning
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ca0b651258a0 [SPARK-59410][SQL] Derive PartitionPredicate from identity fields of a mixed partitioning
ca0b651258a0 is described below
commit ca0b651258a047548cafb8d7866a866fde0304c6
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Fri Sep 18 10:48:36 2026 -0700
[SPARK-59410][SQL] Derive PartitionPredicate from identity fields of a mixed partitioning
### What changes were proposed in this pull request?
`PushDownUtils.getPartitionPredicateSchema` returned a schema only when every transform in
`Table.partitioning()` is an identity transform. It now accepts any partitioning with at least
one identity transform: identity fields can be referenced by a `PartitionPredicate`, other
fields keep their ordinal but are never referenced, so a filter on the source column of a
non-identity transform stays a data filter.
The connector contract is unchanged: `PartitionPredicate.eval` still receives the full
partition key, and `references()` still reports ordinals into `Table.partitioning()`.
This applies to all users of the schema: the static second pass, runtime filter pushdown
(DPP and scalar subqueries), the metadata-only DELETE rewrite, and the scan of a row-level
UPDATE, MERGE or DELETE.
Two in-memory test tables are corrected so they model a connector faithfully on a mixed
partitioning. The V2 filter table accepts only predicates of the shape it can evaluate, a
column against a literal, returning anything else such as a predicate over a cast. The
partition predicate delete table matches a filter against the partition key only for identity
transforms, since the slot of any other transform holds that transform's value rather than its
source column.
### Why are the changes needed?
A table partitioned by, for example, `dt STRING` (identity) and `bucket(16, user_id)` received
no `PartitionPredicate` at all. A filter such as `dt = DATE'2026-09-01'` is analyzed as
`cast(dt AS DATE) = DATE'2026-09-01'`; a connector that does not evaluate casts returns it
from the first pass (and without ANSI mode it is not translatable at all), so only a
`PartitionPredicate` can prune with it. On such a table it could not prune partitions in
either the static or the runtime path, and could not drive a metadata-only DELETE, while the
same filter on an identity-only table can. Mixed partitionings are the common case for
connectors that support partition transforms.
The all-identity check was raised in the SPARK-55596 review (#54459) and kept only because
the partition-key contract was still open then. The contract that shipped (full key,
partitioning order) is what makes lifting the check safe.
### Does this PR introduce _any_ user-facing change?
No. A connector that opts into iterative pushdown and has a mixed partitioning now receives
`PartitionPredicate`s over its identity fields; the `eval` contract is unchanged.
### How was this patch tested?
New tests on a partitioning of one identity column plus a bucket transform. A predicate on the
identity field is pushed and prunes in the static pass, via DPP, in a metadata-only DELETE, and
in the scan of a group-based UPDATE and MERGE, including the `dt STRING` against `DATE` literal
case whose type-coercion cast the source cannot evaluate. An identity field placed after the
bucket transform binds to its own ordinal. A filter on the bucket source column is not turned
into a predicate, stays post-scan on a read, and is declined by the metadata-only DELETE path.
A bucket-only partitioning yields no predicate. The predicate survives Java and Kryo
round-trips.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
Closes #58702 from pan3793/partition-predicate-mixed-spec.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../connector/PartitionPredicateField.scala | 17 +--
.../connector/PartitionPredicateImpl.scala | 24 +++--
.../InMemoryPartitionPredicateDeleteTable.scala | 6 +-
.../catalog/InMemoryTableWithV2Filter.scala | 19 ++--
.../connector/PartitionPredicateImplSuite.scala | 33 +++++-
.../execution/datasources/v2/PushDownUtils.scala | 31 ++++--
.../DataSourceV2EnhancedDeleteFilterSuite.scala | 119 +++++++++++++++++++++
.../DataSourceV2EnhancedPartitionFilterSuite.scala | 100 +++++++++++++++++
...urceV2EnhancedRuntimePartitionFilterSuite.scala | 55 ++++++++++
.../datasources/v2/PushDownUtilsSuite.scala | 2 +-
10 files changed, 371 insertions(+), 35 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateField.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateField.scala
index 993c8706feec..e1c76183f997 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateField.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/connector/PartitionPredicateField.scala
@@ -20,14 +20,19 @@ package org.apache.spark.sql.internal.connector
import org.apache.spark.sql.catalyst.expressions.AttributeReference
/**
- * Metadata for one partition field.
+ * Metadata for one field of `Table.partitioning()`. A partition predicate is built over the
+ * fields in partitioning order, so their ordinals match the partition key a connector passes to
+ * `PartitionPredicate.eval`.
*
* @param fieldNames the multi-part field name from the table's partitioning
- * (e.g. `Seq("s", "tz")`).
- * @param attrRef the [[AttributeReference]] for the partition field.
- * Created from the resolved partition field so it carries the
- * flattened dotted name (e.g. `"s.tz"`) for nested fields.
+ * (e.g. `Seq("s", "tz")`) for an identity transform, or the transform's
+ * description (e.g. `Seq("bucket(4, id)")`) otherwise.
+ * @param attrRef the [[AttributeReference]] a filter can reference, for an identity transform.
+ * Created from the resolved partition field so it carries the flattened dotted
+ * name (e.g. `"s.tz"`) for nested fields. None for any other transform: for now
+ * Spark does not evaluate a filter against its partition value, so no filter
+ * references it, but the field keeps its ordinal.
*/
case class PartitionPredicateField(
fieldNames: Seq[String],
- attrRef: AttributeReference)
+ attrRef: Option[AttributeReference])
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 187e86dbb7d4..0121f1af027f 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
@@ -20,9 +20,10 @@ 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}
+import org.apache.spark.sql.catalyst.expressions.{AttributeReference, BindReferences, Expression => CatalystExpression, ExprId, Predicate => CatalystPredicate}
import org.apache.spark.sql.connector.expressions.NamedReference
import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.types.NullType
/**
* An implementation for [[PartitionPredicate]] that wraps a Catalyst Expression representing a
@@ -47,15 +48,24 @@ class PartitionPredicateImpl private (
private val keepOnEvalFailure: Boolean)
extends PartitionPredicate with Logging {
+ /** Ordinal of each identity partition field, keyed by the attribute a filter references. */
@transient private lazy val exprIdToIndex: Map[ExprId, Int] =
- partitionFields.zipWithIndex.map { case (f, i) => f.attrRef.exprId -> i }.toMap
+ partitionFields.zipWithIndex.collect {
+ case (PartitionPredicateField(_, Some(attr)), i) => attr.exprId -> i
+ }.toMap
/** The wrapped partition filter Catalyst Expression. */
def expression: CatalystExpression = catalystExpr
/** Bound predicate, computed once and reused for all partition rows. */
@transient private lazy val boundPredicate: InternalRow => Boolean = {
- val boundExpr = BindReferences.bindReference(catalystExpr, partitionFields.map(_.attrRef))
+ // One attribute per partition field, so that ordinals match the full partition key. A field
+ // of a non-identity transform has no attribute a filter can reference; a placeholder keeps
+ // its slot.
+ val input = partitionFields.map { f =>
+ f.attrRef.getOrElse(AttributeReference(f.fieldNames.mkString("."), NullType)())
+ }
+ val boundExpr = BindReferences.bindReference(catalystExpr, input)
val predicate = CatalystPredicate.createInterpreted(boundExpr)
predicate.eval
}
@@ -123,16 +133,18 @@ object PartitionPredicateImpl extends Logging {
return None
}
- val partitionExprIds = partitionFields.map(_.attrRef.exprId).toSet
+ val partitionExprIds = partitionFields.flatMap(_.attrRef).map(_.exprId).toSet
val unmatchedRefs = catalystExpr.references.filterNot(r => partitionExprIds.contains(r.exprId))
if (unmatchedRefs.nonEmpty) {
logWarning(
log"Cannot create partition predicate ${MDC(LogKeys.EXPR, catalystExpr.sql)}: " +
log"expression references " +
log"${MDC(LogKeys.FIELD_NAME, unmatchedRefs.map(_.name).mkString(", "))} " +
- log"not found in partition fields " +
+ log"not found in identity partition fields " +
log"${MDC(LogKeys.PARTITION_SPECIFICATION,
- partitionFields.map(_.fieldNames.mkString(".")).mkString(", "))}. " +
+ partitionFields.collect {
+ case PartitionPredicateField(names, Some(_)) => names.mkString(".")
+ }.mkString(", "))}. " +
log"Skipping pushdown for this predicate.")
return None
}
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala
index 91f26741a08a..12b5dc59318e 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala
@@ -58,7 +58,11 @@ class InMemoryPartitionPredicateDeleteTable(
properties.getOrDefault(
InMemoryPartitionPredicateDeleteTable.AcceptDataPredicatesKey, "false").toBoolean
- private val partPaths = partCols.map(_.mkString(".")).toSet
+ // Only an identity transform keeps its source value in the partition key, so only its column
+ // can be matched against that key. `partCols` also holds the source columns of the other
+ // transforms, e.g. `pk` of `bucket(4, pk)`, whose slot holds the bucket value instead.
+ private val partPaths =
+ identityPartitionReferences.map(_.fieldNames().mkString(".")).toSet
private def refsOnlyPartCols(p: Predicate): Boolean =
p.references().forall(ref => partPaths.contains(ref.fieldNames().mkString(".")))
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
index 6922c874cf73..49346769c031 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2Filter.scala
@@ -203,14 +203,19 @@ object InMemoryTableWithV2Filter {
}
}
+ /**
+ * Whether every predicate has a shape [[evalPredicate]] can evaluate: a plain column, or a
+ * column and a literal. A predicate over an expression, e.g. a cast, is not supported and
+ * returned to Spark, as a real connector without expression support would do.
+ */
def supportsPredicates(predicates: Array[Predicate]): Boolean = {
- predicates.flatMap(splitAnd).forall {
- case p: Predicate if p.name().equals("=") => true
- case p: Predicate if p.name().equals("<=>") => true
- case p: Predicate if p.name().equals("IS_NULL") => true
- case p: Predicate if p.name().equals("IS_NOT_NULL") => true
- case p: Predicate if p.name().equals("ALWAYS_TRUE") => true
- case _ => false
+ predicates.flatMap(splitAnd).forall { p =>
+ (p.name(), p.children().toSeq) match {
+ case ("=" | "<=>", Seq(_: NamedReference, _: LiteralValue[_])) => true
+ case ("IS_NULL" | "IS_NOT_NULL", Seq(_: NamedReference)) => true
+ case ("ALWAYS_TRUE", _) => true
+ case _ => false
+ }
}
}
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 197ee710a212..e2fda0ff096f 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
@@ -56,7 +56,7 @@ class PartitionPredicateImplSuite extends SparkFunSuite with QueryErrorsBase {
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))
+ val fields = Seq(PartitionPredicateField(Seq("p"), Some(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"))
@@ -100,11 +100,38 @@ class PartitionPredicateImplSuite extends SparkFunSuite with QueryErrorsBase {
assert(lenient.eval(wrongWidth) === true)
}
+ test("non-identity partition field: predicate binds by ordinal and never references it") {
+ val ref = DataTypeUtils.toAttribute(StructField("p", StringType, nullable = true))
+ val fields = Seq(
+ PartitionPredicateField(Seq("bucket(4, id)"), None),
+ PartitionPredicateField(Seq("p"), Some(ref)))
+ val predicate = PartitionPredicateImpl(GreaterThan(ref, Literal("m")), fields).get
+
+ // The partition key carries one value per field; the bucket value at ordinal 0 is skipped.
+ assert(predicate.eval(InternalRow(3, UTF8String.fromString("z"))) === true)
+ assert(predicate.eval(InternalRow(3, UTF8String.fromString("a"))) === false)
+ assert(refsWithOrdinals(predicate.references.toSeq) === Seq(("p", 1)))
+
+ // A filter on the source column of the bucket transform has no field to bind to.
+ val id = DataTypeUtils.toAttribute(StructField("id", IntegerType, nullable = true))
+ assert(PartitionPredicateImpl(GreaterThan(id, Literal(1)), fields).isEmpty)
+
+ Seq(new JavaSerializer(new SparkConf()), new KryoSerializer(new SparkConf())).foreach { s =>
+ val serializer = s.newInstance()
+ val deserialized = serializer.deserialize[PartitionPredicateImpl](
+ serializer.serialize(predicate))
+ assert(deserialized.eval(InternalRow(3, UTF8String.fromString("z"))) === true)
+ assert(deserialized.eval(InternalRow(3, UTF8String.fromString("a"))) === false)
+ assert(refsWithOrdinals(deserialized.references.toSeq) === Seq(("p", 1)))
+ assert(deserialized.equals(predicate))
+ }
+ }
+
private def checkPartitionPredicateImplAfterSerialization(
serializer: SerializerInstance): Unit = {
val ref = DataTypeUtils.toAttribute(StructField("p", IntegerType, nullable = true))
val expr = GreaterThan(ref, Literal(5))
- val fields = Seq(PartitionPredicateField(Seq("p"), ref))
+ val fields = Seq(PartitionPredicateField(Seq("p"), Some(ref)))
val predicate = PartitionPredicateImpl(expr, fields).get
val deserialized = serializer.deserialize[PartitionPredicateImpl](
@@ -125,7 +152,7 @@ class PartitionPredicateImplSuite extends SparkFunSuite with QueryErrorsBase {
serializer: SerializerInstance): Unit = {
val ref = DataTypeUtils.toAttribute(StructField("ts.timezone", StringType, nullable = false))
val expr = GreaterThan(ref, Literal("x"))
- val fields = Seq(PartitionPredicateField(Seq("ts", "timezone"), ref))
+ val fields = Seq(PartitionPredicateField(Seq("ts", "timezone"), Some(ref)))
val predicate = PartitionPredicateImpl(expr, fields).get
val deserialized = serializer.deserialize[PartitionPredicateImpl](
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 949c80c7a114..ff3acdfca7e2 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
@@ -357,8 +357,8 @@ object PushDownUtils extends Logging {
}
/**
- * Returns a Seq of [[PartitionPredicateField]] representing partition transform expression types,
- * if schema is supported for [[PartitionPredicate]] push down. None if not supported.
+ * Returns one [[PartitionPredicateField]] per transform of `relation.table.partitioning`, if
+ * the partitioning supports [[PartitionPredicate]] push down. None if not supported.
*/
def getPartitionPredicateSchema(relation: DataSourceV2Relation)
: Option[Seq[PartitionPredicateField]] = {
@@ -366,8 +366,8 @@ object PushDownUtils extends Logging {
}
/**
- * Returns a Seq of [[PartitionPredicateField]] representing partition transform expression types,
- * if schema is supported for [[PartitionPredicate]] push down. None if not supported.
+ * Returns one [[PartitionPredicateField]] per transform of `table.partitioning`, if the
+ * partitioning supports [[PartitionPredicate]] push down. None if not supported.
*/
def getPartitionPredicateSchema(table: Table, output: Seq[AttributeReference])
: Option[Seq[PartitionPredicateField]] = {
@@ -375,8 +375,15 @@ object PushDownUtils extends Logging {
}
/**
- * Returns a Seq of [[PartitionPredicateField]] representing partition transform expression types,
- * if schema is supported for [[PartitionPredicate]] push down. None if not supported.
+ * Returns one [[PartitionPredicateField]] per transform, in partitioning order, if the
+ * partitioning supports [[PartitionPredicate]] push down. None if not supported.
+ *
+ * Only an identity transform yields a field with an attribute, so only filters over identity
+ * partition columns become partition predicates. Any other transform is kept as a field without
+ * an attribute: Spark cannot evaluate a filter against its partition value, but the field must
+ * keep its ordinal since a predicate is evaluated against the full partition key. The
+ * partitioning is not supported when it is empty, has no identity transform, or has an identity
+ * transform that does not resolve against `output`.
*
* Use this overload when the caller has access to the partition transforms but not the
* full [[Table]].
@@ -392,11 +399,11 @@ object PushDownUtils extends Logging {
val fields = transforms.flatMap {
case t: IdentityTransform =>
resolveIdentityPartitionField(t, rootStruct).map { sf =>
- PartitionPredicateField(t.ref.fieldNames().toSeq, DataTypeUtils.toAttribute(sf))
+ PartitionPredicateField(t.ref.fieldNames().toSeq, Some(DataTypeUtils.toAttribute(sf)))
}
- case _ => None
+ case t => Some(PartitionPredicateField(Seq(t.describe()), None))
}
- if (fields.length == transforms.length) {
+ if (fields.length == transforms.length && fields.exists(_.attrRef.isDefined)) {
Some(fields.toSeq)
} else {
None
@@ -445,7 +452,7 @@ object PushDownUtils extends Logging {
partitionFields: Seq[PartitionPredicateField],
keepOnEvalFailure: Boolean = false)
: (Seq[PartitionPredicateImpl], Seq[Expression]) = {
- val partitionAttributes = partitionFields.map(_.attrRef)
+ val partitionAttributes = partitionFields.flatMap(_.attrRef)
val (partFilters, nonPartitionFilters) =
DataSourceUtils.getPartitionFiltersAndDataFilters(partitionAttributes, flattenedFilters)
val (pushable, nonPushable) = partFilters.partition(isPushablePartitionFilter(_))
@@ -537,7 +544,9 @@ object PushDownUtils extends Logging {
filters: Seq[Expression],
partitionFields: Seq[PartitionPredicateField])
: Map[Expression, Expression] = {
- val pathToAttr = partitionFields.map(f => f.fieldNames -> f.attrRef).toMap
+ val pathToAttr = partitionFields.collect {
+ case PartitionPredicateField(names, Some(attr)) => names -> attr
+ }.toMap
filters.map(f => doNormalizePartitionFilters(f, pathToAttr) -> f).toMap
}
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 953eb9e38121..b2c753e0ab61 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
@@ -191,6 +191,69 @@ class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession
}
}
+ // Mixed partitioning: the bucket field keeps its ordinal but is never referenced, so the
+ // IN on the identity column still becomes a PartitionPredicate over the full partition key.
+ test("second pass accepted: identity column next to a bucket transform") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep, bucket(4, pk))")
+ sql(s"INSERT INTO $deleteTableName VALUES " +
+ "(1, 'hr', 100), (2, 'software', 200), (3, 'marketing', 300)")
+
+ assertDeleteWithFilters(
+ s"DELETE FROM $deleteTableName WHERE dep IN ('hr', 'software')",
+ expectedNumConditions = 1,
+ expectedNumPartitionPredicates = 1,
+ expectedOrdinalsPerPredicate = Seq(Array(0)),
+ expectedPartitionFieldNames = Array("dep", "bucket(4, pk)"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Row(3, "marketing", 300) :: Nil)
+ }
+ }
+
+ // `dt = DATE'...'` is analyzed as `cast(dt AS DATE) = DATE'...'`, which the table cannot
+ // evaluate in the first pass; the second pass turns it into a PartitionPredicate.
+ test("second pass accepted: cast on a string identity column next to a bucket transform") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dt STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dt, bucket(4, pk))")
+ sql(s"INSERT INTO $deleteTableName VALUES " +
+ "(1, '2026-09-01', 100), (2, '2026-09-02', 200), (3, '2026-09-03', 300)")
+
+ assertDeleteWithFilters(
+ s"DELETE FROM $deleteTableName WHERE dt = DATE'2026-09-02'",
+ expectedNumConditions = 1,
+ expectedNumPartitionPredicates = 1,
+ expectedOrdinalsPerPredicate = Seq(Array(0)),
+ expectedPartitionFieldNames = Array("dt", "bucket(4, pk)"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "2026-09-01", 100), Row(3, "2026-09-03", 300)))
+ }
+ }
+
+ // A filter on the source column of a bucket transform is a data filter: its value is not in
+ // the partition key, whose slot holds the bucket instead. The metadata-only path must decline
+ // it and fall back, rather than compare `pk` with a bucket value.
+ test("second pass skipped: filter on the source column of a bucket transform") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep, bucket(4, pk))")
+ sql(s"INSERT INTO $deleteTableName VALUES " +
+ "(1, 'hr', 100), (5, 'hr', 500), (2, 'software', 200)")
+
+ // `bucket(4, 5)` is 1, so comparing the bucket slot with 5 would match nothing.
+ assertDeleteWithRowLevel(s"DELETE FROM $deleteTableName WHERE pk = 5")
+
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "hr", 100), Row(2, "software", 200)))
+ }
+ }
+
// Table property disables PartitionPredicate acceptance;
// both passes rejected, falls back to row-level operation.
test("first and second pass rejected: table rejects all") {
@@ -368,6 +431,62 @@ class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession
}
}
+ // A mixed partitioning: the bucket field keeps its ordinal but is never referenced, so the
+ // IN on the identity column still becomes a PartitionPredicate over the full partition key.
+ test("group-based UPDATE prunes by the identity field of a mixed partitioning") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep, bucket(4, pk)) " +
+ "TBLPROPERTIES('iterative-row-level-pushdown' = 'true')")
+ sql(s"INSERT INTO $deleteTableName VALUES " +
+ "(1, 'hr', 100), (2, 'software', 200), (3, 'marketing', 300)")
+
+ val plan = executeAndKeepPlan {
+ sql(s"UPDATE $deleteTableName SET salary = salary + 1 WHERE dep IN ('hr', 'software')")
+ }
+ assertRowLevelScanPrunedByPartitionPredicate(plan,
+ expectedOrdinals = Array(0),
+ expectedPartitionFieldNames = Array("dep", "bucket(4, pk)"),
+ expectedReplacedDeps = Set("hr", "software"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "hr", 101), Row(2, "software", 201), Row(3, "marketing", 300)))
+ }
+ }
+
+ test("group-based MERGE prunes by the identity field of a mixed partitioning") {
+ withTable(deleteTableName) {
+ withTempView("source") {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep, bucket(4, pk)) " +
+ "TBLPROPERTIES('iterative-row-level-pushdown' = 'true')")
+ sql(s"INSERT INTO $deleteTableName VALUES " +
+ "(1, 'hr', 100), (2, 'software', 200), (3, 'marketing', 300)")
+ Seq((1, 1000), (3, 3000)).toDF("pk", "salary").createOrReplaceTempView("source")
+
+ val plan = executeAndKeepPlan {
+ sql(
+ s"""MERGE INTO $deleteTableName t
+ |USING source s
+ |ON t.pk = s.pk AND t.dep IN ('hr', 'software')
+ |WHEN MATCHED THEN
+ | UPDATE SET salary = s.salary
+ |""".stripMargin)
+ }
+ assertRowLevelScanPrunedByPartitionPredicate(plan,
+ expectedOrdinals = Array(0),
+ expectedPartitionFieldNames = Array("dep", "bucket(4, pk)"),
+ expectedReplacedDeps = Set("hr", "software"))
+
+ // Row 3 is in a pruned partition, so it is never read and stays unchanged.
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "hr", 1000), Row(2, "software", 200), Row(3, "marketing", 300)))
+ }
+ }
+ }
+
private def executeAndKeepPlan(func: => Unit): SparkPlan = keepingPlan(func)._2
/**
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 8fc01533b252..f0fcdd31f3f2 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
@@ -426,6 +426,106 @@ class DataSourceV2EnhancedPartitionFilterSuite
}
}
+ test("mixed partitioning: second-pass PartitionPredicate on the identity field") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (part_col string, id int, data string) " +
+ s"USING $v2Source PARTITIONED BY (part_col, bucket(4, id))")
+ sql(s"INSERT INTO $partFilterTableName VALUES ('a', 1, 'x'), ('A', 2, 'y'), ('b', 3, 'z')")
+
+ spark.udf.register("my_upper", (s: String) =>
+ if (s == null) null else s.toUpperCase(Locale.ROOT))
+
+ // Untranslatable, Partition Filter on the identity field; 2nd Pass Accepted.
+ // The bucket field keeps ordinal 1 but is never referenced.
+ val df = sql(s"SELECT * FROM $partFilterTableName WHERE my_upper(part_col) = 'A'")
+ checkAnswer(df, Seq(Row("a", 1, "x"), Row("A", 2, "y")))
+ assertPushedPartitionPredicates(df, 1)
+ assertScanReturnsPartitionKeys(df, Set("a/1", "A/2"))
+ assertReferencedPartitionFieldOrdinals(df, Array(0), Array("part_col", "bucket(4, id)"))
+ }
+ }
+
+ test("mixed partitioning: filter on the source column of a bucket transform stays post-scan") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (part_col string, id int, data string) " +
+ s"USING $v2Source PARTITIONED BY (part_col, bucket(4, id))")
+ sql(s"INSERT INTO $partFilterTableName VALUES ('a', 1, 'x'), ('A', 2, 'y'), ('b', 3, 'z')")
+
+ spark.udf.register("my_upper", (s: String) =>
+ if (s == null) null else s.toUpperCase(Locale.ROOT))
+ spark.udf.register("my_plus1", (i: Int) => i + 1)
+
+ // Only the identity conjunct becomes a PartitionPredicate. Spark cannot evaluate the
+ // bucket conjunct against the partition key, so it is a data filter applied after the scan.
+ val df = sql(s"SELECT * FROM $partFilterTableName " +
+ "WHERE my_upper(part_col) = 'A' AND my_plus1(id) = 3")
+ checkAnswer(df, Seq(Row("A", 2, "y")))
+ assertPushedPartitionPredicates(df, 1)
+ assertScanReturnsPartitionKeys(df, Set("a/1", "A/2"))
+ assertReferencedPartitionFieldOrdinals(df, Array(0), Array("part_col", "bucket(4, id)"))
+ val postScanFilter = df.queryExecution.executedPlan.collectFirst {
+ case f: FilterExec => f
+ }.getOrElse(fail("Filter on the bucket source column should remain as a post-scan Filter"))
+ val referenced = postScanFilter.condition.references.map(_.name).toSet
+ assert(referenced.contains("id") && !referenced.contains("part_col"),
+ s"Post-scan Filter should reference id but not part_col, got $referenced")
+ }
+ }
+
+ test("bucket-only partitioning: no identity transform -> no PartitionPredicate") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (id int, data string) " +
+ s"USING $v2Source PARTITIONED BY (bucket(4, id))")
+ sql(s"INSERT INTO $partFilterTableName VALUES (1, 'x'), (2, 'y'), (3, 'z')")
+
+ spark.udf.register("my_plus1", (i: Int) => i + 1)
+
+ val df = sql(s"SELECT * FROM $partFilterTableName WHERE my_plus1(id) = 3")
+ checkAnswer(df, Seq(Row(2, "y")))
+ assertPushedPartitionPredicates(df, 0)
+ assertScanReturnsPartitionKeys(df, Set("1", "2", "3"))
+ }
+ }
+
+ test("mixed partitioning: identity field after a bucket transform -> ordinal 1") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (part_col string, id int, data string) " +
+ s"USING $v2Source PARTITIONED BY (bucket(4, id), part_col)")
+ sql(s"INSERT INTO $partFilterTableName VALUES ('a', 1, 'x'), ('A', 2, 'y'), ('b', 3, 'z')")
+
+ spark.udf.register("my_upper_second", (s: String) =>
+ if (s == null) null else s.toUpperCase(Locale.ROOT))
+
+ // The bucket field at ordinal 0 keeps its slot, so `part_col` binds to the second
+ // partition-key value and the reference reports ordinal 1.
+ val df = sql(s"SELECT * FROM $partFilterTableName WHERE my_upper_second(part_col) = 'A'")
+ checkAnswer(df, Seq(Row("a", 1, "x"), Row("A", 2, "y")))
+ assertPushedPartitionPredicates(df, 1)
+ assertScanReturnsPartitionKeys(df, Set("1/a", "2/A"))
+ assertReferencedPartitionFieldOrdinals(
+ df, Array(1), Array("bucket(4, id)", "part_col"))
+ }
+ }
+
+ test("mixed partitioning: cast from type coercion on the identity field is pruned by " +
+ "the second pass") {
+ withTable(partFilterTableName) {
+ sql(s"CREATE TABLE $partFilterTableName (dt string, id int, data string) " +
+ s"USING $v2Source PARTITIONED BY (dt, bucket(4, id))")
+ sql(s"INSERT INTO $partFilterTableName VALUES " +
+ "('2026-09-01', 1, 'x'), ('2026-09-02', 2, 'y'), ('2026-09-03', 3, 'z')")
+
+ // `dt = DATE'...'` is analyzed as `cast(dt AS DATE) = DATE'...'`. The source cannot
+ // evaluate a predicate over a cast and returns it in the first pass; the second pass
+ // evaluates it against the `dt` value of the full partition key.
+ val df = sql(s"SELECT * FROM $partFilterTableName WHERE dt = DATE'2026-09-02'")
+ checkAnswer(df, Seq(Row("2026-09-02", 2, "y")))
+ assertPushedPartitionPredicates(df, 1)
+ assertScanReturnsPartitionKeys(df, Set("2026-09-02/2"))
+ assertReferencedPartitionFieldOrdinals(df, Array(0), Array("dt", "bucket(4, id)"))
+ }
+ }
+
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/connector/DataSourceV2EnhancedRuntimePartitionFilterSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedRuntimePartitionFilterSuite.scala
index 07a214c24fbc..c6215d39dd0b 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedRuntimePartitionFilterSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedRuntimePartitionFilterSuite.scala
@@ -219,6 +219,61 @@ class DataSourceV2EnhancedRuntimePartitionFilterSuite
}
}
+ test("mixed partitioning: DPP on the identity col next to a bucket transform -> " +
+ "PartitionPredicate") {
+ val fact = s"$catalogName.fact_mixed"
+ val dim = s"$catalogName.dim_mixed"
+ withTable(fact, dim) {
+ sql(s"CREATE TABLE $fact (id INT, part INT) USING $v2Source " +
+ "PARTITIONED BY (part, bucket(4, id))")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $fact VALUES ($i, $i)")
+ }
+ sql(s"CREATE TABLE $dim (dim_id INT, dim_val STRING) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (2, 'two')")
+
+ withDPPConf {
+ val df = sql(
+ s"""SELECT f.id, f.part FROM $fact f JOIN $dim d
+ |ON f.part = d.dim_id WHERE d.dim_val = 'two'""".stripMargin)
+ checkAnswer(df, Row(2, 2))
+
+ assertDPPRuntimeFilters(df)
+ assertPushedPartitionPredicates(df, 1)
+ assertScanReturnsPartitionKeys(df, Set("2/2"))
+ assertReferencedPartitionFieldOrdinals(df, Array(0), Array("part", "bucket(4, id)"))
+ }
+ }
+ }
+
+ test("mixed partitioning: DPP on the source col of a bucket transform -> " +
+ "no PartitionPredicate") {
+ val fact = s"$catalogName.fact_mixed2"
+ val dim = s"$catalogName.dim_mixed2"
+ withTable(fact, dim) {
+ sql(s"CREATE TABLE $fact (id INT, part INT) USING $v2Source " +
+ "PARTITIONED BY (part, bucket(4, id))")
+ for (i <- 0 until 5) {
+ sql(s"INSERT INTO $fact VALUES ($i, $i)")
+ }
+ sql(s"CREATE TABLE $dim (dim_id INT, dim_val STRING) USING $v2Source")
+ sql(s"INSERT INTO $dim VALUES (2, 'two')")
+
+ withDPPConf {
+ val df = sql(
+ s"""SELECT f.id, f.part FROM $fact f JOIN $dim d
+ |ON f.id = d.dim_id WHERE d.dim_val = 'two'""".stripMargin)
+ checkAnswer(df, Row(2, 2))
+
+ // Spark cannot evaluate `id IN (...)` against the bucket value, so no PartitionPredicate
+ // is derived and the scan keeps every partition.
+ assertDPPRuntimeFilters(df)
+ assertPushedPartitionPredicates(df, 0)
+ assertScanReturnsPartitionKeys(df, Set("0/0", "1/1", "2/2", "3/3", "4/0"))
+ }
+ }
+ }
+
test("case 3: scalar subquery translatable, rejected in 1st pass -> PartitionPredicate") {
val tbl = s"$catalogName.tbl"
val dim = s"$catalogName.dim"
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
index 1165c7c35f31..f411ca4eef47 100644
--- 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
@@ -29,7 +29,7 @@ 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))
+ val fields = Seq(PartitionPredicateField(Seq("p"), Some(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"))
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org