diff --git a/docs/source/user_guide/data_types.rst b/docs/source/user_guide/data_types.rst index add537adb..8c06a7e05 100644 --- a/docs/source/user_guide/data_types.rst +++ b/docs/source/user_guide/data_types.rst @@ -197,9 +197,14 @@ and `Arrow DataTypes & schema) { return false; } +Status VectorUtils::ValidateVectorTypeEvolution( + const std::shared_ptr& previous_type, + const std::shared_ptr& 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(*previous_type); + const auto& new_vector = checked_cast(*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: diff --git a/src/paimon/common/utils/arrow/vector_utils.h b/src/paimon/common/utils/arrow/vector_utils.h index 0031a5de7..9eebc34c3 100644 --- a/src/paimon/common/utils/arrow/vector_utils.h +++ b/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& previous_type, + const std::shared_ptr& new_type); }; } // namespace paimon diff --git a/src/paimon/common/utils/arrow/vector_utils_test.cpp b/src/paimon/common/utils/arrow/vector_utils_test.cpp index 1cce6853e..9560a0242 100644 --- a/src/paimon/common/utils/arrow/vector_utils_test.cpp +++ b/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( diff --git a/src/paimon/core/schema/schema_manager.cpp b/src/paimon/core/schema/schema_manager.cpp index 2425cc4a6..4e31508c0 100644 --- a/src/paimon/core/schema/schema_manager.cpp +++ b/src/paimon/core/schema/schema_manager.cpp @@ -22,6 +22,7 @@ #include #include +#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> SchemaManager::CreateTable( return Status::Invalid("create table failed, should not be here"); } +Result> SchemaManager::CommitSchema( + const std::vector& fields, int32_t highest_field_id, + const std::map& options) { + PAIMON_ASSIGN_OR_RAISE(std::optional> current_schema, Latest()); + if (!current_schema) { + return Status::NotExist("Cannot commit schema because the table schema does not exist."); + } + + const std::shared_ptr& current = current_schema.value(); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr 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 committed_schema(std::move(next_schema)); + schema_cache_[committed_schema->Id()] = committed_schema; + return committed_schema; +} + } // namespace paimon diff --git a/src/paimon/core/schema/schema_manager.h b/src/paimon/core/schema/schema_manager.h index 382fa3033..2aa64066b 100644 --- a/src/paimon/core/schema/schema_manager.h +++ b/src/paimon/core/schema/schema_manager.h @@ -54,6 +54,15 @@ class SchemaManager { const std::vector& primary_keys, const std::map& 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> CommitSchema( + const std::vector& fields, int32_t highest_field_id, + const std::map& options); + std::string SchemaDirectory() const; Result SchemaExists(int64_t id) const; Result> ListAllIds() const; diff --git a/src/paimon/core/schema/schema_manager_test.cpp b/src/paimon/core/schema/schema_manager_test.cpp index 83f995a99..6e028fffd 100644 --- a/src/paimon/core/schema/schema_manager_test.cpp +++ b/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(); + 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 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 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 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 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 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(); std::string table_root = diff --git a/src/paimon/core/schema/schema_validation.cpp b/src/paimon/core/schema/schema_validation.cpp index 90f508b47..417791d03 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& 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 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& 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 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, diff --git a/src/paimon/core/schema/schema_validation.h b/src/paimon/core/schema/schema_validation.h index 388ab42e2..a200b4423 100644 --- a/src/paimon/core/schema/schema_validation.h +++ b/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: diff --git a/src/paimon/core/schema/schema_validation_test.cpp b/src/paimon/core/schema/schema_validation_test.cpp index 2137970a3..8dd401f2b 100644 --- a/src/paimon/core/schema/schema_validation_test.cpp +++ b/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 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 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 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 diff --git a/src/paimon/core/schema/table_schema.h b/src/paimon/core/schema/table_schema.h index eab109cf1..a5bbf1309 100644 --- a/src/paimon/core/schema/table_schema.h +++ b/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 { public: @@ -125,6 +127,7 @@ class TableSchema : public DataSchema, public Jsonizable { } private: + friend class SchemaManager; JSONIZABLE_FRIEND_AND_DEFAULT_CTOR(TableSchema); static Result> InitSchema( diff --git a/src/paimon/core/utils/nested_projection_utils.cpp b/src/paimon/core/utils/nested_projection_utils.cpp index c786936be..83fe136e6 100644 --- a/src/paimon/core/utils/nested_projection_utils.cpp +++ b/src/paimon/core/utils/nested_projection_utils.cpp @@ -35,6 +35,7 @@ #include "fmt/format.h" #include "paimon/common/data/variant/variant_access_utils.h" #include "paimon/common/data/variant/variant_type_utils.h" +#include "paimon/common/utils/arrow/vector_utils.h" #include "paimon/common/utils/checked_cast.h" #include "paimon/common/utils/string_utils.h" #include "paimon/core/casting/casting_utils.h" @@ -164,6 +165,13 @@ Result EqualWithFieldIds(const std::shared_ptr& a, if (a->id() != b->id() || a->num_fields() != b->num_fields()) { return false; } + if (a->id() == arrow::Type::FIXED_SIZE_LIST) { + const auto& vector_a = checked_cast(*a); + const auto& vector_b = checked_cast(*b); + if (vector_a.list_size() != vector_b.list_size()) { + return false; + } + } if (a->num_fields() == 0) { return a->Equals(*b); } @@ -238,6 +246,7 @@ Result IsVariantAccessSubstitution(const std::shared_ptr& Result> PruneRepeatedItemType( const std::shared_ptr& read_type, const std::shared_ptr& data_type, const char* container) { + PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorTypeEvolution(data_type, read_type)); PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(read_type, data_type)); if (same) { return data_type; @@ -315,6 +324,7 @@ Result> PruneRepeatedItemType( Result>> NestedProjectionUtils::PruneDataType( const std::shared_ptr& read_type, const std::shared_ptr& data_type) { + PAIMON_RETURN_NOT_OK(VectorUtils::ValidateVectorTypeEvolution(data_type, read_type)); // Identical types (including paimon field IDs) need no pruning. PAIMON_ASSIGN_OR_RAISE(bool same, EqualWithFieldIds(read_type, data_type)); if (same) { diff --git a/src/paimon/core/utils/nested_projection_utils_test.cpp b/src/paimon/core/utils/nested_projection_utils_test.cpp index 153d90e17..35c1132fc 100644 --- a/src/paimon/core/utils/nested_projection_utils_test.cpp +++ b/src/paimon/core/utils/nested_projection_utils_test.cpp @@ -105,6 +105,25 @@ TEST(NestedProjectionUtilsTest, PruneDataTypeAtomicType) { ASSERT_TRUE(result.value()->Equals(data_type)); } +TEST(NestedProjectionUtilsTest, PruneDataTypeRejectsVectorDimensionChange) { + auto data_type = arrow::fixed_size_list(arrow::float32(), 3); + auto read_type = arrow::fixed_size_list(arrow::float32(), 5); + + ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(read_type, data_type), + "VECTOR type mismatch during schema evolution: previous " + "fixed_size_list[3] vs new fixed_size_list[5]"); +} + +TEST(NestedProjectionUtilsTest, PruneDataTypeRejectsNestedVectorDimensionChange) { + auto data_type = + arrow::struct_({MakeField("embedding", arrow::fixed_size_list(arrow::float32(), 3), 1)}); + auto read_type = + arrow::struct_({MakeField("embedding", arrow::fixed_size_list(arrow::float32(), 5), 1)}); + + ASSERT_NOK_WITH_MSG(NestedProjectionUtils::PruneDataType(read_type, data_type), + "VECTOR type mismatch during schema evolution"); +} + TEST(NestedProjectionUtilsTest, PruneDataTypeStructPruneSubset) { // data: STRUCT // read: STRUCT diff --git a/src/paimon/testing/utils/test_helper.h b/src/paimon/testing/utils/test_helper.h index 925a59ad3..daec2c259 100644 --- a/src/paimon/testing/utils/test_helper.h +++ b/src/paimon/testing/utils/test_helper.h @@ -351,20 +351,10 @@ class TestHelper { const std::vector& fields, int32_t highest_field_id, const std::map& options) { SchemaManager schema_manager(file_system, table_path); - PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema_opt, - schema_manager.Latest()); - if (!latest_schema_opt) { - return Status::Invalid("table schema does not exist"); - } - auto next_schema = std::make_shared(*latest_schema_opt.value()); - next_schema->id_ = latest_schema_opt.value()->Id() + 1; - next_schema->fields_ = fields; - next_schema->highest_field_id_ = highest_field_id; - next_schema->options_ = options; - PAIMON_ASSIGN_OR_RAISE(std::string schema_content, next_schema->ToJsonString()); - std::string schema_path = PathUtil::JoinPath(schema_manager.SchemaDirectory(), - fmt::format("schema-{}", next_schema->Id())); - return file_system->AtomicStore(schema_path, schema_content); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr committed_schema, + schema_manager.CommitSchema(fields, highest_field_id, options)); + (void)committed_schema; + return Status::OK(); } static void CheckCommitMessages(std::vector> expected, diff --git a/test/inte/data_evolution_table_test.cpp b/test/inte/data_evolution_table_test.cpp index 303c6d700..6acedd8ed 100644 --- a/test/inte/data_evolution_table_test.cpp +++ b/test/inte/data_evolution_table_test.cpp @@ -1657,6 +1657,117 @@ TEST_P(DataEvolutionTableTest, TestPartitionWithPredicate) { } } +TEST_P(DataEvolutionTableTest, TestVectorSchemaEvolution) { + if (FileFormat() != "parquet") { + GTEST_SKIP() << "VECTOR currently only supports Parquet data files"; + } + + auto retained_vector_type = + arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 3); + auto dropped_vector_type = + arrow::fixed_size_list(arrow::field("item", arrow::int64(), /*nullable=*/false), 2); + auto added_vector_type = + arrow::fixed_size_list(arrow::field("item", arrow::float64(), /*nullable=*/false), 2); + arrow::FieldVector fields_v0 = { + arrow::field("id", arrow::int32()), + arrow::field("retained_embedding", retained_vector_type), + arrow::field("dropped_embedding", dropped_vector_type), + }; + std::map options = { + {Options::MANIFEST_FORMAT, "orc"}, {Options::FILE_FORMAT, "parquet"}, + {Options::FILE_SYSTEM, "local"}, {Options::BUCKET, "-1"}, + {Options::ROW_TRACKING_ENABLED, "true"}, {Options::DATA_EVOLUTION_ENABLED, "true"}, + }; + CreateTable(fields_v0, /*partition_keys=*/{}, options); + std::string table_path = PathUtil::JoinPath(dir_->Str(), "foo.db/bar"); + + auto initial_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_v0), R"([ + [1, [1.0, 2.0, 3.0], [10, 11]], + [2, null, [20, 21]] + ])") + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN( + std::vector> initial_commit_msgs, + WriteArray(table_path, arrow::schema(fields_v0)->field_names(), initial_array)); + ASSERT_OK(Commit(table_path, initial_commit_msgs)); + + // Drop one VECTOR column and add another. Files written with schema 0 must still expose the + // retained VECTOR and null-fill the newly added one. + arrow::FieldVector fields_v1 = { + fields_v0[0], + fields_v0[1], + arrow::field("added_embedding", added_vector_type), + }; + ASSERT_OK(TestHelper::WriteNextSchema( + dir_->GetFileSystem(), table_path, + {DataField(0, fields_v1[0]), DataField(1, fields_v1[1]), DataField(3, fields_v1[2])}, + /*highest_field_id=*/3, options)); + + auto expected_after_evolution = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_v1), R"([ + [1, [1.0, 2.0, 3.0], null], + [2, null, null] + ])") + .ValueOrDie()); + ASSERT_OK( + ScanAndRead(table_path, arrow::schema(fields_v1)->field_names(), expected_after_evolution)); + + // A partial write under schema 1 fills the added VECTOR for the same row range. The read + // merges it with scalar and VECTOR columns from the schema-0 file. + auto added_array = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_({fields_v1[2]}), R"([ + [[100.0, 101.0]], + [null] + ])") + .ValueOrDie()); + ASSERT_OK_AND_ASSIGN(std::vector> added_commit_msgs, + WriteArray(table_path, {"added_embedding"}, added_array)); + SetFirstRowId(/*reset_first_row_id=*/0, added_commit_msgs); + ASSERT_OK(Commit(table_path, added_commit_msgs)); + + auto expected_after_partial_write = std::dynamic_pointer_cast( + arrow::ipc::internal::json::ArrayFromJSON(arrow::struct_(fields_v1), R"([ + [1, [1.0, 2.0, 3.0], [100.0, 101.0]], + [2, null, null] + ])") + .ValueOrDie()); + ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields_v1)->field_names(), + expected_after_partial_write)); + + // Reusing a field id with a different dimension is not a compatible schema evolution. + auto incompatible_vector_type = + arrow::fixed_size_list(arrow::field("item", arrow::float32(), /*nullable=*/false), 5); + arrow::FieldVector incompatible_fields = { + fields_v1[0], + arrow::field("retained_embedding", incompatible_vector_type), + fields_v1[2], + }; + ASSERT_NOK_WITH_MSG(TestHelper::WriteNextSchema(dir_->GetFileSystem(), table_path, + {DataField(0, incompatible_fields[0]), + DataField(1, incompatible_fields[1]), + DataField(3, incompatible_fields[2])}, + /*highest_field_id=*/3, options), + "VECTOR type mismatch during schema evolution"); + + // Changing the element type is equally incompatible, even when the dimension is unchanged. + auto incompatible_element_type = + arrow::fixed_size_list(arrow::field("item", arrow::float64(), /*nullable=*/false), 3); + incompatible_fields[1] = arrow::field("retained_embedding", incompatible_element_type); + ASSERT_NOK_WITH_MSG(TestHelper::WriteNextSchema(dir_->GetFileSystem(), table_path, + {DataField(0, incompatible_fields[0]), + DataField(1, incompatible_fields[1]), + DataField(3, incompatible_fields[2])}, + /*highest_field_id=*/3, options), + "VECTOR type mismatch during schema evolution"); + + SchemaManager schema_manager(dir_->GetFileSystem(), table_path); + ASSERT_OK_AND_ASSIGN(bool schema_2_exists, schema_manager.SchemaExists(/*id=*/2)); + ASSERT_FALSE(schema_2_exists); + ASSERT_OK(ScanAndRead(table_path, arrow::schema(fields_v1)->field_names(), + expected_after_partial_write)); +} + TEST_P(DataEvolutionTableTest, TestAlterTable) { auto file_format = FileFormat(); if (file_format == "mosaic") {