diff --git a/be/src/format_v2/parquet/native_schema_desc.cpp b/be/src/format_v2/parquet/native_schema_desc.cpp index 28ac3da4ccf9f9..f9da3adfbefe80 100644 --- a/be/src/format_v2/parquet/native_schema_desc.cpp +++ b/be/src/format_v2/parquet/native_schema_desc.cpp @@ -286,7 +286,7 @@ class ScopedBoolOverride { Status validate_variant_layout(const NativeFieldSchema& group_field, std::optional 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); @@ -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) { @@ -360,12 +360,12 @@ Status validate_variant_layout(const NativeFieldSchema& group_field, std::function validate_typed_value; std::function 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; @@ -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 || diff --git a/be/src/format_v2/parquet/native_schema_desc.h b/be/src/format_v2/parquet/native_schema_desc.h index 38cd3bf0269d19..88757edb6eaa7b 100644 --- a/be/src/format_v2/parquet/native_schema_desc.h +++ b/be/src/format_v2/parquet/native_schema_desc.h @@ -89,7 +89,7 @@ struct NativeFieldSchema { Status validate_variant_layout(const NativeFieldSchema& group_field, std::optional 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 { diff --git a/be/test/format_v2/parquet/parquet_schema_test.cpp b/be/test/format_v2/parquet/parquet_schema_test.cpp index c475348edc030b..1193420b9ebfb8 100644 --- a/be/test/format_v2/parquet/parquet_schema_test.cpp +++ b/be/test/format_v2/parquet/parquet_schema_test.cpp @@ -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> 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 schema) { + NativeFieldDescriptor descriptor; + ASSERT_TRUE(descriptor.parse_from_thrift(schema).ok()); + std::vector> 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()) << 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()) << annotated_status; +} + TEST(ParquetSchemaTest, RejectsMalformedUnannotatedVariantOverride) { auto schema = unshredded_variant_schema(); schema[1].__isset.logicalType = false; diff --git a/be/test/format_v2/parquet/variant_column_reader_test.cpp b/be/test/format_v2/parquet/variant_column_reader_test.cpp index 9e2fc0774baf72..5e7a92734c2526 100644 --- a/be/test/format_v2/parquet/variant_column_reader_test.cpp +++ b/be/test/format_v2/parquet/variant_column_reader_test.cpp @@ -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(); + typed->name = "typed_value"; + typed->kind = ParquetColumnSchemaKind::LIST; + auto element = std::make_unique(); + element->name = "element"; + element->kind = ParquetColumnSchemaKind::STRUCT; + auto value = std::make_unique(); + value->name = "value"; + value->kind = ParquetColumnSchemaKind::PRIMITIVE; + value->type = std::make_shared(); + 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(); @@ -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 first_value { + static_cast(static_cast(VariantPrimitiveId::INT8) + << VARIANT_VALUE_HEADER_SHIFT), + 3}; + const std::array second_value { + static_cast(static_cast(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 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())->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(*output); + const auto& variants = assert_cast(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 int_seven { static_cast(static_cast(VariantPrimitiveId::INT8)