github-actions[bot] commented on code in PR #65851: URL: https://github.com/apache/doris/pull/65851#discussion_r3662648914
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java: ########## @@ -0,0 +1,468 @@ +// 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. + +package org.apache.doris.datasource.iceberg; + +import org.apache.doris.catalog.Column; +import org.apache.doris.datasource.mvcc.MvccSnapshot; +import org.apache.doris.datasource.mvcc.MvccUtil; +import org.apache.doris.nereids.exceptions.AnalysisException; +import org.apache.doris.nereids.trees.expressions.Expression; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Array; +import org.apache.doris.nereids.trees.expressions.functions.scalar.CreateMap; +import org.apache.doris.nereids.trees.expressions.functions.scalar.CreateNamedStruct; +import org.apache.doris.nereids.trees.expressions.functions.scalar.Unhex; +import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral; +import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral; +import org.apache.doris.nereids.trees.expressions.literal.BooleanLiteral; +import org.apache.doris.nereids.trees.expressions.literal.DateTimeV2Literal; +import org.apache.doris.nereids.trees.expressions.literal.DateV2Literal; +import org.apache.doris.nereids.trees.expressions.literal.DecimalV3Literal; +import org.apache.doris.nereids.trees.expressions.literal.DoubleLiteral; +import org.apache.doris.nereids.trees.expressions.literal.FloatLiteral; +import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral; +import org.apache.doris.nereids.trees.expressions.literal.Literal; +import org.apache.doris.nereids.trees.expressions.literal.MapLiteral; +import org.apache.doris.nereids.trees.expressions.literal.NullLiteral; +import org.apache.doris.nereids.trees.expressions.literal.StringLiteral; +import org.apache.doris.nereids.trees.expressions.literal.StructLiteral; +import org.apache.doris.nereids.trees.expressions.literal.TimestampTzLiteral; +import org.apache.doris.nereids.trees.expressions.literal.VarBinaryLiteral; +import org.apache.doris.nereids.types.DataType; +import org.apache.doris.nereids.types.DateTimeV2Type; +import org.apache.doris.nereids.types.DecimalV3Type; +import org.apache.doris.nereids.types.StructType; +import org.apache.doris.nereids.types.TimeStampTzType; +import org.apache.doris.nereids.types.VarBinaryType; +import org.apache.doris.nereids.util.TypeCoercionUtils; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import com.google.common.io.BaseEncoding; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.Table; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.SnapshotUtil; + +import java.math.BigDecimal; +import java.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.UUID; + +/** + * Statement-scoped Iceberg write schema and write-default values. + * + * <p>The context pins one Iceberg schema before analysis. The analyzer, planner sink and + * transaction preflight must all use this same instance so a concurrent schema change cannot + * combine expressions from one schema with a writer schema from another one. + */ +public final class IcebergWriteSchemaContext { + private final long tableId; + private final String tableName; + private final Schema schema; + private final int formatVersion; + private final Optional<String> branchName; + private final String schemaJson; + private final String mergeSchemaJson; + private final List<Column> columns; + private final List<Column> mergeColumns; + private final Map<Integer, Types.NestedField> fieldsById; + private final Map<Integer, Expression> writeDefaultsById; + + /** Pin the statement snapshot's current table schema under the catalog authentication boundary. */ + public static IcebergWriteSchemaContext create( + IcebergExternalTable dorisTable, Optional<String> branchName) { + Objects.requireNonNull(dorisTable, "dorisTable should not be null"); + Objects.requireNonNull(branchName, "branchName should not be null"); + try { + return dorisTable.getCatalog().getExecutionAuthenticator().execute(() -> { + Table table = dorisTable.getIcebergTable(); + Schema schema = branchName.isPresent() + ? resolveBranchSchema(table, branchName.get(), dorisTable.getName()) + : resolveStatementSchema(table, dorisTable); + int formatVersion = IcebergUtils.getFormatVersion(table); + return new IcebergWriteSchemaContext( + dorisTable.getId(), dorisTable.getName(), schema, formatVersion, branchName, + dorisTable.getCatalog().getEnableMappingVarbinary(), + dorisTable.getCatalog().getEnableMappingTimestampTz()); + }); + } catch (Exception e) { + throw new AnalysisException("Failed to pin Iceberg write schema for table " + + dorisTable.getName() + ": " + e.getMessage(), e); + } + } + + @VisibleForTesting + public static IcebergWriteSchemaContext forSchema(Schema schema, int formatVersion, + boolean enableMappingVarbinary, boolean enableMappingTimestampTz) { + return new IcebergWriteSchemaContext(-1L, "test_table", schema, formatVersion, + Optional.empty(), enableMappingVarbinary, enableMappingTimestampTz); + } + + private IcebergWriteSchemaContext(long tableId, String tableName, Schema schema, + int formatVersion, Optional<String> branchName, + boolean enableMappingVarbinary, boolean enableMappingTimestampTz) { + this.tableId = tableId; + this.tableName = Objects.requireNonNull(tableName, "tableName should not be null"); + this.schema = Objects.requireNonNull(schema, "schema should not be null"); + this.formatVersion = formatVersion; + this.branchName = Objects.requireNonNull(branchName, "branchName should not be null"); + this.schemaJson = SchemaParser.toJson(schema); + Schema mergeSchema = formatVersion >= IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION + ? IcebergUtils.appendRowLineageFieldsForV3(schema) : schema; + this.mergeSchemaJson = SchemaParser.toJson(mergeSchema); + + List<Column> parsedColumns = IcebergUtils.parseSchema( + schema, enableMappingVarbinary, enableMappingTimestampTz); + this.columns = ImmutableList.copyOf(parsedColumns); + List<Column> writerColumns = new ArrayList<>(parsedColumns); + writerColumns.add(IcebergRowId.createHiddenColumn()); + if (formatVersion >= IcebergUtils.ICEBERG_ROW_LINEAGE_MIN_VERSION) { + Column rowIdColumn = IcebergUtils.parseField( + org.apache.iceberg.MetadataColumns.ROW_ID, + enableMappingVarbinary, enableMappingTimestampTz); + rowIdColumn.setIsVisible(false); + writerColumns.add(rowIdColumn); + Column sequenceColumn = IcebergUtils.parseField( + org.apache.iceberg.MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER, + enableMappingVarbinary, enableMappingTimestampTz); + sequenceColumn.setIsVisible(false); + writerColumns.add(sequenceColumn); + } + this.mergeColumns = ImmutableList.copyOf(writerColumns); + + ImmutableMap.Builder<Integer, Types.NestedField> byId = ImmutableMap.builder(); + ImmutableMap.Builder<Integer, Expression> defaults = ImmutableMap.builder(); + for (Types.NestedField field : schema.columns()) { + byId.put(field.fieldId(), field); + if (field.writeDefault() != null) { + DataType targetType = DataType.fromCatalogType(IcebergUtils.icebergTypeToDorisType( + field.type(), enableMappingVarbinary, enableMappingTimestampTz)); + defaults.put(field.fieldId(), toDorisExpression( + field.type(), field.writeDefault(), targetType, + enableMappingVarbinary, enableMappingTimestampTz)); + } + } + this.fieldsById = byId.build(); + this.writeDefaultsById = defaults.build(); + } + + private static Schema resolveBranchSchema(Table table, String branchName, String tableName) { + SnapshotRef ref = table.refs().get(branchName); + if (ref == null) { + throw new AnalysisException(branchName + " is not founded in " + tableName); + } + if (!ref.isBranch()) { + throw new AnalysisException(branchName + + " is a tag, not a branch. Tags cannot be targets for producing snapshots"); + } + return SnapshotUtil.schemaFor(table, ref.snapshotId()); + } + + private static Schema resolveStatementSchema(Table table, IcebergExternalTable dorisTable) { + Optional<MvccSnapshot> snapshot = MvccUtil.getSnapshotFromContext(dorisTable); + if (!snapshot.isPresent()) { + return table.schema(); + } + Preconditions.checkState(snapshot.get() instanceof IcebergMvccSnapshot, + "Expected an Iceberg MVCC snapshot for table %s", dorisTable.getName()); + long schemaId = ((IcebergMvccSnapshot) snapshot.get()) + .getSnapshotCacheValue().getSnapshot().getSchemaId(); + Schema schema = table.schemas().get(Math.toIntExact(schemaId)); + return Preconditions.checkNotNull(schema, + "Iceberg schema %s is not available in the statement table metadata for %s", + schemaId, dorisTable.getName()); + } + + /** Resolve the value used for an omitted column or an explicit DEFAULT. */ + public Expression resolveWriteDefault(Column column) { + Types.NestedField field = fieldsById.get(column.getUniqueId()); + if (field == null) { + throw new AnalysisException("Column " + column.getName() + + " is not present in pinned Iceberg schema " + getSchemaId()); + } + Expression writeDefault = writeDefaultsById.get(field.fieldId()); + if (writeDefault != null) { + return writeDefault; + } + DataType targetType = DataType.fromCatalogType(column.getType()); + if (field.isOptional()) { + return new NullLiteral(targetType); + } + throw new AnalysisException("Column has no write default and is required, column=" + field.name()); + } + + /** Validate that the table still exposes the schema and format pinned during analysis. */ + public void validateCurrentSchema(Table table) { + Schema currentSchema = branchName.isPresent() + ? resolveBranchSchema(table, branchName.get(), tableName) + : table.schema(); + int currentFormatVersion = IcebergUtils.getFormatVersion(table); Review Comment: [P1] Fence all writer metadata before accepting the fresh table This preflight only compares schema ID and Iceberg format version, but the BE has already been configured from the cached table's partition spec, sort order, `write.format.default`, and metrics policy. The transaction then uses the independently loaded fresh table, and `IcebergWriterHelper.convertToWriterResult(transaction.table(), ...)` reconstructs each `DataFile` from that fresh table's `spec()`, file format, schema, metrics config, and sort order; ordinary data-file commit records carry partition values but not the spec ID or physical file format that BE actually used. A concurrent partition-spec replacement or `write.format.default` change therefore passes this check and can commit old-spec partition values under the new spec, or describe a Parquet file as ORC. Please either pin and validate every piece of writer metadata consumed at commit, or carry the actual spec ID/file format and related metadata in the commit result and reconstruct the `DataFile` from those pinned value s. ########## fe/fe-core/src/main/java/org/apache/doris/planner/IcebergTableSink.java: ########## @@ -121,21 +130,32 @@ public void bindDataSink(Optional<InsertCommandContext> insertCtx) tSink.setTbName(targetTable.getName()); boolean isRewriting = false; + Optional<IcebergWriteSchemaContext> executorWriteSchemaContext = Optional.empty(); if (insertCtx.isPresent() && insertCtx.get() instanceof IcebergInsertCommandContext) { IcebergInsertCommandContext context = (IcebergInsertCommandContext) insertCtx.get(); isRewriting = context.isRewriting(); + executorWriteSchemaContext = context.getWriteSchemaContext(); if (isRewriting) { tSink.setWriteType(TIcebergWriteType.REWRITE); } } + if (!executorWriteSchemaContext.equals(writeSchemaContext)) { + throw new AnalysisException("Iceberg write schema context differs between plan and executor"); + } - Schema schema = icebergTable.schema(); + Schema schema = writeSchemaContext Review Comment: [P1] Carry a complete branch-compatible writer metadata bundle This pins the branch-head schema, but planning and commit still consume table-current metadata. For example, if main adds partition field `p` after branch `b`, an insert into `b` sends a schema without `p` plus the current spec containing `p`. BE's `id_to_column_idx[partition_field.source_id()]` inserts the missing key as index 0, so it silently commits column 0 as partition `p`. Even without partitioning or sorting, keep STRING field `s` on `b` and drop it on main: BE emits bounds keyed by `s`, while `IcebergWriterHelper` filters them with `transaction.table().schema()`. The default truncate mode then gets `schema.findType(s) == null` and `truncateBound()` dereferences that null type, failing the valid branch commit. Please carry the pinned branch schema plus compatible spec, sort order, file and metrics policy through commit; reject incompatible sources before dispatch, and make the BE lookup fail instead of defaulting to column zero. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/source/IcebergScanNode.java: ########## @@ -505,17 +521,55 @@ private List<String> getOrderedPathPartitionKeys() { } public void createScanRangeLocations() throws UserException { - super.createScanRangeLocations(); + Schema scanSchema = getQuerySchema(); + Set<Integer> equalityDeleteFieldIds = Collections.emptySet(); + if (!isSystemTable) { + equalityDeleteFieldIds = getEqualityDeleteFieldIdsForScan(); + boolean requiresCurrentSemantics = requiresRecursiveInitialDefaultMaterialization( + scanSchema, desc.getSlots()) || !equalityDeleteFieldIds.isEmpty(); + if (!requiresCurrentSemantics + && hasSmoothUpgradeSourceBackend(backendPolicy.getBackends())) { + requiresCurrentSemantics = requiresMissingRequiredFieldRejection( Review Comment: [P2] Scope the required-field fence to the selected target This compares a time-travel or ref schema with the table's complete current schema list, including schemas committed after or outside the selected snapshot lineage. If snapshot A has required `id` and a later main-line schema drops `id`, a query of A marks `id` potentially missing and rejects every smooth-upgrade source BE, although every file visible at A was written while `id` was required and needs no new missing-field behavior. Please derive this history from the selected snapshot's parent lineage, with only the target-relative fallback needed for schema-only commits, and add a later-drop or off-lineage regression. ########## be/src/format_v2/table/iceberg_reader.cpp: ########## @@ -80,61 +90,447 @@ static bool is_projected_iceberg_rowid(const format::ColumnDefinition& column) { return column.name == BeConsts::ICEBERG_ROWID_COL; } +static int iceberg_hex_value(char value) { + if (value >= '0' && value <= '9') { + return value - '0'; + } + if (value >= 'a' && value <= 'f') { + return value - 'a' + 10; + } + if (value >= 'A' && value <= 'F') { + return value - 'A' + 10; + } + return -1; +} + +static Status decode_iceberg_hex(std::string_view encoded, std::string* decoded) { + DORIS_CHECK(decoded != nullptr); + if ((encoded.size() & 1U) != 0) { + return Status::InvalidArgument("Invalid odd-length Iceberg binary default"); + } + decoded->resize(encoded.size() / 2); + for (size_t index = 0; index < encoded.size(); index += 2) { + const int high = iceberg_hex_value(encoded[index]); + const int low = iceberg_hex_value(encoded[index + 1]); + if (high < 0 || low < 0) { + return Status::InvalidArgument("Invalid hexadecimal Iceberg binary default"); + } + (*decoded)[index / 2] = static_cast<char>((high << 4) | low); + } + return Status::OK(); +} + +static Status decode_iceberg_json_binary(std::string_view encoded, std::string* decoded) { + DORIS_CHECK(decoded != nullptr); + const bool is_uuid = encoded.size() == 36 && encoded[8] == '-' && encoded[13] == '-' && + encoded[18] == '-' && encoded[23] == '-'; + if (!is_uuid) { + return decode_iceberg_hex(encoded, decoded); + } + + std::string uuid_hex; + uuid_hex.reserve(32); + for (size_t index = 0; index < encoded.size(); ++index) { + if (index != 8 && index != 13 && index != 18 && index != 23) { + uuid_hex.push_back(encoded[index]); + } + } + return decode_iceberg_hex(uuid_hex, decoded); +} + +static std::string iceberg_json_scalar_text(const rapidjson::Value& value) { + if (value.IsString()) { + return {value.GetString(), value.GetStringLength()}; + } + rapidjson::StringBuffer buffer; + rapidjson::Writer<rapidjson::StringBuffer> writer(buffer); + value.Accept(writer); + return {buffer.GetString(), buffer.GetSize()}; +} + +static void normalize_iceberg_json_timestamp(PrimitiveType primitive_type, std::string* value) { + if (primitive_type != TYPE_DATETIME && primitive_type != TYPE_DATETIMEV2 && + primitive_type != TYPE_TIMESTAMPTZ) { + return; + } + if (const size_t separator = value->find('T'); separator != std::string::npos) { + (*value)[separator] = ' '; + } + if (primitive_type == TYPE_TIMESTAMPTZ) { + return; + } + if (value->ends_with('Z')) { + value->pop_back(); + return; + } + const size_t time_start = value->find(' '); + if (time_start == std::string::npos) { + return; + } + const size_t offset = value->find_first_of("+-", time_start + 1); + if (offset != std::string::npos) { + value->erase(offset); + } +} + +static Status build_v2_null_default(const format::ColumnDefinition& field, + const DataTypePtr& data_type, Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(result != nullptr); + if (field.is_optional.has_value() && !*field.is_optional) { + return Status::InvalidArgument("Required Iceberg field '{}' has a null default", + field.name); + } + if (!data_type->is_nullable()) { + return Status::InternalError( + "Optional Iceberg field '{}' has a null default, but its Doris type '{}' is not " + "nullable", + field.name, data_type->get_name()); + } + *result = Field(); + return Status::OK(); +} + +static const format::ColumnDefinition* find_v2_struct_child(const format::ColumnDefinition& field, + const std::string& name) { + const auto exact_child = std::ranges::find_if( + field.children, [&](const auto& candidate) { return iequal(candidate.name, name); }); + if (exact_child != field.children.end()) { + return &*exact_child; + } + const auto aliased_child = std::ranges::find_if(field.children, [&](const auto& candidate) { + return std::ranges::any_of(candidate.name_mapping, + [&](const auto& alias) { return iequal(alias, name); }); + }); + return aliased_child == field.children.end() ? nullptr : &*aliased_child; +} + +static Status build_v2_initial_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + std::deque<std::string>* binary_storage, + Field* result); + +static Status build_v2_json_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result); + +static Status build_v2_json_struct_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsObject()) { + return Status::InvalidArgument("Invalid Iceberg struct default for field '{}'", field.name); + } + + const auto& struct_type = assert_cast<const DataTypeStruct&>(*value_type); + Struct struct_value; + struct_value.reserve(struct_type.get_elements().size()); + for (size_t index = 0; index < struct_type.get_elements().size(); ++index) { + const auto* child = find_v2_struct_child(field, struct_type.get_element_name(index)); + if (child == nullptr || !child->has_identifier_field_id()) { + return Status::InvalidArgument( + "Iceberg struct default for field '{}' has incomplete child metadata", + field.name); + } + + const std::string child_id = std::to_string(child->get_identifier_field_id()); + const auto member = json_value.FindMember(child_id.c_str()); + Field child_value; + if (member == json_value.MemberEnd()) { + RETURN_IF_ERROR(build_v2_initial_default_field(*child, struct_type.get_element(index), + binary_storage, &child_value)); + } else { + RETURN_IF_ERROR(build_v2_json_default_field(*child, struct_type.get_element(index), + member->value, binary_storage, + &child_value)); + } + struct_value.push_back(std::move(child_value)); + } + *result = Field::create_field<TYPE_STRUCT>(std::move(struct_value)); + return Status::OK(); +} + +// The child ColumnDefinition, recursively transported from the item TField, describes the element +// schema and its field-level default metadata. It cannot represent a particular list literal's +// length or per-position values, so the parent initial-default keeps those values in Iceberg's +// single-value JSON array. +static Status build_v2_json_array_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsArray() || field.children.size() != 1) { + return Status::InvalidArgument("Invalid Iceberg list default for field '{}'", field.name); + } + + const auto& array_type = assert_cast<const DataTypeArray&>(*value_type); + Array array_value; + array_value.reserve(json_value.Size()); + for (const auto& json_element : json_value.GetArray()) { + Field element_value; + RETURN_IF_ERROR(build_v2_json_default_field(field.children.front(), + array_type.get_nested_type(), json_element, + binary_storage, &element_value)); + array_value.push_back(std::move(element_value)); + } + *result = Field::create_field<TYPE_ARRAY>(std::move(array_value)); + return Status::OK(); +} + +// The child ColumnDefinitions, recursively transported from the key/value TFields, describe entry +// schemas and field-level default metadata. They cannot represent the number, order, or concrete +// values of map entries, so the parent initial-default keeps the entries in Iceberg's single-value +// JSON key/value arrays. +static Status build_v2_json_map_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + if (!json_value.IsObject() || !json_value.HasMember("keys") || !json_value["keys"].IsArray() || + !json_value.HasMember("values") || !json_value["values"].IsArray() || + field.children.size() != 2) { + return Status::InvalidArgument("Invalid Iceberg map default for field '{}'", field.name); + } + const auto& keys = json_value["keys"]; + const auto& values = json_value["values"]; + if (keys.Size() != values.Size()) { + return Status::InvalidArgument( + "Iceberg map default for field '{}' has {} keys but {} values", field.name, + keys.Size(), values.Size()); + } + + const auto& map_type = assert_cast<const DataTypeMap&>(*value_type); + Array key_fields; + Array value_fields; + key_fields.reserve(keys.Size()); + value_fields.reserve(values.Size()); + for (rapidjson::SizeType index = 0; index < keys.Size(); ++index) { + Field key_value; + Field mapped_value; + RETURN_IF_ERROR(build_v2_json_default_field(field.children[0], map_type.get_key_type(), + keys[index], binary_storage, &key_value)); + RETURN_IF_ERROR(build_v2_json_default_field(field.children[1], map_type.get_value_type(), + values[index], binary_storage, &mapped_value)); + key_fields.push_back(std::move(key_value)); + value_fields.push_back(std::move(mapped_value)); + } + Map map_value; + map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(key_fields))); + map_value.push_back(Field::create_field<TYPE_ARRAY>(std::move(value_fields))); + *result = Field::create_field<TYPE_MAP>(std::move(map_value)); + return Status::OK(); +} + +static Status build_v2_json_scalar_default(const format::ColumnDefinition& field, + const DataTypePtr& value_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + const auto primitive_type = value_type->get_primitive_type(); + std::string serialized_value = iceberg_json_scalar_text(json_value); + const bool binary_like = + field.initial_default_value_is_base64 || primitive_type == TYPE_VARBINARY; + if (binary_like) { + if (!json_value.IsString()) { + return Status::InvalidArgument( + "Iceberg binary default for field '{}' is not a JSON string", field.name); + } + binary_storage->emplace_back(); + RETURN_IF_ERROR(decode_iceberg_json_binary(serialized_value, &binary_storage->back())); + if (primitive_type == TYPE_VARBINARY) { + *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); + } else if (is_string_type(primitive_type)) { + *result = Field::create_field<TYPE_STRING>(binary_storage->back()); + } else { + return Status::InvalidArgument( + "Iceberg binary default for field '{}' has incompatible Doris type '{}'", + field.name, value_type->get_name()); + } + return Status::OK(); + } + + if (is_string_type(primitive_type)) { + if (!json_value.IsString()) { + return Status::InvalidArgument("Iceberg string default for field '{}' is not a string", + field.name); + } + *result = Field::create_field<TYPE_STRING>(std::move(serialized_value)); + return Status::OK(); + } + normalize_iceberg_json_timestamp(primitive_type, &serialized_value); + RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(serialized_value, *result)); + return Status::OK(); +} + +static Status build_v2_json_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + const rapidjson::Value& json_value, + std::deque<std::string>* binary_storage, Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(binary_storage != nullptr); + DORIS_CHECK(result != nullptr); + if (json_value.IsNull()) { + return build_v2_null_default(field, data_type, result); + } + + const auto value_type = remove_nullable(data_type); + switch (value_type->get_primitive_type()) { + case TYPE_STRUCT: + return build_v2_json_struct_default(field, value_type, json_value, binary_storage, result); + case TYPE_ARRAY: + return build_v2_json_array_default(field, value_type, json_value, binary_storage, result); + case TYPE_MAP: + return build_v2_json_map_default(field, value_type, json_value, binary_storage, result); + default: + return build_v2_json_scalar_default(field, value_type, json_value, binary_storage, result); + } +} + +static Status build_v2_initial_default_field(const format::ColumnDefinition& field, + const DataTypePtr& data_type, + std::deque<std::string>* binary_storage, + Field* result) { + DORIS_CHECK(data_type != nullptr); + DORIS_CHECK(binary_storage != nullptr); + DORIS_CHECK(result != nullptr); + if (!field.initial_default_value.has_value()) { + if (field.is_optional.has_value() && !*field.is_optional) { + return Status::InvalidArgument( + "Required Iceberg field '{}' is missing from the data file and has no initial " + "default", + field.name); + } + return build_v2_null_default(field, data_type, result); + } + + const auto value_type = remove_nullable(data_type); + const auto primitive_type = value_type->get_primitive_type(); + if (is_complex_type(primitive_type)) { + rapidjson::Document document; + document.Parse(field.initial_default_value->data(), field.initial_default_value->size()); + if (document.HasParseError()) { + return Status::InvalidArgument("Invalid Iceberg JSON initial default for field '{}'", + field.name); + } + if (primitive_type == TYPE_STRUCT && + (!document.IsObject() || document.MemberCount() != 0)) { + return Status::InvalidArgument( + "Iceberg struct field '{}' has a non-empty initial default", field.name); + } + return build_v2_json_default_field(field, data_type, document, binary_storage, result); + } + + if (field.initial_default_value_is_base64 || primitive_type == TYPE_VARBINARY) { + binary_storage->emplace_back(); + if (!base64_decode(*field.initial_default_value, &binary_storage->back())) { + return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field {}", + field.name); + } + if (primitive_type == TYPE_VARBINARY) { + *result = Field::create_field<TYPE_VARBINARY>(StringView(binary_storage->back())); + } else if (is_string_type(primitive_type)) { + *result = Field::create_field<TYPE_STRING>(binary_storage->back()); + } else { + return Status::InvalidArgument( + "Base64 Iceberg initial default has incompatible Doris type {} for field {}", + data_type->get_name(), field.name); + } + return Status::OK(); + } + + RETURN_IF_ERROR(value_type->get_serde()->from_fe_string(*field.initial_default_value, *result)); + return Status::OK(); +} + +static Status build_initial_default_literal(const format::ColumnDefinition& table_field, + VExprSPtr* literal) { + DORIS_CHECK(table_field.type != nullptr); + DORIS_CHECK(table_field.initial_default_value.has_value()); + DORIS_CHECK(literal != nullptr); + + std::deque<std::string> binary_storage; + Field initial_default; + RETURN_IF_ERROR(build_v2_initial_default_field(table_field, table_field.type, &binary_storage, + &initial_default)); + // VLiteral inserts the Field into an owning column before binary_storage is destroyed. + *literal = VLiteral::create_shared(table_field.type, initial_default); + return Status::OK(); +} + +static Status build_initial_default_exprs(format::ColumnDefinition* column) { + DORIS_CHECK(column != nullptr); + if (column->initial_default_value.has_value()) { + VExprSPtr literal; + RETURN_IF_ERROR(build_initial_default_literal(*column, &literal)); + column->default_expr = VExprContext::create_shared(std::move(literal)); + } + for (auto& child : column->children) { + RETURN_IF_ERROR(build_initial_default_exprs(&child)); + } + return Status::OK(); +} + static Status build_missing_equality_delete_key_expr(const format::ColumnDefinition& table_field, const DataTypePtr& delete_key_type, + bool require_complete_metadata, VExprSPtr* key_expr) { DORIS_CHECK(delete_key_type != nullptr); DORIS_CHECK(key_expr != nullptr); if (!table_field.initial_default_value.has_value()) { + if (require_complete_metadata && !table_field.is_optional.has_value()) { + return Status::InvalidArgument( + "Iceberg equality delete field '{}' is missing optionality metadata", + table_field.name); + } + if (table_field.is_optional.has_value() && !*table_field.is_optional) { + return Status::InvalidArgument("Missing required field: {}", table_field.name); + } // A newly added optional field without an initial default is logically NULL in older // files. EqualityDeletePredicate treats NULL == NULL as a match. *key_expr = VLiteral::create_shared(make_nullable(delete_key_type), Field()); return Status::OK(); } VExprSPtr literal; - if (table_field.initial_default_value_is_base64 || - table_field.type->get_primitive_type() == TYPE_VARBINARY) { - // New FE versions mark every Iceberg UUID/BINARY/FIXED default as Base64 regardless of its - // Doris mapping. Keep the VARBINARY fallback for scan descriptors produced before that - // marker existed. Decode before parsing so STRING/CHAR and VARBINARY all compare against - // the raw bytes stored in equality-delete files. - std::string decoded_default; - if (!base64_decode(*table_field.initial_default_value, &decoded_default)) { - return Status::InvalidArgument("Invalid Base64 Iceberg initial default for field {}", - table_field.name); - } - if (table_field.type->get_primitive_type() == TYPE_VARBINARY) { - const auto initial_default = - Field::create_field<TYPE_VARBINARY>(StringView(decoded_default)); - // VLiteral must copy the borrowed StringView while decoded_default is alive; UUID and - // long FIXED defaults otherwise retain a pointer into freed decode storage. - literal = VLiteral::create_shared(table_field.type, initial_default); - } else { - DORIS_CHECK(is_string_type(table_field.type->get_primitive_type())); - literal = VLiteral::create_shared(table_field.type, - Field::create_field<TYPE_STRING>(decoded_default)); - } - } else { - // An added field's initial default is its logical value in every older data file that lacks - // the physical column. FE normalizes the string for the current Doris table type. - Field initial_default; - RETURN_IF_ERROR(table_field.type->get_serde()->from_fe_string( - *table_field.initial_default_value, initial_default)); - literal = VLiteral::create_shared(table_field.type, initial_default); - } - - DORIS_CHECK(literal != nullptr); + RETURN_IF_ERROR(build_initial_default_literal(table_field, &literal)); if (table_field.type->equals(*delete_key_type)) { *key_expr = std::move(literal); return Status::OK(); } auto cast_expr = Cast::create_shared(delete_key_type); - cast_expr->add_child(std::move(literal)); + cast_expr->add_child(literal); *key_expr = std::move(cast_expr); return Status::OK(); } +Status IcebergTableReader::annotate_projected_column(const TFileScanSlotInfo& slot_info, + format::ProjectedColumnBuildContext* context, + format::ColumnDefinition* column) const { + RETURN_IF_ERROR(format::TableReader::annotate_projected_column(slot_info, context, column)); + DORIS_CHECK(context != nullptr); + DORIS_CHECK(column != nullptr); + if (!supports_iceberg_scan_semantics_v2(context->scan_params)) { + return Status::OK(); + } + if (!context->schema_column.has_value()) { + return Status::OK(); + } + + auto& schema_column = *context->schema_column; + RETURN_IF_ERROR(build_initial_default_exprs(&schema_column)); Review Comment: [P1] Prefer an exact child name before aliases when assigning complex types This now prepares every recursive default from `schema_column`, but that tree's `find_struct_child_type_by_external_field()` performs a single ordered scan where a historical alias can win before a later exact current name. For example, rename nested complex `b` to `renamed_b`, add a different complex field named `b`, then move the new `b` before `renamed_b`. The renamed field matches the new `b` through alias `b`; unlike primitive leaves, `restore_current_primitive_type()` keeps the fallback type for STRUCT/LIST/MAP. A valid STRUCT initial default `{}` can then be parsed as the new ARRAY/MAP type and fail the scan, or descendant metadata can be associated with the wrong shape. Please do a full exact-current-name pass before aliases (or resolve by field ID) and cover reordered same-name reuse with different complex types. -- 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]
