szehon-ho commented on code in PR #58702:
URL: https://github.com/apache/spark/pull/58702#discussion_r4008027851
##########
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:
Test-only follow-up: `partCols` includes references from all transforms. For
`(dep, bucket(4, pk))`, this may accept `pk = 4` and evaluate it against the
bucket-value slot. This does not affect production code or the
mixed-partitioning tests in this PR, but could make the fixture model ordinary
filter pushdown incorrectly. Should `refsOnlyPartCols` use only identity
transforms?
##########
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:
Good catch. Row-level rewrites replace the original relation table with this
wrapper, so without this delegation `getPartitionPredicateSchema` sees the
default empty partitioning and disables second-pass `PartitionPredicate`
pushdown. This enables UPDATE/MERGE/DELETE, including mixed partitioning.
--
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]
