dongjoon-hyun commented on PR #55579: URL: https://github.com/apache/spark/pull/55579#issuecomment-5768680813
Thanks for picking this up, @LuciferYang. I read through the diff and the surrounding paths, and the correctness core looks sound to me. Two comments plus a nit on the description. ### 1. Runtime filters prune after bin-packing, which costs parallelism `planInputPartitionsWithRuntimeFilters` filters files out of `FilePartition`s that `partitions` already bin-packed, and `maxSplitBytes` was derived there from the *unpruned* file set. V1 does the opposite: `FileSourceScanExec.createReadRDD` recomputes `FilePartition.maxSplitBytes` from `dynamicallySelectedPartitions` and bin-packs only the survivors. Concretely: a 12.8 GB fact table over 10 partition directories, with the default `spark.sql.files.maxPartitionBytes` (128 MB) and leaf parallelism 200, gives `maxSplitBytes = min(128MB, 12.8GB/200) = 64MB` and ~200 `FilePartition`s. A DPP filter selecting one directory (1.28 GB) leaves the ~20 bins that covered it, so the scan runs ~20 tasks of 64 MB. The V1 plan for the same query recomputes `maxSplitBytes` from 1.28 GB (6.4 MB) and produces ~200 tasks -- roughly a 10x parallelism difference. Bins that straddle two directories are additionally left under-filled by `part.copy(files = kept)`. The same loop also evaluates the predicate once per `PartitionedFile` rather than once per partition directory, so the driver does one interpreted `InSubqueryExec` lookup per file instead of per distinct partition value. I see the tradeoff you documented (one listing per scan, honoring whatever a subclass did in `partitions`), and I think both can be kept by re-running `FilePartition.maxSplitBytes` / `FilePartition.getFilePartitions` over the surviving `PartitionedFile`s instead of returning the filtered bins as they are. If you'd rather leave it, it seems worth listing next to the missing pruning metrics in the description. ### 2. Stale comment in the new suite In `DataSourceV2FileSourceDPPSuite`, the "a runtime filter is ANDed with the compile-time partition filters" test explains the mechanism as "`buildPartitions` gets `partitionFilters ++ expressions`", but there is no `buildPartitions` anywhere in the tree. The actual mechanism is that `planInputPartitionsWithRuntimeFilters` filters `plannedPartitions`, which `partitions` built with `partitionFilters` applied through `fileIndex.listFiles`. Looks like a leftover from an earlier iteration. ### Nit on the description > Such a subquery filter is no longer evaluated above the scan, so the scan is its only evaluator and a file whose partition value cannot be evaluated (an ANSI cast, an overflow) fails the query rather than being filtered out row by row This reads as more user-visible than it is. Whenever the partition directory holds rows, the pre-change plan evaluates the same cast in the `FilterExec` above the scan (or in the join key) and fails identically -- the `fact_str` test in this PR is exactly that shape. The genuinely new exposure looks narrow (e.g. a partition directory whose files carry no rows). -- 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]
