dongjoon-hyun commented on PR #58895: URL: https://github.com/apache/spark/pull/58895#issuecomment-5821627153
Here is a summary of my review comments (https://github.com/apache/spark/pull/58895#pullrequestreview-5309775040) on `c255f88`. I didn't find a wrong-result issue under the default settings. **Correctness** 1. The `parquet.filter.columnindex.enabled=false` escape hatch only covers phase 0. Phase 2 still selects pages and synchronizes rows through the offset index, so a file with a corrupt offset index can return non-key values from the wrong rows while the post-scan `Filter` passes them (key hash is correct). 2. `maxSplicedRowGroupBytes` is checked only when an accumulator fills `capacity` rows, and the post-phase-1 check ignores `splicedBytes`. Large variable-length keys can exceed the cap by a wide margin, and the conf doc ("holds no more than the plain read path does") is not accurate. 3. The key-slot/queue pairing in `publishSurvivorKeyVectors` silently depends on `ParquetStorageFilter.create` sorting the ordinals. 4. `FileFormat.supportsStorageFilter` receives canonicalized expressions because `ExpressionSet.filter` applies the predicate to `e.canonicalized` (undocumented contract for third-party formats). **Performance** 5. The row-range budget is multiplied by the number of phase-2 leaf columns (per-reader copies in `ParquetReadState`). Wide projections give the filter up after phase 1 has done all its work, repeatedly for every row group, which is slower than the feature off. Sharing `RowRanges.getRanges()` would fix it at the root. 6. A missing offset index is detected only via `MissingOffsetIndexException` in phase 2, although it is known from the footer (`getOffsetIndexReference() == null`). This costs wasted work and a stack-trace WARN per split, loses whole-row-group skips afterwards, and gives up the whole file when only a key column lacks an offset index. 7. The materialized bloom filter is duplicated into the task binary and deserialized twice per task (about 16MiB extra heap per concurrent task with the default max size). 8. Because `Cast` is not in the `canEvaluateUnconditionally` whitelist, even never-failing widening casts (e.g., `int` joined with `bigint`) are never pushed. **Design** 9. The correctness of the byte metrics and reads relies on the implicit ordering of Parquet's per-block `ColumnIndexStore` cache (`MISSING_INDEX_STORE.getOffsetIndex()` returns `null`, which would NPE inside Parquet if the ordering changes). 10. A Parquet-named conf and Parquet-specific metrics live in the format-agnostic `FileSourceStrategy` / `FileSourceScanLike`. **Tests** 11. The `>= 0` metric assertions can never fail because `SQLMetric` ignores negative values. 12. The `MissingOffsetIndexException` path is untested, although such a file can be written with the `ParquetFileWriter.writeDataPage` overloads without a row count. 13. No AQE-on test verifies that the storage filter is actually applied (only result equality, which the post-scan `Filter` guarantees anyway). **Cleanup** 14. Duplicated code (`closeSplicingState` vs `abandonSplicing`, phase-0 ranges in `recordFileSkipped`) and redundant give-up signaling (`filterGivenUp`, `spliceCurrentRowGroup`, `accumulate`, `null` return). 15. Stale references to the nonexistent `extractStorageFilters` and outdated test comments. -- 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]
