github-actions[bot] commented on code in PR #66302:
URL: https://github.com/apache/doris/pull/66302#discussion_r3682593700
##########
be/src/format_v2/column_mapper.cpp:
##########
@@ -2004,6 +2125,7 @@ Status
TableColumnMapper::_create_mapping_for_column(const ColumnDefinition& tab
mapping->global_index = global_index;
mapping->table_column_name = table_column.name;
mapping->table_type = table_column.type;
+ mapping->variant_access_paths = table_column.variant_access_paths;
Review Comment:
[P1] Propagate nested Variant paths through recursive mappings
This assignment runs only while creating a root mapping. `AccessPathParser`
attaches `{"x"}` to the nested `info.payload` `ColumnDefinition`, but recursive
child construction at `_create_direct_mapping()` copies only its name and type.
`build_complex_projection()` then reaches that Variant mapping with no child
mappings and projects the complete wrapper, so the new regression's
`VariantLeafProjections > 0` assertion cannot pass. Propagate
`variant_access_paths` into recursive mappings and invoke the Variant
projection builder at nested STRUCT/ARRAY/MAP terminals.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -55,6 +56,209 @@ static bool is_map_node(const tparquet::SchemaElement&
schema) {
(schema.__isset.logicalType && schema.logicalType.__isset.MAP);
}
+static bool is_variant_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.logicalType && schema.logicalType.__isset.VARIANT;
+}
+
+class ScopedBoolOverride {
+public:
+ ScopedBoolOverride(bool& target, bool value) : _target(target),
_original(target) {
+ _target = value;
+ }
+ ~ScopedBoolOverride() { _target = _original; }
+
+private:
+ bool& _target;
+ bool _original;
+};
+
+static Status validate_variant_layout(const tparquet::SchemaElement&
group_schema,
+ const NativeFieldSchema& group_field) {
+ const auto& annotation = group_schema.logicalType.VARIANT;
+ if (annotation.__isset.specification_version &&
annotation.specification_version != 1) {
+ return Status::NotSupported("Parquet Variant specification version {}
is not supported",
+ annotation.specification_version);
+ }
+ if (group_field.children.size() < 2 || group_field.children.size() > 3) {
+ return Status::Corruption(
+ "Parquet Variant {} must contain metadata, value, and optional
typed_value",
+ group_schema.name);
+ }
+
+ const NativeFieldSchema* metadata = nullptr;
+ const NativeFieldSchema* value = nullptr;
+ const NativeFieldSchema* typed_value = nullptr;
+ for (const auto& child : group_field.children) {
+ const NativeFieldSchema** target = nullptr;
+ if (child.name == "metadata") {
+ target = &metadata;
+ } else if (child.name == "value") {
+ target = &value;
+ } else if (child.name == "typed_value") {
+ target = &typed_value;
+ } else {
+ return Status::Corruption("Parquet Variant {} has unexpected child
{}",
+ group_schema.name, child.name);
+ }
+ if (*target != nullptr) {
+ return Status::Corruption("Parquet Variant {} has duplicate child
{}",
+ group_schema.name, child.name);
+ }
+ *target = &child;
+ }
+ if (metadata == nullptr || value == nullptr) {
+ return Status::Corruption("Parquet Variant {} requires metadata and
value children",
+ group_schema.name);
+ }
+ if (!metadata->children.empty() || metadata->physical_type !=
tparquet::Type::BYTE_ARRAY ||
+ metadata->parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::REQUIRED) {
+ return Status::Corruption("Parquet Variant {} metadata must be a
required BYTE_ARRAY",
+ group_schema.name);
+ }
+ const auto expected_value_repetition = typed_value == nullptr
+ ?
tparquet::FieldRepetitionType::REQUIRED
+ :
tparquet::FieldRepetitionType::OPTIONAL;
+ // SQL nullability belongs to the outer Variant group. Only shredding
makes value optional,
+ // because typed_value may carry all or part of the logical value instead.
+ if (!value->children.empty() || value->physical_type !=
tparquet::Type::BYTE_ARRAY ||
+ value->parquet_schema.repetition_type != expected_value_repetition) {
+ return Status::Corruption("Parquet Variant {} value must be a {}
BYTE_ARRAY",
+ group_schema.name,
+ typed_value == nullptr ? "required" :
"optional");
+ }
+ if (typed_value != nullptr &&
+ typed_value->parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::OPTIONAL) {
+ return Status::Corruption("Parquet Variant {} typed_value must be
optional",
+ group_schema.name);
+ }
+
+ enum class WrapperContext : uint8_t { OBJECT_FIELD, ARRAY_ELEMENT };
+ std::function<Status(const NativeFieldSchema&)> validate_typed_value;
+ std::function<Status(const NativeFieldSchema&, WrapperContext)>
validate_wrapper;
+ validate_wrapper = [&](const NativeFieldSchema& wrapper, WrapperContext
context) -> Status {
+ if (!wrapper.parquet_schema.__isset.repetition_type ||
+ wrapper.parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::REQUIRED) {
+ return Status::Corruption("Parquet Variant shredded wrapper {}
must be required",
+ wrapper.name);
+ }
+ const NativeFieldSchema* fallback = nullptr;
+ const NativeFieldSchema* typed = nullptr;
+ for (const auto& child : wrapper.children) {
+ if (child.name == "value") {
+ if (fallback != nullptr) {
+ return Status::Corruption(
+ "Parquet Variant wrapper {} has duplicate value
child", wrapper.name);
+ }
+ fallback = &child;
+ } else if (child.name == "typed_value") {
+ if (typed != nullptr) {
+ return Status::Corruption(
+ "Parquet Variant wrapper {} has duplicate
typed_value child",
+ wrapper.name);
+ }
+ typed = &child;
+ } else {
+ return Status::Corruption("Parquet Variant wrapper {} has
unexpected child {}",
+ wrapper.name, child.name);
+ }
+ }
+ if (fallback == nullptr && typed == nullptr) {
+ return Status::Corruption(
+ "Parquet Variant shredded wrapper {} requires at least one
of value or "
+ "typed_value",
+ wrapper.name);
+ }
+ // Object fields always retain the fallback value carrier; only
typed_value is optional.
+ // Array elements may omit either carrier when every element uses the
remaining one.
+ if (context == WrapperContext::OBJECT_FIELD && fallback == nullptr) {
+ return Status::Corruption(
+ "Parquet Variant object wrapper {} requires an optional
value child",
+ wrapper.name);
+ }
+ 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)) {
+ return Status::Corruption(
+ "Parquet Variant wrapper {} value must be an optional
BYTE_ARRAY",
+ wrapper.name);
+ }
+ if (typed != nullptr) {
+ if (!typed->parquet_schema.__isset.repetition_type ||
+ typed->parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::OPTIONAL) {
+ return Status::Corruption("Parquet Variant wrapper {}
typed_value must be optional",
+ wrapper.name);
+ }
+ return validate_typed_value(*typed);
+ }
+ return Status::OK();
+ };
+ validate_typed_value = [&](const NativeFieldSchema& typed) -> Status {
+ if (!typed.unsupported_reason.empty()) {
+ return Status::NotSupported("Parquet Variant typed value {} is not
supported: {}",
+ typed.name, typed.unsupported_reason);
+ }
+ if (typed.children.empty()) {
+ const auto& physical = typed.parquet_schema;
+ if (physical.__isset.logicalType &&
physical.logicalType.__isset.INTEGER &&
+ !physical.logicalType.INTEGER.isSigned) {
+ return Status::Corruption(
+ "Parquet Variant unsigned integers are not valid typed
values");
+ }
+ if (physical.__isset.converted_type &&
+ (physical.converted_type == tparquet::ConvertedType::UINT_8 ||
+ physical.converted_type == tparquet::ConvertedType::UINT_16 ||
+ physical.converted_type == tparquet::ConvertedType::UINT_32 ||
+ physical.converted_type == tparquet::ConvertedType::UINT_64))
{
+ return Status::Corruption(
+ "Parquet Variant unsigned integers are not valid typed
values");
+ }
+ if (physical.__isset.logicalType &&
physical.logicalType.__isset.TIME) {
+ const auto& time = physical.logicalType.TIME;
+ // Variant v1 has one canonical TIME representation: local
wall-clock MICROS.
+ // Accepting adjusted or lower-precision forms would make
projection-dependent
+ // reconstruction disagree with the canonical Variant value.
+ if (time.isAdjustedToUTC) {
+ return Status::Corruption(
+ "Parquet Variant TIME must have
isAdjustedToUTC=false");
+ }
+ if (!time.unit.__isset.MICROS) {
+ return Status::Corruption(
+ "Parquet Variant TIME(MILLIS) is not supported;
use TIME(MICROS)");
+ }
+ }
+ if (physical.__isset.converted_type &&
+ physical.converted_type ==
tparquet::ConvertedType::TIME_MILLIS) {
+ return Status::Corruption(
+ "Parquet Variant TIME(MILLIS) is not supported; use
TIME(MICROS)");
+ }
+ if (physical.__isset.logicalType &&
physical.logicalType.__isset.TIMESTAMP &&
+ physical.logicalType.TIMESTAMP.unit.__isset.NANOS) {
+ // Reject at schema open so full reconstruction and direct
typed-leaf access have
+ // the same precision contract instead of diverging after
projection planning.
+ return Status::NotSupported("Parquet Variant TIMESTAMP(NANOS)
is not supported");
+ }
+ return Status::OK();
+ }
+
+ const PrimitiveType primitive =
remove_nullable(typed.data_type)->get_primitive_type();
+ if (primitive == TYPE_STRUCT) {
+ for (const auto& child : typed.children) {
Review Comment:
[P1] Reject duplicate shredded object fields before planning
This loop validates wrappers independently but never requires unique names.
With two `typed_value` wrappers named `x`, full materialization skips an absent
first wrapper and emits the populated second one, while
`find_file_child_by_name()`/`find_child()` bind a direct `v['x']` projection to
the first wrapper and return NULL; if both are populated, the full path instead
errors on the duplicate key. The same file therefore produces
query-shape-dependent values. Reject duplicate wrapper names here before
projection planning, with first-absent/second-present and both-present negative
fixtures.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -55,6 +56,209 @@ static bool is_map_node(const tparquet::SchemaElement&
schema) {
(schema.__isset.logicalType && schema.logicalType.__isset.MAP);
}
+static bool is_variant_node(const tparquet::SchemaElement& schema) {
+ return schema.__isset.logicalType && schema.logicalType.__isset.VARIANT;
+}
+
+class ScopedBoolOverride {
+public:
+ ScopedBoolOverride(bool& target, bool value) : _target(target),
_original(target) {
+ _target = value;
+ }
+ ~ScopedBoolOverride() { _target = _original; }
+
+private:
+ bool& _target;
+ bool _original;
+};
+
+static Status validate_variant_layout(const tparquet::SchemaElement&
group_schema,
+ const NativeFieldSchema& group_field) {
+ const auto& annotation = group_schema.logicalType.VARIANT;
+ if (annotation.__isset.specification_version &&
annotation.specification_version != 1) {
+ return Status::NotSupported("Parquet Variant specification version {}
is not supported",
+ annotation.specification_version);
+ }
+ if (group_field.children.size() < 2 || group_field.children.size() > 3) {
+ return Status::Corruption(
+ "Parquet Variant {} must contain metadata, value, and optional
typed_value",
+ group_schema.name);
+ }
+
+ const NativeFieldSchema* metadata = nullptr;
+ const NativeFieldSchema* value = nullptr;
+ const NativeFieldSchema* typed_value = nullptr;
+ for (const auto& child : group_field.children) {
+ const NativeFieldSchema** target = nullptr;
+ if (child.name == "metadata") {
+ target = &metadata;
+ } else if (child.name == "value") {
+ target = &value;
+ } else if (child.name == "typed_value") {
+ target = &typed_value;
+ } else {
+ return Status::Corruption("Parquet Variant {} has unexpected child
{}",
+ group_schema.name, child.name);
+ }
+ if (*target != nullptr) {
+ return Status::Corruption("Parquet Variant {} has duplicate child
{}",
+ group_schema.name, child.name);
+ }
+ *target = &child;
+ }
+ if (metadata == nullptr || value == nullptr) {
+ return Status::Corruption("Parquet Variant {} requires metadata and
value children",
+ group_schema.name);
+ }
+ if (!metadata->children.empty() || metadata->physical_type !=
tparquet::Type::BYTE_ARRAY ||
+ metadata->parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::REQUIRED) {
+ return Status::Corruption("Parquet Variant {} metadata must be a
required BYTE_ARRAY",
+ group_schema.name);
+ }
+ const auto expected_value_repetition = typed_value == nullptr
+ ?
tparquet::FieldRepetitionType::REQUIRED
+ :
tparquet::FieldRepetitionType::OPTIONAL;
+ // SQL nullability belongs to the outer Variant group. Only shredding
makes value optional,
+ // because typed_value may carry all or part of the logical value instead.
+ if (!value->children.empty() || value->physical_type !=
tparquet::Type::BYTE_ARRAY ||
+ value->parquet_schema.repetition_type != expected_value_repetition) {
+ return Status::Corruption("Parquet Variant {} value must be a {}
BYTE_ARRAY",
+ group_schema.name,
+ typed_value == nullptr ? "required" :
"optional");
+ }
+ if (typed_value != nullptr &&
+ typed_value->parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::OPTIONAL) {
+ return Status::Corruption("Parquet Variant {} typed_value must be
optional",
+ group_schema.name);
+ }
+
+ enum class WrapperContext : uint8_t { OBJECT_FIELD, ARRAY_ELEMENT };
+ std::function<Status(const NativeFieldSchema&)> validate_typed_value;
+ std::function<Status(const NativeFieldSchema&, WrapperContext)>
validate_wrapper;
+ validate_wrapper = [&](const NativeFieldSchema& wrapper, WrapperContext
context) -> Status {
+ if (!wrapper.parquet_schema.__isset.repetition_type ||
+ wrapper.parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::REQUIRED) {
+ return Status::Corruption("Parquet Variant shredded wrapper {}
must be required",
+ wrapper.name);
+ }
+ const NativeFieldSchema* fallback = nullptr;
+ const NativeFieldSchema* typed = nullptr;
+ for (const auto& child : wrapper.children) {
+ if (child.name == "value") {
+ if (fallback != nullptr) {
+ return Status::Corruption(
+ "Parquet Variant wrapper {} has duplicate value
child", wrapper.name);
+ }
+ fallback = &child;
+ } else if (child.name == "typed_value") {
+ if (typed != nullptr) {
+ return Status::Corruption(
+ "Parquet Variant wrapper {} has duplicate
typed_value child",
+ wrapper.name);
+ }
+ typed = &child;
+ } else {
+ return Status::Corruption("Parquet Variant wrapper {} has
unexpected child {}",
+ wrapper.name, child.name);
+ }
+ }
+ if (fallback == nullptr && typed == nullptr) {
+ return Status::Corruption(
+ "Parquet Variant shredded wrapper {} requires at least one
of value or "
+ "typed_value",
+ wrapper.name);
+ }
+ // Object fields always retain the fallback value carrier; only
typed_value is optional.
+ // Array elements may omit either carrier when every element uses the
remaining one.
+ if (context == WrapperContext::OBJECT_FIELD && fallback == nullptr) {
+ return Status::Corruption(
+ "Parquet Variant object wrapper {} requires an optional
value child",
+ wrapper.name);
+ }
+ 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)) {
+ return Status::Corruption(
+ "Parquet Variant wrapper {} value must be an optional
BYTE_ARRAY",
+ wrapper.name);
+ }
+ if (typed != nullptr) {
+ if (!typed->parquet_schema.__isset.repetition_type ||
+ typed->parquet_schema.repetition_type !=
tparquet::FieldRepetitionType::OPTIONAL) {
+ return Status::Corruption("Parquet Variant wrapper {}
typed_value must be optional",
+ wrapper.name);
+ }
+ return validate_typed_value(*typed);
+ }
+ return Status::OK();
+ };
+ validate_typed_value = [&](const NativeFieldSchema& typed) -> Status {
+ if (!typed.unsupported_reason.empty()) {
+ return Status::NotSupported("Parquet Variant typed value {} is not
supported: {}",
+ typed.name, typed.unsupported_reason);
+ }
+ if (typed.children.empty()) {
+ const auto& physical = typed.parquet_schema;
+ if (physical.__isset.logicalType &&
physical.logicalType.__isset.INTEGER &&
+ !physical.logicalType.INTEGER.isSigned) {
+ return Status::Corruption(
+ "Parquet Variant unsigned integers are not valid typed
values");
+ }
+ if (physical.__isset.converted_type &&
+ (physical.converted_type == tparquet::ConvertedType::UINT_8 ||
+ physical.converted_type == tparquet::ConvertedType::UINT_16 ||
+ physical.converted_type == tparquet::ConvertedType::UINT_32 ||
+ physical.converted_type == tparquet::ConvertedType::UINT_64))
{
+ return Status::Corruption(
+ "Parquet Variant unsigned integers are not valid typed
values");
+ }
+ if (physical.__isset.logicalType &&
physical.logicalType.__isset.TIME) {
+ const auto& time = physical.logicalType.TIME;
+ // Variant v1 has one canonical TIME representation: local
wall-clock MICROS.
+ // Accepting adjusted or lower-precision forms would make
projection-dependent
+ // reconstruction disagree with the canonical Variant value.
+ if (time.isAdjustedToUTC) {
+ return Status::Corruption(
+ "Parquet Variant TIME must have
isAdjustedToUTC=false");
+ }
+ if (!time.unit.__isset.MICROS) {
+ return Status::Corruption(
+ "Parquet Variant TIME(MILLIS) is not supported;
use TIME(MICROS)");
+ }
+ }
+ if (physical.__isset.converted_type &&
+ physical.converted_type ==
tparquet::ConvertedType::TIME_MILLIS) {
+ return Status::Corruption(
+ "Parquet Variant TIME(MILLIS) is not supported; use
TIME(MICROS)");
+ }
+ if (physical.__isset.logicalType &&
physical.logicalType.__isset.TIMESTAMP &&
+ physical.logicalType.TIMESTAMP.unit.__isset.NANOS) {
+ // Reject at schema open so full reconstruction and direct
typed-leaf access have
+ // the same precision contract instead of diverging after
projection planning.
+ return Status::NotSupported("Parquet Variant TIMESTAMP(NANOS)
is not supported");
+ }
+ return Status::OK();
Review Comment:
[P1] Enforce the Variant typed-value type matrix
This primitive branch accepts every type that generic Parquet inference can
represent, but the [Variant shredding
contract](https://parquet.apache.org/docs/file-format/types/variantshredding/)
has an exact physical/logical pair matrix. Consequently FLOAT16,
JSON/ENUM/BSON, INT96, unannotated FIXED_LEN_BYTE_ARRAY, and mismatched
integer/decimal pairs reach `append_typed_scalar()` and are re-encoded as
plausible float/binary/timestamp/decimal Variant values instead of being
rejected. That silently changes the source value's logical identity. Please
validate the complete matrix here, including physical width, annotation,
signedness, fixed length, precision/scale, and time unit, and add negative
fixtures for unsupported pairs.
##########
be/src/format_v2/parquet/parquet_reader.cpp:
##########
@@ -57,6 +58,99 @@ struct ParquetReaderScanState {
bool enable_strict_mode = false;
};
+const ParquetColumnSchema* projected_schema_child(const ParquetColumnSchema&
schema,
+ int32_t local_id) {
+ const auto child_it = std::ranges::find_if(
+ schema.children, [local_id](const auto& child) { return
child->local_id == local_id; });
+ return child_it == schema.children.end() ? nullptr : child_it->get();
+}
+
+const ParquetColumnSchema* schema_child_by_name(const ParquetColumnSchema&
schema,
+ std::string_view name) {
+ const auto child_it = std::ranges::find_if(
+ schema.children, [name](const auto& child) { return child->name ==
name; });
+ return child_it == schema.children.end() ? nullptr : child_it->get();
+}
+
+bool collect_variant_residual_leaf_ids(const ParquetColumnSchema& schema,
+ const format::LocalColumnIndex&
projection,
+ std::vector<int>* residual_leaf_ids) {
+ DORIS_CHECK(residual_leaf_ids != nullptr);
+ const auto* value = schema_child_by_name(schema, "value");
+ const auto* typed_value = schema_child_by_name(schema, "typed_value");
+ if (value != nullptr && typed_value != nullptr) {
+ if (value->kind != ParquetColumnSchemaKind::PRIMITIVE ||
value->leaf_column_id < 0) {
+ return false;
+ }
+ residual_leaf_ids->push_back(value->leaf_column_id);
+ }
+ for (const auto& child_projection : projection.children) {
+ const auto* child = projected_schema_child(schema,
child_projection.local_id());
+ if (child == nullptr ||
+ !collect_variant_residual_leaf_ids(*child, child_projection,
residual_leaf_ids)) {
+ return false;
+ }
+ }
+ return true;
+}
+
+bool detail::variant_projection_is_fully_shredded(const
tparquet::FileMetaData& metadata,
+ const ParquetColumnSchema&
schema,
+ const
format::LocalColumnIndex& projection) {
+ if (schema.kind != ParquetColumnSchemaKind::VARIANT ||
+ !format::is_partial_projection(&projection)) {
+ return false;
+ }
+ std::vector<int> residual_leaf_ids;
+ if (!collect_variant_residual_leaf_ids(schema, projection,
&residual_leaf_ids)) {
+ return false;
+ }
+ std::ranges::sort(residual_leaf_ids);
+ residual_leaf_ids.erase(std::unique(residual_leaf_ids.begin(),
residual_leaf_ids.end()),
+ residual_leaf_ids.end());
+ for (const auto& row_group : metadata.row_groups) {
+ for (const int leaf_id : residual_leaf_ids) {
+ if (leaf_id < 0 || leaf_id >=
static_cast<int>(row_group.columns.size())) {
+ return false;
+ }
+ const auto& chunk = row_group.columns[leaf_id];
+ if (!chunk.__isset.meta_data ||
!chunk.meta_data.__isset.statistics ||
+ !chunk.meta_data.statistics.__isset.null_count ||
+ chunk.meta_data.statistics.null_count != row_group.num_rows) {
+ return false;
+ }
+ }
+ }
+ return true;
+}
+
+size_t finalize_variant_leaf_projections(
+ const NativeParquetMetadata& metadata,
+ const std::vector<std::unique_ptr<ParquetColumnSchema>>& file_schema,
+ std::vector<format::LocalColumnIndex>* projections) {
+ DORIS_CHECK(projections != nullptr);
+ size_t retained = 0;
+ for (auto& projection : *projections) {
+ const int32_t local_id = projection.local_id();
+ if (local_id < 0 || local_id >=
static_cast<int32_t>(file_schema.size()) ||
+ file_schema[local_id]->kind != ParquetColumnSchemaKind::VARIANT ||
Review Comment:
[P1] Finalize nested Variant projections recursively
This condition limits residual completeness validation to top-level Variant
columns. Once the nested mapping is wired, a STRUCT/LIST/MAP-rooted request
skips `variant_projection_is_fully_shredded()` and unsafe-node expansion, yet
native planning still builds an incomplete nested Variant state. If that
wrapper has a populated residual `value`, direct leaf lookup can then return
only the typed leaf and silently drop the residual contribution. Walk the
schema/projection recursively, expand each unsafe nested Variant before reader
creation, and count retained nested projections. For ARRAY/MAP nesting, prove
fallback absence in the repeated Variant-instance domain rather than comparing
leaf `null_count` with top-level Row Group rows.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java:
##########
@@ -725,6 +725,7 @@ private Plan
bindIcebergTableSink(MatchingContext<UnboundIcebergTableSink<Plan>>
Pair<IcebergExternalDatabase, IcebergExternalTable> pair =
bind(ctx.cascadesContext, sink);
IcebergExternalDatabase database = pair.first;
IcebergExternalTable table = pair.second;
+ IcebergUtils.validateWriteSchema(table.getFullSchema());
Review Comment:
[P1] Apply the read-only gate to direct Iceberg merge sinks
This is the only production call to `validateWriteSchema()`, so it protects
`UnboundIcebergTableSink` but not UPDATE/MERGE: both commands construct
`LogicalIcebergMergeSink` directly and its expression binder never invokes this
check. UPDATE also projects every base column, so changing an unrelated field
rewrites the unchanged Variant through the writer that this PR declares
unsupported, risking lossy data. Enforce the gate at a common data-writing
Iceberg sink boundary (or explicitly in both commands) and cover UPDATE plus
matched-update/not-matched-insert MERGE.
##########
be/src/format_v2/parquet/native_schema_desc.cpp:
##########
@@ -517,6 +726,21 @@ Status NativeFieldDescriptor::parse_group_field(
const std::vector<tparquet::SchemaElement>& t_schemas, size_t curr_pos,
NativeFieldSchema* group_field) {
auto& group_schema = t_schemas[curr_pos];
+ group_field->parquet_schema = group_schema;
+ if (is_variant_node(group_schema)) {
Review Comment:
[P1] Preserve repetition when parsing Variant groups
This branch runs before generic repeated-group handling, although the
[Variant encoding
contract](https://parquet.apache.org/docs/file-format/types/variantencoding/)
places no additional restriction on group repetition. For `repeated group v
(VARIANT(1))`, `parse_struct_field()` neither increments the
repetition/definition levels nor constructs the ARRAY wrapper, and
`build_native_node_schema()` exposes a scalar Variant with maximum repetition
level zero while the leaf pages use repeated levels. A valid file can therefore
fail decoding or return the wrong row shape. Preserve the repeated wrapper and
its levels here (or reject it explicitly if unsupported), and add
required/optional/repeated Variant schema and read fixtures.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java:
##########
@@ -263,6 +263,15 @@ protected void doInitialize() throws UserException {
}
}
super.doInitialize();
+ // This gate must run during shared initialization: batch split
assignment bypasses
+ // doGetSplits(), but it must never assign a semantic Variant
projection to an old BE.
+
checkVariantBackendCompatibilityForCurrentScan(backendPolicy.getBackends());
+ }
+
+ void checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend>
backends)
+ throws UserException {
+ boolean projectsVariant = !isTableLevelCountStarPushdown() &&
projectsVariant(desc);
Review Comment:
[P1] Gate COUNT(*) only after a metadata count is proven
This treats every syntactic COUNT(*) as non-Variant during `doInitialize()`,
but only later does `isBatchMode()` discover whether `getCountFromSnapshot()`
returned a nonnegative metadata count. Missing counters or equality/non-ignored
position deletes fall back to normal data splits; the V1 Iceberg reader then
disables its underlying COUNT reader for delete files and performs a real
carrier-column read. Together with the matching legacy-scanner exemption, a
Variant-only table can therefore reach an old/legacy reader that cannot decode
it. Grant this exemption only after a nonnegative table-level count is
attached; otherwise require current backends and FileScannerV2.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
