diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc index c4d3fe5a8a5a..008e192cd633 100644 --- a/cpp/src/parquet/decoder.cc +++ b/cpp/src/parquet/decoder.cc @@ -854,7 +854,26 @@ class PlainByteArrayDecoder : public PlainDecoder { class PlainFLBADecoder : public PlainDecoder, public FLBADecoder { public: using Base = PlainDecoder; + using Base::Decode; // keep Decode(FixedLenByteArray*, int) using Base::PlainDecoder; + + // PLAIN-encoded FLBA values are already contiguous in the page buffer, so + // decode them with a single memcpy into the caller's buffer. This is the same + // copy used by PlainDecoder::DecodeArrow, without the builder. + int Decode(uint8_t* buffer, int max_values) override { + max_values = std::min(max_values, this->num_values_); + const int64_t bytes_to_decode = static_cast(this->type_length_) * max_values; + if (bytes_to_decode > this->len_) { + ParquetException::EofException(); + } + if (bytes_to_decode > 0) { + memcpy(buffer, this->data_, static_cast(bytes_to_decode)); + } + this->data_ += bytes_to_decode; + this->len_ -= static_cast(bytes_to_decode); + this->num_values_ -= max_values; + return max_values; + } }; // ---------------------------------------------------------------------- @@ -1431,6 +1450,36 @@ class DictByteArrayDecoderImpl : public DictDecoderImpl { } }; +// Dictionary decoder for FIXED_LEN_BYTE_ARRAY that can decode directly into a +// caller-owned, densely packed byte buffer. DictDecoderImpl on its own +// does not inherit FLBADecoder, so this thin subclass adds the dense Decode +// overload (mirroring the DeltaByteArray and ByteStreamSplit FLBA decoders). +class DictFLBADecoder : public DictDecoderImpl, public FLBADecoder { + public: + using Base = DictDecoderImpl; + using Base::Decode; // keep Decode(FixedLenByteArray*, int) + using Base::DictDecoderImpl; + + // Read one index per value and copy that dictionary entry's type_length bytes + // contiguously into the caller's buffer. Mirrors DecodeArrow without nulls. + int Decode(uint8_t* buffer, int max_values) override { + max_values = std::min(max_values, this->num_values_); + const auto* dict_values = this->dictionary_->data_as(); + const int64_t type_length = this->type_length_; + for (int i = 0; i < max_values; ++i) { + int32_t index; + if (ARROW_PREDICT_FALSE(!this->idx_decoder_.Get(&index))) { + throw ParquetException("Dict decoding failed"); + } + PARQUET_THROW_NOT_OK(this->IndexInBounds(index)); + memcpy(buffer + i * type_length, dict_values[index].ptr, + static_cast(type_length)); + } + this->num_values_ -= max_values; + return max_values; + } +}; + // ---------------------------------------------------------------------- // DELTA_BINARY_PACKED decoder @@ -1730,12 +1779,19 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl { DecodeLengths(); } - int Decode(ByteArray* buffer, int max_values) override { - // Decode up to `max_values` strings into an internal buffer - // and reference them into `buffer`. + // 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 GetBatchView(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; } @@ -1747,7 +1803,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"); } @@ -1756,13 +1811,24 @@ 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; + *lengths = length_ptr; + *data = data_ + bytes_offset; + this->num_values_ -= max_values; + num_valid_values_ -= max_values; + return max_values; + } + + 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 = GetBatchView(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; } - this->num_values_ -= max_values; - num_valid_values_ -= max_values; return max_values; } @@ -1983,7 +2049,8 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { suffix_decoder_(nullptr, pool), last_value_in_previous_page_(""), buffered_prefix_length_(AllocateBuffer(pool, 0)), - buffered_data_(AllocateBuffer(pool, 0)) {} + buffered_data_(AllocateBuffer(pool, 0)), + buffered_suffix_(AllocateBuffer(pool, 0)) {} void SetData(int num_values, const uint8_t* data, int len) override { this->num_values_ = num_values; @@ -2033,93 +2100,171 @@ 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; + // Storage-agnostic reconstruction of DELTA_BYTE_ARRAY values. + // + // Each value is a `prefix` (a prefix of the previous value) followed by a + // `suffix` produced by `suffix_decoder_`. `GetInternalImpl` drives the shared + // decoding and delegates the actual materialization to an `Output` policy, so + // callers can either reference values as `ByteArray`s (backed by + // `buffered_data_`) or write fixed-length values contiguously into a + // caller-provided buffer without allocating per-value pointers. + + // Output that reconstructs values into `ByteArray`s backed by `buffered_data_`, + // reproducing the original DELTA_BYTE_ARRAY behavior including the zero-copy + // optimizations for empty prefixes/suffixes. + 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_begin = nullptr; + uint8_t* data_ptr = nullptr; + int64_t data_size = 0; + + 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) { + 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_begin = buffered_data->mutable_data(); + data_ptr = data_begin; } - 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 (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`. + memcpy(data_ptr, prefix.data(), prefix_len); + memcpy(data_ptr + prefix_len, suffix.ptr, suffix.len); + const uint32_t full_len = suffix.len + static_cast(prefix_len); + buffer[i].ptr = data_ptr; + buffer[i].len = full_len; + data_ptr += full_len; + return std::string_view{buffer[i]}; + } + + void Finish(int max_values) { + DCHECK_EQ(data_ptr - data_begin, data_size); + if constexpr (std::is_same_v) { + // Checks all values + for (int i = 0; i < max_values; ++i) { + if (buffer[i].len != static_cast(type_length)) { + throw ParquetException("FLBA type requires fixed-length ", type_length, + " but got ", buffer[i].len); + } + } } } - // 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->GetBatchView(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("Fixed length byte array length mismatch"); + } + // 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)}; + } + + void Finish(int max_values) {} + }; + + 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; @@ -2129,19 +2274,30 @@ 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); - } - } - } - + output.Finish(max_values); return max_values; } + // Resize the ByteArray scratch buffer and return its data. Used to decode + // suffixes and to expose reconstructed value pointers without a per-call + // heap allocation. + ByteArray* ResizeSuffixScratch(int max_values) { + PARQUET_THROW_NOT_OK(buffered_suffix_->Resize(sizeof(ByteArray) * max_values)); + return buffered_suffix_->mutable_data_as(); + } + + int GetInternal(ByteArray* buffer, int max_values) { + ByteArrayOutput output{buffer, buffered_data_.get(), this->type_length_}; + return GetInternalImpl(output, max_values); + } + + // Decode fixed-length values contiguously into `out` without materializing + // per-value pointers. Only meaningful for `FLBAType`. + int DecodeDense(uint8_t* out, int max_values) { + DenseOutput output{out, this->type_length_}; + return GetInternalImpl(output, max_values); + } + Status DecodeArrowDense(int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset, typename EncodingTraits::Accumulator* out, @@ -2199,6 +2355,9 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl { // buffer for decoded strings, which guarantees the lifetime of the decoded strings // until the next call of Decode. std::shared_ptr buffered_data_; + // Scratch buffer of ByteArray used to decode suffixes / expose value pointers + // without a per-call heap allocation. + std::shared_ptr buffered_suffix_; }; class DeltaByteArrayDecoder : public DeltaByteArrayDecoderImpl { @@ -2219,19 +2378,22 @@ 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; + // Reconstruct values as ByteArrays backed by `buffered_data_`, then expose + // their pointers; the fixed length is implied by the column descriptor and + // validated by `GetInternal`. + ByteArray* values = this->ResizeSuffixScratch(max_values); + const int decoded_values_size = this->GetInternal(values, max_values); + for (int i = 0; i < decoded_values_size; ++i) { + buffer[i].ptr = values[i].ptr; } return decoded_values_size; } + + // Decode the bytes contiguously into the caller's buffer without + // materializing per-value pointers. + int Decode(uint8_t* buffer, int max_values) override { + return this->DecodeDense(buffer, max_values); + } }; // ---------------------------------------------------------------------- @@ -2370,6 +2532,12 @@ class ByteStreamSplitDecoder : public ByteStreamSplitDecoderBaseDecodeRaw(buffer, max_values); + } }; } // namespace @@ -2475,7 +2643,7 @@ std::unique_ptr MakeDictDecoder(Type::type type_num, case Type::BYTE_ARRAY: return std::make_unique(descr, pool); case Type::FIXED_LEN_BYTE_ARRAY: - return std::make_unique>(descr, pool); + return std::make_unique(descr, pool); default: break; } diff --git a/cpp/src/parquet/encoding.h b/cpp/src/parquet/encoding.h index e3de4f2aa60b..9a0cc55aa185 100644 --- a/cpp/src/parquet/encoding.h +++ b/cpp/src/parquet/encoding.h @@ -410,12 +410,23 @@ class BooleanDecoder : virtual public TypedDecoder { class FLBADecoder : virtual public TypedDecoder { public: + using TypedDecoder::Decode; using TypedDecoder::DecodeSpaced; - // TODO(wesm): As possible follow-up to PARQUET-1508, we should examine if - // there is value in adding specialized read methods for - // FIXED_LEN_BYTE_ARRAY. If only Decimal data can occur with this data type - // then perhaps not + /// \brief Decode values into a densely packed buffer + /// + /// Unlike Decode(FixedLenByteArray*, int), which writes one pointer per + /// value, this writes the raw fixed-width values back to back, with no + /// per-value pointers and no gaps. + /// + /// \param[in] buffer destination for decoded values; caller owns it and + /// must size it to at least max_values * descr->type_length() bytes. + /// \param[in] max_values max values to decode. + /// \return The number of values decoded. Should be identical to max_values + /// except at the end of the current data page. + /// + /// \note API EXPERIMENTAL + virtual int Decode(uint8_t* buffer, int max_values) = 0; }; PARQUET_EXPORT diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc index 831829e4a210..74b422cb606f 100644 --- a/cpp/src/parquet/encoding_test.cc +++ b/cpp/src/parquet/encoding_test.cc @@ -2660,4 +2660,167 @@ TEST(DeltaByteArrayEncodingAdHoc, ArrowDirectPut) { } } +// ---------------------------------------------------------------------- +// Dense FIXED_LEN_BYTE_ARRAY decode tests +// +// FLBADecoder::Decode(uint8_t*, int) writes decoded values back to back into a +// densely packed buffer, with no per-value FixedLenByteArray pointers. Verify +// it for every encoding that overrides it: PLAIN, RLE_DICTIONARY, +// DELTA_BYTE_ARRAY and BYTE_STREAM_SPLIT. + +class TestFLBADenseDecode : public ::testing::Test { + public: + void SetUp() override { + descr_ = ExampleDescr(); + type_length_ = descr_->type_length(); + draws_.resize(kNumValues); + GenerateData(kNumValues, draws_.data(), &data_buffer_); + } + + // Decode densely and compare each value against the original draw. + void CheckDenseDecode(FLBADecoder* decoder, const std::vector& decode_sizes) { + ASSERT_NE(nullptr, decoder); + std::vector dense(static_cast(type_length_) * kNumValues); + + int values_decoded = 0; + for (int decode_size : decode_sizes) { + const int expected_decoded = std::min(decode_size, kNumValues - values_decoded); + ASSERT_EQ(expected_decoded, + decoder->Decode( + dense.data() + static_cast(values_decoded) * type_length_, + decode_size)); + values_decoded += expected_decoded; + ASSERT_EQ(kNumValues - values_decoded, decoder->values_left()); + } + ASSERT_EQ(kNumValues, values_decoded); + ASSERT_EQ(0, decoder->Decode(dense.data(), /*max_values=*/1)); + ASSERT_EQ(0, decoder->values_left()); + + for (int i = 0; i < kNumValues; ++i) { + ASSERT_EQ(0, memcmp(dense.data() + static_cast(i) * type_length_, + draws_[i].ptr, type_length_)) + << "mismatch at value " << i; + } + } + + void CheckDenseDecode(FLBADecoder* decoder) { + CheckDenseDecode(decoder, {1, 17, kNumValues / 2, kNumValues}); + } + + protected: + static constexpr int kNumValues = 1000; + int type_length_; + std::vector draws_; + std::vector data_buffer_; + std::shared_ptr descr_; +}; + +// These encodings are all built the same way, so exercise them with a single +// body. RLE_DICTIONARY needs a dictionary and is tested separately below. +TEST_F(TestFLBADenseDecode, NonDictionaryEncodings) { + for (auto encoding : + {Encoding::PLAIN, Encoding::DELTA_BYTE_ARRAY, Encoding::BYTE_STREAM_SPLIT}) { + SCOPED_TRACE(EncodingToString(encoding)); + + auto encoder = + MakeTypedEncoder(encoding, /*use_dictionary=*/false, descr_.get()); + encoder->Put(draws_.data(), kNumValues); + auto buffer = encoder->FlushValues(); + + auto decoder = MakeTypedDecoder(encoding, descr_.get()); + decoder->SetData(kNumValues, buffer->data(), static_cast(buffer->size())); + // EXPECT rather than ASSERT so one failing encoding doesn't hide the others. + EXPECT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast(decoder.get()))); + } +} + +TEST_F(TestFLBADenseDecode, Dictionary) { + auto base_encoder = MakeEncoder(::parquet::Type::FIXED_LEN_BYTE_ARRAY, Encoding::PLAIN, + /*use_dictionary=*/true, descr_.get()); + auto encoder = dynamic_cast*>(base_encoder.get()); + auto dict_traits = dynamic_cast*>(base_encoder.get()); + + encoder->Put(draws_.data(), kNumValues); + auto dict_buffer = + AllocateBuffer(default_memory_pool(), dict_traits->dict_encoded_size()); + dict_traits->WriteDict(dict_buffer->mutable_data()); + auto indices = encoder->FlushValues(); + + auto dict_decoder = MakeTypedDecoder(Encoding::PLAIN, descr_.get()); + dict_decoder->SetData(dict_traits->num_entries(), dict_buffer->data(), + static_cast(dict_buffer->size())); + + auto decoder = MakeDictDecoder(descr_.get()); + decoder->SetDict(dict_decoder.get()); + decoder->SetData(kNumValues, indices->data(), static_cast(indices->size())); + // dict_decoder must outlive the decode: the decoded bytes are owned by it. + ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast(decoder.get()))); +} + +// DELTA_BYTE_ARRAY stores each value as a prefix shared with its predecessor plus +// a suffix, so the dense decode path has to reconstruct the two halves. The +// random draws used above hardly ever share a prefix, which leaves that +// reconstruction untested; use prefixed data here instead. +TEST_F(TestFLBADenseDecode, DeltaByteArrayPrefixedData) { + for (double prefixed_probability : {0.5, 1.0}) { + SCOPED_TRACE("prefixed_probability=" + std::to_string(prefixed_probability)); + GeneratePrefixedData(kNumValues, draws_.data(), &data_buffer_, + prefixed_probability); + + auto encoder = MakeTypedEncoder(Encoding::DELTA_BYTE_ARRAY, + /*use_dictionary=*/false, descr_.get()); + encoder->Put(draws_.data(), kNumValues); + auto buffer = encoder->FlushValues(); + + auto decoder = MakeTypedDecoder(Encoding::DELTA_BYTE_ARRAY, descr_.get()); + decoder->SetData(kNumValues, buffer->data(), static_cast(buffer->size())); + EXPECT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast(decoder.get()))); + } +} + +TEST_F(TestFLBADenseDecode, PlainRejectsTruncatedDenseBuffer) { + auto encoder = + MakeTypedEncoder(Encoding::PLAIN, /*use_dictionary=*/false, descr_.get()); + encoder->Put(draws_.data(), kNumValues); + auto buffer = encoder->FlushValues(); + + auto decoder = MakeTypedDecoder(Encoding::PLAIN, descr_.get()); + decoder->SetData(kNumValues, buffer->data(), static_cast(buffer->size() - 1)); + + std::vector dense(static_cast(type_length_) * kNumValues); + ASSERT_THROW(decoder->Decode(dense.data(), kNumValues), ParquetException); +} + +TEST_F(TestFLBADenseDecode, DictionaryRejectsOutOfBoundsDenseIndex) { + auto dict_decoder = MakeTypedDecoder(Encoding::PLAIN, descr_.get()); + dict_decoder->SetData(/*num_values=*/1, draws_[0].ptr, type_length_); + + auto decoder = MakeDictDecoder(descr_.get()); + decoder->SetDict(dict_decoder.get()); + + // RLE_DICTIONARY data: bit width 1, followed by an RLE run of one index value + // 1. The dictionary has only one entry, so the index is out of bounds. + const std::vector indices = {1, 2, 1}; + decoder->SetData(/*num_values=*/1, indices.data(), static_cast(indices.size())); + + auto flba_decoder = dynamic_cast(decoder.get()); + ASSERT_NE(nullptr, flba_decoder); + std::vector dense(static_cast(type_length_)); + ASSERT_THROW(flba_decoder->Decode(dense.data(), /*max_values=*/1), ParquetException); +} + +TEST_F(TestFLBADenseDecode, DeltaByteArrayRejectsWrongFLBALengthDense) { + std::string suffix(static_cast(type_length_ - 1), 'x'); + auto buffer = + ::arrow::ConcatenateBuffers({DeltaEncode({0}), DeltaEncode({type_length_ - 1}), + std::make_shared(suffix)}) + .ValueOrDie(); + + auto decoder = MakeTypedDecoder(Encoding::DELTA_BYTE_ARRAY, descr_.get()); + decoder->SetData(/*num_values=*/1, buffer->data(), static_cast(buffer->size())); + + std::vector dense(static_cast(type_length_)); + ASSERT_THROW(decoder->Decode(dense.data(), /*max_values=*/1), ParquetException); +} + } // namespace parquet::test