fix(parquet): support nullable fixed-size lists for vector (#231)

Author: lxy-9602

cmake_modules/arrow.diff

@@ -48,10 +48,71 @@ index b36c38c6d4..f974a33073 100644
 
    /// \brief Return zero-copy string_view to upcoming bytes.
    ///
+diff --git a/cpp/src/arrow/util/bit_run_reader.h b/cpp/src/arrow/util/bit_run_reader.h
+index a436a503a0..27d483978c 100644
+--- a/cpp/src/arrow/util/bit_run_reader.h
++++ b/cpp/src/arrow/util/bit_run_reader.h
+@@ -168,6 +168,26 @@ class ARROW_EXPORT BitRunReader {
+ using BitRunReader = BitRunReaderLinear;
+ #endif
+
++template <typename Visit>
++inline Status VisitBitRuns(const uint8_t* bitmap, int64_t offset, int64_t length,
++                           Visit&& visit) {
++  if (bitmap == NULLPTR) {
++    // Assuming all set (as in a null bitmap)
++    return visit(static_cast<int64_t>(0), length, true);
++  }
++  BitRunReader reader(bitmap, offset, length);
++  int64_t position = 0;
++  while (true) {
++    const auto run = reader.NextRun();
++    if (run.length == 0) {
++      break;
++    }
++    ARROW_RETURN_NOT_OK(visit(position, run.length, run.set));
++    position += run.length;
++  }
++  return Status::OK();
++}
++
+ struct SetBitRun {
+   int64_t position;
+   int64_t length;
 diff --git a/cpp/src/parquet/arrow/reader.cc b/cpp/src/parquet/arrow/reader.cc
-index 285e2a5973..db919d7ef8 100644
+index 285e2a5973..52f42cf5b3 100644
 --- a/cpp/src/parquet/arrow/reader.cc
 +++ b/cpp/src/parquet/arrow/reader.cc
+@@ -19,12 +19,14 @@
+
+ #include <algorithm>
+ #include <cstring>
++#include <iterator>
+ #include <memory>
+ #include <unordered_set>
+ #include <utility>
+ #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,12 +34,14 @@
+ #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"
+ #include "arrow/util/logging.h"
+ #include "arrow/util/parallel.h"
+ #include "arrow/util/range.h"
++#include "arrow/util/span.h"
+ #include "arrow/util/tracing_internal.h"
+ #include "parquet/arrow/reader_internal.h"
+ #include "parquet/column_reader.h"
 @@ -254,6 +254,11 @@ class FileReaderImpl : public FileReader {
      return GetColumn(i, AllRowGroupsFactory(), out);
    }
@@ -151,7 +212,87 @@ index 285e2a5973..db919d7ef8 100644
    virtual ::arrow::Result<std::shared_ptr<ChunkedArray>> AssembleArray(
        std::shared_ptr<ArrayData> data) {
      if (field_->type()->id() == ::arrow::Type::MAP) {
-@@ -709,6 +776,39 @@ class PARQUET_NO_EXPORT StructReader : public ColumnReaderImpl {
+@@ -642,8 +713,10 @@ class ListReader : public ColumnReaderImpl {
+
+   const std::shared_ptr<Field> field() override { return field_; }
+
+- private:
++ protected:
+   std::shared_ptr<ReaderContext> ctx_;
++
++ private:
+   std::shared_ptr<Field> field_;
+   ::parquet::internal::LevelInfo level_info_;
+   std::unique_ptr<ColumnReaderImpl> item_reader_;
+@@ -662,12 +735,62 @@ class PARQUET_NO_EXPORT FixedSizeListReader : public ListReader<int32_t> {
+     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;
++      ::arrow::util::span<const int32_t> run_offsets(
++          offsets + start, static_cast<size_t>(length + 1));
++      const auto first_invalid_offset = std::adjacent_find(
++          run_offsets.begin(), run_offsets.end(),
++          [&](int32_t left, int32_t right) { return right - left != expected_size; });
++      if (first_invalid_offset != run_offsets.end()) {
++        const int64_t x =
++            start + std::distance(run_offsets.begin(), first_invalid_offset);
++        const int32_t size = offsets[x + 1] - offsets[x];
++        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) {
++      // Rebuild the child array run-by-run so null fixed-size list slots still
++      // contribute list_size child values in the final layout.
++      ::arrow::ArrayVector child_arrays;
++
++      auto visit_run = [&](int64_t start, int64_t length, bool has_elements) -> Status {
++        RETURN_NOT_OK(validate_offsets(start, length, has_elements));
++
++        const int64_t child_length = length * list_size;
++        // Valid runs reuse the decoded child slice; null runs materialize null
++        // children to preserve the fixed-size list shape.
++        if (!has_elements) {
++          ARROW_ASSIGN_OR_RAISE(
++              auto null_array,
++              ::arrow::MakeArrayOfNull(type.value_type(), child_length, ctx_->pool));
++          child_arrays.push_back(std::move(null_array));
++          return Status::OK();
++        }
++        child_arrays.push_back(
++            ::arrow::MakeArray(data->child_data[0]->Slice(offsets[start], child_length)));
++        return Status::OK();
++      };
++
++      DCHECK_NE(data->buffers[0], nullptr);
++      RETURN_NOT_OK(::arrow::internal::VisitBitRuns(
++          data->buffers[0]->data(), data->offset, data->length, visit_run));
++
++      // TODO(GH-50271): Build one padded child array directly instead of creating
++      // one temporary Array/ArrayData per validity run and concatenating them.
++      ARROW_ASSIGN_OR_RAISE(auto child_array_with_padding,
++                            ::arrow::Concatenate(child_arrays, ctx_->pool));
++      data->child_data[0] = child_array_with_padding->data();
++    } else {
++      RETURN_NOT_OK(validate_offsets(/*start=*/0, data->length, /*valid=*/true));
+     }
+     data->buffers.resize(1);
+     std::shared_ptr<Array> result = ::arrow::MakeArray(data);
+@@ -709,6 +832,39 @@ class PARQUET_NO_EXPORT StructReader : public ColumnReaderImpl {
      }
      return Status::OK();
    }
@@ -191,7 +332,7 @@ index 285e2a5973..db919d7ef8 100644
    Status BuildArray(int64_t length_upper_bound,
                      std::shared_ptr<ChunkedArray>* out) override;
    Status GetDefLevels(const int16_t** data, int64_t* length) override;
-@@ -1013,25 +1113,32 @@ Status FileReaderImpl::GetRecordBatchReader(const std::vector<int>& row_groups,
+@@ -1013,25 +1169,32 @@ Status FileReaderImpl::GetRecordBatchReader(const std::vector<int>& row_groups,
      return Status::OK();
    }
 
@@ -230,7 +371,7 @@ index 285e2a5973..db919d7ef8 100644
 
          RETURN_NOT_OK(::arrow::internal::OptionalParallelFor(
              reader_properties_.use_threads(), static_cast<int>(readers.size()),
-@@ -1224,6 +1331,23 @@ Status FileReaderImpl::GetColumn(int i, FileColumnIteratorFactory iterator_facto
+@@ -1224,6 +1387,23 @@ Status FileReaderImpl::GetColumn(int i, FileColumnIteratorFactory iterator_facto
    return Status::OK();
  }
 
@@ -400,10 +541,49 @@ index ec3890a41f..943f69bb6c 100644
      return Status::OK();
    }
 diff --git a/cpp/src/parquet/arrow/writer.cc b/cpp/src/parquet/arrow/writer.cc
-index 4fd7ef1b47..87326a54f1 100644
+index 4fd7ef1b47..feff99c99b 100644
 --- a/cpp/src/parquet/arrow/writer.cc
 +++ b/cpp/src/parquet/arrow/writer.cc
-@@ -314,6 +314,14 @@ class FileWriterImpl : public FileWriter {
+@@ -26,6 +26,7 @@
+ #include <vector>
+
+ #include "arrow/array.h"
++#include "arrow/array/concatenate.h"
+ #include "arrow/extension_type.h"
+ #include "arrow/ipc/writer.h"
+ #include "arrow/record_batch.h"
+@@ -142,13 +143,24 @@ class ArrowColumnWriterV2 {
+             leaf_idx, ctx, [&](const MultipathLevelBuilderResult& result) {
+               size_t visited_component_size = result.post_list_visited_elements.size();
+               DCHECK_GT(visited_component_size, 0);
+-              if (visited_component_size != 1) {
+-                return Status::NotImplemented(
+-                    "Lists with non-zero length null components are not supported");
++              std::shared_ptr<Array> values_array;
++              if (visited_component_size == 1) {
++                const ElementRange& range = result.post_list_visited_elements[0];
++                values_array = result.leaf_array->Slice(range.start, range.Size());
++              } else {
++                // Multiple leaf ranges can be produced when child values are
++                // skipped, such as null fixed-size-list slots, or when
++                // list-view ranges are non-contiguous. Concatenate the slices
++                // in logical write order.
++                ::arrow::ArrayVector arrays;
++                arrays.reserve(visited_component_size);
++                for (const auto& range : result.post_list_visited_elements) {
++                  DCHECK(!range.Empty());
++                  arrays.push_back(result.leaf_array->Slice(range.start, range.Size()));
++                }
++                ARROW_ASSIGN_OR_RAISE(values_array,
++                                      ::arrow::Concatenate(arrays, ctx->memory_pool));
+               }
+-              const ElementRange& range = result.post_list_visited_elements[0];
+-              std::shared_ptr<Array> values_array =
+-                  result.leaf_array->Slice(range.start, range.Size());
+
+               return column_writer->WriteArrow(result.def_levels, result.rep_levels,
+                                                result.def_rep_level_count, *values_array,
+@@ -314,6 +326,14 @@ class FileWriterImpl : public FileWriter {
      return Status::OK();
    }
 
@@ -418,7 +598,7 @@ index 4fd7ef1b47..87326a54f1 100644
    Status Close() override {
      if (!closed_) {
        // Make idempotent
-@@ -418,10 +426,13 @@ class FileWriterImpl : public FileWriter {
+@@ -418,10 +438,13 @@ class FileWriterImpl : public FileWriter {
 
      // Max number of rows allowed in a row group.
      const int64_t max_row_group_length = this->properties().max_row_group_length();

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.
+       Paimon C++ also reads Parquet files written by Paimon Rust or Python whose
+       embedded Arrow schema restores VECTOR columns as ``FixedSizeList``,
+       including NULL vector values.
 
    * - ``MAP<kt, vt>``
      - Map

src/paimon/core/io/vector_file_batch_reader.cpp

@@ -119,6 +119,11 @@ Result<std::shared_ptr<arrow::Array>> CastListToVector(
             fmt::format("Cannot restore VECTOR from type {}", array->type()->ToString()));
     }
     PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*array));
+    if (array->null_count() == array->length()) {
+        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> result,
+                                          arrow::MakeArrayOfNull(read_type, array->length(), pool));
+        return result;
+    }
     arrow::compute::ExecContext exec_context(pool);
     arrow::TypeHolder type_holder(read_type.get());
     arrow::compute::CastOptions options = arrow::compute::CastOptions::Safe();

src/paimon/format/parquet/CMakeLists.txt

@@ -20,7 +20,6 @@ set(PAIMON_PARQUET_FILE_FORMAT
     file_reader_wrapper.cpp
     page_filtered_row_group_reader.cpp
     parquet_timestamp_converter.cpp
-    parquet_vector_converter.cpp
     parquet_file_batch_reader.cpp
     parquet_file_format_factory.cpp
     parquet_format_writer.cpp
@@ -56,7 +55,6 @@ if(PAIMON_BUILD_TESTS)
                     file_reader_wrapper_test.cpp
                     page_filtered_row_group_reader_test.cpp
                     parquet_timestamp_converter_test.cpp
-                    parquet_vector_converter_test.cpp
                     parquet_vector_io_test.cpp
                     parquet_field_id_converter_test.cpp
                     parquet_file_batch_reader_test.cpp

src/paimon/format/parquet/parquet_format_writer.cpp

@@ -23,7 +23,6 @@
 #include <string_view>
 #include <utility>
 
-#include "arrow/array/array_nested.h"
 #include "arrow/c/bridge.h"
 #include "arrow/memory_pool.h"
 #include "arrow/record_batch.h"
@@ -32,9 +31,7 @@
 #include "paimon/common/metrics/metrics_impl.h"
 #include "paimon/common/utils/arrow/arrow_output_stream_adapter.h"
 #include "paimon/common/utils/arrow/status_utils.h"
-#include "paimon/common/utils/checked_cast.h"
 #include "paimon/format/parquet/parquet_format_defs.h"
-#include "paimon/format/parquet/parquet_vector_converter.h"
 #include "parquet/arrow/writer.h"
 #include "parquet/properties.h"
 
@@ -58,33 +55,17 @@ Result<std::unique_ptr<ParquetFormatWriter>> ParquetFormatWriter::Create(
     ::parquet::ArrowWriterProperties::Builder arrow_properties_builder;
     auto arrow_writer_properties =
         arrow_properties_builder.enable_deprecated_int96_timestamps()->build();
-    auto logical_type = arrow::struct_(schema->fields());
-    auto write_type =
-        checked_pointer_cast<arrow::StructType>(ParquetVectorConverter::GetWriteType(logical_type));
-    auto write_schema = arrow::schema(write_type->fields(), schema->metadata());
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
         std::unique_ptr<::parquet::arrow::FileWriter> file_writer,
-        ::parquet::arrow::FileWriter::Open(*write_schema, pool.get(), out, writer_properties,
+        ::parquet::arrow::FileWriter::Open(*schema, pool.get(), out, writer_properties,
                                            arrow_writer_properties));
-    return std::unique_ptr<ParquetFormatWriter>(new ParquetFormatWriter(
-        std::move(file_writer), out, schema, max_memory_use,
-        /*needs_vector_conversion=*/!logical_type->Equals(write_type), pool));
+    return std::unique_ptr<ParquetFormatWriter>(
+        new ParquetFormatWriter(std::move(file_writer), out, schema, max_memory_use, pool));
 }
 
 Status ParquetFormatWriter::AddBatch(ArrowArray* batch) {
     PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<::arrow::RecordBatch> record_batch,
                                       arrow::ImportRecordBatch(batch, schema_));
-    if (needs_vector_conversion_) {
-        // TODO(ChaomingZhangCN): Remove this conversion after upgrading Arrow. Arrow 17
-        // mishandles nullable FixedSizeList values when writing them as Parquet LIST.
-        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::StructArray> struct_array,
-                                          record_batch->ToStructArray());
-        std::shared_ptr<arrow::Array> array = struct_array;
-        PAIMON_ASSIGN_OR_RAISE(array,
-                               ParquetVectorConverter::ConvertToWriteType(array, pool_.get()));
-        PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(record_batch,
-                                          arrow::RecordBatch::FromStructArray(array, pool_.get()));
-    }
     if (static_cast<uint64_t>(pool_->bytes_allocated()) > max_memory_use_) {
         PAIMON_RETURN_NOT_OK_FROM_ARROW(writer_->NewBufferedRowGroup());
     }
@@ -132,14 +113,13 @@ Result<uint64_t> ParquetFormatWriter::GetEstimateLength() const {
 ParquetFormatWriter::ParquetFormatWriter(std::unique_ptr<::parquet::arrow::FileWriter> writer,
                                          const std::shared_ptr<ArrowOutputStreamAdapter>& out,
                                          const std::shared_ptr<arrow::Schema>& schema,
-                                         uint64_t max_memory_use, bool needs_vector_conversion,
+                                         uint64_t max_memory_use,
                                          const std::shared_ptr<arrow::MemoryPool>& pool)
     : pool_(pool),
       out_(out),
       writer_(std::move(writer)),
       schema_(schema),
       metrics_(std::make_shared<MetricsImpl>()),
-      max_memory_use_(max_memory_use),
-      needs_vector_conversion_(needs_vector_conversion) {}
+      max_memory_use_(max_memory_use) {}
 
 }  // namespace paimon::parquet

src/paimon/format/parquet/parquet_format_writer.h

@@ -72,7 +72,6 @@ class ParquetFormatWriter : public FormatWriter {
     ParquetFormatWriter(std::unique_ptr<::parquet::arrow::FileWriter> writer,
                         const std::shared_ptr<ArrowOutputStreamAdapter>& out,
                         const std::shared_ptr<arrow::Schema>& schema, uint64_t max_memory_use,
-                        bool needs_vector_conversion,
                         const std::shared_ptr<arrow::MemoryPool>& pool);
 
     Result<uint64_t> GetEstimateLength() const;
@@ -84,7 +83,6 @@ class ParquetFormatWriter : public FormatWriter {
     std::shared_ptr<Metrics> metrics_;
     int64_t total_records_written_ = 0;
     uint64_t max_memory_use_;
-    bool needs_vector_conversion_;
 };
 
 }  // namespace paimon::parquet