diff --git a/cpp/benchmarks/CMakeLists.txt b/cpp/benchmarks/CMakeLists.txt index 1b57e3b23666..d03bf72ee9e8 100644 --- a/cpp/benchmarks/CMakeLists.txt +++ b/cpp/benchmarks/CMakeLists.txt @@ -296,8 +296,8 @@ ConfigureNVBench( # ################################################################################################## # * parquet reader benchmark ---------------------------------------------------------------------- ConfigureNVBench( - PARQUET_READER_NVBENCH io/parquet/parquet_reader_input.cpp io/parquet/parquet_reader_options.cpp - io/parquet/reader_common.cpp + PARQUET_READER_NVBENCH io/parquet/parquet_reader_input.cpp io/parquet/parquet_reader_encoding.cpp + io/parquet/parquet_reader_options.cpp io/parquet/reader_common.cpp ) # ################################################################################################## diff --git a/cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp b/cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp new file mode 100644 index 000000000000..fdedf02a137c --- /dev/null +++ b/cpp/benchmarks/io/parquet/parquet_reader_encoding.cpp @@ -0,0 +1,103 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include "reader_common.hpp" + +#include +#include +#include + +#include +#include + +#include + +#include +#include +#include +#include + +// Benchmarks decoding pages written with an explicitly requested column encoding. The writer's +// defaults never choose the DELTA_* encodings, so `parquet_read_decode` does not exercise their +// decode kernels; this benchmark covers them (with PLAIN as the baseline encoding). + +namespace { + +cudf::io::column_encoding retrieve_column_encoding_enum(std::string_view encoding_string) +{ + if (encoding_string == "PLAIN") { return cudf::io::column_encoding::PLAIN; } + if (encoding_string == "DELTA_BINARY_PACKED") { + return cudf::io::column_encoding::DELTA_BINARY_PACKED; + } + if (encoding_string == "DELTA_LENGTH_BYTE_ARRAY") { + return cudf::io::column_encoding::DELTA_LENGTH_BYTE_ARRAY; + } + if (encoding_string == "DELTA_BYTE_ARRAY") { return cudf::io::column_encoding::DELTA_BYTE_ARRAY; } + CUDF_FAIL("Unsupported column encoding: " + std::string(encoding_string)); +} + +void bench_read_encoding(nvbench::state& state, std::vector const& d_types) +{ + auto const encoding = retrieve_column_encoding_enum(state.get_string("encoding")); + auto const source_type = retrieve_io_type_enum(state.get_string("io_type")); + auto const data_size = static_cast(state.get_int64("data_size")); + auto const cardinality = static_cast(state.get_int64("cardinality")); + auto const run_length = static_cast(state.get_int64("run_length")); + cuio_source_sink_pair source_sink(source_type); + + auto const num_rows_written = [&]() { + auto const tbl = create_random_table( + cycle_dtypes(d_types, num_cols), + table_size_bytes{data_size}, + data_profile_builder().cardinality(cardinality).avg_run_length(run_length)); + auto const view = tbl->view(); + + cudf::io::table_input_metadata metadata(view); + for (auto& col_meta : metadata.column_metadata) { + col_meta.set_encoding(encoding); + } + + cudf::io::parquet_writer_options write_opts = + cudf::io::parquet_writer_options::builder(source_sink.make_sink_info(), view) + .metadata(std::move(metadata)) + .compression(cudf::io::compression_type::NONE) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .write_v2_headers(true); + cudf::io::write_parquet(write_opts); + return view.num_rows(); + }(); + + parquet_read_common(num_rows_written, num_cols, source_sink, state); +} + +} // namespace + +void BM_parquet_read_delta_binary(nvbench::state& state) +{ + bench_read_encoding(state, {cudf::type_id::INT32, cudf::type_id::INT64}); +} + +void BM_parquet_read_delta_string(nvbench::state& state) +{ + bench_read_encoding(state, {cudf::type_id::STRING}); +} + +NVBENCH_BENCH(BM_parquet_read_delta_binary) + .set_name("parquet_read_delta_binary") + .add_string_axis("encoding", {"PLAIN", "DELTA_BINARY_PACKED"}) + .add_string_axis("io_type", {"DEVICE_BUFFER"}) + .set_min_samples(4) + .add_int64_axis("cardinality", {0, 1000}) + .add_int64_axis("run_length", {1, 32}) + .add_int64_axis("data_size", {512 << 20}); + +NVBENCH_BENCH(BM_parquet_read_delta_string) + .set_name("parquet_read_delta_string") + .add_string_axis("encoding", {"PLAIN", "DELTA_LENGTH_BYTE_ARRAY", "DELTA_BYTE_ARRAY"}) + .add_string_axis("io_type", {"DEVICE_BUFFER"}) + .set_min_samples(4) + .add_int64_axis("cardinality", {0, 1000}) + .add_int64_axis("run_length", {1, 32}) + .add_int64_axis("data_size", {512 << 20}); diff --git a/cpp/src/io/parquet/compact_protocol_writer.cpp b/cpp/src/io/parquet/compact_protocol_writer.cpp index d354015b1d47..e3de1594a22f 100644 --- a/cpp/src/io/parquet/compact_protocol_writer.cpp +++ b/cpp/src/io/parquet/compact_protocol_writer.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2018-2025, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -263,6 +263,56 @@ size_t CompactProtocolWriter::write(SortingColumn const& sc) return c.value(); } +size_t CompactProtocolWriter::write(DataPageHeader const& pg_hdr) +{ + CompactProtocolFieldWriter c(*this); + c.field_int(1, pg_hdr.num_values); + c.field_int(2, static_cast(pg_hdr.encoding)); + c.field_int(3, static_cast(pg_hdr.definition_level_encoding)); + c.field_int(4, static_cast(pg_hdr.repetition_level_encoding)); + return c.value(); +} + +size_t CompactProtocolWriter::write(DictionaryPageHeader const& dict_pg_hdr) +{ + CompactProtocolFieldWriter c(*this); + c.field_int(1, dict_pg_hdr.num_values); + c.field_int(2, static_cast(dict_pg_hdr.encoding)); + return c.value(); +} + +size_t CompactProtocolWriter::write(DataPageHeaderV2 const& pg_hdr_v2) +{ + CompactProtocolFieldWriter c(*this); + c.field_int(1, pg_hdr_v2.num_values); + c.field_int(2, pg_hdr_v2.num_nulls); + c.field_int(3, pg_hdr_v2.num_rows); + c.field_int(4, static_cast(pg_hdr_v2.encoding)); + c.field_int(5, pg_hdr_v2.definition_levels_byte_length); + c.field_int(6, pg_hdr_v2.repetition_levels_byte_length); + c.field_bool(7, pg_hdr_v2.is_compressed); + return c.value(); +} + +size_t CompactProtocolWriter::write(PageHeader const& page_hdr) +{ + CompactProtocolFieldWriter c(*this); + c.field_int(1, static_cast(page_hdr.type)); + c.field_int(2, page_hdr.uncompressed_page_size); + c.field_int(3, page_hdr.compressed_page_size); + // Exactly one page-specific header is set, selected by `type`; field ids (5/7/8) match + // CompactProtocolReader::read(PageHeader) and the GPU encoder in gpuEncodePageHeaders. + switch (page_hdr.type) { + case PageType::DATA_PAGE: c.field_struct(5, page_hdr.data_page_header); break; + case PageType::DICTIONARY_PAGE: c.field_struct(7, page_hdr.dictionary_page_header); break; + case PageType::DATA_PAGE_V2: c.field_struct(8, page_hdr.data_page_header_v2); break; + default: + CUDF_FAIL("Trying to write an invalid PageType " + + std::to_string(static_cast(page_hdr.type))); + } + return c.value(); +} + void CompactProtocolFieldWriter::put_byte(uint8_t v) { writer.m_buf.push_back(v); } void CompactProtocolFieldWriter::put_byte(uint8_t const* raw, uint32_t len) diff --git a/cpp/src/io/parquet/compact_protocol_writer.hpp b/cpp/src/io/parquet/compact_protocol_writer.hpp index 47dcc9480783..b89d54914540 100644 --- a/cpp/src/io/parquet/compact_protocol_writer.hpp +++ b/cpp/src/io/parquet/compact_protocol_writer.hpp @@ -1,11 +1,12 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2018-2025, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2018-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ #pragma once #include +#include #include #include @@ -13,7 +14,8 @@ #include #include -namespace cudf::io::parquet::detail { +namespace CUDF_EXPORT cudf { +namespace io::parquet::detail { /** * @brief Class for parsing Parquet's Thrift Compact Protocol encoded metadata @@ -43,6 +45,10 @@ class CompactProtocolWriter { size_t write(ColumnOrder const&); size_t write(PageEncodingStats const&); size_t write(SortingColumn const&); + size_t write(DataPageHeader const&); + size_t write(DictionaryPageHeader const&); + size_t write(DataPageHeaderV2 const&); + size_t write(PageHeader const&); protected: std::vector& m_buf; @@ -117,4 +123,5 @@ template <> inline void CompactProtocolFieldWriter::field_int_list(int field, std::vector const& val); -} // namespace cudf::io::parquet::detail +} // namespace io::parquet::detail +} // namespace CUDF_EXPORT cudf diff --git a/cpp/src/io/parquet/delta_binary.cuh b/cpp/src/io/parquet/delta_binary.cuh index 6dac50ce80bd..a9c234a87b0f 100644 --- a/cpp/src/io/parquet/delta_binary.cuh +++ b/cpp/src/io/parquet/delta_binary.cuh @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2023-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2023-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -7,6 +7,11 @@ #include "page_decode.cuh" +#include +#include + +#include + namespace cudf::io::parquet::detail { // DELTA_XXX encoding support @@ -29,16 +34,24 @@ namespace cudf::io::parquet::detail { // block to ensure that all encoded values are positive. The deltas for each mini-block are bit // packed using the same encoding as the RLE/Bit-Packing Hybrid encoder. -// The largest mini-block size we can currently support. -constexpr int max_delta_mini_block_size = 64; - -// The first pass decodes `values_per_mb` values, and then the second pass does another -// batch of size `values_per_mb`. The largest value for values_per_miniblock among the -// major writers seems to be 64, so 2 * 64 should be good. We save the first value separately -// since it is not encoded in the first mini-block. -// The extra 1 is for the first value, from the block header. It's not stored in the buffer, but it -// still impacts buffer indexing and we need to account for it to avoid race conditions. -constexpr int delta_rolling_buf_size = (2 * max_delta_mini_block_size) + 1; +// The DELTA_BINARY_PACKED spec requires the number of values in a mini-block to be a multiple of +// 32. The decoders rely on the coincidence that this also equals warp size; they produce values +// in warp_size-wide passes, so it must divide every spec-valid mini-block size. +constexpr int delta_mini_block_size_multiple = 32; +static_assert(delta_mini_block_size_multiple % cudf::detail::warp_size == 0, + "warp_size must divide the DELTA mini-block size multiple; the pass-based decoders " + "assume warp_size divides every spec-valid mini-block size"); + +// The decode loops produce up to two (warp_size-wide) passes per iteration: pages whose +// mini-blocks hold at least two passes keep the two-pass batch the loops have always used, and +// running several passes back to back amortizes the per-iteration synchronization. +constexpr int delta_max_batch_size = 2 * cudf::detail::warp_size; + +// The rolling buffer must hold two batches in flight (the consumer drains one batch while the +// producer decodes the next), plus one slot for the first value from the block header: it is not +// stored in the buffer, but it still impacts buffer indexing and we need to account for it to +// avoid race conditions. +constexpr int delta_rolling_buf_size = (2 * delta_max_batch_size) + 1; /** * @brief Read a ULEB128 varint integer @@ -86,6 +99,7 @@ struct delta_binary_decoder { uint32_t values_per_mb; // block_size / mini_block_count, must be multiple of 32 uint32_t current_value_idx; // current value index, initialized to 0 at start of block + uint32_t cur_pass; // current pass within the mini-block zigzag128_t cur_min_delta; // min delta for the block uint32_t cur_mb; // index of the current mini-block within the block @@ -110,6 +124,13 @@ struct delta_binary_decoder { return value_count == 0 ? 0 : all_values ? value_count : value_count - 1; } + // index just past the values decode_next_pass() has produced so far (0 before the first pass, + // even though the header value already occupies index 0) + __device__ uint32_t next_pass_start_idx() + { + return current_value_idx + cur_pass * cudf::detail::warp_size; + } + // read mini-block header into state object. should only be called from init_binary_block or // setup_next_mini_block. header format is: // @@ -149,10 +170,14 @@ struct delta_binary_decoder { last_value = first_value; current_value_idx = 0; + cur_pass = 0; error = false; - // Validate header against the DELTA_BINARY_PACKED spec invariants - if (mini_block_count == 0 or block_size == 0 or (block_size % mini_block_count) != 0) { + // Validate the header against the DELTA_BINARY_PACKED spec: the mini-block count must evenly + // divide the block size, and each mini-block must hold a multiple of 32 values. The decoders + // rely on the latter to advance from one mini-block to the next. + if (mini_block_count == 0 or block_size == 0 or (block_size % mini_block_count) != 0 or + ((block_size / mini_block_count) % delta_mini_block_size_multiple) != 0) { error = true; value_count = 0; values_per_mb = 1; @@ -184,12 +209,12 @@ struct delta_binary_decoder { // just set pointer to start of next mini_block if (cur_mb < mini_block_count - 1) { - cur_mb_start += cur_bitwidths[cur_mb] * values_per_mb / 8; + cur_mb_start += cur_bitwidths[cur_mb] * values_per_mb / CHAR_BIT; cur_mb++; } // out of mini-blocks, start a new block else { - block_start = cur_mb_start + cur_bitwidths[cur_mb] * values_per_mb / 8; + block_start = cur_mb_start + cur_bitwidths[cur_mb] * values_per_mb / CHAR_BIT; init_mini_block(is_decode); } } @@ -216,111 +241,117 @@ struct delta_binary_decoder { return new_end; } - // decode the current mini-batch of deltas, and convert to values. - // called by all threads in a warp, currently only one warp supported. - inline __device__ void calc_mini_block_values(int lane_id) + // account for the first value from the block header before the first mini-block is decoded. + // the first value is not encoded in the mini-block data, but it still occupies index 0 of the + // value stream. returns true if there are more values to decode after the header value. + // called by all threads in a single warp `warp`. + inline __device__ bool advance_past_first_value( + cg::thread_block_tile const& warp) { - using cudf::detail::warp_size; - if (current_value_idx >= value_count) { return; } + if (current_value_idx >= value_count) { return false; } - // need to account for the first value from header on first pass if (current_value_idx == 0) { // make sure all threads access current_value_idx above before incrementing - __syncwarp(); - if (lane_id == 0) { current_value_idx++; } - __syncwarp(); - if (current_value_idx >= value_count) { return; } + warp.sync(); + if (warp.thread_rank() == 0) { current_value_idx++; } + warp.sync(); + if (current_value_idx >= value_count) { return false; } } + return true; + } + + // decode a single warp_size-wide pass (indexed by `pass`) of the current mini-block and convert + // the deltas to values (see decode_next_pass). called by all threads in a single warp `warp`. + inline __device__ void calc_mini_block_pass( + uint32_t pass, cg::thread_block_tile const& warp) + { + using cudf::detail::warp_size; + auto const lane_id = static_cast(warp.thread_rank()); uint32_t const mb_bits = cur_bitwidths[cur_mb]; - // need to do in multiple passes if values_per_mb != 32 - uint32_t const num_pass = values_per_mb / warp_size; - - auto d_start = cur_mb_start; - - for (int i = 0; i < num_pass; i++) { - // position at end of the current mini-block since the following calculates - // negative indexes - d_start += (warp_size * mb_bits) / 8; - - // unpack deltas. modified from version in decode_dictionary_indices(), but - // that one only unpacks up to bitwidths of 24. simplified some since this - // will always do batches of 32. - // NOTE: because this needs to handle up to 64 bits, the branching used in the other - // implementation has been replaced with a loop. While this uses more registers, the - // looping version is just as fast and easier to read. Might need to revisit this when - // DELTA_BYTE_ARRAY is implemented. - zigzag128_t delta = 0; - if (lane_id + current_value_idx < value_count) { - int32_t ofs = (lane_id - warp_size) * mb_bits; - uint8_t const* p = d_start + (ofs >> 3); - ofs &= 7; - if (p < block_end) { - uint32_t c = 8 - ofs; // 0 - 7 bits - delta = (*p++) >> ofs; - - while (c < mb_bits && p < block_end) { - delta |= static_cast(*p++) << c; - c += 8; - } - delta &= (static_cast(1) << mb_bits) - 1; + // position at the end of this pass's values since the following calculates negative indexes + auto const d_start = cur_mb_start + (pass + 1) * (warp_size * mb_bits / CHAR_BIT); + + // unpack deltas. modified from version in decode_dictionary_indices(), but + // that one only unpacks up to bitwidths of 24. simplified some since this + // will always do batches of 32. + // NOTE: because this needs to handle up to 64 bits, the branching used in the other + // implementation has been replaced with a loop. While this uses more registers, the + // looping version is just as fast and easier to read. + zigzag128_t delta = 0; + if (current_value_idx + pass * warp_size + lane_id < value_count) { + // ofs is non-positive, so the arithmetic shift and mask compute the byte offset and leading + // bit position as floored division/modulo by CHAR_BIT (a plain / and % would round + // toward 0) + int32_t ofs = (lane_id - warp_size) * mb_bits; + uint8_t const* p = d_start + (ofs >> 3); + ofs &= 7; + if (p < block_end) { + uint32_t c = CHAR_BIT - ofs; // 0 - 7 bits + delta = (*p++) >> ofs; + + while (c < mb_bits && p < block_end) { + delta |= static_cast(*p++) << c; + c += CHAR_BIT; } + delta &= (static_cast(1) << mb_bits) - 1; } - - // add min delta to get true delta - delta += cur_min_delta; - - // do inclusive scan to get value - first_value at each position - __shared__ cub::WarpScan::TempStorage temp_storage; - cub::WarpScan(temp_storage).InclusiveSum(delta, delta); - - // now add first value from header or last value from previous block to get true value - delta += last_value; - int const value_idx = - rolling_index(current_value_idx + warp_size * i + lane_id); - value[value_idx] = delta; - - // save value from last lane in warp. this will become the 'first value' added to the - // deltas calculated in the next iteration (or invocation). - if (lane_id == warp_size - 1) { last_value = delta; } - __syncwarp(); } + + // add min delta to get true delta + delta += cur_min_delta; + + // do inclusive scan to get value - first_value at each position. cg::inclusive_scan is + // shuffle-based and carries no shared storage, so any number of delta decoders (e.g. the + // prefix and suffix decoder warps of the DELTA_BYTE_ARRAY kernels) can run it concurrently, + // each over its own warp tile, with no risk of aliasing. + delta = cg::inclusive_scan(warp, delta, cg::plus{}); + + // now add first value from header or last value from previous pass to get true value + delta += last_value; + int const value_idx = + rolling_index(current_value_idx + warp_size * pass + lane_id); + value[value_idx] = delta; + + // save value from last lane in warp. this will become the 'first value' added to the + // deltas calculated in the next pass (or invocation). + if (lane_id == warp_size - 1) { last_value = delta; } + warp.sync(); } - // decodes and skips values until the block containing the value after `skip` is reached. - // called by all threads in a thread block. - inline __device__ void skip_values(int skip) + // decodes and discards values so the decoder resumes at the pass boundary at or just past + // `skip`. the up to warp_size - 1 values decoded beyond `skip` stay resident in the rolling + // buffer for the consumer, which resumes reading at `skip`. works for any mini-block size. + // called by all threads in a thread block (`block`); the decode runs on warp 0 (`warp`). + inline __device__ void skip_values( + int skip, + cg::thread_block const& block, + cg::thread_block_tile const& warp) { - using cudf::detail::warp_size; - int const t = threadIdx.x; - int const lane_id = t % warp_size; - - while (current_value_idx < skip && current_value_idx < num_encoded_values(true)) { - // calc_mini_block_values only runs in warp 0, but writes to current_value_idx, - // so everyone must sync before we diverge - __syncthreads(); - if (t < warp_size) { - calc_mini_block_values(lane_id); - if (lane_id == 0) { setup_next_mini_block(true); } - } - __syncthreads(); + while (next_pass_start_idx() < static_cast(skip) && + current_value_idx < num_encoded_values(true)) { + // decode_next_pass only runs in warp 0, but advances decoder state everyone reads, + // so everyone must sync around it + block.sync(); + if (warp.meta_group_rank() == 0) { decode_next_pass(warp); } + block.sync(); } } - // Decodes and skips values until the block containing the value after `skip` is reached. - // Keeps a running sum of the values and returns that upon exit. Called by all threads in a - // warp 0. Result is only valid on thread 0. - // This is intended for use only by the DELTA_LENGTH_BYTE_ARRAY decoder. - inline __device__ size_t skip_values_and_sum(int skip) + // Decodes and skips values until the pass containing `skip` has been decoded, keeping a + // running sum of the skipped values (indices below `skip`) and returning it. Values decoded + // beyond `skip` stay resident in the rolling buffer for the consumer. Works for any + // mini-block size. Called by all threads in warp 0 (`warp`); the result is only valid on + // thread 0. This is intended for use only by the DELTA_LENGTH_BYTE_ARRAY decoder. + inline __device__ size_t skip_values_and_sum( + int skip, cg::thread_block_tile const& warp) { using cudf::detail::warp_size; // DELTA_LENGTH_BYTE_ARRAY lengths are encoded as INT32 by convention (since the PLAIN encoding // uses 4-byte lengths). using delta_length_type = int32_t; - using warp_reduce = cub::WarpReduce; - __shared__ warp_reduce::TempStorage temp_storage; - int const t = threadIdx.x; + auto const t = warp.thread_rank(); // initialize sum with first value, which is stored in the block header. cast to // `delta_length_type` to ensure the value is interpreted properly before promoting it @@ -330,40 +361,48 @@ struct delta_binary_decoder { // if only skipping one value, we're done already if (skip == 1) { return sum; } - // need to do in multiple passes if values_per_mb != 32 - uint32_t const num_pass = values_per_mb / warp_size; - - while (current_value_idx < skip && current_value_idx < num_encoded_values(true)) { - calc_mini_block_values(t); - - int const idx = current_value_idx + t; - - for (uint32_t p = 0; p < num_pass; p++) { - auto const pidx = idx + p * warp_size; - size_t const val = pidx < skip ? static_cast(value_at(pidx)) : 0; - auto const warp_sum = warp_reduce(temp_storage).Sum(val); - if (t == 0) { sum += warp_sum; } - } - if (t == 0) { setup_next_mini_block(true); } - __syncwarp(); + while (next_pass_start_idx() < static_cast(skip) && + current_value_idx < num_encoded_values(true)) { + // the pass decoded below produces indices [pass_first, pass_first + warp_size); the + // header value at index 0 is not part of any pass and is already in `sum` + auto const pass_first = max(next_pass_start_idx(), 1u); + decode_next_pass(warp); + + auto const idx = pass_first + t; + size_t const val = idx < static_cast(skip) && idx < value_count + ? static_cast(value_at(idx)) + : 0; + auto const warp_sum = cg::reduce(warp, val, cg::plus{}); + if (t == 0) { sum += warp_sum; } + warp.sync(); } return sum; } - // decodes the current mini block and stores the values obtained. should only be called by - // a single warp. - inline __device__ void decode_batch() + // decode the next warp_size-wide pass of the current mini-block into db->value, advancing to + // the next mini-block once all of its passes have been decoded. Decoding a single pass at a + // time keeps the rolling buffer footprint independent of the mini-block size. Should only be + // called by a single warp `warp`. NOTE: lane 0's state updates are not synchronized on exit; + // the caller must synchronize the warp (or block) before the next call so all lanes observe + // them. + inline __device__ void decode_next_pass( + cg::thread_block_tile const& warp) { using cudf::detail::warp_size; - int const t = threadIdx.x; - int const lane_id = t % warp_size; - // unpack deltas and save in db->value - calc_mini_block_values(lane_id); + if (not advance_past_first_value(warp)) { return; } - // set up for next mini-block - if (lane_id == 0) { setup_next_mini_block(true); } + // unpack one pass of deltas and save in db->value + calc_mini_block_pass(cur_pass, warp); + + // advance within the mini-block; move to the next mini-block once all passes are decoded + if (warp.thread_rank() == 0) { + if (++cur_pass == values_per_mb / warp_size) { + cur_pass = 0; + setup_next_mini_block(true); + } + } } }; diff --git a/cpp/src/io/parquet/page_delta_decode.cu b/cpp/src/io/parquet/page_delta_decode.cu index b73c8b88ab61..e7210d9a3ca5 100644 --- a/cpp/src/io/parquet/page_delta_decode.cu +++ b/cpp/src/io/parquet/page_delta_decode.cu @@ -24,6 +24,12 @@ namespace cg = cooperative_groups; constexpr int decode_block_size = 128; constexpr int decode_delta_binary_block_size = 96; +// Size of the ring buffer that maps leaf-value ordinals to output rows (nz_idx). The level +// decoder runs up to two batches ahead of the value consumer and, on nested pages, overshoots +// its target by up to a warp of values, so this needs to exceed 3 * delta_max_batch_size + +// 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; + // 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 @@ -33,7 +39,10 @@ struct delta_byte_array_decoder { uint8_t const* last_string; // pointer to last decoded string...needed for its prefix uint8_t const* suffix_char_data; // pointer to the start of character data - uint8_t* temp_buf; // buffer used when skipping values + uint8_t* temp_buf; // scratch for strings skipped over by a leading row range; the next + // batch overwrites it from its start each round + uint8_t* prefix_seed; // one reserved slot ahead of temp_buf holding a durable copy of the + // last decoded string, used to seed the next batch's first prefix uint32_t start_val; // decoded strings up to this index will be dumped to temp_buf uint32_t last_string_len; // length of the last decoded string @@ -41,13 +50,18 @@ struct delta_byte_array_decoder { delta_binary_decoder suffixes; // state of decoder for suffix lengths // initialize the prefixes and suffixes blocks - __device__ void init(uint8_t const* start, uint8_t const* end, uint32_t start_idx, uint8_t* temp) + __device__ void init( + uint8_t const* start, uint8_t const* end, uint32_t start_idx, uint8_t* temp, size_t temp_size) { auto const* suffix_start = prefixes.find_end_of_block(start, end); suffix_char_data = suffixes.find_end_of_block(suffix_start, end); last_string = nullptr; - temp_buf = temp; - start_val = start_idx; + // the temp allocation holds one leading string slot (see the string-size prepass) followed by + // delta_max_batch_size scratch slots. reserve the leading slot for the last decoded string so + // it stays clear of the scratch, which each round overwrites from its start. + prefix_seed = temp; + temp_buf = temp + temp_size / (delta_max_batch_size + 1); + start_val = start_idx; } // kind of like an inclusive scan for strings. takes prefix_len bytes from preceding @@ -197,6 +211,16 @@ struct delta_byte_array_decoder { __syncwarp(); } + // the next batch overwrites the temp scratch from its start, so if the last decoded string + // lives there, preserve it in the reserved seed slot ahead of the scratch + if (end_idx <= start_val && last_string != prefix_seed) { + if (lane_id == 0) { + memcpy(prefix_seed, last_string, last_string_len); + last_string = prefix_seed; + } + __syncwarp(); + } + return string_total; } @@ -252,45 +276,51 @@ struct delta_byte_array_decoder { __syncwarp(); } + // the next batch overwrites the temp scratch from its start, so if the last decoded string + // lives there, preserve it in the reserved seed slot ahead of the scratch + if (end_idx <= start_val && last_string != prefix_seed) { + if (lane_id == 0) { + memcpy(prefix_seed, last_string, last_string_len); + last_string = prefix_seed; + } + __syncwarp(); + } + return string_total; } - // dump strings before start_val to temp buf - __device__ void skip(bool use_char_ll) + // dump strings before start_val to temp buf. decodes one warp_size-wide pass per round, so + // any mini-block size is supported. called by all threads in a thread block. + __device__ void skip(bool use_char_ll, + cg::thread_block const& block, + cg::thread_block_tile const& warp) { using cudf::detail::warp_size; - int const t = threadIdx.x; - int const lane_id = t % warp_size; // is this even necessary? return if asking to skip the whole block. if (start_val >= prefixes.num_encoded_values(true)) { return; } - // prefixes and suffixes will have the same parameters (it's checked earlier) - auto const batch_size = prefixes.values_per_mb; - uint32_t skip_pos = 0; - while (prefixes.current_value_idx < start_val) { - // warp 0 gets prefixes and warp 1 gets suffixes - auto* const db = t < 32 ? &prefixes : &suffixes; - - // this will potentially decode past start_val, but that's ok - if (t < 64) { db->decode_batch(); } - __syncthreads(); - - // warp 0 decodes the batch. - if (t < 32) { - auto const num_to_decode = min(batch_size, start_val - skip_pos); - auto const bytes_written = - use_char_ll ? calculate_string_values_cp(temp_buf, skip_pos, num_to_decode, lane_id) - : calculate_string_values(temp_buf, skip_pos, num_to_decode, lane_id); - // store last_string someplace safe in temp buffer - if (t == 0) { - memcpy(temp_buf + bytes_written, last_string, last_string_len); - last_string = temp_buf + bytes_written; + while (skip_pos < start_val) { + // warp 0 decodes a pass of prefixes and warp 1 a pass of suffixes. this will potentially + // decode past start_val, and those values stay resident in the rolling buffers for the + // decode loop that follows. + auto* const db = warp.meta_group_rank() == 0 ? &prefixes : &suffixes; + if (warp.meta_group_rank() < 2) { db->decode_next_pass(warp); } + block.sync(); + + // warp 0 reconstructs this round's skipped strings into the temp scratch (the helpers + // preserve the round's last string past the scratch area for the next round's prefixes) + if (warp.meta_group_rank() == 0) { + auto const num_to_decode = min(static_cast(warp_size), start_val - skip_pos); + if (use_char_ll) { + calculate_string_values_cp(temp_buf, skip_pos, num_to_decode, warp.thread_rank()); + } else { + calculate_string_values(temp_buf, skip_pos, num_to_decode, warp.thread_rank()); } } - skip_pos += prefixes.values_per_mb; - __syncthreads(); + skip_pos += warp_size; + block.sync(); } } }; @@ -310,7 +340,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) { __shared__ __align__(16) delta_binary_decoder db_state; __shared__ __align__(16) full_page_decode_state state_g; - __shared__ __align__(16) page_state_buffers_s state_buffers; + __shared__ __align__(16) page_state_buffers_s state_buffers; auto* const s = &state_g; auto* const sb = &state_buffers; @@ -360,8 +390,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) if (block.thread_rank() == 0) { db->init_binary_block(s->stream.data_start, s->stream.data_end); } block.sync(); - auto const batch_size = db->values_per_mb; - if (db->error or batch_size > max_delta_mini_block_size) { + if (db->error) { if (block.thread_rank() == 0) { set_error(static_cast(decode_error::DELTA_PARAMS_UNSUPPORTED), error_code); @@ -369,9 +398,21 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) return; } + bool const is_skip_resume = skipped_leaf_values > 0; + + // Number of values produced per main-loop iteration: up to two warp_size passes, so pages whose + // mini-blocks hold at least two passes keep the schedule of the whole-mini-block decoder. When + // resuming after skip_values() the producer emits a single pass per iteration: the skip leaves + // up to warp_size not-yet-consumed values in the rolling buffer, and a larger batch could wrap + // around and overwrite them before the consumer reads them. + 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 skipped_leaf_values is non-zero, then we need to decode up to the first mini-block // that has a value we need. - if (skipped_leaf_values > 0) { db->skip_values(skipped_leaf_values); } + if (is_skip_resume) { db->skip_values(skipped_leaf_values, block, warp); } while (s->setup.error == 0 && (s->progress.input_value_count < s->setup.num_input_values || s->progress.src_pos < s->progress.nz_count)) { @@ -395,10 +436,15 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) // - update validity vectors // - updates offsets (for nested columns) // - produces non-NULL value indices in s->nz_idx for subsequent decoding - gpuDecodeLevels(s, sb, target_pos, rep, def, warp); + gpuDecodeLevels(s, sb, target_pos, rep, def, warp); } else if (warp.meta_group_rank() == 1) { // warp 1 - db->decode_batch(); + for (uint32_t i = 0; i < passes_per_batch; i++) { + // make lane 0's state updates from the previous pass visible to the whole warp; the + // block-wide sync below covers the last pass of the iteration + if (i > 0) { warp.sync(); } + db->decode_next_pass(warp); + } } else if (src_pos < target_pos) { // warp 2 // nesting level that is storing actual leaf values @@ -408,7 +454,7 @@ CUDF_KERNEL void __launch_bounds__(decode_delta_binary_block_size) for (uint32_t sp = src_pos + warp.thread_rank(); sp < src_pos + batch_size; sp += warp.size()) { // the position in the output column/buffer - int32_t dst_pos = sb->nz_idx[rolling_index(sp)]; + int32_t dst_pos = sb->nz_idx[rolling_index(sp)]; // handle skip_rows here. flat hierarchies can just skip up to first_row. if (!has_repetition) { dst_pos -= s->setup.first_row; } @@ -467,7 +513,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) { __shared__ __align__(16) delta_byte_array_decoder db_state; __shared__ __align__(16) full_page_decode_state state_g; - __shared__ __align__(16) page_state_buffers_s state_buffers; + __shared__ __align__(16) page_state_buffers_s state_buffers; auto* const s = &state_g; auto* const sb = &state_buffers; @@ -529,7 +575,8 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) dba->init(s->stream.data_start, s->stream.data_end, s->setup.page.start_val, - s->setup.page.temp_string_buf); + s->setup.page.temp_string_buf, + s->setup.page.temp_string_size); } block.sync(); @@ -555,22 +602,21 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) int const leaf_level_index = s->setup.col.max_nesting_depth - 1; auto strings_data = nesting_info_base[leaf_level_index].string_out; - // sanity check to make sure we can process this page - auto const batch_size = prefix_db->values_per_mb; - if (batch_size > max_delta_mini_block_size) { - if (block.thread_rank() == 0) { - set_error(static_cast(decode_error::DELTA_PARAMS_UNSUPPORTED), - error_code); - } - return; - } - // if this is a bounds page and nested, then we need to skip up front. non-nested will work // its way through the page. int string_pos = has_repetition ? s->setup.page.start_val : 0; auto const is_bounds_pg = is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); - if (is_bounds_pg && string_pos > 0) { dba->skip(use_char_ll); } + bool const is_skip_resume = is_bounds_pg and string_pos > 0; + + // Number of values produced per main-loop iteration (see decode_delta_binary_kernel for why + // skip-resume pages must produce a single warp_size pass per iteration). + uint32_t const batch_size = + is_skip_resume ? cudf::detail::warp_size + : min(prefix_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) { dba->skip(use_char_ll, block, warp); } while (!s->setup.error && (s->progress.input_value_count < s->setup.num_input_values || s->progress.src_pos < s->progress.nz_count)) { @@ -594,13 +640,21 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // - update validity vectors // - updates offsets (for nested columns) // - produces non-NULL value indices in s->nz_idx for subsequent decoding - gpuDecodeLevels(s, sb, target_pos, rep, def, warp); + gpuDecodeLevels(s, sb, target_pos, rep, def, warp); } else if (warp.meta_group_rank() == 1) { // warp 1 - prefix_db->decode_batch(); + for (uint32_t i = 0; i < passes_per_batch; i++) { + // make lane 0's state updates from the previous pass visible to the whole warp; the + // block-wide sync below covers the last pass of the iteration + if (i > 0) { warp.sync(); } + prefix_db->decode_next_pass(warp); + } } else if (warp.meta_group_rank() == 2) { // warp 2 - suffix_db->decode_batch(); + for (uint32_t i = 0; i < passes_per_batch; i++) { + if (i > 0) { warp.sync(); } + suffix_db->decode_next_pass(warp); + } } else if (warp.meta_group_rank() == 3 and src_pos < target_pos) { // warp 3 int const nproc = min(batch_size, s->setup.page.end_val - string_pos); @@ -614,7 +668,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) for (uint32_t sp = src_pos + warp.thread_rank(); sp < src_pos + batch_size; sp += warp.size()) { // the position in the output column/buffer - int dst_pos = sb->nz_idx[rolling_index(sp)]; + int dst_pos = sb->nz_idx[rolling_index(sp)]; // handle skip_rows here. flat hierarchies can just skip up to first_row. if (!has_repetition) { dst_pos -= s->setup.first_row; } @@ -681,7 +735,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) { __shared__ __align__(16) delta_binary_decoder db_state; __shared__ __align__(16) full_page_decode_state state_g; - __shared__ __align__(16) page_state_buffers_s state_buffers; + __shared__ __align__(16) page_state_buffers_s state_buffers; __shared__ __align__(8) uint8_t const* page_string_data; __shared__ size_t string_offset; @@ -741,9 +795,9 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) } block.sync(); - // sanity check to make sure we can process this page - auto const batch_size = db->values_per_mb; - if (db->error or batch_size > max_delta_mini_block_size) { + // The decode loop below sub-batches each mini-block into warp_size-wide passes, so any mini-block + // size is supported (see decode_next_pass). + if (db->error) { if (block.thread_rank() == 0) { set_error(static_cast(decode_error::DELTA_PARAMS_UNSUPPORTED), error_code); @@ -760,10 +814,21 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // string data block. auto const is_bounds_pg = is_bounds_page(s->setup.page, s->setup.col.start_row, min_row, num_rows, has_repetition); - if (is_bounds_pg && s->setup.page.start_val > 0) { + bool const is_skip_resume = is_bounds_pg and s->setup.page.start_val > 0; + + // Only nested pages resume the decoder mid-page; flat pages re-init it below and can keep the + // full batch. Mid-page resumption must produce a single warp_size pass per iteration (see + // decode_delta_binary_kernel for why). + bool const resumes_mid_page = is_skip_resume and has_repetition; + uint32_t const batch_size = + resumes_mid_page ? cudf::detail::warp_size + : min(db->values_per_mb, static_cast(delta_max_batch_size)); + uint32_t const passes_per_batch = batch_size / cudf::detail::warp_size; + + if (is_skip_resume) { if (warp.meta_group_rank() == 0) { // string_off is only valid on thread 0 - auto const string_off = db->skip_values_and_sum(s->setup.page.start_val); + auto const string_off = db->skip_values_and_sum(s->setup.page.start_val, warp); // Threads in the warp might diverge and read in skip_values_and_sum // after lane 0 reinits below. warp.sync(); @@ -803,10 +868,15 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) // - update validity vectors // - updates offsets (for nested columns) // - produces non-NULL value indices in s->nz_idx for subsequent decoding - gpuDecodeLevels(s, sb, target_pos, rep, def, warp); + gpuDecodeLevels(s, sb, target_pos, rep, def, warp); } else if (warp.meta_group_rank() == 1) { // warp 1 - db->decode_batch(); + for (uint32_t i = 0; i < passes_per_batch; i++) { + // make lane 0's state updates from the previous pass visible to the whole warp; the + // block-wide sync below covers the last pass of the iteration + if (i > 0) { warp.sync(); } + db->decode_next_pass(warp); + } } else if (warp.meta_group_rank() == 2 && src_pos < target_pos) { // warp 2 int const nproc = min(batch_size, s->setup.page.end_val - string_pos); @@ -816,7 +886,7 @@ CUDF_KERNEL void __launch_bounds__(decode_block_size) for (uint32_t sp = src_pos + warp.thread_rank(); sp < src_pos + batch_size; sp += warp.size()) { // the position in the output column/buffer - int dst_pos = sb->nz_idx[rolling_index(sp)]; + int dst_pos = sb->nz_idx[rolling_index(sp)]; // handle skip_rows here. flat hierarchies can just skip up to first_row. if (!has_repetition) { dst_pos -= s->setup.first_row; } diff --git a/cpp/src/io/parquet/page_string_decode.cu b/cpp/src/io/parquet/page_string_decode.cu index 9633bb0b50ac..fb316bb5a208 100644 --- a/cpp/src/io/parquet/page_string_decode.cu +++ b/cpp/src/io/parquet/page_string_decode.cu @@ -14,6 +14,7 @@ #include #include +#include #include #include #include @@ -411,25 +412,24 @@ __device__ cuda::std::pair totalDeltaByteArraySize(uint8_t const int end_value) { using cudf::detail::warp_size; - using WarpReduce = cub::WarpReduce; - - __shared__ typename WarpReduce::TempStorage temp_storage[2]; __shared__ __align__(16) delta_binary_decoder prefixes; __shared__ __align__(16) delta_binary_decoder suffixes; - int const t = threadIdx.x; - int const lane_id = t % warp_size; - int const warp_id = t / warp_size; + auto const block = cg::this_thread_block(); + auto const warp = cg::tiled_partition(block); + int const t = block.thread_rank(); + int const lane_id = warp.thread_rank(); if (t == 0) { auto const* suffix_start = prefixes.find_end_of_block(data, end); suffixes.init_binary_block(suffix_start, end); } - __syncthreads(); + block.sync(); // two warps will traverse the prefixes and suffixes and sum them up - auto const db = t < warp_size ? &prefixes : t < 2 * warp_size ? &suffixes : nullptr; + auto const warp_id = warp.meta_group_rank(); + auto const db = (warp_id == 0) ? &prefixes : warp_id == 1 ? &suffixes : nullptr; size_t total_bytes = 0; uleb128_t max_len = 0; @@ -442,12 +442,17 @@ __device__ cuda::std::pair totalDeltaByteArraySize(uint8_t const uleb128_t lane_max = 0; while (db->current_value_idx < end_value && db->current_value_idx < db->num_encoded_values(true)) { - // calculate values for current mini-block - db->calc_mini_block_values(lane_id); + if (not db->advance_past_first_value(warp)) { break; } - // get per lane sum for mini-block - for (uint32_t i = 0; i < db->values_per_mb; i += 32) { - uint32_t const idx = db->current_value_idx + i + lane_id; + // decode one warp_size-wide pass at a time and read it back immediately: a whole + // mini-block is only fully resident in the rolling buffer while it fits, but a single + // pass always is + uint32_t const num_pass = db->values_per_mb / warp_size; + for (uint32_t p = 0; p < num_pass; p++) { + db->calc_mini_block_pass(p, warp); + + // get per lane sum for this pass + uint32_t const idx = db->current_value_idx + p * warp_size + lane_id; if (idx >= start_value && idx < end_value && idx < db->value_count) { lane_sum += db->value[rolling_index(idx)]; } @@ -458,31 +463,30 @@ __device__ cuda::std::pair totalDeltaByteArraySize(uint8_t const } if (lane_id == 0) { db->setup_next_mini_block(true); } - __syncwarp(); + warp.sync(); } - // get sum for warp. - // note: warp_sum will only be valid on lane 0. - auto const warp_sum = WarpReduce(temp_storage[warp_id]).Sum(lane_sum); - __syncwarp(); - auto const warp_max = WarpReduce(temp_storage[warp_id]).Reduce(lane_max, cuda::maximum{}); + // get sum and max for warp (valid on all lanes; only lane 0 is consumed below). + auto const warp_sum = cg::reduce(warp, lane_sum, cg::plus{}); + auto const warp_max = cg::reduce(warp, lane_max, cg::greater{}); if (lane_id == 0) { total_bytes += warp_sum; max_len = warp_max; } } - __syncthreads(); + block.sync(); // now sum up total_bytes from the two warps auto const final_bytes = cudf::detail::single_lane_block_sum_reduce(total_bytes); - // Sum up prefix and suffix max lengths to get a max possible string length. Multiply that - // by the number of strings in a mini-block, plus one to save the last string. + // Sum up prefix and suffix max lengths to get a max possible string length. Multiply that by + // the largest number of strings a decode-loop iteration can stage, plus one to save the last + // string. auto const temp_bytes = cudf::detail::single_lane_block_sum_reduce(max_len) * - (db->values_per_mb + 1); + (delta_max_batch_size + 1); return {final_bytes, temp_bytes}; } @@ -635,11 +639,8 @@ CUDF_KERNEL void __launch_bounds__(delta_preproc_block_size) // only need temp space if we're skipping values if (start_value > 0) { - // just need to parse the header of the first delta binary block to get values_per_mb - delta_binary_decoder db; - db.init_binary_block(s->stream.data_start, s->stream.data_end); - // save enough for one mini-block plus some extra to save the last_string - pp->temp_string_size = s->output_cvt.dtype_len_in * (db.values_per_mb + 1); + // save enough for one decode batch plus some extra to save the last_string + pp->temp_string_size = s->output_cvt.dtype_len_in * (delta_max_batch_size + 1); } } } else { @@ -695,11 +696,10 @@ CUDF_KERNEL void __launch_bounds__(delta_length_block_size) size_t num_rows) { using cudf::detail::warp_size; - using WarpReduce = cub::WarpReduce; - __shared__ typename WarpReduce::TempStorage temp_storage; __shared__ __align__(16) string_size_scan_state state_g; __shared__ __align__(16) delta_binary_decoder string_lengths; + auto const warp = cg::tiled_partition(cg::this_thread_block()); auto* const s = &state_g; int const page_idx = blockIdx.x; int const t = threadIdx.x; @@ -754,7 +754,7 @@ CUDF_KERNEL void __launch_bounds__(delta_length_block_size) auto const end_value = pp->end_val; if (t == 0) { string_lengths.init_binary_block(s->stream.data_start, s->stream.data_end); } - __syncwarp(); + warp.sync(); size_t total_bytes = 0; @@ -766,24 +766,29 @@ CUDF_KERNEL void __launch_bounds__(delta_length_block_size) uleb128_t lane_sum = 0; while (string_lengths.current_value_idx < end_value && string_lengths.current_value_idx < string_lengths.num_encoded_values(true)) { - // calculate values for current mini-block - string_lengths.calc_mini_block_values(t); + if (not string_lengths.advance_past_first_value(warp)) { break; } + + // decode one warp_size-wide pass at a time and read it back immediately: a whole + // mini-block is only fully resident in the rolling buffer while it fits, but a single + // pass always is + uint32_t const num_pass = string_lengths.values_per_mb / warp_size; + for (uint32_t p = 0; p < num_pass; p++) { + string_lengths.calc_mini_block_pass(p, warp); - // get per lane sum for mini-block - for (uint32_t i = 0; i < string_lengths.values_per_mb; i += warp_size) { - uint32_t const idx = string_lengths.current_value_idx + i + t; + // get per lane sum for this pass + uint32_t const idx = string_lengths.current_value_idx + p * warp_size + t; if (idx >= start_value && idx < end_value && idx < string_lengths.value_count) { lane_sum += string_lengths.value[rolling_index(idx)]; } } if (t == 0) { string_lengths.setup_next_mini_block(true); } - __syncwarp(); + warp.sync(); } // get sum for warp. // note: warp_sum will only be valid on lane 0. - auto const warp_sum = WarpReduce(temp_storage).Sum(lane_sum); + auto const warp_sum = cg::reduce(warp, lane_sum, cg::plus{}); if (t == 0) { total_bytes += warp_sum; diff --git a/cpp/tests/io/parquet_delta_test_utils.hpp b/cpp/tests/io/parquet_delta_test_utils.hpp new file mode 100644 index 000000000000..9a1058a3b4c5 --- /dev/null +++ b/cpp/tests/io/parquet_delta_test_utils.hpp @@ -0,0 +1,698 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#pragma once + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +// Builders for single-page Parquet files with DELTA-family encodings, used to test mini-block +// sizes no stock writer emits (the cudf and parquet-mr writers put 32 values in a mini-block, +// pyarrow and arrow-rs at most 64, while the format allows any multiple of 32). Callers pass +// the column values plus the DELTA block geometry (block_size, mini_block_count) and get back +// the complete file bytes. +// +// Both the page header and the file footer are serialized with cudf's production +// CompactProtocolWriter (the same CompactProtocolWriter::write() overloads the writer uses), so +// only the DELTA-encoded page body -- the part under test, which no stock writer emits -- is built +// by hand. + +// Parquet metadata (page header + footer) serialization via the production compact protocol writer +namespace parquet_delta_test { + +// serialize a V1 data page header for a flat REQUIRED column (no repetition/definition levels) +inline std::vector serialize_data_page_header(int num_values, + cudf::io::parquet::Encoding encoding, + int64_t page_size) +{ + namespace pq = cudf::io::parquet; + pq::PageHeader ph; + ph.type = pq::PageType::DATA_PAGE; + ph.uncompressed_page_size = static_cast(page_size); + ph.compressed_page_size = static_cast(page_size); + ph.data_page_header.num_values = num_values; + ph.data_page_header.encoding = encoding; + ph.data_page_header.definition_level_encoding = pq::Encoding::RLE; // no levels + ph.data_page_header.repetition_level_encoding = pq::Encoding::RLE; // no levels + std::vector out; + pq::detail::CompactProtocolWriter{&out}.write(ph); + return out; +} + +// serialize a V2 data page header (the LIST builders carry bit-packed repetition/definition levels) +inline std::vector serialize_data_page_header_v2(int num_values, + int num_nulls, + int num_rows, + cudf::io::parquet::Encoding encoding, + int64_t definition_levels_byte_length, + int64_t repetition_levels_byte_length, + int64_t page_size) +{ + namespace pq = cudf::io::parquet; + pq::PageHeader ph; + ph.type = pq::PageType::DATA_PAGE_V2; + ph.uncompressed_page_size = static_cast(page_size); + ph.compressed_page_size = static_cast(page_size); + auto& v2 = ph.data_page_header_v2; + v2.num_values = num_values; + v2.num_nulls = num_nulls; + v2.num_rows = num_rows; + v2.encoding = encoding; + v2.definition_levels_byte_length = static_cast(definition_levels_byte_length); + v2.repetition_levels_byte_length = static_cast(repetition_levels_byte_length); + v2.is_compressed = false; + std::vector out; + pq::detail::CompactProtocolWriter{&out}.write(ph); + return out; +} + +// serialize a FileMetaData footer with the production CompactProtocolWriter +inline std::vector serialize_footer(cudf::io::parquet::FileMetaData const& file_metadata) +{ + std::vector out; + cudf::io::parquet::detail::CompactProtocolWriter writer(&out); + writer.write(file_metadata); + return out; +} + +} // namespace parquet_delta_test + +// DELTA_BINARY_PACKED stream encoder + +// pack values (padded with 0 up to `count`) at `width` bits each, LSB-first, consecutively -- +// the same layout the RLE/bit-packing hybrid and the delta mini-blocks use +inline void bitpack_into(std::vector& out, + std::vector const& vals, + int width, + int count) +{ + size_t const base = out.size(); + out.resize(base + static_cast(count) * width / 8, 0); + size_t pos = 0; + for (auto const v : vals) { + for (int b = 0; b < width; b++) { + if ((v >> b) & 1) { out[base + pos / 8] |= 1 << (pos % 8); } + pos++; + } + } +} + +// append `v` to `out` as an unsigned LEB128 varint +inline void append_uleb128(std::vector& out, uint64_t v) +{ + while (true) { + uint8_t const b = v & 0x7f; + v >>= 7; + if (v) { + out.push_back(b | 0x80); + } else { + out.push_back(b); + return; + } + } +} + +// append `v` to `out` as a zigzag-encoded LEB128 varint +inline void append_zigzag128(std::vector& out, int64_t v) +{ + append_uleb128(out, (static_cast(v) << 1) ^ static_cast(v >> 63)); +} + +// complete DELTA_BINARY_PACKED stream: header (block_size, mini_block_count, value count, first +// value), then per block a zigzag min-delta, one bit-width byte per mini-block, and the +// bit-packed deltas +inline std::vector encode_delta_binary_packed(std::vector const& values, + int block_size, + int mini_block_count) +{ + CUDF_EXPECTS(block_size % mini_block_count == 0 && (block_size / mini_block_count) % 32 == 0, + "DELTA mini-block size (block_size / mini_block_count) must be a multiple of 32"); + std::vector out; + append_uleb128(out, block_size); + append_uleb128(out, mini_block_count); + append_uleb128(out, values.size()); // total value count, including the first value below + append_zigzag128(out, values.empty() ? 0 : values.front()); + if (values.size() <= 1) { return out; } + + std::vector deltas(values.size() - 1); + for (size_t i = 0; i + 1 < values.size(); i++) { + deltas[i] = values[i + 1] - values[i]; + } + + int const vpm = block_size / mini_block_count; + for (size_t bstart = 0; bstart < deltas.size(); bstart += block_size) { + auto const bend = std::min(bstart + block_size, deltas.size()); + auto const min_delta = *std::min_element(deltas.begin() + bstart, deltas.begin() + bend); + append_zigzag128(out, min_delta); + + // per mini-block bit widths, then the packed deltas (empty trailing mini-blocks get width 0 + // and no data) + std::vector widths(mini_block_count, 0); + std::vector> rel(mini_block_count); + for (int m = 0; m < mini_block_count; m++) { + auto const mstart = bstart + static_cast(m) * vpm; + auto const mend = std::min(mstart + vpm, bend); + for (size_t i = mstart; i < mend; i++) { + auto const r = static_cast(deltas[i] - min_delta); // >= 0 by construction + rel[m].push_back(r); + int w = 0; + while (r >> w) { + w++; + } + widths[m] = std::max(widths[m], w); + } + } + for (auto const w : widths) { + out.push_back(static_cast(w)); + } + for (int m = 0; m < mini_block_count; m++) { + if (!rel[m].empty()) { bitpack_into(out, rel[m], widths[m], vpm); } + } + } + return out; +} + +// raw DELTA_BINARY_PACKED header only (block/mini-block geometry, value count, first value), for +// negative tests that need a geometry encode_delta_binary_packed would reject. The reader validates +// the geometry in init_binary_block before decoding any deltas, so no mini-block data follows. +inline std::vector encode_delta_binary_header(int block_size, + int mini_block_count, + int64_t value_count, + int64_t first_value) +{ + std::vector out; + append_uleb128(out, static_cast(block_size)); + append_uleb128(out, static_cast(mini_block_count)); + append_uleb128(out, static_cast(value_count)); + append_zigzag128(out, first_value); + return out; +} + +// single-page file assembly + +// V1 data page + footer around `body` for a single REQUIRED flat column "a" +inline std::vector wrap_single_page_parquet(std::vector const& body, + int num_values, + cudf::io::parquet::Type physical_type, + cudf::io::parquet::Encoding encoding, + bool utf8) +{ + namespace pq = cudf::io::parquet; + + auto const page_header = + parquet_delta_test::serialize_data_page_header(num_values, encoding, body.size()); + + int const data_page_offset = 4; // immediately after the leading "PAR1" magic + auto const chunk_size = static_cast(page_header.size() + body.size()); + + pq::FileMetaData file_metadata; + file_metadata.version = 1; + file_metadata.num_rows = num_values; + + // schema: root group with a single REQUIRED leaf "a". The metadata structs are built in place + // (emplace_back + reference) so that structs holding std::optional members are never copied + // through an initializer_list, which trips GCC's -Wmaybe-uninitialized on the empty optionals. + file_metadata.schema.reserve(2); + auto& root = file_metadata.schema.emplace_back(); + root.name = "schema"; + root.num_children = 1; + root.repetition_type = pq::FieldRepetitionType::UNSPECIFIED; // the root carries no repetition + auto& col = file_metadata.schema.emplace_back(); + col.type = physical_type; + col.repetition_type = pq::FieldRepetitionType::REQUIRED; + col.name = "a"; + if (utf8) { col.converted_type = pq::ConvertedType::UTF8; } + + auto& row_group = file_metadata.row_groups.emplace_back(); + row_group.total_byte_size = chunk_size; + row_group.num_rows = num_values; + auto& chunk = row_group.columns.emplace_back(); + chunk.file_offset = data_page_offset; + auto& meta = chunk.meta_data; + meta.type = physical_type; + meta.encodings = {pq::Encoding::RLE, encoding}; + meta.path_in_schema = {"a"}; + meta.codec = pq::Compression::UNCOMPRESSED; + meta.num_values = num_values; + meta.total_uncompressed_size = chunk_size; + meta.total_compressed_size = chunk_size; + meta.data_page_offset = data_page_offset; + + auto const footer = parquet_delta_test::serialize_footer(file_metadata); + + std::vector out; + auto append = [&out](auto const& bytes) { out.insert(out.end(), bytes.begin(), bytes.end()); }; + out.insert(out.end(), {'P', 'A', 'R', '1'}); + append(page_header); + append(body); + append(footer); + auto const flen = static_cast(footer.size()); + for (int i = 0; i < 4; i++) { + out.push_back((flen >> (8 * i)) & 0xff); + } + out.insert(out.end(), {'P', 'A', 'R', '1'}); + return out; +} + +// complete file: one DELTA_BINARY_PACKED INT64 column "a" +inline std::vector build_delta_binary_parquet(std::vector const& values, + int block_size, + int mini_block_count) +{ + auto const body = encode_delta_binary_packed(values, block_size, mini_block_count); + return wrap_single_page_parquet(body, + values.size(), + cudf::io::parquet::Type::INT64, + cudf::io::parquet::Encoding::DELTA_BINARY_PACKED, + false); +} + +// deterministic test data (self-contained splitmix64 so results never vary across platforms) +inline uint64_t delta_test_rand(uint64_t& state) +{ + state += 0x9e3779b97f4a7c15ull; + uint64_t z = state; + z = (z ^ (z >> 30)) * 0xbf58476d1ce4e5b9ull; + z = (z ^ (z >> 27)) * 0x94d049bb133111ebull; + return z ^ (z >> 31); +} + +// values whose deltas vary within [-2, regime_max], with the magnitude regime switching every 64 +// values so consecutive mini-blocks get different, non-zero bit widths +inline std::vector delta_test_int64_values(int n, uint64_t seed = 101) +{ + constexpr int64_t regime_max[] = {5, 220, 3000, 60000}; + std::vector out(n); + int64_t v = 0; + for (int i = 0; i < n; i++) { + out[i] = v; + auto const hi = regime_max[(i / 64) % 4]; + v += static_cast(delta_test_rand(seed) % static_cast(hi + 3)) - 2; + } + return out; +} + +// string encodings + +// strings mixing ASCII and valid non-ASCII UTF-8 sequences, with lengths varying in +// [1, max_length]; with shared_prefixes, each string keeps a random-length prefix of its +// predecessor so the DELTA_BYTE_ARRAY prefix-length stream also has varying non-zero deltas. The +// multi-byte code points exercise the byte-wise length/prefix/suffix reconstruction paths. +inline std::vector delta_test_strings(int n, + bool shared_prefixes, + uint64_t seed = 201, + size_t max_length = 20) +{ + constexpr char alphabet[] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + // 2- and 3-byte UTF-8 code points (e-acute, n-tilde, u-umlaut, euro, CJK, lambda) written as + // explicit bytes so the encoding does not depend on the compiler's execution character set + constexpr std::string_view utf8_tokens[] = { + "\xc3\xa9", "\xc3\xb1", "\xc3\xbc", "\xe2\x82\xac", "\xe4\xb8\xad", "\xce\xbb"}; + std::vector out; + out.reserve(n); + std::string prev; + for (int i = 0; i < n; i++) { + auto const length = 1 + static_cast(delta_test_rand(seed) % max_length); + std::string s; + if (shared_prefixes && !prev.empty()) { + auto const keep = delta_test_rand(seed) % (std::min(prev.size(), length) + 1); + s = prev.substr(0, keep); + } + while (s.size() < length) { + auto const r = delta_test_rand(seed); + auto const remaining = length - s.size(); + // roughly one position in four appends a multi-byte token when it fits; the rest stay ASCII. + // appending only whole tokens that fit keeps the byte length exactly `length`. + if (remaining >= 2 && r % 4 == 0) { + auto const& token = utf8_tokens[(r >> 2) % (sizeof(utf8_tokens) / sizeof(utf8_tokens[0]))]; + if (token.size() <= remaining) { + s += token; + continue; + } + } + s += alphabet[r % (sizeof(alphabet) - 1)]; + } + out.push_back(s); + prev = std::move(s); + } + return out; +} + +// complete file: one DELTA_LENGTH_BYTE_ARRAY string column "a" (delta-encoded lengths followed +// by the concatenated string bytes) +inline std::vector build_delta_length_byte_array_parquet( + std::vector const& strings, int block_size, int mini_block_count) +{ + std::vector lengths(strings.size()); + std::transform(strings.begin(), strings.end(), lengths.begin(), [](auto const& s) { + return static_cast(s.size()); + }); + auto body = encode_delta_binary_packed(lengths, block_size, mini_block_count); + for (auto const& s : strings) { + body.insert(body.end(), s.begin(), s.end()); + } + return wrap_single_page_parquet(body, + strings.size(), + cudf::io::parquet::Type::BYTE_ARRAY, + cudf::io::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY, + true); +} + +// complete file: one DELTA_BYTE_ARRAY string column "a" (front compression: delta-encoded +// shared-prefix lengths, then delta-encoded suffix lengths, then the concatenated suffixes) +inline std::vector build_delta_byte_array_parquet(std::vector const& strings, + int block_size, + int mini_block_count) +{ + std::vector prefix_lens, suffix_lens; + std::string suffix_bytes; + std::string prev; + for (auto const& s : strings) { + size_t lcp = 0; + auto const end = std::min(prev.size(), s.size()); + while (lcp < end && prev[lcp] == s[lcp]) { + lcp++; + } + prefix_lens.push_back(lcp); + suffix_lens.push_back(s.size() - lcp); + suffix_bytes.append(s, lcp, std::string::npos); + prev = s; + } + auto body = encode_delta_binary_packed(prefix_lens, block_size, mini_block_count); + auto const suffix_stream = encode_delta_binary_packed(suffix_lens, block_size, mini_block_count); + body.insert(body.end(), suffix_stream.begin(), suffix_stream.end()); + body.insert(body.end(), suffix_bytes.begin(), suffix_bytes.end()); + return wrap_single_page_parquet(body, + strings.size(), + cudf::io::parquet::Type::BYTE_ARRAY, + cudf::io::parquet::Encoding::DELTA_BYTE_ARRAY, + true); +} + +// LIST: one optional list column "col" of optional int64 "element" (max_def_level 3, +// max_rep_level 1), no null lists or elements -- empty lists only. Emitted as a single +// uncompressed V2 data page whose rep/def levels are RLE/bit-packed hybrid runs. + +// encode `levels` at `width` bits as one bit-packed hybrid run (padded to a multiple of 8) +inline std::vector encode_levels_bit_packed(std::vector const& levels, int width) +{ + auto const groups = (levels.size() + 7) / 8; + std::vector out; + // RLE/bit-pack hybrid run header: LSB set marks a bit-packed run of `groups` 8-value groups + append_uleb128(out, (groups << 1) | 1); + std::vector vals(levels.begin(), levels.end()); + bitpack_into(out, vals, width, groups * 8); + return out; +} + +// per-value repetition/definition levels for a list column with the shape given by `sizes` +// (max_def_level 3, max_rep_level 1, no null lists or elements -- empty lists only) +struct list_levels { + std::vector rep; + std::vector def; + int num_nulls = 0; +}; + +inline list_levels compute_list_levels(std::vector const& sizes) +{ + list_levels out; + for (auto const size : sizes) { + if (size == 0) { // empty list: one level entry, def < max_def, no leaf value + out.rep.push_back(0); + out.def.push_back(1); + out.num_nulls++; + continue; + } + for (size_t j = 0; j < size; j++) { + out.rep.push_back(j == 0 ? 0 : 1); + out.def.push_back(3); + } + } + return out; +} + +// V2 data page + footer around the encoded `values` of a single list column "col" of leaf +// "element" with the given physical type and encoding +inline std::vector wrap_single_page_list_parquet( + list_levels const& levels, + int num_rows, + std::vector const& values, + cudf::io::parquet::Type leaf_physical_type, + cudf::io::parquet::Encoding encoding, + bool utf8) +{ + namespace pq = cudf::io::parquet; + + auto const num_values = static_cast(levels.rep.size()); + auto const rep = encode_levels_bit_packed(levels.rep, 1); // max_rep_level 1 + auto const dfn = encode_levels_bit_packed(levels.def, 2); // max_def_level 3 + auto const page_size = static_cast(rep.size() + dfn.size() + values.size()); + + auto const page_header = parquet_delta_test::serialize_data_page_header_v2( + num_values, levels.num_nulls, num_rows, encoding, dfn.size(), rep.size(), page_size); + + int const data_page_offset = 4; + auto const chunk_size = static_cast(page_header.size()) + page_size; + + pq::FileMetaData file_metadata; + file_metadata.version = 2; + file_metadata.num_rows = num_rows; + + // schema: root -> optional LIST group "col" -> repeated group "list" -> optional leaf "element". + // The metadata structs are built in place (emplace_back + reference) so that structs holding + // std::optional members are never copied through an initializer_list, which trips GCC's + // -Wmaybe-uninitialized on the empty optionals. + file_metadata.schema.reserve(4); + auto& root = file_metadata.schema.emplace_back(); + root.name = "schema"; + root.num_children = 1; + root.repetition_type = pq::FieldRepetitionType::UNSPECIFIED; + auto& list_col = file_metadata.schema.emplace_back(); + list_col.repetition_type = pq::FieldRepetitionType::OPTIONAL; + list_col.name = "col"; + list_col.num_children = 1; + list_col.converted_type = pq::ConvertedType::LIST; + list_col.logical_type = pq::LogicalType{pq::LogicalType::LIST}; + auto& list_group = file_metadata.schema.emplace_back(); + list_group.repetition_type = pq::FieldRepetitionType::REPEATED; + list_group.name = "list"; + list_group.num_children = 1; + auto& element = file_metadata.schema.emplace_back(); + element.type = leaf_physical_type; + element.repetition_type = pq::FieldRepetitionType::OPTIONAL; + element.name = "element"; + if (utf8) { element.converted_type = pq::ConvertedType::UTF8; } + + auto& row_group = file_metadata.row_groups.emplace_back(); + row_group.total_byte_size = chunk_size; + row_group.num_rows = num_rows; + auto& chunk = row_group.columns.emplace_back(); + chunk.file_offset = data_page_offset; + auto& meta = chunk.meta_data; + meta.type = leaf_physical_type; + meta.encodings = {pq::Encoding::RLE, encoding}; + meta.path_in_schema = {"col", "list", "element"}; + meta.codec = pq::Compression::UNCOMPRESSED; + meta.num_values = num_values; // counts level entries incl. empties + meta.total_uncompressed_size = chunk_size; + meta.total_compressed_size = chunk_size; + meta.data_page_offset = data_page_offset; + + auto const footer = parquet_delta_test::serialize_footer(file_metadata); + + std::vector out; + auto append = [&out](auto const& bytes) { out.insert(out.end(), bytes.begin(), bytes.end()); }; + out.insert(out.end(), {'P', 'A', 'R', '1'}); + append(page_header); + append(rep); + append(dfn); + append(values); + append(footer); + auto const flen = static_cast(footer.size()); + for (int i = 0; i < 4; i++) { + out.push_back((flen >> (8 * i)) & 0xff); + } + out.insert(out.end(), {'P', 'A', 'R', '1'}); + return out; +} + +inline std::vector build_delta_binary_list_parquet( + std::vector> const& lists, int block_size, int mini_block_count) +{ + std::vector sizes(lists.size()); + std::vector leaf_values; + for (size_t i = 0; i < lists.size(); i++) { + sizes[i] = lists[i].size(); + leaf_values.insert(leaf_values.end(), lists[i].begin(), lists[i].end()); + } + auto const values = encode_delta_binary_packed(leaf_values, block_size, mini_block_count); + return wrap_single_page_list_parquet(compute_list_levels(sizes), + lists.size(), + values, + cudf::io::parquet::Type::INT64, + cudf::io::parquet::Encoding::DELTA_BINARY_PACKED, + false); +} + +// per-value repetition/definition levels for a list column whose leaves may be null (definition +// level 2). std::nullopt is a null leaf element; an empty inner vector is an empty list (definition +// level 1). Same LIST shape as compute_list_levels (max_def_level 3, max_rep_level 1). +inline list_levels compute_list_levels_with_leaf_nulls( + std::vector>> const& lists) +{ + list_levels out; + for (auto const& list : lists) { + if (list.empty()) { // empty list: one level entry, def < max_def, no leaf value + out.rep.push_back(0); + out.def.push_back(1); + out.num_nulls++; + continue; + } + for (size_t j = 0; j < list.size(); j++) { + out.rep.push_back(j == 0 ? 0 : 1); + out.def.push_back(list[j].has_value() ? 3 : 2); // 3: leaf present, 2: null leaf element + if (not list[j].has_value()) { out.num_nulls++; } + } + } + return out; +} + +// complete file: one LIST column whose leaves may be null. Only the non-null leaf values are +// DELTA_BINARY_PACKED encoded; the null leaves are carried by the definition levels. +inline std::vector build_delta_binary_list_with_leaf_nulls_parquet( + std::vector>> const& lists, + int block_size, + int mini_block_count) +{ + std::vector leaf_values; + for (auto const& list : lists) { + for (auto const& e : list) { + if (e.has_value()) { leaf_values.push_back(*e); } + } + } + auto const values = encode_delta_binary_packed(leaf_values, block_size, mini_block_count); + return wrap_single_page_list_parquet(compute_list_levels_with_leaf_nulls(lists), + lists.size(), + values, + cudf::io::parquet::Type::INT64, + cudf::io::parquet::Encoding::DELTA_BINARY_PACKED, + false); +} + +// complete file: one LIST column, leaf strings DELTA_LENGTH_BYTE_ARRAY encoded +inline std::vector build_delta_length_byte_array_list_parquet( + std::vector> const& lists, int block_size, int mini_block_count) +{ + std::vector sizes(lists.size()); + std::vector lengths; + std::string chars; + for (size_t i = 0; i < lists.size(); i++) { + sizes[i] = lists[i].size(); + for (auto const& s : lists[i]) { + lengths.push_back(s.size()); + chars += s; + } + } + auto body = encode_delta_binary_packed(lengths, block_size, mini_block_count); + body.insert(body.end(), chars.begin(), chars.end()); + return wrap_single_page_list_parquet(compute_list_levels(sizes), + lists.size(), + body, + cudf::io::parquet::Type::BYTE_ARRAY, + cudf::io::parquet::Encoding::DELTA_LENGTH_BYTE_ARRAY, + true); +} + +// complete file: one LIST column, leaf strings DELTA_BYTE_ARRAY (front compression) +// encoded over the flattened string sequence +inline std::vector build_delta_byte_array_list_parquet( + std::vector> const& lists, int block_size, int mini_block_count) +{ + std::vector sizes(lists.size()); + std::vector prefix_lens, suffix_lens; + std::string suffix_bytes; + std::string prev; + for (size_t i = 0; i < lists.size(); i++) { + sizes[i] = lists[i].size(); + for (auto const& s : lists[i]) { + size_t lcp = 0; + auto const end = std::min(prev.size(), s.size()); + while (lcp < end && prev[lcp] == s[lcp]) { + lcp++; + } + prefix_lens.push_back(lcp); + suffix_lens.push_back(s.size() - lcp); + suffix_bytes.append(s, lcp, std::string::npos); + prev = s; + } + } + auto body = encode_delta_binary_packed(prefix_lens, block_size, mini_block_count); + auto const suffix_stream = encode_delta_binary_packed(suffix_lens, block_size, mini_block_count); + body.insert(body.end(), suffix_stream.begin(), suffix_stream.end()); + body.insert(body.end(), suffix_bytes.begin(), suffix_bytes.end()); + return wrap_single_page_list_parquet(compute_list_levels(sizes), + lists.size(), + body, + cudf::io::parquet::Type::BYTE_ARRAY, + cudf::io::parquet::Encoding::DELTA_BYTE_ARRAY, + true); +} + +// lists of alphanumeric strings with the same shape as delta_test_lists (varying lengths 1..8, +// empties mixed in, a trailing empty list); the flattened string sequence comes from +// delta_test_strings so prefix and suffix lengths vary +inline std::vector> delta_test_string_lists(int n_lists, + bool shared_prefixes, + uint64_t seed = 501, + size_t max_length = 20) +{ + std::vector lengths(n_lists); + size_t n_leaf = 0; + for (int i = 0; i < n_lists; i++) { + bool const empty = (i + 1 == n_lists) || delta_test_rand(seed) % 6 == 0; + lengths[i] = empty ? 0 : 1 + delta_test_rand(seed) % 8; + n_leaf += lengths[i]; + } + auto const strings = delta_test_strings(n_leaf, shared_prefixes, seed, max_length); + std::vector> out(n_lists); + size_t pos = 0; + for (int i = 0; i < n_lists; i++) { + out[i].assign(strings.begin() + pos, strings.begin() + pos + lengths[i]); + pos += lengths[i]; + } + return out; +} + +// lists of varying lengths 1..8 with empties mixed in (including a trailing empty list), leaf +// values from the varying-delta generator above +inline std::vector> delta_test_lists(int n_lists, uint64_t seed = 401) +{ + std::vector lengths(n_lists); + size_t n_leaf = 0; + for (int i = 0; i < n_lists; i++) { + bool const empty = (i + 1 == n_lists) || delta_test_rand(seed) % 6 == 0; + lengths[i] = empty ? 0 : 1 + delta_test_rand(seed) % 8; + n_leaf += lengths[i]; + } + auto const values = delta_test_int64_values(n_leaf, seed); + std::vector> out(n_lists); + size_t pos = 0; + for (int i = 0; i < n_lists; i++) { + out[i].assign(values.begin() + pos, values.begin() + pos + lengths[i]); + pos += lengths[i]; + } + return out; +} diff --git a/cpp/tests/io/parquet_reader_test.cpp b/cpp/tests/io/parquet_reader_test.cpp index 6ac067e74d3f..08e48dd213eb 100644 --- a/cpp/tests/io/parquet_reader_test.cpp +++ b/cpp/tests/io/parquet_reader_test.cpp @@ -7,6 +7,7 @@ #include "io/utilities/time_utils.hpp" #include "io_test_utils.hpp" #include "parquet_common.hpp" +#include "parquet_delta_test_utils.hpp" #include #include @@ -15,6 +16,7 @@ #include #include +#include #include #include #include @@ -34,6 +36,7 @@ #include #include #include +#include #include #include @@ -1098,6 +1101,383 @@ TEST_F(ParquetReaderTest, DecimalRead) } } +// Reading Parquet DELTA-encoded pages whose mini-blocks hold more than 64 values. No stock +// writer emits over 64 values/mini-block (cudf and parquet-mr write 32, pyarrow and arrow-rs at +// most 64), so the files are built in-memory by the parquet_delta_test_utils.hpp helpers. +namespace { + +cudf::io::parquet_reader_options delta_fixture_reader_options( + std::vector const& file_bytes, + int64_t skip_rows, + std::optional num_rows) +{ + auto builder = cudf::io::parquet_reader_options::builder( + cudf::io::source_info{cudf::host_span{ + reinterpret_cast(file_bytes.data()), file_bytes.size()}}); + if (skip_rows > 0) { builder.skip_rows(skip_rows); } + if (num_rows.has_value()) { builder.num_rows(*num_rows); } + return builder.build(); +} + +// read a DELTA_BINARY_PACKED file, optionally trimmed to [skip_rows, skip_rows + num_rows), and +// compare with the matching slice of `expected` +void delta_large_mini_block_read_test(std::vector const& file_bytes, + std::vector const& expected, + int64_t skip_rows = 0, + std::optional num_rows = std::nullopt) +{ + auto const result = + cudf::io::read_parquet(delta_fixture_reader_options(file_bytes, skip_rows, num_rows)); + auto const first = expected.begin() + skip_rows; + auto const last = num_rows.has_value() ? first + *num_rows : expected.end(); + auto const expected_col = cudf::test::fixed_width_column_wrapper(first, last); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(0), expected_col); +} + +// same as above for the DELTA_BYTE_ARRAY / DELTA_LENGTH_BYTE_ARRAY string files +void delta_large_mini_block_string_read_test(std::vector const& file_bytes, + std::vector const& expected, + int64_t skip_rows = 0, + std::optional num_rows = std::nullopt) +{ + auto const result = + cudf::io::read_parquet(delta_fixture_reader_options(file_bytes, skip_rows, num_rows)); + auto const first = expected.begin() + skip_rows; + auto const last = num_rows.has_value() ? first + *num_rows : expected.end(); + auto const expected_col = cudf::test::strings_column_wrapper(first, last); + CUDF_TEST_EXPECT_COLUMNS_EQUAL(result.tbl->view().column(0), expected_col); +} + +// same as above for the LIST files; the expected column is built from the lists and +// sliced to the requested row range +void delta_large_mini_block_list_read_test(std::vector const& file_bytes, + std::vector> const& expected, + int64_t skip_rows = 0, + std::optional num_rows = std::nullopt) +{ + auto const result = + cudf::io::read_parquet(delta_fixture_reader_options(file_bytes, skip_rows, num_rows)); + + std::vector offsets{0}; + std::vector leaf_values; + for (auto const& list : expected) { + leaf_values.insert(leaf_values.end(), list.begin(), list.end()); + offsets.push_back(leaf_values.size()); + } + auto offsets_col = + cudf::test::fixed_width_column_wrapper(offsets.begin(), offsets.end()); + auto child = + cudf::test::fixed_width_column_wrapper(leaf_values.begin(), leaf_values.end()); + auto const num_lists = static_cast(expected.size()); + auto const expected_col = cudf::make_lists_column( + num_lists, offsets_col.release(), child.release(), 0, rmm::device_buffer{}); + + auto const start = static_cast(skip_rows); + auto const end = num_rows.has_value() ? start + *num_rows : num_lists; + auto const expected_sliced = cudf::slice(expected_col->view(), {start, end}).front(); + CUDF_TEST_EXPECT_COLUMNS_EQUIVALENT(result.tbl->view().column(0), expected_sliced); +} + +// same as above for LIST files +void delta_large_mini_block_string_list_read_test( + std::vector const& file_bytes, + std::vector> const& expected, + int64_t skip_rows = 0, + std::optional num_rows = std::nullopt) +{ + auto const result = + cudf::io::read_parquet(delta_fixture_reader_options(file_bytes, skip_rows, num_rows)); + + std::vector offsets{0}; + std::vector leaf; + for (auto const& list : expected) { + leaf.insert(leaf.end(), list.begin(), list.end()); + offsets.push_back(leaf.size()); + } + auto offsets_col = + cudf::test::fixed_width_column_wrapper(offsets.begin(), offsets.end()); + auto child = cudf::test::strings_column_wrapper(leaf.begin(), leaf.end()); + auto const num_lists = static_cast(expected.size()); + auto const expected_col = cudf::make_lists_column( + num_lists, offsets_col.release(), child.release(), 0, rmm::device_buffer{}); + + auto const start = static_cast(skip_rows); + auto const end = num_rows.has_value() ? start + *num_rows : num_lists; + auto const expected_sliced = cudf::slice(expected_col->view(), {start, end}).front(); + CUDF_TEST_EXPECT_COLUMNS_EQUIVALENT(result.tbl->view().column(0), expected_sliced); +} + +} // namespace + +TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock128) +{ + // block_size=128, mini_block_count=1 -> 128 values/mini-block: the reader previously rejected + // mini-blocks over 64 values with DELTA_PARAMS_UNSUPPORTED (0x100). + auto const values = delta_test_int64_values(173); + delta_large_mini_block_read_test(build_delta_binary_parquet(values, 128, 1), values); +} + +TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock256) +{ + // block_size=256, mini_block_count=1 -> 256 values/mini-block (8 passes), beyond what + // simply raising the cap to 128 would cover. + auto const values = delta_test_int64_values(301); + delta_large_mini_block_read_test(build_delta_binary_parquet(values, 256, 1), values); +} + +TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlock96) +{ + // block_size=384, mini_block_count=4 -> 96 values/mini-block: exercises multiple + // mini-blocks per block (both the within-block and next-block advance paths) and a + // non-power-of-two pass count (3). + auto const values = delta_test_int64_values(141); + delta_large_mini_block_read_test(build_delta_binary_parquet(values, 384, 4), values); +} + +TEST_F(ParquetReaderTest, DeltaBinaryMiniBlock64) +{ + // block_size=256, mini_block_count=4 -> 64 values/mini-block: the mini-block size pyarrow and + // arrow-rs write for INT64, and the boundary the rolling buffer and the per-iteration batch cap + // are sized for. The in-repo writer emits 32, so nothing else covers it. + auto const values = delta_test_int64_values(157); + delta_large_mini_block_read_test(build_delta_binary_parquet(values, 256, 4), values); +} + +TEST_F(ParquetReaderTest, DeltaBinaryLargeMiniBlockSkipRows) +{ + // Row-range trims on flat columns are handled on the consumer side (first_row), so leading + // skips decode correctly for any mini-block size. + auto const v128 = delta_test_int64_values(173); + auto const f128 = build_delta_binary_parquet(v128, 128, 1); + delta_large_mini_block_read_test(f128, v128, 40); + delta_large_mini_block_read_test(f128, v128, 40, 50); + + auto const v256 = delta_test_int64_values(301); + delta_large_mini_block_read_test(build_delta_binary_parquet(v256, 256, 1), v256, 100, 100); + + auto const v96 = delta_test_int64_values(141); + delta_large_mini_block_read_test(build_delta_binary_parquet(v96, 384, 4), v96, 0, 60); + + auto const v64 = delta_test_int64_values(157); + delta_large_mini_block_read_test(build_delta_binary_parquet(v64, 256, 4), v64, 65, 64); +} + +TEST_F(ParquetReaderTest, DeltaByteArrayLargeMiniBlock96) +{ + // DELTA_BYTE_ARRAY pages with mini-blocks larger than 64 values. The string-size prepass + // reads decoded lengths back per warp-size pass, so any mini-block size is supported. + auto const strings = delta_test_strings(141, true); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(strings, 384, 4), strings); +} + +TEST_F(ParquetReaderTest, DeltaByteArrayLargeMiniBlock128) +{ + auto const strings = delta_test_strings(173, true); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(strings, 128, 1), strings); +} + +TEST_F(ParquetReaderTest, DeltaByteArrayLargeMiniBlock256) +{ + auto const strings = delta_test_strings(301, true); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(strings, 256, 1), strings); +} + +TEST_F(ParquetReaderTest, DeltaByteArrayLargeMiniBlockExactFill) +{ + // n = 257 exactly fills the single 256-value mini-block (plus the header value), the shape + // where a mis-sized string allocation causes out-of-bounds writes. + auto const strings = delta_test_strings(257, true); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(strings, 256, 1), strings); +} + +TEST_F(ParquetReaderTest, DeltaByteArrayLargeMiniBlockSkipRows) +{ + // Flat DELTA_BYTE_ARRAY bounds pages stage leading skipped strings through temp_string_buf + // inside the decode loop, so row-range reads work at any mini-block size. + auto const s256 = delta_test_strings(301, true); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(s256, 256, 1), s256, 100); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(s256, 256, 1), s256, 160); + + auto const s128 = delta_test_strings(173, true); + delta_large_mini_block_string_read_test( + build_delta_byte_array_parquet(s128, 128, 1), s128, 40, 60); + + auto const s96 = delta_test_strings(141, true); + delta_large_mini_block_string_read_test(build_delta_byte_array_parquet(s96, 384, 4), s96, 0, 70); +} + +TEST_F(ParquetReaderTest, DeltaLengthByteArrayLargeMiniBlock96) +{ + // DELTA_LENGTH_BYTE_ARRAY pages with mini-blocks larger than 64 values. + auto const strings = delta_test_strings(141, false); + delta_large_mini_block_string_read_test(build_delta_length_byte_array_parquet(strings, 384, 4), + strings); +} + +TEST_F(ParquetReaderTest, DeltaLengthByteArrayLargeMiniBlock128) +{ + auto const strings = delta_test_strings(173, false); + delta_large_mini_block_string_read_test(build_delta_length_byte_array_parquet(strings, 128, 1), + strings); +} + +TEST_F(ParquetReaderTest, DeltaLengthByteArrayLargeMiniBlock256) +{ + auto const strings = delta_test_strings(301, false); + delta_large_mini_block_string_read_test(build_delta_length_byte_array_parquet(strings, 256, 1), + strings); +} + +TEST_F(ParquetReaderTest, DeltaLengthByteArrayLargeMiniBlockTrimmed) +{ + // A num_rows-trimmed read makes the page a bounds page, which sizes the string output from + // the delta-decoded lengths (the path that mis-summed lengths when a mini-block did not fit + // the rolling buffer). + auto const s256 = delta_test_strings(301, false); + delta_large_mini_block_string_read_test( + build_delta_length_byte_array_parquet(s256, 256, 1), s256, 0, 200); + + auto const s128 = delta_test_strings(173, false); + delta_large_mini_block_string_read_test( + build_delta_length_byte_array_parquet(s128, 128, 1), s128, 0, 100); +} + +TEST_F(ParquetReaderTest, DeltaLengthByteArrayLargeMiniBlockSkipRows) +{ + // Leading row-range skips sum the skipped lengths one pass at a time (skip_values_and_sum), + // so they work for any mini-block size. + auto const s256 = delta_test_strings(301, false); + delta_large_mini_block_string_read_test( + build_delta_length_byte_array_parquet(s256, 256, 1), s256, 100); + + auto const s96 = delta_test_strings(141, false); + delta_large_mini_block_string_read_test( + build_delta_length_byte_array_parquet(s96, 384, 4), s96, 40, 60); +} + +TEST_F(ParquetReaderTest, DeltaBinaryListMiniBlock64) +{ + // LIST with 64 values/mini-block. The leading-skip read resumes the delta decoder + // mid-page after skip_values(), the path that requires the single-pass-per-iteration batch. + auto const lists = delta_test_lists(150); + auto const file = build_delta_binary_list_parquet(lists, 256, 4); + delta_large_mini_block_list_read_test(file, lists); + delta_large_mini_block_list_read_test(file, lists, 40); + delta_large_mini_block_list_read_test(file, lists, 40, 60); +} + +TEST_F(ParquetReaderTest, DeltaBinaryListLargeMiniBlock) +{ + // LIST with mini-blocks larger than the decode batch: a leading skip fast-forwards the + // decoder one pass at a time (skip_values) and resumes at a pass boundary, so any mini-block + // size works. + auto const lists = delta_test_lists(150); + for (auto const& [block_size, mini_block_count] : + {std::pair{384, 4}, std::pair{128, 1}, std::pair{256, 1}}) { + auto const file = build_delta_binary_list_parquet(lists, block_size, mini_block_count); + delta_large_mini_block_list_read_test(file, lists); + delta_large_mini_block_list_read_test(file, lists, 40); + delta_large_mini_block_list_read_test(file, lists, 40, 60); + } +} + +TEST_F(ParquetReaderTest, DeltaByteArrayListLargeMiniBlock) +{ + // LIST with DELTA_BYTE_ARRAY leaves: a leading skip reconstructs the skipped strings + // into temp_string_buf one pass at a time (delta_byte_array_decoder::skip), so any mini-block + // size works. The long-string variant takes the character-parallel reconstruction path. + auto const lists = delta_test_string_lists(150, true); + for (auto const& [block_size, mini_block_count] : + {std::pair{256, 4}, std::pair{384, 4}, std::pair{256, 1}}) { + auto const file = build_delta_byte_array_list_parquet(lists, block_size, mini_block_count); + delta_large_mini_block_string_list_read_test(file, lists); + delta_large_mini_block_string_list_read_test(file, lists, 40); + delta_large_mini_block_string_list_read_test(file, lists, 40, 60); + } + + auto const long_lists = delta_test_string_lists(150, true, 502, 90); + auto const file = build_delta_byte_array_list_parquet(long_lists, 384, 4); + delta_large_mini_block_string_list_read_test(file, long_lists, 40); +} + +TEST_F(ParquetReaderTest, DeltaLengthByteArrayListLargeMiniBlock) +{ + // LIST with DELTA_LENGTH_BYTE_ARRAY leaves: a leading skip resumes the length decoder + // mid-page (skip_values_and_sum) at a pass boundary. + auto const lists = delta_test_string_lists(150, false); + for (auto const& [block_size, mini_block_count] : + {std::pair{256, 4}, std::pair{384, 4}, std::pair{256, 1}}) { + auto const file = + build_delta_length_byte_array_list_parquet(lists, block_size, mini_block_count); + delta_large_mini_block_string_list_read_test(file, lists); + delta_large_mini_block_string_list_read_test(file, lists, 40); + delta_large_mini_block_string_list_read_test(file, lists, 40, 60); + } +} + +TEST_F(ParquetReaderTest, DeltaBinaryUnsupportedMiniBlockGeometry) +{ + // block_size=96, mini_block_count=2 -> 48 values/mini-block, not a multiple of 32: + // init_binary_block rejects the geometry and the reader fails with DELTA_PARAMS_UNSUPPORTED + // (0x100) before decoding any deltas, rather than reading a malformed mini-block. + auto const body = encode_delta_binary_header(96, 2, 100, 0); + auto const file = wrap_single_page_parquet(body, + 100, + cudf::io::parquet::Type::INT64, + cudf::io::parquet::Encoding::DELTA_BINARY_PACKED, + false); + EXPECT_THROW(cudf::io::read_parquet(delta_fixture_reader_options(file, 0, std::nullopt)), + cudf::logic_error); +} + +TEST_F(ParquetReaderTest, DeltaBinaryListLargeMiniBlockLeafNulls) +{ + // LIST whose leaves are ~1/5 null (definition level 2). Only the non-null leaves are + // DELTA-encoded, so leaf-null handling and the value->output mapping are exercised at mini-block + // sizes past 64 values. + auto const base = delta_test_lists(150); // list shapes and non-null leaf values + std::vector>> lists; + size_t leaf_idx = 0; + for (auto const& list : base) { + auto& out = lists.emplace_back(); + for (auto const v : list) { + // interleave null and present leaves deterministically + out.push_back((leaf_idx++ % 5 == 0) ? std::nullopt : std::optional{v}); + } + } + + // expected LIST with a nullable child, built from the same lists + std::vector offsets{0}; + std::vector leaf_values; + std::vector leaf_valid; + for (auto const& list : lists) { + for (auto const& e : list) { + leaf_values.push_back(e.value_or(0)); + leaf_valid.push_back(e.has_value()); + } + offsets.push_back(static_cast(leaf_values.size())); + } + auto const valid_it = cudf::detail::make_counting_transform_iterator( + 0, [&leaf_valid](auto i) { return leaf_valid[i]; }); + auto child = cudf::test::fixed_width_column_wrapper( + leaf_values.begin(), leaf_values.end(), valid_it); + auto offsets_col = + cudf::test::fixed_width_column_wrapper(offsets.begin(), offsets.end()); + auto const expected = cudf::make_lists_column(static_cast(lists.size()), + offsets_col.release(), + child.release(), + 0, + rmm::device_buffer{}); + + for (auto const& [block_size, mini_block_count] : + {std::pair{128, 1}, std::pair{384, 4}, std::pair{256, 1}}) { + auto const file = + build_delta_binary_list_with_leaf_nulls_parquet(lists, block_size, mini_block_count); + auto const result = cudf::io::read_parquet(delta_fixture_reader_options(file, 0, std::nullopt)); + CUDF_TEST_EXPECT_COLUMNS_EQUIVALENT(result.tbl->view().column(0), expected->view()); + } +} + TEST_F(ParquetReaderTest, EmptyOutput) { cudf::test::fixed_width_column_wrapper c0; @@ -2728,6 +3108,113 @@ TEST_F(ParquetReaderTest, DeltaByteArraySkipAllValid) result.tbl->view()); } +namespace { +// read `buffer` trimmed to [skip, skip + n) and compare column 0 with the matching slice of +// `expected` +void delta_byte_array_nested_skip_check(std::vector const& buffer, + cudf::table_view const& expected, + cudf::size_type skip, + cudf::size_type n) +{ + auto const result = + 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(skip) + .num_rows(n) + .build()); + CUDF_TEST_EXPECT_COLUMNS_EQUIVALENT(result.tbl->view().column(0), + cudf::slice(expected, {skip, skip + n}).front().column(0)); +} +} // namespace + +TEST_F(ParquetReaderTest, DeltaByteArrayStructSkipRows) +{ + // STRUCT whose string field is DELTA_BYTE_ARRAY encoded. A struct adds a definition level + // (the field is nullable here) but no repetition, so leading skips stage the skipped strings + // through the in-loop temp_string_buf and carry the prefix seed across rounds -- the flat-column + // path, but exercised under a nested schema with nulls. + constexpr cudf::size_type num_rows = 400; + auto const strings = delta_test_strings(num_rows, true); + auto const str_valids = + cudf::detail::make_counting_transform_iterator(0, [](auto i) { return i % 5 != 0; }); + + std::vector> children; + children.push_back(cudf::purge_nonempty_nulls( + cudf::test::strings_column_wrapper(strings.begin(), strings.end(), str_valids))); + auto const struct_col = cudf::make_structs_column(num_rows, std::move(children), 0, {}); + auto const expected = cudf::table_view({struct_col->view()}); + + cudf::io::table_input_metadata md(expected); + md.column_metadata[0].set_name("s"); + md.column_metadata[0].child(0).set_name("str").set_encoding( + cudf::io::column_encoding::DELTA_BYTE_ARRAY); + + std::vector buffer; + cudf::io::write_parquet( + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, expected) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .write_v2_headers(true) + .metadata(std::move(md)) + .build()); + + for (auto const& [skip, n] : std::vector>{ + {0, num_rows}, {40, num_rows - 40}, {100, 200}, {160, 33}}) { + delta_byte_array_nested_skip_check(buffer, expected, skip, n); + } +} + +TEST_F(ParquetReaderTest, DeltaByteArrayMapSkipRows) +{ + // MAP is physically LIST>, so both string leaves decode under + // repetition. A leading skip resumes each DELTA_BYTE_ARRAY leaf mid-page via + // delta_byte_array_decoder::skip, which stages the skipped strings through temp_string_buf and + // carries the prefix seed across rounds. Varying map sizes (including empties) exercise the + // definition levels. + constexpr cudf::size_type num_rows = 250; + uint64_t seed = 601; + std::vector offsets{0}; + for (cudf::size_type i = 0; i < num_rows; i++) { + auto const sz = (delta_test_rand(seed) % 6 == 0) + ? 0 + : 1 + static_cast(delta_test_rand(seed) % 6); + offsets.push_back(offsets.back() + sz); + } + auto const n_leaf = static_cast(offsets.back()); + auto const keys = delta_test_strings(n_leaf, true, 602); + auto const vals = delta_test_strings(n_leaf, true, 603); + + auto keys_col = cudf::test::strings_column_wrapper(keys.begin(), keys.end()); + auto vals_col = cudf::test::strings_column_wrapper(vals.begin(), vals.end()); + auto struct_col = cudf::test::structs_column_wrapper({keys_col, vals_col}).release(); + auto offsets_col = + cudf::test::fixed_width_column_wrapper(offsets.begin(), offsets.end()); + auto const map_col = cudf::make_lists_column( + num_rows, offsets_col.release(), std::move(struct_col), 0, rmm::device_buffer{}); + auto const expected = cudf::table_view({map_col->view()}); + + cudf::io::table_input_metadata md(expected); + md.column_metadata[0].set_name("m"); + md.column_metadata[0].set_list_column_as_map(); + md.column_metadata[0].child(1).child(0).set_name("key").set_encoding( + cudf::io::column_encoding::DELTA_BYTE_ARRAY); + md.column_metadata[0].child(1).child(1).set_name("value").set_encoding( + cudf::io::column_encoding::DELTA_BYTE_ARRAY); + + std::vector buffer; + cudf::io::write_parquet( + cudf::io::parquet_writer_options::builder(cudf::io::sink_info{&buffer}, expected) + .dictionary_policy(cudf::io::dictionary_policy::NEVER) + .write_v2_headers(true) + .metadata(std::move(md)) + .build()); + + for (auto const& [skip, n] : std::vector>{ + {0, num_rows}, {40, num_rows - 40}, {70, 100}, {130, 40}}) { + delta_byte_array_nested_skip_check(buffer, expected, skip, n); + } +} + // test that using page stats is working for full reads and various skip rows TEST_F(ParquetReaderTest, StringsWithPageStats) {