From dd5c36aedce84ac4becda26a9939468cb640264a Mon Sep 17 00:00:00 2001 From: Marcin Krystianc Date: Mon, 14 Sep 2026 12:40:33 +0200 Subject: [PATCH 1/2] Make DELTA_BYTE_ARRAY value reconstruction storage-agnostic --- cpp/src/parquet/decoder.cc | 276 ++++++++++++++++---------- cpp/src/parquet/encoding_benchmark.cc | 113 +++++++++++ 2 files changed, 285 insertions(+), 104 deletions(-) diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index 47f2b46ebaac..07185d28f53f 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -1782,9 +1782,30 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl { int Decode(ByteArray* buffer, int max_values) override { // Decode up to `max_values` strings into an internal buffer // and reference them into `buffer`. + const int32_t* lengths = nullptr; + const uint8_t* data_ptr = nullptr; + max_values = NextBatch(max_values, &lengths, &data_ptr); + for (int i = 0; i < max_values; ++i) { + buffer[i].len = static_cast(lengths[i]); + buffer[i].ptr = data_ptr; + data_ptr += buffer[i].len; + } + return max_values; + } + + // Advance over up to `max_values` values and expose them as a view: the + // per-value lengths, and a pointer to the start of their (contiguous) data. + // Returns the number of values available. + // + // The lengths are already fully decoded by `DecodeLengths`, and the values are + // stored contiguously in the page, so callers that only need the bytes can walk + // this view directly instead of materializing a `ByteArray` per value. + int NextBatch(int max_values, const int32_t** lengths, const uint8_t** data) { max_values = std::min(max_values, num_valid_values_); DCHECK_GE(max_values, 0); if (max_values == 0) { + *lengths = nullptr; + *data = nullptr; return 0; } @@ -1796,7 +1817,6 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl { if (ARROW_PREDICT_FALSE(len < 0)) { throw ParquetException("negative string delta length"); } - buffer[i].len = len; if (AddWithOverflow(data_size, len, &data_size)) { throw ParquetException("excess expansion in DELTA_(LENGTH_)BYTE_ARRAY"); } @@ -1805,11 +1825,8 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl { if (ARROW_PREDICT_FALSE(!decoder_->Advance(8 * static_cast(data_size)))) { ParquetException::EofException(); } - const uint8_t* data_ptr = data_ + bytes_offset; - for (int i = 0; i < max_values; ++i) { - buffer[i].ptr = data_ptr; - data_ptr += buffer[i].len; - } + *lengths = length_ptr; + *data = data_ + bytes_offset; this->num_values_ -= max_values; num_valid_values_ -= max_values; return max_values; @@ -2082,93 +2099,154 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { } protected: - template - static void BuildBufferInternal(const int32_t* prefix_len_ptr, int i, ByteArray* buffer, - std::string_view* prefix, uint8_t** data_ptr) { - if (ARROW_PREDICT_FALSE(static_cast(prefix_len_ptr[i]) > prefix->length())) { - throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY"); - } - // For now, `buffer` points to string suffixes, and the suffix decoder - // ensures that the suffix data has sufficient lifetime. - if (prefix_len_ptr[i] == 0) { - // prefix is empty: buffer[i] already points to the suffix. - *prefix = std::string_view{buffer[i]}; - return; + struct ByteArrayOutput { + // `buffer` is used both as the suffix decode target and as the output. + ByteArray* buffer; + ResizableBuffer* buffered_data; + int type_length; + uint8_t* data_ptr = nullptr; + const uint8_t* data_end = nullptr; + + int FetchSuffixes(DeltaLengthByteArrayDecoder* suffix_decoder, int max_values) { + // Decode the suffixes in place; `Emit` reads each one before overwriting it + // with the reconstructed value. + return suffix_decoder->Decode(buffer, max_values); + } + + void Prepare(const int32_t* prefix_len_ptr, int max_values) { + int64_t data_size = 0; + for (int i = 0; i < max_values; ++i) { + if (prefix_len_ptr[i] == 0) { + // We don't need to copy the suffix if the prefix length is 0. + continue; + } + if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) { + throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY"); + } + if (buffer[i].len == 0 && i != 0) { + // We don't need to copy the prefix if the suffix length is 0 + // and this is not the first run (that is, the prefix doesn't point + // to the mutable `last_value_`). + continue; + } + if (ARROW_PREDICT_FALSE( + AddWithOverflow(data_size, prefix_len_ptr[i], &data_size) || + AddWithOverflow(data_size, buffer[i].len, &data_size))) { + throw ParquetException("excess expansion in DELTA_BYTE_ARRAY"); + } + } + PARQUET_THROW_NOT_OK(buffered_data->Resize(data_size)); + data_ptr = buffered_data->mutable_data(); + data_end = data_ptr + data_size; } - DCHECK_EQ(is_first_run, i == 0); - if constexpr (!is_first_run) { - if (buffer[i].len == 0) { + + std::string_view Emit(int i, int32_t prefix_len, std::string_view prefix) { + // Read the suffix before `buffer[i]` is overwritten with the output. + const ByteArray suffix = buffer[i]; + if (ARROW_PREDICT_FALSE(static_cast(prefix_len) > prefix.length())) { + throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY"); + } + if constexpr (std::is_same_v) { + // The value is `prefix_len` bytes of the previous value followed by the + // suffix, whichever branch below produces it. + if (ARROW_PREDICT_FALSE(static_cast(prefix_len) + suffix.len != + type_length)) { + throw ParquetException("FLBA type requires fixed-length ", type_length, + " but got ", prefix_len + suffix.len); + } + } + if (prefix_len == 0) { + // prefix is empty: buffer[i] already points to the suffix. + return std::string_view{buffer[i]}; + } + if (i != 0 && suffix.len == 0) { // suffix is empty: buffer[i] can simply point to the prefix. // This is not possible for the first run since the prefix // would point to the mutable `last_value_`. - *prefix = prefix->substr(0, prefix_len_ptr[i]); - buffer[i] = ByteArray(*prefix); - return; + prefix = prefix.substr(0, prefix_len); + buffer[i] = ByteArray(prefix); + return prefix; } + // Both prefix and suffix are non-empty, so decode the string into + // `data_ptr`. + const uint32_t full_len = suffix.len + static_cast(prefix_len); + DCHECK_LE(data_ptr + full_len, data_end); + memcpy(data_ptr, prefix.data(), prefix_len); + memcpy(data_ptr + prefix_len, suffix.ptr, suffix.len); + buffer[i].ptr = data_ptr; + buffer[i].len = full_len; + data_ptr += full_len; + return std::string_view{buffer[i]}; } - // Both prefix and suffix are non-empty, so we need to decode the string - // into `data_ptr`. - // 1. Copy the prefix - memcpy(*data_ptr, prefix->data(), prefix_len_ptr[i]); - // 2. Copy the suffix. - memcpy(*data_ptr + prefix_len_ptr[i], buffer[i].ptr, buffer[i].len); - // 3. Point buffer[i] to the decoded string. - buffer[i].ptr = *data_ptr; - buffer[i].len += prefix_len_ptr[i]; - *data_ptr += buffer[i].len; - *prefix = std::string_view{buffer[i]}; - } + }; - int GetInternal(ByteArray* buffer, int max_values) { - // Decode up to `max_values` strings into an internal buffer - // and reference them into `buffer`. + // Output that writes fixed-length values contiguously into a caller-provided + // buffer. Used by the FLBA dense decode path; avoids `buffered_data_` and any + // temporary per-value `ByteArray` storage. + struct DenseOutput { + uint8_t* out; + int type_length; + // View of the suffixes still to be consumed: their lengths, and a cursor into + // their contiguous data. No per-value pointer is ever materialized. + const int32_t* suffix_lengths = nullptr; + const uint8_t* suffix_data = nullptr; + + int FetchSuffixes(DeltaLengthByteArrayDecoder* suffix_decoder, int max_values) { + return suffix_decoder->NextBatch(max_values, &suffix_lengths, &suffix_data); + } + + void Prepare(const int32_t* prefix_len_ptr, int max_values) {} + + std::string_view Emit(int i, int32_t prefix_len, std::string_view prefix) { + const int32_t suffix_len = suffix_lengths[i]; + const uint8_t* suffix_ptr = suffix_data; + suffix_data += suffix_len; + + if (ARROW_PREDICT_FALSE(static_cast(prefix_len) > prefix.length())) { + throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY"); + } + // Each reconstructed FLBA value must be exactly `type_length` bytes. + if (ARROW_PREDICT_FALSE(static_cast(prefix_len) + suffix_len != + type_length)) { + throw ParquetException("FLBA type requires fixed-length ", type_length, + " but got ", prefix_len + suffix_len); + } + // Copy the prefix and the suffix straight into the caller's buffer. + uint8_t* dst = out + static_cast(i) * type_length; + memcpy(dst, prefix.data(), prefix_len); + memcpy(dst + prefix_len, suffix_ptr, suffix_len); + // The next value's prefix references this value, which now lives in the + // caller's buffer. + return std::string_view{reinterpret_cast(dst), + static_cast(type_length)}; + } + }; + + template + int GetInternalImpl(Output& output, int max_values) { + // Decode up to `max_values` values, delegating materialization to `output`. max_values = std::min(max_values, num_valid_values_); if (max_values == 0) { return max_values; } - int suffix_read = suffix_decoder_.Decode(buffer, max_values); + const int suffix_read = output.FetchSuffixes(&suffix_decoder_, max_values); if (ARROW_PREDICT_FALSE(suffix_read != max_values)) { ParquetException::EofException("Read " + std::to_string(suffix_read) + ", expecting " + std::to_string(max_values) + " from suffix decoder"); } - int64_t data_size = 0; const int32_t* prefix_len_ptr = buffered_prefix_length_->data_as() + prefix_len_offset_; - for (int i = 0; i < max_values; ++i) { - if (prefix_len_ptr[i] == 0) { - // We don't need to copy the suffix if the prefix length is 0. - continue; - } - if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) { - throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY"); - } - if (buffer[i].len == 0 && i != 0) { - // We don't need to copy the prefix if the suffix length is 0 - // and this is not the first run (that is, the prefix doesn't point - // to the mutable `last_value_`). - continue; - } - if (ARROW_PREDICT_FALSE(AddWithOverflow(data_size, prefix_len_ptr[i], &data_size) || - AddWithOverflow(data_size, buffer[i].len, &data_size))) { - throw ParquetException("excess expansion in DELTA_BYTE_ARRAY"); - } - } - PARQUET_THROW_NOT_OK(buffered_data_->Resize(data_size)); + + output.Prepare(prefix_len_ptr, max_values); std::string_view prefix{last_value_}; - uint8_t* data_ptr = buffered_data_->mutable_data(); - if (max_values > 0) { - BuildBufferInternal(prefix_len_ptr, 0, buffer, &prefix, - &data_ptr); - } - for (int i = 1; i < max_values; ++i) { - BuildBufferInternal(prefix_len_ptr, i, buffer, &prefix, - &data_ptr); + for (int i = 0; i < max_values; ++i) { + prefix = output.Emit(i, prefix_len_ptr[i], prefix); } - DCHECK_EQ(data_ptr - buffered_data_->mutable_data(), data_size); + prefix_len_offset_ += max_values; this->num_values_ -= max_values; num_valid_values_ -= max_values; @@ -2178,19 +2256,31 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { last_value_in_previous_page_ = last_value_; } - if constexpr (std::is_same_v) { - // Checks all values - for (int i = 0; i < max_values; i++) { - if (buffer[i].len != static_cast(this->type_length_)) { - throw ParquetException("FLBA type requires fixed-length ", this->type_length_, - " but got ", buffer[i].len); - } - } - } - return max_values; } + int GetInternal(ByteArray* buffer, int max_values) { + ByteArrayOutput output{buffer, buffered_data_.get(), this->type_length_}; + return GetInternalImpl(output, max_values); + } + + int DecodeDense(uint8_t* out, int max_values) { + DenseOutput output{out, this->type_length_}; + return GetInternalImpl(output, max_values); + } + + int DecodePointers(FixedLenByteArray* buffer, int max_values) { + PARQUET_THROW_NOT_OK( + buffered_data_->Resize(static_cast(max_values) * this->type_length_, + /*shrink_to_fit=*/false)); + uint8_t* values = buffered_data_->mutable_data(); + const int decoded_values_size = DecodeDense(values, max_values); + for (int i = 0; i < decoded_values_size; ++i) { + buffer[i].ptr = values + static_cast(i) * this->type_length_; + } + return decoded_values_size; + } + Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset, typename EncodingTraits::Accumulator* out, @@ -2268,35 +2358,13 @@ class DeltaByteArrayFLBADecoder : public DeltaByteArrayDecoderImpl, using Base::pool_; int Decode(FixedLenByteArray* buffer, int max_values) override { - // GetInternal currently only support ByteArray. - std::vector decode_byte_array(max_values); - const int decoded_values_size = GetInternal(decode_byte_array.data(), max_values); - const uint32_t type_length = static_cast(this->type_length_); - - for (int i = 0; i < decoded_values_size; i++) { - if (ARROW_PREDICT_FALSE(decode_byte_array[i].len != type_length)) { - throw ParquetException("Fixed length byte array length mismatch"); - } - buffer[i].ptr = decode_byte_array[i].ptr; - } - return decoded_values_size; + return this->DecodePointers(buffer, max_values); } // Same internal decode as above, but copy the bytes contiguously into the // caller's buffer instead of materializing per-value pointers. int Decode(uint8_t* buffer, int max_values) override { - std::vector decode_byte_array(max_values); - const int decoded_values_size = GetInternal(decode_byte_array.data(), max_values); - const uint32_t type_length = static_cast(this->type_length_); - - for (int i = 0; i < decoded_values_size; i++) { - if (ARROW_PREDICT_FALSE(decode_byte_array[i].len != type_length)) { - throw ParquetException("Fixed length byte array length mismatch"); - } - memcpy(buffer + static_cast(i) * type_length, decode_byte_array[i].ptr, - type_length); - } - return decoded_values_size; + return this->DecodeDense(buffer, max_values); } }; diff --git a/cpp/src/parquet/encoding_benchmark.cc b/cpp/src/parquet/encoding_benchmark.cc index bea1a5807a2a..52341917b20d 100644 --- a/cpp/src/parquet/encoding_benchmark.cc +++ b/cpp/src/parquet/encoding_benchmark.cc @@ -1012,6 +1012,119 @@ static void ByteArrayDeltaCustomArguments(benchmark::internal::Benchmark* b) { BENCHMARK(BM_DeltaEncodingByteArray)->Apply(ByteArrayDeltaCustomArguments); BENCHMARK(BM_DeltaDecodingByteArray)->Apply(ByteArrayDeltaCustomArguments); +struct DeltaFLBAState { + int32_t type_length; + int32_t array_length; + int32_t total_data_size; + double prefixed_probability; + std::vector buf; + std::shared_ptr descr; + + explicit DeltaFLBAState(const benchmark::State& state) + : type_length(static_cast(state.range(0))), + array_length(static_cast(state.range(1))), + total_data_size(type_length * array_length), + prefixed_probability(state.range(2) / 100.0), + descr(std::make_shared( + PrimitiveNode::Make("flba", Repetition::REQUIRED, Type::FIXED_LEN_BYTE_ARRAY, + ConvertedType::NONE, type_length), + /*max_definition_level=*/0, /*max_repetition_level=*/0)) {} + + std::vector MakeRandomFLBA(uint32_t seed) { + std::default_random_engine gen(seed); + std::uniform_int_distribution dist_byte(0, 255); + std::bernoulli_distribution dist_has_prefix(prefixed_probability); + std::uniform_real_distribution dist_prefix_length(0, 1); + + std::vector out(array_length); + buf.resize(total_data_size); + auto buf_ptr = buf.data(); + + for (int32_t i = 0; i < array_length; ++i) { + out[i] = FLBA(buf_ptr); + + bool do_prefix = i > 0 && dist_has_prefix(gen); + int prefix_len = 0; + if (do_prefix) { + prefix_len = static_cast(std::ceil(type_length * dist_prefix_length(gen))); + } + for (int j = 0; j < prefix_len; ++j) { + buf_ptr[j] = out[i - 1].ptr[j]; + } + for (int j = prefix_len; j < type_length; ++j) { + buf_ptr[j] = static_cast(dist_byte(gen)); + } + buf_ptr += type_length; + } + return out; + } +}; + +static void BM_DeltaDecodingFLBA(benchmark::State& state) { + DeltaFLBAState delta_state(state); + const ColumnDescriptor* descr = delta_state.descr.get(); + std::vector values = delta_state.MakeRandomFLBA(/*seed=*/42); + + auto encoder = MakeTypedEncoder(Encoding::DELTA_BYTE_ARRAY, + /*use_dictionary=*/false, descr); + encoder->Put(values.data(), static_cast(values.size())); + std::shared_ptr buf = encoder->FlushValues(); + + auto decoder = MakeTypedDecoder(Encoding::DELTA_BYTE_ARRAY, descr); + for (auto _ : state) { + decoder->SetData(delta_state.array_length, buf->data(), + static_cast(buf->size())); + decoder->Decode(values.data(), static_cast(values.size())); + ::benchmark::DoNotOptimize(values); + } + state.SetItemsProcessed(state.iterations() * delta_state.array_length); + state.SetBytesProcessed(state.iterations() * delta_state.total_data_size); + state.counters["compression_ratio"] = + static_cast(delta_state.total_data_size) / buf->size(); +} + +// Same as above, but decoding densely into a caller-provided buffer instead of +// producing one pointer per value. +static void BM_DeltaDecodingDenseFLBA(benchmark::State& state) { + DeltaFLBAState delta_state(state); + const ColumnDescriptor* descr = delta_state.descr.get(); + std::vector values = delta_state.MakeRandomFLBA(/*seed=*/42); + + auto encoder = MakeTypedEncoder(Encoding::DELTA_BYTE_ARRAY, + /*use_dictionary=*/false, descr); + encoder->Put(values.data(), static_cast(values.size())); + std::shared_ptr buf = encoder->FlushValues(); + + auto decoder = MakeTypedDecoder(Encoding::DELTA_BYTE_ARRAY, descr); + auto* flba_decoder = dynamic_cast(decoder.get()); + std::vector decoded(delta_state.total_data_size); + for (auto _ : state) { + decoder->SetData(delta_state.array_length, buf->data(), + static_cast(buf->size())); + flba_decoder->Decode(decoded.data(), delta_state.array_length); + ::benchmark::DoNotOptimize(decoded); + } + state.SetItemsProcessed(state.iterations() * delta_state.array_length); + state.SetBytesProcessed(state.iterations() * delta_state.total_data_size); + state.counters["compression_ratio"] = + static_cast(delta_state.total_data_size) / buf->size(); +} + +static void FLBADeltaCustomArguments(benchmark::internal::Benchmark* b) { + // 2 bytes is float16, 16 bytes is decimal128. + for (int type_length : {2, 16}) { + for (int batch_size : {512, 2048}) { + for (int prefixed_percent : {10, 90, 99}) { + b->Args({type_length, batch_size, prefixed_percent}); + } + } + } + b->ArgNames({"type-length", "batch-size", "prefixed-percent"}); +} + +BENCHMARK(BM_DeltaDecodingFLBA)->Apply(FLBADeltaCustomArguments); +BENCHMARK(BM_DeltaDecodingDenseFLBA)->Apply(FLBADeltaCustomArguments); + static void BM_RleEncodingBoolean(benchmark::State& state) { std::vector values(state.range(0), true); auto encoder = MakeEncoder(Type::BOOLEAN, Encoding::RLE); From ad356b3a4df25abb5996c01fe54dc5db4b09aab6 Mon Sep 17 00:00:00 2001 From: Marcin Krystianc Date: Thu, 22 Oct 2026 14:45:26 +0200 Subject: [PATCH 2/2] PR remarks --- cpp/src/parquet/decoder.cc | 265 ++++++++++++++------------ cpp/src/parquet/encoding_benchmark.cc | 53 ++++++ cpp/src/parquet/encoding_test.cc | 36 +++- 3 files changed, 230 insertions(+), 124 deletions(-) diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index a4362768e067..6d8484f4e41b 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -2102,21 +2102,18 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { } protected: - struct ByteArrayOutput { - // `buffer` is used both as the suffix decode target and as the output. - ByteArray* buffer; + // Output that produces one `T` (`ByteArray` or `FixedLenByteArray`) per value. + // A value is only copied (into `buffered_data_`) when both its prefix and its + // suffix are non-empty; otherwise it references the suffix in the page or the + // previous value. + struct ValueOutput { + T* buffer; ResizableBuffer* buffered_data; - int type_length; uint8_t* data_ptr = nullptr; const uint8_t* data_end = nullptr; - int FetchSuffixes(DeltaLengthByteArrayDecoder* suffix_decoder, int max_values) { - // Decode the suffixes in place; `Emit` reads each one before overwriting it - // with the reconstructed value. - return suffix_decoder->Decode(buffer, max_values); - } - - void Prepare(const int32_t* prefix_len_ptr, int max_values) { + void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths, + int max_values) { int64_t data_size = 0; for (int i = 0; i < max_values; ++i) { if (prefix_len_ptr[i] == 0) { @@ -2126,7 +2123,7 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) { throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY"); } - if (buffer[i].len == 0 && i != 0) { + if (suffix_lengths[i] == 0 && i != 0) { // We don't need to copy the prefix if the suffix length is 0 // and this is not the first run (that is, the prefix doesn't point // to the mutable `last_value_`). @@ -2134,7 +2131,7 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { } if (ARROW_PREDICT_FALSE( AddWithOverflow(data_size, prefix_len_ptr[i], &data_size) || - AddWithOverflow(data_size, buffer[i].len, &data_size))) { + AddWithOverflow(data_size, suffix_lengths[i], &data_size))) { throw ParquetException("excess expansion in DELTA_BYTE_ARRAY"); } } @@ -2143,81 +2140,47 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { data_end = data_ptr + data_size; } - std::string_view Emit(int i, int32_t prefix_len, std::string_view prefix) { - // Read the suffix before `buffer[i]` is overwritten with the output. - const ByteArray suffix = buffer[i]; - if (ARROW_PREDICT_FALSE(static_cast(prefix_len) > prefix.length())) { - throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY"); - } - if constexpr (std::is_same_v) { - // The value is `prefix_len` bytes of the previous value followed by the - // suffix, whichever branch below produces it. - if (ARROW_PREDICT_FALSE(static_cast(prefix_len) + suffix.len != - type_length)) { - throw ParquetException("FLBA type requires fixed-length ", type_length, - " but got ", prefix_len + suffix.len); - } - } - if (prefix_len == 0) { - // prefix is empty: buffer[i] already points to the suffix. - return std::string_view{buffer[i]}; + std::string_view Emit(int i, std::string_view prefix, std::string_view suffix) { + std::string_view value; + if (prefix.empty()) { + value = suffix; + } else if (suffix.empty() && i != 0) { + // This is not possible for the first run since the prefix would point to + // the mutable `last_value_`. + value = prefix; + } else { + DCHECK_LE(data_ptr + prefix.size() + suffix.size(), data_end); + memcpy(data_ptr, prefix.data(), prefix.size()); + memcpy(data_ptr + prefix.size(), suffix.data(), suffix.size()); + value = {reinterpret_cast(data_ptr), prefix.size() + suffix.size()}; + data_ptr += value.size(); } - if (i != 0 && suffix.len == 0) { - // suffix is empty: buffer[i] can simply point to the prefix. - // This is not possible for the first run since the prefix - // would point to the mutable `last_value_`. - prefix = prefix.substr(0, prefix_len); - buffer[i] = ByteArray(prefix); - return prefix; + buffer[i].ptr = reinterpret_cast(value.data()); + if constexpr (std::is_same_v) { + buffer[i].len = static_cast(value.size()); } - // Both prefix and suffix are non-empty, so decode the string into - // `data_ptr`. - const uint32_t full_len = suffix.len + static_cast(prefix_len); - DCHECK_LE(data_ptr + full_len, data_end); - memcpy(data_ptr, prefix.data(), prefix_len); - memcpy(data_ptr + prefix_len, suffix.ptr, suffix.len); - buffer[i].ptr = data_ptr; - buffer[i].len = full_len; - data_ptr += full_len; - return std::string_view{buffer[i]}; + return value; } }; // Output that writes fixed-length values contiguously into a caller-provided // buffer. Used by the FLBA dense decode path; avoids `buffered_data_` and any - // temporary per-value `ByteArray` storage. + // temporary per-value storage. struct DenseOutput { uint8_t* out; int type_length; - // View of the suffixes still to be consumed: their lengths, and a cursor into - // their contiguous data. No per-value pointer is ever materialized. - const int32_t* suffix_lengths = nullptr; - const uint8_t* suffix_data = nullptr; - - int FetchSuffixes(DeltaLengthByteArrayDecoder* suffix_decoder, int max_values) { - return suffix_decoder->NextBatch(max_values, &suffix_lengths, &suffix_data); - } - void Prepare(const int32_t* prefix_len_ptr, int max_values) {} + void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths, + int max_values) {} - std::string_view Emit(int i, int32_t prefix_len, std::string_view prefix) { - const int32_t suffix_len = suffix_lengths[i]; - const uint8_t* suffix_ptr = suffix_data; - suffix_data += suffix_len; - - if (ARROW_PREDICT_FALSE(static_cast(prefix_len) > prefix.length())) { - throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY"); + std::string_view Emit(int i, std::string_view prefix, std::string_view suffix) { + uint8_t* dst = out + static_cast(i) * type_length; + if (!prefix.empty()) { + memcpy(dst, prefix.data(), prefix.size()); } - // Each reconstructed FLBA value must be exactly `type_length` bytes. - if (ARROW_PREDICT_FALSE(static_cast(prefix_len) + suffix_len != - type_length)) { - throw ParquetException("FLBA type requires fixed-length ", type_length, - " but got ", prefix_len + suffix_len); + if (!suffix.empty()) { + memcpy(dst + prefix.size(), suffix.data(), suffix.size()); } - // Copy the prefix and the suffix straight into the caller's buffer. - uint8_t* dst = out + static_cast(i) * type_length; - memcpy(dst, prefix.data(), prefix_len); - memcpy(dst + prefix_len, suffix_ptr, suffix_len); // The next value's prefix references this value, which now lives in the // caller's buffer. return std::string_view{reinterpret_cast(dst), @@ -2225,6 +2188,73 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { } }; + // Output that appends each value to an Arrow binary builder through `helper`. + // A value is assembled in `scratch` only when both its prefix and its suffix are + // non-empty; otherwise it is appended straight from the page or the previous value. + template + struct ArrowOutput { + Helper* helper; + std::string* scratch; + // Number of slots (values and nulls) the builder receives. + int num_slots; + // Builder data still to be appended, reserved by `Prepare`. + int64_t remaining_data_length = 0; + // Set when there are nulls: the runs of valid slots. The nulls before a run are + // appended when its first value is; `position` is the end of the current run. + ::arrow::internal::SetBitRunReader* valid_runs = nullptr; + int64_t position = 0; + int64_t run_remaining = 0; + + void Prepare(const int32_t* prefix_len_ptr, const int32_t* suffix_lengths, + int max_values) { + // Each value is exactly `prefix_len + suffix_len` bytes, so reserve exactly the + // data the batch appends; `Emit` can then append without capacity checks. + for (int i = 0; i < max_values; ++i) { + if (ARROW_PREDICT_FALSE(prefix_len_ptr[i] < 0)) { + throw ParquetException("negative prefix length in DELTA_BYTE_ARRAY"); + } + remaining_data_length += + static_cast(prefix_len_ptr[i]) + suffix_lengths[i]; + } + PARQUET_THROW_NOT_OK(helper->Prepare(num_slots, remaining_data_length)); + } + + std::string_view Emit(int i, std::string_view prefix, std::string_view suffix) { + std::string_view value; + if (prefix.empty()) { + value = suffix; + } else if (suffix.empty()) { + // For the first value, `prefix` is `last_value_`, which is not modified + // until the whole batch is decoded. + value = prefix; + } else { + if (prefix.data() == scratch->data()) { + // The previous value is in `scratch`: keep its first `prefix.size()` bytes. + scratch->resize(prefix.size()); + } else { + scratch->assign(prefix); + } + scratch->append(suffix); + value = *scratch; + } + if (valid_runs != nullptr) { + if (run_remaining == 0) { + const auto run = valid_runs->NextRun(); + PARQUET_THROW_NOT_OK(helper->AppendNulls(run.position - position)); + position = run.position + run.length; + run_remaining = run.length; + } + --run_remaining; + } + PARQUET_THROW_NOT_OK( + helper->AppendValue(reinterpret_cast(value.data()), + static_cast(value.size()), remaining_data_length)); + remaining_data_length -= value.size(); + DCHECK_GE(remaining_data_length, 0); + return value; + } + }; + template int GetInternalImpl(Output& output, int max_values) { // Decode up to `max_values` values, delegating materialization to `output`. @@ -2233,7 +2263,10 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { return max_values; } - const int suffix_read = output.FetchSuffixes(&suffix_decoder_, max_values); + const int32_t* suffix_lengths = nullptr; + const uint8_t* suffix_data = nullptr; + const int suffix_read = + suffix_decoder_.NextBatch(max_values, &suffix_lengths, &suffix_data); if (ARROW_PREDICT_FALSE(suffix_read != max_values)) { ParquetException::EofException("Read " + std::to_string(suffix_read) + ", expecting " + std::to_string(max_values) + @@ -2243,11 +2276,25 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { const int32_t* prefix_len_ptr = buffered_prefix_length_->data_as() + prefix_len_offset_; - output.Prepare(prefix_len_ptr, max_values); + output.Prepare(prefix_len_ptr, suffix_lengths, max_values); std::string_view prefix{last_value_}; for (int i = 0; i < max_values; ++i) { - prefix = output.Emit(i, prefix_len_ptr[i], prefix); + const int32_t prefix_len = prefix_len_ptr[i]; + const std::string_view suffix{reinterpret_cast(suffix_data), + static_cast(suffix_lengths[i])}; + suffix_data += suffix.size(); + if (ARROW_PREDICT_FALSE(static_cast(prefix_len) > prefix.length())) { + throw ParquetException("prefix length too large in DELTA_BYTE_ARRAY"); + } + if constexpr (std::is_same_v) { + if (ARROW_PREDICT_FALSE(prefix_len + suffix.size() != + static_cast(this->type_length_))) { + throw ParquetException("FLBA type requires fixed-length ", this->type_length_, + " but got ", prefix_len + suffix.size()); + } + } + prefix = output.Emit(i, prefix.substr(0, prefix_len), suffix); } prefix_len_offset_ += max_values; @@ -2262,8 +2309,8 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { return max_values; } - int GetInternal(ByteArray* buffer, int max_values) { - ByteArrayOutput output{buffer, buffered_data_.get(), this->type_length_}; + int GetInternal(T* buffer, int max_values) { + ValueOutput output{buffer, buffered_data_.get()}; return GetInternalImpl(output, max_values); } @@ -2272,54 +2319,30 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { return GetInternalImpl(output, max_values); } - int DecodePointers(FixedLenByteArray* buffer, int max_values) { - PARQUET_THROW_NOT_OK( - buffered_data_->Resize(static_cast(max_values) * this->type_length_, - /*shrink_to_fit=*/false)); - uint8_t* values = buffered_data_->mutable_data(); - const int decoded_values_size = DecodeDense(values, max_values); - for (int i = 0; i < decoded_values_size; ++i) { - buffer[i].ptr = values + static_cast(i) * this->type_length_; - } - return decoded_values_size; - } - Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset, typename EncodingTraits::Accumulator* out, int* out_num_values) { - ARROW_ASSIGN_OR_RAISE( - auto values_buffer, - ::arrow::AllocateBuffer(sizeof(ByteArray) * (num_values - null_count), pool_)); - auto values_data = values_buffer->template mutable_data_as(); - - const int num_valid_values = GetInternal(values_data, num_values - null_count); - if (ARROW_PREDICT_FALSE(num_values - null_count != num_valid_values)) { - throw ParquetException("Expected to decode ", num_values - null_count, - " values, but decoded ", num_valid_values, " values."); + const int values_to_decode = num_values - null_count; + if (ARROW_PREDICT_FALSE(values_to_decode > num_valid_values_)) { + throw ParquetException("Expected to decode ", values_to_decode, + " values, but decoded ", num_valid_values_, " values."); } + std::string scratch; auto visit_binary_helper = [&](auto* helper) { - auto values_ptr = reinterpret_cast(values_data); - int value_idx = 0; - - PARQUET_THROW_NOT_OK( - VisitBitRuns(valid_bits, valid_bits_offset, num_values, - [&](int64_t position, int64_t run_length, bool is_valid) { - if (is_valid) { - for (int64_t i = 0; i < run_length; ++i) { - const auto& val = values_ptr[value_idx]; - RETURN_NOT_OK(helper->AppendValue( - val.ptr, static_cast(val.len))); - ++value_idx; - } - return Status::OK(); - } else { - return helper->AppendNulls(run_length); - } - })); - - *out_num_values = num_valid_values; + ArrowOutput> output{helper, &scratch, + num_values}; + std::optional<::arrow::internal::SetBitRunReader> valid_runs; + if (null_count > 0) { + valid_runs.emplace(valid_bits, valid_bits_offset, num_values); + output.valid_runs = &*valid_runs; + } + GetInternalImpl(output, values_to_decode); + if (null_count > 0) { + RETURN_NOT_OK(helper->AppendNulls(num_values - output.position)); + } + *out_num_values = values_to_decode; return Status::OK(); }; return DispatchArrowBinaryHelper(out, num_values, /*estimated_data_length=*/{}, @@ -2361,7 +2384,7 @@ class DeltaByteArrayFLBADecoder : public DeltaByteArrayDecoderImpl, using Base::pool_; int Decode(FixedLenByteArray* buffer, int max_values) override { - return this->DecodePointers(buffer, max_values); + return this->GetInternal(buffer, max_values); } // Same internal decode as above, but copy the bytes contiguously into the diff --git a/cpp/src/parquet/encoding_benchmark.cc b/cpp/src/parquet/encoding_benchmark.cc index 3f1980f27a6f..e43234ab1c2d 100644 --- a/cpp/src/parquet/encoding_benchmark.cc +++ b/cpp/src/parquet/encoding_benchmark.cc @@ -1084,6 +1084,59 @@ static void ByteArrayDeltaCustomArguments(benchmark::internal::Benchmark* b) { BENCHMARK(BM_DeltaEncodingByteArray)->Apply(ByteArrayDeltaCustomArguments); BENCHMARK(BM_DeltaDecodingByteArray)->Apply(ByteArrayDeltaCustomArguments); +// Same as BM_DeltaDecodingByteArray, but decoding into an Arrow builder with +// `null-percent` of the slots null. +static void BM_DeltaDecodingArrowByteArray(benchmark::State& state) { + DeltaByteArrayState delta_state(state); + std::vector values = delta_state.MakeRandomByteArray(/*seed=*/42); + + auto encoder = MakeTypedEncoder(Encoding::DELTA_BYTE_ARRAY); + encoder->Put(values.data(), static_cast(values.size())); + std::shared_ptr buf = encoder->FlushValues(); + + // Spread the encoded (non-null) values over enough slots to reach `null-percent`. + std::default_random_engine gen(/*seed=*/42); + std::bernoulli_distribution dist_is_null(state.range(3) / 100.0); + std::vector is_valid; + for (int32_t num_valid = 0; num_valid < delta_state.array_length;) { + is_valid.push_back(!dist_is_null(gen)); + num_valid += is_valid.back(); + } + const int num_slots = static_cast(is_valid.size()); + const int null_count = num_slots - delta_state.array_length; + std::vector valid_bits(::arrow::bit_util::BytesForBits(num_slots)); + for (int i = 0; i < num_slots; ++i) { + ::arrow::bit_util::SetBitTo(valid_bits.data(), i, is_valid[i]); + } + + auto decoder = MakeTypedDecoder(Encoding::DELTA_BYTE_ARRAY); + for (auto _ : state) { + decoder->SetData(delta_state.array_length, buf->data(), + static_cast(buf->size())); + EncodingTraits::Accumulator acc; + acc.builder = std::make_unique<::arrow::BinaryBuilder>(default_memory_pool()); + decoder->DecodeArrow(num_slots, null_count, valid_bits.data(), 0, &acc); + ::benchmark::DoNotOptimize(acc); + } + state.SetItemsProcessed(state.iterations() * delta_state.array_length); + state.SetBytesProcessed(state.iterations() * delta_state.total_data_size); +} + +static void ByteArrayDeltaArrowCustomArguments(benchmark::internal::Benchmark* b) { + for (int max_string_length : {8, 64, 1024}) { + for (int batch_size : {512, 2048}) { + for (int prefixed_percent : {10, 90, 99}) { + for (int null_percent : {0, 50}) { + b->Args({max_string_length, batch_size, prefixed_percent, null_percent}); + } + } + } + } + b->ArgNames({"max-string-length", "batch-size", "prefixed-percent", "null-percent"}); +} + +BENCHMARK(BM_DeltaDecodingArrowByteArray)->Apply(ByteArrayDeltaArrowCustomArguments); + struct DeltaFLBAState { int32_t type_length; int32_t array_length; diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc index 142dcbbf106e..672e55a900dc 100644 --- a/cpp/src/parquet/encoding_test.cc +++ b/cpp/src/parquet/encoding_test.cc @@ -2559,9 +2559,11 @@ void TestDeltaByteArrayEncodingDirectPut::CheckRoundtrip() { rag.FixedSizeBinary(/*size=*/0, /*byte_width=*/kByteWidth); CheckDirectPut(values); - for (auto seed : {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) { - values = rag.FixedSizeBinary(kSize + seed, kByteWidth); - CheckDirectPut(values); + for (double null_probability : {0.0, 0.25}) { + for (auto seed : {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}) { + values = rag.FixedSizeBinary(kSize + seed, kByteWidth, null_probability); + CheckDirectPut(values); + } } } @@ -2665,6 +2667,34 @@ TEST(DeltaByteArrayEncodingAdHoc, ArrowDirectPut) { } } +// A value with an empty suffix at the start of a Decode call must not point into the +// decoder's last value, which that call overwrites. +TEST(DeltaByteArrayEncodingAdHoc, FLBAEmptySuffixAtBatchStart) { + auto descr = ExampleDescr(); + const int type_length = descr->type_length(); + const std::string a(static_cast(type_length), 'a'); + const std::string b(static_cast(type_length), 'b'); + // Values: a, a, a, b + auto buffer = + ::arrow::ConcatenateBuffers({DeltaEncode({0, type_length, type_length, 0}), + DeltaEncode({type_length, 0, 0, type_length}), + std::make_shared(a + b)}) + .ValueOrDie(); + + auto decoder = MakeTypedDecoder(Encoding::DELTA_BYTE_ARRAY, descr.get()); + decoder->SetData(/*num_values=*/4, buffer->data(), static_cast(buffer->size())); + + std::vector out(2); + for (const auto& expected : {std::vector{a, a}, std::vector{a, b}}) { + ASSERT_EQ(2, decoder->Decode(out.data(), /*max_values=*/2)); + for (int i = 0; i < 2; ++i) { + ASSERT_EQ(expected[i], + std::string(reinterpret_cast(out[i].ptr), type_length)) + << "value " << i; + } + } +} + // ---------------------------------------------------------------------- // Dense FIXED_LEN_BYTE_ARRAY decode tests //