| /* |
| * 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/common/data/variant/variant_type_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, TestRealtimeOffsetIsGloballyReserved) { |
| auto schema = arrow::schema({arrow::field("_REALTIME_OFFSET", arrow::int64())}); |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema, |
| TableSchema::Create(0, schema, {}, {}, {})); |
| ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), |
| "field name '_REALTIME_OFFSET' in schema cannot be special field"); |
| } |
| |
| 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."); |
| } |
| |
| #ifdef PAIMON_ENABLE_MOSAIC |
| TEST(SchemaValidationTest, TestMosaicDataTypes) { |
| std::map<std::string, std::string> options = {{Options::BUCKET, "-1"}, |
| {Options::FILE_FORMAT, "mosaic"}}; |
| arrow::FieldVector supported_fields = { |
| arrow::field("f0", arrow::boolean()), |
| arrow::field("f1", arrow::int8()), |
| arrow::field("f2", arrow::int16()), |
| arrow::field("f3", arrow::int32()), |
| arrow::field("f4", arrow::int64()), |
| arrow::field("f5", arrow::float32()), |
| arrow::field("f6", arrow::float64()), |
| arrow::field("f7", arrow::utf8()), |
| arrow::field("f8", arrow::binary()), |
| arrow::field("f9", arrow::date32()), |
| arrow::field("f10", arrow::timestamp(arrow::TimeUnit::NANO)), |
| arrow::field("f11", arrow::decimal128(38, 2)), |
| arrow::field("f12", arrow::list(arrow::float32())), |
| arrow::field("f13", arrow::map(arrow::int8(), arrow::int16())), |
| }; |
| ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema, |
| TableSchema::Create(/*schema_id=*/0, arrow::schema(supported_fields), |
| /*partition_keys=*/{}, /*primary_keys=*/{}, options)); |
| ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema)); |
| |
| arrow::FieldVector unsupported_fields = { |
| arrow::field("row", arrow::struct_({arrow::field("value", arrow::int32())})), |
| VariantTypeUtils::ToArrowField("variant"), |
| arrow::field("vector", arrow::fixed_size_list(arrow::float32(), 3)), |
| arrow::field("timestamp", arrow::timestamp(arrow::TimeUnit::SECOND)), |
| arrow::field("nested_row", |
| arrow::list(arrow::struct_({arrow::field("value", arrow::int32())}))), |
| }; |
| std::vector<std::string> expected_errors = {"type ROW", "type VARIANT", "type VECTOR", |
| "TIMESTAMP(0)", "type ROW"}; |
| for (size_t i = 0; i < unsupported_fields.size(); ++i) { |
| SCOPED_TRACE("field=" + unsupported_fields[i]->name()); |
| ASSERT_OK_AND_ASSIGN( |
| table_schema, |
| TableSchema::Create(/*schema_id=*/0, arrow::schema({unsupported_fields[i]}), |
| /*partition_keys=*/{}, /*primary_keys=*/{}, options)); |
| ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), |
| expected_errors[i]); |
| } |
| |
| std::shared_ptr<arrow::Field> blob_field = BlobUtils::ToArrowField("blob", false); |
| std::map<std::string, std::string> blob_options = { |
| {Options::BUCKET, "-1"}, |
| {Options::FILE_FORMAT, "mosaic"}, |
| {Options::ROW_TRACKING_ENABLED, "true"}, |
| {Options::DATA_EVOLUTION_ENABLED, "true"}, |
| }; |
| ASSERT_OK_AND_ASSIGN( |
| table_schema, |
| TableSchema::Create(/*schema_id=*/0, |
| arrow::schema({arrow::field("id", arrow::int32()), blob_field}), |
| /*partition_keys=*/{}, /*primary_keys=*/{}, blob_options)); |
| ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema)); |
| |
| blob_options[Options::BLOB_DESCRIPTOR_FIELD] = "blob"; |
| ASSERT_OK_AND_ASSIGN( |
| table_schema, |
| TableSchema::Create(/*schema_id=*/0, |
| arrow::schema({arrow::field("id", arrow::int32()), blob_field}), |
| /*partition_keys=*/{}, /*primary_keys=*/{}, blob_options)); |
| ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type BLOB"); |
| } |
| #endif |
| |
| 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), |
| "Deletion vectors mode is only supported for " |
| "NONE/INPUT/LOOKUP changelog producer now"); |
| } |
| { |
| std::map<std::string, std::string> options = {{Options::BUCKET, "2"}, |
| {Options::BUCKET_KEY, "f0"}, |
| {Options::DELETION_VECTORS_ENABLED, "true"}, |
| {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_OK(SchemaValidation::ValidateTableSchema(*table_schema)); |
| } |
| { |
| 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."); |
| } |
| { |
| std::map<std::string, std::string> options = { |
| {Options::CHANGELOG_PRODUCER, "input"}, |
| {Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE, "true"}}; |
| 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), |
| "'changelog-producer.row-deduplicate' is only valid for 'lookup' or " |
| "'full-compaction' changelog producer"); |
| } |
| { |
| std::map<std::string, std::string> options = { |
| {Options::CHANGELOG_PRODUCER, "input"}, |
| {Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS, "f1"}}; |
| 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), |
| "'changelog-producer.row-deduplicate-ignore-fields' is only valid when " |
| "'changelog-producer.row-deduplicate' is true"); |
| } |
| { |
| std::map<std::string, std::string> options = { |
| {Options::CHANGELOG_PRODUCER, "input"}, |
| {Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE, "true"}, |
| {Options::CHANGELOG_PRODUCER_ROW_DEDUPLICATE_IGNORE_FIELDS, "missing"}}; |
| 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), |
| "Fields [\"missing\"] configured in " |
| "'changelog-producer.row-deduplicate-ignore-fields' can not be found in table schema"); |
| } |
| { |
| 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), |
| "Only support 'none' and 'lookup' changelog-producer on FIRST_ROW " |
| "merge engine"); |
| } |
| // 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[Options::CHANGELOG_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 " |
| "changelog-file.compression 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."); |
| } |
| { |
| auto options = base_options; |
| options[Options::CHANGELOG_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 " |
| "changelog-file.format 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 |
