Skip to content

[fix](paimon) Accept required Variant array fallback value #66626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yiguolei merged 1 commit into from
Aug 11, 2026
Merged

[fix](paimon) Accept required Variant array fallback value #66626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 22 additions & 12 deletions be/src/format_v2/parquet/native_schema_desc.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,7 +286,7 @@ class ScopedBoolOverride {

Status validate_variant_layout(const NativeFieldSchema& group_field,
std::optional<int8_t> specification_version,
bool allow_optional_shredded_fields) {
bool allow_paimon_shredded_layout) {
if (specification_version.has_value() && *specification_version != 1) {
return Status::NotSupported("Parquet Variant specification version {} is not supported",
*specification_version);
Expand Down Expand Up @@ -332,7 +332,7 @@ Status validate_variant_layout(const NativeFieldSchema& group_field,
// unannotated overrides; row materialization still rejects null metadata for a non-null value.
const bool valid_metadata_repetition =
metadata_repetition == tparquet::FieldRepetitionType::REQUIRED ||
(allow_optional_shredded_fields && typed_value != nullptr &&
(allow_paimon_shredded_layout && typed_value != nullptr &&
metadata_repetition == tparquet::FieldRepetitionType::OPTIONAL);
if (!metadata->children.empty() || metadata->physical_type != tparquet::Type::BYTE_ARRAY ||
!valid_metadata_repetition) {
Expand Down Expand Up @@ -360,12 +360,12 @@ Status validate_variant_layout(const NativeFieldSchema& group_field,
std::function<Status(const NativeFieldSchema&)> validate_typed_value;
std::function<Status(const NativeFieldSchema&, WrapperContext)> validate_wrapper;
validate_wrapper = [&](const NativeFieldSchema& wrapper, WrapperContext context) -> Status {
const bool valid_wrapper_repetition = wrapper.parquet_schema.__isset.repetition_type &&
(wrapper.parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::REQUIRED ||
(allow_optional_shredded_fields &&
wrapper.parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::OPTIONAL));
const bool valid_wrapper_repetition =
wrapper.parquet_schema.__isset.repetition_type &&
(wrapper.parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::REQUIRED ||
(allow_paimon_shredded_layout && wrapper.parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::OPTIONAL));
// The Parquet Variant specification requires wrapper groups. Paimon's unannotated
// physical carrier makes them optional, so accept that representation only through the
// table-format override. Materialization still rejects an actually null array element;
Expand Down Expand Up @@ -408,13 +408,23 @@ Status validate_variant_layout(const NativeFieldSchema& group_field,
"Parquet Variant object wrapper {} requires an optional value child",
wrapper.name);
}
// Paimon makes this leaf required because a fallback-only array element has no typed
// carrier; keep the exception scoped to that exact unannotated layout.
const bool allow_required_fallback = allow_paimon_shredded_layout &&
context == WrapperContext::ARRAY_ELEMENT &&
typed == nullptr;
const bool valid_fallback_repetition =
fallback != nullptr && fallback->parquet_schema.__isset.repetition_type &&
(fallback->parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::OPTIONAL ||
(allow_required_fallback && fallback->parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::REQUIRED));
if (fallback != nullptr &&
(!fallback->children.empty() || fallback->physical_type != tparquet::Type::BYTE_ARRAY ||
!fallback->parquet_schema.__isset.repetition_type ||
fallback->parquet_schema.repetition_type != tparquet::FieldRepetitionType::OPTIONAL)) {
!valid_fallback_repetition)) {
return Status::Corruption(
"Parquet Variant wrapper {} value must be an optional BYTE_ARRAY",
wrapper.name);
"Parquet Variant wrapper {} value must be an {} BYTE_ARRAY", wrapper.name,
allow_required_fallback ? "optional or required" : "optional");
}
if (typed != nullptr) {
if (!typed->parquet_schema.__isset.repetition_type ||
Expand Down
2 changes: 1 addition & 1 deletion be/src/format_v2/parquet/native_schema_desc.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ struct NativeFieldSchema {

Status validate_variant_layout(const NativeFieldSchema& group_field,
std::optional<int8_t> specification_version = std::nullopt,
bool allow_optional_shredded_fields = false);
bool allow_paimon_shredded_layout = false);

// V2 owns this schema tree and parser so footer/schema planning never invokes the V1 reader path.
class NativeFieldDescriptor {
Expand Down
53 changes: 53 additions & 0 deletions be/test/format_v2/parquet/parquet_schema_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,6 +274,59 @@ TEST(ParquetSchemaTest, AppliesPaimonShreddedVariantOverrideWithOptionalFields)
apply_override(std::move(array_schema));
}

TEST(ParquetSchemaTest, AppliesPaimonFallbackOnlyArrayWithRequiredValue) {
auto schema = shredded_array_variant_schema(true, false);
schema[1].__isset.logicalType = false;
schema[2].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
schema[6].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
schema[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);

NativeFieldDescriptor descriptor;
ASSERT_TRUE(descriptor.parse_from_thrift(schema).ok());
std::vector<std::unique_ptr<ParquetColumnSchema>> fields;
ASSERT_TRUE(build_parquet_column_schema(descriptor, &fields).ok());

const std::vector overrides {format::LocalColumnIndex::top_level(format::LocalColumnId(0))};
const auto status = apply_variant_schema_overrides(descriptor, overrides, &fields);
ASSERT_TRUE(status.ok()) << status;
ASSERT_EQ(fields.size(), 1);
EXPECT_EQ(fields[0]->kind, ParquetColumnSchemaKind::VARIANT);
}

TEST(ParquetSchemaTest, RejectsRequiredShreddedFallbackOutsidePaimonFallbackOnlyArray) {
const auto expect_override_corruption = [](std::vector<tparquet::SchemaElement> schema) {
NativeFieldDescriptor descriptor;
ASSERT_TRUE(descriptor.parse_from_thrift(schema).ok());
std::vector<std::unique_ptr<ParquetColumnSchema>> fields;
ASSERT_TRUE(build_parquet_column_schema(descriptor, &fields).ok());
const std::vector overrides {format::LocalColumnIndex::top_level(format::LocalColumnId(0))};
const auto status = apply_variant_schema_overrides(descriptor, overrides, &fields);
EXPECT_TRUE(status.is<ErrorCode::CORRUPTION>()) << status;
};

auto paimon_object = shredded_object_variant_schema();
paimon_object[1].__isset.logicalType = false;
paimon_object[2].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
paimon_object[5].__set_num_children(1);
paimon_object[5].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
paimon_object[6].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
paimon_object.pop_back();
expect_override_corruption(std::move(paimon_object));

auto paimon_array_with_typed_value = shredded_array_variant_schema(true, true);
paimon_array_with_typed_value[1].__isset.logicalType = false;
paimon_array_with_typed_value[2].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
paimon_array_with_typed_value[6].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
paimon_array_with_typed_value[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
expect_override_corruption(std::move(paimon_array_with_typed_value));

auto annotated_array = shredded_array_variant_schema(true, false);
annotated_array[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
NativeFieldDescriptor descriptor;
const auto annotated_status = descriptor.parse_from_thrift(annotated_array);
EXPECT_TRUE(annotated_status.is<ErrorCode::CORRUPTION>()) << annotated_status;
}

TEST(ParquetSchemaTest, RejectsMalformedUnannotatedVariantOverride) {
auto schema = unshredded_variant_schema();
schema[1].__isset.logicalType = false;
Expand Down
59 changes: 59 additions & 0 deletions be/test/format_v2/parquet/variant_column_reader_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,24 @@ ParquetColumnSchema shredded_array_schema() {
return schema;
}

ParquetColumnSchema shredded_fallback_only_array_schema() {
auto schema = unshredded_schema();
auto typed = std::make_unique<ParquetColumnSchema>();
typed->name = "typed_value";
typed->kind = ParquetColumnSchemaKind::LIST;
auto element = std::make_unique<ParquetColumnSchema>();
element->name = "element";
element->kind = ParquetColumnSchemaKind::STRUCT;
auto value = std::make_unique<ParquetColumnSchema>();
value->name = "value";
value->kind = ParquetColumnSchemaKind::PRIMITIVE;
value->type = std::make_shared<DataTypeString>();
element->children.push_back(std::move(value));
typed->children.push_back(std::move(element));
schema.children.push_back(std::move(typed));
return schema;
}

ParquetColumnSchema shredded_mixed_array_schema() {
auto schema = shredded_array_schema();
auto* element = schema.children.back()->children[0].get();
Expand Down Expand Up @@ -1737,6 +1755,47 @@ TEST(VariantColumnReaderTest, MaterializesShreddedArrayElements) {
EXPECT_EQ(value.array_at(1).get_int(), 4);
}

TEST(VariantColumnReaderTest, MaterializesFallbackOnlyArrayFromRequiredValueLeaf) {
const StringRef metadata(VARIANT_EMPTY_METADATA.data(), VARIANT_EMPTY_METADATA.size());
const std::array<char, 2> first_value {
static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
<< VARIANT_VALUE_HEADER_SHIFT),
3};
const std::array<char, 2> second_value {
static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
<< VARIANT_VALUE_HEADER_SHIFT),
4};

auto values = ColumnString::create();
values->insert_data(first_value.data(), first_value.size());
values->insert_data(second_value.data(), second_value.size());
MutableColumns wrapper_fields;
wrapper_fields.push_back(std::move(values));
auto wrappers = ColumnStruct::create(std::move(wrapper_fields));
auto elements = ColumnNullable::create(std::move(wrappers), ColumnUInt8::create(2, 0));
auto offsets = ColumnArray::ColumnOffsets::create();
offsets->insert_value(2);
auto array = ColumnArray::create(std::move(elements), std::move(offsets));

const std::array<char, 1> ignored {0};
MutableColumns root_fields;
root_fields.push_back(nullable_strings({metadata}, {0}));
root_fields.push_back(nullable_strings({{ignored.data(), 0}}, {1}));
root_fields.push_back(ColumnNullable::create(std::move(array), ColumnUInt8::create(1, 0)));
auto physical = root_wrapper(std::move(root_fields));

auto output = make_nullable(std::make_shared<DataTypeVariantV2>())->create_column();
const auto status =
materialize_variant_rows(shredded_fallback_only_array_schema(), *physical, output);
ASSERT_TRUE(status.ok()) << status;
const auto& nullable = assert_cast<const ColumnNullable&>(*output);
const auto& variants = assert_cast<const ColumnVariantV2&>(nullable.get_nested_column());
const VariantRef value = variants.get_value_ref(0);
ASSERT_EQ(value.num_elements(), 2);
EXPECT_EQ(value.array_at(0).get_int(), 3);
EXPECT_EQ(value.array_at(1).get_int(), 4);
}

TEST(VariantColumnReaderTest, RejectsCorruptShreddedWrappersWithoutCrashing) {
const std::array<char, 2> int_seven {
static_cast<char>(static_cast<uint8_t>(VariantPrimitiveId::INT8)
Expand Down
Loading