Support Parquet DELTA encodings with more than 64 values per mini-block - #23314
Support Parquet DELTA encodings with more than 64 values per mini-block#23314pramodsatya wants to merge 15 commits into
Conversation
github-actions
Bot
added
libcudf
|
/ok to test aa8b101 |
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe PR changes GPU DELTA decoding to warp-sized pass processing, updates CUDA string and skip handling, adds compact-protocol page writers and DELTA-encoded Parquet fixtures, expands reader coverage, and registers NVBench benchmarks. ChangesParquet DELTA decoder
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp (1)
6-15: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd direct standard-library includes.
This file directly uses
std::string_view,std::string,std::vector, andstd::moveon Lines [23], [36], and [59], but does not include their defining headers. Avoid relying on transitive includes.Proposed include additions
`#include` "reader_common.hpp" +#include <string> +#include <string_view> +#include <utility> +#include <vector> + `#include` <benchmarks/common/generate_input.hpp>🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp` around lines 6 - 15, Add the direct standard-library headers defining the symbols used by this benchmark: string_view, string, vector, and move. Update the include section of parquet_reader_encoding.cpp without changing the benchmark implementation or relying on transitive includes.Source: Coding guidelines
cpp/src/io/parquet/delta_binary.cuh (1)
32-49: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConstants and the warp-divides-mini-block invariant are sound (
32 % warp_size == 0guaranteeswarp_sizedivides every spec-valid mini-block size), and thedelta_rolling_buf_sizederivation matches the documented "two batches in flight + header slot".Minor: the
static_asserton Line 36 has no diagnostic message. Adding one aids diagnosis if the warp-size assumption is ever violated on a new arch.As per coding guidelines: "use clear `static_assert` messages for template misuse".Proposed message
-static_assert(delta_mini_block_size_multiple % cudf::detail::warp_size == 0); +static_assert(delta_mini_block_size_multiple % cudf::detail::warp_size == 0, + "warp_size must divide the DELTA mini-block size multiple; the pass-based " + "decoders assume warp_size divides every spec-valid mini-block size");🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@cpp/src/io/parquet/delta_binary.cuh` around lines 32 - 49, Add a clear diagnostic message to the static_assert enforcing the delta_mini_block_size_multiple and cudf::detail::warp_size divisibility invariant. Keep the existing condition and surrounding constants unchanged.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@cpp/tests/io/parquet_delta_test_utils.hpp`:
- Around line 354-372: Update delta_test_strings to generate a mixture of ASCII
and valid non-ASCII UTF-8 sequences instead of using only the current
alphanumeric alphabet. Preserve the existing length, shared-prefix, and
deterministic random-generation behavior so DELTA_BYTE_ARRAY and related string
tests exercise byte-wise handling of multi-byte code points.
---
Nitpick comments:
In `@cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp`:
- Around line 6-15: Add the direct standard-library headers defining the symbols
used by this benchmark: string_view, string, vector, and move. Update the
include section of parquet_reader_encoding.cpp without changing the benchmark
implementation or relying on transitive includes.
In `@cpp/src/io/parquet/delta_binary.cuh`:
- Around line 32-49: Add a clear diagnostic message to the static_assert
enforcing the delta_mini_block_size_multiple and cudf::detail::warp_size
divisibility invariant. Keep the existing condition and surrounding constants
unchanged.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 25563803-0815-484a-9274-f4f50ab926e4
📒 Files selected for processing (7)
cpp/benchmarks/CMakeLists.txtcpp/benchmarks/io/parquet/parquet_reader_encoding.cppcpp/src/io/parquet/delta_binary.cuhcpp/src/io/parquet/page_delta_decode.cucpp/src/io/parquet/page_string_decode.cucpp/tests/io/parquet_delta_test_utils.hppcpp/tests/io/parquet_reader_test.cpp
mhaseeb123
added
feature request
|
|
||
| // index just past the values decode_next_pass() has produced so far (0 before the first pass, | ||
| // even though the header value already occupies index 0) | ||
| __device__ constexpr uint32_t next_pass_start_idx() |
There was a problem hiding this comment.
this is not usable in a constexpr context as the member variables are runtime. don't make this constexpr, it is misleading.
|
|
||
| void BM_parquet_read_delta_string(nvbench::state& state) | ||
| { | ||
| bench_read_encoding(state, {cudf::type_id::STRING}); |
There was a problem hiding this comment.
post benchmark results in the pr conversation page for before/after adding the decoding changes.
| // that is safe today because a 32-lane WarpScan over int64_t is shuffle-based and never | ||
| // touches its (empty) TempStorage, but a cub change or a wider scan type could turn this | ||
| // into a race. | ||
| __shared__ cub::WarpScan<int64_t>::TempStorage temp_storage; |
There was a problem hiding this comment.
thanks for finding this issue. let's just fix this now by declaring an array of size 2: temp_storage[2], plus a cudf_assert that we're not calling with 3+ warps.
There was a problem hiding this comment.
this array is now removed with use of cooperative_groups, is that alright?
| // the column values plus the DELTA block geometry (block_size, mini_block_count) and get back | ||
| // the complete file bytes. | ||
|
|
||
| // --------------------------------------------------------------------------------------------- |
There was a problem hiding this comment.
Can we use our compact_protocol_writer here instead of a reimpl?
There was a problem hiding this comment.
Yes this was AI generated, page headers are now serialized with compact_protocol_writer, thanks for pointing it out.
| // The DELTA_BINARY_PACKED spec requires the number of values in a mini-block to be a multiple of | ||
| // 32. That this equals the warp size is a coincidence the decoders below depend on: they produce | ||
| // values in warp_size-wide passes, so warp_size must divide every spec-valid mini-block size. |
There was a problem hiding this comment.
| // The DELTA_BINARY_PACKED spec requires the number of values in a mini-block to be a multiple of | |
| // 32. That this equals the warp size is a coincidence the decoders below depend on: they produce | |
| // values in warp_size-wide passes, so warp_size must divide every spec-valid mini-block size. | |
| // The DELTA_BINARY_PACKED spec requires the number of values in a mini-block to be a multiple of | |
| // 32. The decoders rely on the coincidence that this also equals warp size; they produce values in warp_size-wide passes, so it must divide every spec-valid mini-block size. |
| // The decoders produce values in warp_size-wide passes (see decode_next_pass), so mini-blocks of | ||
| // any size decode with a fixed-size rolling buffer. The decode loops produce up to two passes per | ||
| // iteration: pages whose mini-blocks hold at least two passes keep the two-pass batch the loops | ||
| // have always used, and running several passes back to back amortizes the per-iteration | ||
| // synchronization. |
There was a problem hiding this comment.
| // The decoders produce values in warp_size-wide passes (see decode_next_pass), so mini-blocks of | |
| // any size decode with a fixed-size rolling buffer. The decode loops produce up to two passes per | |
| // iteration: pages whose mini-blocks hold at least two passes keep the two-pass batch the loops | |
| // have always used, and running several passes back to back amortizes the per-iteration | |
| // synchronization. | |
| // The decode loops produce up to two (warp_size-wide) passes per iteration: pages whose mini-blocks | |
| // hold at least two passes keep the two-pass batch the loops have always used, and running several | |
| // passes back to back amortizes the per-iteration synchronization. |
| uint32_t cur_pass; // current warp_size-wide pass within the mini-block, used by | ||
| // decode_next_pass for pipelined single-pass decoding |
There was a problem hiding this comment.
| uint32_t cur_pass; // current warp_size-wide pass within the mini-block, used by | |
| // decode_next_pass for pipelined single-pass decoding | |
| uint32_t cur_pass; // current pass within the mini-block |
| // position at the end of this pass's values since the following calculates negative indexes | ||
| auto const d_start = cur_mb_start + (pass + 1) * (warp_size * mb_bits / 8); | ||
|
|
||
| // unpack deltas. modified from version in decode_dictionary_indices(), but | ||
| // that one only unpacks up to bitwidths of 24. simplified some since this | ||
| // will always do batches of 32. | ||
| // NOTE: because this needs to handle up to 64 bits, the branching used in the other | ||
| // implementation has been replaced with a loop. While this uses more registers, the | ||
| // looping version is just as fast and easier to read. | ||
| zigzag128_t delta = 0; | ||
| if (lane_id + current_value_idx < value_count) { | ||
| int32_t ofs = (lane_id - warp_size) * mb_bits; | ||
| uint8_t const* p = d_start + (ofs >> 3); | ||
| ofs &= 7; | ||
| if (p < block_end) { | ||
| uint32_t c = 8 - ofs; // 0 - 7 bits | ||
| delta = (*p++) >> ofs; | ||
|
|
||
| while (c < mb_bits && p < block_end) { | ||
| delta |= static_cast<zigzag128_t>(*p++) << c; | ||
| c += 8; | ||
| } | ||
| delta &= (static_cast<zigzag128_t>(1) << mb_bits) - 1; |
There was a problem hiding this comment.
Is it possible to get rid of some of these magic numbers if possible.
There was a problem hiding this comment.
bits_per_byte=8 reduces use of magic-numbers here.
| // do inclusive scan to get value - first_value at each position | ||
| // NOTE: this function-scope shared TempStorage is shared by all warps that call this method | ||
| // concurrently (e.g. the prefix and suffix decoder warps of the DELTA_BYTE_ARRAY kernels). | ||
| // that is safe today because a 32-lane WarpScan over int64_t is shuffle-based and never | ||
| // touches its (empty) TempStorage, but a cub change or a wider scan type could turn this | ||
| // into a race. | ||
| __shared__ cub::WarpScan<int64_t>::TempStorage temp_storage; | ||
| cub::WarpScan<int64_t>(temp_storage).InclusiveSum(delta, delta); |
There was a problem hiding this comment.
Optional: We should use (warp) cooperative_group in here for reductions and syncs. Easier and extensible in the future.
There was a problem hiding this comment.
Thanks for the suggestion, refactored accordingly.
| using cudf::detail::warp_size; | ||
| int const t = threadIdx.x; | ||
| int const lane_id = t % warp_size; | ||
| int const t = threadIdx.x; |
There was a problem hiding this comment.
Same nudge regarding use of cg here. Then we don't need to branch out for warp 0 as
if (t < warp_size)and can directly do:
// Branch by warp id directly
if (warp.meta_thread_rank() == 0)
|
Parquet DELTA reader decode benchmark comparison, obtained using NVBench on
DELTA states: mean −0.62%, best −4.80% (DELTA_BYTE_ARRAY), worst +2.19% (DELTA_LENGTH_BYTE_ARRAY, within the ~4% NVBench noise seen on the baseline run). |
|
@pmattione-nvidia , @mhaseeb123 , thanks for the feedback, updated the PR per your suggestions. Could you please take another look? |
| return c.value(); | ||
| } | ||
|
|
||
| size_t CompactProtocolWriter::write(DataPageHeader const& d) |
There was a problem hiding this comment.
Optional nit: If possible, can we replace single letter variable names in here. I know other functions here also use them so you can disregard this too if you don't feel like it. For example:
| size_t CompactProtocolWriter::write(DataPageHeader const& d) | |
| size_t CompactProtocolWriter::write(DataPageHeader const& pg_hdr) |
|
|
||
| auto const block = cg::this_thread_block(); | ||
| auto const warp = cg::tiled_partition<cudf::detail::warp_size>(block); | ||
| int const t = threadIdx.x; |
There was a problem hiding this comment.
| int const t = threadIdx.x; | |
| int const t = block.thread_rank(); |
| block.sync(); | ||
|
|
||
| // two warps will traverse the prefixes and suffixes and sum them up | ||
| auto const db = t < warp_size ? &prefixes : t < 2 * warp_size ? &suffixes : nullptr; |
There was a problem hiding this comment.
| auto const db = t < warp_size ? &prefixes : t < 2 * warp_size ? &suffixes : nullptr; | |
| auto const warp_id = warp.meta_group_rank(); | |
| auto const db = (warp_id == 0) ? &prefixes : warp_id == 1 ? &suffixes : nullptr; |
There was a problem hiding this comment.
Thanks for pointing this out.
| // block_size=128, mini_block_count=1 -> 128 values/mini-block: the reader previously rejected | ||
| // mini-blocks over 64 values with DELTA_PARAMS_UNSUPPORTED (0x100). | ||
| TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock128) | ||
| { |
There was a problem hiding this comment.
nit (style): Please put the comments inside the tests like
| // block_size=128, mini_block_count=1 -> 128 values/mini-block: the reader previously rejected | |
| // mini-blocks over 64 values with DELTA_PARAMS_UNSUPPORTED (0x100). | |
| TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock128) | |
| { | |
| TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock128) | |
| { | |
| // block_size=128, mini_block_count=1 -> 128 values/mini-block: the reader previously rejected | |
| // mini-blocks over 64 values with DELTA_PARAMS_UNSUPPORTED (0x100). |
| // block_size=384, mini_block_count=4 -> 96 values/mini-block: exercises multiple | ||
| // mini-blocks per block (both the within-block and next-block advance paths) and a | ||
| // non-power-of-two pass count (3). | ||
| TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock96) |
There was a problem hiding this comment.
Could we also add a malformed-geometry case (e.g. block_size=96, mini_block_count=2) expecting DELTA_PARAMS_UNSUPPORTED, and a case with null leaf values? The new values_per_mb % 32 validation and null handling at these mini-block sizes are otherwise uncovered.
| // implementation has been replaced with a loop. While this uses more registers, the | ||
| // looping version is just as fast and easier to read. | ||
| zigzag128_t delta = 0; | ||
| if (lane_id + current_value_idx < value_count) { |
There was a problem hiding this comment.
The guard omits pass, so it stops filtering anything past pass 0 (harmless today since p < block_end bounds the reads, but misleading now that a mini-block can hold 8 passes).
| if (lane_id + current_value_idx < value_count) { | |
| if (current_value_idx + pass * warp_size + lane_id < value_count) { |
| constexpr int delta_rolling_buf_size = (2 * max_delta_mini_block_size) + 1; | ||
| // Parquet serializes the bit-packed mini-block deltas as a stream of 8-bit bytes, so a bit count is | ||
| // converted to a byte count by dividing by this. | ||
| constexpr int bits_per_byte = 8; |
There was a problem hiding this comment.
Let's use the macro CHAR_BIT (directly where needed) instead of defining this in the header here which is pulled by every TU
| constexpr int bits_per_byte = 8; |
There was a problem hiding this comment.
Thanks for the suggestion, was uncertain if CHAR_BITS is the right invariant here so had gone with a new const to be safe, updated to CHAR_BITS.
| int block_size, | ||
| int mini_block_count) | ||
| { | ||
| assert(block_size % mini_block_count == 0 && (block_size / mini_block_count) % 32 == 0); |
There was a problem hiding this comment.
Please use:
| assert(block_size % mini_block_count == 0 && (block_size / mini_block_count) % 32 == 0); | |
| CUDF_EXPECTS(block_size % mini_block_count == 0 && (block_size / mini_block_count) % 32 == 0, "your one liner message here"); |
| uint32_t const batch_size = | ||
| is_skip_resume ? cudf::detail::warp_size | ||
| : min(db->values_per_mb, static_cast<uint32_t>(delta_max_batch_size)); |
There was a problem hiding this comment.
| uint32_t const batch_size = | |
| is_skip_resume ? cudf::detail::warp_size | |
| : min(db->values_per_mb, static_cast<uint32_t>(delta_max_batch_size)); | |
| // only nested pages resume the decoder mid-page; flat pages re-init it below and can keep the full batch | |
| bool const resumes_mid_page = is_skip_resume and has_repetition; | |
| uint32_t const batch_size = | |
| resumes_mid_page ? cudf::detail::warp_size | |
| : min(db->values_per_mb, static_cast<uint32_t>(delta_max_batch_size)); |
|
@pmattione-nvidia Looks like there are some merge conflicts. Could you please coordinate with @vyasr to get them resolved |
# Conflicts: # cpp/src/io/parquet/page_delta_decode.cu # cpp/src/io/parquet/page_string_decode.cu
|
/ok to test 6822b1b |
I took care of the conflicts. They were just the struct field changes from my PRs. |
4 participants
Description
The GPU Parquet reader rejects DELTA-encoded pages whose mini-blocks hold more than 64 values
with
DELTA_PARAMS_UNSUPPORTED, although the format allows any multiple of 32 and otherreaders accept such files. This PR removes the mini-block size limit from the
DELTA_BINARY_PACKED, DELTA_BYTE_ARRAY and DELTA_LENGTH_BYTE_ARRAY decoders.
The limit came from the decoder's rolling value buffer, which had to hold two whole
mini-blocks. Instead of growing it, decode and consume mini-blocks one warp-size pass at a
time, so every buffer is sized by the decode pipeline (values in flight per iteration) rather
than by any mini-block geometry:
(
delta_binary_decoder::decode_next_pass()); each page still producesmin(values_per_mb, 64)values per main-loop iteration, so previously-readable pages keeptheir exact iteration schedule. Shared memory for the value buffers is unchanged.
skip_values,skip_values_and_sum,delta_byte_array_decoder::skip) alsoadvance pass by pass and resume at a pass boundary, instead of requiring a whole mini-block
to stay resident; pages resuming after a skip produce one pass per iteration since larger
batches could overwrite the up-to-31 not-yet-consumed values the skip leaves behind.
silently mis-computes
str_bytesonce a mini-block no longer fits the buffer.init_binary_blocknow validatesvalues_per_mb % 32 == 0; malformed headers previouslydecoded garbage instead of erroring.
This also fixes two latent bugs reachable on nested pages with the old code:
nz_idxring (leaf-ordinal to output-row map) shares its size with the value buffers,but on nested pages the level decoder overshoots its target by up to a warp of values and
wraps onto entries the value consumer is reading when a page decodes 64 values per iteration
(
values_per_mb == 64, e.g. arrow-rs INT64 lists; confirmed with compute-sanitizerracecheck). The ring now has its own, larger size.
delta_byte_array_decoder::skipsaved the string needed for the next batch's frontcompression at a stale offset inside the scratch it was about to overwrite; the last decoded
string now lives in a reserved slot past the scratch area.
Since no stock writer emits more than 64 values per mini-block (cudf and parquet-mr write 32,
pyarrow and arrow-rs write 64 for INT64), tests build single-page files in memory from plain
value vectors with new test utilities (
parquet_delta_test_utils.hpp; the builders' output wascross-checked against pyarrow for every geometry the tests use): flat INT64, LIST, and
both string encodings flat and as LIST, at 64/96/128/256 values per mini-block, with
full,
num_rows-trimmed andskip_rowsreads. Also addsparquet_read_delta_binary/parquet_read_delta_stringbenchmarks (the DELTA decode kernelshad no reader benchmark; results are within ~1% of the previous code for DELTA_BINARY_PACKED
and DELTA_BYTE_ARRAY, ~2% for DELTA_LENGTH_BYTE_ARRAY).
Checklist