Google Git Google Git
Sign in
apache/paimon-cpp/242decffdfdefef01dcd0e52e48c9a1e69792479/./src/paimon/core/schema/schema_validation_test.cpp
blob: 9356b0be12e59ffd0a5c2dc8bde657992e5fcb5f [file]
/*
* 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 <string>
#include <vector>
#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/common/utils/checked_cast.h"
#include "paimon/core/schema/table_schema.h"
#include "paimon/defs.h"
#include "paimon/testing/utils/testharness.h"
namespace paimon::test {
namespace {
Result<std::unique_ptr<TableSchema>> MakePrimaryKeyBTreeSchema(
std::map<std::string, std::string> options,
const std::vector<std::string>& primary_keys = {"id"},
const std::shared_ptr<arrow::DataType>& value_type = arrow::int32()) {
options.emplace(Options::PK_BTREE_INDEX_COLUMNS, "value");
auto schema = arrow::schema({arrow::field("id", arrow::int64(), /*nullable=*/false),
arrow::field("value", value_type)});
return TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{}, primary_keys,
options);
}
const std::vector<std::string>& ContainerBlobTypeJsons() {
static const std::vector<std::string> container_blob_types = {
R"({"type":"ARRAY","element":"BLOB"})",
R"({"type":"MAP","key":"STRING","value":"BLOB"})",
};
return container_blob_types;
}
Result<std::unique_ptr<TableSchema>> MakeContainerBlobSchema(const std::string& blob_type_json,
const std::string& partition_keys_json,
const std::string& options_json) {
const std::string schema_json =
R"({
"version": 3,
"id": 0,
"fields": [
{"id": 0, "name": "id", "type": "INT"},
{"id": 1, "name": "blob", "type": )" +
blob_type_json +
R"(}
],
"highestFieldId": 1,
"partitionKeys": )" +
partition_keys_json +
R"(,
"primaryKeys": [],
"options": )" +
options_json +
R"(,
"timeMillis": 0
})";
return TableSchema::CreateFromJson(schema_json);
}
} // namespace
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/lance data files");
std::map<std::string, std::string> changelog_options = {
{Options::BUCKET, "1"},
{Options::FILE_FORMAT, "parquet"},
{Options::CHANGELOG_FILE_FORMAT, "orc"},
};
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{"id"}, changelog_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"changelog-file.format is orc");
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");
ASSERT_OK_AND_ASSIGN(
table_schema, TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{"embedding"},
/*primary_keys=*/{}, parquet_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"in partition field embedding is unsupported");
std::map<std::string, std::string> bucket_key_options = {
{Options::BUCKET, "1"},
{Options::BUCKET_KEY, "embedding"},
{Options::FILE_FORMAT, "parquet"},
};
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{}, bucket_key_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"Nested type cannot be in bucket-key");
std::map<std::string, std::string> sequence_field_options = {
{Options::BUCKET, "1"},
{Options::BUCKET_KEY, "id"},
{Options::FILE_FORMAT, "parquet"},
{Options::SEQUENCE_FIELD, "embedding"},
};
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{"id"}, sequence_field_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"VECTOR field 'embedding' cannot be used as a sequence field.");
std::map<std::string, std::string> sequence_group_options = {
{Options::BUCKET, "1"},
{Options::BUCKET_KEY, "id"},
{Options::FILE_FORMAT, "parquet"},
{Options::MERGE_ENGINE, "partial-update"},
{"fields.embedding.sequence-group", "id"},
};
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{"id"}, sequence_group_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"VECTOR field 'embedding' cannot be used as a sequence-group ordering "
"field.");
primary_key_options[Options::FILE_FORMAT] = "parquet";
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{"id"}, primary_key_options));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
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_OK(SchemaValidation::ValidateTableSchema(*table_schema));
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_OK(SchemaValidation::ValidateTableSchema(*table_schema));
ASSERT_OK_AND_ASSIGN(
table_schema,
TableSchema::Create(/*schema_id=*/0, nested_schema,
/*partition_keys=*/{}, /*primary_keys=*/{}, data_evolution_options));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
data_evolution_options[Options::FILE_FORMAT] = "orc";
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 currently only supports parquet/lance data files");
}
#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));
for (const std::string& blob_type_json : ContainerBlobTypeJsons()) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> container_blob_schema,
MakeContainerBlobSchema(blob_type_json, "[]",
R"({
"bucket": "-1",
"file.format": "mosaic",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true"
})"));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*container_blob_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
#ifdef PAIMON_ENABLE_LANCE
TEST(SchemaValidationTest, TestLanceDataTypes) {
std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
{Options::FILE_FORMAT, "lance"}};
arrow::FieldVector supported_fields = {
arrow::field("boolean", arrow::boolean()),
arrow::field("tinyint", arrow::int8()),
arrow::field("smallint", arrow::int16()),
arrow::field("integer", arrow::int32()),
arrow::field("bigint", arrow::int64()),
arrow::field("float", arrow::float32()),
arrow::field("double", arrow::float64()),
arrow::field("string", arrow::utf8()),
arrow::field("binary", arrow::binary()),
arrow::field("date", arrow::date32()),
arrow::field("timestamp", arrow::timestamp(arrow::TimeUnit::MICRO)),
arrow::field("decimal", arrow::decimal128(38, 2)),
arrow::field("array", arrow::list(arrow::float32())),
arrow::field("row", arrow::struct_({arrow::field("value", arrow::int32())}),
/*nullable=*/false),
arrow::field("nullable_row", arrow::struct_({arrow::field("value", arrow::int32())})),
arrow::field("nested_nullable_row",
arrow::struct_({arrow::field(
"child", arrow::struct_({arrow::field("value", arrow::int32())}))}),
/*nullable=*/false),
arrow::field("row_array",
arrow::list(arrow::struct_({arrow::field("value", arrow::int32())}))),
arrow::field("vector", arrow::fixed_size_list(arrow::float32(), 3)),
};
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));
for (const std::string& blob_type_json : ContainerBlobTypeJsons()) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> container_blob_schema,
MakeContainerBlobSchema(blob_type_json, "[]",
R"({
"bucket": "-1",
"file.format": "lance",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true"
})"));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*container_blob_schema));
}
arrow::FieldVector unsupported_fields = {
arrow::field("map", arrow::map(arrow::int32(), arrow::utf8())),
arrow::field("ltz", arrow::timestamp(arrow::TimeUnit::MICRO, "UTC")),
VariantTypeUtils::ToArrowField("variant"),
arrow::field("time_millis", arrow::time32(arrow::TimeUnit::MILLI)),
arrow::field("nested_time",
arrow::struct_({arrow::field(
"values", arrow::list(arrow::time32(arrow::TimeUnit::MILLI)))})),
};
std::vector<std::string> expected_errors = {"type MAP", "LOCAL_ZONED_TIMESTAMP", "type VARIANT",
"type time32", "type time32"};
for (size_t i = 0; i < unsupported_fields.size(); ++i) {
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]);
}
for (const auto& field : arrow::FieldVector{
arrow::field("time_seconds", arrow::time32(arrow::TimeUnit::SECOND)),
arrow::field("nested_time",
arrow::struct_({arrow::field(
"values", arrow::list(arrow::time32(arrow::TimeUnit::SECOND)))}))}) {
ASSERT_NOK_WITH_MSG(
TableSchema::Create(/*schema_id=*/0, arrow::schema({field}),
/*partition_keys=*/{}, /*primary_keys=*/{}, options),
"Only millisecond TIME is supported");
}
for (const auto& [option_key, option_value] : std::vector<std::pair<std::string, std::string>>{
{Options::FILE_FORMAT_PER_LEVEL, "1:lance"},
{Options::CHANGELOG_FILE_FORMAT, "lance"}}) {
std::map<std::string, std::string> alternate_format_options = {
{Options::BUCKET, "-1"}, {Options::FILE_FORMAT, "parquet"}, {option_key, option_value}};
ASSERT_OK_AND_ASSIGN(
table_schema,
TableSchema::Create(
/*schema_id=*/0,
arrow::schema({arrow::field("map", arrow::map(arrow::int32(), arrow::utf8()))}),
/*partition_keys=*/{}, /*primary_keys=*/{}, alternate_format_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type MAP");
}
}
#endif
#ifdef PAIMON_ENABLE_VORTEX
TEST(SchemaValidationTest, TestVortexDataTypes) {
std::shared_ptr<TableSchema> table_schema;
// Vortex as the default file format: unsupported types are rejected at validation time.
std::map<std::string, std::string> vortex_options = {{Options::BUCKET, "-1"},
{Options::FILE_FORMAT, "vortex"}};
ASSERT_OK_AND_ASSIGN(
table_schema,
TableSchema::Create(
/*schema_id=*/0,
arrow::schema({arrow::field("map", arrow::map(arrow::int32(), arrow::utf8()))}),
/*partition_keys=*/{}, /*primary_keys=*/{}, vortex_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type MAP");
// Vortex selected per level or for the changelog must be validated too, not only the default.
for (const auto& [option_key, option_value] : std::vector<std::pair<std::string, std::string>>{
{Options::FILE_FORMAT_PER_LEVEL, "1:vortex"},
{Options::CHANGELOG_FILE_FORMAT, "vortex"}}) {
std::map<std::string, std::string> alternate_format_options = {
{Options::BUCKET, "-1"}, {Options::FILE_FORMAT, "parquet"}, {option_key, option_value}};
ASSERT_OK_AND_ASSIGN(
table_schema,
TableSchema::Create(
/*schema_id=*/0,
arrow::schema({arrow::field("map", arrow::map(arrow::int32(), arrow::utf8()))}),
/*partition_keys=*/{}, /*primary_keys=*/{}, alternate_format_options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema), "type MAP");
}
}
#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, TestContainerBlobRowTracking) {
for (const std::string& blob_type_json : ContainerBlobTypeJsons()) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> table_schema,
MakeContainerBlobSchema(blob_type_json, "[]",
R"({
"bucket": "-1",
"row-tracking.enabled": "true",
"data-evolution.enabled": "false"
})"));
ASSERT_OK_AND_ASSIGN(CoreOptions options, CoreOptions::FromMap(table_schema->Options()));
ASSERT_NOK_WITH_MSG(
SchemaValidation::ValidateRowTracking(*table_schema, options),
"Data evolution config must be enabled for table with BLOB type column.");
ASSERT_OK_AND_ASSIGN(table_schema, MakeContainerBlobSchema(blob_type_json, R"(["blob"])",
R"({
"bucket": "-1",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true"
})"));
ASSERT_OK_AND_ASSIGN(options, CoreOptions::FromMap(table_schema->Options()));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateRowTracking(*table_schema, options),
"Blob field blob cannot be a partition key.");
}
}
TEST(SchemaValidationTest, TestMapBlobKeyType) {
const std::string options_json = R"({
"bucket": "-1",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true"
})";
ASSERT_OK_AND_ASSIGN(
std::unique_ptr<TableSchema> table_schema,
MakeContainerBlobSchema(R"json({"type":"MAP","key":"STRING","value":"BLOB"})json", "[]",
options_json));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
// Checked without any blob option naming the field.
ASSERT_OK_AND_ASSIGN(
table_schema,
MakeContainerBlobSchema(R"json({"type":"MAP","key":"TIME(3)","value":"BLOB"})json", "[]",
options_json));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"Field 'blob' has unsupported MAP<..., BLOB> key type: time32[ms].");
std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
{Options::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{Options::BLOB_FIELD, "blob_map"}};
auto schema = arrow::schema(
{arrow::field("id", arrow::int32()),
arrow::field("blob_map", arrow::map(arrow::float32(), BlobUtils::ToArrowField(
"value", /*nullable=*/true)))});
ASSERT_OK_AND_ASSIGN(table_schema,
TableSchema::Create(/*schema_id=*/0, schema, /*partition_keys=*/{},
/*primary_keys=*/{}, options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"Field 'blob_map' has unsupported MAP<..., BLOB> key type: float.");
}
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));
}
for (const std::string& blob_type_json : ContainerBlobTypeJsons()) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> table_schema,
MakeContainerBlobSchema(blob_type_json, "[]",
R"({
"bucket": "-1",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true",
"blob-field": "blob"
})"));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
}
const std::vector<std::pair<std::string, std::string>> inline_blob_options = {
{std::string(Options::BLOB_DESCRIPTOR_FIELD),
R"({"bucket":"-1","row-tracking.enabled":"true","data-evolution.enabled":"true","blob-descriptor-field":"blob"})"},
{std::string(Options::BLOB_VIEW_FIELD),
R"({"bucket":"-1","row-tracking.enabled":"true","data-evolution.enabled":"true","blob-view-field":"blob"})"},
};
for (const std::string& blob_type_json : ContainerBlobTypeJsons()) {
for (const auto& [option_key, options_json] : inline_blob_options) {
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> table_schema,
MakeContainerBlobSchema(blob_type_json, "[]", options_json));
ASSERT_NOK_WITH_MSG(
SchemaValidation::ValidateTableSchema(*table_schema),
"Field 'blob' in '" + option_key + "' must be a BLOB field in 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, ARRAY<BLOB> or "
"MAP<..., 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, TestTimePartitionKey) {
auto schema = arrow::schema({arrow::field("id", arrow::int32()),
arrow::field("time", arrow::time32(arrow::TimeUnit::MILLI))});
for (const std::vector<std::string>& partition_keys :
{std::vector<std::string>{}, std::vector<std::string>{"time"}}) {
ASSERT_OK_AND_ASSIGN(auto table_schema,
TableSchema::Create(0, schema, partition_keys, {},
{{"file.format", "parquet"}, {"bucket", "-1"}}));
if (partition_keys.empty()) {
ASSERT_OK(SchemaValidation::ValidateTableSchema(*table_schema));
} else {
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"partition field time cannot be TIME");
}
}
}
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, ValidatePrimaryKeyBTreeIndexes) {
{
std::map<std::string, std::string> options = {
{Options::BUCKET, "1"},
{Options::DELETION_VECTORS_ENABLED, "true"},
{"fields.value.pk-btree.index.options", R"({"cache-size":"16 mb"})"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*schema));
}
{
std::map<std::string, std::string> options = {{Options::BUCKET, "-2"},
{Options::DELETION_VECTORS_ENABLED, "true"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_OK(SchemaValidation::ValidateTableSchema(*schema));
}
{
std::map<std::string, std::string> options = {{Options::BUCKET, "1"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"require deletion-vectors.enabled = true");
}
{
std::map<std::string, std::string> options = {{Options::BUCKET, "1"},
{Options::BUCKET_KEY, "id"},
{Options::DELETION_VECTORS_ENABLED, "true"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options, /*primary_keys=*/{}));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"require a primary-key table");
}
{
std::map<std::string, std::string> options = {{Options::BUCKET, "-1"},
{Options::DELETION_VECTORS_ENABLED, "true"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"require fixed or postpone bucket mode");
}
{
std::map<std::string, std::string> options = {{Options::BUCKET, "1"},
{Options::DELETION_VECTORS_ENABLED, "true"},
{"deletion-vectors.merge-on-read", "true"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"require deletion-vectors.merge-on-read = false");
}
{
std::map<std::string, std::string> options = {{Options::BUCKET, "1"},
{Options::DELETION_VECTORS_ENABLED, "true"},
{Options::PK_CLUSTERING_OVERRIDE, "true"}};
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"unsupported by the C++ commit path");
}
}
TEST(SchemaValidationTest, ValidatePrimaryKeyBTreeIndexColumnsAndOptions) {
std::map<std::string, std::string> valid_options = {
{Options::BUCKET, "1"}, {Options::DELETION_VECTORS_ENABLED, "true"}};
{
std::map<std::string, std::string> options = valid_options;
options.emplace(Options::PK_BTREE_INDEX_COLUMNS, "missing");
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"entry 'missing' must reference an existing column");
}
{
ASSERT_OK_AND_ASSIGN(
std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(valid_options, /*primary_keys=*/{"id"}, arrow::binary()));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"entry 'value' has unsupported type binary");
}
{
std::map<std::string, std::string> options = valid_options;
options.emplace(Options::PK_BTREE_INDEX_COLUMNS, "value,value");
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema),
"contains duplicate column 'value'");
}
{
std::map<std::string, std::string> options = valid_options;
options.emplace("fields.value.pk-btree.index.options", R"({"cache-size":"not-a-size"})");
ASSERT_OK_AND_ASSIGN(std::unique_ptr<TableSchema> schema,
MakePrimaryKeyBTreeSchema(options));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*schema), "not-a-size");
}
}
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 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::ROW_TRACKING_ENABLED, "true"},
{Options::DATA_EVOLUTION_ENABLED, "true"},
{"fields.f1.map.storage-layout", "shared-shredding"},
};
const std::string loaded_schema = R"json({
"version": 3,
"id": 0,
"fields": [
{"id": 0, "name": "f0", "type": "STRING"},
{"id": 1, "name": "f1",
"type": {"type": "MAP", "key": "STRING", "value": "BLOB"}}
],
"highestFieldId": 1,
"partitionKeys": [],
"primaryKeys": [],
"options": {
"bucket": "-1",
"row-tracking.enabled": "true",
"data-evolution.enabled": "true",
"fields.f1.map.storage-layout": "shared-shredding"
},
"timeMillis": 0
})json";
ASSERT_OK_AND_ASSIGN(std::shared_ptr<TableSchema> table_schema,
TableSchema::CreateFromJson(loaded_schema));
auto loaded_map = checked_pointer_cast<arrow::MapType>(table_schema->Fields()[1].Type());
ASSERT_TRUE(BlobUtils::IsBlobField(loaded_map->item_field()));
ASSERT_NOK_WITH_MSG(SchemaValidation::ValidateTableSchema(*table_schema),
"MAP shared-shredding currently cannot contain BLOB fields.");
auto nested_schema = arrow::schema({
arrow::field("f0", arrow::utf8()),
arrow::field("f1", nested_blob_map),
});
ASSERT_NOK_WITH_MSG(TableSchema::Create(/*schema_id=*/0, nested_schema,
/*partition_keys=*/{}, /*primary_keys=*/{}, options),
"BLOB field must be a top-level field or the direct element/value of a "
"top-level ARRAY/MAP 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