Skip to content

feat(vector): support VECTOR as a non-key value column in Parquet-backed primary-key tables. #275

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ChaomingZhangCN wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(vector): support VECTOR as a non-key value column in Parquet-backed primary-key tables. #275

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 6 additions & 5 deletions docs/source/user_guide/data_types.rst
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,12 +194,13 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
``SMALLINT``, ``INT``, ``BIGINT``, ``FLOAT``, or ``DOUBLE``. A VECTOR
value may be NULL, but its elements cannot be NULL.

Paimon C++ currently supports VECTOR columns only in append-only tables
backed by Parquet data files. They use the standard Parquet LIST
Paimon C++ supports VECTOR value columns in append-only and primary-key
tables backed by Parquet data files. They use the standard Parquet LIST
representation on disk and are restored as Arrow ``FixedSizeList``
values on read. Primary-key tables and data-evolution tables containing
VECTOR fields are rejected. VECTOR columns also cannot be partition or
bucket keys. Dedicated vector storage is not included yet.
values on read. VECTOR is not supported in data-evolution tables. VECTOR
columns cannot be primary, partition, or bucket keys, nor comparator-based
ordering fields such as sequence and sequence-group fields. Dedicated
vector storage is not included yet.

Paimon C++ also reads Parquet files written by Paimon Rust or Python whose
embedded Arrow schema restores VECTOR columns as ``FixedSizeList``,
Expand Down
8 changes: 8 additions & 0 deletions src/paimon/common/data/columnar/columnar_array.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,14 @@ Timestamp ColumnarArray::GetTimestamp(int32_t pos, int32_t precision) const {
}

std::shared_ptr<InternalArray> ColumnarArray::GetArray(int32_t pos) const {
if (array_->type_id() == arrow::Type::FIXED_SIZE_LIST) {
auto fixed_size_list_array = checked_cast<const arrow::FixedSizeListArray*>(array_);
auto fixed_size_list_type =
checked_pointer_cast<arrow::FixedSizeListType>(fixed_size_list_array->type());
int32_t offset = static_cast<int32_t>(fixed_size_list_array->value_offset(offset_ + pos));
return std::make_shared<ColumnarArray>(fixed_size_list_array->values().get(), pool_, offset,
fixed_size_list_type->list_size());
}
auto list_array = checked_cast<const arrow::ListArray*>(array_);
int32_t offset = list_array->value_offset(offset_ + pos);
int32_t length = list_array->value_length(offset_ + pos);
Expand Down
9 changes: 9 additions & 0 deletions src/paimon/common/data/columnar/columnar_row.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,15 @@ std::shared_ptr<InternalRow> ColumnarRow::GetRow(int32_t pos, int32_t num_fields
}

std::shared_ptr<InternalArray> ColumnarRow::GetArray(int32_t pos) const {
if (array_vec_[pos]->type_id() == arrow::Type::FIXED_SIZE_LIST) {
auto fixed_size_list_array =
checked_cast<const arrow::FixedSizeListArray*>(array_vec_[pos]);
auto fixed_size_list_type =
checked_pointer_cast<arrow::FixedSizeListType>(fixed_size_list_array->type());
int32_t offset = static_cast<int32_t>(fixed_size_list_array->value_offset(row_id_));
return std::make_shared<ColumnarArray>(fixed_size_list_array->values().get(), pool_, offset,
fixed_size_list_type->list_size());
}
auto list_array = checked_cast<const arrow::ListArray*>(array_vec_[pos]);
int32_t offset = list_array->value_offset(row_id_);
int32_t length = list_array->value_length(row_id_);
Expand Down
9 changes: 9 additions & 0 deletions src/paimon/common/data/columnar/columnar_row_ref.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,15 @@ std::shared_ptr<InternalRow> ColumnarRowRef::GetRow(int32_t pos, int32_t num_fie
}

std::shared_ptr<InternalArray> ColumnarRowRef::GetArray(int32_t pos) const {
if (ctx_->array_vec[pos]->type_id() == arrow::Type::FIXED_SIZE_LIST) {
auto fixed_size_list_array =
checked_cast<const arrow::FixedSizeListArray*>(ctx_->array_vec[pos].get());
auto fixed_size_list_type =
checked_pointer_cast<arrow::FixedSizeListType>(fixed_size_list_array->type());
int32_t offset = static_cast<int32_t>(fixed_size_list_array->value_offset(row_id_));
return std::make_shared<ColumnarArray>(fixed_size_list_array->values().get(), ctx_->pool,
offset, fixed_size_list_type->list_size());
}
auto list_array = checked_cast<const arrow::ListArray*>(ctx_->array_vec[pos].get());
int32_t offset = list_array->value_offset(row_id_);
int32_t length = list_array->value_length(row_id_);
Expand Down
3 changes: 2 additions & 1 deletion src/paimon/common/data/internal_row.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,8 @@ Result<InternalRow::FieldGetterFunc> InternalRow::CreateFieldGetter(
};
break;
}
case arrow::Type::type::LIST: {
case arrow::Type::type::LIST:
case arrow::Type::type::FIXED_SIZE_LIST: {
field_getter = [field_idx](const InternalRow& row) -> VariantType {
return row.GetArray(field_idx);
};
Expand Down
6 changes: 3 additions & 3 deletions src/paimon/common/data/serializer/binary_serializer_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,7 @@ Result<std::shared_ptr<BinaryArray>> BinarySerializerUtils::WriteBinaryArray(
return binary_array;
}
auto binary_array = std::make_shared<BinaryArray>();
auto list_type = checked_pointer_cast<arrow::ListType>(type);
auto value_type = list_type->value_type();
auto value_type = type->field(0)->type();
// TODO(xinyu.lxy): reuse BinaryWriter
BinaryArrayWriter binary_writer(binary_array.get(), value->Size(),
BinaryArrayWriter::GetElementSize(value_type->id()), pool);
Expand Down Expand Up @@ -183,7 +182,8 @@ Status BinarySerializerUtils::WriteBinaryData(const std::shared_ptr<arrow::DataT
}
break;
}
case arrow::Type::type::LIST: {
case arrow::Type::type::LIST:
case arrow::Type::type::FIXED_SIZE_LIST: {
auto internal_array = getter->GetArray(pos);
PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<BinaryArray> binary_array,
WriteBinaryArray(internal_array, type, pool));
Expand Down
6 changes: 4 additions & 2 deletions src/paimon/common/data/serializer/row_compacted_serializer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,7 +285,8 @@ Result<RowCompactedSerializer::FieldReader> RowCompactedSerializer::CreateFieldR
};
break;
}
case arrow::Type::type::LIST: {
case arrow::Type::type::LIST:
case arrow::Type::type::FIXED_SIZE_LIST: {
field_reader = [](int32_t pos, RowReader* reader) -> Result<VariantType> {
PAIMON_ASSIGN_OR_RAISE(VariantType value, reader->ReadArray());
return value;
Expand Down Expand Up @@ -414,7 +415,8 @@ Result<RowCompactedSerializer::FieldWriter> RowCompactedSerializer::CreateFieldW
};
break;
}
case arrow::Type::type::LIST: {
case arrow::Type::type::LIST:
case arrow::Type::type::FIXED_SIZE_LIST: {
field_writer = [field_type](int32_t pos, const VariantType& field,
RowWriter* writer) -> Status {
return writer->WriteArray(
Expand Down
30 changes: 25 additions & 5 deletions src/paimon/core/io/key_value_in_memory_record_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,22 +111,42 @@ void KeyValueInMemoryRecordReader::Close() {
Result<std::shared_ptr<arrow::NumericArray<arrow::UInt64Type>>>
KeyValueInMemoryRecordReader::SortBatch() const {
std::vector<arrow::compute::SortKey> sort_keys;
arrow::FieldVector sort_fields;
arrow::ArrayVector sort_columns;
sort_keys.reserve(primary_keys_.size() + user_defined_sequence_fields_.size());
sort_fields.reserve(primary_keys_.size() + user_defined_sequence_fields_.size());
sort_columns.reserve(primary_keys_.size() + user_defined_sequence_fields_.size());
const arrow::StructType* value_type = value_struct_array_->struct_type();
auto append_sort_key = [&](const std::string& name, arrow::compute::SortOrder order) -> Status {
int32_t field_index = value_type->GetFieldIndex(name);
if (field_index < 0) {
return Status::Invalid(fmt::format("cannot find field {} in data batch", name));
}
sort_keys.emplace_back(name, order);
sort_fields.push_back(value_type->field(field_index));
sort_columns.push_back(value_struct_array_->field(field_index));
return Status::OK();
};
for (const auto& name : primary_keys_) {
sort_keys.emplace_back(name, arrow::compute::SortOrder::Ascending);
PAIMON_RETURN_NOT_OK(append_sort_key(name, arrow::compute::SortOrder::Ascending));
}
const auto sequence_sort_order = sequence_fields_ascending_
? arrow::compute::SortOrder::Ascending
: arrow::compute::SortOrder::Descending;
for (const auto& name : user_defined_sequence_fields_) {
sort_keys.emplace_back(name, sequence_sort_order);
PAIMON_RETURN_NOT_OK(append_sort_key(name, sequence_sort_order));
}
auto sort_options =
arrow::compute::SortOptions(sort_keys, arrow::compute::NullPlacement::AtStart);
arrow::compute::ExecContext exec_context(arrow_pool_.get());
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array> sorted_indices,
arrow::compute::SortIndices(arrow::Datum(value_struct_array_),
sort_options, &exec_context));
// Arrow's StructArray sorting path may inspect value columns outside the sort keys. Restrict
// the batch to the requested fields so non-sortable values, such as VECTOR, are never compared.
std::shared_ptr<arrow::RecordBatch> sort_batch =

@lxy-9602 lxy-9602 Sep 3, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I’d like to better understand what issues inspect value columns can cause here. Does it lead to incorrect sorting, crashes, or explicit errors? Also, is this considered a bug in Arrow, and is there any ongoing fix for it?

arrow::RecordBatch::Make(arrow::schema(std::move(sort_fields)),
value_struct_array_->length(), std::move(sort_columns));
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
std::shared_ptr<arrow::Array> sorted_indices,
arrow::compute::SortIndices(arrow::Datum(sort_batch), sort_options, &exec_context));
if (!sorted_indices || sorted_indices->type_id() != arrow::Type::UINT64) {
return Status::Invalid("cannot cast sorted indices to UInt64Array");
}
Expand Down
36 changes: 36 additions & 0 deletions src/paimon/core/io/row_to_arrow_array_converter.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,12 @@ Status RowToArrowArrayConverter<T, R>::Reserve(arrow::ArrayBuilder* array_builde
PAIMON_RETURN_NOT_OK(Reserve(list_builder->value_builder(), idx));
break;
}
case arrow::Type::type::FIXED_SIZE_LIST: {
PAIMON_ASSIGN_OR_RAISE(auto* list_builder,
CastToTypedBuilder<arrow::FixedSizeListBuilder>(array_builder));
PAIMON_RETURN_NOT_OK(Reserve(list_builder->value_builder(), idx));
break;
}
case arrow::Type::type::MAP: {
PAIMON_ASSIGN_OR_RAISE(auto* map_builder,
CastToTypedBuilder<arrow::MapBuilder>(array_builder));
Expand Down Expand Up @@ -224,6 +230,11 @@ Status RowToArrowArrayConverter<T, R>::Accumulate(const arrow::Array* array, int
PAIMON_RETURN_NOT_OK(Accumulate(list_array->values().get(), idx));
break;
}
case arrow::Type::type::FIXED_SIZE_LIST: {
auto list_array = checked_cast<const arrow::FixedSizeListArray*>(array);
PAIMON_RETURN_NOT_OK(Accumulate(list_array->values().get(), idx));
break;
}
case arrow::Type::type::MAP: {
auto map_array = checked_cast<const arrow::MapArray*>(array);
PAIMON_RETURN_NOT_OK(Accumulate(map_array->keys().get(), idx));
Expand Down Expand Up @@ -432,6 +443,31 @@ RowToArrowArrayConverter<T, R>::AppendField(bool use_view, arrow::ArrayBuilder*
return arrow::Status::OK();
});
}
case arrow::Type::type::FIXED_SIZE_LIST: {
PAIMON_ASSIGN_OR_RAISE(auto* list_builder,
CastToTypedBuilder<arrow::FixedSizeListBuilder>(array_builder));
std::shared_ptr<arrow::FixedSizeListType> list_type =
checked_pointer_cast<arrow::FixedSizeListType>(list_builder->type());
int32_t list_size = list_type->list_size();
PAIMON_ASSIGN_OR_RAISE(AppendValueFunc value_func,
(RowToArrowArrayConverter<T, R>::AppendField(
use_view, list_builder->value_builder(), reserve_count)));
return RowToArrowArrayConverter<T, R>::AppendValueFunc(
[list_builder, list_size, value_func](const DataGetters& data_getter,
int32_t pos) -> arrow::Status {
CHECK_AND_APPEND_NULL(data_getter, list_builder, pos);
std::shared_ptr<InternalArray> sub_array = data_getter.GetArray(pos);
if (!sub_array || sub_array->Size() != list_size) {
return arrow::Status::Invalid(
"VECTOR length does not match its declared dimension");
}
ARROW_RETURN_NOT_OK(list_builder->Append());
for (int32_t i = 0; i < list_size; ++i) {
ARROW_RETURN_NOT_OK(value_func(*sub_array, i));
}
return arrow::Status::OK();
});
}
case arrow::Type::type::MAP: {
PAIMON_ASSIGN_OR_RAISE(auto* map_builder,
CastToTypedBuilder<arrow::MapBuilder>(array_builder));
Expand Down
8 changes: 5 additions & 3 deletions src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,12 +73,12 @@ Result<bool> EqualRows(const std::shared_ptr<InternalRow>& lhs,

Result<bool> EqualArrays(const std::shared_ptr<InternalArray>& lhs,
const std::shared_ptr<InternalArray>& rhs,
const std::shared_ptr<arrow::ListType>& type) {
const std::shared_ptr<arrow::DataType>& type) {
if (!lhs || !rhs || lhs->Size() != rhs->Size()) {
return lhs == rhs;
}
for (int32_t i = 0; i < lhs->Size(); ++i) {
PAIMON_ASSIGN_OR_RAISE(bool equal, EqualGetters(*lhs, i, *rhs, i, type->value_type()));
PAIMON_ASSIGN_OR_RAISE(bool equal, EqualGetters(*lhs, i, *rhs, i, type->field(0)->type()));
if (!equal) {
return false;
}
Expand Down Expand Up @@ -174,6 +174,7 @@ Result<VariantType> FieldAggregateUtils::GetValue(const DataGetters& getters, in
getters.GetDecimal(pos, decimal_type->precision(), decimal_type->scale()));
}
case arrow::Type::LIST:
case arrow::Type::FIXED_SIZE_LIST:
return VariantType(getters.GetArray(pos));
case arrow::Type::MAP:
return VariantType(getters.GetMap(pos));
Expand Down Expand Up @@ -229,9 +230,10 @@ Result<bool> FieldAggregateUtils::Equals(const VariantType& lhs, const VariantTy
DataDefine::GetVariantValue<std::shared_ptr<InternalRow>>(rhs),
checked_pointer_cast<arrow::StructType>(type));
case arrow::Type::LIST:
case arrow::Type::FIXED_SIZE_LIST:
return EqualArrays(DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(lhs),
DataDefine::GetVariantValue<std::shared_ptr<InternalArray>>(rhs),
checked_pointer_cast<arrow::ListType>(type));
type);
case arrow::Type::MAP:
return EqualMaps(DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(lhs),
DataDefine::GetVariantValue<std::shared_ptr<InternalMap>>(rhs),
Expand Down
7 changes: 3 additions & 4 deletions src/paimon/core/mergetree/compact/internal_row_equalizer.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,11 +144,10 @@ class InternalRowEqualizer {
.CompareTo(rhs.GetDecimal(rhs_pos, precision, scale)) == 0;
});
}
case arrow::Type::LIST: {
std::shared_ptr<arrow::ListType> list_type =
checked_pointer_cast<arrow::ListType>(type);
case arrow::Type::LIST:
case arrow::Type::FIXED_SIZE_LIST: {
PAIMON_ASSIGN_OR_RAISE(ValueEqualizer element_equalizer,
CreateValueEqualizer(list_type->value_type()));
CreateValueEqualizer(type->field(0)->type()));
return ValueEqualizer([element_equalizer = std::move(element_equalizer)](
const DataGetters& lhs, int32_t lhs_pos,
const DataGetters& rhs, int32_t rhs_pos) {
Expand Down
5 changes: 5 additions & 0 deletions src/paimon/core/mergetree/in_memory_sort_buffer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,11 @@ Result<int64_t> InMemorySortBuffer::EstimateMemoryUse(const std::shared_ptr<arro
PAIMON_ASSIGN_OR_RAISE(int64_t value_mem, EstimateMemoryUse(list_array->values()));
return null_bits_size_in_bytes + value_mem;
}
case arrow::Type::type::FIXED_SIZE_LIST: {
auto list_array = checked_cast<const arrow::FixedSizeListArray*>(array.get());
PAIMON_ASSIGN_OR_RAISE(int64_t value_mem, EstimateMemoryUse(list_array->values()));
return null_bits_size_in_bytes + value_mem;
}
case arrow::Type::type::MAP: {
auto map_array = checked_cast<const arrow::MapArray*>(array.get());
PAIMON_ASSIGN_OR_RAISE(int64_t key_mem, EstimateMemoryUse(map_array->keys()));
Expand Down
21 changes: 20 additions & 1 deletion src/paimon/core/realtime/realtime_primary_key_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,17 +87,36 @@ Result<std::shared_ptr<arrow::StructArray>> CreateRealtimePrimaryKeyTransportBat
arrow::StructArray::Make(std::move(columns), transport_schema->fields()));

std::vector<arrow::compute::SortKey> sort_keys;
std::vector<std::string> sort_field_names;
sort_keys.reserve(trimmed_primary_keys.size() + 1);
sort_field_names.reserve(trimmed_primary_keys.size() + 1);
for (const std::string& key : trimmed_primary_keys) {
sort_keys.emplace_back(key, arrow::compute::SortOrder::Ascending);
sort_field_names.push_back(key);
}
sort_keys.emplace_back(SpecialFields::SequenceNumber().Name(),
arrow::compute::SortOrder::Ascending);
sort_field_names.push_back(SpecialFields::SequenceNumber().Name());
arrow::compute::ExecContext context(arrow_pool);
arrow::compute::SortOptions options(sort_keys, arrow::compute::NullPlacement::AtStart);
arrow::FieldVector sort_fields;
arrow::ArrayVector sort_columns;
sort_fields.reserve(sort_keys.size());
sort_columns.reserve(sort_keys.size());
const arrow::StructType* transport_type = transport->struct_type();
for (const std::string& name : sort_field_names) {
int32_t field_index = transport_type->GetFieldIndex(name);
if (field_index < 0) {
return Status::Invalid("PK sort field is missing from transport batch: ", name);
}
sort_fields.push_back(transport_type->field(field_index));
sort_columns.push_back(transport->field(field_index));
}
std::shared_ptr<arrow::RecordBatch> sort_batch = arrow::RecordBatch::Make(
arrow::schema(std::move(sort_fields)), transport->length(), std::move(sort_columns));
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
arrow::Datum indices,
arrow::compute::SortIndices(arrow::Datum(transport), options, &context));
arrow::compute::SortIndices(arrow::Datum(sort_batch), options, &context));
PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
arrow::Datum sorted,
arrow::compute::Take(arrow::Datum(transport), indices,
Expand Down
Loading