github-actions[bot] commented on code in PR #66413: URL: https://github.com/apache/doris/pull/66413#discussion_r3791010143
########## be/src/format_v2/parquet/reader/variant_column_reader.cpp: ########## @@ -0,0 +1,1332 @@ +// 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 "format_v2/parquet/reader/variant_column_reader.h" + +#include <algorithm> +#include <array> +#include <cmath> +#include <cstdint> +#include <cstring> +#include <limits> +#include <mutex> +#include <optional> +#include <string_view> +#include <vector> + +#include "common/exception.h" +#include "core/assert_cast.h" +#include "core/column/column_array.h" +#include "core/column/column_decimal.h" +#include "core/column/column_map.h" +#include "core/column/column_nullable.h" +#include "core/column/column_struct.h" +#include "core/column/column_vector.h" +#include "core/column/variant_v2/column_variant_v2.h" +#include "core/column/variant_v2/column_variant_v2_typed_column.h" +#include "core/data_type/data_type_nullable.h" +#include "core/data_type/data_type_variant_v2.h" +#include "core/value/variant/variant_batch_builder.h" +#include "core/value/variant/variant_metadata.h" +#include "format_v2/parquet/parquet_column_schema.h" + +namespace doris::format::parquet { +namespace { + +struct Cell { + const IColumn* column = nullptr; + bool is_null = false; +}; + +Cell cell_at(const IColumn& column, size_t row) { + if (row >= column.size()) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant row {} exceeds column size {}", row, + column.size()); + } + if (const auto* nullable = check_and_get_column<ColumnNullable>(column)) { + return {.column = &nullable->get_nested_column(), + .is_null = nullable->get_null_map_data()[row] != 0}; + } + return {.column = &column, .is_null = false}; +} + +const ParquetColumnSchema* find_child(const ParquetColumnSchema& schema, std::string_view name, + size_t* index) { + for (size_t i = 0; i < schema.children.size(); ++i) { + if (schema.children[i]->name == name) { + if (index != nullptr) { + *index = i; + } + return schema.children[i].get(); + } + } + return nullptr; +} + +Cell struct_child_at(const ParquetColumnSchema& schema, const IColumn& physical, size_t row, + std::string_view name, const ParquetColumnSchema** child_schema) { + const auto& structure = assert_cast<const ColumnStruct&>(physical); + size_t index = 0; + const auto* child = find_child(schema, name, &index); + if (child == nullptr || index >= structure.tuple_size()) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has no physical child {}", + schema.name, name); + } + if (child_schema != nullptr) { + *child_schema = child; + } + return cell_at(structure.get_column(index), row); +} + +uint8_t decimal_width(int precision) { + if (precision <= 0 || precision > 38) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant decimal precision {} is outside [1, 38]", precision); + } + return precision <= 9 ? 4 : (precision <= 18 ? 8 : 16); +} + +uint8_t integer_width(const ParquetColumnSchema& schema, PrimitiveType type) { + if (schema.type_descriptor.is_unsigned_integer) { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Unsigned integers are not valid Parquet Variant typed values"); + } + if (schema.type_descriptor.integer_bit_width > 0) { + switch (schema.type_descriptor.integer_bit_width) { + case 8: + return 1; + case 16: + return 2; + case 32: + return 4; + case 64: + return 8; + default: + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant integer width {}", + schema.type_descriptor.integer_bit_width); + } + } + switch (type) { + case TYPE_TINYINT: + return 1; + case TYPE_SMALLINT: + return 2; + case TYPE_INT: + return 4; + case TYPE_BIGINT: + return 8; + default: + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant integer type {}", type); + } +} + +void append_typed_scalar(const ParquetColumnSchema& schema, const IColumn& column, size_t row, + VariantBatchBuilder::Row& builder) { + const PrimitiveType type = remove_nullable(schema.type)->get_primitive_type(); + switch (type) { + case TYPE_BOOLEAN: + builder.add_bool(assert_cast<const ColumnUInt8&>(column).get_data()[row] != 0); + return; + case TYPE_TINYINT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt8&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_SMALLINT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt16&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_INT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt32&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_BIGINT: + builder.add_scalar( + VariantScalarRef::integer(assert_cast<const ColumnInt64&>(column).get_data()[row], + integer_width(schema, type))); + return; + case TYPE_FLOAT: + builder.add_float(assert_cast<const ColumnFloat32&>(column).get_data()[row]); + return; + case TYPE_DOUBLE: + builder.add_double(assert_cast<const ColumnFloat64&>(column).get_data()[row]); + return; + case TYPE_DECIMAL128I: { + const auto value = assert_cast<const ColumnDecimal128V3&>(column).get_data()[row].value; + builder.add_decimal(value, static_cast<uint8_t>(schema.type_descriptor.decimal_scale), + decimal_width(schema.type_descriptor.decimal_precision)); + return; + } + case TYPE_TIMEV2: { + const double seconds = assert_cast<const ColumnTimeV2&>(column).get_data()[row]; + if (!std::isfinite(seconds) || + std::abs(seconds) > static_cast<double>(std::numeric_limits<int64_t>::max()) / 1e6) { + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant TIME value"); + } + builder.add_time_ntz_micros(static_cast<int64_t>(std::llround(seconds * 1e6))); + return; + } + case TYPE_DATETIMEV2: { + if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) { + // Native DATETIMEV2 is microsecond based. Reject before returning a silently truncated + // value; a raw INT64 nanos decoder can be added independently. + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Parquet Variant TIMESTAMP(NANOS) is not supported"); + } + const auto& value = assert_cast<const ColumnDateTimeV2&>(column).get_data()[row]; + builder.add_timestamp_micros( + variant_timestamp_micros(value, row, "Parquet Variant TIMESTAMP"), + schema.type_descriptor.timestamp_is_adjusted_to_utc); + return; + } + case TYPE_TIMESTAMPTZ: { + if (schema.type_descriptor.time_unit == ParquetTimeUnit::NANOS) { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Parquet Variant TIMESTAMP(NANOS) is not supported"); + } + const auto& value = assert_cast<const ColumnTimeStampTz&>(column).get_data()[row]; + builder.add_timestamp_micros( + variant_timestamp_micros(value, row, "Parquet Variant TIMESTAMP"), true); + return; + } + case TYPE_VARBINARY: { + const StringRef value = column.get_data_at(row); + if (!schema.type_descriptor.is_uuid) { + builder.add_binary(value); + return; + } + if (value.size != 16) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant UUID has {} bytes instead of 16", value.size); + } + std::array<uint8_t, 16> uuid {}; + std::memcpy(uuid.data(), value.data, uuid.size()); + builder.add_uuid(uuid); + return; + } + case TYPE_STRING: { + const StringRef value = column.get_data_at(row); + if (schema.type_descriptor.is_uuid) { + if (value.size != 16) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant UUID has {} bytes instead of 16", value.size); + } + std::array<uint8_t, 16> uuid {}; + std::memcpy(uuid.data(), value.data, uuid.size()); + builder.add_uuid(uuid); + } else if (schema.type_descriptor.is_string_annotation) { + builder.add_string(value); + } else { + builder.add_binary(value); + } + return; + } + default: + if (!is_supported_variant_typed_identity(type)) { + throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, + "Parquet Variant typed value {} is not supported", + remove_nullable(schema.type)->get_name()); + } + dispatch_variant_typed_column( + column, type, [&]<PrimitiveType Type>(const auto& typed_column) { + with_variant_typed_scalar<Type>( + typed_column, row, + static_cast<uint8_t>(remove_nullable(schema.type)->get_scale()), + [&](const VariantScalarRef& scalar) { builder.add_scalar(scalar); }); + }); + } +} + +enum class WrapperContext { ROOT, ARRAY_ELEMENT, OBJECT_FIELD }; + +bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper, size_t row, + VariantMetadataRef metadata, VariantBatchBuilder::Row& builder, + WrapperContext context); + +void append_typed_value(const ParquetColumnSchema& schema, const IColumn& column, size_t row, + VariantMetadataRef metadata, const VariantRef* residual, + VariantBatchBuilder::Row& builder) { + switch (schema.kind) { + case ParquetColumnSchemaKind::PRIMITIVE: + if (static_cast<bool>(residual)) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant scalar typed_value cannot have residual value bytes"); + } + append_typed_scalar(schema, column, row, builder); + return; + case ParquetColumnSchemaKind::STRUCT: { + if (static_cast<bool>(residual) && residual->basic_type() != VariantBasicType::OBJECT) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant object typed_value has non-object residual value"); + } + const auto& structure = assert_cast<const ColumnStruct&>(column); + if (structure.tuple_size() != schema.children.size()) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant object {} physical field count mismatch", schema.name); + } + auto object = builder.start_object(); + if (static_cast<bool>(residual)) { + for (uint32_t i = 0; i < residual->num_elements(); ++i) { + uint32_t field_id = 0; + const VariantRef child = residual->object_value_at(i, &field_id); + object.add_key(residual->metadata.key_at(field_id)); + builder.add_value(child); + } + } + for (size_t i = 0; i < schema.children.size(); ++i) { + const auto& child_schema = *schema.children[i]; + const Cell child = cell_at(structure.get_column(i), row); + if (child.is_null) { + // Shredded object fields are optional wrapper groups. A missing group means the + // key is absent, which differs from a present wrapper encoding a Variant null. + continue; + } + // A null/null wrapper means this object field is absent. Delay add_key until its + // presence is known so absent shredded fields do not turn into Variant nulls. + size_t value_index = 0; + const auto* value_schema = find_child(child_schema, "value", &value_index); + const auto& child_struct = assert_cast<const ColumnStruct&>(*child.column); + const bool value_present = value_schema != nullptr && + !cell_at(child_struct.get_column(value_index), row).is_null; + size_t typed_index = 0; + const auto* typed_schema = find_child(child_schema, "typed_value", &typed_index); + const bool typed_present = typed_schema != nullptr && + !cell_at(child_struct.get_column(typed_index), row).is_null; + if (!value_present && !typed_present) { + continue; + } + object.add_key(StringRef(child_schema.name)); + (void)append_wrapper(child_schema, *child.column, row, metadata, builder, + WrapperContext::OBJECT_FIELD); + } + object.finish(); + return; + } + case ParquetColumnSchemaKind::LIST: { + if (static_cast<bool>(residual)) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant array typed_value cannot have residual value bytes"); + } + if (schema.children.size() != 1) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant array {} has invalid element schema", schema.name); + } + const auto& array = assert_cast<const ColumnArray&>(column); + const size_t begin = array.offset_at(static_cast<ssize_t>(row)); + const size_t end = array.get_offsets()[row]; + auto scope = builder.start_array(); + for (size_t element = begin; element < end; ++element) { + const Cell cell = cell_at(array.get_data(), element); + if (cell.is_null) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant shredded array element wrapper is null"); + } + (void)append_wrapper(*schema.children[0], *cell.column, element, metadata, builder, + WrapperContext::ARRAY_ELEMENT); + } + scope.finish(); + return; + } + case ParquetColumnSchemaKind::MAP: + case ParquetColumnSchemaKind::VARIANT: + throw Exception(ErrorCode::CORRUPTION, "Invalid Parquet Variant typed_value schema {}", + schema.name); + } +} + +bool append_wrapper(const ParquetColumnSchema& schema, const IColumn& wrapper, size_t row, + VariantMetadataRef metadata, VariantBatchBuilder::Row& builder, + WrapperContext context) { + Cell value; + if (find_child(schema, "value", nullptr) != nullptr) { + value = struct_child_at(schema, wrapper, row, "value", nullptr); + } else { + value.is_null = true; + } + const ParquetColumnSchema* typed_schema = nullptr; + Cell typed; + if (find_child(schema, "typed_value", nullptr) != nullptr) { + typed = struct_child_at(schema, wrapper, row, "typed_value", &typed_schema); + } else { + typed.is_null = true; + } + + if (find_child(schema, "value", nullptr) == nullptr && typed_schema == nullptr) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant wrapper {} has neither value nor typed_value", + schema.name); + } + if (value.is_null && typed.is_null) { + if (context == WrapperContext::OBJECT_FIELD) { + return false; + } + if (context == WrapperContext::ARRAY_ELEMENT) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant array element is missing"); + } + builder.add_null(); + return true; + } + + VariantRef residual {.metadata = metadata, .value = {}}; + if (!value.is_null) { + residual.value = value.column->get_data_at(row); + } + if (typed.is_null) { + builder.add_value(residual); + return true; + } + append_typed_value(*typed_schema, *typed.column, row, metadata, + value.is_null ? nullptr : &residual, builder); + return true; +} + +void encode_variant_range(const ParquetColumnSchema& schema, const IColumn& wrapper, + const ColumnNullable* outer_nullable, size_t begin, size_t end, + bool require_metadata, ColumnVariantV2& variants) { + try { + VariantBatchBuilder builder(VariantBatchBuilder::ReserveHint {.rows = end - begin}); + for (size_t row = begin; row < end; ++row) { + auto output_row = builder.begin_row(); + if (outer_nullable != nullptr && outer_nullable->get_null_map_data()[row] != 0) { + output_row.add_null(); + output_row.finish(); + continue; + } + VariantMetadataRef metadata; + if (find_child(schema, "metadata", nullptr) != nullptr) { + const Cell metadata_cell = + struct_child_at(schema, wrapper, row, "metadata", nullptr); + if (metadata_cell.is_null) { + throw Exception(ErrorCode::CORRUPTION, + "Parquet Variant {} has null metadata at row {}", schema.name, + row); + } + const StringRef metadata_bytes = metadata_cell.column->get_data_at(row); + metadata = {metadata_bytes.data, metadata_bytes.size}; + metadata.validate(); + } else if (require_metadata) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} has no root metadata", + schema.name); + } + (void)append_wrapper(schema, wrapper, row, metadata, output_row, WrapperContext::ROOT); + output_row.finish(); + } + VariantBatchBuilder batch = builder.finish_batch(); + variants.insert_encoded_batch(batch); + } catch (...) { + if (end - begin <= 1) { + throw; + } + // A single builder has one metadata dictionary. If heterogeneous file rows cannot fit in + // that dictionary, split without changing the destination column's already-valid batches. + // Corrupt input still reaches a one-row range and propagates its original exception. + const size_t middle = begin + (end - begin) / 2; + encode_variant_range(schema, wrapper, outer_nullable, begin, middle, require_metadata, + variants); + encode_variant_range(schema, wrapper, outer_nullable, middle, end, require_metadata, + variants); + } +} + +ColumnVariantV2::MutablePtr encode_variant_column(const ParquetColumnSchema& schema, + const IColumn& physical, + bool require_metadata = true) { + if (schema.kind != ParquetColumnSchemaKind::VARIANT) { + throw Exception(ErrorCode::INVALID_ARGUMENT, "Parquet column {} is not Variant", + schema.name); + } + const auto* outer_nullable = check_and_get_column<ColumnNullable>(physical); + const IColumn& wrapper = + outer_nullable == nullptr ? physical : outer_nullable->get_nested_column(); + const auto& structure = assert_cast<const ColumnStruct&>(wrapper); + if (structure.tuple_size() != schema.children.size()) { + throw Exception(ErrorCode::CORRUPTION, "Parquet Variant {} physical field count mismatch", + schema.name); + } + + auto variants = ColumnVariantV2::create(); + constexpr size_t MAX_RECONSTRUCTION_BATCH_ROWS = 4096; + for (size_t begin = 0; begin < physical.size(); begin += MAX_RECONSTRUCTION_BATCH_ROWS) { + encode_variant_range(schema, wrapper, outer_nullable, begin, + std::min(physical.size(), begin + MAX_RECONSTRUCTION_BATCH_ROWS), + require_metadata, *variants); + } + return variants; +} + +std::unique_ptr<ParquetColumnSchema> clone_schema( + const ParquetColumnSchema& source, const format::LocalColumnIndex* projection = nullptr) { + auto result = std::make_unique<ParquetColumnSchema>(); + result->local_id = source.local_id; + result->parquet_field_id = source.parquet_field_id; + result->name = source.name; + result->type = source.type; + result->variant_physical_type = source.variant_physical_type; + result->leaf_column_id = source.leaf_column_id; + result->type_descriptor = source.type_descriptor; + result->kind = source.kind; + result->contains_variant = source.contains_variant; + result->max_definition_level = source.max_definition_level; + result->max_repetition_level = source.max_repetition_level; + result->nullable_definition_level = source.nullable_definition_level; + result->definition_level = source.definition_level; + result->repetition_level = source.repetition_level; + result->repeated_ancestor_definition_level = source.repeated_ancestor_definition_level; + result->repeated_repetition_level = source.repeated_repetition_level; + const bool partial = format::is_partial_projection(projection); + result->children.reserve(partial ? projection->children.size() : source.children.size()); + if (partial) { + // NativeColumnReader emits a partial STRUCT in projection order, so the retained schema + // must use that same order or field names will address the wrong physical tuple element. + for (const auto& child_projection : projection->children) { + const auto child = std::ranges::find_if(source.children, [&](const auto& candidate) { + return candidate->local_id == child_projection.local_id(); + }); + DORIS_CHECK(child != source.children.end()); + result->children.push_back(clone_schema(**child, &child_projection)); + } + } else { + for (const auto& child : source.children) { + result->children.push_back(clone_schema(*child)); + } + } + return result; +} + +ColumnPtr unwrap_nullable(ColumnPtr column) { + if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) { + return nullable->get_nested_column_ptr(); + } + return column; +} + +ColumnPtr struct_child(const ParquetColumnSchema& schema, ColumnPtr column, std::string_view name, + const ParquetColumnSchema** child_schema) { + column = unwrap_nullable(std::move(column)); + const auto* structure = check_and_get_column<ColumnStruct>(*column); + if (structure == nullptr) { + return nullptr; + } + size_t index = 0; + const auto* child = find_child(schema, name, &index); + if (child == nullptr || index >= structure->tuple_size()) { + return nullptr; + } + if (child_schema != nullptr) { + *child_schema = child; + } + return structure->get_column_ptr(index); +} + +bool has_present_value(const ColumnPtr& column) { + if (const auto* nullable = check_and_get_column<ColumnNullable>(*column)) { + return std::ranges::any_of(nullable->get_null_map_data(), + [](uint8_t is_null) { return is_null == 0; }); + } + return !column->empty(); +} + +bool supports_direct_typed_variant_state(const ParquetColumnSchema& schema) { + if (schema.type == nullptr || schema.kind != ParquetColumnSchemaKind::PRIMITIVE) { + return false; + } + // ColumnVariantV2 typed state carries only a Doris type. Binary/UUID annotations, temporal + // units, and other Parquet-only identity must therefore reconstruct canonical Variant bytes. + switch (remove_nullable(schema.type)->get_primitive_type()) { + case TYPE_BOOLEAN: + case TYPE_TINYINT: + case TYPE_SMALLINT: + case TYPE_INT: + case TYPE_BIGINT: + case TYPE_FLOAT: + case TYPE_DOUBLE: + case TYPE_DECIMAL128I: Review Comment: [P1] Preserve the Parquet decimal width on this direct path. Native schema conversion maps every DECIMAL annotation to `TYPE_DECIMAL128I`, including precision 9 and 18, while full reconstruction correctly derives Variant `DECIMAL4`/`DECIMAL8` from `type_descriptor.decimal_precision`. Admitting that Doris type here loses the descriptor; the typed-state encoder later hardcodes width 16, so selecting a shredded decimal leaf changes its Variant identity to `DECIMAL16` depending on projection. Normalize precision <= 18 here (or retain the Parquet width in typed state), and cover production-shaped `DataTypeDecimal128(9, ...)` and `(18, ...)` extraction/serialization. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java: ########## @@ -194,6 +205,86 @@ public PluginDrivenScanNode(PlanNodeId id, TupleDescriptor desc, this.currentHandle = tableHandle; } + @Override + protected void doInitialize() throws UserException { + super.doInitialize(); + // Pin before projection pruning so every later connector decision uses this scan's snapshot. + // The Variant compatibility fence itself runs in finalize, after Nereids prunes scan slots. + pinMvccSnapshot(); + } + + void checkVariantBackendCompatibilityForCurrentScan(Iterable<Backend> backends) + throws UserException { + boolean projectsVariant = projectsComputeVariant(desc); + boolean metadataCountProven = false; + ConnectorScanPlanProvider scanProvider = resolveScanProvider(); + if (isTableLevelCountStarPushdown() && conjuncts.isEmpty() && scanProvider != null) { Review Comment: [P1] Do not use the metadata-count escape hatch when TABLESAMPLE will disable count pushdown. `getSplits()` later computes `applySample` from `tableSample` plus `supportsTableSample()` and forces `countPushdown=false`, but this earlier check can still set `metadataCountProven=true` and skip the Variant old-BE fence. A connector exposing both public capabilities (`VARIANT_COMPUTE_V2` and byte-proportional TABLESAMPLE) then plans sampled data ranges that actually decode Variant on an unsupported backend. Apply the same sampling gate here (or defer compatibility until the actual ranges are known), and cover sampled `COUNT(*)` on a one-Variant-column table. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/scan/PluginDrivenScanNode.java: ########## @@ -1663,6 +1777,10 @@ private boolean computeBatchMode() { if (tableSample != null && onPluginClassLoader(scanProvider, scanProvider::supportsTableSample)) { return false; } + if (variantCompatibilityDeferred) { Review Comment: [P1] Keep metadata-proven Variant counts out of this partition-batch path. The earlier `canServeMetadataOnlyCount()` result leaves `variantCompatibilityDeferred=false`, so a provider that also implements `supportsBatchScan()` reaches this branch on a large partition set. `startSplit()` then builds the batch request without `countPushdown` and never applies `plannedScanDecodesVariant()` to the returned ranges; execution-v11 BEs can therefore receive ordinary V2 data ranges even though the metadata proof skipped their compatibility fence. Preserve count pushdown per batch or force these scans through the synchronous range-level proof, and cover the combined public SPI capabilities. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java: ########## @@ -366,6 +373,53 @@ private static boolean hasMeaningfulTypeParameters(String typeName) { || "TIMESTAMPTZ".equals(typeName); } + static void validateWriteSchema(List<ConnectorColumn> columns, boolean writesDataFiles) { + if (!writesDataFiles) { + return; + } + if (columns.stream().anyMatch(column -> containsVariant(column.getType()))) { + // Reject the whole data-file write: validating only selected columns would let an + // unchanged Variant target flow through a writer that cannot preserve its physical identity. + throw new DorisConnectorException( + "Iceberg VARIANT columns are read-only and cannot be written"); + } + } + + static void validateWriteSchema(ConnectorWriteHandle handle) { + // The explicit INSERT column list can omit an unchanged Variant column, but the data writer + // still binds the complete target schema and therefore must validate that complete shape. + validateWriteSchema(handle.getBoundTargetColumns(), handle.isWritesDataFiles()); + } + + private static void validateNestedPartitionWriteCompatibility( + ConnectorWriteHandle handle, IcebergWriteSchemaContext schemaContext) { + WriteOperation operation = handle.getWriteOperation(); + boolean writesDataFiles = operation != WriteOperation.DELETE + && ((operation != WriteOperation.UPDATE && operation != WriteOperation.MERGE) + || handle.isWritesDataFiles()); + if (!writesDataFiles || schemaContext == null Review Comment: [P1] Keep this fence for delete-only MERGE on nested partition specs. Execution-v11 BEs ignore the new `writes_data_files=false` field and follow the legacy merge-sink path, which unconditionally constructs and initializes the table writer. That writer binds the serialized partition spec before seeing any rows, so a nested source still fails as `outside writer schema`; having zero insert rows only avoids creating partition writers later. Exempt the separate DELETE sink, but reject old-BE MERGE whenever the retained spec has a nested source, and change the new v11 `doesNotThrow` case accordingly. -- 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]
