dongjoon-hyun commented on code in PR #58756:
URL: https://github.com/apache/spark/pull/58756#discussion_r4001846479
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala:
##########
@@ -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()
Review Comment:
With this delegation, row-level scans now go through the second-pass
`PartitionPredicate` path. When a `PartitionPredicate` is accepted, Spark drops
the original filter: from the post-scan `Filter` for delta-based operations
(`V2ScanRelationPushDown`), and from the MERGE join condition for group-based
MERGE (the filter ends up in `evaluatedFilters`, which
`optimizeMergeJoinCondition` removes). But `PartitionPredicateImpl.eval` fails
open: it returns `true` on an evaluation exception or a field-count mismatch.
For example, under ANSI mode, `MERGE INTO t USING s ON t.id = s.id AND
CAST(t.dep AS INT) = 1 WHEN MATCHED THEN UPDATE ...` should fail with
`CAST_INVALID_INPUT` when a row in the `hr` partition matches by `id`. With
this change, the fail-open `eval` keeps the `hr` partition, the conjunct is
dropped from the join condition, and the matching `hr` rows are updated without
any error. A delta-based `UPDATE t SET ... WHERE CAST(dep AS INT) = 1` has the
same problem. Group-based UPDATE/DELETE are safe because the rewrite plan
evaluates the condition again.
The fail-open behavior itself comes from SPARK-55596, but this PR extends it
from reads to UPDATE/MERGE writes. For row-level operations, should we keep the
original filter (the post-scan filter or the join conjunct) even when a
`PartitionPredicate` is accepted?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala:
##########
@@ -107,6 +113,72 @@ class InMemoryPartitionPredicateDeleteTable(
}
}
+ /**
+ * Row-level scans push V2 predicates iteratively, so a group-based
operation receives a
+ * second-pass [[PartitionPredicate]] the same way a metadata-only DELETE
does. Only partition
+ * predicates prune, by partition key; a data predicate is always returned
since the scan
+ * cannot filter rows.
+ */
+ override protected def newRowLevelScanBuilder(
+ options: CaseInsensitiveStringMap)(
+ onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
+ new PartitionPredicateRowLevelScanBuilder(onBuild)
+ }
+
+ class PartitionPredicateRowLevelScanBuilder(onBuild: BatchScanBaseClass =>
Unit)
+ extends ScanBuilder with SupportsPushDownV2Filters with
SupportsPushDownRequiredColumns {
+
+ private var readSchema: StructType = schema
+ private val pushed = ArrayBuffer.empty[Predicate]
+
+ override def supportsIterativePushdown(): Boolean = true
+
+ override def pushPredicates(predicates: Array[Predicate]):
Array[Predicate] = {
+ val (accepted, returned) = predicates.partition {
+ case _: PartitionPredicate => acceptPartitionPredicates
+ case p => refsOnlyPartCols(p) &&
InMemoryTableWithV2Filter.supportsPredicates(Array(p))
Review Comment:
This accepts any `=`, `<=>`, `IS_NULL`, etc. whose references are all
partition columns, whatever its children are. `build()` then assumes that
`children(0)` is a partition column name and `children(1)` is a `LiteralValue`
(`filtersToKeys` -> `evalPredicate`). For example, `UPDATE t SET salary = 0
WHERE upper(dep) = 'HR'` translates to `=(UPPER(dep), 'HR')`. This builder
accepts it, and planning then fails in `InMemoryBaseTable.extractValue` with
`Unknown filter attribute: UPPER(dep)`. Could we accept only a column reference
compared with a literal, and return everything else?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala:
##########
@@ -107,6 +113,72 @@ class InMemoryPartitionPredicateDeleteTable(
}
}
+ /**
+ * Row-level scans push V2 predicates iteratively, so a group-based
operation receives a
+ * second-pass [[PartitionPredicate]] the same way a metadata-only DELETE
does. Only partition
+ * predicates prune, by partition key; a data predicate is always returned
since the scan
+ * cannot filter rows.
+ */
+ override protected def newRowLevelScanBuilder(
Review Comment:
This override skips everything the parent `newRowLevelScanBuilder` does: the
`use-catalyst-runtime-filtering` branch, `options`, the ordering-aware scan,
and `recordScanEvent`. `PartitionPredicateRowLevelBatchScan` also has no
runtime filtering, so `RowLevelOperationRuntimeGroupFiltering` can no longer
inject group filters for this table. Delta-based operations are affected too,
since they call the same builder.
As a result, the existing fallback tests in this suite (e.g. `first and
second pass rejected: table rejects all`) used to prune through the V1 `In`
pushdown (`canEvaluate`) and now rewrite the whole table. They still pass
because they check only the final rows, so this coverage loss goes unnoticed.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala:
##########
@@ -241,6 +245,31 @@ 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") {
Review Comment:
This test covers only group-based UPDATE. The PR description, and the
updated class doc, also cover MERGE, delta-based scans
(`V2ScanRelationPushDown`), and runtime filter pushdown. Could you add at least
a group-based MERGE test (e.g. `ON t.pk = s.pk AND t.dep IN ('hr',
'software')`) that checks both the join condition and the result, plus a
delta-based (`supports-deltas`) UPDATE test? MERGE matters most, because an
accepted `PartitionPredicate` now removes the conjunct from the join condition.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala:
##########
@@ -107,6 +113,72 @@ class InMemoryPartitionPredicateDeleteTable(
}
}
+ /**
+ * Row-level scans push V2 predicates iteratively, so a group-based
operation receives a
+ * second-pass [[PartitionPredicate]] the same way a metadata-only DELETE
does. Only partition
+ * predicates prune, by partition key; a data predicate is always returned
since the scan
+ * cannot filter rows.
+ */
+ override protected def newRowLevelScanBuilder(
+ options: CaseInsensitiveStringMap)(
+ onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
+ new PartitionPredicateRowLevelScanBuilder(onBuild)
+ }
+
+ class PartitionPredicateRowLevelScanBuilder(onBuild: BatchScanBaseClass =>
Unit)
Review Comment:
nit: This builder and `PartitionPredicateRowLevelBatchScan` mostly duplicate
`InMemoryEnhancedPartitionFilterScanBuilder` and
`InMemoryEnhancedPartitionFilterBatchScan` in
`InMemoryEnhancedPartitionFilterTable`: the accumulating `pushPredicates`,
`build()` with `filtersToKeys` plus `PartitionPredicate.eval`, and a scan that
holds `pushedPartitionPredicates`. `pruneColumns` also duplicates
`InMemoryScanBuilder.pruneColumns`. Could we move the shared part into
`InMemoryBaseTable` so both fixtures use it?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryPartitionPredicateDeleteTable.scala:
##########
@@ -107,6 +113,72 @@ class InMemoryPartitionPredicateDeleteTable(
}
}
+ /**
+ * Row-level scans push V2 predicates iteratively, so a group-based
operation receives a
+ * second-pass [[PartitionPredicate]] the same way a metadata-only DELETE
does. Only partition
+ * predicates prune, by partition key; a data predicate is always returned
since the scan
+ * cannot filter rows.
+ */
+ override protected def newRowLevelScanBuilder(
+ options: CaseInsensitiveStringMap)(
+ onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
+ new PartitionPredicateRowLevelScanBuilder(onBuild)
+ }
+
+ class PartitionPredicateRowLevelScanBuilder(onBuild: BatchScanBaseClass =>
Unit)
+ extends ScanBuilder with SupportsPushDownV2Filters with
SupportsPushDownRequiredColumns {
+
+ private var readSchema: StructType = schema
+ private val pushed = ArrayBuffer.empty[Predicate]
+
+ override def supportsIterativePushdown(): Boolean = true
+
+ override def pushPredicates(predicates: Array[Predicate]):
Array[Predicate] = {
+ val (accepted, returned) = predicates.partition {
+ case _: PartitionPredicate => acceptPartitionPredicates
+ case p => refsOnlyPartCols(p) &&
InMemoryTableWithV2Filter.supportsPredicates(Array(p))
+ }
+ pushed ++= accepted
+ returned
+ }
+
+ override def pushedPredicates(): Array[Predicate] = pushed.toArray
+
+ 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 (partPreds, stdPreds) =
pushed.toArray.partition(_.isInstanceOf[PartitionPredicate])
+ val partitionPredicates =
partPreds.map(_.asInstanceOf[PartitionPredicate])
+ val keys = InMemoryTableWithV2Filter.filtersToKeys(
+ data.map(_.key).toImmutableArraySeq,
Review Comment:
nit: `data` rebuilds `dataMap.values.flatten.toArray` on every call. Here it
is called twice (for the keys and for the partitions), outside
`dataMap.synchronized`. How about computing it once, as
`InMemoryEnhancedPartitionFilterTable.build()` does with `allPartitions`?
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2EnhancedDeleteFilterSuite.scala:
##########
@@ -357,6 +386,35 @@ 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}")
+ val scans = collect(plan) { case s: BatchScanExec => s }
+ val scan = scans.map(_.scan).collectFirst {
+ case s:
InMemoryPartitionPredicateDeleteTable#PartitionPredicateRowLevelBatchScan => s
+ }.getOrElse(fail("Expected the row-level scan of the in-memory table"))
+ assertPartitionFieldReferences(
+ scan.pushedPartitionPredicates.toArray, Seq(expectedOrdinals),
expectedPartitionFieldNames)
+
+ val table = scans.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.toSet === expectedReplacedDeps &&
Review Comment:
nit: `replacedPartitions` is already `distinct` (see
`PartitionBasedReplaceData.doCommit`), so the size check adds nothing. Also,
joining two `===` with `&&` loses ScalaTest's diff message. How about
`assert(replacedDeps.sorted === expectedReplacedDeps.toSeq.sorted)`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
