From 8ba5f096f3302ae939f35e1213f0855e8403510e Mon Sep 17 00:00:00 2001 From: ykiran Date: Fri, 7 Aug 2026 14:56:51 -0700 Subject: [PATCH 1/4] Added Parquet Read benchmark --- cpp/benchmarks/CMakeLists.txt | 2 +- .../io/parquet/parquet_reader_dict.cpp | 402 ++++++++++++++++++ 2 files changed, 403 insertions(+), 1 deletion(-) create mode 100644 cpp/benchmarks/io/parquet/parquet_reader_dict.cpp diff --git a/cpp/benchmarks/CMakeLists.txt b/cpp/benchmarks/CMakeLists.txt index d03bf72ee9e8..96401e863baf 100644 --- a/cpp/benchmarks/CMakeLists.txt +++ b/cpp/benchmarks/CMakeLists.txt @@ -297,7 +297,7 @@ ConfigureNVBench( # * parquet reader benchmark ---------------------------------------------------------------------- ConfigureNVBench( 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 + io/parquet/parquet_reader_options.cpp io/parquet/parquet_reader_dict.cpp io/parquet/reader_common.cpp ) # ################################################################################################## diff --git a/cpp/benchmarks/io/parquet/parquet_reader_dict.cpp b/cpp/benchmarks/io/parquet/parquet_reader_dict.cpp new file mode 100644 index 000000000000..a0aa26f98d78 --- /dev/null +++ b/cpp/benchmarks/io/parquet/parquet_reader_dict.cpp @@ -0,0 +1,402 @@ +/* + * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +// Benchmark for the parquet-dictionary -> cudf DICTIONARY32 transcode fast path enabled by +// `parquet_reader_options::output_dict_columns`. A flat fully dictionary-encoded string column is +// read under three modes, selected by the `mode` axis, so the transcode path can be judged against +// both a lower and an upper reference: +// +// - "materialize_string": reader default; the column materializes as STRING. The cheapest +// possible +// read (no dictionary built); serves as the lower-bound reference. +// - "materialize_string_and_encode_dict": materialize as STRING, then `cudf::dictionary::encode` +// to DICTIONARY32. The pre-existing way to obtain a dictionary column and the +// fair apples-to-apples baseline the transcode fast path aims to beat. +// - "direct_dict_transcode": `output_dict_columns=true`; the reader keeps the dictionary +// representation and produces DICTIONARY32 directly, skipping string +// materialization. +// +// Both "materialize_string_and_encode_dict" and "direct_dict_transcode" produce DICTIONARY32 +// output, so their times and peak memory are directly comparable; "materialize_string" shows the +// floor cost of just decoding. A relative comparison table (materialize_string_and_encode_dict = +// 100%%) is printed at program exit (see comparison_collector). +// +// The sweep varies cardinality, rows per row group, and rows per data page at a fixed table size +// A single column (num_cols == 1) is used, so a row group can hold as many distinct values as +// possible. The writer picks the dictionary index bit width per row group from the distinct values +// it contains, capped at MAX_DICT_BITS (24). Cardinality therefore ranges up to 2^24, the point at +// which 24-bit indices are required. At high distinct-per-row-group counts the writer may abandon +// dictionary encoding (indices exceed 24 bits, or plain encoding is smaller); when that leaves the +// column ineligible for transcode, that state is skipped rather than measured (the benchmark +// omits skipped states from its own table). + +namespace { + +constexpr cudf::size_type num_cols = 1; + +enum class bench_mode { + materialize_string, + materialize_string_and_encode_dict, + direct_dict_transcode +}; + +[[nodiscard]] bench_mode parse_mode(std::string const& mode) +{ + if (mode == "materialize_string") { return bench_mode::materialize_string; } + if (mode == "materialize_string_and_encode_dict") { + return bench_mode::materialize_string_and_encode_dict; + } + if (mode == "direct_dict_transcode") { return bench_mode::direct_dict_transcode; } + CUDF_FAIL("Unknown benchmark mode: " + mode); +} + +// Upper-bound estimate of the dictionary index bit width the writer will use for a row group. The +// width is derived per row group from its distinct value count, which is at most +// min(cardinality, rows in the row group). This over-estimates when the (last) row group is shorter +// than `row_group_size_rows` or when hash collisions reduce distinct counts. +[[nodiscard]] int approx_dict_bits(std::int64_t cardinality, std::int64_t row_group_size_rows) +{ + auto const distinct = std::min(cardinality, row_group_size_rows); + if (distinct <= 1) { return 1; } + return static_cast(std::bit_width(static_cast(distinct - 1))); +} + +// nvbench invokes the benchmark once per axis combination, prints its own results table, and omits +// skipped states from it; there is also no cross-state hook, so a single invocation cannot group +// the three modes of a configuration together. This collector accumulates each run's CPU/GPU mean +// time (and any direct_dict_transcode skip reason), keyed by every setting except `mode`, and +// prints one row per configuration from its destructor -- i.e. at program exit, after nvbench's own +// output +// -- with all three modes in fixed order (materialize_string, materialize_string_and_encode_dict, +// direct_dict_transcode) so each configuration is +// grouped and ordered regardless of nvbench's state ordering or its omission of skipped states. +struct run_settings { + std::int64_t cardinality; + std::int64_t data_size; + std::int64_t row_group_size_rows; + std::int64_t max_page_size_rows; + std::int64_t avg_string_length; + + bool operator<(run_settings const& o) const + { + return std::tie( + cardinality, data_size, row_group_size_rows, max_page_size_rows, avg_string_length) < + std::tie(o.cardinality, + o.data_size, + o.row_group_size_rows, + o.max_page_size_rows, + o.avg_string_length); + } +}; + +struct mode_timing { + double cpu_ms = 0.0; + double gpu_ms = 0.0; + bool present = false; +}; + +class comparison_collector { + public: + void record(run_settings const& key, bench_mode mode, double cpu_ms, double gpu_ms) + { + auto& r = _rows[key]; + auto& slot = (mode == bench_mode::materialize_string_and_encode_dict) + ? r.materialize_string_and_encode_dict + : ((mode == bench_mode::direct_dict_transcode) ? r.direct_dict_transcode + : r.materialize_string); + slot = mode_timing{cpu_ms, gpu_ms, true}; + } + + // Record that direct_dict_transcode was skipped for a configuration, with a short reason shown in + // the table. `direct_dict_transcode` is the only mode this benchmark ever skips. + void record_skip(run_settings const& key, std::string reason) + { + _rows[key].direct_dict_transcode_note = std::move(reason); + } + + ~comparison_collector() { print(); } + + private: + struct row { + mode_timing materialize_string; + mode_timing materialize_string_and_encode_dict; + mode_timing direct_dict_transcode; + // reason shown when `direct_dict_transcode` was skipped as ineligible + std::string direct_dict_transcode_note; + }; + + void print() const + { + if (_rows.empty()) { return; } + + std::printf( + "\n# Per-configuration mode comparison " + "(order: materialize_string, materialize_string_and_encode_dict, " + "direct_dict_transcode)\n\n"); + std::printf( + "| cardinality | ~dict_bits | data_size (MiB) | row_group_size_rows | max_page_size_rows | " + "materialize_string CPU (ms) | materialize_string_and_encode_dict CPU (ms) | " + "direct_dict_transcode CPU (ms) | materialize_string GPU (ms) | " + "materialize_string_and_encode_dict GPU (ms) | direct_dict_transcode GPU (ms) | " + "direct_dict_transcode CPU speedup %% | direct_dict_transcode GPU speedup %% |\n"); + std::printf("|---|---|---|---|---|---|---|---|---|---|---|---|---|\n"); + + auto const num = [](double v) { + std::array buf{}; + std::snprintf(buf.data(), buf.size(), "%.3f", v); + return std::string{buf.data()}; + }; + // Timing cell: the value if the mode ran, else the skip reason (direct_dict_transcode only) or + // "-". + auto const cell = + [&](mode_timing const& t, double mode_timing::* field, std::string const& note) { + if (t.present) { return num(t.*field); } + return note.empty() ? std::string{"-"} : note; + }; + + // Speedup of direct_dict_transcode over the materialize_string_and_encode_dict baseline, as a + // signed percentage of baseline time saved: + // 100 * (materialize_string_and_encode_dict - direct_dict_transcode) / + // materialize_string_and_encode_dict. + // Positive = direct_dict_transcode faster, negative = slower. "-" when either mode is missing. + auto const speedup = + [](mode_timing const& base, mode_timing const& cand, double mode_timing::* field) { + if (not(base.present and cand.present)) { return std::string{"-"}; } + std::array buf{}; + std::snprintf( + buf.data(), buf.size(), "%+.1f%%", 100.0 * (base.*field - cand.*field) / (base.*field)); + return std::string{buf.data()}; + }; + + for (auto const& [key, r] : _rows) { + std::printf( + "| %lld | %d | %lld | %lld | %lld | %s | %s | %s | %s | %s | %s | %s | %s |\n", + static_cast(key.cardinality), + approx_dict_bits(key.cardinality, key.row_group_size_rows), + static_cast(key.data_size >> 20), + static_cast(key.row_group_size_rows), + static_cast(key.max_page_size_rows), + cell(r.materialize_string, &mode_timing::cpu_ms, std::string{}).c_str(), + cell(r.materialize_string_and_encode_dict, &mode_timing::cpu_ms, std::string{}).c_str(), + cell(r.direct_dict_transcode, &mode_timing::cpu_ms, r.direct_dict_transcode_note).c_str(), + cell(r.materialize_string, &mode_timing::gpu_ms, std::string{}).c_str(), + cell(r.materialize_string_and_encode_dict, &mode_timing::gpu_ms, std::string{}).c_str(), + cell(r.direct_dict_transcode, &mode_timing::gpu_ms, r.direct_dict_transcode_note).c_str(), + speedup(r.materialize_string_and_encode_dict, r.direct_dict_transcode, &mode_timing::cpu_ms) + .c_str(), + speedup(r.materialize_string_and_encode_dict, r.direct_dict_transcode, &mode_timing::gpu_ms) + .c_str()); + } + std::printf("\n"); + } + + std::map _rows; +}; + +comparison_collector g_comparison_collector; + +// The transcode fast path requires every data page of an eligible column to be dictionary-encoded. +// Forcing `dictionary_policy::ALWAYS` maximizes the chance of full dictionary encoding; the writer +// can still fall back to plain when indices exceed MAX_DICT_BITS or plain is smaller, in which case +// the direct_dict_transcode state is skipped by the caller. +void write_dict_encoded_parquet(cudf::table_view const& view, + cuio_source_sink_pair& source_sink, + std::int64_t row_group_size_rows, + std::int64_t max_page_size_rows) +{ + cudf::io::parquet_writer_options write_opts = + cudf::io::parquet_writer_options::builder(source_sink.make_sink_info(), view) + .compression(cudf::io::compression_type::NONE) + .dictionary_policy(cudf::io::dictionary_policy::ALWAYS) + .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN); + if (row_group_size_rows > 0) { + write_opts.set_row_group_size_rows(static_cast(row_group_size_rows)); + } + if (max_page_size_rows > 0) { + write_opts.set_max_page_size_rows(static_cast(max_page_size_rows)); + } + cudf::io::write_parquet(write_opts); +} + +} // namespace + +void BM_parquet_read_dict_transcode(nvbench::state& state) +{ + auto const cardinality = static_cast(state.get_int64("cardinality")); + auto const data_size = static_cast(state.get_int64("data_size")); + auto const rg_size_rows = state.get_int64("row_group_size_rows"); + auto const page_size_rows = state.get_int64("max_page_size_rows"); + auto const mode = parse_mode(state.get_string("mode")); + auto const avg_string_length = static_cast(state.get_int64("avg_string_length")); + auto const source_type = retrieve_io_type_enum(state.get_string("io_type")); + + // corresponds to 3 sigma (full width 6 sigma: 99.7% of range) + auto const half_width = avg_string_length >> 3; + auto const length_min = avg_string_length - half_width; + auto const length_max = avg_string_length + half_width; + + data_profile const profile = + data_profile_builder() + .cardinality(cardinality) + .avg_run_length(1) + .distribution(data_type::STRING, distribution_id::NORMAL, length_min, length_max); + + auto const d_type = get_type_or_group(static_cast(data_type::STRING)); + auto const tbl = + create_random_table(cycle_dtypes(d_type, num_cols), table_size_bytes{data_size}, profile); + auto const view = tbl->view(); + + cuio_source_sink_pair source_sink(source_type); + write_dict_encoded_parquet(view, source_sink, rg_size_rows, page_size_rows); + + cudf::io::parquet_reader_options read_opts = + cudf::io::parquet_reader_options::builder(source_sink.make_source_info()) + .output_dict_columns(mode == bench_mode::direct_dict_transcode); + + // Perform the full work for the selected mode: read, and for `materialize_string_and_encode_dict` + // additionally encode each STRING column to DICTIONARY32. Returns the resulting table so it can + // be reused for both the outside-the-timed-region verification and the timed measurement. + auto const run_mode = [&]() -> std::unique_ptr { + auto result = cudf::io::read_parquet(read_opts); + if (mode == bench_mode::materialize_string_and_encode_dict) { + std::vector> encoded; + encoded.reserve(result.tbl->num_columns()); + for (auto const& col : result.tbl->view()) { + encoded.push_back(cudf::dictionary::encode(col)); + } + return std::make_unique(std::move(encoded)); + } + return std::move(result.tbl); + }; + + // Verification (outside the timed region, run for every mode so warm-up is symmetric). For + // `direct_dict_transcode`, the writer may have fallen back to plain encoding at high cardinality + // / large row groups, leaving the column ineligible for the fast path; in that case skip the + // state rather than silently measuring the plain path or aborting the whole sweep. + { + auto const probe = run_mode(); + // Bind the table_view to a local: `probe->view()` returns a temporary, so calling it separately + // for begin() and end() would yield iterators into two different temporaries (mismatched- + // iterator UB). Iterate a single view instead. + auto const probe_view = probe->view(); + CUDF_EXPECTS(probe_view.num_columns() == num_cols, "Unexpected number of columns"); + auto const all_of_type = [&](cudf::type_id id) { + return std::all_of(probe_view.begin(), probe_view.end(), [id](auto const& col) { + return col.type().id() == id; + }); + }; + auto const actual_type_id = static_cast(probe_view.column(0).type().id()); + if (mode == bench_mode::materialize_string) { + if (not all_of_type(cudf::type_id::STRING)) { + state.skip( + "materialize_string produced unexpected type_id=" + std::to_string(actual_type_id) + + " (expected STRING=" + std::to_string(static_cast(cudf::type_id::STRING)) + ")"); + return; + } + } else if (mode == bench_mode::materialize_string_and_encode_dict) { + if (not all_of_type(cudf::type_id::DICTIONARY32)) { + state.skip("materialize_string_and_encode_dict produced unexpected type_id=" + + std::to_string(actual_type_id) + " (expected DICTIONARY32=" + + std::to_string(static_cast(cudf::type_id::DICTIONARY32)) + ")"); + return; + } + } else if (not all_of_type(cudf::type_id::DICTIONARY32)) { + // Record the skip so the end-of-program per-configuration table can show why + // direct_dict_transcode has no + // timing for this configuration (nvbench omits skipped states from its own table). + g_comparison_collector.record_skip(run_settings{cardinality, + static_cast(data_size), + rg_size_rows, + page_size_rows, + avg_string_length}, + "skipped: plain fallback"); + state.skip( + "direct_dict_transcode did not produce DICTIONARY32: at this cardinality / row-group size " + "the writer " + "fell back to plain encoding, making the column ineligible for the fast path"); + return; + } + } + + auto mem_stats_logger = cudf::memory_stats_logger(); + state.set_cuda_stream(nvbench::make_cuda_stream_view(cudf::get_default_stream().value())); + state.exec(nvbench::exec_tag::sync | nvbench::exec_tag::timer, + [&](nvbench::launch& launch, auto& timer) { + drop_page_cache_if_enabled(read_opts.get_source().filepaths()); + + timer.start(); + auto const result = run_mode(); + timer.stop(); + + CUDF_EXPECTS(result->num_columns() == num_cols, "Unexpected number of columns"); + }); + + auto const time = state.get_summary("nv/cold/time/gpu/mean").get_float64("value"); + auto const cpu_time = state.get_summary("nv/cold/time/cpu/mean").get_float64("value"); + state.add_element_count(static_cast(data_size) / time, "bytes_per_second"); + state.add_element_count(static_cast(view.num_rows()) / time, "rows_per_sec"); + state.add_buffer_size( + mem_stats_logger.peak_memory_usage(), "peak_memory_usage", "peak_memory_usage"); + state.add_buffer_size(source_sink.size(), "encoded_file_size", "encoded_file_size"); + + // Record this run for the end-of-program + // direct_dict_transcode-vs-materialize_string_and_encode_dict comparison table. Times are + // reported by nvbench in seconds; store as milliseconds. + g_comparison_collector.record(run_settings{cardinality, + static_cast(data_size), + rg_size_rows, + page_size_rows, + avg_string_length}, + mode, + cpu_time * 1e3, + time * 1e3); +} + +NVBENCH_BENCH(BM_parquet_read_dict_transcode) + .set_name("parquet_read_dict_transcode") + .add_string_axis("io_type", {"DEVICE_BUFFER"}) + .set_min_samples(4) + .add_string_axis( + "mode", {"materialize_string", "materialize_string_and_encode_dict", "direct_dict_transcode"}) + // Cardinality: low, mid, and 2^24 -- the point at which per-row-group dictionary indices need the + // maximum 24 bits the writer supports (MAX_DICT_BITS); beyond that the writer abandons dictionary + // encoding. Achieved bits = ceil(log2(min(cardinality, rows per row group))). + .add_int64_axis("cardinality", {1 << 10, 1 << 20, 1 << 24}) + // Fixed table size, kept modest so the sweep stays light for local/CI runs (peak memory and + // per-state runtime scale with this). + .add_int64_axis("data_size", {std::int64_t{512} << 20}) + // Rows per row group: small (many row groups -> stresses per-row-group key concatenation), + // default (1M), and very large (>= 2^24 so a single row group can reach the 24-bit dict + // boundary). + .add_int64_axis("row_group_size_rows", {100'000, 1'000'000, 20'000'000}) + // Rows per data page: small and large (page size is a second-order factor for this benchmark). + .add_int64_axis("max_page_size_rows", {20'000, 1'000'000}) + .add_int64_axis("avg_string_length", {16}); From 0f0bc95581d3aeeccca60a6a4b65fb8858806efd Mon Sep 17 00:00:00 2001 From: ykiran Date: Wed, 12 Aug 2026 14:40:51 -0700 Subject: [PATCH 2/4] Folded into parquet_reader_options.cpp --- cpp/benchmarks/CMakeLists.txt | 2 +- cpp/benchmarks/io/nvbench_helpers.hpp | 13 + .../io/parquet/parquet_reader_dict.cpp | 402 ------------------ .../io/parquet/parquet_reader_options.cpp | 38 +- 4 files changed, 49 insertions(+), 406 deletions(-) delete mode 100644 cpp/benchmarks/io/parquet/parquet_reader_dict.cpp diff --git a/cpp/benchmarks/CMakeLists.txt b/cpp/benchmarks/CMakeLists.txt index 96401e863baf..d03bf72ee9e8 100644 --- a/cpp/benchmarks/CMakeLists.txt +++ b/cpp/benchmarks/CMakeLists.txt @@ -297,7 +297,7 @@ ConfigureNVBench( # * parquet reader benchmark ---------------------------------------------------------------------- ConfigureNVBench( PARQUET_READER_NVBENCH io/parquet/parquet_reader_input.cpp io/parquet/parquet_reader_encoding.cpp - io/parquet/parquet_reader_options.cpp io/parquet/parquet_reader_dict.cpp io/parquet/reader_common.cpp + io/parquet/parquet_reader_options.cpp io/parquet/reader_common.cpp ) # ################################################################################################## diff --git a/cpp/benchmarks/io/nvbench_helpers.hpp b/cpp/benchmarks/io/nvbench_helpers.hpp index ebffb930c275..99aaa55d4a5a 100644 --- a/cpp/benchmarks/io/nvbench_helpers.hpp +++ b/cpp/benchmarks/io/nvbench_helpers.hpp @@ -137,6 +137,8 @@ NVBENCH_DECLARE_ENUM_TYPE_STRINGS( enum class converts_strings : bool { YES, NO }; +enum class output_dict : bool { YES, NO }; + enum class uses_pandas_metadata : bool { YES, NO }; NVBENCH_DECLARE_ENUM_TYPE_STRINGS( @@ -150,6 +152,17 @@ NVBENCH_DECLARE_ENUM_TYPE_STRINGS( }, [](auto) { return std::string{}; }) +NVBENCH_DECLARE_ENUM_TYPE_STRINGS( + output_dict, + [](auto value) { + switch (value) { + case output_dict::YES: return "YES"; + case output_dict::NO: return "NO"; + default: return "Unknown"; + } + }, + [](auto) { return std::string{}; }) + NVBENCH_DECLARE_ENUM_TYPE_STRINGS( uses_pandas_metadata, [](auto value) { diff --git a/cpp/benchmarks/io/parquet/parquet_reader_dict.cpp b/cpp/benchmarks/io/parquet/parquet_reader_dict.cpp deleted file mode 100644 index a0aa26f98d78..000000000000 --- a/cpp/benchmarks/io/parquet/parquet_reader_dict.cpp +++ /dev/null @@ -1,402 +0,0 @@ -/* - * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. - * SPDX-License-Identifier: Apache-2.0 - */ - -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include - -#include - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -// Benchmark for the parquet-dictionary -> cudf DICTIONARY32 transcode fast path enabled by -// `parquet_reader_options::output_dict_columns`. A flat fully dictionary-encoded string column is -// read under three modes, selected by the `mode` axis, so the transcode path can be judged against -// both a lower and an upper reference: -// -// - "materialize_string": reader default; the column materializes as STRING. The cheapest -// possible -// read (no dictionary built); serves as the lower-bound reference. -// - "materialize_string_and_encode_dict": materialize as STRING, then `cudf::dictionary::encode` -// to DICTIONARY32. The pre-existing way to obtain a dictionary column and the -// fair apples-to-apples baseline the transcode fast path aims to beat. -// - "direct_dict_transcode": `output_dict_columns=true`; the reader keeps the dictionary -// representation and produces DICTIONARY32 directly, skipping string -// materialization. -// -// Both "materialize_string_and_encode_dict" and "direct_dict_transcode" produce DICTIONARY32 -// output, so their times and peak memory are directly comparable; "materialize_string" shows the -// floor cost of just decoding. A relative comparison table (materialize_string_and_encode_dict = -// 100%%) is printed at program exit (see comparison_collector). -// -// The sweep varies cardinality, rows per row group, and rows per data page at a fixed table size -// A single column (num_cols == 1) is used, so a row group can hold as many distinct values as -// possible. The writer picks the dictionary index bit width per row group from the distinct values -// it contains, capped at MAX_DICT_BITS (24). Cardinality therefore ranges up to 2^24, the point at -// which 24-bit indices are required. At high distinct-per-row-group counts the writer may abandon -// dictionary encoding (indices exceed 24 bits, or plain encoding is smaller); when that leaves the -// column ineligible for transcode, that state is skipped rather than measured (the benchmark -// omits skipped states from its own table). - -namespace { - -constexpr cudf::size_type num_cols = 1; - -enum class bench_mode { - materialize_string, - materialize_string_and_encode_dict, - direct_dict_transcode -}; - -[[nodiscard]] bench_mode parse_mode(std::string const& mode) -{ - if (mode == "materialize_string") { return bench_mode::materialize_string; } - if (mode == "materialize_string_and_encode_dict") { - return bench_mode::materialize_string_and_encode_dict; - } - if (mode == "direct_dict_transcode") { return bench_mode::direct_dict_transcode; } - CUDF_FAIL("Unknown benchmark mode: " + mode); -} - -// Upper-bound estimate of the dictionary index bit width the writer will use for a row group. The -// width is derived per row group from its distinct value count, which is at most -// min(cardinality, rows in the row group). This over-estimates when the (last) row group is shorter -// than `row_group_size_rows` or when hash collisions reduce distinct counts. -[[nodiscard]] int approx_dict_bits(std::int64_t cardinality, std::int64_t row_group_size_rows) -{ - auto const distinct = std::min(cardinality, row_group_size_rows); - if (distinct <= 1) { return 1; } - return static_cast(std::bit_width(static_cast(distinct - 1))); -} - -// nvbench invokes the benchmark once per axis combination, prints its own results table, and omits -// skipped states from it; there is also no cross-state hook, so a single invocation cannot group -// the three modes of a configuration together. This collector accumulates each run's CPU/GPU mean -// time (and any direct_dict_transcode skip reason), keyed by every setting except `mode`, and -// prints one row per configuration from its destructor -- i.e. at program exit, after nvbench's own -// output -// -- with all three modes in fixed order (materialize_string, materialize_string_and_encode_dict, -// direct_dict_transcode) so each configuration is -// grouped and ordered regardless of nvbench's state ordering or its omission of skipped states. -struct run_settings { - std::int64_t cardinality; - std::int64_t data_size; - std::int64_t row_group_size_rows; - std::int64_t max_page_size_rows; - std::int64_t avg_string_length; - - bool operator<(run_settings const& o) const - { - return std::tie( - cardinality, data_size, row_group_size_rows, max_page_size_rows, avg_string_length) < - std::tie(o.cardinality, - o.data_size, - o.row_group_size_rows, - o.max_page_size_rows, - o.avg_string_length); - } -}; - -struct mode_timing { - double cpu_ms = 0.0; - double gpu_ms = 0.0; - bool present = false; -}; - -class comparison_collector { - public: - void record(run_settings const& key, bench_mode mode, double cpu_ms, double gpu_ms) - { - auto& r = _rows[key]; - auto& slot = (mode == bench_mode::materialize_string_and_encode_dict) - ? r.materialize_string_and_encode_dict - : ((mode == bench_mode::direct_dict_transcode) ? r.direct_dict_transcode - : r.materialize_string); - slot = mode_timing{cpu_ms, gpu_ms, true}; - } - - // Record that direct_dict_transcode was skipped for a configuration, with a short reason shown in - // the table. `direct_dict_transcode` is the only mode this benchmark ever skips. - void record_skip(run_settings const& key, std::string reason) - { - _rows[key].direct_dict_transcode_note = std::move(reason); - } - - ~comparison_collector() { print(); } - - private: - struct row { - mode_timing materialize_string; - mode_timing materialize_string_and_encode_dict; - mode_timing direct_dict_transcode; - // reason shown when `direct_dict_transcode` was skipped as ineligible - std::string direct_dict_transcode_note; - }; - - void print() const - { - if (_rows.empty()) { return; } - - std::printf( - "\n# Per-configuration mode comparison " - "(order: materialize_string, materialize_string_and_encode_dict, " - "direct_dict_transcode)\n\n"); - std::printf( - "| cardinality | ~dict_bits | data_size (MiB) | row_group_size_rows | max_page_size_rows | " - "materialize_string CPU (ms) | materialize_string_and_encode_dict CPU (ms) | " - "direct_dict_transcode CPU (ms) | materialize_string GPU (ms) | " - "materialize_string_and_encode_dict GPU (ms) | direct_dict_transcode GPU (ms) | " - "direct_dict_transcode CPU speedup %% | direct_dict_transcode GPU speedup %% |\n"); - std::printf("|---|---|---|---|---|---|---|---|---|---|---|---|---|\n"); - - auto const num = [](double v) { - std::array buf{}; - std::snprintf(buf.data(), buf.size(), "%.3f", v); - return std::string{buf.data()}; - }; - // Timing cell: the value if the mode ran, else the skip reason (direct_dict_transcode only) or - // "-". - auto const cell = - [&](mode_timing const& t, double mode_timing::* field, std::string const& note) { - if (t.present) { return num(t.*field); } - return note.empty() ? std::string{"-"} : note; - }; - - // Speedup of direct_dict_transcode over the materialize_string_and_encode_dict baseline, as a - // signed percentage of baseline time saved: - // 100 * (materialize_string_and_encode_dict - direct_dict_transcode) / - // materialize_string_and_encode_dict. - // Positive = direct_dict_transcode faster, negative = slower. "-" when either mode is missing. - auto const speedup = - [](mode_timing const& base, mode_timing const& cand, double mode_timing::* field) { - if (not(base.present and cand.present)) { return std::string{"-"}; } - std::array buf{}; - std::snprintf( - buf.data(), buf.size(), "%+.1f%%", 100.0 * (base.*field - cand.*field) / (base.*field)); - return std::string{buf.data()}; - }; - - for (auto const& [key, r] : _rows) { - std::printf( - "| %lld | %d | %lld | %lld | %lld | %s | %s | %s | %s | %s | %s | %s | %s |\n", - static_cast(key.cardinality), - approx_dict_bits(key.cardinality, key.row_group_size_rows), - static_cast(key.data_size >> 20), - static_cast(key.row_group_size_rows), - static_cast(key.max_page_size_rows), - cell(r.materialize_string, &mode_timing::cpu_ms, std::string{}).c_str(), - cell(r.materialize_string_and_encode_dict, &mode_timing::cpu_ms, std::string{}).c_str(), - cell(r.direct_dict_transcode, &mode_timing::cpu_ms, r.direct_dict_transcode_note).c_str(), - cell(r.materialize_string, &mode_timing::gpu_ms, std::string{}).c_str(), - cell(r.materialize_string_and_encode_dict, &mode_timing::gpu_ms, std::string{}).c_str(), - cell(r.direct_dict_transcode, &mode_timing::gpu_ms, r.direct_dict_transcode_note).c_str(), - speedup(r.materialize_string_and_encode_dict, r.direct_dict_transcode, &mode_timing::cpu_ms) - .c_str(), - speedup(r.materialize_string_and_encode_dict, r.direct_dict_transcode, &mode_timing::gpu_ms) - .c_str()); - } - std::printf("\n"); - } - - std::map _rows; -}; - -comparison_collector g_comparison_collector; - -// The transcode fast path requires every data page of an eligible column to be dictionary-encoded. -// Forcing `dictionary_policy::ALWAYS` maximizes the chance of full dictionary encoding; the writer -// can still fall back to plain when indices exceed MAX_DICT_BITS or plain is smaller, in which case -// the direct_dict_transcode state is skipped by the caller. -void write_dict_encoded_parquet(cudf::table_view const& view, - cuio_source_sink_pair& source_sink, - std::int64_t row_group_size_rows, - std::int64_t max_page_size_rows) -{ - cudf::io::parquet_writer_options write_opts = - cudf::io::parquet_writer_options::builder(source_sink.make_sink_info(), view) - .compression(cudf::io::compression_type::NONE) - .dictionary_policy(cudf::io::dictionary_policy::ALWAYS) - .stats_level(cudf::io::statistics_freq::STATISTICS_COLUMN); - if (row_group_size_rows > 0) { - write_opts.set_row_group_size_rows(static_cast(row_group_size_rows)); - } - if (max_page_size_rows > 0) { - write_opts.set_max_page_size_rows(static_cast(max_page_size_rows)); - } - cudf::io::write_parquet(write_opts); -} - -} // namespace - -void BM_parquet_read_dict_transcode(nvbench::state& state) -{ - auto const cardinality = static_cast(state.get_int64("cardinality")); - auto const data_size = static_cast(state.get_int64("data_size")); - auto const rg_size_rows = state.get_int64("row_group_size_rows"); - auto const page_size_rows = state.get_int64("max_page_size_rows"); - auto const mode = parse_mode(state.get_string("mode")); - auto const avg_string_length = static_cast(state.get_int64("avg_string_length")); - auto const source_type = retrieve_io_type_enum(state.get_string("io_type")); - - // corresponds to 3 sigma (full width 6 sigma: 99.7% of range) - auto const half_width = avg_string_length >> 3; - auto const length_min = avg_string_length - half_width; - auto const length_max = avg_string_length + half_width; - - data_profile const profile = - data_profile_builder() - .cardinality(cardinality) - .avg_run_length(1) - .distribution(data_type::STRING, distribution_id::NORMAL, length_min, length_max); - - auto const d_type = get_type_or_group(static_cast(data_type::STRING)); - auto const tbl = - create_random_table(cycle_dtypes(d_type, num_cols), table_size_bytes{data_size}, profile); - auto const view = tbl->view(); - - cuio_source_sink_pair source_sink(source_type); - write_dict_encoded_parquet(view, source_sink, rg_size_rows, page_size_rows); - - cudf::io::parquet_reader_options read_opts = - cudf::io::parquet_reader_options::builder(source_sink.make_source_info()) - .output_dict_columns(mode == bench_mode::direct_dict_transcode); - - // Perform the full work for the selected mode: read, and for `materialize_string_and_encode_dict` - // additionally encode each STRING column to DICTIONARY32. Returns the resulting table so it can - // be reused for both the outside-the-timed-region verification and the timed measurement. - auto const run_mode = [&]() -> std::unique_ptr { - auto result = cudf::io::read_parquet(read_opts); - if (mode == bench_mode::materialize_string_and_encode_dict) { - std::vector> encoded; - encoded.reserve(result.tbl->num_columns()); - for (auto const& col : result.tbl->view()) { - encoded.push_back(cudf::dictionary::encode(col)); - } - return std::make_unique(std::move(encoded)); - } - return std::move(result.tbl); - }; - - // Verification (outside the timed region, run for every mode so warm-up is symmetric). For - // `direct_dict_transcode`, the writer may have fallen back to plain encoding at high cardinality - // / large row groups, leaving the column ineligible for the fast path; in that case skip the - // state rather than silently measuring the plain path or aborting the whole sweep. - { - auto const probe = run_mode(); - // Bind the table_view to a local: `probe->view()` returns a temporary, so calling it separately - // for begin() and end() would yield iterators into two different temporaries (mismatched- - // iterator UB). Iterate a single view instead. - auto const probe_view = probe->view(); - CUDF_EXPECTS(probe_view.num_columns() == num_cols, "Unexpected number of columns"); - auto const all_of_type = [&](cudf::type_id id) { - return std::all_of(probe_view.begin(), probe_view.end(), [id](auto const& col) { - return col.type().id() == id; - }); - }; - auto const actual_type_id = static_cast(probe_view.column(0).type().id()); - if (mode == bench_mode::materialize_string) { - if (not all_of_type(cudf::type_id::STRING)) { - state.skip( - "materialize_string produced unexpected type_id=" + std::to_string(actual_type_id) + - " (expected STRING=" + std::to_string(static_cast(cudf::type_id::STRING)) + ")"); - return; - } - } else if (mode == bench_mode::materialize_string_and_encode_dict) { - if (not all_of_type(cudf::type_id::DICTIONARY32)) { - state.skip("materialize_string_and_encode_dict produced unexpected type_id=" + - std::to_string(actual_type_id) + " (expected DICTIONARY32=" + - std::to_string(static_cast(cudf::type_id::DICTIONARY32)) + ")"); - return; - } - } else if (not all_of_type(cudf::type_id::DICTIONARY32)) { - // Record the skip so the end-of-program per-configuration table can show why - // direct_dict_transcode has no - // timing for this configuration (nvbench omits skipped states from its own table). - g_comparison_collector.record_skip(run_settings{cardinality, - static_cast(data_size), - rg_size_rows, - page_size_rows, - avg_string_length}, - "skipped: plain fallback"); - state.skip( - "direct_dict_transcode did not produce DICTIONARY32: at this cardinality / row-group size " - "the writer " - "fell back to plain encoding, making the column ineligible for the fast path"); - return; - } - } - - auto mem_stats_logger = cudf::memory_stats_logger(); - state.set_cuda_stream(nvbench::make_cuda_stream_view(cudf::get_default_stream().value())); - state.exec(nvbench::exec_tag::sync | nvbench::exec_tag::timer, - [&](nvbench::launch& launch, auto& timer) { - drop_page_cache_if_enabled(read_opts.get_source().filepaths()); - - timer.start(); - auto const result = run_mode(); - timer.stop(); - - CUDF_EXPECTS(result->num_columns() == num_cols, "Unexpected number of columns"); - }); - - auto const time = state.get_summary("nv/cold/time/gpu/mean").get_float64("value"); - auto const cpu_time = state.get_summary("nv/cold/time/cpu/mean").get_float64("value"); - state.add_element_count(static_cast(data_size) / time, "bytes_per_second"); - state.add_element_count(static_cast(view.num_rows()) / time, "rows_per_sec"); - state.add_buffer_size( - mem_stats_logger.peak_memory_usage(), "peak_memory_usage", "peak_memory_usage"); - state.add_buffer_size(source_sink.size(), "encoded_file_size", "encoded_file_size"); - - // Record this run for the end-of-program - // direct_dict_transcode-vs-materialize_string_and_encode_dict comparison table. Times are - // reported by nvbench in seconds; store as milliseconds. - g_comparison_collector.record(run_settings{cardinality, - static_cast(data_size), - rg_size_rows, - page_size_rows, - avg_string_length}, - mode, - cpu_time * 1e3, - time * 1e3); -} - -NVBENCH_BENCH(BM_parquet_read_dict_transcode) - .set_name("parquet_read_dict_transcode") - .add_string_axis("io_type", {"DEVICE_BUFFER"}) - .set_min_samples(4) - .add_string_axis( - "mode", {"materialize_string", "materialize_string_and_encode_dict", "direct_dict_transcode"}) - // Cardinality: low, mid, and 2^24 -- the point at which per-row-group dictionary indices need the - // maximum 24 bits the writer supports (MAX_DICT_BITS); beyond that the writer abandons dictionary - // encoding. Achieved bits = ceil(log2(min(cardinality, rows per row group))). - .add_int64_axis("cardinality", {1 << 10, 1 << 20, 1 << 24}) - // Fixed table size, kept modest so the sweep stays light for local/CI runs (peak memory and - // per-state runtime scale with this). - .add_int64_axis("data_size", {std::int64_t{512} << 20}) - // Rows per row group: small (many row groups -> stresses per-row-group key concatenation), - // default (1M), and very large (>= 2^24 so a single row group can reach the 24-bit dict - // boundary). - .add_int64_axis("row_group_size_rows", {100'000, 1'000'000, 20'000'000}) - // Rows per data page: small and large (page size is a second-order factor for this benchmark). - .add_int64_axis("max_page_size_rows", {20'000, 1'000'000}) - .add_int64_axis("avg_string_length", {16}); diff --git a/cpp/benchmarks/io/parquet/parquet_reader_options.cpp b/cpp/benchmarks/io/parquet/parquet_reader_options.cpp index 07c12ab41fab..af4045aa088c 100644 --- a/cpp/benchmarks/io/parquet/parquet_reader_options.cpp +++ b/cpp/benchmarks/io/parquet/parquet_reader_options.cpp @@ -1,5 +1,5 @@ /* - * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION. + * SPDX-FileCopyrightText: Copyright (c) 2022-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. * SPDX-License-Identifier: Apache-2.0 */ @@ -37,18 +37,21 @@ template void BM_parquet_read_options(nvbench::state& state, nvbench::type_list, nvbench::enum_type, nvbench::enum_type, nvbench::enum_type, + nvbench::enum_type, nvbench::enum_type>) { auto const num_chunks = RowSelection == row_selection::ALL ? 1 : chunked_read_num_chunks; - auto constexpr str_to_categories = ConvertsStrings == converts_strings::YES; - auto constexpr uses_pd_metadata = UsesPandasMetadata == uses_pandas_metadata::YES; + auto constexpr str_to_categories = ConvertsStrings == converts_strings::YES; + auto constexpr output_dict_columns = OutputDict == output_dict::YES; + auto constexpr uses_pd_metadata = UsesPandasMetadata == uses_pandas_metadata::YES; auto const ts_type = cudf::data_type{Timestamp}; @@ -84,6 +87,7 @@ void BM_parquet_read_options(nvbench::state& state, cudf::io::parquet_reader_options::builder(source_sink.make_source_info()) .column_names(cols_to_read) .convert_strings_to_categories(str_to_categories) + .output_dict_columns(output_dict_columns) .use_pandas_metadata(uses_pd_metadata) .timestamp_type(ts_type); @@ -139,12 +143,14 @@ NVBENCH_BENCH_TYPES(BM_parquet_read_options, row_selections, nvbench::enum_type_list, nvbench::enum_type_list, + nvbench::enum_type_list, nvbench::enum_type_list)) .set_name("parquet_read_row_selection") .set_type_axes_names({"column_selection", "row_selection", "str_to_categories", "uses_pandas_metadata", + "output_dict_columns", "timestamp_type"}) .set_min_samples(4) // NOTE: row_selection::ROW_GROUPS reads a fraction of row groups; non-zero @@ -161,12 +167,14 @@ NVBENCH_BENCH_TYPES(BM_parquet_read_options, nvbench::enum_type_list, nvbench::enum_type_list, nvbench::enum_type_list, + nvbench::enum_type_list, nvbench::enum_type_list)) .set_name("parquet_read_column_selection") .set_type_axes_names({"column_selection", "row_selection", "str_to_categories", "uses_pandas_metadata", + "output_dict_columns", "timestamp_type"}) .set_min_samples(4) .add_int64_axis("row_group_size_bytes", {0}) @@ -178,13 +186,37 @@ NVBENCH_BENCH_TYPES( nvbench::enum_type_list, nvbench::enum_type_list, nvbench::enum_type_list, + nvbench::enum_type_list, nvbench::enum_type_list)) .set_name("parquet_read_misc_options") .set_type_axes_names({"column_selection", "row_selection", "str_to_categories", "uses_pandas_metadata", + "output_dict_columns", "timestamp_type"}) .set_min_samples(4) .add_int64_axis("row_group_size_bytes", {0}) .add_int64_axis("row_group_size_rows", {0}); + +// Sweep `output_dict_columns` on/off. Only flat STRING columns are dictionary-transcoded, so this +// case reports read throughput and peak memory for both the direct transcode (YES) and the plain +// STRING materialization (NO). Varying `row_group_size_rows` exercises the single-row-group fast +// path (few, large row groups) versus the multi-row-group concatenate path (many, small ones). +NVBENCH_BENCH_TYPES(BM_parquet_read_options, + NVBENCH_TYPE_AXES(nvbench::enum_type_list, + nvbench::enum_type_list, + nvbench::enum_type_list, + nvbench::enum_type_list, + nvbench::enum_type_list, + nvbench::enum_type_list)) + .set_name("parquet_read_dict_output") + .set_type_axes_names({"column_selection", + "row_selection", + "str_to_categories", + "uses_pandas_metadata", + "output_dict_columns", + "timestamp_type"}) + .set_min_samples(4) + .add_int64_axis("row_group_size_bytes", {0}) + .add_int64_axis("row_group_size_rows", {0, 1'000'000}); From 566eb8feb668db77e672e23e012128ee7d16f614 Mon Sep 17 00:00:00 2001 From: ykiran Date: Wed, 12 Aug 2026 14:57:55 -0700 Subject: [PATCH 3/4] MR feedback --- .../all_cuda-133_arch-x86_64.yaml | 2 +- cpp/benchmarks/CMakeLists.txt | 3 +- .../io/parquet/parquet_reader_options.cpp | 4 +- .../io/parquet/reader_impl_dict_transcode.cu | 44 ++++++++++++++----- 4 files changed, 39 insertions(+), 14 deletions(-) diff --git a/conda/environments/all_cuda-133_arch-x86_64.yaml b/conda/environments/all_cuda-133_arch-x86_64.yaml index fcb76af8b6b2..f6a369816b8d 100644 --- a/conda/environments/all_cuda-133_arch-x86_64.yaml +++ b/conda/environments/all_cuda-133_arch-x86_64.yaml @@ -121,4 +121,4 @@ dependencies: - zstandard - pip: - nvidia-sphinx-theme -name: all_cuda-133_arch-x86_64 +name: cudf_dev \ No newline at end of file diff --git a/cpp/benchmarks/CMakeLists.txt b/cpp/benchmarks/CMakeLists.txt index d03bf72ee9e8..8a458dc8c0ca 100644 --- a/cpp/benchmarks/CMakeLists.txt +++ b/cpp/benchmarks/CMakeLists.txt @@ -297,7 +297,8 @@ ConfigureNVBench( # * parquet reader benchmark ---------------------------------------------------------------------- ConfigureNVBench( 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 + io/parquet/parquet_reader_options.cpp io/parquet/parquet_reader_dict.cpp + io/parquet/reader_common.cpp ) # ################################################################################################## diff --git a/cpp/benchmarks/io/parquet/parquet_reader_options.cpp b/cpp/benchmarks/io/parquet/parquet_reader_options.cpp index af4045aa088c..c0e114c73d6f 100644 --- a/cpp/benchmarks/io/parquet/parquet_reader_options.cpp +++ b/cpp/benchmarks/io/parquet/parquet_reader_options.cpp @@ -219,4 +219,6 @@ NVBENCH_BENCH_TYPES(BM_parquet_read_options, "timestamp_type"}) .set_min_samples(4) .add_int64_axis("row_group_size_bytes", {0}) - .add_int64_axis("row_group_size_rows", {0, 1'000'000}); + // 0 == cuDF default (1,000,000 rows/RG → few, large row groups); 100,000 forces ~10x more, + // smaller row groups, exercising the multi-row-group concatenate path. + .add_int64_axis("row_group_size_rows", {0, 100'000}); diff --git a/cpp/src/io/parquet/reader_impl_dict_transcode.cu b/cpp/src/io/parquet/reader_impl_dict_transcode.cu index 796c48447662..f106698c08a6 100644 --- a/cpp/src/io/parquet/reader_impl_dict_transcode.cu +++ b/cpp/src/io/parquet/reader_impl_dict_transcode.cu @@ -247,6 +247,23 @@ void reader_impl::assemble_dict_transcoded_columns( auto const& pass = *_pass_itm_data; + // Batched keys: every string chunk's dictionary entries live contiguously in + // `pass.str_dict_index` (each `chunk.str_dict_index` is a pointer into that one buffer). So all + // per-chunk keys can be materialized by a single `make_strings_column` instead of one launch per + // chunk. Build that column lazily on first multi-row-group use -- columns that all take the + // single-row-group fast path never need it -- and hand out zero-copy slices below. + std::unique_ptr all_keys; + auto ensure_all_keys = [&]() -> column_view { + if (all_keys == nullptr) { + all_keys = + make_keys_column_from_index_pairs(pass.str_dict_index.data(), + static_cast(pass.str_dict_index.size()), + _stream, + get_current_device_resource_ref()); + } + return all_keys->view(); + }; + // For each eligible input column, collect its chunks in row-group order, build a per-chunk // DICTIONARY32 segment (local 0-based indices + per-chunk keys column), and concatenate. // @@ -355,13 +372,16 @@ void reader_impl::assemble_dict_transcoded_columns( } // Build a per-chunk DICTIONARY32 *view* that aliases the shared decoded INT32 buffer (no - // copy): keys = this chunk's STRING column, indices = `indices_view`. The row range, null - // mask, and null count must all live on the *parent* view (via offset/size), not the indices - // child, because `get_indices_annotated()` rebuilds the indices from the child's `head()` - // plus the parent's offset/size/null_mask -- anything set on the child is ignored. A wrong - // null count (e.g. a hardcoded 0) would silently turn nulls into a valid index once - // `cudf::detail::concatenate` remaps the indices against the unified keys. - std::vector> seg_keys_owners(chunk_indices.size()); + // copy). Per-chunk keys are zero-copy slices of the single batched `all_keys` column: chunk + // `k`'s entries occupy `[key_offset, key_offset + chunk_key_counts[k])` in + // `pass.str_dict_index`, where `key_offset` is recovered from the chunk's stored pointer into + // that buffer. The row range, null mask, and null count must all live on the *parent* view + // (via offset/size), not the indices child, because `get_indices_annotated()` rebuilds the + // indices from the child's `head()` plus the parent's offset/size/null_mask. A wrong null + // count would silently turn nulls into a valid index once `cudf::detail::concatenate` remaps + // the indices against the unified keys. `all_keys` owns the key data and outlives the + // concatenate below, so the slices stay valid. + auto const keys_base = ensure_all_keys(); std::vector dict_segment_views(chunk_indices.size()); std::transform( cuda::counting_iterator{0}, @@ -371,8 +391,10 @@ void reader_impl::assemble_dict_transcoded_columns( auto const chunk_idx = chunk_indices[k]; auto const& chunk = pass.chunks[chunk_idx]; - seg_keys_owners[k] = make_keys_column_from_index_pairs( - chunk.str_dict_index, chunk_key_counts[k], _stream, get_current_device_resource_ref()); + auto const key_offset = + static_cast(chunk.str_dict_index - pass.str_dict_index.data()); + auto const seg_keys = + cudf::detail::slice(keys_base, key_offset, key_offset + chunk_key_counts[k], _stream); auto const seg_begin = chunk_row_offsets[k]; auto const seg_end = chunk_row_offsets[k + 1]; @@ -383,10 +405,10 @@ void reader_impl::assemble_dict_transcoded_columns( indices_view.null_mask(), // shared with indices_view seg_null_counts[k], seg_begin, // reslices shared indices child + null mask - {indices_view, seg_keys_owners[k]->view()}}; + {indices_view, seg_keys}}; }); - // `cudf::detail::concatenate` deduplicates + sorts keys and recomputes indices. + // `cudf::detail::concatenate` deduplicates keys and recomputes indices. out_columns[out_idx] = cudf::detail::concatenate(dict_segment_views, _stream, _mr); }); } From c23c5d6e266344c7dde1357b66b74840c88f1403 Mon Sep 17 00:00:00 2001 From: ykiran Date: Fri, 14 Aug 2026 13:40:18 -0700 Subject: [PATCH 4/4] Fix mangled rebase --- .../all_cuda-133_arch-x86_64.yaml | 2 +- cpp/benchmarks/CMakeLists.txt | 3 +- .../io/parquet/reader_impl_dict_transcode.cu | 44 +++++-------------- 3 files changed, 13 insertions(+), 36 deletions(-) diff --git a/conda/environments/all_cuda-133_arch-x86_64.yaml b/conda/environments/all_cuda-133_arch-x86_64.yaml index f6a369816b8d..fcb76af8b6b2 100644 --- a/conda/environments/all_cuda-133_arch-x86_64.yaml +++ b/conda/environments/all_cuda-133_arch-x86_64.yaml @@ -121,4 +121,4 @@ dependencies: - zstandard - pip: - nvidia-sphinx-theme -name: cudf_dev \ No newline at end of file +name: all_cuda-133_arch-x86_64 diff --git a/cpp/benchmarks/CMakeLists.txt b/cpp/benchmarks/CMakeLists.txt index 8a458dc8c0ca..d03bf72ee9e8 100644 --- a/cpp/benchmarks/CMakeLists.txt +++ b/cpp/benchmarks/CMakeLists.txt @@ -297,8 +297,7 @@ ConfigureNVBench( # * parquet reader benchmark ---------------------------------------------------------------------- ConfigureNVBench( PARQUET_READER_NVBENCH io/parquet/parquet_reader_input.cpp io/parquet/parquet_reader_encoding.cpp - io/parquet/parquet_reader_options.cpp io/parquet/parquet_reader_dict.cpp - io/parquet/reader_common.cpp + io/parquet/parquet_reader_options.cpp io/parquet/reader_common.cpp ) # ################################################################################################## diff --git a/cpp/src/io/parquet/reader_impl_dict_transcode.cu b/cpp/src/io/parquet/reader_impl_dict_transcode.cu index f106698c08a6..796c48447662 100644 --- a/cpp/src/io/parquet/reader_impl_dict_transcode.cu +++ b/cpp/src/io/parquet/reader_impl_dict_transcode.cu @@ -247,23 +247,6 @@ void reader_impl::assemble_dict_transcoded_columns( auto const& pass = *_pass_itm_data; - // Batched keys: every string chunk's dictionary entries live contiguously in - // `pass.str_dict_index` (each `chunk.str_dict_index` is a pointer into that one buffer). So all - // per-chunk keys can be materialized by a single `make_strings_column` instead of one launch per - // chunk. Build that column lazily on first multi-row-group use -- columns that all take the - // single-row-group fast path never need it -- and hand out zero-copy slices below. - std::unique_ptr all_keys; - auto ensure_all_keys = [&]() -> column_view { - if (all_keys == nullptr) { - all_keys = - make_keys_column_from_index_pairs(pass.str_dict_index.data(), - static_cast(pass.str_dict_index.size()), - _stream, - get_current_device_resource_ref()); - } - return all_keys->view(); - }; - // For each eligible input column, collect its chunks in row-group order, build a per-chunk // DICTIONARY32 segment (local 0-based indices + per-chunk keys column), and concatenate. // @@ -372,16 +355,13 @@ void reader_impl::assemble_dict_transcoded_columns( } // Build a per-chunk DICTIONARY32 *view* that aliases the shared decoded INT32 buffer (no - // copy). Per-chunk keys are zero-copy slices of the single batched `all_keys` column: chunk - // `k`'s entries occupy `[key_offset, key_offset + chunk_key_counts[k])` in - // `pass.str_dict_index`, where `key_offset` is recovered from the chunk's stored pointer into - // that buffer. The row range, null mask, and null count must all live on the *parent* view - // (via offset/size), not the indices child, because `get_indices_annotated()` rebuilds the - // indices from the child's `head()` plus the parent's offset/size/null_mask. A wrong null - // count would silently turn nulls into a valid index once `cudf::detail::concatenate` remaps - // the indices against the unified keys. `all_keys` owns the key data and outlives the - // concatenate below, so the slices stay valid. - auto const keys_base = ensure_all_keys(); + // copy): keys = this chunk's STRING column, indices = `indices_view`. The row range, null + // mask, and null count must all live on the *parent* view (via offset/size), not the indices + // child, because `get_indices_annotated()` rebuilds the indices from the child's `head()` + // plus the parent's offset/size/null_mask -- anything set on the child is ignored. A wrong + // null count (e.g. a hardcoded 0) would silently turn nulls into a valid index once + // `cudf::detail::concatenate` remaps the indices against the unified keys. + std::vector> seg_keys_owners(chunk_indices.size()); std::vector dict_segment_views(chunk_indices.size()); std::transform( cuda::counting_iterator{0}, @@ -391,10 +371,8 @@ void reader_impl::assemble_dict_transcoded_columns( auto const chunk_idx = chunk_indices[k]; auto const& chunk = pass.chunks[chunk_idx]; - auto const key_offset = - static_cast(chunk.str_dict_index - pass.str_dict_index.data()); - auto const seg_keys = - cudf::detail::slice(keys_base, key_offset, key_offset + chunk_key_counts[k], _stream); + seg_keys_owners[k] = make_keys_column_from_index_pairs( + chunk.str_dict_index, chunk_key_counts[k], _stream, get_current_device_resource_ref()); auto const seg_begin = chunk_row_offsets[k]; auto const seg_end = chunk_row_offsets[k + 1]; @@ -405,10 +383,10 @@ void reader_impl::assemble_dict_transcoded_columns( indices_view.null_mask(), // shared with indices_view seg_null_counts[k], seg_begin, // reslices shared indices child + null mask - {indices_view, seg_keys}}; + {indices_view, seg_keys_owners[k]->view()}}; }); - // `cudf::detail::concatenate` deduplicates keys and recomputes indices. + // `cudf::detail::concatenate` deduplicates + sorts keys and recomputes indices. out_columns[out_idx] = cudf::detail::concatenate(dict_segment_views, _stream, _mr); }); }