pan3793 commented on code in PR #58756:
URL: https://github.com/apache/spark/pull/58756#discussion_r4036687275
##########
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:
Fixed in SPARK-59572: `eval` now propagates the failure instead of reporting
a match, except
for runtime filters, which Spark evaluates again after the scan. It was not
specific to
row-level operations; the static second pass and the metadata-only DELETE
rewrite were already
in that position before this PR. This PR is rebased on it.
##########
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:
Added in 0abbdb2: a group-based MERGE test, which also asserts the conjunct
is dropped from the
join condition and that the pruned partition's row is untouched, and a
delta-based UPDATE test,
which reaches the second pass through `V2ScanRelationPushDown` rather than
the group-based rule.
##########
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:
Fixed in 0abbdb2. The V2 builder is now opt-in via an
`iterative-row-level-pushdown` table
property, so every existing test keeps the inherited builder, with its V1
pushdown, the
`use-catalyst-runtime-filtering` branch, the ordering-aware scan and
`recordScanEvent`. The
pruning coverage you describe is therefore intact.
One builder cannot serve both paths: Spark takes the V1 branch whenever a
builder offers
`SupportsPushDownFilters`, and only the V2 interface has a second pass. An
opted-in table
consequently gets no runtime group filtering, which is now stated in the
fixture comment.
##########
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:
Fixed in 0abbdb2, returning more than you suggested: the builder accepts only
PartitionPredicates and returns every other predicate, so nothing it cannot
evaluate reaches
the partition key lookup. The partition-column IN still prunes, because it
comes back as a
PartitionPredicate in the second pass.
##########
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:
Left as is. Accepting only PartitionPredicates removed `filtersToKeys`, the
partition-name
plumbing and the shape check, so what still overlaps is an accumulating
`pushPredicates` and a
scan that holds its pushed predicates. Moving that into `InMemoryBaseTable`
would also turn
this into a manual merge on branch-4.2 and branch-4.3, which this fix
targets.
--
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]
