This is an automated email from the ASF dual-hosted git repository.
Gabriel39 pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 9ffd818bec3 [fix](parquet) Share Variant layout compatibility across
table formats (#69148)
9ffd818bec3 is described below
commit 9ffd818bec378f95e18a35c78955d9966d1d9467
Author: Gabriel <[email protected]>
AuthorDate: Fri Oct 16 15:57:18 2026 +0800
[fix](parquet) Share Variant layout compatibility across table formats
(#69148)
### What problem does this PR solve?
Related PR: #69127 (merged into branch-4.1)
Annotated Parquet Variant files may contain a required BYTE_ARRAY
fallback in an array element without a typed carrier. The same physical
layout was accepted through Paimon schema overrides but rejected for
VARIANT(1)-annotated files, including files read through Iceberg.
Port #69127: remove `allow_paimon_shredded_layout` and share the
existing compatibility for fallback-only arrays, optional shredded
metadata, and optional wrappers. Missing metadata and null array
elements remain invalid at materialization; object fields and arrays
with typed carriers still require optional fallbacks.
The schema acceptance tests call NativeFieldDescriptor directly because
diagnostic PR #69119 is still open. The file-reader regression still
covers actual Parquet footers. The diagnostic rejection fixture in
#69119 is updated separately to remain invalid after this compatibility
fix, so the PRs can merge in either order.
The port includes schema tests, a native file-reader regression, and two
entirely synthetic six-row Parquet fixtures with a reproducible
generator. No external data is included.
### Release note
Fix reading annotated Parquet Variant columns with compatible shredded
layouts previously accepted only through Paimon schema overrides.
### Check List (For Author)
- Test
- [x] Unit Test (ported coverage; target-branch execution pending CI)
- [x] Manual test: regenerated both fixtures byte-for-byte and verified
VARIANT(1), six rows and three row groups per file.
Repository clang-format 16 and `git diff --check` passed. BE unit tests
and full builds were not rerun locally for this port; `run buildall` is
requested. The original PR contains the prior focused ASAN and native
reader validation results.
- Behavior changed:
- [x] Yes. Annotated Variant columns share the existing shredded-layout
compatibility rules.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label
---
be/src/format_v2/parquet/native_schema_desc.cpp | 33 ++++----
be/src/format_v2/parquet/native_schema_desc.h | 3 +-
be/src/format_v2/parquet/parquet_column_schema.cpp | 2 +-
be/test/format_v2/parquet/parquet_reader_test.cpp | 70 +++++++++++++++++
be/test/format_v2/parquet/parquet_schema_test.cpp | 41 ++++++++--
.../iceberg/generate_variant_fallback_array.py | 84 +++++++++++++++++++++
.../variant_fallback_array_optional.parquet | Bin 0 -> 2940 bytes
.../variant_fallback_array_required.parquet | Bin 0 -> 2924 bytes
8 files changed, 204 insertions(+), 29 deletions(-)
diff --git a/be/src/format_v2/parquet/native_schema_desc.cpp
b/be/src/format_v2/parquet/native_schema_desc.cpp
index bcb3abcf91f..6a21ce165cd 100644
--- a/be/src/format_v2/parquet/native_schema_desc.cpp
+++ b/be/src/format_v2/parquet/native_schema_desc.cpp
@@ -285,8 +285,7 @@ private:
};
Status validate_variant_layout(const NativeFieldSchema& group_field,
- std::optional<int8_t> specification_version,
- bool allow_paimon_shredded_layout) {
+ std::optional<int8_t> specification_version) {
if (specification_version.has_value() && *specification_version != 1) {
return Status::NotSupported("Parquet Variant specification version {}
is not supported",
*specification_version);
@@ -328,11 +327,11 @@ Status validate_variant_layout(const NativeFieldSchema&
group_field,
group_field.name);
}
const auto metadata_repetition = metadata->parquet_schema.repetition_type;
- // Paimon makes every field in its shredded carrier optional. Restrict
that compatibility to
- // unannotated overrides; row materialization still rejects null metadata
for a non-null value.
+ // Shredded writers may declare metadata optional regardless of the
VARIANT annotation.
+ // Row materialization still rejects null metadata for a non-null Variant.
const bool valid_metadata_repetition =
metadata_repetition == tparquet::FieldRepetitionType::REQUIRED ||
- (allow_paimon_shredded_layout && typed_value != nullptr &&
+ (typed_value != nullptr &&
metadata_repetition == tparquet::FieldRepetitionType::OPTIONAL);
if (!metadata->children.empty() || metadata->physical_type !=
tparquet::Type::BYTE_ARRAY ||
!valid_metadata_repetition) {
@@ -364,15 +363,13 @@ Status validate_variant_layout(const NativeFieldSchema&
group_field,
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;
- // an absent object wrapper represents a missing key.
+ wrapper.parquet_schema.repetition_type ==
tparquet::FieldRepetitionType::OPTIONAL);
+ // An optional wrapper is a writer layout choice, not a table-format
identifier. A null
+ // object wrapper means an absent key; materialization still rejects
null array elements.
if (!valid_wrapper_repetition) {
- return Status::Corruption("Parquet Variant shredded wrapper {}
must be required",
- wrapper.name);
+ return Status::Corruption(
+ "Parquet Variant shredded wrapper {} must be required or
optional",
+ wrapper.name);
}
const NativeFieldSchema* fallback = nullptr;
const NativeFieldSchema* typed = nullptr;
@@ -408,11 +405,11 @@ 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;
+ // Both annotated and unannotated writers use a required fallback-only
array carrier.
+ // Even a null element has Variant null bytes; object fields must
still represent absence,
+ // and a sibling typed_value needs a nullable fallback to avoid
conflicting carriers.
+ const bool allow_required_fallback =
+ context == WrapperContext::ARRAY_ELEMENT && typed == nullptr;
const bool valid_fallback_repetition =
fallback != nullptr &&
fallback->parquet_schema.__isset.repetition_type &&
(fallback->parquet_schema.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 80683cf0efe..1f788bef999 100644
--- a/be/src/format_v2/parquet/native_schema_desc.h
+++ b/be/src/format_v2/parquet/native_schema_desc.h
@@ -91,8 +91,7 @@ struct NativeFieldSchema {
};
Status validate_variant_layout(const NativeFieldSchema& group_field,
- std::optional<int8_t> specification_version =
std::nullopt,
- bool allow_paimon_shredded_layout = false);
+ std::optional<int8_t> specification_version =
std::nullopt);
// V2 owns this schema tree and parser so footer/schema planning never invokes
the V1 reader path.
class NativeFieldDescriptor {
diff --git a/be/src/format_v2/parquet/parquet_column_schema.cpp
b/be/src/format_v2/parquet/parquet_column_schema.cpp
index 984e0259165..484dad564ef 100644
--- a/be/src/format_v2/parquet/parquet_column_schema.cpp
+++ b/be/src/format_v2/parquet/parquet_column_schema.cpp
@@ -284,7 +284,7 @@ Status apply_variant_schema_override(const
NativeFieldSchema& native_schema,
return Status::Corruption("Parquet Variant {} must use a group
carrier",
native_schema.name);
}
- RETURN_IF_ERROR(validate_variant_layout(native_schema, std::nullopt,
true));
+ RETURN_IF_ERROR(validate_variant_layout(native_schema));
field->variant_physical_type = field->type;
DataTypePtr variant_type = std::make_shared<DataTypeVariantV2>();
field->type = field->type->is_nullable() ?
make_nullable(std::move(variant_type))
diff --git a/be/test/format_v2/parquet/parquet_reader_test.cpp
b/be/test/format_v2/parquet/parquet_reader_test.cpp
index 5aa1aaf16d7..cf222bbe9b5 100644
--- a/be/test/format_v2/parquet/parquet_reader_test.cpp
+++ b/be/test/format_v2/parquet/parquet_reader_test.cpp
@@ -2879,6 +2879,76 @@ TEST(ParquetVariantProjectionTest,
FinalizesNestedVariantProjectionPerRowGroup)
EXPECT_TRUE(repeated.children[0].project_all_children);
}
+TEST_F(NewParquetReaderTest, ReadsAnnotatedVariantFallbackArrayLayouts) {
+ const char* source_root = std::getenv("ROOT");
+ ASSERT_NE(source_root, nullptr);
+ for (const auto* layout : {"required", "optional"}) {
+ SCOPED_TRACE(layout);
+ // Synthetic fixtures preserve the required binary element carrier
without external data.
+ _file_path = std::string(source_root) +
+
"/regression-test/data/external_table_p0/iceberg/variant_fallback_array_" +
+ layout + ".parquet";
+ auto reader = create_reader();
+ reader->set_batch_size(2);
+ RuntimeState state {TQueryOptions(), TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 2);
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->non_predicate_columns.push_back(
+ format::LocalColumnIndex::local(schema[1].local_id));
+
request->local_positions.emplace(format::LocalColumnId(schema[1].local_id),
+ format::LocalIndex(0));
+ ASSERT_TRUE(reader->open(request).ok());
+ size_t rows = 0;
+ bool eof = false;
+ while (!eof) {
+ Block block;
+ block.insert({schema[1].type->create_column(), schema[1].type,
"payload"});
+ size_t batch_rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &batch_rows, &eof).ok());
+ const auto& nullable =
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(0).column);
+ const auto& variants =
+ assert_cast<const
ColumnVariantV2&>(nullable.get_nested_column());
+ for (size_t row = 0; row < batch_rows; ++row, ++rows) {
+ if (rows == 2) {
+ EXPECT_TRUE(nullable.is_null_at(row));
+ continue;
+ }
+ ASSERT_FALSE(nullable.is_null_at(row));
+ const auto value = variants.get_value_ref(row);
+ if (rows == 3) {
+ EXPECT_TRUE(value.is_null());
+ continue;
+ }
+ VariantRef events;
+ if (rows == 5) {
+ EXPECT_FALSE(value.object_find(StringRef("events"),
&events));
+ EXPECT_EQ(value.num_elements(), 0);
+ continue;
+ }
+ ASSERT_TRUE(value.object_find(StringRef("events"), &events));
+ if (rows == 0) {
+ ASSERT_EQ(events.num_elements(), 3);
+ EXPECT_EQ(events.array_at(0).get_int(), 3);
+ EXPECT_TRUE(events.array_at(1).is_null());
+ EXPECT_EQ(events.array_at(2).get_int(), 4);
+ } else if (rows == 1) {
+ EXPECT_EQ(events.num_elements(), 0);
+ } else {
+ ASSERT_EQ(rows, 4);
+ ASSERT_EQ(events.num_elements(), 1);
+ EXPECT_TRUE(events.array_at(0).is_null());
+ }
+ }
+ }
+ EXPECT_EQ(rows, 6);
+ ASSERT_TRUE(reader->close().ok());
+ }
+}
+
TEST_F(NewParquetReaderTest, ReadsFullyShreddedVariantTypedLeafProjection) {
const char* source_root = std::getenv("ROOT");
ASSERT_NE(source_root, nullptr);
diff --git a/be/test/format_v2/parquet/parquet_schema_test.cpp
b/be/test/format_v2/parquet/parquet_schema_test.cpp
index f7f96ea2c7e..97a10c3bd69 100644
--- a/be/test/format_v2/parquet/parquet_schema_test.cpp
+++ b/be/test/format_v2/parquet/parquet_schema_test.cpp
@@ -337,7 +337,33 @@ TEST(ParquetSchemaTest,
AppliesPaimonFallbackOnlyArrayWithRequiredValue) {
EXPECT_EQ(fields[0]->kind, ParquetColumnSchemaKind::VARIANT);
}
-TEST(ParquetSchemaTest,
RejectsRequiredShreddedFallbackOutsidePaimonFallbackOnlyArray) {
+TEST(ParquetSchemaTest, AcceptsAnnotatedFallbackOnlyArrayWithRequiredValue) {
+ auto schema = shredded_array_variant_schema(true, false);
+ schema[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ NativeFieldDescriptor descriptor;
+ ASSERT_TRUE(descriptor.parse_from_thrift(schema).ok());
+}
+
+TEST(ParquetSchemaTest, AcceptsAnnotatedShreddedVariantWithOptionalMetadata) {
+ auto schema = shredded_object_variant_schema();
+ schema[2].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
+ NativeFieldDescriptor descriptor;
+ ASSERT_TRUE(descriptor.parse_from_thrift(schema).ok());
+}
+
+TEST(ParquetSchemaTest, AcceptsAnnotatedShreddedVariantWithOptionalWrappers) {
+ auto object_schema = shredded_object_variant_schema();
+
object_schema[5].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
+ NativeFieldDescriptor descriptor;
+ ASSERT_TRUE(descriptor.parse_from_thrift(object_schema).ok());
+
+ auto array_schema = shredded_array_variant_schema(true, false);
+
array_schema[6].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
+
array_schema[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
+ ASSERT_TRUE(descriptor.parse_from_thrift(array_schema).ok());
+}
+
+TEST(ParquetSchemaTest,
RejectsRequiredShreddedFallbackOutsideFallbackOnlyArray) {
const auto expect_override_corruption =
[](std::vector<tparquet::SchemaElement> schema) {
NativeFieldDescriptor descriptor;
ASSERT_TRUE(descriptor.parse_from_thrift(schema).ok());
@@ -364,7 +390,7 @@ TEST(ParquetSchemaTest,
RejectsRequiredShreddedFallbackOutsidePaimonFallbackOnly
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);
+ auto annotated_array = shredded_array_variant_schema(true, true);
annotated_array[7].__set_repetition_type(tparquet::FieldRepetitionType::REQUIRED);
NativeFieldDescriptor descriptor;
const auto annotated_status =
descriptor.parse_from_thrift(annotated_array);
@@ -386,11 +412,11 @@ TEST(ParquetSchemaTest,
RejectsMalformedUnannotatedVariantOverride) {
EXPECT_NE(status.to_string().find("unexpected child"), std::string::npos);
}
-TEST(ParquetSchemaTest, RejectsOptionalMetadataOutsidePaimonShreddedOverride) {
- auto annotated_shredded = shredded_object_variant_schema();
-
annotated_shredded[2].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
+TEST(ParquetSchemaTest, RejectsOptionalMetadataForUnshreddedVariant) {
+ auto annotated_unshredded = unshredded_variant_schema();
+
annotated_unshredded[2].__set_repetition_type(tparquet::FieldRepetitionType::OPTIONAL);
NativeFieldDescriptor descriptor;
- const auto annotated_status =
descriptor.parse_from_thrift(annotated_shredded);
+ const auto annotated_status =
descriptor.parse_from_thrift(annotated_unshredded);
EXPECT_TRUE(annotated_status.is<ErrorCode::CORRUPTION>()) <<
annotated_status;
auto unannotated_unshredded = unshredded_variant_schema();
@@ -510,8 +536,7 @@ TEST(ParquetSchemaTest,
NativeVariantValidatesEveryShreddedWrapperAndScalar) {
const auto optional_wrapper_status =
descriptor.parse_from_thrift(shredded_object_variant_schema(true,
false));
- EXPECT_TRUE(optional_wrapper_status.is<ErrorCode::CORRUPTION>()) <<
optional_wrapper_status;
- EXPECT_NE(optional_wrapper_status.to_string().find("wrapper"),
std::string::npos);
+ EXPECT_TRUE(optional_wrapper_status.ok()) << optional_wrapper_status;
}
TEST(ParquetSchemaTest, NativeVariantRejectsDuplicateObjectFieldNames) {
diff --git
a/regression-test/data/external_table_p0/iceberg/generate_variant_fallback_array.py
b/regression-test/data/external_table_p0/iceberg/generate_variant_fallback_array.py
new file mode 100644
index 00000000000..7ebc80e80f3
--- /dev/null
+++
b/regression-test/data/external_table_p0/iceberg/generate_variant_fallback_array.py
@@ -0,0 +1,84 @@
+# 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.
+
+"""Generate synthetic Variant fixtures; no external data is read.
+
+Run from the repository root with:
+ uv run --with pyarrow==24.0.0 --with thriftpy2==0.5.3 python \
+
regression-test/data/external_table_p0/iceberg/generate_variant_fallback_array.py
+"""
+from pathlib import Path
+import struct
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import thriftpy2
+from thriftpy2.protocol import TCompactProtocolFactory
+from thriftpy2.transport import TMemoryBuffer
+
+ROOT = Path(__file__).resolve().parents[4]
+PARQUET = thriftpy2.load(str(ROOT / "gensrc/thrift/parquet.thrift"),
+ module_name="fixture_parquet_thrift")
+# A one-key metadata dictionary containing only the synthetic key "events".
+METADATA = bytes([1, 1, 0, 6]) + b"events"
+
+
+def generate(optional_wrappers):
+ element = pa.struct([pa.field("value", pa.binary(), nullable=False)])
+ values = pa.list_(pa.field("element", element, nullable=optional_wrappers))
+ wrapper = pa.struct([pa.field("value", pa.binary()),
+ pa.field("typed_value", values)])
+ typed = pa.struct([pa.field("events", wrapper,
nullable=optional_wrappers)])
+ variant = pa.struct([pa.field("metadata", pa.binary(),
nullable=optional_wrappers),
+ pa.field("value", pa.binary()),
pa.field("typed_value", typed)])
+ schema = pa.schema([pa.field("id", pa.int32(), nullable=False),
+ pa.field("payload", variant)])
+
+ def array(items):
+ return {"metadata": METADATA, "value": None, "typed_value": {
+ "events": {"value": None, "typed_value": [{"value": v} for v in
items]}}}
+
+ # 0x0c is the Variant int8 header. 0x00 is Variant null, not a Parquet
null.
+ payloads = [array([b"\x0c\x03", b"\x00", b"\x0c\x04"]), array([]), None,
+ {"metadata": METADATA, "value": b"\x00", "typed_value": None},
+ array([b"\x00"]),
+ {"metadata": METADATA, "value": None, "typed_value": {
+ "events": None if optional_wrappers else {"value": None,
"typed_value": None}}}]
+ rows = [{"id": i + 1, "payload": value} for i, value in
enumerate(payloads)]
+ suffix = "optional" if optional_wrappers else "required"
+ path = Path(__file__).with_name(f"variant_fallback_array_{suffix}.parquet")
+ pq.write_table(pa.Table.from_pylist(rows, schema), path,
compression="zstd",
+ row_group_size=2, store_schema=False)
+ data = path.read_bytes()
+ length = struct.unpack("<I", data[-8:-4])[0]
+ footer_start = len(data) - 8 - length
+ metadata = PARQUET.FileMetaData()
+
metadata.read(TCompactProtocolFactory().get_protocol(TMemoryBuffer(data[footer_start:-8])))
+ # Arrow writes the physical carrier; add the standard logical annotation
to its footer.
+ payload = next(field for field in metadata.schema if field.name ==
"payload")
+ payload.logicalType =
PARQUET.LogicalType(VARIANT=PARQUET.VariantType(specification_version=1))
+ transport = TMemoryBuffer()
+ metadata.write(TCompactProtocolFactory().get_protocol(transport))
+ footer = transport.getvalue()
+ path.write_bytes(data[:footer_start] + footer + struct.pack("<I",
len(footer)) + b"PAR1")
+ assert pq.read_table(path).num_rows == 6
+ print(path.name, path.stat().st_size)
+
+
+if __name__ == "__main__":
+ generate(False)
+ generate(True)
diff --git
a/regression-test/data/external_table_p0/iceberg/variant_fallback_array_optional.parquet
b/regression-test/data/external_table_p0/iceberg/variant_fallback_array_optional.parquet
new file mode 100644
index 00000000000..c0703b34ce1
Binary files /dev/null and
b/regression-test/data/external_table_p0/iceberg/variant_fallback_array_optional.parquet
differ
diff --git
a/regression-test/data/external_table_p0/iceberg/variant_fallback_array_required.parquet
b/regression-test/data/external_table_p0/iceberg/variant_fallback_array_required.parquet
new file mode 100644
index 00000000000..2a267c821a1
Binary files /dev/null and
b/regression-test/data/external_table_p0/iceberg/variant_fallback_array_required.parquet
differ
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
