LuciferYang commented on PR #55579: URL: https://github.com/apache/spark/pull/55579#issuecomment-5772975648
Thanks @dongjoon-hyun. All three hold up; the first one is fixed, with a caveat about your example. ### 1. Packing Fixed. The runtime path now collects the surviving `PartitionedFile`s, recomputes `FilePartition.maxSplitBytes` from their own size, and runs `getFilePartitions` over them, instead of returning `part.copy(files = kept)`. A new test compares the layout against the same partitions selected at compile time (`f.part >= 8`); reverting only that hunk fails it with `1 did not equal 4 expected one partition per surviving file, got 1 for 4 files`. The caveat: this does not change your example, and the reason is worth stating. `getFilePartitions` packs but never splits, so the floor on the task count is the number of `PartitionedFile` pieces, which `partitions` already fixed at the unpruned `maxSplitBytes`. With 64 MB pieces the repack gives each piece its own bin and you still get ~20 tasks, not 200. Where it does match `createReadRDD` exactly is the many-small-files shape: a file below `maxSplitBytes` is never split, so repacking the survivors at the pruned set's `maxSplitBytes` reproduces V1's layout. The new test also measures the boundary between the two: raising `spark.sql.files.minPartitionNum` until the pruned set's `maxSplitBytes` drops below one file's length makes the compile-time arm cut its files finer, and the arms separate (6 partitions against the runtime arm's 4). Closing that means re-splitting, which needs the pre-split file list, i.e. the second read of the index that the earlier revision did and that r40 48745686 asked me to drop. I changed that mechanism without re-checking the packing, which is how this got in. The residual is now in the description next to the missing metrics. On the per-file evaluation: `InSubqueryExec.eval` delegates to an `InSet` built once from `result`, so each file costs a hash lookup. I left it per file rather than grouping by partition values. ### 2. Stale comment Fixed, thanks. It described the previous mechanism. It now says that `partitions` applies `partitionFilters` through `fileIndex.listFiles` and `planInputPartitionsWithRuntimeFilters` filters what that planned. ### 3. Description You are right, and it is narrower still: `PartitioningAwareFileIndex.prunePartitions` evaluates the same expression once per partition directory with no tolerance either, so V1 fails even for a directory holding no files, while the scan here evaluates per file. Reworded to say that the new exposure is a directory contributing no rows, and that whenever a directory does contribute, the cast already fails above the scan or in the join key. -- 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]
