LuciferYang commented on PR #55579: URL: https://github.com/apache/spark/pull/55579#issuecomment-5966973332
@dongjoon-hyun Items 1 and 2 set the direction for several of the others, so before picking one I ran both alternatives against `FileScan` subclasses modeled on the ones you linked. The probe is a throwaway suite with three subclasses: one shaped like Kyuubi's `HiveScan` (`FileScan with SupportsRuntimeFiltering`, re-listing with the runtime filters in `partitions`), one like SPYT's `YtScan` (the same, except it caches `partitions` and clears the cache in `filter`), and one like Sedona's `ShapefileScan` (its own partition type from `planInputPartitions()`, read by a factory that rejects any other type; Sedona's `RasterScan`, `NetCdfMetadataScan` and `GeoTiffMetadataScan` have the same shape). Each runs a plain `SELECT`, a DPP join and a scalar subquery on the partition column, next to built-in parquet. | | built-in parquet | Kyuubi-like | SPYT-like | Sedona-like | |---|---|---|---|---| | master | no pruning | DPP prunes | DPP prunes | works, no pruning | | this PR | DPP and subquery prune | every query fails at planning | every query fails at planning | DPP and subquery fail: `Unexpected partition type: FilePartition` | | A: per-format opt-in | DPP and subquery prune | same as master | same as master | same as master | | B: overridable hook | DPP and subquery prune | every query fails at planning | every query fails at planning | same as master | | B + precedence rule, memo behind the flag | DPP and subquery prune | same as master | same as master | same as master | In the table, every query that ran returned the right rows. The planning failure is the `INTERNAL_ERROR` from the dual-interface check. Option A moves `SupportsRuntimeCatalystFiltering` and its three methods off `FileScan` into a trait that `ParquetScan`, `OrcScan`, `AvroScan` and `TextBasedFileScan` (CSV, JSON, text) mix in, so any other `FileScan` subclass sees what it sees on master. The memo has to move into that trait as well. When I left it in `FileScan.planInputPartitions()`, the SPYT-like scan's DPP stopped pruning without any error: its fact scan read 100 rows instead of 10, and only the join kept the answer right. `supportsColumnar` fills the memo at planning, and the `filter()` that later clears YtScan's own cache never reaches it. With A, items 4 and 12 also apply only to the scans that mix the trait in. Option B keeps the interface on `FileScan` and returns no filter attributes unless a subclass turns on an overridable flag, which the built-in scans do. On its own that fixes Sedona but not Kyuubi or SPYT: both dual-interface checks (DataSourceV2Relation.scala:309, PushDownUtils.scala:190) match on the type, and a subclass cannot drop a trait it inherits. Dropping the checks returns wrong rows: the Kyuubi-like and SPYT-like scans then inherit a `fullyPushedFilterAttributes()` that returns their own partition columns, Spark removes the `FilterExec`, and their `filter()` ignores the scalar subquery, so that query returned 100 rows instead of 10. What worked in the probe is the last row: a precedence rule, where a dual-interface scan takes the V2 path and Spark ignores its `fullyPushedFilterAttributes()`, plus the memo behind the same flag. Without the memo change the SPYT-like fact scan read 100 rows under DPP, the same silent loss as leaving the memo in `FileScan` under A. The prec edence rule replaces the one-interface rule SPARK-58523 set, and the probe covers only the shapes above. The other route is that each such connector moves to the Catalyst interface. @pan3793, for Kyuubi the second route means `HiveScan`, `KyuubiParquetScan` and `KyuubiOrcScan` drop `SupportsRuntimeFiltering` and each override `planInputPartitionsWithRuntimeFilters`. All three re-list with the runtime filters today, `HiveScan` through `listHiveFiles(partitionFilters ++ runtimeFilters)` and the Parquet and ORC wrappers through a sibling scan built with `partitionFilters ++ runtimeFilters`, while the inherited default filters a listing made without them. The filters would arrive as Catalyst expressions rather than `In` filters, and since `FileScan` declares the partition columns fully pushed, each override has to apply every expression it gets, or return nothing from `fullyPushedFilterAttributes()`. The interface is already in 4.3, with a mutating `filter(Array[Expression])` that this PR replaces, and the connector is one binary for Spark 3.5 through 4.2 (compiled against 3.5, tested on 4.0 to 4.2), so these scans would need separately compiled classes for 4.4, as this PR is meant for branch-4.x too. Does that match how you see it? My preference is A: among `FileScan` subclasses it changes behavior only for the built-in scans, and it leaves the one-interface rule alone. A third-party `FileScan` that wants runtime pruning can mix the trait in itself. I'd like to hear what you both think before I change it. I'll answer the other comments inline. -- 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]
