[SYSTEMDS-3949] Column API parquet decode for Delta frame reads#2535
[SYSTEMDS-3949] Column API parquet decode for Delta frame reads#2535Jakob-al28 wants to merge 3 commits into
Conversation
|
Overall this looks good, but i would change some things.
Lets me know what you think? |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #2535 +/- ##
============================================
+ Coverage 71.59% 71.62% +0.02%
- Complexity 49684 49876 +192
============================================
Files 1596 1603 +7
Lines 192298 193230 +932
Branches 37675 37825 +150
============================================
+ Hits 137680 138402 +722
- Misses 43909 44058 +149
- Partials 10709 10770 +61 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
7c7de72 to
06c3a4b
Compare
|
I agree, the changes have been applied. The engine decorator and the config flag are gone, the column decode is now the path for a table's data files whenever it applies, it writes directly into the preallocated frame columns. The kernel is still used for the log/snapshot and for the cases where the data files alone aren't enough: deletion vectors and partitioned tables fall back to the kernel engine read. This simplification improved the speedup from 1.89x to 2.08x. Updated numbers are in the PR description. |
Follow-up to #2515, as discussed in #2528.
Delta Kernel's default engine decodes parquet through parquet-mr's row-record API (ParquetReader + ReadSupport/RecordMaterializer), which is built to reassemble nested records; each cell goes through a PrimitiveConverter callback. Frames are flat in SystemDS, so this adds a thin Engine decorator that overrides ParquetHandler.readParquetFiles: flat schemas decode through the column API (ColumnReadStoreImpl/ColumnReader). Everything else (predicates, nested/decimal/timestamp columns, unknown metadata columns) falls through to the wrapped default engine, and deletion vectors, partition values and column mapping are still applied by the kernel above the handler.
On TPC-H lineitem, 30M rows (median of 7 runs, cloud VM, n2-highmem-8, 40GB heap): 101.0s to 48.5s (2.08x). On the DeltaFrameRead benchmark from #2515: 1.73x serial, 1.51x parallel (k=8).
All runs at the same settings: ZSTD, no dictionary for Parquet and Delta; Binary Block uncompressed.
Two cases needed explicit handling: tables with deletion vectors enabled ask every read for a synthetic _metadata.row_index column that isn't actually in the file, and older files in a table that's had a column added later don't contain that column at all. Both would have failed otherwise; now the first is computed and the second comes back as null, covered by tests (DeltaFrameSparkContractTest).
Notes: the ParquetHandler contract has grown across kernel versions (the row_index requirement came with deletion vector support), so delta-kernel version bumps should re-run the contract tests. The decode relies on parquet-mr's column API (org.apache.parquet.column.impl); its signatures are unchanged from parquet 1.13 through 1.17.
The write path stays on the default engine. Two checks suggest there is little to gain there: the parquet writer rewrite in #2528, which removed the per-row Group materialization, only moved write time from 92s to 87s, and with matched codecs the Delta write is within about 8% of that plain parquet writer (2.7s Delta vs 2.5s parquet on 1M lineitem rows, both set to snappy for this check). Neither points to a large win available on the write side.