This is an automated email from the ASF dual-hosted git repository.
lxy-9602 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 a42670e0 feat(vector): support VECTOR as a non-key value column in
Parquet-backed primary-key tables. (#275)
a42670e0 is described below
commit a42670e066bba60100317cc0aa5a854b2273bf98
Author: 小明同学 <[email protected]>
AuthorDate: Mon Sep 7 21:06:58 2026 +0800
feat(vector): support VECTOR as a non-key value column in Parquet-backed
primary-key tables. (#275)
---
docs/source/user_guide/data_types.rst | 11 +-
src/paimon/common/data/columnar/columnar_array.cpp | 8 +
src/paimon/common/data/columnar/columnar_row.cpp | 9 +
.../common/data/columnar/columnar_row_ref.cpp | 9 +
src/paimon/common/data/internal_row.cpp | 3 +-
.../data/serializer/binary_serializer_utils.cpp | 6 +-
.../data/serializer/row_compacted_serializer.cpp | 6 +-
.../io/key_value_in_memory_record_reader_test.cpp | 63 +++++++
src/paimon/core/io/row_to_arrow_array_converter.h | 36 ++++
.../compact/aggregate/field_aggregate_utils.cpp | 8 +-
.../mergetree/compact/internal_row_equalizer.h | 7 +-
.../core/mergetree/in_memory_sort_buffer.cpp | 5 +
src/paimon/core/schema/schema_validation.cpp | 17 +-
src/paimon/core/schema/schema_validation_test.cpp | 57 ++++++-
test/inte/realtime_write_inte_test.cpp | 29 ++++
test/inte/write_and_read_inte_test.cpp | 184 +++++++++++++++++++++
test/inte/write_inte_test.cpp | 76 +++++++++
17 files changed, 504 insertions(+), 30 deletions(-)
diff --git a/docs/source/user_guide/data_types.rst
b/docs/source/user_guide/data_types.rst
index add537ad..57f0eafd 100644
--- a/docs/source/user_guide/data_types.rst
+++ b/docs/source/user_guide/data_types.rst
@@ -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``,
diff --git a/src/paimon/common/data/columnar/columnar_array.cpp
b/src/paimon/common/data/columnar/columnar_array.cpp
index 6566aea9..11a80dfc 100644
--- a/src/paimon/common/data/columnar/columnar_array.cpp
+++ b/src/paimon/common/data/columnar/columnar_array.cpp
@@ -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());
+ auto 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);
diff --git a/src/paimon/common/data/columnar/columnar_row.cpp
b/src/paimon/common/data/columnar/columnar_row.cpp
index 5099b42b..db732d90 100644
--- a/src/paimon/common/data/columnar/columnar_row.cpp
+++ b/src/paimon/common/data/columnar/columnar_row.cpp
@@ -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());
+ auto 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_);
diff --git a/src/paimon/common/data/columnar/columnar_row_ref.cpp
b/src/paimon/common/data/columnar/columnar_row_ref.cpp
index d73e6b77..8ee635de 100644
--- a/src/paimon/common/data/columnar/columnar_row_ref.cpp
+++ b/src/paimon/common/data/columnar/columnar_row_ref.cpp
@@ -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());
+ auto 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_);
diff --git a/src/paimon/common/data/internal_row.cpp
b/src/paimon/common/data/internal_row.cpp
index 24b6b5be..b7c011b2 100644
--- a/src/paimon/common/data/internal_row.cpp
+++ b/src/paimon/common/data/internal_row.cpp
@@ -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);
};
diff --git a/src/paimon/common/data/serializer/binary_serializer_utils.cpp
b/src/paimon/common/data/serializer/binary_serializer_utils.cpp
index c0248cd6..d72a2b0b 100644
--- a/src/paimon/common/data/serializer/binary_serializer_utils.cpp
+++ b/src/paimon/common/data/serializer/binary_serializer_utils.cpp
@@ -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);
@@ -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));
diff --git a/src/paimon/common/data/serializer/row_compacted_serializer.cpp
b/src/paimon/common/data/serializer/row_compacted_serializer.cpp
index 411dae68..1fd618a3 100644
--- a/src/paimon/common/data/serializer/row_compacted_serializer.cpp
+++ b/src/paimon/common/data/serializer/row_compacted_serializer.cpp
@@ -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;
@@ -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(
diff --git a/src/paimon/core/io/key_value_in_memory_record_reader_test.cpp
b/src/paimon/core/io/key_value_in_memory_record_reader_test.cpp
index 7604064e..ca6f331e 100644
--- a/src/paimon/core/io/key_value_in_memory_record_reader_test.cpp
+++ b/src/paimon/core/io/key_value_in_memory_record_reader_test.cpp
@@ -26,6 +26,7 @@
#include "arrow/array/array_nested.h"
#include "arrow/ipc/json_simple.h"
#include "gtest/gtest.h"
+#include "paimon/common/data/internal_array.h"
#include "paimon/common/types/data_field.h"
#include "paimon/common/types/row_kind.h"
#include "paimon/common/utils/fields_comparator.h"
@@ -349,6 +350,68 @@ TEST_F(KeyValueInMemoryRecordReaderTest,
TestStableSortWithDuplicateKeys) {
ASSERT_FALSE(eof_iter);
}
+TEST_F(KeyValueInMemoryRecordReaderTest, TestSortWithVectorValues) {
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ auto src_type = arrow::struct_(
+ {arrow::field("pk", arrow::int32()), arrow::field("seq",
arrow::int32()),
+ arrow::field("embedding", vector_type),
+ arrow::field("payload", arrow::struct_({arrow::field("embedding",
vector_type)}))});
+ auto src_array = std::dynamic_pointer_cast<arrow::StructArray>(
+ arrow::ipc::internal::json::ArrayFromJSON(src_type, R"([
+ [2, 20, [0, 1, 2], [[0, 1, 2]]],
+ [1, 20, null, null],
+ [2, 10, [2, 3, 4], [null]],
+ [1, null, [3, 4, 5], [[3, 4, 5]]],
+ [2, 20, [4, 5, 6], [[4, 5, 6]]]
+ ])")
+ .ValueOrDie());
+ ASSERT_OK_AND_ASSIGN(
+ std::shared_ptr<FieldsComparator> key_comparator,
+ FieldsComparator::Create({DataField(0, arrow::field("pk",
arrow::int32()))},
+ /*is_ascending_order=*/true));
+
+ auto check_sort = [&](const std::vector<std::string>& sequence_fields,
bool ascending,
+ const std::vector<int64_t>& expected_indices) {
+ KeyValueInMemoryRecordReader reader(
+ /*last_sequence_num=*/0, src_array,
std::vector<RecordBatch::RowKind>{},
+ std::vector<std::string>{"pk"}, sequence_fields, ascending,
key_comparator, pool_);
+ ASSERT_OK_AND_ASSIGN(
+ std::vector<KeyValue> results,
+
(ReadResultCollector::CollectKeyValueResult<KeyValueInMemoryRecordReader,
+
KeyValueRecordReader::Iterator>(&reader)));
+ ASSERT_EQ(expected_indices.size(), results.size());
+ for (size_t i = 0; i < results.size(); ++i) {
+ const int64_t source_index = expected_indices[i];
+ SCOPED_TRACE(source_index);
+ ASSERT_EQ(source_index, results[i].sequence_number);
+ const auto& value = results[i].value;
+ ASSERT_EQ(source_index == 1 || source_index == 3 ? 1 : 2,
results[i].key->GetInt(0));
+ ASSERT_EQ(source_index == 1, value->IsNullAt(2));
+ ASSERT_EQ(source_index == 1, value->IsNullAt(3));
+ if (source_index == 1) {
+ continue;
+ }
+ const std::vector<float> expected_vector =
{static_cast<float>(source_index),
+
static_cast<float>(source_index + 1),
+
static_cast<float>(source_index + 2)};
+ ASSERT_OK_AND_ASSIGN(std::vector<float> vector,
value->GetArray(2)->ToFloatArray());
+ ASSERT_EQ(expected_vector, vector);
+ auto payload = value->GetRow(3, 1);
+ ASSERT_EQ(source_index == 2, payload->IsNullAt(0));
+ if (source_index != 2) {
+ ASSERT_OK_AND_ASSIGN(std::vector<float> nested_vector,
+ payload->GetArray(0)->ToFloatArray());
+ ASSERT_EQ(expected_vector, nested_vector);
+ }
+ }
+ };
+ // Equal sort keys retain input order regardless of the non-sortable value
columns.
+ check_sort({}, /*ascending=*/true, {1, 3, 0, 2, 4});
+ check_sort({"seq"}, /*ascending=*/true, {3, 1, 2, 0, 4});
+ check_sort({"seq"}, /*ascending=*/false, {3, 1, 0, 4, 2});
+}
+
TEST_F(KeyValueInMemoryRecordReaderTest, TestVariantType) {
// test null, repeated key, sequence fields, out of order in variant data
type
// precondition: fields[0] is key fields, fields[1] is sequence field,
src_array are like:
diff --git a/src/paimon/core/io/row_to_arrow_array_converter.h
b/src/paimon/core/io/row_to_arrow_array_converter.h
index 6caa9f61..336d9857 100644
--- a/src/paimon/core/io/row_to_arrow_array_converter.h
+++ b/src/paimon/core/io/row_to_arrow_array_converter.h
@@ -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));
@@ -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));
@@ -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));
diff --git
a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
index 9397ca4e..22e02f10 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
+++ b/src/paimon/core/mergetree/compact/aggregate/field_aggregate_utils.cpp
@@ -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;
}
@@ -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));
@@ -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),
diff --git a/src/paimon/core/mergetree/compact/internal_row_equalizer.h
b/src/paimon/core/mergetree/compact/internal_row_equalizer.h
index 22f15d8b..a0a0b4c5 100644
--- a/src/paimon/core/mergetree/compact/internal_row_equalizer.h
+++ b/src/paimon/core/mergetree/compact/internal_row_equalizer.h
@@ -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) {
diff --git a/src/paimon/core/mergetree/in_memory_sort_buffer.cpp
b/src/paimon/core/mergetree/in_memory_sort_buffer.cpp
index 5f53d598..272d34b9 100644
--- a/src/paimon/core/mergetree/in_memory_sort_buffer.cpp
+++ b/src/paimon/core/mergetree/in_memory_sort_buffer.cpp
@@ -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()));
diff --git a/src/paimon/core/schema/schema_validation.cpp
b/src/paimon/core/schema/schema_validation.cpp
index 5e524a37..26d4100c 100644
--- a/src/paimon/core/schema/schema_validation.cpp
+++ b/src/paimon/core/schema/schema_validation.cpp
@@ -127,6 +127,16 @@ Status ValidatePerLevelOption(
return Status::OK();
}
+Status ValidateVectorComparatorField(const TableSchema& schema, const
std::string& field_name,
+ const std::string& role) {
+ PAIMON_ASSIGN_OR_RAISE(DataField field, schema.GetField(field_name));
+ if (VectorUtils::ContainsVectorField(field.ArrowField())) {
+ return Status::Invalid(
+ fmt::format("VECTOR field '{}' cannot be used as {}.", field_name,
role));
+ }
+ return Status::OK();
+}
+
} // namespace
bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>&
field) {
@@ -378,6 +388,8 @@ Status SchemaValidation::ValidateSequenceGroup(const
TableSchema& schema,
fmt::format("The sequence field group: {} can not be found
in table schema.",
sequence_field_name));
}
+ PAIMON_RETURN_NOT_OK(ValidateVectorComparatorField(schema,
sequence_field_name,
+ "a
sequence-group ordering field"));
}
for (const auto& field : StringUtils::Split(v,
Options::FIELDS_SEPARATOR)) {
@@ -445,6 +457,7 @@ Status SchemaValidation::ValidateSequenceField(const
TableSchema& schema,
PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
std::find(field_names.begin(), field_names.end(), field) !=
field_names.end(),
fmt::format("Sequence field: '{}' cannot be found in table
schema.", field)));
+ PAIMON_RETURN_NOT_OK(ValidateVectorComparatorField(schema, field,
"a sequence field"));
PAIMON_ASSIGN_OR_RAISE(std::optional<std::string> agg_func,
options.GetFieldAggFunc(field));
@@ -776,10 +789,6 @@ Status SchemaValidation::ValidateVectorFields(const
TableSchema& schema,
if (!has_vector) {
return Status::OK();
}
- if (!schema.PrimaryKeys().empty()) {
- return Status::NotImplemented(
- "VECTOR fields in primary-key tables are not implemented yet.");
- }
if (options.DataEvolutionEnabled()) {
return Status::NotImplemented(
"VECTOR fields in data-evolution tables are not implemented yet.");
diff --git a/src/paimon/core/schema/schema_validation_test.cpp
b/src/paimon/core/schema/schema_validation_test.cpp
index 3e708f87..f34b25fa 100644
--- a/src/paimon/core/schema/schema_validation_test.cpp
+++ b/src/paimon/core/schema/schema_validation_test.cpp
@@ -87,12 +87,54 @@ TEST(SchemaValidationTest, TestVectorType) {
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"in primary key field embedding is unsupported");
+ ASSERT_OK_AND_ASSIGN(
+ table_schema, TableSchema::Create(/*schema_id=*/0, schema,
/*partition_keys=*/{"embedding"},
+ /*primary_keys=*/{},
parquet_options));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+ "in partition field embedding is unsupported");
+
+ std::map<std::string, std::string> bucket_key_options = {
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "embedding"},
+ {Options::FILE_FORMAT, "parquet"},
+ };
+ ASSERT_OK_AND_ASSIGN(table_schema,
+ TableSchema::Create(/*schema_id=*/0, schema,
/*partition_keys=*/{},
+ /*primary_keys=*/{},
bucket_key_options));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+ "Nested type cannot be in bucket-key");
+
+ std::map<std::string, std::string> sequence_field_options = {
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "id"},
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::SEQUENCE_FIELD, "embedding"},
+ };
+ ASSERT_OK_AND_ASSIGN(table_schema,
+ TableSchema::Create(/*schema_id=*/0, schema,
/*partition_keys=*/{},
+ /*primary_keys=*/{"id"},
sequence_field_options));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+ "VECTOR field 'embedding' cannot be used as a sequence
field.");
+
+ std::map<std::string, std::string> sequence_group_options = {
+ {Options::BUCKET, "1"},
+ {Options::BUCKET_KEY, "id"},
+ {Options::FILE_FORMAT, "parquet"},
+ {Options::MERGE_ENGINE, "partial-update"},
+ {"fields.embedding.sequence-group", "id"},
+ };
+ ASSERT_OK_AND_ASSIGN(table_schema,
+ TableSchema::Create(/*schema_id=*/0, schema,
/*partition_keys=*/{},
+ /*primary_keys=*/{"id"},
sequence_group_options));
+ ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+ "VECTOR field 'embedding' cannot be used as a
sequence-group ordering "
+ "field.");
+
primary_key_options[Options::FILE_FORMAT] = "parquet";
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema,
/*partition_keys=*/{},
/*primary_keys=*/{"id"},
primary_key_options));
- ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
- "VECTOR fields in primary-key tables are not
implemented yet.");
+ ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
auto nested_schema = arrow::schema({
arrow::field("id", arrow::int64()),
@@ -102,8 +144,7 @@ TEST(SchemaValidationTest, TestVectorType) {
table_schema,
TableSchema::Create(/*schema_id=*/0, nested_schema,
/*partition_keys=*/{}, /*primary_keys=*/{"id"},
primary_key_options));
- ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
- "VECTOR fields in primary-key tables are not
implemented yet.");
+ ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
std::map<std::string, std::string> data_evolution_options = {
{Options::BUCKET, "-1"},
@@ -117,10 +158,10 @@ TEST(SchemaValidationTest, TestVectorType) {
/*primary_keys=*/{},
data_evolution_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"VECTOR fields in data-evolution tables are not
implemented yet.");
- ASSERT_OK_AND_ASSIGN(table_schema,
- TableSchema::Create(/*schema_id=*/0, nested_schema,
- /*partition_keys=*/{},
- /*primary_keys=*/{},
data_evolution_options));
+ ASSERT_OK_AND_ASSIGN(
+ table_schema,
+ TableSchema::Create(/*schema_id=*/0, nested_schema,
+ /*partition_keys=*/{}, /*primary_keys=*/{},
data_evolution_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"VECTOR fields in data-evolution tables are not
implemented yet.");
}
diff --git a/test/inte/realtime_write_inte_test.cpp
b/test/inte/realtime_write_inte_test.cpp
index 95adf74f..03e55c24 100644
--- a/test/inte/realtime_write_inte_test.cpp
+++ b/test/inte/realtime_write_inte_test.cpp
@@ -1164,6 +1164,35 @@ TEST_F(RealtimeWriteInteTest, TestPkRead) {
read_array.reset();
}
+TEST_F(RealtimeWriteInteTest, TestPkVector) {
+ options_[Options::FILE_FORMAT] = "parquet";
+ std::shared_ptr<arrow::DataType> vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ fields_ = {arrow::field("id", arrow::int64()), arrow::field("embedding",
vector_type)};
+ schema_ = arrow::schema(fields_);
+ CreatePkTable();
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
+ RealtimeContext::Create());
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<FileStoreWrite> writer,
+ CreateRealtimeWriter(realtime_context));
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> batch,
MakeUnpartitionedBatchFromJson(R"([
+ [0, 2, [2.0, 2.0, 2.0]],
+ [1, 3, null],
+ [2, 1, [1.0, 1.0, 1.0]],
+ [3, 2, [3.0, 3.0, 3.0]]
+ ])"));
+ ASSERT_OK(writer->Write(std::move(batch)));
+
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<Plan> plan,
+ CreatePlan(realtime_context, /*predicate=*/nullptr));
+ ReadPlanWithSchemaAndCheck(plan, realtime_context, schema_,
+ R"([[0, 1, [1.0, 1.0, 1.0]],
+ [0, 2, [3.0, 3.0, 3.0]],
+ [0, 3, null]])");
+ ASSERT_OK(writer->Close());
+}
+
TEST_F(RealtimeWriteInteTest, TestPkRealtimeReadOptimizedScanUnsupported) {
CreatePkTable();
ASSERT_OK_AND_ASSIGN(std::shared_ptr<RealtimeContext> realtime_context,
diff --git a/test/inte/write_and_read_inte_test.cpp
b/test/inte/write_and_read_inte_test.cpp
index c5d9b9d2..7ce6f2e6 100644
--- a/test/inte/write_and_read_inte_test.cpp
+++ b/test/inte/write_and_read_inte_test.cpp
@@ -69,6 +69,7 @@
#include "rapidjson/writer.h"
namespace paimon::test {
+
// This is a sdk end-to-end test demo that supports write, commit, scan, and
read operations.
class WriteAndReadInteTest
: public ::testing::Test,
@@ -723,6 +724,189 @@ TEST_P(WriteAndReadInteTest,
TestPKListAggPreservesResultsAcrossKeys) {
ASSERT_TRUE(success);
}
+TEST_P(WriteAndReadInteTest, TestPKVector) {
+ auto [file_format, file_system] = GetParam();
+ if (file_format != "parquet") {
+ return;
+ }
+
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::int64()),
+ arrow::field("embedding", vector_type),
+ };
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::TARGET_FILE_SIZE, "1024"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(
+ auto helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options,
/*is_streaming_mode=*/true));
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> initial_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([[1, [1.0, 2.0,
3.0]], [2, null]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(initial_batch),
/*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> update_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
+ R"([[1, [4.0, 5.0, 6.0]], [3, [7.0, 8.0,
9.0]]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(update_batch),
/*commit_identifier=*/1,
+
/*expected_commit_messages=*/std::nullopt));
+
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ ASSERT_OK(CompactAndCommit(table_path, options, /*commit_identifier=*/2));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
+ helper->NewScan(StartupMode::LatestFull(),
/*snapshot_id=*/std::nullopt));
+ arrow::FieldVector result_fields = fields;
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND",
arrow::int8()));
+ ASSERT_OK_AND_ASSIGN(
+ bool success, helper->ReadAndCheckResult(
+ arrow::struct_(result_fields), data_splits,
+ R"([[0, 1, [4.0, 5.0, 6.0]], [0, 2, null], [0, 3,
[7.0, 8.0, 9.0]]])"));
+ ASSERT_TRUE(success);
+}
+
+TEST_P(WriteAndReadInteTest, TestPKNestedVector) {
+ auto [file_format, file_system] = GetParam();
+ if (file_format != "parquet") {
+ return;
+ }
+
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::int64()),
+ arrow::field("payload", arrow::struct_({arrow::field("embedding",
vector_type),
+ arrow::field("tag",
arrow::utf8())})),
+ };
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::TARGET_FILE_SIZE, "1024"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(
+ auto helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options,
/*is_streaming_mode=*/true));
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> initial_batch,
+ TestHelper::MakeRecordBatch(
+ arrow::struct_(fields),
+ R"([[1, [[1.0, 2.0, 3.0], "initial"]], [2, [null, "null-vector"]],
[3, null]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(initial_batch),
/*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> update_batch,
+ TestHelper::MakeRecordBatch(
+ arrow::struct_(fields),
+ R"([[1, [[4.0, 5.0, 6.0], "updated"]], [2, [[7.0,
8.0, 9.0], null]]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(update_batch),
/*commit_identifier=*/1,
+
/*expected_commit_messages=*/std::nullopt));
+
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ ASSERT_OK(CompactAndCommit(table_path, options, /*commit_identifier=*/2));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
+ helper->NewScan(StartupMode::LatestFull(),
/*snapshot_id=*/std::nullopt));
+ arrow::FieldVector result_fields = fields;
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND",
arrow::int8()));
+ ASSERT_OK_AND_ASSIGN(bool success,
+
helper->ReadAndCheckResult(arrow::struct_(result_fields), data_splits, R"([
+ [0, 1, [[4.0, 5.0, 6.0], "updated"]],
+ [0, 2, [[7.0, 8.0, 9.0], null]],
+ [0, 3, null]
+ ])"));
+ ASSERT_TRUE(success);
+}
+
+TEST_P(WriteAndReadInteTest, TestPKVectorWithListagg) {
+ auto [file_format, file_system] = GetParam();
+ if (file_format != "parquet") {
+ return;
+ }
+
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ arrow::FieldVector fields = {
+ arrow::field("pk", arrow::int64()),
+ arrow::field("embedding", vector_type),
+ arrow::field("tags", arrow::utf8()),
+ };
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::TARGET_FILE_SIZE, "1024"},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, file_system},
+ {Options::MERGE_ENGINE, "aggregation"},
+ {"fields.embedding.aggregate-function", "last_non_null_value"},
+ {"fields.tags.aggregate-function", "listagg"},
+ };
+ if (file_system == "jindo") {
+ options = AddOptionsForJindo(options);
+ }
+ ASSERT_OK_AND_ASSIGN(
+ auto helper,
+ TestHelper::Create(test_dir_, arrow::schema(fields),
/*partition_keys=*/{},
+ /*primary_keys=*/{"pk"}, options,
/*is_streaming_mode=*/true));
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> initial_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields),
+ R"([[1, [1.0, 2.0, 3.0], "alpha"], [2,
null, "one"]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(initial_batch),
/*commit_identifier=*/0,
+
/*expected_commit_messages=*/std::nullopt));
+
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<RecordBatch> update_batch,
+ TestHelper::MakeRecordBatch(
+ arrow::struct_(fields),
+ R"([[1, [4.0, 5.0, 6.0], "beta"], [2, [7.0, 8.0,
9.0], "two"]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(update_batch),
/*commit_identifier=*/1,
+
/*expected_commit_messages=*/std::nullopt));
+
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<RecordBatch> null_vector_batch,
+ TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([[1, null,
"gamma"]])",
+ /*partition_map=*/{}, /*bucket=*/0, {}));
+ ASSERT_OK(helper->WriteAndCommit(std::move(null_vector_batch),
/*commit_identifier=*/2,
+
/*expected_commit_messages=*/std::nullopt));
+
+ std::string table_path = PathUtil::JoinPath(test_dir_, "foo.db/bar");
+ ASSERT_OK(CompactAndCommit(table_path, options, /*commit_identifier=*/3));
+
+ ASSERT_OK_AND_ASSIGN(std::vector<std::shared_ptr<Split>> data_splits,
+ helper->NewScan(StartupMode::LatestFull(),
/*snapshot_id=*/std::nullopt));
+ arrow::FieldVector result_fields = fields;
+ result_fields.insert(result_fields.begin(), arrow::field("_VALUE_KIND",
arrow::int8()));
+ ASSERT_OK_AND_ASSIGN(bool success,
+
helper->ReadAndCheckResult(arrow::struct_(result_fields), data_splits, R"([
+ [0, 1, [4.0, 5.0, 6.0], "alpha,beta,gamma"],
+ [0, 2, [7.0, 8.0, 9.0], "one,two"]
+ ])"));
+ ASSERT_TRUE(success);
+}
+
TEST_P(WriteAndReadInteTest, TestInputChangelogStreamRead) {
arrow::FieldVector fields = {
arrow::field("pk", arrow::utf8()),
diff --git a/test/inte/write_inte_test.cpp b/test/inte/write_inte_test.cpp
index 27bcea67..3c6d2111 100644
--- a/test/inte/write_inte_test.cpp
+++ b/test/inte/write_inte_test.cpp
@@ -106,6 +106,7 @@ class TableSchema;
} // namespace paimon
namespace paimon::test {
+
class WriteInteTest : public testing::Test, public
::testing::WithParamInterface<std::string> {
public:
void SetUp() override {
@@ -4407,6 +4408,81 @@ TEST_P(WriteInteTest,
TestPkSpillableIntermediateMergeWithTempFileTracking) {
ASSERT_OK(ScanAndVerifyResult(table_path, fields, expected));
}
+TEST_P(WriteInteTest, TestPkSpillableVector) {
+ auto file_format = GetParam();
+ if (file_format != "parquet") {
+ return;
+ }
+
+ auto dir = UniqueTestDirectory::Create();
+ auto vector_type =
+ arrow::fixed_size_list(arrow::field("item", arrow::float32(),
/*nullable=*/false), 3);
+ arrow::FieldVector fields = {
+ arrow::field("f0", arrow::utf8()),
+ arrow::field("pt", arrow::int32()),
+ arrow::field("embedding", vector_type),
+ };
+ auto data_type = arrow::struct_(fields);
+ std::map<std::string, std::string> options = {
+ {Options::FILE_FORMAT, file_format},
+ {Options::BUCKET, "1"},
+ {Options::FILE_SYSTEM, "local"},
+ {Options::WRITE_BUFFER_SIZE, "1"},
+ {Options::WRITE_BUFFER_SPILLABLE, "true"},
+ {Options::LOCAL_SORT_MAX_NUM_FILE_HANDLES, "2"},
+ {Options::WRITE_ONLY, "true"},
+ };
+ auto schema = arrow::schema(fields);
+ ::ArrowSchema c_schema;
+ ASSERT_TRUE(arrow::ExportSchema(*schema, &c_schema).ok());
+ ASSERT_OK_AND_ASSIGN(auto table_path, CreateTestTable(dir->Str(), "db",
"tbl", &c_schema,
+
/*partition_keys=*/{"pt"},
+
/*primary_keys=*/{"pt", "f0"}, options));
+
+ std::string tmp_dir = PathUtil::JoinPath(dir->Str(), "tmp");
+ WriteContextBuilder write_builder(table_path, "commit_user_1");
+ write_builder.WithStreamingMode(true).WithTempDirectory(tmp_dir);
+ ASSERT_OK_AND_ASSIGN(std::unique_ptr<WriteContext> write_context,
write_builder.Finish());
+ ASSERT_OK_AND_ASSIGN(auto file_store_write,
FileStoreWrite::Create(std::move(write_context)));
+
+ auto write_array = [](FileStoreWrite* writer, const
std::shared_ptr<arrow::Array>& array) {
+ ArrowArray c_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, &c_array));
+ auto batch = std::make_unique<RecordBatch>(std::map<std::string,
std::string>{{"pt", "10"}},
+ /*bucket=*/0,
+
std::vector<RecordBatch::RowKind>{}, &c_array);
+ return writer->Write(std::move(batch));
+ };
+
+ auto batch1 =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([["Alice", 10,
[1.0, 2.0, 3.0]]])")
+ .ValueOrDie();
+ auto batch2 =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([["Bob", 10,
[4.0, 5.0, 6.0]]])")
+ .ValueOrDie();
+ auto batch3 =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type, R"([["Alice", 10,
[7.0, 8.0, 9.0]]])")
+ .ValueOrDie();
+
+ ASSERT_OK(write_array(file_store_write.get(), batch1));
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
+ ASSERT_OK(write_array(file_store_write.get(), batch2));
+ ASSERT_EQ(1, TestHelper::CountChannelFiles(file_system_, tmp_dir));
+ ASSERT_OK(write_array(file_store_write.get(), batch3));
+ ASSERT_EQ(2, TestHelper::CountChannelFiles(file_system_, tmp_dir));
+
+ ASSERT_OK_AND_ASSIGN(auto commit_messages,
+
file_store_write->PrepareCommit(/*wait_compaction=*/false,
+
/*commit_identifier=*/0));
+ ASSERT_EQ(0, TestHelper::CountChannelFiles(file_system_, tmp_dir));
+ ASSERT_OK(CommitMessages(table_path, commit_messages));
+ ASSERT_OK(file_store_write->Close());
+
+ ASSERT_OK(ScanAndVerifyResult(table_path, fields,
+ R"([[0, "Alice", 10, [7.0, 8.0, 9.0]],
+ [0, "Bob", 10, [4.0, 5.0, 6.0]]])"));
+}
+
TEST_P(WriteInteTest, TestPkSpillableMultiBucketMultiRoundDataCorrectness) {
auto dir = UniqueTestDirectory::Create();
arrow::FieldVector fields = {
