This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 05d6497b fix(parquet): support nullable fixed-size lists for vector
(#231)
05d6497b is described below
commit 05d6497bdd8111fbec45eba731ea5b2c06ce0d0b
Author: lxy <[email protected]>
AuthorDate: Fri Aug 21 19:41:52 2026 +0800
fix(parquet): support nullable fixed-size lists for vector (#231)
---
cmake_modules/arrow.diff | 194 ++++++++++++++++++++-
docs/source/user_guide/data_types.rst | 8 +-
src/paimon/core/io/vector_file_batch_reader.cpp | 5 +
src/paimon/format/parquet/CMakeLists.txt | 2 -
.../format/parquet/parquet_format_writer.cpp | 30 +---
src/paimon/format/parquet/parquet_format_writer.h | 2 -
.../format/parquet/parquet_vector_converter.cpp | 174 ------------------
.../format/parquet/parquet_vector_converter.h | 46 -----
.../parquet/parquet_vector_converter_test.cpp | 95 ----------
.../format/parquet/parquet_vector_io_test.cpp | 72 +++++---
.../parquet/vector_compatibility/README.md | 8 +-
11 files changed, 251 insertions(+), 385 deletions(-)
diff --git a/cmake_modules/arrow.diff b/cmake_modules/arrow.diff
index 75e3bb51..b8b83517 100644
--- a/cmake_modules/arrow.diff
+++ b/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();
diff --git a/docs/source/user_guide/data_types.rst
b/docs/source/user_guide/data_types.rst
index 9fdecf6e..add537ad 100644
--- a/docs/source/user_guide/data_types.rst
+++ b/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
diff --git a/src/paimon/core/io/vector_file_batch_reader.cpp
b/src/paimon/core/io/vector_file_batch_reader.cpp
index a4573eef..f16025c4 100644
--- a/src/paimon/core/io/vector_file_batch_reader.cpp
+++ b/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();
diff --git a/src/paimon/format/parquet/CMakeLists.txt
b/src/paimon/format/parquet/CMakeLists.txt
index c31e3cc3..96854658 100644
--- a/src/paimon/format/parquet/CMakeLists.txt
+++ b/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
diff --git a/src/paimon/format/parquet/parquet_format_writer.cpp
b/src/paimon/format/parquet/parquet_format_writer.cpp
index 6e69e694..0a8e38b4 100644
--- a/src/paimon/format/parquet/parquet_format_writer.cpp
+++ b/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
diff --git a/src/paimon/format/parquet/parquet_format_writer.h
b/src/paimon/format/parquet/parquet_format_writer.h
index f8f44119..4ab58d73 100644
--- a/src/paimon/format/parquet/parquet_format_writer.h
+++ b/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
diff --git a/src/paimon/format/parquet/parquet_vector_converter.cpp
b/src/paimon/format/parquet/parquet_vector_converter.cpp
deleted file mode 100644
index 5b6446d2..00000000
--- a/src/paimon/format/parquet/parquet_vector_converter.cpp
+++ /dev/null
@@ -1,174 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#include "paimon/format/parquet/parquet_vector_converter.h"
-
-#include <cstdint>
-#include <limits>
-#include <memory>
-#include <vector>
-
-#include "arrow/array.h"
-#include "arrow/array/array_nested.h"
-#include "arrow/array/builder_primitive.h"
-#include "arrow/compute/api.h"
-#include "arrow/type.h"
-#include "paimon/common/utils/arrow/status_utils.h"
-#include "paimon/common/utils/arrow/vector_utils.h"
-#include "paimon/common/utils/checked_cast.h"
-#include "paimon/status.h"
-
-namespace paimon::parquet {
-namespace {
-
-Result<std::shared_ptr<arrow::Array>> CastToListType(
- const std::shared_ptr<arrow::Array>& array, const
std::shared_ptr<arrow::DataType>& write_type,
- arrow::MemoryPool* pool) {
- arrow::compute::ExecContext exec_context(pool);
- arrow::TypeHolder type_holder(write_type.get());
- arrow::compute::CastOptions options = arrow::compute::CastOptions::Safe();
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
- std::shared_ptr<arrow::Array> result,
- arrow::compute::Cast(*array, type_holder, options, &exec_context));
- return result;
-}
-
-/// Rebuilds a nullable VECTOR as a LIST whose null slots have a zero length,
dropping the
-/// values Arrow keeps for them.
-///
-/// TODO(ChaomingZhangCN): Cast the whole array once Arrow is upgraded. Arrow
17 casts a null
-/// FixedSizeList row to a null LIST slot spanning `list_size` values, and the
Parquet writer
-/// rejects a LIST with non-zero length null slots.
-Result<std::shared_ptr<arrow::Array>> CompactNullVectorsToList(
- const arrow::FixedSizeListArray& vector_array,
- const std::shared_ptr<arrow::DataType>& write_type, arrow::MemoryPool*
pool) {
- const auto& vector_type = checked_cast<const
arrow::FixedSizeListType&>(*vector_array.type());
- const int32_t vector_length = vector_type.list_size();
- if (vector_array.length() > std::numeric_limits<int32_t>::max() /
vector_length) {
- return Status::Invalid("VECTOR values exceed the maximum Parquet LIST
offset");
- }
-
- arrow::Int32Builder offsets_builder(pool);
- arrow::Int64Builder indices_builder(pool);
- arrow::BooleanBuilder validity_builder(pool);
-
PAIMON_RETURN_NOT_OK_FROM_ARROW(offsets_builder.Reserve(vector_array.length() +
1));
-
PAIMON_RETURN_NOT_OK_FROM_ARROW(indices_builder.Reserve(vector_array.length() *
vector_length));
-
PAIMON_RETURN_NOT_OK_FROM_ARROW(validity_builder.Reserve(vector_array.length()));
- PAIMON_RETURN_NOT_OK_FROM_ARROW(offsets_builder.Append(0));
-
- int32_t offset = 0;
- for (int64_t i = 0; i < vector_array.length(); ++i) {
- bool valid = !vector_array.IsNull(i);
- PAIMON_RETURN_NOT_OK_FROM_ARROW(validity_builder.Append(valid));
- if (valid) {
- int64_t value_offset = (vector_array.offset() + i) * vector_length;
- for (int32_t j = 0; j < vector_length; ++j) {
-
PAIMON_RETURN_NOT_OK_FROM_ARROW(indices_builder.Append(value_offset + j));
- }
- offset += vector_length;
- }
- PAIMON_RETURN_NOT_OK_FROM_ARROW(offsets_builder.Append(offset));
- }
-
- std::shared_ptr<arrow::Array> offsets;
- std::shared_ptr<arrow::Array> indices;
- std::shared_ptr<arrow::Array> validity;
- PAIMON_RETURN_NOT_OK_FROM_ARROW(offsets_builder.Finish(&offsets));
- PAIMON_RETURN_NOT_OK_FROM_ARROW(indices_builder.Finish(&indices));
- PAIMON_RETURN_NOT_OK_FROM_ARROW(validity_builder.Finish(&validity));
-
- arrow::compute::ExecContext exec_context(pool);
- PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
- arrow::Datum values,
- arrow::compute::Take(arrow::Datum(vector_array.values()),
arrow::Datum(indices),
- arrow::compute::TakeOptions::NoBoundsCheck(),
&exec_context));
- return std::make_shared<arrow::ListArray>(
- write_type, vector_array.length(), offsets->data()->buffers[1],
values.make_array(),
- validity->data()->buffers[1], vector_array.null_count());
-}
-
-} // namespace
-
-std::shared_ptr<arrow::DataType> ParquetVectorConverter::GetWriteType(
- const std::shared_ptr<arrow::DataType>& logical_type) {
- switch (logical_type->id()) {
- case arrow::Type::FIXED_SIZE_LIST: {
- const auto& vector_type = checked_cast<const
arrow::FixedSizeListType&>(*logical_type);
- return arrow::list(
-
vector_type.value_field()->WithType(GetWriteType(vector_type.value_type())));
- }
- case arrow::Type::STRUCT: {
- arrow::FieldVector fields;
- fields.reserve(logical_type->num_fields());
- for (const auto& field : logical_type->fields()) {
- fields.push_back(field->WithType(GetWriteType(field->type())));
- }
- return arrow::struct_(fields);
- }
- case arrow::Type::LIST:
- return arrow::list(
-
logical_type->field(0)->WithType(GetWriteType(logical_type->field(0)->type())));
- case arrow::Type::MAP: {
- const auto& map_type = checked_cast<const
arrow::MapType&>(*logical_type);
- return std::make_shared<arrow::MapType>(
- map_type.value_field()->WithType(arrow::struct_(
-
{map_type.key_field()->WithType(GetWriteType(map_type.key_type())),
-
map_type.item_field()->WithType(GetWriteType(map_type.item_type()))})),
- map_type.keys_sorted());
- }
- default:
- return logical_type;
- }
-}
-
-Result<std::shared_ptr<arrow::Array>>
ParquetVectorConverter::ConvertToWriteType(
- const std::shared_ptr<arrow::Array>& array, arrow::MemoryPool* pool) {
- if (!VectorUtils::ContainsVectorType(array->type())) {
- return array;
- }
- std::shared_ptr<arrow::DataType> write_type = GetWriteType(array->type());
- switch (array->type_id()) {
- case arrow::Type::FIXED_SIZE_LIST: {
- PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorElements(*array));
- const auto& vector_array = checked_cast<const
arrow::FixedSizeListArray&>(*array);
- if (vector_array.null_count() == 0) {
- return CastToListType(array, write_type, pool);
- }
- return CompactNullVectorsToList(vector_array, write_type, pool);
- }
- case arrow::Type::STRUCT:
- case arrow::Type::LIST:
- case arrow::Type::MAP: {
- std::vector<std::shared_ptr<arrow::ArrayData>> children;
- children.reserve(array->data()->child_data.size());
- for (const auto& child_data : array->data()->child_data) {
- PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> child,
-
ConvertToWriteType(arrow::MakeArray(child_data), pool));
- children.push_back(child->data());
- }
- std::shared_ptr<arrow::ArrayData> data = array->data()->Copy();
- data->child_data = std::move(children);
- data->type = write_type;
- return arrow::MakeArray(data);
- }
- default:
- return array;
- }
-}
-
-} // namespace paimon::parquet
diff --git a/src/paimon/format/parquet/parquet_vector_converter.h
b/src/paimon/format/parquet/parquet_vector_converter.h
deleted file mode 100644
index a265e2d1..00000000
--- a/src/paimon/format/parquet/parquet_vector_converter.h
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#pragma once
-
-#include <memory>
-
-#include "arrow/memory_pool.h"
-#include "paimon/result.h"
-
-namespace arrow {
-class Array;
-class DataType;
-} // namespace arrow
-
-namespace paimon::parquet {
-
-/// Converts logical FixedSizeList VECTOR arrays to Parquet LIST arrays.
-class ParquetVectorConverter {
- public:
- ParquetVectorConverter() = delete;
- ~ParquetVectorConverter() = delete;
-
- static Result<std::shared_ptr<arrow::Array>> ConvertToWriteType(
- const std::shared_ptr<arrow::Array>& array, arrow::MemoryPool* pool);
-
- static std::shared_ptr<arrow::DataType> GetWriteType(
- const std::shared_ptr<arrow::DataType>& logical_type);
-};
-
-} // namespace paimon::parquet
diff --git a/src/paimon/format/parquet/parquet_vector_converter_test.cpp
b/src/paimon/format/parquet/parquet_vector_converter_test.cpp
deleted file mode 100644
index 6e1c0b0d..00000000
--- a/src/paimon/format/parquet/parquet_vector_converter_test.cpp
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#include "paimon/format/parquet/parquet_vector_converter.h"
-
-#include <memory>
-
-#include "arrow/api.h"
-#include "arrow/ipc/json_simple.h"
-#include "gtest/gtest.h"
-#include "paimon/common/utils/checked_cast.h"
-#include "paimon/testing/utils/testharness.h"
-
-namespace paimon::parquet::test {
-
-TEST(ParquetVectorConverterTest, ConvertNullableVectorToList) {
- auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
- auto vector_array = arrow::ipc::internal::json::ArrayFromJSON(
- vector_type, R"([[1.0, 2.0, 3.0], null, [4.0, 5.0,
6.0]])")
- .ValueOrDie();
-
- ASSERT_OK_AND_ASSIGN(
- std::shared_ptr<arrow::Array> converted,
- ParquetVectorConverter::ConvertToWriteType(vector_array,
arrow::default_memory_pool()));
- ASSERT_EQ(converted->type()->id(), arrow::Type::LIST);
- auto list_array = checked_pointer_cast<arrow::ListArray>(converted);
- ASSERT_EQ(list_array->value_length(0), 3);
- ASSERT_TRUE(list_array->IsNull(1));
- // The Parquet writer rejects a null LIST slot spanning values, so the
values Arrow keeps for
- // a null VECTOR row are dropped.
- ASSERT_EQ(list_array->value_length(1), 0);
- ASSERT_EQ(list_array->value_length(2), 3);
- ASSERT_EQ(list_array->values()->length(), 6);
- auto values =
checked_pointer_cast<arrow::FloatArray>(list_array->values());
- ASSERT_FLOAT_EQ(values->Value(3), 4.0f);
-}
-
-TEST(ParquetVectorConverterTest, ConvertNestedVectorsToList) {
- auto vector_type =
- arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 2);
- auto nested_type = arrow::struct_({
- arrow::field("vectors", arrow::list(vector_type)),
- arrow::field("by_name", arrow::map(arrow::utf8(), vector_type)),
- });
- auto nested_array =
- arrow::ipc::internal::json::ArrayFromJSON(nested_type,
- R"([[[[1.0, 2.0], null],
[["a", [3.0, 4.0]]]],
- [null, [["b", null]]]])")
- .ValueOrDie();
-
- ASSERT_OK_AND_ASSIGN(
- std::shared_ptr<arrow::Array> physical_array,
- ParquetVectorConverter::ConvertToWriteType(nested_array,
arrow::default_memory_pool()));
- auto physical_type =
checked_pointer_cast<arrow::StructType>(physical_array->type());
- auto physical_list =
checked_pointer_cast<arrow::ListType>(physical_type->field(0)->type());
- auto physical_map =
checked_pointer_cast<arrow::MapType>(physical_type->field(1)->type());
- ASSERT_EQ(physical_list->value_type()->id(), arrow::Type::LIST);
- ASSERT_EQ(physical_map->item_type()->id(), arrow::Type::LIST);
-}
-
-TEST(ParquetVectorConverterTest, ConvertSlicedVectorToList) {
- auto vector_type = arrow::fixed_size_list(arrow::float64(), 2);
- auto vector_array =
- arrow::ipc::internal::json::ArrayFromJSON(vector_type, R"([[1.0, 2.0],
[3.0, 4.0], null])")
- .ValueOrDie()
- ->Slice(1, 2);
-
- ASSERT_OK_AND_ASSIGN(
- std::shared_ptr<arrow::Array> converted,
- ParquetVectorConverter::ConvertToWriteType(vector_array,
arrow::default_memory_pool()));
- auto list_array = checked_pointer_cast<arrow::ListArray>(converted);
- ASSERT_EQ(list_array->length(), 2);
- ASSERT_EQ(list_array->value_length(0), 2);
- ASSERT_TRUE(list_array->IsNull(1));
- auto values =
checked_pointer_cast<arrow::DoubleArray>(list_array->values());
- ASSERT_DOUBLE_EQ(values->Value(0), 3.0);
- ASSERT_DOUBLE_EQ(values->Value(1), 4.0);
-}
-
-} // namespace paimon::parquet::test
diff --git a/src/paimon/format/parquet/parquet_vector_io_test.cpp
b/src/paimon/format/parquet/parquet_vector_io_test.cpp
index f45dad1e..ad60caef 100644
--- a/src/paimon/format/parquet/parquet_vector_io_test.cpp
+++ b/src/paimon/format/parquet/parquet_vector_io_test.cpp
@@ -258,6 +258,45 @@ TEST_F(ParquetVectorIoTest, WriteAndReadVector) {
R"([[1, [1.0, 2.0, 3.0]], [2, null], [3, [4.0, 5.0,
6.0]]])");
}
+TEST_F(ParquetVectorIoTest, WriteAndReadAllNullVector) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ auto struct_type = checked_pointer_cast<arrow::StructType>(arrow::struct_(
+ {arrow::field("id", arrow::int32()), arrow::field("embedding",
vector_type)}));
+ WriteAndCheck("all-null-vector-list.parquet", struct_type, struct_type,
+ R"([[1, null], [2, null], [3, null]])");
+}
+
+TEST_F(ParquetVectorIoTest, WriteAndReadAllNullFixedSizeListWithArrowSchema) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ auto logical_type =
checked_pointer_cast<arrow::StructType>(arrow::struct_({
+ arrow::field("id", arrow::int32()),
+ arrow::field("embedding", vector_type),
+ }));
+ const std::string json = R"([[1, null], [2, null], [3, null]])";
+ std::string file_path = dir_->Str() + "/all-null-vector.parquet";
+ WriteWithArrowWriter(file_path, logical_type, json);
+
+ 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(logical_type->fields()),
/*predicate=*/nullptr,
+ /*options=*/{}, /*batch_size=*/10, &reader);
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::ChunkedArray> actual,
+
paimon::test::ReadResultCollector::CollectResult(reader.get()));
+ arrow::Result<std::shared_ptr<arrow::Array>> expected_result =
+ arrow::ipc::internal::json::ArrayFromJSON(logical_type, json);
+ ASSERT_TRUE(expected_result.ok()) << expected_result.status().ToString();
+
ASSERT_TRUE(std::make_shared<arrow::ChunkedArray>(std::move(expected_result).ValueOrDie())
+ ->Equals(actual))
+ << actual->ToString();
+}
+
TEST_F(ParquetVectorIoTest, ReadOrdinaryParquetListAsVector) {
auto physical_type = checked_pointer_cast<arrow::StructType>(
arrow::struct_({arrow::field("id", arrow::int32()),
@@ -370,6 +409,13 @@ TEST_F(ParquetVectorIoTest, ReadNullableJavaFixture) {
{{{1.0f, 2.0f, 3.0f}}, std::nullopt, {{4.0f, 5.0f,
6.0f}}});
}
+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}}});
+}
+
// A table can hold files from several writers, and Paimon Java stores VECTOR
as Parquet LIST
// while Paimon Rust stores it as FixedSizeList. Reading both with the table
schema must produce
// batches of one Arrow type, otherwise they cannot be combined into a single
result.
@@ -386,7 +432,7 @@ TEST_F(ParquetVectorIoTest,
ReadMixedListAndFixedSizeListFixtures) {
// the whole result has been consumed.
std::vector<std::unique_ptr<FileBatchReader>> readers;
arrow::ArrayVector chunks;
- for (const char* file_name : {"java_vector_nullable.parquet",
"rust_vector.parquet"}) {
+ for (const char* file_name : {"java_vector_nullable.parquet",
"rust_vector_nullable.parquet"}) {
std::string file_path =
paimon::test::GetDataDir() + "/parquet/vector_compatibility/" +
file_name;
std::unique_ptr<FileBatchReader> reader;
@@ -406,7 +452,7 @@ TEST_F(ParquetVectorIoTest,
ReadMixedListAndFixedSizeListFixtures) {
arrow::Result<std::shared_ptr<arrow::Array>> expected_result =
arrow::ipc::internal::json::ArrayFromJSON(
logical_type, R"([[1, [1.0, 2.0, 3.0]], [2, null], [3, [4.0, 5.0,
6.0]],
- [1, [1.0, 2.0, 3.0]], [2, [7.0, 8.0, 9.0]], [3,
[4.0, 5.0, 6.0]]])");
+ [1, [1.0, 2.0, 3.0]], [2, null], [3, [4.0, 5.0,
6.0]]])");
ASSERT_TRUE(expected_result.ok()) << expected_result.status().ToString();
std::shared_ptr<arrow::ChunkedArray> merged =
std::move(merged_result).ValueOrDie();
ASSERT_TRUE(std::make_shared<arrow::ChunkedArray>(std::move(expected_result).ValueOrDie())
@@ -414,26 +460,4 @@ TEST_F(ParquetVectorIoTest,
ReadMixedListAndFixedSizeListFixtures) {
<< merged->ToString();
}
-// 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");
-}
-
} // namespace paimon::parquet::test
diff --git a/test/test_data/parquet/vector_compatibility/README.md
b/test/test_data/parquet/vector_compatibility/README.md
index 15eb2ef3..8a2fe25c 100644
--- a/test/test_data/parquet/vector_compatibility/README.md
+++ b/test/test_data/parquet/vector_compatibility/README.md
@@ -22,11 +22,9 @@ VECTOR columns, with and without null vectors.
`(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.
+`fixed_size_list`. The bundled Arrow 17 patch backports the Arrow community
fix that pads the
+decoded child array for null fixed-size-list slots, so
`rust_vector_nullable.parquet` is readable
+as a nullable VECTOR.
SHA-256 checksums:
