From 2f5a101ee57043b9b4f9bf26957acea82ac471b4 Mon Sep 17 00:00:00 2001 From: Krisztian Szucs Date: Thu, 1 Oct 2026 19:26:07 +0200 Subject: [PATCH] [C++][Parquet] Speed up content-defined chunking with a local gear hash The rolling hash state was stored to memory for every hashed byte, because the values read through byte pointers may alias it. GetChunks() now works on a local copy of a GearHash and stores it back once, keeping the state in registers: chunking is 1.2 to 1.6 times faster on a single thread and multi-threaded writes about 2 times faster. BM_WriteContentDefinedChunking measures both. --- .../parquet/arrow/reader_writer_benchmark.cc | 42 ++++++ cpp/src/parquet/chunker_internal.cc | 131 ++++++++++-------- 2 files changed, 119 insertions(+), 54 deletions(-) diff --git a/cpp/src/parquet/arrow/reader_writer_benchmark.cc b/cpp/src/parquet/arrow/reader_writer_benchmark.cc index 2f288fd2eb0f..a5e84243b7ff 100644 --- a/cpp/src/parquet/arrow/reader_writer_benchmark.cc +++ b/cpp/src/parquet/arrow/reader_writer_benchmark.cc @@ -273,6 +273,48 @@ BENCHMARK(BM_WriteBinaryColumn) ->Args({50, kInfiniteUniqueValues}) ->Args({99, kInfiniteUniqueValues}); +// Writes int32 columns with content-defined chunking into a single row group, with +// use_threads the columns are chunked by different threads in parallel +static void BM_WriteContentDefinedChunking(::benchmark::State& state) { + const auto num_columns = static_cast(state.range(0)); + const bool use_threads = state.range(1) != 0; + constexpr int64_t kNumRows = 1024 * 1024; + + ::arrow::random::RandomArrayGenerator generator(/*seed=*/500); + ::arrow::FieldVector fields; + ::arrow::ArrayVector columns; + for (int i = 0; i < num_columns; i++) { + fields.push_back(::arrow::field("column" + std::to_string(i), ::arrow::int32(), + /*nullable=*/false)); + columns.push_back(generator.Int32(kNumRows, /*min=*/0, /*max=*/1 << 30)); + } + auto schema = ::arrow::schema(fields); + auto batch = ::arrow::RecordBatch::Make(schema, kNumRows, columns); + auto properties = WriterProperties::Builder() + .enable_content_defined_chunking() + ->disable_dictionary() + ->build(); + auto arrow_properties = + ArrowWriterProperties::Builder().set_use_threads(use_threads)->build(); + + for (auto _ : state) { + auto output = CreateOutputStream(); + auto writer = arrow::FileWriter::Open(*schema, ::arrow::default_memory_pool(), output, + properties, arrow_properties); + EXIT_NOT_OK(writer.status()); + EXIT_NOT_OK((*writer)->WriteRecordBatch(*batch)); + EXIT_NOT_OK((*writer)->Close()); + } + state.SetBytesProcessed(state.iterations() * num_columns * kNumRows * + static_cast(sizeof(int32_t))); +} + +BENCHMARK(BM_WriteContentDefinedChunking) + ->ArgNames({"columns", "use_threads"}) + ->Args({16, 0}) + ->Args({16, 1}) + ->UseRealTime(); + template struct Examples { static constexpr std::array values() { return {127, 128}; } diff --git a/cpp/src/parquet/chunker_internal.cc b/cpp/src/parquet/chunker_internal.cc index a6f161fafeb8..fc5768b28d7e 100644 --- a/cpp/src/parquet/chunker_internal.cc +++ b/cpp/src/parquet/chunker_internal.cc @@ -113,14 +113,11 @@ uint64_t CalculateMask(int64_t min_chunk_size, int64_t max_chunk_size, int norm_ } } -} // namespace - -class ContentDefinedChunker::Impl { +/// The gear hash of a column together with the state deciding the chunk boundaries +class GearHash { public: - Impl(const LevelInfo& level_info, int64_t min_chunk_size, int64_t max_chunk_size, - int norm_level) - : level_info_(level_info), - min_chunk_size_(min_chunk_size), + GearHash(int64_t min_chunk_size, int64_t max_chunk_size, int norm_level) + : min_chunk_size_(min_chunk_size), max_chunk_size_(max_chunk_size), rolling_hash_mask_(CalculateMask(min_chunk_size, max_chunk_size, norm_level)) {} @@ -202,6 +199,40 @@ class ContentDefinedChunker::Impl { return false; } + private: + // Minimum chunk size in bytes, the rolling hash will not be updated until this size is + // reached for each chunk. Note that all data sent through the hash function is counted + // towards the chunk size, including definition and repetition levels. + int64_t min_chunk_size_; + int64_t max_chunk_size_; + // The mask to match the rolling hash against to determine if a new chunk should be + // created. The mask is calculated based on min/max chunk size and the normalization + // level. + uint64_t rolling_hash_mask_; + + // Whether the rolling hash has matched the mask since the last chunk creation. This + // flag is set true by the Roll() function when the mask is matched and reset to false + // by NeedNewChunk() method. + bool has_matched_ = false; + // The current run of the rolling hash, used to normalize the chunk size distribution + // by requiring multiple consecutive matches to create a new chunk. + int8_t nth_run_ = 0; + // Current chunk size in bytes, reset to 0 when a new chunk is created. + int64_t chunk_size_ = 0; + // Rolling hash state, never reset only initialized once for the entire column. + uint64_t rolling_hash_ = 0; +}; + +} // namespace + +class ContentDefinedChunker::Impl { + public: + Impl(const LevelInfo& level_info, int64_t min_chunk_size, int64_t max_chunk_size, + int norm_level) + : level_info_(level_info), gearhash_(min_chunk_size, max_chunk_size, norm_level) {} + + uint64_t GetRollingHashMask() const { return gearhash_.GetRollingHashMask(); } + void ValidateChunks(const std::vector& chunks, int64_t num_levels) const { // chunks must be non-empty and monotonic increasing ARROW_DCHECK(!chunks.empty()); @@ -249,6 +280,9 @@ class ContentDefinedChunker::Impl { // requirements, we create a new chunk. See the `NeedNewChunk()` method for more // details. std::vector chunks; + // Roll a local copy of the hash so the compiler can keep it in registers instead of + // storing it to memory for every hashed byte + GearHash gearhash = gearhash_; int64_t offset; int64_t prev_offset = 0; int64_t prev_value_offset = 0; @@ -258,8 +292,8 @@ class ContentDefinedChunker::Impl { if (!has_rep_levels && !has_def_levels) { // fastest path for non-nested non-null data for (offset = 0; offset < num_levels; ++offset) { - RollValue(offset); - if (NeedNewChunk()) { + RollValue(gearhash, offset); + if (gearhash.NeedNewChunk()) { chunks.push_back({prev_offset, prev_offset, offset - prev_offset}); prev_offset = offset; } @@ -272,11 +306,11 @@ class ContentDefinedChunker::Impl { for (int64_t offset = 0; offset < num_levels; ++offset) { def_level = def_levels[offset]; - Roll(&def_level); + gearhash.Roll(&def_level); if (def_level == level_info_.def_level) { - RollValue(offset); + RollValue(gearhash, offset); } - if (NeedNewChunk()) { + if (gearhash.NeedNewChunk()) { chunks.push_back({prev_offset, prev_offset, offset - prev_offset}); prev_offset = offset; } @@ -293,13 +327,13 @@ class ContentDefinedChunker::Impl { def_level = def_levels[offset]; rep_level = rep_levels[offset]; - Roll(&def_level); - Roll(&rep_level); + gearhash.Roll(&def_level); + gearhash.Roll(&rep_level); if (def_level == level_info_.def_level) { - RollValue(value_offset); + RollValue(gearhash, value_offset); } - if (rep_level == 0 && NeedNewChunk()) { + if (rep_level == 0 && gearhash.NeedNewChunk()) { // if we are at a record boundary and need a new chunk, we create a new chunk auto levels_to_write = offset - prev_offset; if (levels_to_write > 0) { @@ -314,6 +348,7 @@ class ContentDefinedChunker::Impl { } } } + gearhash_ = gearhash; // add the last chunk if we have any levels left if (prev_offset < num_levels) { @@ -333,9 +368,10 @@ class ContentDefinedChunker::Impl { const uint8_t* raw_values = values.data()->GetValues(/*i=*/1, /*absolute_offset=*/0) + values.offset() * kByteWidth; - return Calculate(def_levels, rep_levels, num_levels, [&](int64_t i) { - return Roll(&raw_values[i * kByteWidth]); - }); + return Calculate(def_levels, rep_levels, num_levels, + [&](GearHash& gearhash, int64_t i) { + return gearhash.Roll(&raw_values[i * kByteWidth]); + }); } template @@ -343,22 +379,24 @@ class ContentDefinedChunker::Impl { const int16_t* rep_levels, int64_t num_levels, const ::arrow::Array& values) { const auto& array = checked_cast(values); - return Calculate(def_levels, rep_levels, num_levels, [&](int64_t i) { - typename ArrayType::offset_type length; - const uint8_t* value = array.GetValue(i, &length); - Roll(value, length); - }); + return Calculate(def_levels, rep_levels, num_levels, + [&](GearHash& gearhash, int64_t i) { + typename ArrayType::offset_type length; + const uint8_t* value = array.GetValue(i, &length); + gearhash.Roll(value, length); + }); } std::vector CalculateBinaryView(const int16_t* def_levels, const int16_t* rep_levels, int64_t num_levels, const ::arrow::Array& values) { const auto& array = checked_cast(values); - return Calculate(def_levels, rep_levels, num_levels, [&](int64_t i) { - const std::string_view value = array.GetView(i); - Roll(reinterpret_cast(value.data()), - static_cast(value.size())); - }); + return Calculate(def_levels, rep_levels, num_levels, + [&](GearHash& gearhash, int64_t i) { + const std::string_view value = array.GetView(i); + gearhash.Roll(reinterpret_cast(value.data()), + static_cast(value.size())); + }); } std::vector GetChunks(const int16_t* def_levels, const int16_t* rep_levels, @@ -366,16 +404,19 @@ class ContentDefinedChunker::Impl { auto handle_type = [&](auto&& type) -> std::vector { using ArrowType = std::decay_t; if constexpr (ArrowType::type_id == ::arrow::Type::NA) { - return Calculate(def_levels, rep_levels, num_levels, [](int64_t) {}); + return Calculate(def_levels, rep_levels, num_levels, [](GearHash&, int64_t) {}); } else if constexpr (ArrowType::type_id == ::arrow::Type::BOOL) { const auto& array = static_cast(values); - return Calculate(def_levels, rep_levels, num_levels, - [&](int64_t i) { Roll(array.Value(i)); }); + return Calculate( + def_levels, rep_levels, num_levels, + [&](GearHash& gearhash, int64_t i) { gearhash.Roll(array.Value(i)); }); } else if constexpr (ArrowType::type_id == ::arrow::Type::FIXED_SIZE_BINARY) { const auto& array = static_cast(values); const auto byte_width = array.byte_width(); return Calculate(def_levels, rep_levels, num_levels, - [&](int64_t i) { Roll(array.GetValue(i), byte_width); }); + [&](GearHash& gearhash, int64_t i) { + gearhash.Roll(array.GetValue(i), byte_width); + }); } else if constexpr (ArrowType::type_id == ::arrow::Type::EXTENSION) { const auto& array = static_cast(values); return GetChunks(def_levels, rep_levels, num_levels, *array.storage()); @@ -408,27 +449,9 @@ class ContentDefinedChunker::Impl { private: // The column's level information const internal::LevelInfo level_info_; - // Minimum chunk size in bytes, the rolling hash will not be updated until this size is - // reached for each chunk. Note that all data sent through the hash function is counted - // towards the chunk size, including definition and repetition levels. - const int64_t min_chunk_size_; - const int64_t max_chunk_size_; - // The mask to match the rolling hash against to determine if a new chunk should be - // created. The mask is calculated based on min/max chunk size and the normalization - // level. - const uint64_t rolling_hash_mask_; - - // Whether the rolling hash has matched the mask since the last chunk creation. This - // flag is set true by the Roll() function when the mask is matched and reset to false - // by NeedNewChunk() method. - bool has_matched_ = false; - // The current run of the rolling hash, used to normalize the chunk size distribution - // by requiring multiple consecutive matches to create a new chunk. - int8_t nth_run_ = 0; - // Current chunk size in bytes, reset to 0 when a new chunk is created. - int64_t chunk_size_ = 0; - // Rolling hash state, never reset only initialized once for the entire column. - uint64_t rolling_hash_ = 0; + // The gear hash carried over between the calls, so the chunking continues across + // the pages and the row groups + GearHash gearhash_; }; ContentDefinedChunker::ContentDefinedChunker(const LevelInfo& level_info,