Skip to content

Commit 88a6e9a

Browse files
ChaomingZhangCNChaomingZhangCN
authored
feat(vector): support data evolution reads and writes (#295)
1 parent c0e506d commit 88a6e9a

9 files changed

Lines changed: 380 additions & 12 deletions

File tree

‎docs/source/user_guide/data_types.rst‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -197,8 +197,11 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
197197
Paimon C++ supports VECTOR value columns in append-only and primary-key
198198
tables backed by Parquet data files. They use the standard Parquet LIST
199199
representation on disk and are restored as Arrow ``FixedSizeList``
200-
values on read. VECTOR is not supported in data-evolution tables. VECTOR
201-
columns cannot be primary, partition, or bucket keys, nor comparator-based
200+
values on read. Row-tracking append-only tables with data evolution also
201+
support full-row and partial-column VECTOR writes, including reads across
202+
schema versions that add or drop VECTOR columns. Changing a VECTOR's
203+
dimension or element type is not supported. VECTOR columns cannot be
204+
primary, partition, or bucket keys, nor comparator-based
202205
ordering fields such as sequence and sequence-group fields. Dedicated
203206
vector storage is not included yet.
204207

‎src/paimon/common/utils/arrow/vector_utils.cpp‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,26 @@ bool VectorUtils::ContainsVector(const std::shared_ptr<arrow::Schema>& schema) {
112112
return false;
113113
}
114114

115+
Status VectorUtils::ValidateVectorEvolution(const std::shared_ptr<arrow::DataType>& read_type,
116+
const std::shared_ptr<arrow::DataType>& data_type) {
117+
bool read_is_vector = read_type->id() == arrow::Type::FIXED_SIZE_LIST;
118+
bool data_is_vector = data_type->id() == arrow::Type::FIXED_SIZE_LIST;
119+
if (!read_is_vector && !data_is_vector) {
120+
return Status::OK();
121+
}
122+
if (read_is_vector && data_is_vector) {
123+
const auto& read_vector = checked_cast<const arrow::FixedSizeListType&>(*read_type);
124+
const auto& data_vector = checked_cast<const arrow::FixedSizeListType&>(*data_type);
125+
if (read_vector.list_size() == data_vector.list_size() &&
126+
read_vector.value_type()->Equals(data_vector.value_type())) {
127+
return Status::OK();
128+
}
129+
}
130+
return Status::Invalid(
131+
fmt::format("VECTOR type mismatch during schema evolution: data {} vs read {}",
132+
data_type->ToString(), read_type->ToString()));
133+
}
134+
115135
Status VectorUtils::ValidateVectorElements(const arrow::Array& array) {
116136
switch (array.type_id()) {
117137
case arrow::Type::LIST:

‎src/paimon/common/utils/arrow/vector_utils.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,11 @@ class PAIMON_EXPORT VectorUtils {
4646

4747
static bool ContainsVector(const std::shared_ptr<arrow::Schema>& schema);
4848

49+
/// Rejects VECTOR dimension or element type changes and VECTOR/non-VECTOR conversions.
50+
/// Checks only these types; callers validate nested children recursively.
51+
static Status ValidateVectorEvolution(const std::shared_ptr<arrow::DataType>& read_type,
52+
const std::shared_ptr<arrow::DataType>& data_type);
53+
4954
/// Rejects VECTOR values whose elements are not fully materialized or contain nulls.
5055
/// `array` must be the List or FixedSizeList array holding the VECTOR values.
5156
static Status ValidateVectorElements(const arrow::Array& array);

‎src/paimon/common/utils/arrow/vector_utils_test.cpp‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include "paimon/common/utils/arrow/vector_utils.h"
2121

2222
#include <memory>
23+
#include <vector>
2324

2425
#include "arrow/api.h"
2526
#include "arrow/ipc/json_simple.h"
@@ -58,6 +59,33 @@ TEST(VectorUtilsTest, TestContainsVector) {
5859
ASSERT_FALSE(VectorUtils::ContainsVector(nullptr));
5960
}
6061

62+
TEST(VectorUtilsTest, TestValidateVectorEvolution) {
63+
auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
64+
ASSERT_OK(VectorUtils::ValidateVectorEvolution(vector_type, vector_type));
65+
auto read_type =
66+
arrow::fixed_size_list(arrow::field("element", arrow::float32(), /*nullable=*/false), 3);
67+
ASSERT_OK(VectorUtils::ValidateVectorEvolution(read_type, vector_type));
68+
ASSERT_OK(VectorUtils::ValidateVectorEvolution(vector_type, read_type));
69+
ASSERT_OK(VectorUtils::ValidateVectorEvolution(arrow::int64(), arrow::int32()));
70+
}
71+
72+
TEST(VectorUtilsTest, TestValidateVectorEvolutionRejectsIncompatibleTypes) {
73+
auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
74+
std::vector<std::shared_ptr<arrow::DataType>> incompatible_types = {
75+
arrow::fixed_size_list(arrow::float32(), 5),
76+
arrow::fixed_size_list(arrow::float64(), 3),
77+
arrow::list(arrow::float32()),
78+
arrow::int32(),
79+
};
80+
for (const auto& incompatible_type : incompatible_types) {
81+
SCOPED_TRACE(incompatible_type->ToString());
82+
ASSERT_NOK_WITH_MSG(VectorUtils::ValidateVectorEvolution(incompatible_type, vector_type),
83+
"VECTOR type mismatch during schema evolution");
84+
ASSERT_NOK_WITH_MSG(VectorUtils::ValidateVectorEvolution(vector_type, incompatible_type),
85+
"VECTOR type mismatch during schema evolution");
86+
}
87+
}
88+
6189
TEST(VectorUtilsTest, TestValidateVectorElements) {
6290
auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
6391
ASSERT_OK(VectorUtils::ValidateVectorElements(

‎src/paimon/core/schema/schema_validation.cpp‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -925,10 +925,6 @@ Status SchemaValidation::ValidateVectorFields(const TableSchema& schema,
925925
if (!has_vector) {
926926
return Status::OK();
927927
}
928-
if (options.DataEvolutionEnabled()) {
929-
return Status::NotImplemented(
930-
"VECTOR fields in data-evolution tables are not implemented yet.");
931-
}
932928
PAIMON_RETURN_NOT_OK(
933929
ValidateVectorFileFormat(Options::FILE_FORMAT, options.GetFileFormat()->Identifier()));
934930
return ValidatePerLevelOption(options.ToMap(), Options::FILE_FORMAT_PER_LEVEL,

‎src/paimon/core/schema/schema_validation_test.cpp‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -156,14 +156,19 @@ TEST(SchemaValidationTest, TestVectorType) {
156156
TableSchema::Create(/*schema_id=*/0, schema,
157157
/*partition_keys=*/{},
158158
/*primary_keys=*/{}, data_evolution_options));
159-
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
160-
"VECTOR fields in data-evolution tables are not implemented yet.");
159+
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
161160
ASSERT_OK_AND_ASSIGN(
162161
table_schema,
163162
TableSchema::Create(/*schema_id=*/0, nested_schema,
164163
/*partition_keys=*/{}, /*primary_keys=*/{}, data_evolution_options));
164+
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
165+
166+
data_evolution_options[Options::FILE_FORMAT] = "orc";
167+
ASSERT_OK_AND_ASSIGN(table_schema,
168+
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
169+
/*primary_keys=*/{}, data_evolution_options));
165170
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
166-
"VECTOR fields in data-evolution tables are not implemented yet.");
171+
"VECTOR currently only supports parquet data files");
167172
}
168173

169174
#ifdef PAIMON_ENABLE_MOSAIC

‎src/paimon/core/utils/nested_projection_utils.cpp‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
#include "paimon/common/data/blob_utils.h"
3737
#include "paimon/common/data/variant/variant_access_utils.h"
3838
#include "paimon/common/data/variant/variant_type_utils.h"
39+
#include "paimon/common/utils/arrow/vector_utils.h"
3940
#include "paimon/common/utils/checked_cast.h"
4041
#include "paimon/common/utils/string_utils.h"
4142
#include "paimon/core/casting/casting_utils.h"
@@ -165,6 +166,13 @@ Result<bool> EqualWithFieldIds(const std::shared_ptr<arrow::DataType>& a,
165166
if (a->id() != b->id() || a->num_fields() != b->num_fields()) {
166167
return false;
167168
}
169+
if (a->id() == arrow::Type::FIXED_SIZE_LIST) {
170+
const auto& vector_a = checked_cast<const arrow::FixedSizeListType&>(*a);
171+
const auto& vector_b = checked_cast<const arrow::FixedSizeListType&>(*b);
172+
if (vector_a.list_size() != vector_b.list_size()) {
173+
return false;
174+
}
175+
}
168176
if (a->num_fields() == 0) {
169177
return a->Equals(*b);
170178
}
@@ -197,6 +205,7 @@ Result<bool> EqualWithFieldIds(const std::shared_ptr<arrow::DataType>& a,
197205
/// projections and nothing else changed (matching paimon field IDs).
198206
Result<bool> IsVariantAccessSubstitution(const std::shared_ptr<arrow::DataType>& read_type,
199207
const std::shared_ptr<arrow::DataType>& data_type) {
208+
PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorEvolution(read_type, data_type));
200209
PAIMON_ASSIGN_OR_RAISE(bool equal, EqualWithFieldIds(read_type, data_type));
201210
if (equal) {
202211
return true;
@@ -239,6 +248,7 @@ Result<bool> IsVariantAccessSubstitution(const std::shared_ptr<arrow::DataType>&
239248
Result<std::shared_ptr<arrow::DataType>> PruneRepeatedItemType(
240249
const std::shared_ptr<arrow::DataType>& read_type,
241250
const std::shared_ptr<arrow::DataType>& data_type, const char* container) {
251+
PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorEvolution(read_type, data_type));
242252
PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(read_type, data_type));
243253
if (same) {
244254
return data_type;
@@ -316,6 +326,7 @@ Result<std::shared_ptr<arrow::DataType>> PruneRepeatedItemType(
316326
Result<std::optional<std::shared_ptr<arrow::DataType>>> NestedProjectionUtils::PruneDataType(
317327
const std::shared_ptr<arrow::DataType>& read_type,
318328
const std::shared_ptr<arrow::DataType>& data_type) {
329+
PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorEvolution(read_type, data_type));
319330
// Identical types (including paimon field IDs) need no pruning.
320331
PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(read_type, data_type));
321332
if (same) {

‎src/paimon/core/utils/nested_projection_utils_test.cpp‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -106,6 +106,60 @@ TEST(NestedProjectionUtilsTest, PruneDataTypeAtomicType) {
106106
ASSERT_TRUE(result.value()->Equals(data_type));
107107
}
108108

109+
TEST(NestedProjectionUtilsTest, PruneDataTypeVector) {
110+
auto data_type = arrow::fixed_size_list(arrow::float32(), 3);
111+
auto read_type =
112+
arrow::fixed_size_list(arrow::field("element", arrow::float32(), /*nullable=*/false), 3);
113+
ASSERT_OK_AND_ASSIGN(auto result, NestedProjectionUtils::PruneDataType(read_type, data_type));
114+
ASSERT_TRUE(result.has_value());
115+
ASSERT_TRUE(result.value()->Equals(data_type));
116+
}
117+
118+
TEST(NestedProjectionUtilsTest, PruneDataTypeRejectsIncompatibleVectorEvolution) {
119+
auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
120+
std::vector<std::shared_ptr<arrow::DataType>> incompatible_types = {
121+
arrow::fixed_size_list(arrow::float32(), 5),
122+
arrow::fixed_size_list(arrow::float64(), 3),
123+
arrow::list(arrow::float32()),
124+
arrow::int32(),
125+
};
126+
for (const auto& incompatible_type : incompatible_types) {
127+
SCOPED_TRACE(incompatible_type->ToString());
128+
ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(incompatible_type, vector_type),
129+
"VECTOR type mismatch during schema evolution");
130+
ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(vector_type, incompatible_type),
131+
"VECTOR type mismatch during schema evolution");
132+
}
133+
}
134+
135+
TEST(NestedProjectionUtilsTest, PruneDataTypeRejectsNestedVectorEvolution) {
136+
auto data_vector = arrow::fixed_size_list(arrow::float32(), 3);
137+
for (const auto& read_vector : {arrow::fixed_size_list(arrow::float32(), 5),
138+
arrow::fixed_size_list(arrow::float64(), 3)}) {
139+
SCOPED_TRACE(read_vector->ToString());
140+
ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(
141+
arrow::struct_({MakeField("embedding", read_vector, 1)}),
142+
arrow::struct_({MakeField("embedding", data_vector, 1)})),
143+
"VECTOR type mismatch during schema evolution");
144+
ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(arrow::list(read_vector),
145+
arrow::list(data_vector)),
146+
"VECTOR type mismatch during schema evolution");
147+
ASSERT_NOK_WITH_MSG(
148+
NestedProjectionUtils::PruneDataType(arrow::map(arrow::utf8(), read_vector),
149+
arrow::map(arrow::utf8(), data_vector)),
150+
"VECTOR type mismatch during schema evolution");
151+
auto read_struct = arrow::struct_({MakeField("embedding", read_vector, 1)});
152+
auto data_struct = arrow::struct_({MakeField("embedding", data_vector, 1)});
153+
ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(arrow::list(read_struct),
154+
arrow::list(data_struct)),
155+
"VECTOR type mismatch during schema evolution");
156+
ASSERT_NOK_WITH_MSG(
157+
NestedProjectionUtils::PruneDataType(arrow::map(arrow::utf8(), read_struct),
158+
arrow::map(arrow::utf8(), data_struct)),
159+
"VECTOR type mismatch during schema evolution");
160+
}
161+
}
162+
109163
TEST(NestedProjectionUtilsTest, PruneDataTypeStructPruneSubset) {
110164
// data: STRUCT<x:INT(id=1), y:STRING(id=2), z:DOUBLE(id=3)>
111165
// read: STRUCT<x:INT(id=1)>

0 commit comments

Comments
 (0)