[PR] feat(schema): add schema validation utilities [paimon-cpp]

["lucasfang (via GitHub)" <gi...@apache.org>]

lucasfang opened a new pull request, #55:
URL: https://github.com/apache/paimon-cpp/pull/55

   ### Purpose
   
   Linked issue: No linked issue
   
   This change adds schema validation support for table schemas and Arrow schemas.
   
   Included changes:
   
   - Table schema validation:
     - Adds schema validation helpers for checking table schema definitions.
     - Adds validation coverage for fields, partition keys, primary keys, options, and schema constraints.
   - Arrow schema validation:
     - Adds Arrow schema validator utilities for checking compatibility with table schema definitions.
     - Adds validation coverage for supported Arrow field types, nested structures, and incompatible schema cases.
   
   ### Tests
   
   Not run. Local compile, CMake, and gtest environment checks are not part of this PR description.
   
   Test coverage included in this change:
   
   - `SchemaValidationTest`
   - `ArrowSchemaValidatorTest`
   
   ### API and Format
   
   No public API, storage format, or protocol changes.
   
   ### Documentation
   
   No documentation changes required.
   
   ### Generative AI tooling
   
   Migrate-by: Aone Copilot (GPT-5.5)


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(schema): add schema validation utilities [paimon-cpp]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #55:
URL: https://github.com/apache/paimon-cpp/pull/55#discussion_r3371373097


##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -0,0 +1,511 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <algorithm>
+#include <cassert>
+#include <map>
+#include <optional>
+#include <set>
+#include <sstream>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/preconditions.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/options/changelog_producer.h"
+#include "paimon/core/options/expire_config.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/defs.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) {
+    return (field->type()->id() == arrow::Type::TIMESTAMP ||
+            field->type()->id() == arrow::Type::DECIMAL || BlobUtils::IsBlobField(field));
+}
+
+Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
+    const auto& field_names = schema.FieldNames();
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.BucketKeys(), "bucket key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PartitionKeys(), "partition key"));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PartitionKeys()),
+                                  "Table column {} should include all partition fields {}",
+                                  field_names, schema.PartitionKeys()));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PrimaryKeys()),
+                                  "Table column {} should include all primary key constraint {}",
+                                  field_names, schema.PrimaryKeys()));
+
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition"));
+    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for
+    // now.
+    PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys()));
+
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options()));
+    PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options));
+    // PAIMON_RETURN_NOT_OK(ValidateDefaultValues(schema));
+    // PAIMON_RETURN_NOT_OK(ValidateStartupMode(options));
+    PAIMON_RETURN_NOT_OK(ValidateFieldsPrefix(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceField(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceGroup(schema, options));
+
+    ChangelogProducer changelog_producer = options.GetChangelogProducer();
+    if (schema.PrimaryKeys().empty() && changelog_producer != ChangelogProducer::NONE) {
+        return Status::Invalid(
+            fmt::format("Can not set {} on table without primary keys, please define primary keys.",
+                        Options::CHANGELOG_PRODUCER));
+    }
+    PAIMON_RETURN_NOT_OK(ValidateChangelogProducer(options));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() > 0,
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should be at least 1"));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() <=
+            options.GetExpireConfig().GetSnapshotRetainMax(),
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should not be larger than " +
+            std::string(Options::SNAPSHOT_NUM_RETAINED_MAX)));
+
+    // TODO(yonghao.fyh): check changelog num retain
+    // TODO(yonghao.fyh): support file format validate data fields
+    for (const auto& field_name : field_names) {
+        if (SpecialFields::IsSpecialFieldName(field_name)) {
+            return Status::Invalid(
+                fmt::format("field name '{}' in schema cannot be special field.", field_name));
+        }
+        if (StringUtils::StartsWith(field_name, SpecialFields::KEY_FIELD_PREFIX)) {
+            return Status::Invalid(fmt::format("field name '{}' in schema cannot start with '{}'.",
+                                               field_name, SpecialFields::KEY_FIELD_PREFIX));
+        }
+    }
+    // TODO(yonghao.fyh): check streaming read overwrite
+    // TODO(yonghao.fyh): check 'partition.expiration-time'
+    // TODO(yonghao.fyh): check 'rowkind.field'
+    if (options.DeletionVectorsEnabled()) {
+        PAIMON_RETURN_NOT_OK(ValidateForDeletionVectors(options));
+    }
+
+    PAIMON_RETURN_NOT_OK(ValidateRowTracking(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateBlobFields(schema, options));
+    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);
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        duplicate_field_names.empty(),
+        fmt::format("{} [{}] must not contain duplicate fields. Found: [{}]", error_message_intro,
+                    fmt::join(field_names, ", "), fmt::join(duplicate_field_names, ", "))));
+    return Status::OK();
+}
+
+Status SchemaValidation::ValidateOnlyContainPrimitiveType(
+    const std::vector<DataField>& fields, const std::vector<std::string>& field_names,
+    const std::string& error_message_intro) {
+    if (field_names.empty()) {
+        return Status::OK();
+    }
+    std::unordered_map<std::string, std::shared_ptr<arrow::DataType>> fields_map;
+    for (const auto& field : fields) {
+        fields_map[field.Name()] = field.Type();
+    }
+    for (const auto& field_name : field_names) {
+        auto it = fields_map.find(field_name);
+        if (it != fields_map.end()) {
+            auto data_type = it->second;
+            if (ArrowSchemaValidator::IsNestedType(data_type)) {
+                return Status::Invalid(fmt::format("The type {} in {} field {} is unsupported",
+                                                   data_type->ToString(), error_message_intro,
+                                                   it->first));
+            }
+        } else {
+            assert(false);
+            return Status::Invalid(
+                fmt::format("unexpected error, field {} not found in fields map", field_name));
+        }
+    }

Review Comment:
   The `assert(false)` makes this path abort in debug builds, which is risky even if earlier validations usually prevent it. Prefer returning a normal error without asserting (or use a `Preconditions::CheckState` that returns a `Status`) to avoid process termination on malformed input.



##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -0,0 +1,511 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <algorithm>
+#include <cassert>
+#include <map>
+#include <optional>
+#include <set>
+#include <sstream>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/preconditions.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/options/changelog_producer.h"
+#include "paimon/core/options/expire_config.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/defs.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) {
+    return (field->type()->id() == arrow::Type::TIMESTAMP ||
+            field->type()->id() == arrow::Type::DECIMAL || BlobUtils::IsBlobField(field));
+}
+
+Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
+    const auto& field_names = schema.FieldNames();
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.BucketKeys(), "bucket key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PartitionKeys(), "partition key"));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PartitionKeys()),
+                                  "Table column {} should include all partition fields {}",
+                                  field_names, schema.PartitionKeys()));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PrimaryKeys()),
+                                  "Table column {} should include all primary key constraint {}",
+                                  field_names, schema.PrimaryKeys()));
+
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition"));
+    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for
+    // now.
+    PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys()));
+
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options()));
+    PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options));
+    // PAIMON_RETURN_NOT_OK(ValidateDefaultValues(schema));
+    // PAIMON_RETURN_NOT_OK(ValidateStartupMode(options));
+    PAIMON_RETURN_NOT_OK(ValidateFieldsPrefix(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceField(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceGroup(schema, options));
+
+    ChangelogProducer changelog_producer = options.GetChangelogProducer();
+    if (schema.PrimaryKeys().empty() && changelog_producer != ChangelogProducer::NONE) {
+        return Status::Invalid(
+            fmt::format("Can not set {} on table without primary keys, please define primary keys.",
+                        Options::CHANGELOG_PRODUCER));
+    }
+    PAIMON_RETURN_NOT_OK(ValidateChangelogProducer(options));

Review Comment:
   The validations are internally inconsistent: `ValidateChangelogProducer` rejects any producer other than `NONE`, but `ValidateForDeletionVectors` explicitly claims DV supports `INPUT/LOOKUP`. As written, DV can never work with `INPUT/LOOKUP` because `ValidateChangelogProducer` fails first. Decide the intended behavior and make it consistent (e.g., if C++ truly doesn’t support any non-NONE producer yet, update `ValidateForDeletionVectors` to require NONE and adjust its error message; otherwise, skip/relax `ValidateChangelogProducer` when DV is enabled).



##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -0,0 +1,511 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <algorithm>
+#include <cassert>
+#include <map>
+#include <optional>
+#include <set>
+#include <sstream>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/preconditions.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/options/changelog_producer.h"
+#include "paimon/core/options/expire_config.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/defs.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) {
+    return (field->type()->id() == arrow::Type::TIMESTAMP ||
+            field->type()->id() == arrow::Type::DECIMAL || BlobUtils::IsBlobField(field));
+}
+
+Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
+    const auto& field_names = schema.FieldNames();
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.BucketKeys(), "bucket key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PartitionKeys(), "partition key"));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PartitionKeys()),
+                                  "Table column {} should include all partition fields {}",
+                                  field_names, schema.PartitionKeys()));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PrimaryKeys()),
+                                  "Table column {} should include all primary key constraint {}",
+                                  field_names, schema.PrimaryKeys()));
+
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition"));
+    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for
+    // now.
+    PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys()));
+
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options()));
+    PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options));
+    // PAIMON_RETURN_NOT_OK(ValidateDefaultValues(schema));
+    // PAIMON_RETURN_NOT_OK(ValidateStartupMode(options));
+    PAIMON_RETURN_NOT_OK(ValidateFieldsPrefix(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceField(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceGroup(schema, options));
+
+    ChangelogProducer changelog_producer = options.GetChangelogProducer();
+    if (schema.PrimaryKeys().empty() && changelog_producer != ChangelogProducer::NONE) {
+        return Status::Invalid(
+            fmt::format("Can not set {} on table without primary keys, please define primary keys.",
+                        Options::CHANGELOG_PRODUCER));
+    }
+    PAIMON_RETURN_NOT_OK(ValidateChangelogProducer(options));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() > 0,
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should be at least 1"));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() <=
+            options.GetExpireConfig().GetSnapshotRetainMax(),
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should not be larger than " +
+            std::string(Options::SNAPSHOT_NUM_RETAINED_MAX)));
+
+    // TODO(yonghao.fyh): check changelog num retain
+    // TODO(yonghao.fyh): support file format validate data fields
+    for (const auto& field_name : field_names) {
+        if (SpecialFields::IsSpecialFieldName(field_name)) {
+            return Status::Invalid(
+                fmt::format("field name '{}' in schema cannot be special field.", field_name));
+        }
+        if (StringUtils::StartsWith(field_name, SpecialFields::KEY_FIELD_PREFIX)) {
+            return Status::Invalid(fmt::format("field name '{}' in schema cannot start with '{}'.",
+                                               field_name, SpecialFields::KEY_FIELD_PREFIX));
+        }
+    }
+    // TODO(yonghao.fyh): check streaming read overwrite
+    // TODO(yonghao.fyh): check 'partition.expiration-time'
+    // TODO(yonghao.fyh): check 'rowkind.field'
+    if (options.DeletionVectorsEnabled()) {
+        PAIMON_RETURN_NOT_OK(ValidateForDeletionVectors(options));
+    }
+
+    PAIMON_RETURN_NOT_OK(ValidateRowTracking(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateBlobFields(schema, options));
+    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);
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        duplicate_field_names.empty(),
+        fmt::format("{} [{}] must not contain duplicate fields. Found: [{}]", error_message_intro,
+                    fmt::join(field_names, ", "), fmt::join(duplicate_field_names, ", "))));
+    return Status::OK();
+}
+
+Status SchemaValidation::ValidateOnlyContainPrimitiveType(
+    const std::vector<DataField>& fields, const std::vector<std::string>& field_names,
+    const std::string& error_message_intro) {
+    if (field_names.empty()) {
+        return Status::OK();
+    }
+    std::unordered_map<std::string, std::shared_ptr<arrow::DataType>> fields_map;
+    for (const auto& field : fields) {
+        fields_map[field.Name()] = field.Type();
+    }
+    for (const auto& field_name : field_names) {
+        auto it = fields_map.find(field_name);
+        if (it != fields_map.end()) {
+            auto data_type = it->second;
+            if (ArrowSchemaValidator::IsNestedType(data_type)) {
+                return Status::Invalid(fmt::format("The type {} in {} field {} is unsupported",
+                                                   data_type->ToString(), error_message_intro,
+                                                   it->first));
+            }
+        } else {
+            assert(false);
+            return Status::Invalid(
+                fmt::format("unexpected error, field {} not found in fields map", field_name));
+        }
+    }
+    return Status::OK();
+}
+
+Status SchemaValidation::ValidateNotContainComplexType(
+    const std::vector<DataField>& fields, const std::vector<std::string>& field_names) {
+    if (field_names.empty()) {
+        return Status::OK();
+    }
+    std::unordered_map<std::string, std::shared_ptr<arrow::Field>> fields_map;
+    for (const auto& field : fields) {
+        fields_map[field.Name()] = field.ArrowField();
+    }
+    for (const auto& field_name : field_names) {
+        auto it = fields_map.find(field_name);
+        if (it != fields_map.end()) {
+            auto field = it->second;
+            if (IsComplexType(field)) {
+                return Status::Invalid(
+                    fmt::format("The field {} in partition field {} is unsupported",
+                                field->ToString(), it->first));
+            }
+        } else {
+            assert(false);
+            return Status::Invalid(fmt::format(
+                "unexpected error, partition field {} not found in schema", field_name));
+        }
+    }
+    return Status::OK();
+}
+
+bool SchemaValidation::IsPostponeBucketTable(const TableSchema& schema, int32_t bucket) {
+    return !schema.PrimaryKeys().empty() && bucket == BucketModeDefine::POSTPONE_BUCKET;
+}
+
+Status SchemaValidation::ValidateBucket(const TableSchema& schema, const CoreOptions& options) {
+    int32_t bucket = options.GetBucket();
+    if (bucket == -1) {
+        if (options.ToMap().count(Options::BUCKET_KEY)) {
+            return Status::Invalid(
+                fmt::format("Cannot define '{}' with bucket -1, please specify a bucket number.",
+                            Options::BUCKET_KEY));
+        }
+        if (schema.PrimaryKeys().empty() &&
+            options.ToMap().count("full-compaction.delta-commits")) {
+            return Status::Invalid(
+                "AppendOnlyTable of unware or dynamic bucket does not support "
+                "'full-compaction.delta-commits'");
+        }

Review Comment:
   Correct typo in user-facing error message: 'unware' → 'unaware'.



##########
src/paimon/core/schema/schema_validation_test.cpp:
##########
@@ -0,0 +1,811 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <map>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/defs.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(SchemaValidationTest, TestSimple) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestRowTracking) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestWithBlobField) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    std::shared_ptr<arrow::Field> f3 = BlobUtils::ToArrowField("f3", false);
+    std::shared_ptr<arrow::Field> f4 = BlobUtils::ToArrowField("f4", false);
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f4"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_VIEW_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f0' in 'blob-descriptor-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_VIEW_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f3' in 'blob-view-field' can not also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f4' in 'blob-external-storage-field' must also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "'blob-external-storage-path' must be set when "
+                            "'blob-external-storage-field' is configured.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "false"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Data evolution config must be enabled for table with BLOB type column.");
+    }
+    {
+        arrow::FieldVector fields = {f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Table with BLOB type column must have other normal columns.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "non-exist"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Get field non-exist failed: not exist in table schema");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Field 'f0' in 'blob-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f3"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateRowTracking(*table_schema, core_options),
+                            "Blob field f3 cannot be a partition key.");
+    }
+}
+
+TEST(SchemaValidationTest, TestDuplicateField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // duplicate primary keys
+        std::vector<std::string> dup_primary_keys = {"f0", "f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 dup_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "primary key [f0, f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate partition keys
+        std::vector<std::string> dup_partition_keys = {"f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, dup_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "partition key [f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate bucket keys
+        std::map<std::string, std::string> dup_options = {{Options::BUCKET, "2"},
+                                                          {Options::BUCKET_KEY, "f0,f0"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 primary_keys, dup_options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "bucket key [f0, f0] must not contain duplicate fields. Found: [f0]");
+    }
+}
+
+TEST(SchemaValidationTest, TestNonExistField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // non-exist primary keys
+        std::vector<std::string> non_exist_primary_keys = {"f0", "f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 non_exist_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all primary key constraint ["f0", "f1", "non-exist"])");
+    }
+    {
+        // non-exist partition keys
+        std::vector<std::string> non_exist_partition_keys = {"f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, non_exist_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all partition fields ["f1", "non-exist"])");
+    }
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyList) {
+    auto value_field = arrow::field("values", arrow::int32());
+    auto f0 = arrow::field("f0", arrow::list(value_field));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyMap) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePartitionKeyStruct) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto child1 = arrow::field("inner1", arrow::int32());
+    auto child2 = arrow::field("inner2", arrow::float64());
+    auto f1 = arrow::field("f1", arrow::struct_({child1, child2}));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::decimal128(5, 2));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = BlobUtils::ToArrowField("f1");
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, /*primary_keys=*/{}, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestDateTypePartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::date32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, ValidateFieldsPrefix) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f3", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "f3 can not be found in table schema.");
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f2", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"},
+            {Options::BUCKET_KEY, "f0"},
+            {Options::FIELDS_DEFAULT_AGG_FUNC, "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.", "f1"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "invalid options key fields.");
+    }
+}
+
+TEST(SchemaValidationTest, ValidateBucket) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    {
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::BUCKET_KEY, "f0"}};

Review Comment:
   This test expects an error message that doesn’t match the implementation. `ValidateBucket` currently returns: `Cannot define 'bucket-key' with bucket -1, please specify a bucket number.` (and includes the option key). With `ASSERT_NOK_WITH_MSG` doing exact-match, this will fail. Update the expected string to match the implementation (or adjust the production message if the shorter text is the intended standard).



##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -0,0 +1,511 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <algorithm>
+#include <cassert>
+#include <map>
+#include <optional>
+#include <set>
+#include <sstream>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/preconditions.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/options/changelog_producer.h"
+#include "paimon/core/options/expire_config.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/defs.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) {
+    return (field->type()->id() == arrow::Type::TIMESTAMP ||
+            field->type()->id() == arrow::Type::DECIMAL || BlobUtils::IsBlobField(field));
+}
+
+Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
+    const auto& field_names = schema.FieldNames();
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.BucketKeys(), "bucket key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PartitionKeys(), "partition key"));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PartitionKeys()),
+                                  "Table column {} should include all partition fields {}",
+                                  field_names, schema.PartitionKeys()));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PrimaryKeys()),
+                                  "Table column {} should include all primary key constraint {}",
+                                  field_names, schema.PrimaryKeys()));
+
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition"));
+    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for
+    // now.
+    PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys()));
+
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options()));
+    PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options));
+    // PAIMON_RETURN_NOT_OK(ValidateDefaultValues(schema));
+    // PAIMON_RETURN_NOT_OK(ValidateStartupMode(options));
+    PAIMON_RETURN_NOT_OK(ValidateFieldsPrefix(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceField(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceGroup(schema, options));
+
+    ChangelogProducer changelog_producer = options.GetChangelogProducer();
+    if (schema.PrimaryKeys().empty() && changelog_producer != ChangelogProducer::NONE) {
+        return Status::Invalid(
+            fmt::format("Can not set {} on table without primary keys, please define primary keys.",
+                        Options::CHANGELOG_PRODUCER));
+    }
+    PAIMON_RETURN_NOT_OK(ValidateChangelogProducer(options));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() > 0,
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should be at least 1"));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() <=
+            options.GetExpireConfig().GetSnapshotRetainMax(),
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should not be larger than " +
+            std::string(Options::SNAPSHOT_NUM_RETAINED_MAX)));
+
+    // TODO(yonghao.fyh): check changelog num retain
+    // TODO(yonghao.fyh): support file format validate data fields
+    for (const auto& field_name : field_names) {
+        if (SpecialFields::IsSpecialFieldName(field_name)) {
+            return Status::Invalid(
+                fmt::format("field name '{}' in schema cannot be special field.", field_name));
+        }
+        if (StringUtils::StartsWith(field_name, SpecialFields::KEY_FIELD_PREFIX)) {
+            return Status::Invalid(fmt::format("field name '{}' in schema cannot start with '{}'.",
+                                               field_name, SpecialFields::KEY_FIELD_PREFIX));
+        }
+    }
+    // TODO(yonghao.fyh): check streaming read overwrite
+    // TODO(yonghao.fyh): check 'partition.expiration-time'
+    // TODO(yonghao.fyh): check 'rowkind.field'
+    if (options.DeletionVectorsEnabled()) {
+        PAIMON_RETURN_NOT_OK(ValidateForDeletionVectors(options));
+    }

Review Comment:
   The validations are internally inconsistent: `ValidateChangelogProducer` rejects any producer other than `NONE`, but `ValidateForDeletionVectors` explicitly claims DV supports `INPUT/LOOKUP`. As written, DV can never work with `INPUT/LOOKUP` because `ValidateChangelogProducer` fails first. Decide the intended behavior and make it consistent (e.g., if C++ truly doesn’t support any non-NONE producer yet, update `ValidateForDeletionVectors` to require NONE and adjust its error message; otherwise, skip/relax `ValidateChangelogProducer` when DV is enabled).



##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -0,0 +1,511 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <algorithm>
+#include <cassert>
+#include <map>
+#include <optional>
+#include <set>
+#include <sstream>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/preconditions.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/options/changelog_producer.h"
+#include "paimon/core/options/expire_config.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/defs.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) {
+    return (field->type()->id() == arrow::Type::TIMESTAMP ||
+            field->type()->id() == arrow::Type::DECIMAL || BlobUtils::IsBlobField(field));
+}

Review Comment:
   `arrow::Type::DECIMAL` is not the standard Arrow enum value used elsewhere in this PR (e.g., `DECIMAL128`). This can fail to compile depending on the Arrow version. Use `arrow::Type::DECIMAL128` (and likely also treat `DECIMAL256` as complex/unsupported for partition keys if applicable) instead of `arrow::Type::DECIMAL`.



##########
src/paimon/core/schema/schema_validation.cpp:
##########
@@ -0,0 +1,511 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <algorithm>
+#include <cassert>
+#include <map>
+#include <optional>
+#include <set>
+#include <sstream>
+#include <unordered_map>
+#include <unordered_set>
+#include <utility>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/types/data_field.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/common/utils/preconditions.h"
+#include "paimon/common/utils/string_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/options/changelog_producer.h"
+#include "paimon/core/options/expire_config.h"
+#include "paimon/core/options/merge_engine.h"
+#include "paimon/core/schema/arrow_schema_validator.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/core/table/bucket_mode.h"
+#include "paimon/defs.h"
+#include "paimon/result.h"
+
+namespace paimon {
+
+bool SchemaValidation::IsComplexType(const std::shared_ptr<arrow::Field>& field) {
+    return (field->type()->id() == arrow::Type::TIMESTAMP ||
+            field->type()->id() == arrow::Type::DECIMAL || BlobUtils::IsBlobField(field));
+}
+
+Status SchemaValidation::ValidateTableSchema(const TableSchema& schema) {
+    const auto& field_names = schema.FieldNames();
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.BucketKeys(), "bucket key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(ValidateNoDuplicateField(schema.PartitionKeys(), "partition key"));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PartitionKeys()),
+                                  "Table column {} should include all partition fields {}",
+                                  field_names, schema.PartitionKeys()));
+    PAIMON_RETURN_NOT_OK(
+        Preconditions::CheckState(ObjectUtils::ContainsAll(field_names, schema.PrimaryKeys()),
+                                  "Table column {} should include all primary key constraint {}",
+                                  field_names, schema.PrimaryKeys()));
+
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PrimaryKeys(), "primary key"));
+    PAIMON_RETURN_NOT_OK(
+        ValidateOnlyContainPrimitiveType(schema.Fields(), schema.PartitionKeys(), "partition"));
+    // TODO(lisizhuo.lsz): C++ Paimon do not support timestamp & decimal type in partition keys for
+    // now.
+    PAIMON_RETURN_NOT_OK(ValidateNotContainComplexType(schema.Fields(), schema.PartitionKeys()));
+
+    PAIMON_ASSIGN_OR_RAISE(CoreOptions options, CoreOptions::FromMap(schema.Options()));
+    PAIMON_RETURN_NOT_OK(ValidateBucket(schema, options));
+    // PAIMON_RETURN_NOT_OK(ValidateDefaultValues(schema));
+    // PAIMON_RETURN_NOT_OK(ValidateStartupMode(options));
+    PAIMON_RETURN_NOT_OK(ValidateFieldsPrefix(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceField(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateSequenceGroup(schema, options));
+
+    ChangelogProducer changelog_producer = options.GetChangelogProducer();
+    if (schema.PrimaryKeys().empty() && changelog_producer != ChangelogProducer::NONE) {
+        return Status::Invalid(
+            fmt::format("Can not set {} on table without primary keys, please define primary keys.",
+                        Options::CHANGELOG_PRODUCER));
+    }
+    PAIMON_RETURN_NOT_OK(ValidateChangelogProducer(options));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() > 0,
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should be at least 1"));
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetExpireConfig().GetSnapshotRetainMin() <=
+            options.GetExpireConfig().GetSnapshotRetainMax(),
+        std::string(Options::SNAPSHOT_NUM_RETAINED_MIN) + " should not be larger than " +
+            std::string(Options::SNAPSHOT_NUM_RETAINED_MAX)));
+
+    // TODO(yonghao.fyh): check changelog num retain
+    // TODO(yonghao.fyh): support file format validate data fields
+    for (const auto& field_name : field_names) {
+        if (SpecialFields::IsSpecialFieldName(field_name)) {
+            return Status::Invalid(
+                fmt::format("field name '{}' in schema cannot be special field.", field_name));
+        }
+        if (StringUtils::StartsWith(field_name, SpecialFields::KEY_FIELD_PREFIX)) {
+            return Status::Invalid(fmt::format("field name '{}' in schema cannot start with '{}'.",
+                                               field_name, SpecialFields::KEY_FIELD_PREFIX));
+        }
+    }
+    // TODO(yonghao.fyh): check streaming read overwrite
+    // TODO(yonghao.fyh): check 'partition.expiration-time'
+    // TODO(yonghao.fyh): check 'rowkind.field'
+    if (options.DeletionVectorsEnabled()) {
+        PAIMON_RETURN_NOT_OK(ValidateForDeletionVectors(options));
+    }
+
+    PAIMON_RETURN_NOT_OK(ValidateRowTracking(schema, options));
+    PAIMON_RETURN_NOT_OK(ValidateBlobFields(schema, options));
+    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);
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        duplicate_field_names.empty(),
+        fmt::format("{} [{}] must not contain duplicate fields. Found: [{}]", error_message_intro,
+                    fmt::join(field_names, ", "), fmt::join(duplicate_field_names, ", "))));
+    return Status::OK();
+}
+
+Status SchemaValidation::ValidateOnlyContainPrimitiveType(
+    const std::vector<DataField>& fields, const std::vector<std::string>& field_names,
+    const std::string& error_message_intro) {
+    if (field_names.empty()) {
+        return Status::OK();
+    }
+    std::unordered_map<std::string, std::shared_ptr<arrow::DataType>> fields_map;
+    for (const auto& field : fields) {
+        fields_map[field.Name()] = field.Type();
+    }
+    for (const auto& field_name : field_names) {
+        auto it = fields_map.find(field_name);
+        if (it != fields_map.end()) {
+            auto data_type = it->second;
+            if (ArrowSchemaValidator::IsNestedType(data_type)) {
+                return Status::Invalid(fmt::format("The type {} in {} field {} is unsupported",
+                                                   data_type->ToString(), error_message_intro,
+                                                   it->first));
+            }
+        } else {
+            assert(false);
+            return Status::Invalid(
+                fmt::format("unexpected error, field {} not found in fields map", field_name));
+        }
+    }
+    return Status::OK();
+}
+
+Status SchemaValidation::ValidateNotContainComplexType(
+    const std::vector<DataField>& fields, const std::vector<std::string>& field_names) {
+    if (field_names.empty()) {
+        return Status::OK();
+    }
+    std::unordered_map<std::string, std::shared_ptr<arrow::Field>> fields_map;
+    for (const auto& field : fields) {
+        fields_map[field.Name()] = field.ArrowField();
+    }
+    for (const auto& field_name : field_names) {
+        auto it = fields_map.find(field_name);
+        if (it != fields_map.end()) {
+            auto field = it->second;
+            if (IsComplexType(field)) {
+                return Status::Invalid(
+                    fmt::format("The field {} in partition field {} is unsupported",
+                                field->ToString(), it->first));
+            }
+        } else {
+            assert(false);
+            return Status::Invalid(fmt::format(
+                "unexpected error, partition field {} not found in schema", field_name));
+        }
+    }
+    return Status::OK();
+}
+
+bool SchemaValidation::IsPostponeBucketTable(const TableSchema& schema, int32_t bucket) {
+    return !schema.PrimaryKeys().empty() && bucket == BucketModeDefine::POSTPONE_BUCKET;
+}
+
+Status SchemaValidation::ValidateBucket(const TableSchema& schema, const CoreOptions& options) {
+    int32_t bucket = options.GetBucket();
+    if (bucket == -1) {
+        if (options.ToMap().count(Options::BUCKET_KEY)) {
+            return Status::Invalid(
+                fmt::format("Cannot define '{}' with bucket -1, please specify a bucket number.",
+                            Options::BUCKET_KEY));
+        }
+        if (schema.PrimaryKeys().empty() &&
+            options.ToMap().count("full-compaction.delta-commits")) {
+            return Status::Invalid(
+                "AppendOnlyTable of unware or dynamic bucket does not support "
+                "'full-compaction.delta-commits'");
+        }
+    } else if (bucket < 1 && !IsPostponeBucketTable(schema, bucket)) {
+        return Status::Invalid("The number of buckets needs to be greater than 0.");
+    } else {
+        if (schema.CrossPartitionUpdate()) {
+            return Status::Invalid(fmt::format(
+                "You should use dynamic bucket (bucket = -1) mode in cross partition update case "
+                "(Primary key constraint '{}' not include all partition fields '{}').",
+                fmt::join(schema.PrimaryKeys(), ", "), fmt::join(schema.PartitionKeys(), ", ")));
+        }
+        if (schema.PrimaryKeys().empty() && schema.BucketKeys().empty()) {
+            return Status::Invalid("You should define a 'bucket-key' for bucketed append mode.");
+        }
+        if (!schema.BucketKeys().empty()) {
+            std::vector<std::string> bucket_keys = schema.BucketKeys();
+            std::vector<std::string> nested_fields;
+
+            for (const auto& field : schema.Fields()) {
+                if (std::find(bucket_keys.begin(), bucket_keys.end(), field.Name()) !=
+                        bucket_keys.end() &&
+                    ArrowSchemaValidator::IsNestedType(field.Type())) {
+                    nested_fields.push_back(field.Name());
+                }
+            }
+
+            if (!nested_fields.empty()) {
+                return Status::Invalid(fmt::format(
+                    "Nested type cannot be in bucket-key, in your table these keys are: {}",
+                    fmt::join(nested_fields, ", ")));
+            }
+        }
+    }
+    return Status::OK();
+}
+
+Status SchemaValidation::ValidateChangelogProducer(const CoreOptions& options) {
+    return Preconditions::CheckState(options.GetChangelogProducer() == ChangelogProducer::NONE,
+                                     "C++ Paimon does not support changelog-producer yet. Please "
+                                     "keep changelog-producer as 'none'.");
+}
+
+Status SchemaValidation::ValidateForDeletionVectors(const CoreOptions& options) {
+    PAIMON_RETURN_NOT_OK(Preconditions::CheckState(
+        options.GetChangelogProducer() == ChangelogProducer::NONE ||
+            options.GetChangelogProducer() == ChangelogProducer::INPUT ||
+            options.GetChangelogProducer() == ChangelogProducer::LOOKUP,
+        "Deletion vectors mode is only supported for NONE/INPUT/LOOKUP changelog producer now."));

Review Comment:
   The validations are internally inconsistent: `ValidateChangelogProducer` rejects any producer other than `NONE`, but `ValidateForDeletionVectors` explicitly claims DV supports `INPUT/LOOKUP`. As written, DV can never work with `INPUT/LOOKUP` because `ValidateChangelogProducer` fails first. Decide the intended behavior and make it consistent (e.g., if C++ truly doesn’t support any non-NONE producer yet, update `ValidateForDeletionVectors` to require NONE and adjust its error message; otherwise, skip/relax `ValidateChangelogProducer` when DV is enabled).



##########
src/paimon/core/schema/schema_validation_test.cpp:
##########
@@ -0,0 +1,811 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <map>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/defs.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(SchemaValidationTest, TestSimple) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestRowTracking) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestWithBlobField) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    std::shared_ptr<arrow::Field> f3 = BlobUtils::ToArrowField("f3", false);
+    std::shared_ptr<arrow::Field> f4 = BlobUtils::ToArrowField("f4", false);
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f4"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_VIEW_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f0' in 'blob-descriptor-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_VIEW_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f3' in 'blob-view-field' can not also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f4' in 'blob-external-storage-field' must also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "'blob-external-storage-path' must be set when "
+                            "'blob-external-storage-field' is configured.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "false"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Data evolution config must be enabled for table with BLOB type column.");
+    }
+    {
+        arrow::FieldVector fields = {f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Table with BLOB type column must have other normal columns.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "non-exist"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Get field non-exist failed: not exist in table schema");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Field 'f0' in 'blob-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f3"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateRowTracking(*table_schema, core_options),
+                            "Blob field f3 cannot be a partition key.");
+    }
+}
+
+TEST(SchemaValidationTest, TestDuplicateField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // duplicate primary keys
+        std::vector<std::string> dup_primary_keys = {"f0", "f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 dup_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "primary key [f0, f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate partition keys
+        std::vector<std::string> dup_partition_keys = {"f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, dup_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "partition key [f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate bucket keys
+        std::map<std::string, std::string> dup_options = {{Options::BUCKET, "2"},
+                                                          {Options::BUCKET_KEY, "f0,f0"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 primary_keys, dup_options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "bucket key [f0, f0] must not contain duplicate fields. Found: [f0]");
+    }
+}
+
+TEST(SchemaValidationTest, TestNonExistField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // non-exist primary keys
+        std::vector<std::string> non_exist_primary_keys = {"f0", "f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 non_exist_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all primary key constraint ["f0", "f1", "non-exist"])");
+    }
+    {
+        // non-exist partition keys
+        std::vector<std::string> non_exist_partition_keys = {"f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, non_exist_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all partition fields ["f1", "non-exist"])");
+    }
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyList) {
+    auto value_field = arrow::field("values", arrow::int32());
+    auto f0 = arrow::field("f0", arrow::list(value_field));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyMap) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePartitionKeyStruct) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto child1 = arrow::field("inner1", arrow::int32());
+    auto child2 = arrow::field("inner2", arrow::float64());
+    auto f1 = arrow::field("f1", arrow::struct_({child1, child2}));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::decimal128(5, 2));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = BlobUtils::ToArrowField("f1");
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, /*primary_keys=*/{}, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestDateTypePartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::date32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, ValidateFieldsPrefix) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f3", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "f3 can not be found in table schema.");
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f2", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"},
+            {Options::BUCKET_KEY, "f0"},
+            {Options::FIELDS_DEFAULT_AGG_FUNC, "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.", "f1"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "invalid options key fields.");
+    }
+}
+
+TEST(SchemaValidationTest, ValidateBucket) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    {
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::BUCKET_KEY, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "please specify a bucket number.");

Review Comment:
   This test expects an error message that doesn’t match the implementation. `ValidateBucket` currently returns: `Cannot define 'bucket-key' with bucket -1, please specify a bucket number.` (and includes the option key). With `ASSERT_NOK_WITH_MSG` doing exact-match, this will fail. Update the expected string to match the implementation (or adjust the production message if the shorter text is the intended standard).



##########
src/paimon/core/schema/schema_validation_test.cpp:
##########
@@ -0,0 +1,811 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <map>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/defs.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(SchemaValidationTest, TestSimple) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestRowTracking) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestWithBlobField) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    std::shared_ptr<arrow::Field> f3 = BlobUtils::ToArrowField("f3", false);
+    std::shared_ptr<arrow::Field> f4 = BlobUtils::ToArrowField("f4", false);
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f4"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_VIEW_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f0' in 'blob-descriptor-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_VIEW_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f3' in 'blob-view-field' can not also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f4' in 'blob-external-storage-field' must also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "'blob-external-storage-path' must be set when "
+                            "'blob-external-storage-field' is configured.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "false"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Data evolution config must be enabled for table with BLOB type column.");
+    }
+    {
+        arrow::FieldVector fields = {f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Table with BLOB type column must have other normal columns.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "non-exist"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Get field non-exist failed: not exist in table schema");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Field 'f0' in 'blob-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f3"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateRowTracking(*table_schema, core_options),
+                            "Blob field f3 cannot be a partition key.");
+    }
+}
+
+TEST(SchemaValidationTest, TestDuplicateField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // duplicate primary keys
+        std::vector<std::string> dup_primary_keys = {"f0", "f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 dup_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "primary key [f0, f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate partition keys
+        std::vector<std::string> dup_partition_keys = {"f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, dup_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "partition key [f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate bucket keys
+        std::map<std::string, std::string> dup_options = {{Options::BUCKET, "2"},
+                                                          {Options::BUCKET_KEY, "f0,f0"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 primary_keys, dup_options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "bucket key [f0, f0] must not contain duplicate fields. Found: [f0]");
+    }
+}
+
+TEST(SchemaValidationTest, TestNonExistField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // non-exist primary keys
+        std::vector<std::string> non_exist_primary_keys = {"f0", "f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 non_exist_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all primary key constraint ["f0", "f1", "non-exist"])");
+    }
+    {
+        // non-exist partition keys
+        std::vector<std::string> non_exist_partition_keys = {"f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, non_exist_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all partition fields ["f1", "non-exist"])");
+    }
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyList) {
+    auto value_field = arrow::field("values", arrow::int32());
+    auto f0 = arrow::field("f0", arrow::list(value_field));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyMap) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePartitionKeyStruct) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto child1 = arrow::field("inner1", arrow::int32());
+    auto child2 = arrow::field("inner2", arrow::float64());
+    auto f1 = arrow::field("f1", arrow::struct_({child1, child2}));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::decimal128(5, 2));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = BlobUtils::ToArrowField("f1");
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, /*primary_keys=*/{}, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestDateTypePartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::date32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, ValidateFieldsPrefix) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f3", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "f3 can not be found in table schema.");

Review Comment:
   The implementation in `ValidateFieldsPrefix` emits `Field f3 can not be found in table schema.` (note the leading `Field`). With exact-match, this assertion will fail. Align the expected error string with the implementation (or standardize the production error messages so tests can be consistent across validation paths).



##########
src/paimon/core/schema/schema_validation_test.cpp:
##########
@@ -0,0 +1,811 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <map>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/defs.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(SchemaValidationTest, TestSimple) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestRowTracking) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestWithBlobField) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    std::shared_ptr<arrow::Field> f3 = BlobUtils::ToArrowField("f3", false);
+    std::shared_ptr<arrow::Field> f4 = BlobUtils::ToArrowField("f4", false);
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f4"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_VIEW_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f0' in 'blob-descriptor-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_VIEW_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f3' in 'blob-view-field' can not also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f4' in 'blob-external-storage-field' must also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "'blob-external-storage-path' must be set when "
+                            "'blob-external-storage-field' is configured.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "false"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Data evolution config must be enabled for table with BLOB type column.");
+    }
+    {
+        arrow::FieldVector fields = {f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Table with BLOB type column must have other normal columns.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "non-exist"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Get field non-exist failed: not exist in table schema");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Field 'f0' in 'blob-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f3"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateRowTracking(*table_schema, core_options),
+                            "Blob field f3 cannot be a partition key.");
+    }
+}
+
+TEST(SchemaValidationTest, TestDuplicateField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // duplicate primary keys
+        std::vector<std::string> dup_primary_keys = {"f0", "f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 dup_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "primary key [f0, f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate partition keys
+        std::vector<std::string> dup_partition_keys = {"f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, dup_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "partition key [f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate bucket keys
+        std::map<std::string, std::string> dup_options = {{Options::BUCKET, "2"},
+                                                          {Options::BUCKET_KEY, "f0,f0"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 primary_keys, dup_options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "bucket key [f0, f0] must not contain duplicate fields. Found: [f0]");
+    }
+}
+
+TEST(SchemaValidationTest, TestNonExistField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // non-exist primary keys
+        std::vector<std::string> non_exist_primary_keys = {"f0", "f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 non_exist_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all primary key constraint ["f0", "f1", "non-exist"])");
+    }
+    {
+        // non-exist partition keys
+        std::vector<std::string> non_exist_partition_keys = {"f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, non_exist_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all partition fields ["f1", "non-exist"])");
+    }
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyList) {
+    auto value_field = arrow::field("values", arrow::int32());
+    auto f0 = arrow::field("f0", arrow::list(value_field));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");

Review Comment:
   This expected message doesn’t match `ValidateOnlyContainPrimitiveType`, which currently formats errors as `The type <...> in <intro> field <name> is unsupported`. With exact-match assertions, this test will fail. Update the expected message to the current formatted output (or change production formatting to the simpler wording if that’s required for compatibility).



##########
src/paimon/core/schema/schema_validation_test.cpp:
##########
@@ -0,0 +1,811 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "paimon/core/schema/schema_validation.h"
+
+#include <map>
+
+#include "arrow/api.h"
+#include "gtest/gtest.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/core/schema/table_schema.h"
+#include "paimon/defs.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(SchemaValidationTest, TestSimple) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestRowTracking) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {
+        {Options::BUCKET, "-1"},
+        {Options::ROW_TRACKING_ENABLED, "true"},
+        {Options::DATA_EVOLUTION_ENABLED, "true"},
+    };
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, TestWithBlobField) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    std::shared_ptr<arrow::Field> f3 = BlobUtils::ToArrowField("f3", false);
+    std::shared_ptr<arrow::Field> f4 = BlobUtils::ToArrowField("f4", false);
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f4"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_VIEW_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f0' in 'blob-descriptor-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_VIEW_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f3' in 'blob-view-field' can not also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3, f4};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "-1"},
+            {Options::ROW_TRACKING_ENABLED, "true"},
+            {Options::DATA_EVOLUTION_ENABLED, "true"},
+            {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+            {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f4"},
+            {Options::BLOB_EXTERNAL_STORAGE_PATH, "FILE:///tmp/blob_external_storage/"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Field 'f4' in 'blob-external-storage-field' must also be in 'blob-descriptor-field'.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_DESCRIPTOR_FIELD, "f3"},
+                                                      {Options::BLOB_EXTERNAL_STORAGE_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "'blob-external-storage-path' must be set when "
+                            "'blob-external-storage-field' is configured.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "false"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Data evolution config must be enabled for table with BLOB type column.");
+    }
+    {
+        arrow::FieldVector fields = {f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Table with BLOB type column must have other normal columns.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "non-exist"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Get field non-exist failed: not exist in table schema");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3,f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "Field 'f0' in 'blob-field' must be a BLOB field in table schema.");
+    }
+    {
+        arrow::FieldVector fields = {f0, f1, f2, f3};
+        auto schema = arrow::schema(fields);
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f3"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::ROW_TRACKING_ENABLED, "true"},
+                                                      {Options::DATA_EVOLUTION_ENABLED, "true"},
+                                                      {Options::BLOB_FIELD, "f3"}};
+        ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateRowTracking(*table_schema, core_options),
+                            "Blob field f3 cannot be a partition key.");
+    }
+}
+
+TEST(SchemaValidationTest, TestDuplicateField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // duplicate primary keys
+        std::vector<std::string> dup_primary_keys = {"f0", "f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 dup_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "primary key [f0, f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate partition keys
+        std::vector<std::string> dup_partition_keys = {"f1", "f1"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, dup_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "partition key [f1, f1] must not contain duplicate fields. Found: [f1]");
+    }
+    {
+        // duplicate bucket keys
+        std::map<std::string, std::string> dup_options = {{Options::BUCKET, "2"},
+                                                          {Options::BUCKET_KEY, "f0,f0"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 primary_keys, dup_options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "bucket key [f0, f0] must not contain duplicate fields. Found: [f0]");
+    }
+}
+
+TEST(SchemaValidationTest, TestNonExistField) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    {
+        // non-exist primary keys
+        std::vector<std::string> non_exist_primary_keys = {"f0", "f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 non_exist_primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all primary key constraint ["f0", "f1", "non-exist"])");
+    }
+    {
+        // non-exist partition keys
+        std::vector<std::string> non_exist_partition_keys = {"f1", "non-exist"};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, non_exist_partition_keys,
+                                                 primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            R"(Table column ["f0", "f1", "f2"] should include all partition fields ["f1", "non-exist"])");
+    }
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyList) {
+    auto value_field = arrow::field("values", arrow::int32());
+    auto f0 = arrow::field("f0", arrow::list(value_field));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePrimaryKeyMap) {
+    auto f0 = arrow::field("f0", arrow::map(arrow::utf8(), arrow::int32()));
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f0 is unsupported");
+}
+
+TEST(SchemaValidationTest, NonPrimitivePartitionKeyStruct) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto child1 = arrow::field("inner1", arrow::int32());
+    auto child2 = arrow::field("inner2", arrow::float64());
+    auto f1 = arrow::field("f1", arrow::struct_({child1, child2}));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                  {Options::BUCKET_KEY, "f0"}};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::decimal128(5, 2));
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestComplexPartitionKeyWithBlob) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = BlobUtils::ToArrowField("f1");
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, /*primary_keys=*/{}, {}));
+    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                        "partition field f1 is unsupported");
+}
+
+TEST(SchemaValidationTest, TestDateTypePartitionKey) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::date32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    ASSERT_OK_AND_ASSIGN(
+        std::shared_ptr<TableSchema> table_schema,
+        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, {}));
+    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+}
+
+TEST(SchemaValidationTest, ValidateFieldsPrefix) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f3", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "f3 can not be found in table schema.");
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.f0,f1,f2", "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"},
+            {Options::BUCKET_KEY, "f0"},
+            {Options::FIELDS_DEFAULT_AGG_FUNC, "some_value"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
+    }
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"}, {Options::BUCKET_KEY, "f0"}, {"fields.", "f1"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "invalid options key fields.");
+    }
+}
+
+TEST(SchemaValidationTest, ValidateBucket) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    {
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
+                                                      {Options::BUCKET_KEY, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "please specify a bucket number.");
+    }
+    {
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f1"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "0"},
+                                                      {Options::BUCKET_KEY, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "The number of buckets needs to be greater than 0.");
+    }
+    {
+        std::vector<std::string> primary_keys = {"f0", "f1"};
+        std::vector<std::string> partition_keys = {"f2"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                      {Options::BUCKET_KEY, "f0"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "You should use dynamic bucket (bucket = -1) mode in cross partition update case");
+    }
+    {
+        std::vector<std::string> primary_keys = {};
+        std::vector<std::string> partition_keys = {"f2"};
+        std::map<std::string, std::string> options = {{Options::BUCKET, "2"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "You should define a 'bucket-key' for bucketed append mode");
+    }
+    {
+        std::vector<std::string> partition_keys = {"f2"};
+        std::map<std::string, std::string> options = {{"full-compaction.delta-commits", "2"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, schema, partition_keys,
+                                                 /*primary_keys=*/{}, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "AppendOnlyTable of unware or dynamic bucket does not support "
+                            "'full-compaction.delta-commits");
+    }
+    {
+        auto f3 = arrow::field("f3", arrow::map(arrow::utf8(), arrow::int32()));
+        arrow::FieldVector new_fields = {f0, f1, f2, f3};
+        auto new_schema = arrow::schema(new_fields);
+        std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
+                                                      {Options::BUCKET_KEY, "f3"}};
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
+                             TableSchema::Create(/*schema_id=*/0, new_schema, /*partition_keys=*/{},
+                                                 /*primary_keys=*/{}, options));
+        ASSERT_NOK_WITH_MSG(
+            SchemaValidation::ValidateTableSchema(*table_schema),
+            "Nested type cannot be in bucket-key, in your table these keys are: f3");
+    }
+}
+
+TEST(SchemaValidationTest, ValidateDeletionVector) {
+    auto f0 = arrow::field("f0", arrow::utf8());
+    auto f1 = arrow::field("f1", arrow::int32());
+    auto f2 = arrow::field("f2", arrow::float64());
+    arrow::FieldVector fields = {f0, f1, f2};
+    auto schema = arrow::schema(fields);
+    std::vector<std::string> primary_keys = {"f0", "f1"};
+    std::vector<std::string> partition_keys = {"f1"};
+    {
+        std::map<std::string, std::string> options = {
+            {Options::BUCKET, "2"},
+            {Options::BUCKET_KEY, "f0"},
+            {Options::DELETION_VECTORS_ENABLED, "true"},
+            {Options::CHANGELOG_PRODUCER, "full-compaction"}};
+        ASSERT_OK_AND_ASSIGN(
+            std::shared_ptr<TableSchema> table_schema,
+            TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
+        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
+                            "C++ Paimon does not support changelog-producer yet");

Review Comment:
   `ValidateChangelogProducer` currently returns `C++ Paimon does not support changelog-producer yet. Please keep changelog-producer as 'none'.` (with additional guidance and punctuation). This exact-match test will fail unless the expected message is updated (or the production message is changed back to the shorter form).



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] feat(schema): add schema validation utilities [paimon-cpp]

["leaves12138 (via GitHub)" <gi...@apache.org>]

leaves12138 merged PR #55:
URL: https://github.com/apache/paimon-cpp/pull/55


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org