fix(arrow): read null values from a FixedSizeList Parquet column

cmake_modules/arrow.diff

@@ -613,6 +613,93 @@ diff --git a/cpp/src/arrow/io/interfaces.h b/cpp/src/arrow/io/interfaces.h
    return Status::OK();
  }
 
+diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc
+--- a/cpp/src/parquet/arrow/reader.cc
++++ b/cpp/src/parquet/arrow/reader.cc
+@@ -25,6 +25,7 @@
+ #include <vector>
+
+ #include "arrow/array.h"
++#include "arrow/array/concatenate.h"
+ #include "arrow/buffer.h"
+ #include "arrow/extension_type.h"
+ #include "arrow/io/memory.h"
+@@ -32,6 +33,7 @@
+ #include "arrow/table.h"
+ #include "arrow/type.h"
+ #include "arrow/util/async_generator.h"
++#include "arrow/util/bit_run_reader.h"
+ #include "arrow/util/bit_util.h"
+ #include "arrow/util/future.h"
+ #include "arrow/util/iterator.h"
+@@ -709,7 +711,7 @@
+
+   const std::shared_ptr<Field> field() override { return field_; }
+
+- private:
++ protected:
+   std::shared_ptr<ReaderContext> ctx_;
+   std::shared_ptr<Field> field_;
+   ::parquet::internal::LevelInfo level_info_;
+@@ -729,12 +731,53 @@
+     DCHECK_EQ(field()->type()->id(), ::arrow::Type::FIXED_SIZE_LIST);
+     const auto& type = checked_cast<::arrow::FixedSizeListType&>(*field()->type());
+     const int32_t* offsets = reinterpret_cast<const int32_t*>(data->buffers[1]->data());
+-    for (int x = 1; x <= data->length; x++) {
+-      int32_t size = offsets[x] - offsets[x - 1];
+-      if (size != type.list_size()) {
+-        return Status::Invalid("Expected all lists to be of size=", type.list_size(),
+-                               " but index ", x, " had size=", size);
++    const int32_t list_size = type.list_size();
++    auto validate_offsets = [&](int64_t start, int64_t length, bool has_elements) -> Status {
++      const int32_t expected_size = has_elements ? list_size : 0;
++      for (int64_t x = start; x < start + length; x++) {
++        const int32_t size = offsets[x + 1] - offsets[x];
++        if (size != expected_size) {
++          if (has_elements) {
++            return Status::Invalid("Expected all lists to be of size=", list_size,
++                                   " but index ", x + 1, " had size=", size);
++          }
++          return Status::Invalid("Expected null fixed-size list at index ", x + 1,
++                                 " to have no child values but had size=", size);
++        }
++      }
++      return Status::OK();
++    };
++    if (data->GetNullCount() != 0) {
++      // Parquet stores a null fixed-size list slot without any child value, so rebuild the
++      // child array run by run to restore the list_size values every slot must have.
++      DCHECK_NE(data->buffers[0], nullptr);
++      ::arrow::ArrayVector child_arrays;
++      ::arrow::internal::BitRunReader run_reader(data->buffers[0]->data(), data->offset,
++                                                 data->length);
++      int64_t start = 0;
++      while (start < data->length) {
++        ::arrow::internal::BitRun run = run_reader.NextRun();
++        if (run.length == 0) {
++          break;
++        }
++        RETURN_NOT_OK(validate_offsets(start, run.length, run.set));
++        const int64_t child_length = run.length * list_size;
++        if (run.set) {
++          child_arrays.push_back(
++              ::arrow::MakeArray(data->child_data[0]->Slice(offsets[start], child_length)));
++        } else {
++          ARROW_ASSIGN_OR_RAISE(
++              std::shared_ptr<::arrow::Array> null_array,
++              ::arrow::MakeArrayOfNull(type.value_type(), child_length, ctx_->pool));
++          child_arrays.push_back(std::move(null_array));
++        }
++        start += run.length;
+       }
++      ARROW_ASSIGN_OR_RAISE(std::shared_ptr<::arrow::Array> padded_child,
++                            ::arrow::Concatenate(child_arrays, ctx_->pool));
++      data->child_data[0] = padded_child->data();
++    } else {
++      RETURN_NOT_OK(validate_offsets(/*start=*/0, data->length, /*has_elements=*/true));
+     }
+     data->buffers.resize(1);
+     std::shared_ptr<Array> result = ::arrow::MakeArray(data);
 --- a/cpp/src/parquet/arrow/reader.h
 +++ b/cpp/src/parquet/arrow/reader.h
 @@ -21,6 +21,7 @@

docs/source/user_guide/data_types.rst

@@ -201,11 +201,9 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
        VECTOR fields are rejected. VECTOR columns also cannot be partition or
        bucket keys. Dedicated vector storage is not included yet.
 
-       **Note:** A data file written by another engine that records the column as
-       Arrow ``FixedSizeList`` instead of ``LIST``, such as Paimon Rust or Python,
-       can only be read while it holds no NULL vector. Parquet stores a NULL list
-       slot with no values, which the Arrow 17 Parquet reader rejects for a
-       ``FixedSizeList`` column.
+       A data file written by another engine is read either way: engines such as
+       Paimon Java record the column as a Parquet ``LIST``, while Paimon Rust and
+       Python record it as Arrow ``FixedSizeList``.
 
    * - ``MAP<kt, vt>``
      - Map

src/paimon/format/parquet/parquet_vector_io_test.cpp

@@ -369,25 +369,13 @@ TEST_F(ParquetVectorIoTest, ReadNullableJavaFixture) {
 }
 
 // A writer that stores the Arrow schema, such as Paimon Rust or Python, exposes the VECTOR column
-// as FixedSizeList. Arrow 17 cannot read a null value from such a column: Parquet stores a null
-// list slot with no values, while FixedSizeListReader::AssembleArray in
-// parquet/arrow/reader.cc requires every slot to span exactly `list_size` values.
-//
-// TODO(ChaomingZhangCN): Turn this into a read check once Arrow is upgraded.
-TEST_F(ParquetVectorIoTest, ReadNullableRustFixtureIsUnsupported) {
-    std::string file_path =
-        paimon::test::GetDataDir() + "/parquet/vector_compatibility/rust_vector_nullable.parquet";
-    std::shared_ptr<arrow::StructType> file_type;
-    ReadFileType(file_path, &file_type);
-    std::shared_ptr<arrow::Field> file_vector_field = file_type->GetFieldByName("embedding");
-    ASSERT_TRUE(file_vector_field);
-    ASSERT_EQ(file_vector_field->type()->id(), arrow::Type::FIXED_SIZE_LIST);
-
-    std::unique_ptr<FileBatchReader> reader;
-    CreateVectorReader(file_path, arrow::schema(file_type->fields()), /*predicate=*/nullptr,
-                       /*options=*/{}, /*batch_size=*/10, &reader);
-    ASSERT_NOK_WITH_MSG(paimon::test::ReadResultCollector::CollectResult(reader.get()),
-                        "Expected all lists to be of size=3");
+// as FixedSizeList. Parquet stores a null list slot without any value, and the Arrow patch of
+// FixedSizeListReader::AssembleArray restores the values every fixed size slot must have.
+TEST_F(ParquetVectorIoTest, ReadNullableRustFixture) {
+    ReadFixtureAndCheck("rust_vector_nullable.parquet", arrow::Type::FIXED_SIZE_LIST,
+                        /*vector_length=*/3, /*expected_ids=*/{1, 2, 3},
+                        /*expected_vectors=*/
+                        {{{1.0f, 2.0f, 3.0f}}, std::nullopt, {{4.0f, 5.0f, 6.0f}}});
 }
 
 }  // namespace paimon::parquet::test

test/test_data/parquet/vector_compatibility/README.md

@@ -21,12 +21,10 @@ VECTOR columns, with and without null vectors.
 - `rust_vector_nullable.parquet` was generated the same way, with the rows `(1, [1, 2, 3])`,
   `(2, null)` and `(3, [4, 5, 6])`.
 
-A file that stores the Arrow schema, as the Rust writer does, is read back as
-`fixed_size_list`. Arrow 17 cannot read a null value from such a column, because Parquet stores a
-null list slot with no values while `FixedSizeListReader::AssembleArray` in
-`parquet/arrow/reader.cc` requires every slot to span exactly `list_size` values. Reading
-`rust_vector_nullable.parquet` therefore fails until Arrow is upgraded, which
-`ParquetVectorIoTest.ReadNullableRustFixtureIsUnsupported` pins.
+A file that stores the Arrow schema, as the Rust writer does, is read back as `fixed_size_list`.
+Parquet stores a null list slot with no values, which stock Arrow 17 rejects for such a column;
+`cmake_modules/arrow.diff` patches `FixedSizeListReader::AssembleArray` to restore the values every
+fixed size slot must have.
 
 SHA-256 checksums: