From e3dc06b27238db200b09d7ae72af4ba2d9347663 Mon Sep 17 00:00:00 2001
From: Marcin Krystianc <marcin.krystianc@gmail.com>
Date: Thu, 2 Jul 2026 14:21:36 +0200
Subject: [PATCH 1/6] [C++][Parquet] Add dense decode path for
 FIXED_LEN_BYTE_ARRAY

---
 cpp/src/parquet/decoder.cc       | 81 ++++++++++++++++++++++++++-
 cpp/src/parquet/encoding.h       | 19 +++++--
 cpp/src/parquet/encoding_test.cc | 94 ++++++++++++++++++++++++++++++++
 3 files changed, 189 insertions(+), 5 deletions(-)

diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc
index c4d3fe5a8a5a..eb06fd8360af 100644
--- a/cpp/src/parquet/decoder.cc
+++ b/cpp/src/parquet/decoder.cc
@@ -855,6 +855,26 @@ class PlainFLBADecoder : public PlainDecoder<FLBAType>, public FLBADecoder {
  public:
   using Base = PlainDecoder<FLBAType>;
   using Base::PlainDecoder;
+  using Base::Decode;  // keep Decode(FixedLenByteArray*, int)
+
+  // 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<FLBAType>::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<int64_t>(this->type_length_) * max_values;
+    if (bytes_to_decode > this->len_ || bytes_to_decode > INT_MAX) {
+      ParquetException::EofException();
+    }
+    if (bytes_to_decode > 0) {
+      memcpy(buffer, this->data_, static_cast<size_t>(bytes_to_decode));
+    }
+    this->data_ += bytes_to_decode;
+    this->len_ -= static_cast<int>(bytes_to_decode);
+    this->num_values_ -= max_values;
+    return max_values;
+  }
 };
 
 // ----------------------------------------------------------------------
@@ -1212,6 +1232,35 @@ int DictDecoderImpl<FLBAType>::DecodeArrow(
   return num_values - null_count;
 }
 
+// Dictionary decoder for FIXED_LEN_BYTE_ARRAY that can decode directly into a
+// caller-owned, densely packed byte buffer. DictDecoderImpl<FLBAType> 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<FLBAType>, public FLBADecoder {
+ public:
+  using Base = DictDecoderImpl<FLBAType>;
+  using Base::DictDecoderImpl;
+  using Base::Decode;  // keep Decode(FixedLenByteArray*, int)
+
+  // 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<FLBA>();
+    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<size_t>(type_length));
+    }
+    this->num_values_ -= max_values;
+    return max_values;
+  }
+};
 template <typename Type>
 int DictDecoderImpl<Type>::DecodeArrow(
     int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset,
@@ -2232,6 +2281,23 @@ class DeltaByteArrayFLBADecoder : public DeltaByteArrayDecoderImpl<FLBAType>,
     }
     return decoded_values_size;
   }
+
+  // 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<ByteArray> decode_byte_array(max_values);
+    const int decoded_values_size = GetInternal(decode_byte_array.data(), max_values);
+    const uint32_t type_length = static_cast<uint32_t>(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<int64_t>(i) * type_length, decode_byte_array[i].ptr,
+             type_length);
+    }
+    return decoded_values_size;
+  }
 };
 
 // ----------------------------------------------------------------------
@@ -2370,10 +2436,23 @@ class ByteStreamSplitDecoder<FLBAType> : public ByteStreamSplitDecoderBase<FLBAT
     }
     return num_decoded;
   }
+
+  // DecodeRaw already unsplits the byte streams into a contiguous buffer, so
+  // decode straight into the caller's buffer with no intermediate scratch.
+  int Decode(uint8_t* buffer, int max_values) override {
+    return this->DecodeRaw(buffer, max_values);
+  }
 };
 
 }  // namespace
 
+// Default for the dense (densely packed) FLBA decode. Encodings override this;
+// the base throws so an unimplemented encoding fails with a clear message.
+int FLBADecoder::Decode(uint8_t* /*buffer*/, int /*max_values*/) {
+  throw ParquetException(
+      "Dense FIXED_LEN_BYTE_ARRAY decoding is not implemented for this encoding");
+}
+
 // ----------------------------------------------------------------------
 // Factory functions
 
@@ -2475,7 +2554,7 @@ std::unique_ptr<Decoder> MakeDictDecoder(Type::type type_num,
     case Type::BYTE_ARRAY:
       return std::make_unique<DictByteArrayDecoderImpl>(descr, pool);
     case Type::FIXED_LEN_BYTE_ARRAY:
-      return std::make_unique<DictDecoderImpl<FLBAType>>(descr, pool);
+      return std::make_unique<DictFLBADecoder>(descr, pool);
     default:
       break;
   }
diff --git a/cpp/src/parquet/encoding.h b/cpp/src/parquet/encoding.h
index e3de4f2aa60b..27b90a5fe9b7 100644
--- a/cpp/src/parquet/encoding.h
+++ b/cpp/src/parquet/encoding.h
@@ -410,12 +410,23 @@ class BooleanDecoder : virtual public TypedDecoder<BooleanType> {
 
 class FLBADecoder : virtual public TypedDecoder<FLBAType> {
  public:
+  using TypedDecoder<FLBAType>::Decode;
   using TypedDecoder<FLBAType>::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);  
 };
 
 PARQUET_EXPORT
diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc
index 831829e4a210..d161b61af5ea 100644
--- a/cpp/src/parquet/encoding_test.cc
+++ b/cpp/src/parquet/encoding_test.cc
@@ -2660,4 +2660,98 @@ 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<FLBAType>();
+    type_length_ = descr_->type_length();
+    draws_.resize(kNumValues);
+    GenerateData<FLBA>(kNumValues, draws_.data(), &data_buffer_);
+  }
+
+  // Decode densely and compare each value against the original draw.
+  void CheckDenseDecode(FLBADecoder* decoder) {
+    ASSERT_NE(nullptr, decoder);
+    std::vector<uint8_t> dense(static_cast<size_t>(type_length_) * kNumValues);
+    int values_decoded = decoder->Decode(dense.data(), kNumValues);
+    ASSERT_EQ(kNumValues, values_decoded);
+    for (int i = 0; i < kNumValues; ++i) {
+      ASSERT_EQ(0, memcmp(dense.data() + static_cast<int64_t>(i) * type_length_,
+                          draws_[i].ptr, type_length_))
+          << "mismatch at value " << i;
+    }
+  }
+
+ protected:
+  static constexpr int kNumValues = 1000;
+  int type_length_;
+  std::vector<FLBA> draws_;
+  std::vector<uint8_t> data_buffer_;
+  std::shared_ptr<ColumnDescriptor> descr_;
+};
+
+TEST_F(TestFLBADenseDecode, Plain) {
+  auto encoder =
+      MakeTypedEncoder<FLBAType>(Encoding::PLAIN, /*use_dictionary=*/false, descr_.get());
+  encoder->Put(draws_.data(), kNumValues);
+  auto buffer = encoder->FlushValues();
+
+  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::PLAIN, descr_.get());
+  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
+  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(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<TypedEncoder<FLBAType>*>(base_encoder.get());
+  auto dict_traits = dynamic_cast<DictEncoder<FLBAType>*>(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<FLBAType>(Encoding::PLAIN, descr_.get());
+  dict_decoder->SetData(dict_traits->num_entries(), dict_buffer->data(),
+                        static_cast<int>(dict_buffer->size()));
+
+  auto decoder = MakeDictDecoder<FLBAType>(descr_.get());
+  decoder->SetDict(dict_decoder.get());
+  decoder->SetData(kNumValues, indices->data(), static_cast<int>(indices->size()));
+  // dict_decoder must outlive the decode: the decoded bytes are owned by it.
+  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
+}
+
+TEST_F(TestFLBADenseDecode, DeltaByteArray) {
+  auto encoder = MakeTypedEncoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY,
+                                            /*use_dictionary=*/false, descr_.get());
+  encoder->Put(draws_.data(), kNumValues);
+  auto buffer = encoder->FlushValues();
+
+  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY, descr_.get());
+  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
+  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
+}
+
+TEST_F(TestFLBADenseDecode, ByteStreamSplit) {
+  auto encoder = MakeTypedEncoder<FLBAType>(Encoding::BYTE_STREAM_SPLIT,
+                                            /*use_dictionary=*/false, descr_.get());
+  encoder->Put(draws_.data(), kNumValues);
+  auto buffer = encoder->FlushValues();
+
+  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::BYTE_STREAM_SPLIT, descr_.get());
+  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
+  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
+}
+
 }  // namespace parquet::test

From 0d3a112eed7a0fcc654ad36f65aee02a6ad381d3 Mon Sep 17 00:00:00 2001
From: Marcin Krystianc <marcin.krystianc@gmail.com>
Date: Mon, 6 Jul 2026 17:02:29 +0200
Subject: [PATCH 2/6] apply Rok's remarks + fix formatting

---
 cpp/src/parquet/decoder.cc       |  7 ++--
 cpp/src/parquet/encoding.h       |  2 +-
 cpp/src/parquet/encoding_test.cc | 66 +++++++++++++++++++++++++++++++-
 3 files changed, 68 insertions(+), 7 deletions(-)

diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc
index eb06fd8360af..0c2a1fa0f4d3 100644
--- a/cpp/src/parquet/decoder.cc
+++ b/cpp/src/parquet/decoder.cc
@@ -854,16 +854,15 @@ class PlainByteArrayDecoder : public PlainDecoder<ByteArrayType> {
 class PlainFLBADecoder : public PlainDecoder<FLBAType>, public FLBADecoder {
  public:
   using Base = PlainDecoder<FLBAType>;
-  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<FLBAType>::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<int64_t>(this->type_length_) * max_values;
+    const int64_t bytes_to_decode = static_cast<int64_t>(this->type_length_) * max_values;
     if (bytes_to_decode > this->len_ || bytes_to_decode > INT_MAX) {
       ParquetException::EofException();
     }
@@ -1239,8 +1238,8 @@ int DictDecoderImpl<FLBAType>::DecodeArrow(
 class DictFLBADecoder : public DictDecoderImpl<FLBAType>, public FLBADecoder {
  public:
   using Base = DictDecoderImpl<FLBAType>;
-  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.
diff --git a/cpp/src/parquet/encoding.h b/cpp/src/parquet/encoding.h
index 27b90a5fe9b7..fc6a42d8976d 100644
--- a/cpp/src/parquet/encoding.h
+++ b/cpp/src/parquet/encoding.h
@@ -426,7 +426,7 @@ class FLBADecoder : virtual public TypedDecoder<FLBAType> {
   /// except at the end of the current data page.
   ///
   /// \note API EXPERIMENTAL
-  virtual int Decode(uint8_t* buffer, int max_values);  
+  virtual int Decode(uint8_t* buffer, int max_values);
 };
 
 PARQUET_EXPORT
diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc
index d161b61af5ea..e7c941184806 100644
--- a/cpp/src/parquet/encoding_test.cc
+++ b/cpp/src/parquet/encoding_test.cc
@@ -2678,11 +2678,24 @@ class TestFLBADenseDecode : public ::testing::Test {
   }
 
   // Decode densely and compare each value against the original draw.
-  void CheckDenseDecode(FLBADecoder* decoder) {
+  void CheckDenseDecode(FLBADecoder* decoder, const std::vector<int>& decode_sizes) {
     ASSERT_NE(nullptr, decoder);
     std::vector<uint8_t> dense(static_cast<size_t>(type_length_) * kNumValues);
-    int values_decoded = decoder->Decode(dense.data(), 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<int64_t>(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<int64_t>(i) * type_length_,
                           draws_[i].ptr, type_length_))
@@ -2690,6 +2703,10 @@ class TestFLBADenseDecode : public ::testing::Test {
     }
   }
 
+  void CheckDenseDecode(FLBADecoder* decoder) {
+    CheckDenseDecode(decoder, {1, 17, kNumValues / 2, kNumValues});
+  }
+
  protected:
   static constexpr int kNumValues = 1000;
   int type_length_;
@@ -2754,4 +2771,49 @@ TEST_F(TestFLBADenseDecode, ByteStreamSplit) {
   ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
 }
 
+TEST_F(TestFLBADenseDecode, PlainRejectsTruncatedDenseBuffer) {
+  auto encoder =
+      MakeTypedEncoder<FLBAType>(Encoding::PLAIN, /*use_dictionary=*/false, descr_.get());
+  encoder->Put(draws_.data(), kNumValues);
+  auto buffer = encoder->FlushValues();
+
+  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::PLAIN, descr_.get());
+  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size() - 1));
+
+  std::vector<uint8_t> dense(static_cast<size_t>(type_length_) * kNumValues);
+  ASSERT_THROW(decoder->Decode(dense.data(), kNumValues), ParquetException);
+}
+
+TEST_F(TestFLBADenseDecode, DictionaryRejectsOutOfBoundsDenseIndex) {
+  auto dict_decoder = MakeTypedDecoder<FLBAType>(Encoding::PLAIN, descr_.get());
+  dict_decoder->SetData(/*num_values=*/1, draws_[0].ptr, type_length_);
+
+  auto decoder = MakeDictDecoder<FLBAType>(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<uint8_t> indices = {1, 2, 1};
+  decoder->SetData(/*num_values=*/1, indices.data(), static_cast<int>(indices.size()));
+
+  auto flba_decoder = dynamic_cast<FLBADecoder*>(decoder.get());
+  ASSERT_NE(nullptr, flba_decoder);
+  std::vector<uint8_t> dense(static_cast<size_t>(type_length_));
+  ASSERT_THROW(flba_decoder->Decode(dense.data(), /*max_values=*/1), ParquetException);
+}
+
+TEST_F(TestFLBADenseDecode, DeltaByteArrayRejectsWrongFLBALengthDense) {
+  std::string suffix(static_cast<size_t>(type_length_ - 1), 'x');
+  auto buffer =
+      ::arrow::ConcatenateBuffers({DeltaEncode({0}), DeltaEncode({type_length_ - 1}),
+                                   std::make_shared<Buffer>(suffix)})
+          .ValueOrDie();
+
+  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY, descr_.get());
+  decoder->SetData(/*num_values=*/1, buffer->data(), static_cast<int>(buffer->size()));
+
+  std::vector<uint8_t> dense(static_cast<size_t>(type_length_));
+  ASSERT_THROW(decoder->Decode(dense.data(), /*max_values=*/1), ParquetException);
+}
+
 }  // namespace parquet::test

From 5f60e743c9e8a8c18cc4daedc86d43408e851961 Mon Sep 17 00:00:00 2001
From: Marcin Krystianc <marcin.krystianc@gmail.com>
Date: Tue, 7 Jul 2026 14:23:52 +0200
Subject: [PATCH 3/6] PR remarks

---
 cpp/src/parquet/decoder.cc | 68 +++++++++++++++++---------------------
 cpp/src/parquet/encoding.h |  2 +-
 2 files changed, 32 insertions(+), 38 deletions(-)

diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc
index 0c2a1fa0f4d3..47f2b46ebaac 100644
--- a/cpp/src/parquet/decoder.cc
+++ b/cpp/src/parquet/decoder.cc
@@ -863,7 +863,7 @@ class PlainFLBADecoder : public PlainDecoder<FLBAType>, public FLBADecoder {
   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<int64_t>(this->type_length_) * max_values;
-    if (bytes_to_decode > this->len_ || bytes_to_decode > INT_MAX) {
+    if (bytes_to_decode > this->len_) {
       ParquetException::EofException();
     }
     if (bytes_to_decode > 0) {
@@ -1231,35 +1231,6 @@ int DictDecoderImpl<FLBAType>::DecodeArrow(
   return num_values - null_count;
 }
 
-// Dictionary decoder for FIXED_LEN_BYTE_ARRAY that can decode directly into a
-// caller-owned, densely packed byte buffer. DictDecoderImpl<FLBAType> 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<FLBAType>, public FLBADecoder {
- public:
-  using Base = DictDecoderImpl<FLBAType>;
-  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<FLBA>();
-    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<size_t>(type_length));
-    }
-    this->num_values_ -= max_values;
-    return max_values;
-  }
-};
 template <typename Type>
 int DictDecoderImpl<Type>::DecodeArrow(
     int num_values, int null_count, const uint8_t* valid_bits, int64_t valid_bits_offset,
@@ -1479,6 +1450,36 @@ class DictByteArrayDecoderImpl : public DictDecoderImpl<ByteArrayType> {
   }
 };
 
+// Dictionary decoder for FIXED_LEN_BYTE_ARRAY that can decode directly into a
+// caller-owned, densely packed byte buffer. DictDecoderImpl<FLBAType> 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<FLBAType>, public FLBADecoder {
+ public:
+  using Base = DictDecoderImpl<FLBAType>;
+  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<FLBA>();
+    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<size_t>(type_length));
+    }
+    this->num_values_ -= max_values;
+    return max_values;
+  }
+};
+
 // ----------------------------------------------------------------------
 // DELTA_BINARY_PACKED decoder
 
@@ -2445,13 +2446,6 @@ class ByteStreamSplitDecoder<FLBAType> : public ByteStreamSplitDecoderBase<FLBAT
 
 }  // namespace
 
-// Default for the dense (densely packed) FLBA decode. Encodings override this;
-// the base throws so an unimplemented encoding fails with a clear message.
-int FLBADecoder::Decode(uint8_t* /*buffer*/, int /*max_values*/) {
-  throw ParquetException(
-      "Dense FIXED_LEN_BYTE_ARRAY decoding is not implemented for this encoding");
-}
-
 // ----------------------------------------------------------------------
 // Factory functions
 
diff --git a/cpp/src/parquet/encoding.h b/cpp/src/parquet/encoding.h
index fc6a42d8976d..9a0cc55aa185 100644
--- a/cpp/src/parquet/encoding.h
+++ b/cpp/src/parquet/encoding.h
@@ -426,7 +426,7 @@ class FLBADecoder : virtual public TypedDecoder<FLBAType> {
   /// except at the end of the current data page.
   ///
   /// \note API EXPERIMENTAL
-  virtual int Decode(uint8_t* buffer, int max_values);
+  virtual int Decode(uint8_t* buffer, int max_values) = 0;
 };
 
 PARQUET_EXPORT

From 90c8a837f1d2246959668158edf8c4245fd1761f Mon Sep 17 00:00:00 2001
From: Marcin Krystianc <marcin.krystianc@gmail.com>
Date: Wed, 5 Aug 2026 10:16:08 +0200
Subject: [PATCH 4/6] Test all non-dictionary encodings in a single loop

---
 cpp/src/parquet/encoding_test.cc | 46 +++++++++++---------------------
 1 file changed, 16 insertions(+), 30 deletions(-)

diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc
index e7c941184806..44a64df8b28a 100644
--- a/cpp/src/parquet/encoding_test.cc
+++ b/cpp/src/parquet/encoding_test.cc
@@ -2715,15 +2715,23 @@ class TestFLBADenseDecode : public ::testing::Test {
   std::shared_ptr<ColumnDescriptor> descr_;
 };
 
-TEST_F(TestFLBADenseDecode, Plain) {
-  auto encoder =
-      MakeTypedEncoder<FLBAType>(Encoding::PLAIN, /*use_dictionary=*/false, descr_.get());
-  encoder->Put(draws_.data(), kNumValues);
-  auto buffer = encoder->FlushValues();
+// 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 decoder = MakeTypedDecoder<FLBAType>(Encoding::PLAIN, descr_.get());
-  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
-  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
+    auto encoder =
+        MakeTypedEncoder<FLBAType>(encoding, /*use_dictionary=*/false, descr_.get());
+    encoder->Put(draws_.data(), kNumValues);
+    auto buffer = encoder->FlushValues();
+
+    auto decoder = MakeTypedDecoder<FLBAType>(encoding, descr_.get());
+    decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
+    // EXPECT rather than ASSERT so one failing encoding doesn't hide the others.
+    EXPECT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
+  }
 }
 
 TEST_F(TestFLBADenseDecode, Dictionary) {
@@ -2749,28 +2757,6 @@ TEST_F(TestFLBADenseDecode, Dictionary) {
   ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
 }
 
-TEST_F(TestFLBADenseDecode, DeltaByteArray) {
-  auto encoder = MakeTypedEncoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY,
-                                            /*use_dictionary=*/false, descr_.get());
-  encoder->Put(draws_.data(), kNumValues);
-  auto buffer = encoder->FlushValues();
-
-  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY, descr_.get());
-  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
-  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
-}
-
-TEST_F(TestFLBADenseDecode, ByteStreamSplit) {
-  auto encoder = MakeTypedEncoder<FLBAType>(Encoding::BYTE_STREAM_SPLIT,
-                                            /*use_dictionary=*/false, descr_.get());
-  encoder->Put(draws_.data(), kNumValues);
-  auto buffer = encoder->FlushValues();
-
-  auto decoder = MakeTypedDecoder<FLBAType>(Encoding::BYTE_STREAM_SPLIT, descr_.get());
-  decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
-  ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
-}
-
 TEST_F(TestFLBADenseDecode, PlainRejectsTruncatedDenseBuffer) {
   auto encoder =
       MakeTypedEncoder<FLBAType>(Encoding::PLAIN, /*use_dictionary=*/false, descr_.get());

From 7bb950b1d4794fedeaa01c22648275b75037429f Mon Sep 17 00:00:00 2001
From: Marcin Krystianc <marcin.krystianc@gmail.com>
Date: Wed, 5 Aug 2026 13:45:07 +0200
Subject: [PATCH 5/6] TestFLBADenseDecode

---
 cpp/src/parquet/encoding_test.cc | 25 +++++++++++++++++++++++--
 1 file changed, 23 insertions(+), 2 deletions(-)

diff --git a/cpp/src/parquet/encoding_test.cc b/cpp/src/parquet/encoding_test.cc
index 44a64df8b28a..74b422cb606f 100644
--- a/cpp/src/parquet/encoding_test.cc
+++ b/cpp/src/parquet/encoding_test.cc
@@ -2718,8 +2718,8 @@ class TestFLBADenseDecode : public ::testing::Test {
 // 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}) {
+  for (auto encoding :
+       {Encoding::PLAIN, Encoding::DELTA_BYTE_ARRAY, Encoding::BYTE_STREAM_SPLIT}) {
     SCOPED_TRACE(EncodingToString(encoding));
 
     auto encoder =
@@ -2757,6 +2757,27 @@ TEST_F(TestFLBADenseDecode, Dictionary) {
   ASSERT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(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<FLBA>(kNumValues, draws_.data(), &data_buffer_,
+                               prefixed_probability);
+
+    auto encoder = MakeTypedEncoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY,
+                                              /*use_dictionary=*/false, descr_.get());
+    encoder->Put(draws_.data(), kNumValues);
+    auto buffer = encoder->FlushValues();
+
+    auto decoder = MakeTypedDecoder<FLBAType>(Encoding::DELTA_BYTE_ARRAY, descr_.get());
+    decoder->SetData(kNumValues, buffer->data(), static_cast<int>(buffer->size()));
+    EXPECT_NO_FATAL_FAILURE(CheckDenseDecode(dynamic_cast<FLBADecoder*>(decoder.get())));
+  }
+}
+
 TEST_F(TestFLBADenseDecode, PlainRejectsTruncatedDenseBuffer) {
   auto encoder =
       MakeTypedEncoder<FLBAType>(Encoding::PLAIN, /*use_dictionary=*/false, descr_.get());

From 550ba671e17aaf1fee62d96b3a006653c5ca4385 Mon Sep 17 00:00:00 2001
From: Marcin Krystianc <marcin.krystianc@gmail.com>
Date: Thu, 6 Aug 2026 11:32:06 +0200
Subject: [PATCH 6/6] Decode DELTA_BYTE_ARRAY / FLBA directly into output
 buffer

---
 cpp/src/parquet/decoder.cc | 308 ++++++++++++++++++++++++-------------
 1 file changed, 202 insertions(+), 106 deletions(-)

diff --git a/cpp/src/parquet/decoder.cc b/cpp/src/parquet/decoder.cc
index 47f2b46ebaac..008e192cd633 100644
--- a/cpp/src/parquet/decoder.cc
+++ b/cpp/src/parquet/decoder.cc
@@ -1779,12 +1779,19 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl<ByteArrayType> {
     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;
     }
 
@@ -1796,7 +1803,6 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl<ByteArrayType> {
       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,13 +1811,24 @@ class DeltaLengthByteArrayDecoder : public TypedDecoderImpl<ByteArrayType> {
     if (ARROW_PREDICT_FALSE(!decoder_->Advance(8 * static_cast<int64_t>(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<uint32_t>(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;
   }
 
@@ -2032,7 +2049,8 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl<DType> {
         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;
@@ -2082,93 +2100,171 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl<DType> {
   }
 
  protected:
-  template <bool is_first_run>
-  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<size_t>(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);
     }
-    DCHECK_EQ(is_first_run, i == 0);
-    if constexpr (!is_first_run) {
-      if (buffer[i].len == 0) {
+
+    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;
+    }
+
+    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<size_t>(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<uint32_t>(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<DType, FLBAType>) {
+        // Checks all values
+        for (int i = 0; i < max_values; ++i) {
+          if (buffer[i].len != static_cast<uint32_t>(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<size_t>(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<int64_t>(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<int64_t>(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<const char*>(dst),
+                              static_cast<size_t>(type_length)};
+    }
+
+    void Finish(int max_values) {}
+  };
+
+  template <typename Output>
+  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<int32_t>() + 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</*is_first_run=*/true>(prefix_len_ptr, 0, buffer, &prefix,
-                                                 &data_ptr);
-    }
-    for (int i = 1; i < max_values; ++i) {
-      BuildBufferInternal</*is_first_run=*/false>(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 +2274,30 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl<DType> {
       last_value_in_previous_page_ = last_value_;
     }
 
-    if constexpr (std::is_same_v<DType, FLBAType>) {
-      // Checks all values
-      for (int i = 0; i < max_values; i++) {
-        if (buffer[i].len != static_cast<uint32_t>(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<ByteArray>();
+  }
+
+  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<DType>::Accumulator* out,
@@ -2248,6 +2355,9 @@ class DeltaByteArrayDecoderImpl : public TypedDecoderImpl<DType> {
   // buffer for decoded strings, which guarantees the lifetime of the decoded strings
   // until the next call of Decode.
   std::shared_ptr<ResizableBuffer> buffered_data_;
+  // Scratch buffer of ByteArray used to decode suffixes / expose value pointers
+  // without a per-call heap allocation.
+  std::shared_ptr<ResizableBuffer> buffered_suffix_;
 };
 
 class DeltaByteArrayDecoder : public DeltaByteArrayDecoderImpl<ByteArrayType> {
@@ -2268,35 +2378,21 @@ class DeltaByteArrayFLBADecoder : public DeltaByteArrayDecoderImpl<FLBAType>,
   using Base::pool_;
 
   int Decode(FixedLenByteArray* buffer, int max_values) override {
-    // GetInternal currently only support ByteArray.
-    std::vector<ByteArray> decode_byte_array(max_values);
-    const int decoded_values_size = GetInternal(decode_byte_array.data(), max_values);
-    const uint32_t type_length = static_cast<uint32_t>(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;
   }
 
-  // Same internal decode as above, but copy the bytes contiguously into the
-  // caller's buffer instead of materializing per-value pointers.
+  // Decode the bytes contiguously into the caller's buffer without
+  // materializing per-value pointers.
   int Decode(uint8_t* buffer, int max_values) override {
-    std::vector<ByteArray> decode_byte_array(max_values);
-    const int decoded_values_size = GetInternal(decode_byte_array.data(), max_values);
-    const uint32_t type_length = static_cast<uint32_t>(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<int64_t>(i) * type_length, decode_byte_array[i].ptr,
-             type_length);
-    }
-    return decoded_values_size;
+    return this->DecodeDense(buffer, max_values);
   }
 };