szehon-ho commented on code in PR #57727:
URL: https://github.com/apache/spark/pull/57727#discussion_r3731865370
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala:
##########
@@ -267,6 +270,94 @@ class InMemoryRowLevelOperationTable private (
override def abort(messages: Array[WriterCommitMessage]): Unit = {}
}
+
+ /**
+ * Builds a scan for row-level operations. When
+ * `use-catalyst-runtime-filtering` is set, the scan implements
+ * [[SupportsRuntimeCatalystFiltering]] so group filtering goes through the
Catalyst path.
+ */
+ private def newRowLevelScanBuilder(
+ options: CaseInsensitiveStringMap)(
+ onBuild: BatchScanBaseClass => Unit): ScanBuilder = {
+ new InMemoryScanBuilder(schema, options) {
+ override protected def createScan(
+ partitions: Seq[InputPartition],
+ readSchema: StructType,
+ tableSchema: StructType,
+ options: CaseInsensitiveStringMap): BatchScanBaseClass = {
+ if (useCatalystRuntimeFiltering) {
+ InMemoryCatalystRowLevelBatchScan(partitions, readSchema,
tableSchema, options)
+ } else {
+ super.createScan(partitions, readSchema, tableSchema, options)
+ }
+ }
+
+ override def build: Scan = {
+ val scan = super.build().asInstanceOf[BatchScanBaseClass]
+ onBuild(scan)
+ scan
+ }
+ }
+ }
+
+ /**
+ * Row-level batch scan that receives runtime filters as Catalyst
expressions.
+ * Evaluates partition-column predicates against each partition key so group
filtering
+ * actually prunes partitions (needed for `replacedPartitions` assertions).
+ */
+ case class InMemoryCatalystRowLevelBatchScan(
+ var _data: Seq[InputPartition],
+ readSchema: StructType,
+ tableSchema: StructType,
+ options: CaseInsensitiveStringMap)
+ extends BatchScanBaseClass(_data, readSchema, tableSchema)
+ with SupportsRuntimeCatalystFiltering {
+
+ private val _catalystPredicates = ArrayBuffer.empty[Expression]
+
+ override def filterAttributes(): Array[NamedReference] = {
+ val scanFields = readSchema.fields.map(_.name).toSet
+ partitioning.flatMap(_.references())
+ .filter(ref => scanFields.contains(ref.fieldNames.mkString(".")))
+ }
+
+ override def filter(expressions: Array[Expression]): Unit = {
Review Comment:
Extracted into a `CatalystRuntimeFilteringScan` trait next to
`BatchScanBaseClass` in `InMemoryBaseTable`, carrying `filter`,
`partitionAttributes` and `pushedCatalystPredicates`, with a `tableSchema`
member for locating partition columns pruned out of `readSchema`. Both scans
now mix it in and override only `filterAttributes()`.
The "keep the partition on eval failure, matching `PartitionPredicateImpl`"
comment is in the shared copy.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -218,6 +222,26 @@ object PushDownUtils extends Logging {
}
translatedFiltersPushed || partPredicatesPushed
+
+ case catalystScan: SupportsRuntimeCatalystFiltering if
runtimeFilters.nonEmpty =>
+ // A DPP filter degrades to TrueLiteral when its subquery is pruned
away; it carries no
+ // information for the source. The V2 path above drops these
implicitly because
+ // translateRuntimeFilterV2 returns None; here we push Catalyst
expressions directly,
+ // so filter them out explicitly.
+ // Screen with the same pushability guard as the V2 PartitionPredicate
path
+ // (deterministic, no subquery, no Python UDF). Keeps
non-deterministic filters
+ // from being the sole evaluator when fullyPushedFilterAttributes
drops FilterExec.
Review Comment:
Both fixed, and thanks for checking the claims rather than the code alone.
The screen now runs before the unwrap with `includeSubquery = true`, exactly
as suggested, so the two sides apply one predicate to one form. The comment
says what the screen is actually for: nothing non-deterministic reaches this
line after the rebase, so what it rejects is a residual subquery or a Python
UDF.
Your `ExtractPythonUDFs` point matches what I found trying to write a
regression test for the `DataSourceV2Strategy` side: the UDF is lifted into a
`BatchEvalPython` node below the `Filter`, so the filter is no longer
scan-adjacent and `PhysicalOperation` never offers it as a post-scan filter. I
could not construct a case where the two screens disagree today, so that gate
is drift protection rather than a live fix -- which is also why there is no
test for it.
--
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]
