github-actions[bot] commented on code in PR #65851: URL: https://github.com/apache/doris/pull/65851#discussion_r3654191169
########## fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergWriteSchemaContext.java: ########## @@ -0,0 +1,452 @@ +// 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.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 current main or branch 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(); + table.refresh(); + Schema schema = resolveSchema(table, branchName, dorisTable.getName()); + 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 resolveSchema(Table table, Optional<String> branchName, String tableName) { + if (!branchName.isPresent()) { + return table.schema(); + } + SnapshotRef ref = table.refs().get(branchName.get()); + if (ref == null) { + throw new AnalysisException(branchName.get() + " is not founded in " + tableName); + } + if (!ref.isBranch()) { + throw new AnalysisException(branchName.get() + + " is a tag, not a branch. Tags cannot be targets for producing snapshots"); + } + return SnapshotUtil.schemaFor(table, ref.snapshotId()); Review Comment: [P1] Validate branch writes with the table's current schema Iceberg has one schema shared across branches: branch reads use the table schema, and branch writes must also be validated/written with `table.schema()`. Returning the branch head snapshot schema here makes Doris accept removed/renamed columns and omit columns added after the branch point (as the updated regressions now expect). For a V3 field whose initial and write defaults differ, such a branch write produces a file without the current field, so later reads materialize the initial default instead of the required write default. Please pin the current table schema for writes to any branch and restore the regressions to require current column names/types/defaults. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergUpdateCommand.java: ########## @@ -115,6 +118,8 @@ public void run(ConnectContext ctx, StmtExecutor executor) throws Exception { IcebergExternalTable icebergTable = (IcebergExternalTable) table; IcebergDmlCommandUtils.checkUpdateMode(icebergTable); + IcebergWriteSchemaContext writeSchemaContext = IcebergWriteSchemaContext.create( Review Comment: [P1] Install the pinned context before rewriting UPDATE defaults This context is carried to the sink, but it is never installed in `StatementContext` before `planner.plan` (the same is true in `IcebergMergeCommand`). `RewriteDefaultExpression` only reads that statement field. Since this patch also removes Iceberg defaults from the generic `Column` channel, `UPDATE t SET c = DEFAULT(c)` and a matched MERGE update now fall back to NULL for an optional field even when its Iceberg `write-default` is non-null. Please scope the pinned context around both run and explain planning for UPDATE/matched MERGE, and cover a field whose initial default, write default, and NULL are distinct. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/IcebergMergeCommand.java: ########## @@ -345,19 +354,8 @@ private List<Expression> buildInsertProjection(MergeNotMatchedClause clause, value = rowItem.child(0); } } - if (value == null) { - if (column.getDefaultValueSql() != null) { - Expression unboundDefaultValue = new NereidsParser() - .parseExpression(column.getDefaultValueSql()); - if (unboundDefaultValue instanceof UnboundAlias) { - unboundDefaultValue = unboundDefaultValue.child(0); - } - value = unboundDefaultValue; - } else if (column.isAllowNull()) { - value = new NullLiteral(DataType.fromCatalogType(column.getType())); - } else { - throw new AnalysisException("Column has no default value, column=" + column.getName()); - } + if (value == null || value instanceof Default || value instanceof DefaultValueSlot) { Review Comment: [P1] Preserve the column referenced by `DEFAULT(column)` `DefaultValueSlot` is bare `DEFAULT`, but `Default` represents `DEFAULT(referenced_column)`. Combining them here discards the latter's child and always resolves the default for the current destination column. As a result, `DEFAULT(no_such_column)` is accepted and `DEFAULT(id)` in the `name` position silently writes `name`'s write-default. Please resolve only `DefaultValueSlot` from the destination position; bind and validate a `Default` expression's referenced field through the normal rewrite path (or explicitly against the pinned schema), with cross-column and unknown-column MERGE tests. -- 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]
