perf(parquet): decode def and rep level streams concurrently via 2D grid

[vyasr]

Description

preprocess_levels_kernel decodes the definition and repetition level streams for each Parquet page. Prior to this change the kernel dispatched one block per page, which decoded both streams sequentially — repetition first, then definition. Because these streams are independent, there is no reason to serialize them.

This PR changes the kernel to a 2D grid with 2 blocks per page: blockIdx.y selects which stream the block owns, so the two decoders run concurrently on separate SMs. Blocks whose stream is absent for a given page (non-list columns have no repetition stream; non-nullable columns have no definition stream) return immediately, so there is no wasted work.

Performance (A100 80GB, parquet_read_decode, 512 MiB, DEVICE_BUFFER, no compression)

data_type	cardinality	run_length	main → target	Δ
INTEGRAL	0	1	9.64 ms → 9.14 ms	−5.2%
INTEGRAL	1000	1	10.42 ms → 9.88 ms	−5.2%
INTEGRAL	0	32	8.27 ms → 7.86 ms	−5.0%
INTEGRAL	1000	32	8.21 ms → 7.82 ms	−4.7%
LIST	0	1	14.99 ms → 14.35 ms	−4.3%
STRUCT	1000	1	14.10 ms → 13.62 ms	−3.4%
STRUCT	1000	32	13.82 ms → 13.35 ms	−3.4%
LIST	1000	1	17.71 ms → 17.34 ms	−2.1%
STRING	1000	32	6.63 ms → 6.48 ms	−2.2%

13 of 16 configs improved; 0 regressions. parquet_read_wide_tables and parquet_read_long_strings/parquet_read_file_shape were within benchmark noise (non-nullable DECIMAL wide tables have no level streams).

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.

[coderabbitai]

📝 Walkthrough

Summary by CodeRabbit

• Performance
  • Improved Parquet page level preprocessing by decoding definition and repetition levels in parallel.
  • Reduced synchronization overhead during level decoding.
• Reliability
  • Added validation to prevent invalid staging ranges during RLE stream initialization.

Walkthrough

Parquet level preprocessing now launches separate blocks for repetition and definition levels. Each block selects and decodes one stream. RLE stream initialization now checks that its staging range is nonnegative.

Changes

Parquet level preprocessing

Layer / File(s)	Summary
Parallel level decoding cpp/src/io/parquet/decode_preprocess.cu	The kernel uses blockIdx.x for pages and blockIdx.y for level streams. Each block initializes one decoder and decodes only its assigned stream. The launch uses two blocks per page.
RLE staging range validation cpp/src/io/parquet/rle_stream.cuh	rle_stream::init adds a device-side assertion that the staging range is nonnegative.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Suggested reviewers: pmattione-nvidia, pointkernel, bdice

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and concisely describes the concurrent definition and repetition level decoding change.
Description check	✅ Passed	The description accurately explains the 2D-grid implementation, absent-stream handling, performance results, and validation status.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches 💡 1

🛠️ Fix failing CI checks 💡

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

cpp/src/io/parquet/rle_stream.cuh (1)

293-298: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win

Validate the range for non-staged streams.

Line 297 runs only when _smem_stage is not null. cpp/src/io/parquet/decode_fixed.cu:1158-1179 calls rle_stream::init without a staging buffer, so those streams bypass the new validation. Compute and validate the range before the staging branch.

Proposed fix

+    auto const len = cuda::std::distance(_start, _end);
+    cudf_assert(len >= 0 && "rle_stream::init: _end must be >= _start");
+
     if (_smem_stage != nullptr) {
       auto* const smem_stage =
         static_cast<uint8_t const*>(cuda::std::assume_aligned<16>(_smem_stage));
-      auto const len = static_cast<int>(cuda::std::distance(_start, _end));
-      cudf_assert(len >= 0 && "rle_stream::init: _end must be >= _start");
       if (len > 0 && len <= stage_capacity) {

🤖 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/rle_stream.cuh` around lines 293 - 298, Move the
_start-to-_end distance calculation and the len >= 0 assertion in
rle_stream::init before the _smem_stage != nullptr branch, so non-staged streams
from decode_fixed also validate the range. Reuse the validated len inside the
staging path without changing its existing capacity handling.

🤖 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.

Outside diff comments:
In `@cpp/src/io/parquet/rle_stream.cuh`:
- Around line 293-298: Move the _start-to-_end distance calculation and the len
>= 0 assertion in rle_stream::init before the _smem_stage != nullptr branch, so
non-staged streams from decode_fixed also validate the range. Reuse the
validated len inside the staging path without changing its existing capacity
handling.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Enterprise

Run ID: fd9800bd-8288-4994-a730-a023c6b21cf0

📥 Commits

Reviewing files that changed from the base of the PR and between 481e42a and 033720a.

📒 Files selected for processing (2)

• cpp/src/io/parquet/decode_preprocess.cu
• cpp/src/io/parquet/rle_stream.cuh

[bdice]

This might be a naive question. Why use a 2D grid rather than a 1D grid with block_rank() % 2 as the selector and block_rank() / 2 as the page index?

[vyasr]

This might be a naive question. Why use a 2D grid rather than a 1D grid with block_rank() % 2 as the selector and block_rank() / 2 as the page index?

No real reason, the two are structurally equivalent and the only differences are semantics.