From cf796a2cc21736ddd852b0b5c373b72e49be3168 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Tue, 22 Sep 2026 22:57:20 +0000 Subject: [PATCH 01/13] feat(parquet): add the flat nz map producer and its DELTA_BINARY consumer Second of the nz map series. This one makes the feature do something, but only under `LIBCUDF_PARQUET_NZ_MAP=1`; unset, nothing below runs and no scratch is allocated. The nz map is `nz_idx[rank] = input position of the rank-th valid value`, plus its length `nz_count`. Naming it for the fields it holds keeps it clear of the reader's own `pass` / `subpass`, which an earlier "prepass" spelling collided with -- this file already has an unrelated string-size prepass. Producer -- `precompute_nz_map` (decode_preprocess.cu) walks a claimed page's already-decoded definition levels once, writing the leaf column's null mask and the map, then records `nz_count`. A required page needs no map -- its map is the identity -- and a null `nz_idx` is what tells the consumer to synthesize it. The block-wide rank is taken from the ballot the loop already computes, rather than from a `cub::BlockScan` over the same predicate, and the map pointer is hoisted out of the loop: `PageInfo*` is not restrict-qualified, so each iteration otherwise re-chased two dependent global loads to set up one 4-byte store. Shared memory drops from 2708 to 2048 bytes per block and the kernel no longer spills. At 40 registers and 128 threads it runs 12 blocks per SM, or 48 warps; the block scan was not what bounded that. Consumer -- `decode_delta_binary_kernel_with_nz_map` reads that map and places values without decoding levels, so it runs two warps rather than the level-decoding kernel's three: a delta producer and a value writer. The writer trails the producer by one batch, gated on `next_pass_start_idx()`. That lag is what lets the two warps run concurrently in one phase instead of two phases split by a barrier. The loop therefore exits only when neither side can advance: gating the writer means a malformed page can leave `nz_count` permanently above anything the delta stream can supply. `decode_delta_binary_kernel` is untouched, including its name. Routing happens through `non_nz_map_page_mask`, seeded on the host with the caller's page mask and then cleared by the producer for every page it claims. The two kernels therefore partition the pages because the producer's claim *is* the routing decision, not because a second kernel re-derives it -- had it been re-derived, the two could disagree and a page would be decoded twice or not at all. The buffer is shared, so the further producers in this series clear their own claims into it; a page has exactly one map kind, so the writes are disjoint. For that partition to hold the consumer must not silently skip a page for any *additional* reason, so the conditions the host already decides are documented rather than re-tested. The one that cannot be decided on the host -- a claimed page the producer wrote no count for -- is a `cudf_assert`, which catches it in a debug build and compiles out in release. It is unreachable today: the producer runs over a superset of the pages this kernel decodes. `classify_nz_map_kind` names the encodings a consumer has been written for, which is DELTA_BINARY_PACKED alone, and widens by one enumerator as each remaining consumer lands. Claiming a page is what moves it off the level-decoding kernel -- a claimed page carries a non-null `nz_map`, and that is what routes it -- so the classifier names only the encodings that have somewhere to go. A flat DELTA_BYTE_ARRAY page is equally decodable from a rank map and is claimed once its consumer lands. Stating that in the classifier rather than in a separate encoding mask keeps one rule instead of two that have to agree. The page mask is kept, rather than testing `has_nz_map()` inside `decode_delta_binary_kernel` directly, so that kernel stays byte-identical while the feature is opt-in -- turning the env var off is then a true rollback. Both forms would read the same host-set fields, so this is not about the two disagreeing. The in-code TODO records when to collapse it. The producer gets its own stream and its own `cuda::event` rather than riding `fork_streams`' shared one, so that only the kernel that reads its output waits on it. The intent is that in a mixed file the PLAIN and dictionary decoders overlap with it rather than queueing behind it; that overlap is not measured here, since every benchmark below sets a single kernel mask. Measured on `parquet_read_delta_binary_nullable_page_sizes` and `parquet_read_delta_binary` (H100, clocks locked, feature off -> on): 50% nulls -25% required columns -4.8% 1% nulls break-even The win scales with how much level-decoding work the map removes. At 50% nulls the bit-packed level walk it replaces is expensive; on required columns no map is allocated at all and the gain is the narrower block alone. At 1% nulls the levels are a cheap RLE run, so the map's own traffic -- 4 bytes per value written, 4 per valid value read, against levels at 1-2 bytes per value -- costs about what it saves. Narrowing the map's indices is the fix for that, and is tracked for before the default flips. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. This is the first PR in the series where the enabled arm decodes anything, so it is also the first real exercise of the nz map under hybrid scan. The test's env-var guard now restores any pre-existing value instead of unsetting it. Unsetting meant that running the suite with `LIBCUDF_PARQUET_NZ_MAP=1` -- which is how this feature is validated -- silently disabled it for every test after this one, the second of 129 in the file. Measured over a full run: 758 of 1723 subpasses saw the feature enabled before the fix, 1721 after. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 171 +++++++++++++++++++ cpp/src/io/parquet/page_delta_decode.cu | 171 +++++++++++++++++++ cpp/src/io/parquet/parquet_gpu.hpp | 67 ++++++-- cpp/src/io/parquet/reader_impl.cpp | 48 +++++- cpp/src/io/parquet/reader_impl.hpp | 11 +- cpp/src/io/parquet/reader_impl_chunking.cu | 8 +- cpp/src/io/parquet/reader_impl_chunking.hpp | 17 +- cpp/src/io/parquet/reader_impl_preprocess.cu | 82 +++++---- cpp/tests/io/parquet_reader_test.cpp | 78 +++++++++ 9 files changed, 569 insertions(+), 84 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 6d7d0a657f8c..732a5023581f 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -11,6 +11,7 @@ #include #include +#include #include @@ -18,6 +19,8 @@ #include #include #include +#include +#include namespace cudf::io::parquet::detail { @@ -29,6 +32,138 @@ namespace { constexpr int preprocess_block_size = 512; constexpr int level_decode_block_size = 128; +/** + * @brief Computes the nz map for one page. + * + * Walks the page's already-decoded definition levels once, writing + * - the output column's null mask for the rows this page contributes + * - `nz_idx[rank] = input position of the rank-th valid value` + * - `nz_count`. + * The matching decode kernel reads that map so it can skip walking levels. + * + * @param[in,out] pages All pages to be processed + * @param[in] chunks All chunks to be processed + * @param[in] page_mask Boolean vector indicating which pages need to be processed + * @param[in] min_row Minimum row index to read + * @param[in] num_rows Number of rows to read starting from min_row + */ +template +CUDF_KERNEL void __launch_bounds__(level_decode_block_size) + precompute_flat_nz_map_kernel(PageInfo* pages, + device_span chunks, + cudf::device_span page_mask, + cudf::device_span non_nz_map_page_mask, + size_t min_row, + size_t num_rows) +{ + static constexpr int num_warps = level_decode_block_size / cudf::detail::warp_size; + __shared__ __align__(16) full_page_decode_state state_g; + __shared__ int warp_valid_counts[num_warps]; + auto const block = cg::this_thread_block(); + int const page_idx = blockIdx.x; + int const t = block.thread_rank(); + PageInfo* const pp = &pages[page_idx]; + if (!pp->has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } + if (!page_mask.empty() && !page_mask[page_idx]) { return; } + + // This kernel must mark that it is producing nz map for this page so that + // downstream decode kernels can use the boolean buffer to determine whether or not to run. + if (t == 0 && !non_nz_map_page_mask.empty()) { non_nz_map_page_mask[page_idx] = false; } + + auto* const s = &state_g; + null_count_back_copier _{s, t}; + if (!setup_local_page_info( + s, pp, chunks, min_row, num_rows, all_types_filter{}, page_processing_stage::DECODE)) { + return; + } + + auto& ni = s->nesting.nesting_info[s->setup.col.max_nesting_depth - 1]; + // The per-page level buffers are slices of a shared allocation, and bounds preprocessing sizes + // each buffer to only the values inside the requested row range, so for a sliced read it holds + // fewer than the page contains; scanning past `num_decoded_level_values` reads the next page's + // levels. A required page has no level streams at all, so zero is its normal state and the whole + // page is in range. + int const decoded_value_limit = + pp->num_decoded_level_values > 0 + ? min(s->setup.page.num_input_values, pp->num_decoded_level_values) + : s->setup.page.num_input_values; + int const first_row = min(s->setup.first_row, decoded_value_limit); + int const value_limit = min(decoded_value_limit, first_row + s->setup.num_rows); + + // A required page's rank map is the identity, so there is nothing to materialize; the consumer + // synthesizes it. `nz_idx` stays null, which is what tells the consumer to do that. + if (!should_process_nulls(s)) { + if (t == 0) { pp->nz_map->nz_count = value_limit; } + block.sync(); + return; + } + + // The other way to reach `num_decoded_level_values == 0`: an optional page that contributes no + // rows to this read gets no level buffer, and there is nothing to map. + auto const* def = reinterpret_cast(pp->lvl_decode_buf[level_type::DEFINITION]); + if (def == nullptr) { + if (t == 0) { pp->nz_map->nz_count = 0; } + block.sync(); + return; + } + + auto* const nz_idx = pp->nz_map->nz_idx; + int const lane = t % cudf::detail::warp_size; + int const warp_id = t / cudf::detail::warp_size; + int valid_count = 0; + + // Hoisted for the same reason as `nz_idx` above: `store_validity` writes through a global + // `bitmask_type*`, which the compiler cannot prove does not alias the shared decode state, so it + // re-reads these from shared memory on every iteration otherwise. + auto* const valid_map = ni.valid_map; + auto const valid_map_offset = ni.valid_map_offset; + + for (int value_base = 0; value_base < value_limit; value_base += level_decode_block_size) { + int const value_pos = value_base + t; + int const is_valid = value_pos < value_limit && def[value_pos] > 0; + auto const valid_mask = ballot(is_valid); + + // Write the output column's null mask, which only requires the ballot. Queuing the write here + // allows us to submit the write operation before the synchronizations below. + { + bool const in_output = value_pos >= first_row && value_pos < value_limit; + auto const output_mask = ballot(in_output); + int const write_start = __ffs(output_mask) - 1; + if (write_start >= 0 && lane == 0 && valid_map != nullptr) { + int const write_end = cudf::detail::warp_size - __clz(output_mask); + store_validity(valid_map_offset + value_base + (t - lane) + write_start - first_row, + valid_map, + valid_mask >> write_start, + write_end - write_start); + } + } + + // The ballot already carries every rank in the warp, so the block-wide exclusive sum is a + // popcount plus a four-element scan, with no shuffle chain and no block scan temp storage. + if (lane == 0) { warp_valid_counts[warp_id] = __popc(valid_mask); } + block.sync(); + int warp_prefix = 0; + int block_valid_count = 0; + for (int w = 0; w < num_warps; ++w) { + if (w < warp_id) { warp_prefix += warp_valid_counts[w]; } + block_valid_count += warp_valid_counts[w]; + } + int const thread_valid_count = warp_prefix + __popc(valid_mask & ((1u << lane) - 1)); + + if (is_valid) { nz_idx[valid_count + thread_valid_count] = value_pos; } + valid_count += block_valid_count; + // This sync ensures that a warp reaching iteration i+1 does not overwrite its + // `warp_valid_counts` entry while another warp is still on iteration i, summing those entries + // above. + block.sync(); + } + + // The consumer recomputes the null count for the rows this page contributes, so the producer + // writes only the rank map and its size. + if (t == 0) { pp->nz_map->nz_count = valid_count; } + block.sync(); +} + using unused_state_buf = page_state_buffers_s<0, 0, 0>; /** @@ -526,4 +661,40 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, } } +void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, + cudf::detail::hostdevice_span chunks, + cudf::device_span page_mask, + cudf::device_span non_nz_map_page_mask, + size_t min_row, + size_t num_rows, + int level_type_size, + cuda::stream_ref stream) +{ + if (pages.size() == 0) { return; } + + // Seed the level-decoding kernels' page mask before claiming anything out of it. + // + // Note: When a second nz_map producer is added it must move out to the caller, or it would wipe + // the claims the first one already made. + if (not non_nz_map_page_mask.empty()) { + auto const policy = rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()); + if (page_mask.empty()) { + thrust::fill(policy, non_nz_map_page_mask.begin(), non_nz_map_page_mask.end(), true); + } else { + thrust::copy(policy, page_mask.begin(), page_mask.end(), non_nz_map_page_mask.begin()); + } + } + + dim3 const grid(pages.size(), 1); + dim3 const block(level_decode_block_size, 1); + if (level_type_size == 1) { + precompute_flat_nz_map_kernel<<>>( + pages.device_ptr(), chunks, page_mask, non_nz_map_page_mask, min_row, num_rows); + } else { + precompute_flat_nz_map_kernel<<>>( + pages.device_ptr(), chunks, page_mask, non_nz_map_page_mask, min_row, num_rows); + } + CUDF_CUDA_TRY(cudaGetLastError()); +} + } // namespace cudf::io::parquet::detail diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 65328dfecd7d..94255b823296 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -9,10 +9,12 @@ #include "page_string_utils.cuh" #include "parquet_gpu.hpp" +#include #include #include +#include #include namespace cudf::io::parquet::detail { @@ -30,6 +32,9 @@ constexpr int decode_delta_binary_block_size = 96; // warp_size; anything smaller lets the level decoder wrap onto entries the consumer is reading. constexpr int delta_nz_buf_size = 4 * delta_max_batch_size; +// One producer warp and one writer warp +constexpr int decode_delta_binary_with_nz_map_block_size = 64; // producer + writer + // DELTA_BYTE_ARRAY encoding (incremental encoding or front compression), is used for BYTE_ARRAY // columns. For each element in a sequence of strings, a prefix length from the preceding string // and a suffix is stored. The prefix lengths are DELTA_BINARY_PACKED encoded. The suffixes are @@ -517,6 +522,155 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } +// Decodes flat DELTA_BINARY_PACKED pages, like `decode_delta_binary_kernel`, but places values +// using the nz map's rank map instead of decoding definition levels itself. +// +// A separate kernel rather than a branch inside that one, so the level-decoding path stays +// untouched while the nz map is opt-in. +CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) + decode_delta_binary_kernel_with_nz_map(PageInfo* pages, + device_span chunks, + size_t min_row, + size_t num_rows, + cudf::device_span page_mask, + kernel_error::pointer error_code) +{ + __shared__ __align__(16) delta_binary_decoder db_state; + __shared__ __align__(16) full_page_decode_state state_g; + + auto* const s = &state_g; + int const page_idx = cg::this_grid().block_rank(); + auto const block = cg::this_thread_block(); + auto const warp = cg::tiled_partition(block); + auto* const db = &db_state; + + if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } + // The producer cleared this page from `decode_delta_binary_kernel`'s mask when it claimed it, + // so every page is decoded by one kernel and none by both. + if (!pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } + + [[maybe_unused]] null_count_back_copier _{s, static_cast(block.thread_rank())}; + if (!setup_local_page_info(s, + &pages[page_idx], + chunks, + min_row, + num_rows, + mask_filter{decode_kernel_mask::DELTA_BINARY}, + page_processing_stage::DECODE)) { + return; + } + + bool const process_nulls = should_process_nulls(s); + auto* const pp = &pages[page_idx]; + auto const* const nz_idx = pp->nz_map->nz_idx; + int const nz_count_in = pp->nz_map->nz_count; + + // This check failing indicates that we somehow failed to write out the null map for a page with + // nulls entering this kernel. + cudf_assert(nz_count_in >= 0 && (not process_nulls || nz_idx != nullptr)); + + auto& ni = s->nesting.nesting_info[s->setup.col.max_nesting_depth - 1]; + if (block.thread_rank() == 0) { + // A sliced page keeps the valid values preceding `first_row` in its map, so the null count for + // the rows this page contributes has to skip them. The map is ascending, so that prefix is the + // distance to `first_row`'s lower bound. Only the first page of a read is sliced, so skip the + // search when there is no prefix to find. + auto const prefix_valid_count = + (process_nulls && s->setup.first_row > 0) + ? static_cast(cuda::std::lower_bound(nz_idx, + nz_idx + nz_count_in, + static_cast(s->setup.first_row)) - + nz_idx) + : static_cast(s->setup.first_row); + ni.null_count = process_nulls ? s->setup.num_rows - (nz_count_in - prefix_valid_count) : 0; + s->progress.nz_count = nz_count_in; + } + block.sync(); + + if (block.thread_rank() == 0) { db->init_binary_block(s->stream.data_start, s->stream.data_end); } + block.sync(); + if (db->error) { + if (block.thread_rank() == 0) { + set_error(static_cast(decode_error::DELTA_PARAMS_UNSUPPORTED), + error_code); + } + return; + } + + uint32_t const skipped_leaf_values = s->setup.page.skipped_leaf_values; + bool const is_skip_resume = skipped_leaf_values > 0; + uint32_t const batch_size = + is_skip_resume ? cudf::detail::warp_size + : min(db->values_per_mb, static_cast(delta_max_batch_size)); + uint32_t const passes_per_batch = batch_size / cudf::detail::warp_size; + if (is_skip_resume) { db->skip_values(skipped_leaf_values, block, warp); } + + // This is a two-warp version of delta binary decoding using a delta producer warp and a value + // writer warp. The writer trails the producer by one batch to allow the two to run concurrently + // in a single phase. `next_pass_start_idx()` is the first stream index the producer has not + // decoded yet. Every index below it is decoded, and is still resident because + // `delta_rolling_buf_size` holds two batches, the one in flight plus the one trailing. Reading it + // only at the barrier keeps the concurrent region free of any access to the decoder's mutable + // scalars. `value_at()` touches `value[]` and `first_value`, and the producer writes neither + // below that index. + uint32_t produced = db->next_pass_start_idx(); + while (s->setup.error == 0 && s->progress.src_pos < s->progress.nz_count) { + uint32_t const src_pos = s->progress.src_pos; + // Decoder indices run ahead of output positions by `skipped_leaf_values` on a resume page. + // Capped at one batch so the writer trails the producer by exactly one, never more: the + // rolling buffer holds two batches, so a writer that fell two behind would read values the + // producer had already overwritten. + uint32_t const target_pos = + min(s->progress.nz_count, + min(produced > skipped_leaf_values ? produced - skipped_leaf_values : 0, + src_pos + batch_size)); + block.sync(); + + if (warp.meta_group_rank() == 0) { + for (uint32_t i = 0; i < passes_per_batch; ++i) { + if (i > 0) { warp.sync(); } + db->decode_next_pass(warp); + } + } else if (warp.meta_group_rank() == 1 && src_pos < target_pos) { + for (uint32_t sp = src_pos + warp.thread_rank(); sp < target_pos; sp += warp.size()) { + auto dst_pos = process_nulls ? static_cast(nz_idx[sp]) : static_cast(sp); + dst_pos -= s->setup.first_row; + if (dst_pos >= 0) { + void* const dst = ni.data_out + dst_pos * s->output_cvt.dtype_len; + auto const val = db->value_at(sp + skipped_leaf_values); + switch (s->output_cvt.dtype_len) { + case 1: *static_cast(dst) = val; break; + case 2: *static_cast(dst) = val; break; + case 4: *static_cast(dst) = val; break; + case 8: *static_cast(dst) = val; break; + } + } + } + if (warp.thread_rank() == 0) { s->progress.src_pos = target_pos; } + } + block.sync(); + + // Bounding the writer by the producer means a malformed page whose delta stream holds fewer + // values than its definition levels imply would leave `nz_count` permanently out of reach, and + // the loop would spin forever. To avoid that, we must stop once neither warp can advance. + // + // Both halves of that test are needed. The producer finishes first on a well-formed page, and + // we need to wait for the writer to drain the trailing batch -- breaking on the producer alone + // would drop it from every page. Conversely, the writer has not run at all on the first + // iteration, so breaking on the writer alone would emit nothing for any page whose values fit + // in one batch. + uint32_t const prev_produced = produced; + produced = db->next_pass_start_idx(); + if (produced == prev_produced && s->progress.src_pos <= src_pos) { break; } + } + + // The loop's barrier is inside its body, so a straggler may still be evaluating the loop + // condition -- which reads s->progress.nz_count -- when a later write lands on that field. + block.sync(); + + if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } +} + // Decode page data that is DELTA_BYTE_ARRAY packed. This encoding consists of a DELTA_BINARY_PACKED // array of prefix lengths, followed by a DELTA_BINARY_PACKED array of suffix lengths, followed by // the suffixes (technically the suffixes are DELTA_LENGTH_BYTE_ARRAY encoded). The latter two can @@ -980,6 +1134,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, size_t min_row, int level_type_size, cudf::device_span page_mask, + cudf::device_span non_nz_map_page_mask, kernel_error::pointer error_code, cuda::stream_ref stream) { @@ -987,6 +1142,22 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, dim3 dim_block(decode_delta_binary_block_size, 1); dim3 dim_grid(pages.size(), 1); // 1 threadblock per page + dim3 dim_block_nz_map(decode_delta_binary_with_nz_map_block_size, 1); + + // The caller sizes this mask only when the nz map producer has pages to claim, so a non-empty + // mask is what tells us the producer ran. Its contents are the complement of what the consumer + // below decodes: the producer cleared the entry for every page it claimed, leaving the rest for + // `decode_delta_binary_kernel`. + auto const use_flat_nz_map = not non_nz_map_page_mask.empty(); + if (use_flat_nz_map) { + // Not templated on `level_t`: this kernel never reads levels, which is the whole point of it. + decode_delta_binary_kernel_with_nz_map<<>>( + pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); + CUDF_CUDA_TRY(cudaGetLastError()); + + // Hand `decode_delta_binary_kernel` everything the consumer above did not claim. + page_mask = non_nz_map_page_mask; + } if (level_type_size == 1) { decode_delta_binary_kernel<<>>( diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index a1f52aef3e3f..89adfbacd1aa 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -333,29 +333,30 @@ struct PageNestingInfo { }; /** - * @brief Which level-prepass consumer a page uses, or NONE for the legacy decoders. + * @brief Which nz map consumer a page uses, or NONE to decode without precomputed validity + * data. * - * The level prepass walks a page's definition levels once up front and publishes a valid-rank + * The nz map producer walks a page's definition levels once up front and precomputes a valid-rank * map, so that the decode kernel can place values without decoding levels * itself. For now, only the DELTA encodings have a consumer for the map. See - * `classify_prepass_family` in reader_impl_preprocess.cu. + * `classify_nz_map_kind` in reader_impl_preprocess.cu. * * uint8_t to minimize the overhead in PageInfo */ -enum class level_prepass_family : uint8_t { +enum class nz_map_kind : uint8_t { NONE = 0, DELTA_FLAT = 1, }; /** - * @brief Level prepass scratch information + * @brief Per-page nz map scratch * - * Contains a valid-rank map computed from the rep and def levels for later decode kernels to use + * Contains a valid-rank map computed from the definition levels for later decode kernels to use * * Reached through a pointer on `PageInfo`, like `PageNestingInfo`, rather than copied into a * per-kernel shared-memory struct: the map is indexed by rank straight out of global memory. */ -struct page_prepass_state { +struct page_nz_map { // `nz_count` value meaning "claimed, but the producer has not run yet". static constexpr int32_t not_yet_produced = -1; @@ -364,8 +365,8 @@ struct page_prepass_state { uint32_t* nz_idx{}; // Negative until the producer runs; the page's valid count afterwards. int32_t nz_count{not_yet_produced}; - // Producer-written count whose meaning depends on the page's family, which is why it is not - // named for one of them. `DELTA_FLAT`, the only family here, uses it for the page's null count. + // A second producer-written count, for map kinds that need one. Unused by `DELTA_FLAT`, whose + // consumer derives the null count from `nz_idx` and `nz_count` itself. int32_t aux_count{}; }; @@ -447,21 +448,21 @@ struct PageInfo { bool is_compressed; // Whether the page is compressed (V2 header) bool has_value_info; // true if str_bytes, num_valids, etc are derivable from page indexes - // prepass_family indicates which prepass consumer a page uses. - // `prepass_state` is null when the selector did not claim this page -- non-null *is* the + // map_kind indicates which nz map consumer a page uses. + // `nz_map` is null when `classify_nz_map_kind` did not claim this page -- non-null *is* the // selection flag. - level_prepass_family prepass_family{level_prepass_family::NONE}; - page_prepass_state* prepass_state{}; + nz_map_kind map_kind{nz_map_kind::NONE}; + page_nz_map* nz_map{}; /** - * @brief True when this page was selected for @p family's prepass. + * @brief True when this page carries an nz map of kind @p kind. * - * @param family Prepass family to test against - * @return True if the page carries prepass scratch for @p family + * @param kind Map kind to test against + * @return True if the page carries nz map scratch of kind @p kind */ - [[nodiscard]] CUDF_HOST_DEVICE constexpr bool is_prepass_family(level_prepass_family family) const + [[nodiscard]] CUDF_HOST_DEVICE constexpr bool has_nz_map(nz_map_kind kind) const { - return prepass_state != nullptr && prepass_family == family; + return nz_map != nullptr && map_kind == kind; } }; @@ -1074,6 +1075,8 @@ void write_final_offsets(host_span offsets, * @param[in] min_row Minimum number of rows to read * @param[in] level_type_size Size in bytes of the type for level decoding * @param[in] page_mask Boolean vector indicating which pages need to be decoded + * @param[in] non_nz_map_page_mask Pages the nz map producer did not claim, empty when it + * claimed none * @param[out] error_code Error code for kernel failures * @param[in] stream CUDA stream to use */ @@ -1083,6 +1086,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, size_t min_row, int level_type_size, cudf::device_span page_mask, + cudf::device_span non_nz_map_page_mask, kernel_error::pointer error_code, cuda::stream_ref stream); @@ -1185,6 +1189,33 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, int level_type_size, cuda::stream_ref stream); +/** + * @brief Launches the kernel that computes the flat nz map + * + * Runs once per subpass, before the decode kernels, over the pages `classify_nz_map_kind` + * claimed. Writes + * `page_nz_map::nz_idx` / `nz_count` and the leaf column's null mask, so that the matching + * decode kernel can place values without walking definition levels itself. + * + * @param[in,out] pages All pages to be processed + * @param[in] chunks All chunks to be processed + * @param[in] page_mask Boolean vector indicating which pages need to be processed + * @param[out] non_nz_map_page_mask Page mask for the level-decoding kernels. Seeded here from + * @p page_mask (all true when that is empty), then cleared for every page claimed + * @param[in] min_row Minimum row index to read + * @param[in] num_rows Number of rows to read starting from min_row + * @param[in] level_type_size Size in bytes of the type for level decoding (1 or 2) + * @param[in] stream CUDA stream to use + */ +void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, + cudf::detail::hostdevice_span chunks, + cudf::device_span page_mask, + cudf::device_span non_nz_map_page_mask, + size_t min_row, + size_t num_rows, + int level_type_size, + cuda::stream_ref stream); + /** * @brief Fills output offset entries for pruned string and list pages * diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index cc8b4d686cb5..2c5793a4391c 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -26,11 +26,13 @@ #include #include +#include #include #include #include #include +#include #include #include @@ -199,9 +201,44 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ // create this before we fork streams kernel_error error_code(_stream); - // get the number of streams we need from the pool and tell them to wait on the H2D copies + auto const has_flat_nz_map = + _nz_map_enabled && + std::any_of(subpass.pages.host_begin(), subpass.pages.host_end(), [](PageInfo const& page) { + return page.has_nz_map(nz_map_kind::DELTA_FLAT); + }); + + // Partitions the pages into those with and without an nz map. Allocated before the fork, like + // `error_code` above: `fork_streams` records its event on `_stream` at fork time, so anything + // allocated afterwards is not ordered against its use on a forked stream under a stream-ordered + // memory resource. + // + // TODO: this mask is transitional. It exists only because the level-decoding kernels are + // deliberately left unmodified, so the page mask they already consult is the only way to tell + // them which pages another kernel has taken. Once every delta kernel has an nz map consumer and + // the feature is on by default, each can test `has_nz_map()` on the page directly and this + // buffer, its seeding and the producer's marking all go away. + rmm::device_uvector non_nz_map_page_mask(has_flat_nz_map ? subpass.pages.size() : 0, + _stream); + + // The producer gets its own stream and its own event rather than `fork_streams`' shared one, so + // that only the delta consumer waits on it and the other decoders overlap with it instead. int const nkernels = std::bitset<32>(kernel_mask).count(); - auto streams = cudf::detail::fork_streams(_stream, nkernels); + auto streams = cudf::detail::fork_streams(_stream, nkernels + (has_flat_nz_map ? 1 : 0)); + std::optional nz_map_done; + + if (has_flat_nz_map) { + auto const nz_map_stream = streams.back(); + auto const page_mask = subpass_page_mask_span(); + precompute_flat_nz_map(subpass.pages, + pass.chunks, + page_mask, + non_nz_map_page_mask, + skip_rows, + num_rows, + level_type_size, + nz_map_stream); + nz_map_done.emplace(nz_map_stream); + } int s_idx = 0; @@ -297,14 +334,19 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ // launch delta binary decoder if (BitAnd(kernel_mask, decode_kernel_mask::DELTA_BINARY) != 0) { + auto const delta_stream = streams[s_idx++]; + // The only kernel that reads the nz map. The others overlap with the producer rather than + // queueing behind it, which is why the producer does not ride `fork_streams`' shared event. + if (has_flat_nz_map) { delta_stream.wait(*nz_map_done); } decode_delta_binary(subpass.pages, pass.chunks, num_rows, skip_rows, level_type_size, subpass_page_mask_span(), + non_nz_map_page_mask, error_code.data(), - streams[s_idx++]); + delta_stream); } // launch byte stream split decoder diff --git a/cpp/src/io/parquet/reader_impl.hpp b/cpp/src/io/parquet/reader_impl.hpp index 6ffe72cc6213..0e1cd14e437f 100644 --- a/cpp/src/io/parquet/reader_impl.hpp +++ b/cpp/src/io/parquet/reader_impl.hpp @@ -180,7 +180,7 @@ class reader_impl { /** * @brief Preprocess string length and bounds information for the subpass. * - * At the end of this process, the `str_bytes` field of the the PageInfo struct + * At the end of this process, the `str_bytes` field of the PageInfo struct * will be populated, and if applicable, the delta_temp_buf in the subpass struct will * be allocated and the pages in the subpass will point into it properly. * @@ -490,7 +490,7 @@ class reader_impl { * @brief Calculate the number of rows read from each source in the output chunk * * @param chunk_start_row The offset of the first row in the output chunk - * @param chunk_num_rows The number of rows in the the output chunk + * @param chunk_num_rows The number of rows in the output chunk * @return Vector of number of rows from each respective data source in the output chunk */ [[nodiscard]] std::vector calculate_output_num_rows_per_source(size_t chunk_start_row, @@ -605,12 +605,11 @@ class reader_impl { // Parquet-dict → DICTIONARY32 transcode. std::vector _dict_transcode_eligible; - // LIBCUDF_PARQUET_LEVEL_PREPASS selector is fixed for the reader lifetime so - // every pass and output chunk agrees on which prepass consumer families may + // LIBCUDF_PARQUET_NZ_MAP selector is fixed for the reader lifetime so + // every pass and output chunk agrees on which nz map consumer kinds may // be selected. Read here rather than in a constructor body because // `hybrid_scan_reader_impl` reaches this class through the default constructor. - bool _level_prepass_enabled{ - cudf::detail::get_bool_env_or("LIBCUDF_PARQUET_LEVEL_PREPASS", false)}; + bool _nz_map_enabled{cudf::detail::get_bool_env_or("LIBCUDF_PARQUET_NZ_MAP", false)}; }; } // namespace cudf::io::parquet::detail diff --git a/cpp/src/io/parquet/reader_impl_chunking.cu b/cpp/src/io/parquet/reader_impl_chunking.cu index 8dbd10cb4b12..17e69657b7b4 100644 --- a/cpp/src/io/parquet/reader_impl_chunking.cu +++ b/cpp/src/io/parquet/reader_impl_chunking.cu @@ -379,12 +379,10 @@ void reader_impl::setup_next_subpass(read_mode mode) } subpass.pages.host_to_device_async(_stream); // The pages carry device pointers into this array, so it has to be uploaded with them. It is - // empty unless the level prepass claimed at least one page. Nothing may copy it back to the - // host and re-upload it later in the subpass: the prepass producer kernels write counts into + // empty unless the nz map producer claimed at least one page. Nothing may copy it back to the + // host and re-upload it later in the subpass: the nz map producer kernels write counts into // it that the decode kernels read in a subsequent launch. - if (!subpass.prepass_state_buf.empty()) { - subpass.prepass_state_buf.host_to_device_async(_stream); - } + if (!subpass.nz_map_buf.empty()) { subpass.nz_map_buf.host_to_device_async(_stream); } // preprocess pages (computes row counts for lists, computes output chunks and computes // the actual row counts we will be able load out of this subpass) diff --git a/cpp/src/io/parquet/reader_impl_chunking.hpp b/cpp/src/io/parquet/reader_impl_chunking.hpp index 773f94124971..31898035fc45 100644 --- a/cpp/src/io/parquet/reader_impl_chunking.hpp +++ b/cpp/src/io/parquet/reader_impl_chunking.hpp @@ -84,8 +84,8 @@ struct subpass_intermediate_data { subpass_intermediate_data(cuda::stream_ref stream) : decomp_page_data(stream, cudf::get_current_device_resource_ref()), level_decode_data(stream, cudf::get_current_device_resource_ref()), - flat_prepass_data(stream, cudf::get_current_device_resource_ref()), - prepass_state_buf(0, stream), + flat_nz_map_data(stream, cudf::get_current_device_resource_ref()), + nz_map_buf(0, stream), page_buf(0, stream), page_src_index{0, stream}, page_string_offset_indices(0, stream), @@ -99,12 +99,13 @@ struct subpass_intermediate_data { cuda::device_buffer decomp_page_data; cuda::device_buffer level_decode_data; - // Backing store for the flat level-prepass valid-rank maps. `allocate_level_decode_space` - // sub-allocates one slice of this per claimed page. Empty unless the prepass claimed a page. - cuda::device_buffer flat_prepass_data; - // Out-of-line prepass scratch, one entry per page. Empty unless the selector claimed at least - // one page of this subpass. See page_prepass_state. - cudf::detail::hostdevice_vector prepass_state_buf; + // Backing store for the flat nz maps. `allocate_level_decode_space` + // sub-allocates one slice of this per claimed page. Empty unless the nz map producer claimed + // a page. + cuda::device_buffer flat_nz_map_data; + // Out-of-line nz map scratch, one entry per page. Empty unless the selector claimed at least + // one page of this subpass. See page_nz_map. + cudf::detail::hostdevice_vector nz_map_buf; cudf::detail::hostdevice_span pages{}; cudf::detail::hostdevice_vector page_buf; diff --git a/cpp/src/io/parquet/reader_impl_preprocess.cu b/cpp/src/io/parquet/reader_impl_preprocess.cu index b264d6cae89f..39e511c1aa75 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess.cu @@ -43,26 +43,22 @@ namespace cudf::io::parquet::detail { namespace { /** - * @brief Decide which level-prepass consumer, if any, can decode @p page. + * @brief Decide which nz map consumer, if any, can decode @p page. * * @param page Page to classify * @param chunk Column chunk the page belongs to - * @return The page's prepass family, or `NONE` if no consumer can decode it + * @return The page's nz map kind, or `NONE` if no consumer can decode it */ -[[nodiscard]] level_prepass_family classify_prepass_family(PageInfo const& page, - ColumnChunkDesc const& chunk) +[[nodiscard]] nz_map_kind classify_nz_map_kind(PageInfo const& page, ColumnChunkDesc const& chunk) { - // For now, only the DELTA encodings have a prepass consumer. - bool const is_delta = BitAnd(page.kernel_mask, - BitOr(decode_kernel_mask::DELTA_BINARY, - decode_kernel_mask::DELTA_BYTE_ARRAY, - decode_kernel_mask::DELTA_LENGTH_BA)) != 0; - if (!is_delta) { return level_prepass_family::NONE; } + // DELTA_BINARY_PACKED is the only encoding with an nz map consumer today. + bool const has_consumer = BitAnd(page.kernel_mask, decode_kernel_mask::DELTA_BINARY) != 0; + if (!has_consumer) { return nz_map_kind::NONE; } // Only flat (depth-1, non-repeated) columns have a consumer today. Lists and structs of delta - // leaves stay on the legacy path. + // leaves keep decoding their own levels. bool const is_flat = chunk.max_level[level_type::REPETITION] == 0 && chunk.max_nesting_depth <= 1; - return is_flat ? level_prepass_family::DELTA_FLAT : level_prepass_family::NONE; + return is_flat ? nz_map_kind::DELTA_FLAT : nz_map_kind::NONE; } // Tests the passed in logical type for a FIXED_LENGTH_BYTE_ARRAY column to see if it should @@ -328,12 +324,12 @@ void reader_impl::allocate_level_decode_space() // Clear any selection left over from a previous subpass, on every page, including masked-out // ones. A subpass's page array is filled by copying whole `PageInfo` structs out of // `pass.pages`, so on the second and later subpasses of a pass these two fields can still hold - // what the previous subpass wrote, including a `prepass_state` pointer into scratch that + // what the previous subpass wrote, including a `nz_map` pointer into scratch that // subpass has since released. The reset has to precede the page-mask `continue` below, or a // masked-out page would keep that stale selection and carry the dangling pointer to the // device. - page.prepass_family = level_prepass_family::NONE; - page.prepass_state = nullptr; + page.map_kind = nz_map_kind::NONE; + page.nz_map = nullptr; // Skip pages that are masked out - no need to allocate level decode space for them if (!page_mask.is_empty() && !page_mask[idx]) { @@ -344,7 +340,7 @@ void reader_impl::allocate_level_decode_space() auto const& chunk = pass.chunks[page.chunk_idx]; // A masked-out page is never classified, so it is never selected and never allocated for. - if (_level_prepass_enabled) { page.prepass_family = classify_prepass_family(page, chunk); } + if (_nz_map_enabled) { page.map_kind = classify_nz_map_kind(page, chunk); } compute_page_level_decode_sizes(page, chunk, @@ -388,35 +384,33 @@ void reader_impl::allocate_level_decode_space() std::max(def_level_sizes[idx], rep_level_sizes[idx]) / pass.level_type_size; } - // Nothing below can do anything with the prepass disabled: `classify_prepass_family` is not + // Nothing below can do anything with the nz map producer disabled: `classify_nz_map_kind` is not // called in that case, so every page is `NONE`. Return before allocating or scanning anything. - if (!_level_prepass_enabled) { return; } + if (!_nz_map_enabled) { return; } - // TODO: count the prepass scratch in the chunked-read budget, before the prepass is enabled by - // default. + // TODO: count the nz map scratch in the chunked-read budget, before the nz map producer is + // enabled by default. // // For now, both `level_decode_data` above and the allocations below are not tracked in the - // chunked-read budget. We will fix that once we start actually leveraging the prepass and + // chunked-read budget. We will fix that once we start actually leveraging the nz map producer and // therefore introduce some of the machinery that will be needed to calculate the memory usage. // - // Hand out the out-of-line prepass scratch. There is exactly one for each claimed page, so most - // pages (those that do not use the prepass) have nothing here. This information is only needed on - // the host. On device, a non-null `PageInfo::prepass_state` always indicates that state exists. - std::vector prepass_slot(num_pages, std::numeric_limits::max()); + // Hand out the out-of-line nz map scratch. There is exactly one for each claimed page, so most + // pages (those that do not use the nz map producer) have nothing here. This information is only + // needed on the host. On device, a non-null `PageInfo::nz_map` always indicates that state + // exists. + std::vector nz_map_slot(num_pages, std::numeric_limits::max()); size_t num_claimed = 0; for (size_t idx = 0; idx < num_pages; ++idx) { - if (pages[idx].prepass_family != level_prepass_family::NONE) { - prepass_slot[idx] = num_claimed++; - } + if (pages[idx].map_kind != nz_map_kind::NONE) { nz_map_slot[idx] = num_claimed++; } } if (num_claimed == 0) { return; } - subpass.prepass_state_buf = - cudf::detail::hostdevice_vector(num_claimed, _stream); + subpass.nz_map_buf = cudf::detail::hostdevice_vector(num_claimed, _stream); - auto host_state = [&](size_t idx) -> page_prepass_state* { - return prepass_slot[idx] != std::numeric_limits::max() - ? subpass.prepass_state_buf.host_ptr(prepass_slot[idx]) + auto host_state = [&](size_t idx) -> page_nz_map* { + return nz_map_slot[idx] != std::numeric_limits::max() + ? subpass.nz_map_buf.host_ptr(nz_map_slot[idx]) : nullptr; }; @@ -427,28 +421,28 @@ void reader_impl::allocate_level_decode_space() // than recomputed so that the loop below, which hands out the slices, cannot disagree with this // one. std::vector flat_map_sizes(num_pages, 0); - size_t flat_prepass_size = 0; + size_t flat_nz_map_size = 0; for (size_t idx = 0; idx < num_pages; ++idx) { auto* const state = host_state(idx); if (state == nullptr) { continue; } - auto& page = pages[idx]; - page.prepass_state = subpass.prepass_state_buf.device_ptr(prepass_slot[idx]); - if (page.is_prepass_family(level_prepass_family::DELTA_FLAT)) { + auto& page = pages[idx]; + page.nz_map = subpass.nz_map_buf.device_ptr(nz_map_slot[idx]); + if (page.has_nz_map(nz_map_kind::DELTA_FLAT)) { auto const& chunk = pass.chunks[page.chunk_idx]; - state->nz_count = page_prepass_state::not_yet_produced; + state->nz_count = page_nz_map::not_yet_produced; if (chunk.max_level[level_type::DEFINITION] != 0) { flat_map_sizes[idx] = static_cast(page.num_input_values) * sizeof(uint32_t); - flat_prepass_size += flat_map_sizes[idx]; + flat_nz_map_size += flat_map_sizes[idx]; } } } - subpass.flat_prepass_data = cuda::device_buffer{ - _stream, cudf::get_current_device_resource_ref(), flat_prepass_size, cuda::no_init}; - auto* flat_prepass_ptr = reinterpret_cast(subpass.flat_prepass_data.data()); + subpass.flat_nz_map_data = cuda::device_buffer{ + _stream, cudf::get_current_device_resource_ref(), flat_nz_map_size, cuda::no_init}; + auto* flat_nz_map_ptr = reinterpret_cast(subpass.flat_nz_map_data.data()); for (size_t idx = 0; idx < num_pages; ++idx) { if (flat_map_sizes[idx] != 0) { - host_state(idx)->nz_idx = reinterpret_cast(flat_prepass_ptr); - flat_prepass_ptr += flat_map_sizes[idx]; + host_state(idx)->nz_idx = reinterpret_cast(flat_nz_map_ptr); + flat_nz_map_ptr += flat_map_sizes[idx]; } } } diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 449b8b5ab145..a8177ce0b268 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -88,6 +88,84 @@ TEST_F(ParquetReaderTest, ManyTinyStringPages) CUDF_TEST_EXPECT_TABLES_EQUAL(input, result.tbl->view()); } +// TODO: delete this test once the nz map is on by default. +// +// This test enables usage of the nz map producer and the corresponding delta binary kernels that +// consume the nz map producer to validate them. Currently those kernels are gated by an environment +// variable that will be removed once the full set of kernels has been merged and the end result +// benchmarked. Once the default flips, the rest of the suite covers those paths and this has +// nothing left to add. +TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) +{ + // The nz map is a second decode path selected by an env var: the same file must read + // identically with it on and off. + // + // DELTA_BINARY_PACKED because it is the only encoding claimed at this point in the series, and + // nullable so that the map itself is exercised. A required column is claimed too, but takes the + // consumer's identity-map path, where no map is allocated. + static constexpr char const* nz_map_env_var = "LIBCUDF_PARQUET_NZ_MAP"; + + constexpr int num_rows = 50000; + auto const values = cudf::detail::make_counting_transform_iterator( + 0, [](auto i) { return static_cast(i * 3 - 7); }); + auto const valids = + cudf::detail::make_counting_transform_iterator(0, [](auto i) { return (i % 7) != 0; }); + cudf::test::fixed_width_column_wrapper const col{values, values + num_rows, valids}; + auto const expected = table_view({col}); + + auto input_metadata = cudf::io::table_input_metadata{expected}; + input_metadata.column_metadata[0].set_encoding(cudf::io::column_encoding::DELTA_BINARY_PACKED); + + std::vector buffer; + cudf::io::write_parquet( + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, expected) + .write_v2_headers(true) + .metadata(input_metadata) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .build()); + + auto const read_back = [&] { + return cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder( + cudf::io::source_info{cudf::host_span{ + reinterpret_cast(buffer.data()), buffer.size()}}) + .build()); + }; + + // Unset is the shipping configuration while the feature is opt-in. + { + tmp_env_var const env{nz_map_env_var, "0"}; + auto const legacy = read_back(); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, legacy.tbl->view()); + } + { + tmp_env_var const env{nz_map_env_var, "1"}; + auto const with_nz_map = read_back(); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, with_nz_map.tbl->view()); + } + // A bounded read puts the same pages on the bounds-page path, where the two routes diverge most. + { + auto const trimmed = [&] { + return cudf::io::read_parquet( + cudf::io::parquet_reader_options::builder( + cudf::io::source_info{cudf::host_span{ + reinterpret_cast(buffer.data()), buffer.size()}}) + .skip_rows(1001) + .num_rows(4321) + .build()); + }; + auto const sliced = cudf::slice(expected, {1001, 1001 + 4321}); + { + tmp_env_var const env{nz_map_env_var, "0"}; + CUDF_TEST_EXPECT_TABLES_EQUAL(sliced, trimmed().tbl->view()); + } + { + tmp_env_var const env{nz_map_env_var, "1"}; + CUDF_TEST_EXPECT_TABLES_EQUAL(sliced, trimmed().tbl->view()); + } + } +} + TEST_F(ParquetReaderTest, UserBounds) { // trying to read more rows than there are should result in From ead9855b9c4de2980f061be2a1ff1590a906ccaf Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Thu, 8 Oct 2026 17:18:40 +0000 Subject: [PATCH 02/13] docs(parquet): say "has an nz map" rather than "claimed" "Claimed" was standing in for a plain fact the code already names: a page either has an nz map or it does not, and `has_nz_map()` is the predicate. The word also framed a selection rule as if pages were being contended for, when nothing competes for them -- `classify_nz_map_kind` names the encodings a consumer has been written for, and that is the whole decision. Comments and doxygen only; no behaviour change. The unrelated uses elsewhere in the reader are a different sense of the word and are left alone: a length that "claims more bytes than remain in the page", `reader_impl_dict_transcode` claiming ownership of a column, and the dictionary slot claims in chunk_dict.cu. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 4 ++-- cpp/src/io/parquet/page_delta_decode.cu | 8 ++++---- cpp/src/io/parquet/parquet_gpu.hpp | 13 ++++++------- cpp/src/io/parquet/reader_impl_chunking.cu | 2 +- cpp/src/io/parquet/reader_impl_chunking.hpp | 7 +++---- cpp/src/io/parquet/reader_impl_preprocess.cu | 4 ++-- cpp/tests/io/parquet_reader_test.cpp | 6 +++--- 7 files changed, 21 insertions(+), 23 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 732a5023581f..4342682c65a6 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -672,10 +672,10 @@ void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, { if (pages.size() == 0) { return; } - // Seed the level-decoding kernels' page mask before claiming anything out of it. + // Seed the level-decoding kernels' page mask before clearing anything out of it. // // Note: When a second nz_map producer is added it must move out to the caller, or it would wipe - // the claims the first one already made. + // the entries the first one already cleared. if (not non_nz_map_page_mask.empty()) { auto const policy = rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()); if (page_mask.empty()) { diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 94255b823296..0cc286995107 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -545,7 +545,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) auto* const db = &db_state; if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } - // The producer cleared this page from `decode_delta_binary_kernel`'s mask when it claimed it, + // The producer cleared this page from `decode_delta_binary_kernel`'s mask when it mapped it, // so every page is decoded by one kernel and none by both. if (!pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } @@ -1144,9 +1144,9 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, dim3 dim_grid(pages.size(), 1); // 1 threadblock per page dim3 dim_block_nz_map(decode_delta_binary_with_nz_map_block_size, 1); - // The caller sizes this mask only when the nz map producer has pages to claim, so a non-empty + // The caller sizes this mask only when the nz map producer has pages to map, so a non-empty // mask is what tells us the producer ran. Its contents are the complement of what the consumer - // below decodes: the producer cleared the entry for every page it claimed, leaving the rest for + // below decodes: the producer cleared the entry for every page it mapped, leaving the rest for // `decode_delta_binary_kernel`. auto const use_flat_nz_map = not non_nz_map_page_mask.empty(); if (use_flat_nz_map) { @@ -1155,7 +1155,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); CUDF_CUDA_TRY(cudaGetLastError()); - // Hand `decode_delta_binary_kernel` everything the consumer above did not claim. + // Hand `decode_delta_binary_kernel` every page the consumer above does not decode. page_mask = non_nz_map_page_mask; } diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index 89adfbacd1aa..9078ad61abae 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -357,7 +357,7 @@ enum class nz_map_kind : uint8_t { * per-kernel shared-memory struct: the map is indexed by rank straight out of global memory. */ struct page_nz_map { - // `nz_count` value meaning "claimed, but the producer has not run yet". + // `nz_count` value meaning "this page has a map, but the producer has not run yet". static constexpr int32_t not_yet_produced = -1; // Valid-rank map: `nz_idx[rank]` is the input position of the rank-th valid value. Null for a @@ -449,8 +449,8 @@ struct PageInfo { bool has_value_info; // true if str_bytes, num_valids, etc are derivable from page indexes // map_kind indicates which nz map consumer a page uses. - // `nz_map` is null when `classify_nz_map_kind` did not claim this page -- non-null *is* the - // selection flag. + // `nz_map` is null when `classify_nz_map_kind` selected no consumer for this page -- + // non-null *is* the selection flag. nz_map_kind map_kind{nz_map_kind::NONE}; page_nz_map* nz_map{}; @@ -1075,8 +1075,7 @@ void write_final_offsets(host_span offsets, * @param[in] min_row Minimum number of rows to read * @param[in] level_type_size Size in bytes of the type for level decoding * @param[in] page_mask Boolean vector indicating which pages need to be decoded - * @param[in] non_nz_map_page_mask Pages the nz map producer did not claim, empty when it - * claimed none + * @param[in] non_nz_map_page_mask Pages with no nz map, empty when no page has one * @param[out] error_code Error code for kernel failures * @param[in] stream CUDA stream to use */ @@ -1193,7 +1192,7 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, * @brief Launches the kernel that computes the flat nz map * * Runs once per subpass, before the decode kernels, over the pages `classify_nz_map_kind` - * claimed. Writes + * selected a consumer for. Writes * `page_nz_map::nz_idx` / `nz_count` and the leaf column's null mask, so that the matching * decode kernel can place values without walking definition levels itself. * @@ -1201,7 +1200,7 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, * @param[in] chunks All chunks to be processed * @param[in] page_mask Boolean vector indicating which pages need to be processed * @param[out] non_nz_map_page_mask Page mask for the level-decoding kernels. Seeded here from - * @p page_mask (all true when that is empty), then cleared for every page claimed + * @p page_mask (all true when that is empty), then cleared for every page mapped * @param[in] min_row Minimum row index to read * @param[in] num_rows Number of rows to read starting from min_row * @param[in] level_type_size Size in bytes of the type for level decoding (1 or 2) diff --git a/cpp/src/io/parquet/reader_impl_chunking.cu b/cpp/src/io/parquet/reader_impl_chunking.cu index 17e69657b7b4..52bc6855dd88 100644 --- a/cpp/src/io/parquet/reader_impl_chunking.cu +++ b/cpp/src/io/parquet/reader_impl_chunking.cu @@ -379,7 +379,7 @@ void reader_impl::setup_next_subpass(read_mode mode) } subpass.pages.host_to_device_async(_stream); // The pages carry device pointers into this array, so it has to be uploaded with them. It is - // empty unless the nz map producer claimed at least one page. Nothing may copy it back to the + // empty unless some page has an nz map. Nothing may copy it back to the // host and re-upload it later in the subpass: the nz map producer kernels write counts into // it that the decode kernels read in a subsequent launch. if (!subpass.nz_map_buf.empty()) { subpass.nz_map_buf.host_to_device_async(_stream); } diff --git a/cpp/src/io/parquet/reader_impl_chunking.hpp b/cpp/src/io/parquet/reader_impl_chunking.hpp index 31898035fc45..1c0cf29e8c5f 100644 --- a/cpp/src/io/parquet/reader_impl_chunking.hpp +++ b/cpp/src/io/parquet/reader_impl_chunking.hpp @@ -100,11 +100,10 @@ struct subpass_intermediate_data { cuda::device_buffer level_decode_data; // Backing store for the flat nz maps. `allocate_level_decode_space` - // sub-allocates one slice of this per claimed page. Empty unless the nz map producer claimed - // a page. + // sub-allocates one slice of this per mapped page. Empty unless some page has an nz map. cuda::device_buffer flat_nz_map_data; - // Out-of-line nz map scratch, one entry per page. Empty unless the selector claimed at least - // one page of this subpass. See page_nz_map. + // Out-of-line nz map scratch, one entry per page. Empty unless some page of this subpass has + // an nz map. See page_nz_map. cudf::detail::hostdevice_vector nz_map_buf; cudf::detail::hostdevice_span pages{}; diff --git a/cpp/src/io/parquet/reader_impl_preprocess.cu b/cpp/src/io/parquet/reader_impl_preprocess.cu index 39e511c1aa75..dcfe21b50b43 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess.cu @@ -395,7 +395,7 @@ void reader_impl::allocate_level_decode_space() // chunked-read budget. We will fix that once we start actually leveraging the nz map producer and // therefore introduce some of the machinery that will be needed to calculate the memory usage. // - // Hand out the out-of-line nz map scratch. There is exactly one for each claimed page, so most + // Hand out the out-of-line nz map scratch. There is exactly one for each mapped page, so most // pages (those that do not use the nz map producer) have nothing here. This information is only // needed on the host. On device, a non-null `PageInfo::nz_map` always indicates that state // exists. @@ -414,7 +414,7 @@ void reader_impl::allocate_level_decode_space() : nullptr; }; - // Point each claimed page at its scratch and size the flat valid-rank maps in a single walk. + // Point each mapped page at its scratch and size the flat valid-rank maps in a single walk. // `PageInfo` travels to the device, so the page carries the device address while the seeding // here goes through `host_state()`. A required page needs no map at all since its rank map is // the identity, which the consumer synthesizes rather than reading. The sizes are stored rather diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index a8177ce0b268..7de2e7b8f2d2 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -100,9 +100,9 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) // The nz map is a second decode path selected by an env var: the same file must read // identically with it on and off. // - // DELTA_BINARY_PACKED because it is the only encoding claimed at this point in the series, and - // nullable so that the map itself is exercised. A required column is claimed too, but takes the - // consumer's identity-map path, where no map is allocated. + // DELTA_BINARY_PACKED because it is the only encoding with a consumer at this point in the + // series, and nullable so that the map itself is exercised. A required column gets one too, but + // takes the consumer's identity-map path, where no map is allocated. static constexpr char const* nz_map_env_var = "LIBCUDF_PARQUET_NZ_MAP"; constexpr int num_rows = 50000; From 135c68bc7bee84d34519760543dffd2f3d480d40 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Thu, 8 Oct 2026 20:06:09 +0000 Subject: [PATCH 03/13] fix(parquet): report a truncated delta stream in the nz map consumer `decode_delta_binary_kernel_with_nz_map` stops once neither of its two warps can advance, which is what keeps a malformed page from spinning forever. But it then returned without an error, and the producer has already written the validity bits for the rows it had counted -- so a page whose delta stream holds fewer values than its definition levels call for read back as valid rows over output slots nothing ever wrote. The stopping condition already knows this happened, so reporting it is one comparison on thread 0 after the loop. It cannot fire on a well-formed page: the only other way out of the loop is the `while` condition, which requires `src_pos >= nz_count`. This does not make a malformed page decode correctly, which is not something the reader can do. It moves the failure from silently wrong data to a reported error. `decode_delta_binary_kernel` has the same gap by a different route -- its writer warp advances `src_pos` by a full batch whether or not the producer supplied those values, so it runs to completion and writes stale rolling-buffer entries. That is pre-existing and is left alone here, so the level-decoding path stays byte-identical while the nz map is opt-in. Worth closing separately. Raised by CodeRabbit on #24501. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/page_delta_decode.cu | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index a2a6ae5b7044..d166d4616da5 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -667,6 +667,13 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) // condition -- which reads s->progress.nz_count -- when a later write lands on that field. block.sync(); + // Values still outstanding means the loop stopped above rather than finishing: the delta stream + // held fewer values than the definition levels call for. Those rows are already marked valid, so + // without an error the read would return whatever their output slots happened to contain. + if (block.thread_rank() == 0 and s->progress.src_pos < s->progress.nz_count) { + s->setup.error |= static_cast(decode_error::DATA_STREAM_OVERRUN); + } + if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } From bf4126c7cc408f696bf851374c5fde720d6f288e Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sat, 10 Oct 2026 22:43:54 +0000 Subject: [PATCH 04/13] refactor(parquet): address review on the nz map producer and its test Four points from review on #24501, with the intrinsic modernization held back so it can be measured on its own. Drop `null_count_back_copier` from `precompute_flat_nz_map_kernel`. The producer never writes a null count -- the only mention of one in the file was the copier's own declaration -- so it copied back the value `setup_local_page_info` initialized, which the consumer then overwrote. `precompute_flat_nz_map` takes `cudf::memory_resources` and seeds the page mask from `mr.get_temporary_mr()`. Behaviour is unchanged: the single-argument constructor captures the current device resource for temporaries, which is what the explicit `get_current_device_resource_ref()` was passing. What it buys is that the temporary resource is now the caller's to choose. `NzMapMatchesLevelDecoder` gains a required column. On that path the producer writes no `nz_idx` and the consumer synthesizes the identity map from the rank alone -- the test's comment claimed to cover it while the table held one nullable column, so it never ran. The same fixture now pins its page geometry through `max_page_fragment_size` and `max_page_size_rows` rather than inheriting the writer's defaults, so what it exercises does not move when those change. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 6 +++--- cpp/src/io/parquet/parquet_gpu.hpp | 3 ++- cpp/src/io/parquet/reader_impl.cpp | 3 ++- cpp/tests/io/parquet_reader_test.cpp | 14 +++++++++++--- 4 files changed, 18 insertions(+), 8 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 4342682c65a6..4a4ea96cf9be 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -71,7 +71,6 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) if (t == 0 && !non_nz_map_page_mask.empty()) { non_nz_map_page_mask[page_idx] = false; } auto* const s = &state_g; - null_count_back_copier _{s, t}; if (!setup_local_page_info( s, pp, chunks, min_row, num_rows, all_types_filter{}, page_processing_stage::DECODE)) { return; @@ -668,7 +667,8 @@ void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, size_t min_row, size_t num_rows, int level_type_size, - cuda::stream_ref stream) + cuda::stream_ref stream, + cudf::memory_resources mr) { if (pages.size() == 0) { return; } @@ -677,7 +677,7 @@ void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, // Note: When a second nz_map producer is added it must move out to the caller, or it would wipe // the entries the first one already cleared. if (not non_nz_map_page_mask.empty()) { - auto const policy = rmm::exec_policy_nosync(stream, cudf::get_current_device_resource_ref()); + auto const policy = rmm::exec_policy_nosync(stream, mr.get_temporary_mr()); if (page_mask.empty()) { thrust::fill(policy, non_nz_map_page_mask.begin(), non_nz_map_page_mask.end(), true); } else { diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index 9078ad61abae..904bc2f66387 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -1213,7 +1213,8 @@ void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, size_t min_row, size_t num_rows, int level_type_size, - cuda::stream_ref stream); + cuda::stream_ref stream, + cudf::memory_resources mr); /** * @brief Fills output offset entries for pruned string and list pages diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index 968bbc5063c8..5f2a54350e5d 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -236,7 +236,8 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ skip_rows, num_rows, level_type_size, - nz_map_stream); + nz_map_stream, + _mr); nz_map_done.emplace(nz_map_stream); } diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 7de2e7b8f2d2..6c5d5d322dd5 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -110,11 +110,17 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) 0, [](auto i) { return static_cast(i * 3 - 7); }); auto const valids = cudf::detail::make_counting_transform_iterator(0, [](auto i) { return (i % 7) != 0; }); - cudf::test::fixed_width_column_wrapper const col{values, values + num_rows, valids}; - auto const expected = table_view({col}); + cudf::test::fixed_width_column_wrapper const nullable_col{ + values, values + num_rows, valids}; + // A required column takes the consumer's identity-map path: the producer writes no `nz_idx` and + // the consumer derives the position from the rank alone. + cudf::test::fixed_width_column_wrapper const required_col(values, values + num_rows); + auto const expected = table_view({nullable_col, required_col}); auto input_metadata = cudf::io::table_input_metadata{expected}; - input_metadata.column_metadata[0].set_encoding(cudf::io::column_encoding::DELTA_BINARY_PACKED); + for (auto& col_meta : input_metadata.column_metadata) { + col_meta.set_encoding(cudf::io::column_encoding::DELTA_BINARY_PACKED); + } std::vector buffer; cudf::io::write_parquet( @@ -122,6 +128,8 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) .write_v2_headers(true) .metadata(input_metadata) .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .max_page_fragment_size(1000) + .max_page_size_rows(1000) .build()); auto const read_back = [&] { From e9079c165fc44ff5b41af74a4d145a93e6c8ab22 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sat, 10 Oct 2026 23:46:59 +0000 Subject: [PATCH 05/13] refactor(parquet): use cuda::std bit ops in the nz map producer Replaces `__ffs`, `__clz` and `__popc` in `precompute_flat_nz_map_kernel` with `cuda::std::countr_zero`, `bit_width` and `popcount`, per review on #24501. The same file already called `cuda::std::countr_zero` for the list-lane scan, so it was inconsistent with itself. The guard had to change with it. `__ffs(mask) - 1` yields -1 on an empty mask, which `write_start >= 0` relied on; `countr_zero` yields `warp_size` instead, so the test is now `write_start < warp_size`. Keeping the old comparison would have sent every warp with no lane in range into the store with `write_start == 32`. `warp_size - __clz(mask)` is exactly `bit_width(mask)`. Separate from the review commit before it because it changes generated code and so needed its own measurement. The SASS is not identical -- normalized over both `level_t` instantiations it goes from 3878 to 3829 instructions, with `FLO.U32.SH` becoming `FLO.U32` plus a negate -- but that does not reach the clock: interleaved A/B, 3 rounds, H100 GPU 2 locked at 1755 MHz, parquet_read_delta_binary_nullable_page_sizes with the nz map enabled nullable_1, page_rows 256 1.003 nullable_1, page_rows 32 1.005 nullable_50, page_rows 256 1.007 nullable_50, page_rows 32 1.007 geomean 1.005 Within-arm spread is 0.3-3.1%, so a 0.5% geomean is a wash rather than a regression -- and the 49 instructions it removes buy nothing either, which says the producer's loop is not issue-bound. Measured on the nullable benchmark because `parquet_read_delta_binary` builds required columns, where the producer returns before reaching the loop these intrinsics are in. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 4a4ea96cf9be..9516ebee148b 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -127,9 +128,11 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) { bool const in_output = value_pos >= first_row && value_pos < value_limit; auto const output_mask = ballot(in_output); - int const write_start = __ffs(output_mask) - 1; - if (write_start >= 0 && lane == 0 && valid_map != nullptr) { - int const write_end = cudf::detail::warp_size - __clz(output_mask); + // `countr_zero` is `warp_size` on an empty mask, which is what rejects a warp with no lane + // in range. + int const write_start = cuda::std::countr_zero(output_mask); + if (write_start < cudf::detail::warp_size && lane == 0 && valid_map != nullptr) { + int const write_end = cuda::std::bit_width(output_mask); store_validity(valid_map_offset + value_base + (t - lane) + write_start - first_row, valid_map, valid_mask >> write_start, @@ -139,7 +142,7 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) // The ballot already carries every rank in the warp, so the block-wide exclusive sum is a // popcount plus a four-element scan, with no shuffle chain and no block scan temp storage. - if (lane == 0) { warp_valid_counts[warp_id] = __popc(valid_mask); } + if (lane == 0) { warp_valid_counts[warp_id] = cuda::std::popcount(valid_mask); } block.sync(); int warp_prefix = 0; int block_valid_count = 0; From 8818749ca07d09e6bede9c01881baf30855b8187 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sun, 11 Oct 2026 00:59:26 +0000 Subject: [PATCH 06/13] refactor(parquet): route delta pages on has_nz_map instead of a second mask Both delta kernels now test the page's own nz map, so `non_nz_map_page_mask` and everything that fed it are gone: the per-subpass `device_uvector`, the thrust fill/copy that seeded it from the caller's page mask, the producer's store clearing each entry, the span threaded through two signatures, and the page-mask swap in the launcher. What replaces all of it is one line in `decode_delta_binary_kernel`: if (pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } The consumer already routed this way. The mask existed only so the level-decoding kernel could be left untouched, and the page mask it already consults was the one channel available for telling it which pages another kernel had taken. That made `page_mask` carry two unrelated meanings -- pruned by hybrid scan, and claimed by the consumer -- which is the interaction review called confusing. It means pruned again, and the map alone decides routing. The launcher keeps a `bool has_flat_nz_map` where the span was, since it still has to decide whether the consumer has any work; both kernels now receive the caller's page mask unmodified. `decode_delta_binary_kernel` is therefore no longer byte-identical, which the previous commit in this branch leaned on. It gains one branch, and the rollback it was protecting is unaffected: with the feature off no page carries a map, so `has_nz_map()` is false everywhere and the branch is never taken. The mask seeding was the only allocation in `precompute_flat_nz_map`, so the `cudf::memory_resources` parameter added for it earlier in this branch has nothing left to serve and is removed, along with the thrust and rmm includes it needed. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. An nsys trace over a LIST delta page and a flat one confirms the partition still holds with no mask: the level-decoding kernel runs for the pages the classifier declines, the consumer for the pages it selects. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 34 ++++--------------------- cpp/src/io/parquet/page_delta_decode.cu | 17 +++++-------- cpp/src/io/parquet/parquet_gpu.hpp | 10 +++----- cpp/src/io/parquet/reader_impl.cpp | 26 +++---------------- 4 files changed, 18 insertions(+), 69 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 9516ebee148b..d8addce6ff90 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -13,15 +13,11 @@ #include #include -#include - #include #include #include #include #include -#include -#include namespace cudf::io::parquet::detail { @@ -53,7 +49,6 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) precompute_flat_nz_map_kernel(PageInfo* pages, device_span chunks, cudf::device_span page_mask, - cudf::device_span non_nz_map_page_mask, size_t min_row, size_t num_rows) { @@ -67,10 +62,6 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) if (!pp->has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } if (!page_mask.empty() && !page_mask[page_idx]) { return; } - // This kernel must mark that it is producing nz map for this page so that - // downstream decode kernels can use the boolean buffer to determine whether or not to run. - if (t == 0 && !non_nz_map_page_mask.empty()) { non_nz_map_page_mask[page_idx] = false; } - auto* const s = &state_g; if (!setup_local_page_info( s, pp, chunks, min_row, num_rows, all_types_filter{}, page_processing_stage::DECODE)) { @@ -666,36 +657,21 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, cudf::detail::hostdevice_span chunks, cudf::device_span page_mask, - cudf::device_span non_nz_map_page_mask, size_t min_row, size_t num_rows, int level_type_size, - cuda::stream_ref stream, - cudf::memory_resources mr) + cuda::stream_ref stream) { if (pages.size() == 0) { return; } - // Seed the level-decoding kernels' page mask before clearing anything out of it. - // - // Note: When a second nz_map producer is added it must move out to the caller, or it would wipe - // the entries the first one already cleared. - if (not non_nz_map_page_mask.empty()) { - auto const policy = rmm::exec_policy_nosync(stream, mr.get_temporary_mr()); - if (page_mask.empty()) { - thrust::fill(policy, non_nz_map_page_mask.begin(), non_nz_map_page_mask.end(), true); - } else { - thrust::copy(policy, page_mask.begin(), page_mask.end(), non_nz_map_page_mask.begin()); - } - } - dim3 const grid(pages.size(), 1); dim3 const block(level_decode_block_size, 1); if (level_type_size == 1) { - precompute_flat_nz_map_kernel<<>>( - pages.device_ptr(), chunks, page_mask, non_nz_map_page_mask, min_row, num_rows); + precompute_flat_nz_map_kernel + <<>>(pages.device_ptr(), chunks, page_mask, min_row, num_rows); } else { - precompute_flat_nz_map_kernel<<>>( - pages.device_ptr(), chunks, page_mask, non_nz_map_page_mask, min_row, num_rows); + precompute_flat_nz_map_kernel + <<>>(pages.device_ptr(), chunks, page_mask, min_row, num_rows); } CUDF_CUDA_TRY(cudaGetLastError()); } diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index d166d4616da5..f897249d96d4 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -377,6 +377,9 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // Exit early if the page is pruned if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } + // A page carrying an nz map is decoded by `decode_delta_binary_kernel_with_nz_map` instead. The + // map is what routes, so the two kernels partition the pages without a second rule to agree with. + if (pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } [[maybe_unused]] null_count_back_copier _{s, static_cast(block.thread_rank())}; @@ -1140,7 +1143,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, size_t min_row, int level_type_size, cudf::device_span page_mask, - cudf::device_span non_nz_map_page_mask, + bool has_flat_nz_map, kernel_error::pointer error_code, cuda::stream_ref stream) { @@ -1150,19 +1153,13 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, dim3 dim_grid(pages.size(), 1); // 1 threadblock per page dim3 dim_block_nz_map(decode_delta_binary_with_nz_map_block_size, 1); - // The caller sizes this mask only when the nz map producer has pages to map, so a non-empty - // mask is what tells us the producer ran. Its contents are the complement of what the consumer - // below decodes: the producer cleared the entry for every page it mapped, leaving the rest for - // `decode_delta_binary_kernel`. - auto const use_flat_nz_map = not non_nz_map_page_mask.empty(); - if (use_flat_nz_map) { + // Both kernels take the caller's page mask and route on the page's own nz map, so this only + // decides whether the consumer has anything to do at all. + if (has_flat_nz_map) { // Not templated on `level_t`: this kernel never reads levels, which is the whole point of it. decode_delta_binary_kernel_with_nz_map<<>>( pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); CUDF_CUDA_TRY(cudaGetLastError()); - - // Hand `decode_delta_binary_kernel` every page the consumer above does not decode. - page_mask = non_nz_map_page_mask; } if (level_type_size == 1) { diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index 904bc2f66387..1ab4f7a5f028 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -1075,7 +1075,7 @@ void write_final_offsets(host_span offsets, * @param[in] min_row Minimum number of rows to read * @param[in] level_type_size Size in bytes of the type for level decoding * @param[in] page_mask Boolean vector indicating which pages need to be decoded - * @param[in] non_nz_map_page_mask Pages with no nz map, empty when no page has one + * @param[in] has_flat_nz_map Whether any page of this subpass carries a flat nz map * @param[out] error_code Error code for kernel failures * @param[in] stream CUDA stream to use */ @@ -1085,7 +1085,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, size_t min_row, int level_type_size, cudf::device_span page_mask, - cudf::device_span non_nz_map_page_mask, + bool has_flat_nz_map, kernel_error::pointer error_code, cuda::stream_ref stream); @@ -1199,8 +1199,6 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, * @param[in,out] pages All pages to be processed * @param[in] chunks All chunks to be processed * @param[in] page_mask Boolean vector indicating which pages need to be processed - * @param[out] non_nz_map_page_mask Page mask for the level-decoding kernels. Seeded here from - * @p page_mask (all true when that is empty), then cleared for every page mapped * @param[in] min_row Minimum row index to read * @param[in] num_rows Number of rows to read starting from min_row * @param[in] level_type_size Size in bytes of the type for level decoding (1 or 2) @@ -1209,12 +1207,10 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, cudf::detail::hostdevice_span chunks, cudf::device_span page_mask, - cudf::device_span non_nz_map_page_mask, size_t min_row, size_t num_rows, int level_type_size, - cuda::stream_ref stream, - cudf::memory_resources mr); + cuda::stream_ref stream); /** * @brief Fills output offset entries for pruned string and list pages diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index 5f2a54350e5d..fcecea023b09 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -207,19 +207,6 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ return page.has_nz_map(nz_map_kind::DELTA_FLAT); }); - // Partitions the pages into those with and without an nz map. Allocated before the fork, like - // `error_code` above: `fork_streams` records its event on `_stream` at fork time, so anything - // allocated afterwards is not ordered against its use on a forked stream under a stream-ordered - // memory resource. - // - // TODO: this mask is transitional. It exists only because the level-decoding kernels are - // deliberately left unmodified, so the page mask they already consult is the only way to tell - // them which pages another kernel has taken. Once every delta kernel has an nz map consumer and - // the feature is on by default, each can test `has_nz_map()` on the page directly and this - // buffer, its seeding and the producer's marking all go away. - rmm::device_uvector non_nz_map_page_mask(has_flat_nz_map ? subpass.pages.size() : 0, - _stream); - // The producer gets its own stream and its own event rather than `fork_streams`' shared one, so // that only the delta consumer waits on it and the other decoders overlap with it instead. int const nkernels = std::bitset<32>(kernel_mask).count(); @@ -229,15 +216,8 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ if (has_flat_nz_map) { auto const nz_map_stream = streams.back(); auto const page_mask = subpass_page_mask_span(); - precompute_flat_nz_map(subpass.pages, - pass.chunks, - page_mask, - non_nz_map_page_mask, - skip_rows, - num_rows, - level_type_size, - nz_map_stream, - _mr); + precompute_flat_nz_map( + subpass.pages, pass.chunks, page_mask, skip_rows, num_rows, level_type_size, nz_map_stream); nz_map_done.emplace(nz_map_stream); } @@ -345,7 +325,7 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ skip_rows, level_type_size, subpass_page_mask_span(), - non_nz_map_page_mask, + has_flat_nz_map, error_code.data(), delta_stream); } From da6847ae0d47d2f6650d5cd9256ec8ad03f5185f Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sun, 11 Oct 2026 04:46:24 +0000 Subject: [PATCH 07/13] fix(parquet): guard the overrun report and drop dead resume handling Two defects found reviewing the branch as a whole. The truncated-stream report added earlier on this branch ORs its code onto `s->setup.error` without checking whether that is already set. A page that fails `setup_local_page_info` never enters the value loop, so `src_pos` stays 0 and the test fires: the reader returns `EMPTY_PAGE | DATA_STREAM_OVERRUN` with the nz map on and `EMPTY_PAGE` with it off, for the same file. Now reported only when nothing else is. `decode_delta_binary_kernel_with_nz_map` carried `skipped_leaf_values` handling copied from the level-decoding kernel, where it is already dead -- `setup_local_page_info` zeroes it for any page without repetition, which is what `classify_nz_map_kind` requires. It is also wrong for this kernel's data model: `nz_count` counts from page position 0, so a rank already spans the prefix below `first_row`, and `value_at(sp + skipped_leaf_values)` would have asked for that prefix twice. The resume branch, the batch-size special case, the `target_pos` subtraction and the index offset are gone, with the invariant stated where the decode index is set up. Also removed: a comment claiming the producer clears this page from a mask, which described machinery deleted in the previous commit; a comment asserting the level-decoding path "stays untouched", which stopped being true when it gained its routing branch; a `__popc` the earlier modernization missed, one line below its replaced sibling; an include left unused when the memory resource parameter went away; and three `block.sync()` calls that each follow a thread-0 write to global memory with nothing after them in the block. `num_claimed` becomes `num_mapped`, which the prose sweep had already done everywhere except this identifier. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 28 +++++----- cpp/src/io/parquet/page_delta_decode.cu | 54 ++++++++------------ cpp/src/io/parquet/reader_impl.cpp | 3 +- cpp/src/io/parquet/reader_impl_chunking.hpp | 3 +- cpp/src/io/parquet/reader_impl_preprocess.cu | 8 +-- 5 files changed, 41 insertions(+), 55 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index d8addce6ff90..4aabb7a90fdd 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -11,7 +11,6 @@ #include #include -#include #include #include @@ -30,7 +29,7 @@ constexpr int preprocess_block_size = 512; constexpr int level_decode_block_size = 128; /** - * @brief Computes the nz map for one page. + * @brief Computes the nz map for each page, one block per page. * * Walks the page's already-decoded definition levels once, writing * - the output column's null mask for the rows this page contributes @@ -60,7 +59,7 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) int const t = block.thread_rank(); PageInfo* const pp = &pages[page_idx]; if (!pp->has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } - if (!page_mask.empty() && !page_mask[page_idx]) { return; } + if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } auto* const s = &state_g; if (!setup_local_page_info( @@ -81,20 +80,18 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) int const first_row = min(s->setup.first_row, decoded_value_limit); int const value_limit = min(decoded_value_limit, first_row + s->setup.num_rows); - // A required page's rank map is the identity, so there is nothing to materialize; the consumer - // synthesizes it. `nz_idx` stays null, which is what tells the consumer to do that. + // A required page's rank map is the identity, so there is nothing to materialize. The consumer + // treats a null `nz_idx` as the identity map. if (!should_process_nulls(s)) { if (t == 0) { pp->nz_map->nz_count = value_limit; } - block.sync(); return; } - // The other way to reach `num_decoded_level_values == 0`: an optional page that contributes no - // rows to this read gets no level buffer, and there is nothing to map. + // An optional page that contributes no rows to this read gets no level buffer; there is nothing + // to map. auto const* def = reinterpret_cast(pp->lvl_decode_buf[level_type::DEFINITION]); if (def == nullptr) { if (t == 0) { pp->nz_map->nz_count = 0; } - block.sync(); return; } @@ -103,9 +100,9 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) int const warp_id = t / cudf::detail::warp_size; int valid_count = 0; - // Hoisted for the same reason as `nz_idx` above: `store_validity` writes through a global - // `bitmask_type*`, which the compiler cannot prove does not alias the shared decode state, so it - // re-reads these from shared memory on every iteration otherwise. + // `store_validity` writes through a global `bitmask_type*` that the compiler cannot prove does + // not alias the shared decode state, so these reload from shared memory every iteration unless + // hoisted. Same for `nz_idx` above. auto* const valid_map = ni.valid_map; auto const valid_map_offset = ni.valid_map_offset; @@ -141,7 +138,8 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) if (w < warp_id) { warp_prefix += warp_valid_counts[w]; } block_valid_count += warp_valid_counts[w]; } - int const thread_valid_count = warp_prefix + __popc(valid_mask & ((1u << lane) - 1)); + int const thread_valid_count = + warp_prefix + cuda::std::popcount(valid_mask & ((1u << lane) - 1)); if (is_valid) { nz_idx[valid_count + thread_valid_count] = value_pos; } valid_count += block_valid_count; @@ -151,10 +149,8 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) block.sync(); } - // The consumer recomputes the null count for the rows this page contributes, so the producer - // writes only the rank map and its size. + // The page's valid count. if (t == 0) { pp->nz_map->nz_count = valid_count; } - block.sync(); } using unused_state_buf = page_state_buffers_s<0, 0, 0>; diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index f897249d96d4..cff901a64e44 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -32,7 +32,7 @@ constexpr int decode_delta_binary_block_size = 96; constexpr int delta_nz_buf_size = 4 * delta_max_batch_size; // One producer warp and one writer warp -constexpr int decode_delta_binary_with_nz_map_block_size = 64; // producer + writer +constexpr int decode_delta_binary_with_nz_map_block_size = 64; // DELTA_BYTE_ARRAY encoding (incremental encoding or front compression), is used for BYTE_ARRAY // columns. For each element in a sequence of strings, a prefix length from the preceding string @@ -377,8 +377,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // Exit early if the page is pruned if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } - // A page carrying an nz map is decoded by `decode_delta_binary_kernel_with_nz_map` instead. The - // map is what routes, so the two kernels partition the pages without a second rule to agree with. + // A page carrying an nz map is decoded by `decode_delta_binary_kernel_with_nz_map` instead. if (pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } [[maybe_unused]] null_count_back_copier _{s, static_cast(block.thread_rank())}; @@ -526,9 +525,6 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // Decodes flat DELTA_BINARY_PACKED pages, like `decode_delta_binary_kernel`, but places values // using the nz map's rank map instead of decoding definition levels itself. -// -// A separate kernel rather than a branch inside that one, so the level-decoding path stays -// untouched while the nz map is opt-in. CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) decode_delta_binary_kernel_with_nz_map(PageInfo* pages, device_span chunks, @@ -547,8 +543,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) auto* const db = &db_state; if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } - // The producer cleared this page from `decode_delta_binary_kernel`'s mask when it mapped it, - // so every page is decoded by one kernel and none by both. + // Pages with a map are decoded here; the level-decoding kernel takes the rest. if (!pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } [[maybe_unused]] null_count_back_copier _{s, static_cast(block.thread_rank())}; @@ -567,8 +562,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) auto const* const nz_idx = pp->nz_map->nz_idx; int const nz_count_in = pp->nz_map->nz_count; - // This check failing indicates that we somehow failed to write out the null map for a page with - // nulls entering this kernel. + // A page with nulls reaching this kernel must have had its nz map produced. cudf_assert(nz_count_in >= 0 && (not process_nulls || nz_idx != nullptr)); auto& ni = s->nesting.nesting_info[s->setup.col.max_nesting_depth - 1]; @@ -599,33 +593,29 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) return; } - uint32_t const skipped_leaf_values = s->setup.page.skipped_leaf_values; - bool const is_skip_resume = skipped_leaf_values > 0; - uint32_t const batch_size = - is_skip_resume ? cudf::detail::warp_size - : min(db->values_per_mb, static_cast(delta_max_batch_size)); + // `setup_local_page_info` zeroes `skipped_leaf_values` for a page without repetition, and the + // classifier selects only such pages, so there is no resume to handle here. The decode index is + // the rank itself: `nz_count` counts from page position 0, so it already spans any prefix below + // `first_row`. + uint32_t const batch_size = min(db->values_per_mb, static_cast(delta_max_batch_size)); uint32_t const passes_per_batch = batch_size / cudf::detail::warp_size; - if (is_skip_resume) { db->skip_values(skipped_leaf_values, block, warp); } // This is a two-warp version of delta binary decoding using a delta producer warp and a value // writer warp. The writer trails the producer by one batch to allow the two to run concurrently // in a single phase. `next_pass_start_idx()` is the first stream index the producer has not // decoded yet. Every index below it is decoded, and is still resident because - // `delta_rolling_buf_size` holds two batches, the one in flight plus the one trailing. Reading it - // only at the barrier keeps the concurrent region free of any access to the decoder's mutable - // scalars. `value_at()` touches `value[]` and `first_value`, and the producer writes neither - // below that index. + // `delta_rolling_buf_size` holds two batches, the one in flight plus the one trailing. Reading + // `next_pass_start_idx()` only at the barrier keeps the concurrent region free of any access to + // the decoder's mutable scalars. `value_at()` touches `value[]` and `first_value`, and the + // producer writes neither below that index. uint32_t produced = db->next_pass_start_idx(); while (s->setup.error == 0 && s->progress.src_pos < s->progress.nz_count) { uint32_t const src_pos = s->progress.src_pos; - // Decoder indices run ahead of output positions by `skipped_leaf_values` on a resume page. // Capped at one batch so the writer trails the producer by exactly one, never more: the // rolling buffer holds two batches, so a writer that fell two behind would read values the // producer had already overwritten. uint32_t const target_pos = - min(s->progress.nz_count, - min(produced > skipped_leaf_values ? produced - skipped_leaf_values : 0, - src_pos + batch_size)); + min(static_cast(s->progress.nz_count), min(produced, src_pos + batch_size)); block.sync(); if (warp.meta_group_rank() == 0) { @@ -639,7 +629,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) dst_pos -= s->setup.first_row; if (dst_pos >= 0) { void* const dst = ni.data_out + dst_pos * s->output_cvt.dtype_len; - auto const val = db->value_at(sp + skipped_leaf_values); + auto const val = db->value_at(sp); switch (s->output_cvt.dtype_len) { case 1: *static_cast(dst) = val; break; case 2: *static_cast(dst) = val; break; @@ -652,9 +642,8 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) } block.sync(); - // Bounding the writer by the producer means a malformed page whose delta stream holds fewer - // values than its definition levels imply would leave `nz_count` permanently out of reach, and - // the loop would spin forever. To avoid that, we must stop once neither warp can advance. + // Stop once neither warp advances: a malformed page whose delta stream is shorter than its + // definition levels imply never reaches `nz_count`. // // Both halves of that test are needed. The producer finishes first on a well-formed page, and // we need to wait for the writer to drain the trailing batch -- breaking on the producer alone @@ -673,7 +662,8 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) // Values still outstanding means the loop stopped above rather than finishing: the delta stream // held fewer values than the definition levels call for. Those rows are already marked valid, so // without an error the read would return whatever their output slots happened to contain. - if (block.thread_rank() == 0 and s->progress.src_pos < s->progress.nz_count) { + if (block.thread_rank() == 0 and s->setup.error == 0 and + s->progress.src_pos < s->progress.nz_count) { s->setup.error |= static_cast(decode_error::DATA_STREAM_OVERRUN); } @@ -1153,10 +1143,10 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, dim3 dim_grid(pages.size(), 1); // 1 threadblock per page dim3 dim_block_nz_map(decode_delta_binary_with_nz_map_block_size, 1); - // Both kernels take the caller's page mask and route on the page's own nz map, so this only - // decides whether the consumer has anything to do at all. + // Both kernels route on the page's own nz map, so `has_flat_nz_map` only skips a launch that + // would find no work. if (has_flat_nz_map) { - // Not templated on `level_t`: this kernel never reads levels, which is the whole point of it. + // Not templated on `level_t`: this kernel never reads levels. decode_delta_binary_kernel_with_nz_map<<>>( pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); CUDF_CUDA_TRY(cudaGetLastError()); diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index fcecea023b09..21546b00bf6a 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -316,8 +316,7 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ // launch delta binary decoder if (BitAnd(kernel_mask, decode_kernel_mask::DELTA_BINARY) != 0) { auto const delta_stream = streams[s_idx++]; - // The only kernel that reads the nz map. The others overlap with the producer rather than - // queueing behind it, which is why the producer does not ride `fork_streams`' shared event. + // The only kernel that reads the nz map. if (has_flat_nz_map) { delta_stream.wait(*nz_map_done); } decode_delta_binary(subpass.pages, pass.chunks, diff --git a/cpp/src/io/parquet/reader_impl_chunking.hpp b/cpp/src/io/parquet/reader_impl_chunking.hpp index 1c0cf29e8c5f..2f4bf8f3db3d 100644 --- a/cpp/src/io/parquet/reader_impl_chunking.hpp +++ b/cpp/src/io/parquet/reader_impl_chunking.hpp @@ -100,7 +100,8 @@ struct subpass_intermediate_data { cuda::device_buffer level_decode_data; // Backing store for the flat nz maps. `allocate_level_decode_space` - // sub-allocates one slice of this per mapped page. Empty unless some page has an nz map. + // sub-allocates one slice of this per mapped page that has definition levels. Empty unless some + // page has an nz map. cuda::device_buffer flat_nz_map_data; // Out-of-line nz map scratch, one entry per page. Empty unless some page of this subpass has // an nz map. See page_nz_map. diff --git a/cpp/src/io/parquet/reader_impl_preprocess.cu b/cpp/src/io/parquet/reader_impl_preprocess.cu index 73bce020fce3..b431254b6c98 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess.cu @@ -397,13 +397,13 @@ void reader_impl::allocate_level_decode_space() // needed on the host. On device, a non-null `PageInfo::nz_map` always indicates that state // exists. std::vector nz_map_slot(num_pages, std::numeric_limits::max()); - size_t num_claimed = 0; + size_t num_mapped = 0; for (size_t idx = 0; idx < num_pages; ++idx) { - if (pages[idx].map_kind != nz_map_kind::NONE) { nz_map_slot[idx] = num_claimed++; } + if (pages[idx].map_kind != nz_map_kind::NONE) { nz_map_slot[idx] = num_mapped++; } } - if (num_claimed == 0) { return; } + if (num_mapped == 0) { return; } - subpass.nz_map_buf = cudf::detail::hostdevice_vector(num_claimed, _stream); + subpass.nz_map_buf = cudf::detail::hostdevice_vector(num_mapped, _stream); auto host_state = [&](size_t idx) -> page_nz_map* { return nz_map_slot[idx] != std::numeric_limits::max() From 24ded12ba27627b250cbafa887cd7089f7d9d85c Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sun, 11 Oct 2026 04:46:41 +0000 Subject: [PATCH 08/13] test(parquet): cover the nz map paths the differential test could not reach `NzMapMatchesLevelDecoder` read one nullable and one required flat column, which left four paths unexercised. An optional column with no null in it. `maybe_has_nulls` short-circuits on a definition-level stream that is one repeated run at the max level, so the producer takes its `!should_process_nulls` return with `nz_idx` allocated and the null mask left to the all-valid fill. The previous validity pattern put a null in every page, so this never ran. A nested column, which `classify_nz_map_kind` declines. Its pages stay on the level-decoding kernel while the flat columns go to the consumer, so the two kernels now partition the pages of one file inside a test rather than only in an nsys trace. A chunked read. The producer runs once per output chunk over the same scratch, each time with that chunk's row range rather than the pass's. A slice landing mid-page. The bounded read skipped 1001 rows against 1000-row pages, leaving `first_row` at 1, so the consumer's prefix search had a single value below it; at 1500 it has 500. The chunked comparison uses TABLES_EQUIVALENT because `cudf::concatenate` drops an all-true validity mask, so the all-valid column comes back non-nullable while the expected table is nullable. The values are identical. Not covered: `level_type_size == 2`, which needs a max definition or repetition level of 256 and so around that many levels of nesting. The `uint16_t` instantiation of the producer is unreachable in practice for the same reason. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. An nsys trace of the test confirms the producer, the consumer and the level-decoding kernel all run. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/tests/io/parquet_reader_test.cpp | 78 ++++++++++++++++++++++++---- 1 file changed, 67 insertions(+), 11 deletions(-) diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 6c5d5d322dd5..69d71d3d0864 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -90,11 +90,11 @@ TEST_F(ParquetReaderTest, ManyTinyStringPages) // TODO: delete this test once the nz map is on by default. // -// This test enables usage of the nz map producer and the corresponding delta binary kernels that -// consume the nz map producer to validate them. Currently those kernels are gated by an environment -// variable that will be removed once the full set of kernels has been merged and the end result -// benchmarked. Once the default flips, the rest of the suite covers those paths and this has -// nothing left to add. +// This test reads one file twice, with the nz map producer and its delta binary consumer off and +// then on, and requires the two to agree. Those kernels are gated by an environment variable that +// will be removed once the full set of kernels has been merged and the end result benchmarked. +// Once the default flips, the rest of the suite covers those paths and this has nothing left to +// add. TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) { // The nz map is a second decode path selected by an env var: the same file must read @@ -115,12 +115,34 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) // A required column takes the consumer's identity-map path: the producer writes no `nz_idx` and // the consumer derives the position from the rank alone. cudf::test::fixed_width_column_wrapper const required_col(values, values + num_rows); - auto const expected = table_view({nullable_col, required_col}); + // Optional but with no null in it. `maybe_has_nulls` short-circuits, so the producer returns + // early with `nz_idx` allocated and the null mask left to the all-valid fill. + auto const all_valid = + cudf::detail::make_counting_transform_iterator(0, [](auto) { return true; }); + cudf::test::fixed_width_column_wrapper const no_nulls_col{ + values, values + num_rows, all_valid}; + // A nested column the classifier declines, so its pages stay on the level-decoding kernel while + // the flat ones above go to the consumer. The two must partition the pages between them. + auto const list_offsets = cudf::detail::make_counting_transform_iterator( + cudf::size_type{0}, [](auto row) { return static_cast(2 * row); }); + auto const nested_col = cudf::make_lists_column( + num_rows, + cudf::test::fixed_width_column_wrapper(list_offsets, + list_offsets + num_rows + 1) + .release(), + cudf::test::fixed_width_column_wrapper(values, values + (2 * num_rows)).release(), + 0, + cudf::create_null_mask(0, cudf::mask_state::UNALLOCATED)); + + auto const expected = table_view({nullable_col, required_col, no_nulls_col, nested_col->view()}); auto input_metadata = cudf::io::table_input_metadata{expected}; for (auto& col_meta : input_metadata.column_metadata) { col_meta.set_encoding(cudf::io::column_encoding::DELTA_BINARY_PACKED); } + // The list's leaf carries the encoding; the list column itself is not a leaf. + input_metadata.column_metadata[3].child(1).set_encoding( + cudf::io::column_encoding::DELTA_BINARY_PACKED); std::vector buffer; cudf::io::write_parquet( @@ -140,11 +162,10 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) .build()); }; - // Unset is the shipping configuration while the feature is opt-in. { tmp_env_var const env{nz_map_env_var, "0"}; - auto const legacy = read_back(); - CUDF_TEST_EXPECT_TABLES_EQUAL(expected, legacy.tbl->view()); + auto const level_decoded = read_back(); + CUDF_TEST_EXPECT_TABLES_EQUAL(expected, level_decoded.tbl->view()); } { tmp_env_var const env{nz_map_env_var, "1"}; @@ -158,11 +179,13 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) cudf::io::parquet_reader_options::builder( cudf::io::source_info{cudf::host_span{ reinterpret_cast(buffer.data()), buffer.size()}}) - .skip_rows(1001) + .skip_rows(1500) .num_rows(4321) .build()); }; - auto const sliced = cudf::slice(expected, {1001, 1001 + 4321}); + // 1500 lands mid-page against 1000-row pages, so the consumer's prefix search has 500 values + // below `first_row` to skip rather than one. + auto const sliced = cudf::slice(expected, {1500, 1500 + 4321}); { tmp_env_var const env{nz_map_env_var, "0"}; CUDF_TEST_EXPECT_TABLES_EQUAL(sliced, trimmed().tbl->view()); @@ -172,6 +195,39 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) CUDF_TEST_EXPECT_TABLES_EQUAL(sliced, trimmed().tbl->view()); } } + + // A chunked read runs the producer once per output chunk over the same scratch, each time with + // that chunk's row range rather than the pass's. + { + auto const read_chunked = [&] { + auto reader = cudf::io::chunked_parquet_reader( + 240'000, + cudf::io::parquet_reader_options::builder( + cudf::io::source_info{cudf::host_span{ + reinterpret_cast(buffer.data()), buffer.size()}}) + .build()); + std::vector> chunks; + while (reader.has_next()) { + chunks.push_back(reader.read_chunk().tbl); + } + std::vector views; + views.reserve(chunks.size()); + for (auto const& c : chunks) { + views.push_back(c->view()); + } + return std::pair{cudf::concatenate(views), std::move(chunks)}; + }; + { + tmp_env_var const env{nz_map_env_var, "0"}; + auto const [joined, keep] = read_chunked(); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected, joined->view()); + } + { + tmp_env_var const env{nz_map_env_var, "1"}; + auto const [joined, keep] = read_chunked(); + CUDF_TEST_EXPECT_TABLES_EQUIVALENT(expected, joined->view()); + } + } } TEST_F(ParquetReaderTest, UserBounds) From e0ad08fa205d1080df5520054989a27955de7ac5 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sun, 11 Oct 2026 05:13:27 +0000 Subject: [PATCH 09/13] refactor(parquet): hoist the consumer's output pointer out of its writer loop `ni.data_out` reaches the output through a `PageNestingDecodeInfo*` that may point at the shared nesting cache or at global memory, and the store it feeds writes through that same pointer, so the compiler cannot prove the store does not alias `ni` and reloads it on every value. In SASS that is a generic `LD.E.64` per value, plus shared loads of `first_row` and `dtype_len` and a switch dispatching on a value it has just reloaded. The producer already hoists `valid_map` and `valid_map_offset` for the same reason; this brings the consumer in line. Performance-neutral, which is the claim being made here -- not a speedup. Interleaved A/B on `parquet_read_delta_binary_nullable_page_sizes` with the nz map enabled, H100 clock-locked at 1755 MHz, 6 rounds with the arm order alternating per round: geomean 0.9954, with the hoist faster in 14 of 24 paired rounds. Register count is unchanged at 40 and neither arm spills, so occupancy is not involved either way. The ordering mattered more than the change did. An earlier run of the same comparison put the hoist 1.9% behind, but it ran the arms in a fixed order every round, and re-running with the order alternating shows the sign following the order rather than the code: 1.0032 when the unhoisted arm ran first, 0.9877 when the hoisted arm did. Whichever arm runs second measures worse by about 1.5% on this benchmark. The apparent regression was an artifact of the harness. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/page_delta_decode.cu | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index cff901a64e44..84b0f32b78f9 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -608,7 +608,13 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) // `next_pass_start_idx()` only at the barrier keeps the concurrent region free of any access to // the decoder's mutable scalars. `value_at()` touches `value[]` and `first_value`, and the // producer writes neither below that index. - uint32_t produced = db->next_pass_start_idx(); + // + // `data_out` and the two scalars are hoisted because the store below writes through `data_out`, + // which the compiler cannot prove does not alias `ni`; left in place they reload once per value. + auto* const data_out = ni.data_out; + auto const dtype_len = s->output_cvt.dtype_len; + auto const first_row = s->setup.first_row; + uint32_t produced = db->next_pass_start_idx(); while (s->setup.error == 0 && s->progress.src_pos < s->progress.nz_count) { uint32_t const src_pos = s->progress.src_pos; // Capped at one batch so the writer trails the producer by exactly one, never more: the @@ -626,11 +632,11 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) } else if (warp.meta_group_rank() == 1 && src_pos < target_pos) { for (uint32_t sp = src_pos + warp.thread_rank(); sp < target_pos; sp += warp.size()) { auto dst_pos = process_nulls ? static_cast(nz_idx[sp]) : static_cast(sp); - dst_pos -= s->setup.first_row; + dst_pos -= first_row; if (dst_pos >= 0) { - void* const dst = ni.data_out + dst_pos * s->output_cvt.dtype_len; + void* const dst = data_out + dst_pos * dtype_len; auto const val = db->value_at(sp); - switch (s->output_cvt.dtype_len) { + switch (dtype_len) { case 1: *static_cast(dst) = val; break; case 2: *static_cast(dst) = val; break; case 4: *static_cast(dst) = val; break; From 6cba3c6b7859fb86ea714cfa19d14d6fdabaf9ac Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sun, 11 Oct 2026 17:38:22 +0000 Subject: [PATCH 10/13] fix(parquet): gate both delta routing decisions on one flag `decode_delta_binary_kernel` skipped any page carrying an nz map, while the kernel that picks those pages up was launched only when the host's `has_flat_nz_map` was also true. Two derivations of the same decision: the host scan requires the env var and `classify_nz_map_kind` requires `kernel_mask & DELTA_BINARY`, the device skip requires neither. They agree today, and by construction rather than by luck -- a page only gets a map when both of those hold, so the host scan cannot be false while a page is mapped. But nothing makes that structural, and the window between `allocate_level_decode_space` and `decode_page_data` already contains a `kernel_mask` rewriter in `maybe_transcode_dict_pages`. If the two ever part, the page is decoded by neither kernel and no error is raised, and because the null mask is filled all-valid beforehand, the uninitialized output reads back as valid values rather than as nulls. Passing the launch flag into the kernel makes the skip conditional on it, so the two cannot disagree. When it is false the level-decoding kernel takes every page, which it decodes correctly: it walks definition levels and never consults the map. This also removes the one way the consumer could meet a page whose producer never ran. `nz_count` is seeded to -1 and is signed, so both the loop condition and the truncated-stream check would read false and the kernel would emit nothing and report nothing, leaving only a `cudf_assert` that compiles out in release. That state is now unreachable rather than merely unreached. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/page_delta_decode.cu | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 84b0f32b78f9..a82ff6345f95 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -362,6 +362,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) size_t min_row, size_t num_rows, cudf::device_span page_mask, + bool has_flat_nz_map, kernel_error::pointer error_code) { __shared__ __align__(16) delta_binary_decoder db_state; @@ -378,7 +379,10 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // Exit early if the page is pruned if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } // A page carrying an nz map is decoded by `decode_delta_binary_kernel_with_nz_map` instead. - if (pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } + // Conditional on the same flag that launches that kernel, so the two routing decisions are one + // bit: were they derived separately, a disagreement would leave a page decoded by neither, with + // its rows already marked valid by the null mask fill. + if (has_flat_nz_map and pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } [[maybe_unused]] null_count_back_copier _{s, static_cast(block.thread_rank())}; @@ -1160,11 +1164,11 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, if (level_type_size == 1) { decode_delta_binary_kernel<<>>( - pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); + pages.device_ptr(), chunks, min_row, num_rows, page_mask, has_flat_nz_map, error_code); CUDF_CUDA_TRY(cudaGetLastError()); } else { decode_delta_binary_kernel<<>>( - pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); + pages.device_ptr(), chunks, min_row, num_rows, page_mask, has_flat_nz_map, error_code); CUDF_CUDA_TRY(cudaGetLastError()); } } From 0975a8a04f451e89de566535d7dce7703c3a9b71 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Sun, 11 Oct 2026 17:48:02 +0000 Subject: [PATCH 11/13] refactor(parquet): drop three things the nz map no longer needs The consumer's `cudf_assert` on `nz_count`. Gating both routing decisions on one flag made the state it checked unreachable rather than merely unreached, so there is nothing left for it to catch. It had also drawn a review suggestion to report the condition through `set_error` instead, which would have been wrong in a way worth recording: every `decode_error` names something about the file, and the reader surfaces them as "Parquet data decode failed", so using that channel for a producer-consumer disagreement would have told a user their data was corrupt to report a bug in cuDF. `page_nz_map::aux_count`. Nothing writes or reads it. The branch that eventually needs it writes and reads it inside a single kernel, where it is a local as well; it becomes cross-kernel state only with the list producer. It can come back then, with the code that uses it. The producer's `first_row` clamp. `min(s->setup.first_row, decoded_value_limit)` guarded against the requested first row running past the page's decoded levels, which cannot happen: the level buffer is sized over the pass's row range and a read's range is a subrange of it. The consumer uses `first_row` unclamped for both the prefix search and the output position, so the clamp also left the two kernels nominally working from different values, resolved only by its being a no-op. Replaced with the invariant it was standing in for, stated once. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 105/105, both with the nz map off and on. The chunked read added earlier in this branch is the case that would notice the clamp, since it re-runs the producer per output chunk with that chunk's row range against a pass-sized level buffer. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 5 ++++- cpp/src/io/parquet/page_delta_decode.cu | 3 --- cpp/src/io/parquet/parquet_gpu.hpp | 3 --- 3 files changed, 4 insertions(+), 7 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 4aabb7a90fdd..9ae749e16322 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -77,7 +77,10 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) pp->num_decoded_level_values > 0 ? min(s->setup.page.num_input_values, pp->num_decoded_level_values) : s->setup.page.num_input_values; - int const first_row = min(s->setup.first_row, decoded_value_limit); + // `first_row` needs no clamp against `decoded_value_limit`: the level buffer is sized over the + // pass's row range and a read's range is a subrange of it, so the rows asked for are always + // within the levels decoded. The consumer uses the same unclamped value to place its output. + int const first_row = s->setup.first_row; int const value_limit = min(decoded_value_limit, first_row + s->setup.num_rows); // A required page's rank map is the identity, so there is nothing to materialize. The consumer diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index a82ff6345f95..59cd450d043f 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -566,9 +566,6 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) auto const* const nz_idx = pp->nz_map->nz_idx; int const nz_count_in = pp->nz_map->nz_count; - // A page with nulls reaching this kernel must have had its nz map produced. - cudf_assert(nz_count_in >= 0 && (not process_nulls || nz_idx != nullptr)); - auto& ni = s->nesting.nesting_info[s->setup.col.max_nesting_depth - 1]; if (block.thread_rank() == 0) { // A sliced page keeps the valid values preceding `first_row` in its map, so the null count for diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index 1ab4f7a5f028..42af088bd1ed 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -365,9 +365,6 @@ struct page_nz_map { uint32_t* nz_idx{}; // Negative until the producer runs; the page's valid count afterwards. int32_t nz_count{not_yet_produced}; - // A second producer-written count, for map kinds that need one. Unused by `DELTA_FLAT`, whose - // consumer derives the null count from `nz_idx` and `nz_count` itself. - int32_t aux_count{}; }; /** From 76242625123b8f4fecf3f9159a5e6252e5016fa1 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Tue, 13 Oct 2026 01:10:02 +0000 Subject: [PATCH 12/13] refactor(parquet): order the nz map consumer by stream instead of an event The producer and the one kernel that reads its output now share a stream, so program order does what a `cuda::event` and an explicit wait did before. The level-decoding kernel stops waiting with them: it reads `map_kind` and `nz_map`, both host-set, and nothing the producer writes, so it never needed that edge. Nested delta pages overlap the producer now rather than queueing behind it. That makes the producer and the level-decoding kernel concurrent writers to the leaf validity buffer. They own disjoint pages, and where their ranges meet inside one bitmask word both are necessarily partial, which is the case `store_validity` handles with atomics. racecheck over the delta and nz map tests with the feature on reports no hazards. No measurement is claimed here. The argument is that it removes a synchronization object and lets one kernel overlap that previously could not; it is not known to be faster, and the benchmarks that could show it are the ones this series has repeatedly found too noisy to resolve changes of this size. Also from review: `static_assert` that `level_decode_block_size` is a multiple of the warp size and that the consumer's block is exactly the two warps its comment claims; `cg::invoke_one` in place of `if (thread_rank() == 0)` where a block is already in scope; the include left behind when the `cudf_assert` went; and the chunked read test now sets both limits explicitly so it produces several subpasses rather than relying on a default to do so. Three comments trimmed. Review called them "AI thinking paste", which is fair: they walked through how a design was reached rather than saying what the code does. The aliasing reason for the writer-loop hoist and the rule about both halves of the loop-exit test are kept, since neither is recoverable from the code; the reasoning around them is not. Validation: PARQUET_TEST 563 passed / 1 pre-existing skip and HYBRID_SCAN_TEST 106/106, both with the nz map off and on. compute-sanitizer racecheck and memcheck: 0 hazards, 0 errors. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/decode_preprocess.cu | 4 ++- cpp/src/io/parquet/page_delta_decode.cu | 46 ++++++++++++------------- cpp/src/io/parquet/parquet_gpu.hpp | 6 ++-- cpp/src/io/parquet/reader_impl.cpp | 23 +++++++------ cpp/tests/io/parquet_reader_test.cpp | 3 +- 5 files changed, 43 insertions(+), 39 deletions(-) diff --git a/cpp/src/io/parquet/decode_preprocess.cu b/cpp/src/io/parquet/decode_preprocess.cu index 9ae749e16322..a359d6ee7b7a 100644 --- a/cpp/src/io/parquet/decode_preprocess.cu +++ b/cpp/src/io/parquet/decode_preprocess.cu @@ -51,7 +51,7 @@ CUDF_KERNEL void __launch_bounds__(level_decode_block_size) size_t min_row, size_t num_rows) { - static constexpr int num_warps = level_decode_block_size / cudf::detail::warp_size; + constexpr int num_warps = level_decode_block_size / cudf::detail::warp_size; __shared__ __align__(16) full_page_decode_state state_g; __shared__ int warp_valid_counts[num_warps]; auto const block = cg::this_thread_block(); @@ -665,6 +665,8 @@ void precompute_flat_nz_map(cudf::detail::hostdevice_span pages, dim3 const grid(pages.size(), 1); dim3 const block(level_decode_block_size, 1); + static_assert(level_decode_block_size % cudf::detail::warp_size == 0, + "level_decode_block_size must be a multiple of the warp size"); if (level_type_size == 1) { precompute_flat_nz_map_kernel <<>>(pages.device_ptr(), chunks, page_mask, min_row, num_rows); diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index 59cd450d043f..b7da78e6fb5c 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -9,7 +9,6 @@ #include "page_string_utils.cuh" #include "parquet_gpu.hpp" -#include #include #include @@ -379,9 +378,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // Exit early if the page is pruned if (page_mask.size() > 0 and not page_mask[page_idx]) { return; } // A page carrying an nz map is decoded by `decode_delta_binary_kernel_with_nz_map` instead. - // Conditional on the same flag that launches that kernel, so the two routing decisions are one - // bit: were they derived separately, a disagreement would leave a page decoded by neither, with - // its rows already marked valid by the null mask fill. + // Gated on the same flag that launches that kernel, so the two routing decisions are one bit. if (has_flat_nz_map and pages[page_idx].has_nz_map(nz_map_kind::DELTA_FLAT)) { return; } [[maybe_unused]] null_count_back_copier _{s, static_cast(block.thread_rank())}; @@ -567,7 +564,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) int const nz_count_in = pp->nz_map->nz_count; auto& ni = s->nesting.nesting_info[s->setup.col.max_nesting_depth - 1]; - if (block.thread_rank() == 0) { + cg::invoke_one(block, [&] { // A sliced page keeps the valid values preceding `first_row` in its map, so the null count for // the rows this page contributes has to skip them. The map is ascending, so that prefix is the // distance to `first_row`'s lower bound. Only the first page of a read is sliced, so skip the @@ -581,16 +578,16 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) : static_cast(s->setup.first_row); ni.null_count = process_nulls ? s->setup.num_rows - (nz_count_in - prefix_valid_count) : 0; s->progress.nz_count = nz_count_in; - } + }); block.sync(); - if (block.thread_rank() == 0) { db->init_binary_block(s->stream.data_start, s->stream.data_end); } + cg::invoke_one(block, [&] { db->init_binary_block(s->stream.data_start, s->stream.data_end); }); block.sync(); if (db->error) { - if (block.thread_rank() == 0) { + cg::invoke_one(block, [&] { set_error(static_cast(decode_error::DELTA_PARAMS_UNSUPPORTED), error_code); - } + }); return; } @@ -610,8 +607,8 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) // the decoder's mutable scalars. `value_at()` touches `value[]` and `first_value`, and the // producer writes neither below that index. // - // `data_out` and the two scalars are hoisted because the store below writes through `data_out`, - // which the compiler cannot prove does not alias `ni`; left in place they reload once per value. + // Hoisted: the store below writes through `data_out`, which the compiler cannot prove does not + // alias `ni`, so in place these reload once per value. auto* const data_out = ni.data_out; auto const dtype_len = s->output_cvt.dtype_len; auto const first_row = s->setup.first_row; @@ -652,11 +649,8 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) // Stop once neither warp advances: a malformed page whose delta stream is shorter than its // definition levels imply never reaches `nz_count`. // - // Both halves of that test are needed. The producer finishes first on a well-formed page, and - // we need to wait for the writer to drain the trailing batch -- breaking on the producer alone - // would drop it from every page. Conversely, the writer has not run at all on the first - // iteration, so breaking on the writer alone would emit nothing for any page whose values fit - // in one batch. + // Both halves are needed: the producer finishes first and the writer still has a batch to + // drain, and on the first iteration the writer has not run at all. uint32_t const prev_produced = produced; produced = db->next_pass_start_idx(); if (produced == prev_produced && s->progress.src_pos <= src_pos) { break; } @@ -669,10 +663,11 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_with_nz_map_block_size) // Values still outstanding means the loop stopped above rather than finishing: the delta stream // held fewer values than the definition levels call for. Those rows are already marked valid, so // without an error the read would return whatever their output slots happened to contain. - if (block.thread_rank() == 0 and s->setup.error == 0 and - s->progress.src_pos < s->progress.nz_count) { - s->setup.error |= static_cast(decode_error::DATA_STREAM_OVERRUN); - } + cg::invoke_one(block, [&] { + if (s->setup.error == 0 and s->progress.src_pos < s->progress.nz_count) { + s->setup.error |= static_cast(decode_error::DATA_STREAM_OVERRUN); + } + }); if (block.thread_rank() == 0 and s->setup.error != 0) { set_error(s->setup.error, error_code); } } @@ -1142,19 +1137,22 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, cudf::device_span page_mask, bool has_flat_nz_map, kernel_error::pointer error_code, - cuda::stream_ref stream) + cuda::stream_ref stream, + cuda::stream_ref nz_map_stream) { CUDF_EXPECTS(pages.size() > 0, "There is no page to decode"); dim3 dim_block(decode_delta_binary_block_size, 1); dim3 dim_grid(pages.size(), 1); // 1 threadblock per page dim3 dim_block_nz_map(decode_delta_binary_with_nz_map_block_size, 1); + static_assert(decode_delta_binary_with_nz_map_block_size == 2 * cudf::detail::warp_size, + "The nz map consumer runs one producer warp and one writer warp"); - // Both kernels route on the page's own nz map, so `has_flat_nz_map` only skips a launch that - // would find no work. + // On the producer's stream, so it is ordered behind the producer without an event. The + // level-decoding kernel below needs no such ordering: it reads only host-set fields. if (has_flat_nz_map) { // Not templated on `level_t`: this kernel never reads levels. - decode_delta_binary_kernel_with_nz_map<<>>( + decode_delta_binary_kernel_with_nz_map<<>>( pages.device_ptr(), chunks, min_row, num_rows, page_mask, error_code); CUDF_CUDA_TRY(cudaGetLastError()); } diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index 6b1f588e8ee3..6457409a4be1 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -1074,7 +1074,8 @@ void write_final_offsets(host_span offsets, * @param[in] page_mask Boolean vector indicating which pages need to be decoded * @param[in] has_flat_nz_map Whether any page of this subpass carries a flat nz map * @param[out] error_code Error code for kernel failures - * @param[in] stream CUDA stream to use + * @param[in] stream CUDA stream for the level-decoding kernel + * @param[in] nz_map_stream Stream the producer ran on, which orders the consumer behind it */ void decode_delta_binary(cudf::detail::hostdevice_span pages, cudf::detail::hostdevice_span chunks, @@ -1084,7 +1085,8 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, cudf::device_span page_mask, bool has_flat_nz_map, kernel_error::pointer error_code, - cuda::stream_ref stream); + cuda::stream_ref stream, + cuda::stream_ref nz_map_stream); /** * @brief Launches kernel for reading the DELTA_BYTE_ARRAY column data stored in the pages diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index 21546b00bf6a..eab74265c069 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -207,18 +207,20 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ return page.has_nz_map(nz_map_kind::DELTA_FLAT); }); - // The producer gets its own stream and its own event rather than `fork_streams`' shared one, so - // that only the delta consumer waits on it and the other decoders overlap with it instead. + // The producer and the consumer that reads its output share a stream, so the other decoders + // overlap with them rather than queueing behind them. int const nkernels = std::bitset<32>(kernel_mask).count(); auto streams = cudf::detail::fork_streams(_stream, nkernels + (has_flat_nz_map ? 1 : 0)); - std::optional nz_map_done; + auto const nz_map_stream = streams.back(); if (has_flat_nz_map) { - auto const nz_map_stream = streams.back(); - auto const page_mask = subpass_page_mask_span(); - precompute_flat_nz_map( - subpass.pages, pass.chunks, page_mask, skip_rows, num_rows, level_type_size, nz_map_stream); - nz_map_done.emplace(nz_map_stream); + precompute_flat_nz_map(subpass.pages, + pass.chunks, + subpass_page_mask_span(), + skip_rows, + num_rows, + level_type_size, + nz_map_stream); } int s_idx = 0; @@ -316,8 +318,6 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ // launch delta binary decoder if (BitAnd(kernel_mask, decode_kernel_mask::DELTA_BINARY) != 0) { auto const delta_stream = streams[s_idx++]; - // The only kernel that reads the nz map. - if (has_flat_nz_map) { delta_stream.wait(*nz_map_done); } decode_delta_binary(subpass.pages, pass.chunks, num_rows, @@ -326,7 +326,8 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ subpass_page_mask_span(), has_flat_nz_map, error_code.data(), - delta_stream); + delta_stream, + nz_map_stream); } // launch byte stream split decoder diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 6fb07cd4d79f..5510c877ff4a 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -203,7 +203,8 @@ TEST_F(ParquetReaderTest, NzMapMatchesLevelDecoder) { auto const read_chunked = [&] { auto reader = cudf::io::chunked_parquet_reader( - 240'000, + 1'024, + 8'192, cudf::io::parquet_reader_options::builder( cudf::io::source_info{cudf::host_span{ reinterpret_cast(buffer.data()), buffer.size()}}) From 0dd235f1fbedd8db549f5f6cac3ad31ad37b5d00 Mon Sep 17 00:00:00 2001 From: Vyas Ramasubramani Date: Thu, 15 Oct 2026 17:43:40 +0000 Subject: [PATCH 13/13] fix(parquet): tighten the nz map's routing, sizing and stream handling Review of #24501 turned up four things. `decode_delta_binary` launched the level-decoding kernel unconditionally, so when every DELTA_BINARY page of a subpass carried a map, a full grid of blocks started only to read the two fields that tell them to return. The host now also computes whether any such page still needs level decoding, in the same walk that decides whether to launch the consumer, so the two halves of the routing decision cannot drift apart. The nz map was sized from `num_input_values`, but the producer never indexes past `min(num_input_values, num_decoded_level_values)`, and for a sliced read the latter is smaller. A page whose levels fall entirely outside the row range now gets no slice at all; the producer already reports a zero count for it and the consumer never reaches the map. `nz_map_stream` was taken from `streams.back()` even when no extra stream had been forked, which aliased another decoder's stream, and would index an empty vector if a subpass ever had no decode kernels. The producer's docstring said it runs once per subpass. It runs once per output chunk, each with its own row range; the allocation is what runs once per subpass. Co-Authored-By: Claude Opus 5 (1M context) --- cpp/src/io/parquet/page_delta_decode.cu | 5 +++++ cpp/src/io/parquet/parquet_gpu.hpp | 12 +++++++---- cpp/src/io/parquet/reader_impl.cpp | 22 ++++++++++++++------ cpp/src/io/parquet/reader_impl_preprocess.cu | 9 +++++++- 4 files changed, 37 insertions(+), 11 deletions(-) diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index b7da78e6fb5c..b0f5bd4bc68d 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -1136,6 +1136,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, int level_type_size, cudf::device_span page_mask, bool has_flat_nz_map, + bool needs_level_decoder, kernel_error::pointer error_code, cuda::stream_ref stream, cuda::stream_ref nz_map_stream) @@ -1157,6 +1158,10 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, CUDF_CUDA_TRY(cudaGetLastError()); } + // When every DELTA_BINARY page of this subpass carries a map, each block of this grid would do + // nothing but read the two fields it takes to decide to return. + if (not needs_level_decoder) { return; } + if (level_type_size == 1) { decode_delta_binary_kernel<<>>( pages.device_ptr(), chunks, min_row, num_rows, page_mask, has_flat_nz_map, error_code); diff --git a/cpp/src/io/parquet/parquet_gpu.hpp b/cpp/src/io/parquet/parquet_gpu.hpp index 6457409a4be1..77940f6e14ed 100644 --- a/cpp/src/io/parquet/parquet_gpu.hpp +++ b/cpp/src/io/parquet/parquet_gpu.hpp @@ -1073,6 +1073,7 @@ void write_final_offsets(host_span offsets, * @param[in] level_type_size Size in bytes of the type for level decoding * @param[in] page_mask Boolean vector indicating which pages need to be decoded * @param[in] has_flat_nz_map Whether any page of this subpass carries a flat nz map + * @param[in] needs_level_decoder Whether any DELTA_BINARY page of this subpass lacks a flat nz map * @param[out] error_code Error code for kernel failures * @param[in] stream CUDA stream for the level-decoding kernel * @param[in] nz_map_stream Stream the producer ran on, which orders the consumer behind it @@ -1084,6 +1085,7 @@ void decode_delta_binary(cudf::detail::hostdevice_span pages, int level_type_size, cudf::device_span page_mask, bool has_flat_nz_map, + bool needs_level_decoder, kernel_error::pointer error_code, cuda::stream_ref stream, cuda::stream_ref nz_map_stream); @@ -1190,10 +1192,12 @@ void preprocess_levels(cudf::detail::hostdevice_span pages, /** * @brief Launches the kernel that computes the flat nz map * - * Runs once per subpass, before the decode kernels, over the pages `classify_nz_map_kind` - * selected a consumer for. Writes - * `page_nz_map::nz_idx` / `nz_count` and the leaf column's null mask, so that the matching - * decode kernel can place values without walking definition levels itself. + * Runs once per output chunk, before the decode kernels, over the pages `classify_nz_map_kind` + * selected a consumer for. Writes `page_nz_map::nz_idx` / `nz_count` and the leaf column's null + * mask, so that the matching decode kernel can place values without walking definition levels + * itself. A subpass may yield several output chunks, each with its own `min_row` / `num_rows`, and + * the map covers one chunk's range at a time — unlike the allocation, which is sized once per + * subpass over the pass's row range. * * @param[in,out] pages All pages to be processed * @param[in] chunks All chunks to be processed diff --git a/cpp/src/io/parquet/reader_impl.cpp b/cpp/src/io/parquet/reader_impl.cpp index eab74265c069..7e07d9a27d68 100644 --- a/cpp/src/io/parquet/reader_impl.cpp +++ b/cpp/src/io/parquet/reader_impl.cpp @@ -201,17 +201,26 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ // create this before we fork streams kernel_error error_code(_stream); - auto const has_flat_nz_map = - _nz_map_enabled && - std::any_of(subpass.pages.host_begin(), subpass.pages.host_end(), [](PageInfo const& page) { - return page.has_nz_map(nz_map_kind::DELTA_FLAT); - }); + // Both halves of the delta routing decision, taken in one walk so they cannot drift apart. The + // second is the complement of the skip that `decode_delta_binary_kernel` applies per page: if it + // ever came out false while some page still needed level decoding, that page would be decoded by + // neither kernel. + auto has_flat_nz_map = false; + auto needs_level_decoder = false; + std::for_each(subpass.pages.host_begin(), subpass.pages.host_end(), [&](PageInfo const& page) { + auto const mapped = _nz_map_enabled && page.has_nz_map(nz_map_kind::DELTA_FLAT); + has_flat_nz_map |= mapped; + needs_level_decoder |= + BitAnd(page.kernel_mask, decode_kernel_mask::DELTA_BINARY) != 0 && not mapped; + }); // The producer and the consumer that reads its output share a stream, so the other decoders // overlap with them rather than queueing behind them. int const nkernels = std::bitset<32>(kernel_mask).count(); auto streams = cudf::detail::fork_streams(_stream, nkernels + (has_flat_nz_map ? 1 : 0)); - auto const nz_map_stream = streams.back(); + // Only forked when there is a map to produce. Without one there is nothing to order, and + // `decode_delta_binary` ignores the stream, so this stands in rather than aliasing a decoder's. + auto const nz_map_stream = has_flat_nz_map ? streams.back() : _stream; if (has_flat_nz_map) { precompute_flat_nz_map(subpass.pages, @@ -325,6 +334,7 @@ void reader_impl::decode_page_data(read_mode mode, size_t skip_rows, size_t num_ level_type_size, subpass_page_mask_span(), has_flat_nz_map, + needs_level_decoder, error_code.data(), delta_stream, nz_map_stream); diff --git a/cpp/src/io/parquet/reader_impl_preprocess.cu b/cpp/src/io/parquet/reader_impl_preprocess.cu index b431254b6c98..15eb0a182349 100644 --- a/cpp/src/io/parquet/reader_impl_preprocess.cu +++ b/cpp/src/io/parquet/reader_impl_preprocess.cu @@ -428,7 +428,14 @@ void reader_impl::allocate_level_decode_space() auto const& chunk = pass.chunks[page.chunk_idx]; state->nz_count = page_nz_map::not_yet_produced; if (chunk.max_level[level_type::DEFINITION] != 0) { - flat_map_sizes[idx] = static_cast(page.num_input_values) * sizeof(uint32_t); + // The producer never indexes past `min(num_input_values, num_decoded_level_values)`, which + // is set above and is the smaller of the two for a sliced read. A page whose levels are + // entirely outside the row range decodes none, so it gets no slice at all; the producer + // takes that page's `def == nullptr` path and reports a zero count, and the consumer never + // reaches the map. + flat_map_sizes[idx] = + static_cast(std::min(page.num_input_values, page.num_decoded_level_values)) * + sizeof(uint32_t); flat_nz_map_size += flat_map_sizes[idx]; } }