924060929 commented on code in PR #65851: URL: https://github.com/apache/doris/pull/65851#discussion_r3654263268
########## 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: [P0] Branch writes now resolve columns against the branch snapshot's schema In Iceberg the schema is table-level metadata; a branch does not carry its own schema. `SnapshotUtil.schemaFor(table, snapshotId)` answers "which schema was this snapshot written with", which is a read/time-travel question. Using it to resolve a write target changes `INSERT INTO t@branch(b)` from the table's current schema to a historical one. Two existing regression contracts were flipped in this PR to accommodate that, which is what makes the change visible: - `iceberg_branch_tag_schema_change_extended.groovy`: writing a column that was just added on main into a branch now has to be asserted as `exception "Unknown column 'new_col' in target table"`, and `.out` changes `3\t30\ttest` to `3\t30\t\N`. - `test_iceberg_schema_ref_actions_matrix.groovy` T09: the comment `// Scenario T09 negative contract: a pre-rename branch write uses main's latest schema.` is inverted to `// uses the branch snapshot's schema`, and a fresh `pre_rename_write_branch` is introduced so the old branch's assertions still pass. Neither the PR description nor the release note mentions this, and it is independent of default values. Beyond the usability loss (you can no longer write a newly added column into a branch), a dropped column becomes a silent data-loss path: ```sql ALTER TABLE t DROP COLUMN c; -- field id 5 leaves the current schema INSERT INTO t@branch(b) (id, c) VALUES (1, 'x'); -- before: column c is not found -- after: accepted, branch snapshot still has id 5 ``` `IcebergTableSink.bindDataSink` sends the pinned schema as `schemaJson`, so BE writes a data file containing field id 5, while the commit records the table's current schema id. No reader can ever see that value. Suggestion: keep `table.schema()` for branch writes and use the branch ref only to pick the commit target. If the historical-schema behavior is genuinely wanted, it deserves its own PR, a release note, and an explicit rule for fields absent from the current schema. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/rules/RewriteDefaultExpression.java: ########## @@ -82,6 +83,16 @@ private static Expression rewrite(ExpressionMatchingContext<Default> context) { + column.getName() + "'"); } + Optional<IcebergWriteSchemaContext> icebergContext = context.cascadesContext + .getStatementContext().getIcebergWriteSchemaContext(); + if (icebergContext.isPresent() + && slotRef.getOriginalTable() + .map(table -> icebergContext.get().isTargetTable(table.getId())) + .orElse(false) + && icebergContext.get().findField(column).isPresent()) { + return icebergContext.get().resolveWriteDefault(column); Review Comment: [P1] `DEFAULT(col)` in UPDATE / MERGE-MATCHED silently degrades to NULL This branch only fires for INSERT, because `StatementContext.icebergWriteSchemaContext` is deliberately left empty for UPDATE and MERGE: ```java // StatementContext.java // Present only while analyzing a normal Iceberg INSERT. UPDATE and MERGE UPDATE deliberately // leave it empty so DEFAULT(column) keeps its existing non-insert semantics. ``` The problem is that the "existing non-insert semantics" were themselves removed by this PR. `IcebergUtils.parseField` no longer populates `Column.defaultValue`, so the fallback below is now unreachable in its useful form: ```java // Column.java:606 public String getDefaultValueSql() { if (defaultValue == null) { return null; } // always taken for Iceberg columns now ``` and `rewrite()` falls through to `if (column.isAllowNull()) return new NullLiteral(targetType);`. Concretely, `UPDATE t SET c = DEFAULT(c)` returned the Iceberg `initial-default` before this PR and returns NULL after it. That also splits one statement in half: inside a single `MERGE`, `WHEN NOT MATCHED THEN INSERT` resolves through `writeSchemaContext.resolveWriteDefault(column)` (correct `write-default`), while `WHEN MATCHED THEN UPDATE SET c = DEFAULT(c)` yields NULL. For an UPDATE the semantically correct value is the `write-default`, same as the INSERT branch. Suggestion: set the statement context for UPDATE/MERGE as well and let this branch handle both, or state the intended UPDATE semantics explicitly and cover it with a regression test — currently neither the old nor the new behavior is pinned. ########## 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(); Review Comment: [P0] `refresh()` mutates the shared cached `Table` and forks the write schema away from the read schema `dorisTable.getIcebergTable()` returns the instance owned by the metadata cache: ```java // IcebergUtils.java:937 public static Table getIcebergTable(ExternalTable dorisTable) { if (useSessionCatalog(dorisTable)) { return loadIcebergTableWithSession(dorisTable); } return icebergExternalMetaCache(dorisTable).getIcebergTable(dorisTable); } ``` So `table.refresh()` here (and again in `validateCurrentSchema`) reloads shared state on behalf of one statement. Three consequences: **1. Write path sees the latest remote schema, analysis path does not.** `refresh()` does not invalidate `ExternalSchemaCache` or the snapshot cache, and everything user-facing goes through them: ```java // IcebergUtils.java:1909 — feeds getFullSchema(), DESC, and every SELECT public static List<Column> getIcebergSchema(ExternalTable dorisTable) { Optional<MvccSnapshot> snapshotFromContext = MvccUtil.getSnapshotFromContext(dorisTable); IcebergSnapshotCacheValue cacheValue = getSnapshotCacheValue(snapshotFromContext, dorisTable); return getSchemaCacheValue(dorisTable, cacheValue).getSchema(); } ``` After an external `ADD COLUMN` and before `REFRESH TABLE`, `bindColumns` comes from the pinned (N+1) schema while the child output comes from the cached (N) schema: ```sql INSERT INTO t SELECT * FROM t; -- insert into cols should be corresponding to the query output. Expected 4 columns but got 3 INSERT INTO t VALUES (...); -- Column count doesn't match value count ``` That divergence is the steady state for an external catalog inside its cache TTL, not a concurrent-DDL race. The PR describes this as rejecting schema skew before the transaction starts, but the skew being rejected is FE cache staleness, which used to be harmless because reads and writes shared it. **2. Cross-statement side effect.** Scan planning reads the live object, e.g. `IcebergUtils.convertToIcebergExpr(conjunct, icebergTable.schema())` in `planFileScanTaskWithManifestCache`. A concurrent INSERT can change `icebergTable.schema()` while another query is mid-planning. **3. Remote I/O under a lock.** `InsertUtils.pinIcebergWriteSchema` is called inside `targetTableIf.readLock()` in `InsertIntoTableCommand.initPlanOnce`, and `getExplainPlan` triggers it too, so `EXPLAIN INSERT` also performs a catalog round trip. Suggestion: pin from the same MVCC snapshot / schema cache the analyzer already used, so reads and writes share one source of truth, and leave the freshness check to `validateCurrentSchema` at commit time. If the newest metadata really is required, fetch it without publishing it into the shared cache entry. -- 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]
