GH-50333: [C++][Parquet] Add dense decode path for FIXED_LEN_BYTE_ARRAY

[marcin-krystianc]

Rationale for this change

For performance reasons, we would like to decode float16 (FLBA) values directly into the caller-provided buffer.

What changes are included in this PR?

• Adds a virtual int Decode(uint8_t* buffer, int max_values) to FLBADecoder in encoding.h. It writes decoded values contiguously into a caller-owned buffer.
• Implements the new overload for all supported decoders. (The base implementation throws ParquetException, so unimplemented encodings fail with a clear message.)
• Replaces the old PARQUET-1508 TODO comment.

Are these changes tested?

Yes

Are there any user-facing changes?

No

PS:

Closes #50333

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - Contributing Overview

[rok]

This looks good @marcin-krystianc !
My only suggestion would be to test the non-happy-path as well to make sure we throw, see diff below.

quick suggestion

diff --git i/cpp/src/parquet/encoding.h w/cpp/src/parquet/encoding.h
index 27b90a5fe9..fc6a42d897 100644
--- i/cpp/src/parquet/encoding.h
+++ w/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 i/cpp/src/parquet/encoding_test.cc w/cpp/src/parquet/encoding_test.cc
index d161b61af5..6e2b21fa62 100644
--- i/cpp/src/parquet/encoding_test.cc
+++ w/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

</details>

For linting errors pre-commit is useful:
```bash
pre-commit run --show-diff-on-failure --color=always --all-files cpp