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/18 05:29:18 UTC
(spark) branch branch-4.x updated: [SPARK-59457][SQL] RowLevelOperationTable should delegate partitioning() to the wrapped table
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 44c2cd7f6c91 [SPARK-59457][SQL] RowLevelOperationTable should delegate partitioning() to the wrapped table
44c2cd7f6c91 is described below
commit 44c2cd7f6c91f6398b2299b498eff0c67b5511e6
Author: Cheng Pan <pa...@gmail.com>
AuthorDate: Fri Sep 18 13:28:46 2026 +0800
[SPARK-59457][SQL] RowLevelOperationTable should delegate partitioning() to the wrapped table
### What changes were proposed in this pull request?
`RowLevelOperationTable`, the internal wrapper a row-level rewrite (UPDATE, MERGE, group-based
DELETE) reads the table through, delegates `name`, `columns`, `capabilities` and `constraints`
to the wrapped table but not `partitioning()`, so it returned the default empty array. It now
delegates `partitioning()` too.
The in-memory row-level test table can opt into pushing V2 predicates iteratively, via an
`iterative-row-level-pushdown` table property. It accepts only partition predicates.
### Why are the changes needed?
Since SPARK-55596 (4.2.0), `PushDownUtils.getPartitionPredicateSchema` reads the partitioning
of the read relation's table to derive `PartitionPredicate`s, in the group-based row-level
scan planning rule, in `V2ScanRelationPushDown` for delta-based scans, and in runtime filter
pushdown. For every row-level operation the table it sees is the wrapper, so the schema was
always empty and no `PartitionPredicate` was ever derived for a row-level scan.
The wrapper has been this way since SPARK-38085 (3.3.0), but nothing read its partitioning
before 4.2.0, so the gap was invisible until then.
### Does this PR introduce _any_ user-facing change?
No. A connector that opts into iterative pushdown now receives `PartitionPredicate`s for
UPDATE, MERGE and DELETE scans the same way it does for SELECT.
### How was this patch tested?
New tests: a group-based UPDATE, a group-based MERGE and a delta-based UPDATE each push a
second-pass `PartitionPredicate` to the row-level scan and prune to the matching partitions.
All fail without the fix.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5.1
Closes #58756 from pan3793/row-level-table-partitioning.
Authored-by: Cheng Pan <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit 110ab709cef8841701047bce5754968546e63c40)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../connector/write/RowLevelOperationTable.scala | 2 +
.../InMemoryPartitionPredicateDeleteTable.scala | 79 +++++++++++-
.../catalog/InMemoryRowLevelOperationTable.scala | 2 +-
.../DataSourceV2EnhancedDeleteFilterSuite.scala | 138 ++++++++++++++++++++-
4 files changed, 215 insertions(+), 6 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
index 50179824e255..0ed78b667db3 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
@@ -21,6 +21,7 @@ import java.util
import org.apache.spark.sql.connector.catalog.{Column, SupportsRead, SupportsRowLevelOperations, SupportsWrite, Table, TableCapability}
import org.apache.spark.sql.connector.catalog.constraints.Constraint
+import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.read.ScanBuilder
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -40,6 +41,7 @@ private[sql] case class RowLevelOperationTable(
override def columns: Array[Column] = table.columns()
override def capabilities: util.Set[TableCapability] = table.capabilities
override def constraints(): Array[Constraint] = table.constraints()
+ override def partitioning(): Array[Transform] = table.partitioning()
override def toString: String = table.toString
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = {
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 497f715776e5..91f26741a08a 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
@@ -19,17 +19,23 @@ package org.apache.spark.sql.connector.catalog
import java.util
+import scala.collection.mutable.ArrayBuffer
+
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
+import org.apache.spark.sql.catalyst.expressions.MetadataStructFieldWithLogicalName
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.MultipartIdentifierHelper
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.expressions.filter.{PartitionPredicate, Predicate}
+import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder, SupportsPushDownRequiredColumns, SupportsPushDownV2Filters}
import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.ArrayImplicits._
/**
* In-memory table that supports row-level operations and accepts [[PartitionPredicate]]s
- * in V2 [[canDeleteWhere]]/[[deleteWhere]] for metadata-only deletes.
+ * in V2 [[canDeleteWhere]]/[[deleteWhere]] for metadata-only deletes, and, when
+ * `iterative-row-level-pushdown` is set, in the scan of a row-level operation.
*
* Contains some knobs to control acceptance of various partition and data predicates.
*/
@@ -44,6 +50,10 @@ class InMemoryPartitionPredicateDeleteTable(
properties.getOrDefault(
InMemoryPartitionPredicateDeleteTable.AcceptPartitionPredicatesKey, "true").toBoolean
+ private val iterativeRowLevelPushdown: Boolean =
+ properties.getOrDefault(
+ InMemoryPartitionPredicateDeleteTable.IterativeRowLevelPushdownKey, "false").toBoolean
+
private val acceptDataPredicates: Boolean =
properties.getOrDefault(
InMemoryPartitionPredicateDeleteTable.AcceptDataPredicatesKey, "false").toBoolean
@@ -107,6 +117,72 @@ class InMemoryPartitionPredicateDeleteTable(
}
}
+ /**
+ * With `iterative-row-level-pushdown`, the scan of a row-level operation pushes V2 predicates
+ * iteratively, so it receives a second-pass [[PartitionPredicate]] the same way a
+ * metadata-only DELETE does. It accepts only partition predicates and returns every other
+ * predicate for Spark to evaluate.
+ *
+ * Otherwise the inherited builder is used, which pushes V1 filters and reports runtime filter
+ * attributes. One builder cannot do both: Spark pushes V1 filters whenever a builder offers
+ * them, and only the V2 interface has a second pass.
+ */
+ override protected def newRowLevelScanBuilder(
+ options: CaseInsensitiveStringMap)(
+ onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
+ if (iterativeRowLevelPushdown) {
+ new PartitionPredicateRowLevelScanBuilder(onBuild)
+ } else {
+ super.newRowLevelScanBuilder(options)(onBuild)
+ }
+ }
+
+ class PartitionPredicateRowLevelScanBuilder(onBuild: BatchScanBaseClass => Unit)
+ extends ScanBuilder with SupportsPushDownV2Filters with SupportsPushDownRequiredColumns {
+
+ private var readSchema: StructType = schema
+ private val pushed = ArrayBuffer.empty[PartitionPredicate]
+
+ override def supportsIterativePushdown(): Boolean = true
+
+ override def pushPredicates(predicates: Array[Predicate]): Array[Predicate] = {
+ val (accepted, returned) = predicates.partition {
+ case _: PartitionPredicate => acceptPartitionPredicates
+ case _ => false
+ }
+ pushed ++= accepted.map(_.asInstanceOf[PartitionPredicate])
+ returned
+ }
+
+ override def pushedPredicates(): Array[Predicate] = pushed.toArray[Predicate]
+
+ override def pruneColumns(requiredSchema: StructType): Unit = {
+ val metadataNames = metadataColumns.map(_.name).toSet
+ val schemaNames = schema.map(_.name).toSet
+ readSchema = StructType(requiredSchema.filter {
+ case MetadataStructFieldWithLogicalName(_, name) => metadataNames.contains(name)
+ case f => schemaNames.contains(f.name)
+ })
+ }
+
+ override def build(): Scan = {
+ val partitions = data.filter(p => pushed.forall(_.eval(p.partitionKey())))
+ val scan = PartitionPredicateRowLevelBatchScan(
+ partitions.map(_.asInstanceOf[InputPartition]).toImmutableArraySeq,
+ readSchema, schema, pushed.toSeq)
+ onBuild(scan)
+ scan
+ }
+ }
+
+ /** Row-level batch scan that records the [[PartitionPredicate]]s it was pruned by. */
+ case class PartitionPredicateRowLevelBatchScan(
+ _data: Seq[InputPartition],
+ readSchema: StructType,
+ tableSchema: StructType,
+ pushedPartitionPredicates: Seq[PartitionPredicate])
+ extends BatchScanBaseClass(_data, readSchema, tableSchema)
+
private def rowMatchesAll(
row: InternalRow,
preds: Array[Predicate],
@@ -123,6 +199,7 @@ class InMemoryPartitionPredicateDeleteTable(
object InMemoryPartitionPredicateDeleteTable {
private[catalog] val AcceptPartitionPredicatesKey = "accept-partition-predicates"
private[catalog] val AcceptDataPredicatesKey = "accept-data-predicates"
+ private[catalog] val IterativeRowLevelPushdownKey = "iterative-row-level-pushdown"
}
class InMemoryPartitionPredicateDeleteCatalog extends InMemoryTableCatalog {
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
index 7556bc96912f..c08dfef04a5b 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
@@ -272,7 +272,7 @@ class InMemoryRowLevelOperationTable private (
* scan mixes in [[CatalystRuntimeFilteringScan]] so group filtering goes through the Catalyst
* path.
*/
- private def newRowLevelScanBuilder(
+ protected def newRowLevelScanBuilder(
options: CaseInsensitiveStringMap)(
onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
new InMemoryScanBuilder(schema, options) {
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 d055871e9a3e..953eb9e38121 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
@@ -19,20 +19,27 @@ package org.apache.spark.sql.connector
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.catalog.{InMemoryPartitionPredicateDeleteCatalog, InMemoryPartitionPredicateDeleteTable}
import org.apache.spark.sql.connector.expressions.PartitionFieldReference
import org.apache.spark.sql.connector.expressions.filter.PartitionPredicate
+import org.apache.spark.sql.connector.write.RowLevelOperationTable
import org.apache.spark.sql.execution.{QueryExecution, SparkPlan}
-import org.apache.spark.sql.execution.datasources.v2.{DeleteFromTableExec, ReplaceDataExec, WriteDeltaExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.execution.datasources.v2.{BatchScanExec, DeleteFromTableExec, ReplaceDataExec, WriteDeltaExec}
+import org.apache.spark.sql.execution.joins.BaseJoinExec
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.util.QueryExecutionListener
/**
* Tests for metadata-only delete optimization using second-pass
- * PartitionPredicate (see SPARK-55596).
+ * PartitionPredicate (see SPARK-55596), and for the second pass in the scan of a
+ * group-based UPDATE or MERGE (GroupBasedRowLevelOperationScanPlanning).
*/
-class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession {
+class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession
+ with AdaptiveSparkPlanHelper {
+
+ import testImplicits._
private val v2Source = classOf[FakeV2ProviderWithCustomSchema].getName
private val catalogName = "ppd_cat"
@@ -270,6 +277,97 @@ class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession {
}
}
+ // A group-based UPDATE reads the table through RowLevelOperationTable. The wrapper reports
+ // the table's partitioning, so the row-level scan, which pushes V2 predicates iteratively,
+ // receives the IN on the partition column as a second-pass PartitionPredicate, and only the
+ // matching partitions are read and replaced.
+ test("SPARK-59457: group-based UPDATE receives a second-pass PartitionPredicate") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep) " +
+ "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"),
+ expectedReplacedDeps = Set("hr", "software"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "hr", 101), Row(2, "software", 201), Row(3, "marketing", 300)))
+ }
+ }
+
+ // The scan evaluates the pushed predicate, so Spark drops that conjunct from the MERGE join
+ // condition. It must therefore have pruned to the partitions the conjunct accepts.
+ test("SPARK-59457: group-based MERGE receives a second-pass PartitionPredicate") {
+ withTable(deleteTableName) {
+ withTempView("source") {
+ sql(s"CREATE TABLE $deleteTableName (pk INT, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep) " +
+ "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"),
+ expectedReplacedDeps = Set("hr", "software"))
+
+ val joins = collect(plan) { case j: BaseJoinExec => j }
+ assert(joins.nonEmpty, "Expected a join in the MERGE plan")
+ joins.foreach { j =>
+ assert(!j.condition.exists(_.references.exists(_.name == "dep")),
+ s"Evaluated IN on dep should be dropped from the join condition: ${j.condition}")
+ }
+
+ // 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)))
+ }
+ }
+ }
+
+ // A delta-based operation plans its scan through V2ScanRelationPushDown rather than
+ // GroupBasedRowLevelOperationScanPlanning, and reads the same wrapped partitioning.
+ test("SPARK-59457: delta-based UPDATE receives a second-pass PartitionPredicate") {
+ withTable(deleteTableName) {
+ sql(s"CREATE TABLE $deleteTableName (pk INT NOT NULL, dep STRING, salary INT) " +
+ s"USING $v2Source PARTITIONED BY (dep) " +
+ "TBLPROPERTIES('supports-deltas' = 'true', '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')")
+ }
+ assert(plan.isInstanceOf[WriteDeltaExec],
+ s"Expected WriteDeltaExec but got: ${plan.getClass.getSimpleName}")
+ assertPartitionFieldReferences(
+ rowLevelScanPartitionPredicates(plan).toArray, Seq(Array(0)), Array("dep"))
+
+ checkAnswer(
+ sql(s"SELECT * FROM $deleteTableName"),
+ Seq(Row(1, "hr", 101), Row(2, "software", 201), Row(3, "marketing", 300)))
+ }
+ }
+
private def executeAndKeepPlan(func: => Unit): SparkPlan = keepingPlan(func)._2
/**
@@ -395,6 +493,38 @@ class DataSourceV2EnhancedDeleteFilterSuite extends SharedSparkSession {
}
}
+ /**
+ * Asserts that the group-based plan's row-level scan was pruned by one PartitionPredicate with
+ * the given references, and that only the partitions with the given `dep` values, the first
+ * partition field, were replaced.
+ */
+ private def assertRowLevelScanPrunedByPartitionPredicate(
+ plan: SparkPlan,
+ expectedOrdinals: Array[Int],
+ expectedPartitionFieldNames: Array[String],
+ expectedReplacedDeps: Set[String]): Unit = {
+ assert(plan.isInstanceOf[ReplaceDataExec],
+ s"Expected ReplaceDataExec but got: ${plan.getClass.getSimpleName}")
+ assertPartitionFieldReferences(
+ rowLevelScanPartitionPredicates(plan).toArray,
+ Seq(expectedOrdinals),
+ expectedPartitionFieldNames)
+
+ val table = collect(plan) { case s: BatchScanExec => s }.map(_.table).collectFirst {
+ case RowLevelOperationTable(t: InMemoryPartitionPredicateDeleteTable, _) => t
+ }.getOrElse(fail("Expected the row-level operation table"))
+ val replacedDeps = table.replacedPartitions.map(_.head.toString)
+ assert(replacedDeps.sorted === expectedReplacedDeps.toSeq.sorted)
+ }
+
+ /** The PartitionPredicates the in-memory row-level scan in `plan` was pruned by. */
+ private def rowLevelScanPartitionPredicates(plan: SparkPlan): Seq[PartitionPredicate] = {
+ collect(plan) { case s: BatchScanExec => s }.map(_.scan).collectFirst {
+ case s: InMemoryPartitionPredicateDeleteTable#PartitionPredicateRowLevelBatchScan => s
+ }.getOrElse(fail("Expected the row-level scan of the in-memory table"))
+ .pushedPartitionPredicates
+ }
+
private def assertDeleteWithRowLevel(query: String): Unit = {
val plan = executeAndKeepPlan { sql(query) }
plan match {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org