[SYSTEMDS-3949] Column API parquet decode for Delta frame reads#2535
Open
Jakob-al28 wants to merge 2 commits into
Open
[SYSTEMDS-3949] Column API parquet decode for Delta frame reads#2535Jakob-al28 wants to merge 2 commits into
Jakob-al28 wants to merge 2 commits into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 53.3s (1.89x). On the DeltaFrameRead benchmark from #2515: 1.68x serial, 1.50x 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). There's also a config kill switch (sysds.io.delta.reader.columnapi, default on).
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 fast path materializes one row group at a time, a bigger per-batch memory footprint than the default engine's fixed-size batches. 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.