feat(vector): support vector values in deduplicate primary-key tables

docs/source/user_guide/data_types.rst

@@ -197,9 +197,14 @@ and `Arrow DataTypes <https://arrow.apache.org/docs/format/Columnar.html#data-ty
        Paimon C++ currently supports VECTOR columns only in append-only 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. Data-evolution tables may add or drop VECTOR columns and
+       continue reading files written with older table schemas. A VECTOR field's
+       element type and dimension cannot be changed through schema evolution;
+       an incompatible next schema is rejected before it is persisted.
+       Primary-key tables containing VECTOR fields are rejected. VECTOR columns
+       also cannot be partition or bucket keys, or 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``,

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

@@ -112,6 +112,30 @@ bool VectorUtils::ContainsVector(const std::shared_ptr<arrow::Schema>& schema) {
     return false;
 }
 
+Status VectorUtils::ValidateVectorTypeEvolution(
+    const std::shared_ptr<arrow::DataType>& previous_type,
+    const std::shared_ptr<arrow::DataType>& new_type) {
+    if (!previous_type || !new_type) {
+        return Status::Invalid("VECTOR schema evolution types cannot be null.");
+    }
+    bool previous_is_vector = previous_type->id() == arrow::Type::FIXED_SIZE_LIST;
+    bool new_is_vector = new_type->id() == arrow::Type::FIXED_SIZE_LIST;
+    if (!previous_is_vector && !new_is_vector) {
+        return Status::OK();
+    }
+    if (previous_is_vector && new_is_vector) {
+        const auto& previous_vector = checked_cast<const arrow::FixedSizeListType&>(*previous_type);
+        const auto& new_vector = checked_cast<const arrow::FixedSizeListType&>(*new_type);
+        if (previous_vector.list_size() == new_vector.list_size() &&
+            previous_vector.value_type()->Equals(new_vector.value_type())) {
+            return Status::OK();
+        }
+    }
+    return Status::Invalid(
+        fmt::format("VECTOR type mismatch during schema evolution: previous {} vs new {}",
+                    previous_type->ToString(), new_type->ToString()));
+}
+
 Status VectorUtils::ValidateVectorElements(const arrow::Array& array) {
     switch (array.type_id()) {
         case arrow::Type::LIST:

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

@@ -49,6 +49,14 @@ class PAIMON_EXPORT VectorUtils {
     /// Rejects VECTOR values whose elements are not fully materialized or contain nulls.
     /// `array` must be the List or FixedSizeList array holding the VECTOR values.
     static Status ValidateVectorElements(const arrow::Array& array);
+
+    /// Validates a type transition involving a top-level VECTOR.
+    ///
+    /// A VECTOR field keeps its element type and dimension throughout schema evolution.
+    /// Adding or dropping a VECTOR is represented by a new or removed field id and therefore
+    /// does not compare the VECTOR against another type here.
+    static Status ValidateVectorTypeEvolution(const std::shared_ptr<arrow::DataType>& previous_type,
+                                              const std::shared_ptr<arrow::DataType>& new_type);
 };
 
 }  // namespace paimon

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

@@ -58,6 +58,22 @@ TEST(VectorUtilsTest, TestContainsVector) {
     ASSERT_FALSE(VectorUtils::ContainsVector(nullptr));
 }
 
+TEST(VectorUtilsTest, TestValidateVectorTypeEvolution) {
+    auto vector3 = arrow::fixed_size_list(arrow::float32(), 3);
+    ASSERT_OK(VectorUtils::ValidateVectorTypeEvolution(vector3, vector3));
+    ASSERT_OK(VectorUtils::ValidateVectorTypeEvolution(arrow::int32(), arrow::int64()));
+
+    ASSERT_NOK_WITH_MSG(VectorUtils::ValidateVectorTypeEvolution(
+                            vector3, arrow::fixed_size_list(arrow::float32(), 5)),
+                        "VECTOR type mismatch during schema evolution");
+    ASSERT_NOK_WITH_MSG(VectorUtils::ValidateVectorTypeEvolution(
+                            vector3, arrow::fixed_size_list(arrow::float64(), 3)),
+                        "VECTOR type mismatch during schema evolution");
+    ASSERT_NOK_WITH_MSG(
+        VectorUtils::ValidateVectorTypeEvolution(arrow::list(arrow::float32()), vector3),
+        "VECTOR type mismatch during schema evolution");
+}
+
 TEST(VectorUtilsTest, TestValidateVectorElements) {
     auto vector_type = arrow::fixed_size_list(arrow::float32(), 3);
     ASSERT_OK(VectorUtils::ValidateVectorElements(

src/paimon/core/schema/schema_manager.cpp

@@ -22,6 +22,7 @@
 #include <algorithm>
 #include <utility>
 
+#include "paimon/common/utils/date_time_utils.h"
 #include "paimon/common/utils/path_util.h"
 #include "paimon/core/schema/schema_validation.h"
 #include "paimon/core/utils/branch_manager.h"
@@ -121,4 +122,29 @@ Result<std::unique_ptr<TableSchema>> SchemaManager::CreateTable(
     return Status::Invalid("create table failed, should not be here");
 }
 
+Result<std::shared_ptr<TableSchema>> SchemaManager::CommitSchema(
+    const std::vector<DataField>& fields, int32_t highest_field_id,
+    const std::map<std::string, std::string>& options) {
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> current_schema, Latest());
+    if (!current_schema) {
+        return Status::NotExist("Cannot commit schema because the table schema does not exist.");
+    }
+
+    const std::shared_ptr<TableSchema>& current = current_schema.value();
+    PAIMON_ASSIGN_OR_RAISE(
+        std::unique_ptr<TableSchema> next_schema,
+        TableSchema::InitSchema(current->Id() + 1, fields, highest_field_id,
+                                current->PartitionKeys(), current->PrimaryKeys(), options,
+                                current->Comment(), DateTimeUtils::GetCurrentUTCTimeUs() / 1000));
+    PAIMON_RETURN_NOT_OK(SchemaValidation::ValidateSchemaEvolution(*current, *next_schema));
+
+    std::string schema_path = ToSchemaPath(next_schema->Id());
+    PAIMON_ASSIGN_OR_RAISE(std::string content, next_schema->ToJsonString());
+    PAIMON_RETURN_NOT_OK(file_system_->AtomicStore(schema_path, content));
+
+    std::shared_ptr<TableSchema> committed_schema(std::move(next_schema));
+    schema_cache_[committed_schema->Id()] = committed_schema;
+    return committed_schema;
+}
+
 }  // namespace paimon

src/paimon/core/schema/schema_manager.h

@@ -54,6 +54,15 @@ class SchemaManager {
         const std::vector<std::string>& primary_keys,
         const std::map<std::string, std::string>& options);
 
+    /// Atomically appends a schema version after validating it against the latest schema.
+    ///
+    /// `fields` is the complete next field list. Existing fields retain their field ids, while
+    /// added fields use ids greater than the latest schema's highest field id. Partition keys,
+    /// primary keys, and the table comment are preserved.
+    Result<std::shared_ptr<TableSchema>> CommitSchema(
+        const std::vector<DataField>& fields, int32_t highest_field_id,
+        const std::map<std::string, std::string>& options);
+
     std::string SchemaDirectory() const;
     Result<bool> SchemaExists(int64_t id) const;
     Result<std::vector<int64_t>> ListAllIds() const;

src/paimon/core/schema/schema_manager_test.cpp

@@ -24,6 +24,7 @@
 
 #include "arrow/type.h"
 #include "gtest/gtest.h"
+#include "paimon/defs.h"
 #include "paimon/fs/local/local_file_system.h"
 #include "paimon/status.h"
 #include "paimon/testing/utils/testharness.h"
@@ -192,6 +193,57 @@ TEST(SchemaManagerTest, TestCreateTableAlreadyExists) {
     ASSERT_NOK_WITH_MSG(manager.CreateTable(schema, {}, {}, {}), "Schema in filesystem exists");
 }
 
+TEST(SchemaManagerTest, TestCommitVectorSchemaEvolution) {
+    auto fs = std::make_shared<LocalFileSystem>();
+    auto dir = UniqueTestDirectory::Create();
+    SchemaManager manager(fs, dir->Str());
+
+    auto vector3 =
+        arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3);
+    auto vector2 =
+        arrow::fixed_size_list(arrow::field("item", arrow::int64(), /*nullable=*/false), 2);
+    auto schema = arrow::schema({
+        arrow::field("id", arrow::int32()),
+        arrow::field("retained_embedding", vector3),
+        arrow::field("dropped_embedding", vector2),
+    });
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "-1"},
+        {Options::FILE_FORMAT, "parquet"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN([[maybe_unused]] std::unique_ptr<TableSchema> created,
+                         manager.CreateTable(schema, /*partition_keys=*/{},
+                                             /*primary_keys=*/{}, options));
+
+    auto added_vector =
+        arrow::fixed_size_list(arrow::field("item", arrow::float64(), /*nullable=*/false), 2);
+    std::vector<DataField> evolved_fields = {
+        DataField(0, arrow::field("id", arrow::int32())),
+        DataField(1, arrow::field("retained_embedding", vector3)),
+        DataField(3, arrow::field("added_embedding", added_vector)),
+    };
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> committed,
+                         manager.CommitSchema(evolved_fields, /*highest_field_id=*/3, options));
+    ASSERT_EQ(committed->Id(), 1);
+    ASSERT_NOK(committed->GetField("dropped_embedding"));
+    ASSERT_OK(committed->GetField("added_embedding"));
+
+    std::vector<DataField> incompatible_fields = evolved_fields;
+    incompatible_fields[1] = DataField(
+        1, arrow::field("retained_embedding", arrow::fixed_size_list(arrow::float32(), 5)));
+    ASSERT_NOK_WITH_MSG(manager.CommitSchema(incompatible_fields, /*highest_field_id=*/3, options),
+                        "VECTOR type mismatch during schema evolution");
+
+    incompatible_fields[1] = DataField(
+        1, arrow::field("retained_embedding", arrow::fixed_size_list(arrow::float64(), 3)));
+    ASSERT_NOK_WITH_MSG(manager.CommitSchema(incompatible_fields, /*highest_field_id=*/3, options),
+                        "VECTOR type mismatch during schema evolution");
+    ASSERT_OK_AND_ASSIGN(bool schema_2_exists, manager.SchemaExists(/*id=*/2));
+    ASSERT_FALSE(schema_2_exists);
+}
+
 TEST(SchemaManagerTest, TestListAllIds) {
     auto fs = std::make_shared<LocalFileSystem>();
     std::string table_root =

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) {
@@ -198,6 +208,46 @@ Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
     return Status::OK();
 }
 
+Status SchemaValidation::ValidateSchemaEvolution(const TableSchema& current_schema,
+                                                 const TableSchema& next_schema) {
+    PAIMON_RETURN_NOT_OK(ValidateTableSchema(next_schema));
+    if (next_schema.Id() != current_schema.Id() + 1) {
+        return Status::Invalid(fmt::format("Next schema id must be {}, but is {}.",
+                                           current_schema.Id() + 1, next_schema.Id()));
+    }
+    if (next_schema.PartitionKeys() != current_schema.PartitionKeys()) {
+        return Status::Invalid("Partition keys cannot be changed by schema evolution.");
+    }
+    if (next_schema.PrimaryKeys() != current_schema.PrimaryKeys()) {
+        return Status::Invalid("Primary keys cannot be changed by schema evolution.");
+    }
+    if (next_schema.HighestFieldId() < current_schema.HighestFieldId()) {
+        return Status::Invalid(fmt::format("Highest field id cannot decrease from {} to {}.",
+                                           current_schema.HighestFieldId(),
+                                           next_schema.HighestFieldId()));
+    }
+
+    std::unordered_map<int32_t, DataField> current_fields;
+    for (const auto& field : current_schema.Fields()) {
+        current_fields.emplace(field.Id(), field);
+    }
+    for (const auto& next_field : next_schema.Fields()) {
+        auto current = current_fields.find(next_field.Id());
+        if (current == current_fields.end()) {
+            if (next_field.Id() <= current_schema.HighestFieldId()) {
+                return Status::Invalid(fmt::format(
+                    "New field '{}' uses id {}, which must be greater than the previous highest "
+                    "field id {}.",
+                    next_field.Name(), next_field.Id(), current_schema.HighestFieldId()));
+            }
+            continue;
+        }
+        PAIMON_RETURN_NOT_OK(
+            VectorUtils::ValidateVectorTypeEvolution(current->second.Type(), next_field.Type()));
+    }
+    return Status::OK();
+}
+
 Status SchemaValidation::ValidateNoDuplicateField(const std::vector<std::string>& field_names,
                                                   const std::string& error_message_intro) {
     auto duplicate_field_names = ObjectUtils::DuplicateItems(field_names);
@@ -383,6 +433,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)) {
@@ -450,6 +502,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));
@@ -785,10 +838,6 @@ Status SchemaValidation::ValidateVectorFields(const TableSchema& schema,
         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.");
-    }
     PAIMON_RETURN_NOT_OK(
         ValidateVectorFileFormat(Options::FILE_FORMAT, options.GetFileFormat()->Identifier()));
     return ValidatePerLevelOption(options.ToMap(), Options::FILE_FORMAT_PER_LEVEL,

src/paimon/core/schema/schema_validation.h

@@ -46,6 +46,13 @@ class SchemaValidation {
 
     static Status ValidateTableSchema(const TableSchema& schema);
 
+    /// Validates an already constructed next schema against the current table schema.
+    ///
+    /// This checks version and field-id invariants together with type transitions that cannot be
+    /// represented by the read path, including VECTOR dimension and element-type changes.
+    static Status ValidateSchemaEvolution(const TableSchema& current_schema,
+                                          const TableSchema& next_schema);
+
     static bool IsPostponeBucketTable(const TableSchema& schema, int32_t bucket);
 
  private:

src/paimon/core/schema/schema_validation_test.cpp

@@ -86,6 +86,49 @@ 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=*/{},
@@ -114,14 +157,12 @@ TEST(SchemaValidationTest, TestVectorType) {
                          TableSchema::Create(/*schema_id=*/0, 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.");
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
     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.");
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
 }
 
 #ifdef PAIMON_ENABLE_MOSAIC

src/paimon/core/schema/table_schema.h

@@ -38,6 +38,8 @@
 struct ArrowSchema;
 
 namespace paimon {
+class SchemaManager;
+
 /// Schema of a table, including schemaId and fieldId.
 class TableSchema : public DataSchema, public Jsonizable<TableSchema> {
  public:
@@ -125,6 +127,7 @@ class TableSchema : public DataSchema, public Jsonizable<TableSchema> {
     }
 
  private:
+    friend class SchemaManager;
     JSONIZABLE_FRIEND_AND_DEFAULT_CTOR(TableSchema);
 
     static Result<std::unique_ptr<TableSchema>> InitSchema(