src/paimon/core/schema/schema_validation_test.cpp

/*
 * 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, TestVectorType) {
    auto vector_field = arrow::field("embedding", arrow::fixed_size_list(arrow::float32(), 3));
    auto schema = arrow::schema({arrow::field("id", arrow::int64()), vector_field});
    std::map<std::string, std::string> parquet_options = {{Options::BUCKET, "-1"},
                                                          {Options::FILE_FORMAT, "parquet"}};
    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                             /*primary_keys=*/{}, parquet_options));
    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));

    parquet_options[Options::FILE_FORMAT] = "PARQUET";
    ASSERT_OK_AND_ASSIGN(table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                             /*primary_keys=*/{}, parquet_options));
    ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));

    std::map<std::string, std::string> orc_options = {{Options::BUCKET, "-1"},
                                                      {Options::FILE_FORMAT, "orc"}};
    ASSERT_OK_AND_ASSIGN(table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                             /*primary_keys=*/{}, orc_options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "VECTOR currently only supports parquet data files");

    std::map<std::string, std::string> primary_key_options = {{Options::BUCKET, "1"}};
    ASSERT_OK_AND_ASSIGN(table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                             /*primary_keys=*/{"embedding"}, primary_key_options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "in primary key field embedding is unsupported");

    primary_key_options[Options::FILE_FORMAT] = "parquet";
    ASSERT_OK_AND_ASSIGN(table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                             /*primary_keys=*/{"id"}, primary_key_options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "VECTOR fields in primary-key tables are not implemented yet.");

    auto nested_schema = arrow::schema({
        arrow::field("id", arrow::int64()),
        arrow::field("payload", arrow::struct_({arrow::field("embedding", vector_field->type())})),
    });
    ASSERT_OK_AND_ASSIGN(
        table_schema,
        TableSchema::Create(/*schema_id=*/0, nested_schema,
                            /*partition_keys=*/{}, /*primary_keys=*/{"id"}, primary_key_options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "VECTOR fields in primary-key tables are not implemented yet.");

    std::map<std::string, std::string> data_evolution_options = {
        {Options::BUCKET, "-1"},
        {Options::FILE_FORMAT, "parquet"},
        {Options::ROW_TRACKING_ENABLED, "true"},
        {Options::DATA_EVOLUTION_ENABLED, "true"},
    };
    ASSERT_OK_AND_ASSIGN(table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema,
                                             /*partition_keys=*/{},
                                             /*primary_keys=*/{}, data_evolution_options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "VECTOR fields in data-evolution tables are not implemented yet.");
    ASSERT_OK_AND_ASSIGN(table_schema,
                         TableSchema::Create(/*schema_id=*/0, nested_schema,
                                             /*partition_keys=*/{},
                                             /*primary_keys=*/{}, data_evolution_options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "VECTOR fields in data-evolution tables are not implemented yet.");
}

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));

    // such a table may also enable deletion vectors, which another engine issues and this one
    // only reads
    options.emplace(Options::DELETION_VECTORS_ENABLED, "true");
    ASSERT_OK_AND_ASSIGN(
        std::shared_ptr<TableSchema> deletion_vector_table_schema,
        TableSchema::Create(/*schema_id=*/0, schema, partition_keys, primary_keys, options));
    ASSERT_OK(SchemaValidation::ValidateTableSchema(*deletion_vector_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));
    }
    {
        // a blob table with data evolution may also enable deletion vectors
        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::DELETION_VECTORS_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"}};
        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};
        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, TestSpecificPartitionKey) {
    {
        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 cannot be TIMESTAMP/DECIMAL/BLOB");
    }
    {
        auto f0 = arrow::field("f0", arrow::utf8());
        auto f1 = arrow::field("f1", arrow::float64());
        arrow::FieldVector fields = {f0, f1};
        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 cannot be FLOAT/DOUBLE");
    }
}

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 cannot be TIMESTAMP/DECIMAL/BLOB");
}

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", "f2"};
        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),
            "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 = {"f1"};
        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 = {"f1"};
        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");
    }
    {
        std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
                                                      {Options::BUCKET_KEY, "f0"},
                                                      {Options::DELETION_VECTORS_ENABLED, "true"},
                                                      {Options::MERGE_ENGINE, "first-row"}};
        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),
                            "First row merge engine does not need deletion vectors because there "
                            "is no deletion of old data in this merge engine.");
    }
}

TEST(SchemaValidationTest, ValidateSequenceField) {
    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::SEQUENCE_FIELD, "f0,f1,f2"}};
        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::SEQUENCE_FIELD, "f0,f1,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),
                            "cannot be found in table schema.");
    }
    {
        std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
                                                      {Options::BUCKET_KEY, "f0"},
                                                      {Options::SEQUENCE_FIELD, "f0,f1,f2"},
                                                      {Options::MERGE_ENGINE, "first-row"}};
        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),
                            "Do not support using sequence field on FIRST_ROW merge engine.");
    }
    {
        std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
                                                      {Options::SEQUENCE_FIELD, "f0,f1,f2"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{"f1"},
                                                 /*primary_keys=*/{"f0", "f2"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "You cannot use sequence.field in cross partition update case (Primary "
                            "key constraint 'f0, f2'  not including all partition fields 'f1').");
    }
}

TEST(SchemaValidationTest, ValidateSequenceGroup) {
    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.sequence-group", "f2"}};
        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::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,f3.sequence-group", "f2"}};
        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 can not be found in table schema.");
    }
    {
        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.sequence-group", "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 can not be found in table schema.");
    }
    {
        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.sequence-group", "f0,f1"},
                                                      {"fields.f2.sequence-group", "f0,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),
                            "defined repeatedly by multiple groups");
    }
    {
        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.sequence-group", "f2"},
            {"fields.f0.aggregate-function", "min"},
        };
        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),
                            "Should not define aggregation function on sequence group");
    }
}

TEST(SchemaValidationTest, ValidateInvalidConfiguration) {
    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::map<std::string, std::string> options = {{Options::CHANGELOG_PRODUCER, "input"}};
        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),
                            "Can not set changelog-producer on table without primary keys, please "
                            "define primary keys.");
    }
    {
        auto invalid_field = arrow::field("_SEQUENCE_NUMBER", arrow::int64());
        arrow::FieldVector invalid_fields = fields;
        invalid_fields.push_back(invalid_field);
        auto invalid_schema = arrow::schema(invalid_fields);
        ASSERT_OK_AND_ASSIGN(
            std::shared_ptr<TableSchema> table_schema,
            TableSchema::Create(/*schema_id=*/0, invalid_schema, /*partition_keys=*/{},
                                /*primary_keys=*/{}, /*options=*/{}));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "field name '_SEQUENCE_NUMBER' in schema cannot be special field.");
    }
    {
        auto invalid_field = arrow::field("_KEY_a", arrow::int64());
        arrow::FieldVector invalid_fields = fields;
        invalid_fields.push_back(invalid_field);
        auto invalid_schema = arrow::schema(invalid_fields);
        ASSERT_OK_AND_ASSIGN(
            std::shared_ptr<TableSchema> table_schema,
            TableSchema::Create(/*schema_id=*/0, invalid_schema, /*partition_keys=*/{},
                                /*primary_keys=*/{}, /*options=*/{}));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "field name '_KEY_a' in schema cannot be special field.");
    }
    {
        std::map<std::string, std::string> options = {{Options::CHANGELOG_PRODUCER, "input"},
                                                      {Options::MERGE_ENGINE, "first-row"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "C++ Paimon does not support changelog-producer yet");
    }
    {
        std::map<std::string, std::string> options = {{Options::CHANGELOG_PRODUCER, "lookup"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "C++ Paimon does not support changelog-producer yet");
    }
    // test for row tracking
    {
        std::map<std::string, std::string> options = {{Options::ROW_TRACKING_ENABLED, "true"},
                                                      {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),
            "Cannot define bucket for row tracking table, it only support bucket = -1");
    }
    {
        std::map<std::string, std::string> options = {{Options::ROW_TRACKING_ENABLED, "true"},
                                                      {Options::BUCKET, "-1"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "Cannot define primary key for row tracking table");
    }
    {
        std::map<std::string, std::string> options = {{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_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "Data evolution config must enabled with row-tracking.enabled");
    }
}

TEST(SchemaValidationTest, TestMapStorageLayout) {
    auto f0 = arrow::field("f0", arrow::utf8());
    auto f1 = arrow::field("f1", arrow::int32());
    auto f2 = arrow::field("f2", arrow::map(arrow::utf8(), arrow::int64()));
    auto f3 = arrow::field("f3", arrow::map(arrow::int32(), arrow::utf8()));

    // Valid: shared-shredding on MAP<STRING, T> column
    {
        arrow::FieldVector fields = {f0, f1, f2};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.f2.map.storage-layout", "shared-shredding"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
    }
    // Invalid: field not in schema failed in ValidateFieldsPrefix
    {
        arrow::FieldVector fields = {f0, f1};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.nonexist.map.storage-layout", "shared-shredding"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "Field nonexist can not be found in table schema");
    }
    // Invalid: field not in schema failed in ValidateMapStorageLayout
    {
        arrow::FieldVector fields = {f0, f1};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.nonexist.map.storage-layout", "shared-shredding"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateMapStorageLayout(*table_schema, core_options),
                            "Column 'nonexist' is configured with map.storage-layout but does not "
                            "exist in table schema.");
    }

    // Invalid: non-MAP column configured with map.storage-layout (any value)
    {
        arrow::FieldVector fields = {f0, f1};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
                                                      {Options::BUCKET_KEY, "f0"},
                                                      {"fields.f1.map.storage-layout", "default"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "not MAP");
    }
    // Invalid: shared-shredding on non-MAP column
    {
        arrow::FieldVector fields = {f0, f1};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.f1.map.storage-layout", "shared-shredding"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "not MAP");
    }
    // Invalid: shared-shredding on MAP with non-STRING key
    {
        arrow::FieldVector fields = {f0, f1, f3};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.f3.map.storage-layout", "shared-shredding"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "not MAP<STRING NOT NULL, T>");
    }
    // Invalid: nested MAP paths are not shared-shredding columns; only top-level columns are
    // addressable by fields.<column>.map.storage-layout.
    {
        auto payload = arrow::field(
            "payload",
            arrow::struct_({arrow::field("attrs", arrow::map(arrow::utf8(), arrow::int64()))}));
        arrow::FieldVector fields = {f0, f1, payload};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.payload.attrs.map.storage-layout", "shared-shredding"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(
            SchemaValidation::ValidateTableSchema(*table_schema),
            "Column 'payload.attrs' is configured with map.storage-layout but does not exist in "
            "table schema.");
    }
    // Valid: default layout on a MAP column
    {
        arrow::FieldVector fields = {f0, f1, f2};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {{Options::BUCKET, "2"},
                                                      {Options::BUCKET_KEY, "f0"},
                                                      {"fields.f2.map.storage-layout", "default"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
    }
    // Invalid: shared-shredding with invalid max-columns
    {
        arrow::FieldVector fields = {f0, f1, f2};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.f2.map.storage-layout", "shared-shredding"},
            {"fields.f2.map.shared-shredding.max-columns", "0"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "options map.shared-shredding.max-columns must > 0");
    }
    // Invalid: shared-shredding with invalid placement policy
    {
        arrow::FieldVector fields = {f0, f1, f2};
        auto schema = arrow::schema(fields);
        std::map<std::string, std::string> options = {
            {Options::BUCKET, "2"},
            {Options::BUCKET_KEY, "f0"},
            {"fields.f2.map.storage-layout", "shared-shredding"},
            {"fields.f2.map.shared-shredding.column-placement-policy", "invalid"}};
        ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                             TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                 /*primary_keys=*/{"f0", "f1"}, options));
        ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                            "invalid map.shared-shredding.column-placement-policy: invalid");
    }
}

TEST(SchemaValidationTest, TestMapSharedShreddingRequiresNonNullableKey) {
    auto nullable_key_map =
        std::make_shared<arrow::MapType>(arrow::field("key", arrow::utf8(), /*nullable=*/true),
                                         arrow::field("value", arrow::int64()));
    auto schema = arrow::schema({
        arrow::field("f0", arrow::utf8()),
        arrow::field("f1", nullable_key_map),
    });
    std::map<std::string, std::string> options = {
        {Options::BUCKET, "1"},
        {Options::BUCKET_KEY, "f0"},
        {"fields.f1.map.storage-layout", "shared-shredding"},
    };
    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),
                        "map key type is nullable");
}

TEST(SchemaValidationTest, TestMapSharedShreddingRejectsBlobValue) {
    auto direct_blob_map =
        arrow::map(arrow::utf8(), BlobUtils::ToArrowField("value", /*nullable=*/true));
    auto nested_blob_map = arrow::map(
        arrow::utf8(), arrow::field("value", arrow::struct_({BlobUtils::ToArrowField("blob")})));
    std::map<std::string, std::string> options = {
        {Options::BUCKET, "1"},
        {Options::BUCKET_KEY, "f0"},
        {"fields.f1.map.storage-layout", "shared-shredding"},
    };

    for (const auto& map_type : {direct_blob_map, nested_blob_map}) {
        auto schema = arrow::schema({
            arrow::field("f0", arrow::utf8()),
            arrow::field("f1", map_type),
        });
        ASSERT_NOK_WITH_MSG(TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                                /*primary_keys=*/{}, options),
                            "Blob field must be a top-level field.");
    }
}

TEST(SchemaValidationTest, TestMapSharedShreddingCompression) {
    auto schema = arrow::schema({
        arrow::field("f0", arrow::utf8()),
        arrow::field("f1", arrow::map(arrow::utf8(), arrow::int64())),
    });
    std::map<std::string, std::string> base_options = {
        {Options::BUCKET, "1"},
        {Options::BUCKET_KEY, "f0"},
        {"fields.f1.map.storage-layout", "shared-shredding"},
    };

    for (const std::string compression : {"none", "lz4", "zstd", "ZSTD"}) {
        auto options = base_options;
        options[Options::FILE_COMPRESSION] = compression;
        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));
    }
    {
        auto options = base_options;
        options[Options::FILE_COMPRESSION] = "snappy";
        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),
                            "MAP shared-shredding only supports none/lz4/zstd compression, but "
                            "file.compression is snappy.");
    }
    {
        auto options = base_options;
        options[Options::FILE_COMPRESSION] = "";
        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),
                            "MAP shared-shredding only supports none/lz4/zstd compression, but "
                            "file.compression is .");
    }
    {
        auto options = base_options;
        options[Options::FILE_COMPRESSION_PER_LEVEL] = "0:lz4,1:snappy";
        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),
                            "MAP shared-shredding only supports none/lz4/zstd compression, but "
                            "file.compression.per.level.1 is snappy.");
    }
    {
        auto options = base_options;
        options.erase("fields.f1.map.storage-layout");
        options[Options::FILE_COMPRESSION] = "snappy";
        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, TestMapSharedShreddingFileFormat) {
    auto schema = arrow::schema({
        arrow::field("f0", arrow::utf8()),
        arrow::field("f1", arrow::map(arrow::utf8(), arrow::int64())),
    });
    std::map<std::string, std::string> base_options = {
        {Options::BUCKET, "1"},
        {Options::BUCKET_KEY, "f0"},
        {"fields.f1.map.storage-layout", "shared-shredding"},
    };

    for (const std::string file_format : {"parquet", "orc"}) {
        auto options = base_options;
        options[Options::FILE_FORMAT] = file_format;
        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));
    }
    {
        auto options = base_options;
        options[Options::FILE_FORMAT] = "avro";
        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),
            "MAP shared-shredding only supports parquet/orc file formats, but file.format is "
            "avro.");
    }
    {
        auto options = base_options;
        options[Options::FILE_FORMAT_PER_LEVEL] = "0:parquet,1:avro";
        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),
                            "MAP shared-shredding only supports parquet/orc file formats, but "
                            "file.format.per.level.1 is avro.");
    }
}

TEST(SchemaValidationTest, TestMapSharedShreddingRejectsPostponeBucketMode) {
    auto schema = arrow::schema({
        arrow::field("f0", arrow::utf8()),
        arrow::field("f1", arrow::map(arrow::utf8(), arrow::int64())),
    });
    std::map<std::string, std::string> options = {
        {Options::BUCKET, "-2"},
        {Options::WRITE_ONLY, "true"},
        {"fields.f1.map.storage-layout", "shared-shredding"},
    };
    ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
                         TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
                                             /*primary_keys=*/{"f0"}, options));
    ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
                        "MAP shared-shredding currently does not support postpone bucket mode.");
}

}  // namespace paimon::test