diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java index 10a0ff9e6f70f3..80a5e3e429c5f9 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java @@ -303,9 +303,8 @@ private void validateBoundWriteColumns(Table table, IcebergWriteSchemaContext sc if (writeOperation == WriteOperation.DELETE || boundColumns.isEmpty()) { return; } - // A branch write is deliberately bound to the branch-head schema, which can lag the current main - // schema. The schema context separately fences branch evolution, so comparing with table.schema() - // rejects valid branch writes without adding concurrency protection. + // Compare with the schema the statement pinned, which is the table-current schema at pin time for main + // and branch writes alike. The schema context fences any later schema change before the commit. Schema writeSchema = schemaContext == null ? table.schema() : schemaContext.getSchema(); List currentColumns = writeSchema.columns(); boolean rowLevelWrite = writeOperation == WriteOperation.UPDATE || writeOperation == WriteOperation.MERGE; diff --git a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java index bb6c4bb7963cab..6d994856a31146 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java +++ b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java @@ -43,9 +43,7 @@ import org.apache.iceberg.TableMetadata; import org.apache.iceberg.transforms.Transforms; import org.apache.iceberg.types.Type; -import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; -import org.apache.iceberg.util.SnapshotUtil; import java.math.BigDecimal; import java.nio.ByteBuffer; @@ -96,11 +94,12 @@ static IcebergWriteSchemaContext create(Table table, String tableName, Objects.requireNonNull(table, "table should not be null"); Objects.requireNonNull(tableName, "tableName should not be null"); Objects.requireNonNull(branchName, "branchName should not be null"); - Schema schema = branchName.isPresent() - ? resolveBranchSchema(table, branchName.get(), tableName) : table.schema(); if (branchName.isPresent()) { - validateBranchWriterSchema(schema, table.schema(), branchName.get(), tableName); + validateTargetBranch(table, branchName.get(), tableName); } + // A branch selects only the parent snapshot / ref. Iceberg validates a branch write against the + // table-current schema and stamps that schema on the new snapshot, and Spark binds it the same way. + Schema schema = table.schema(); int formatVersion = IcebergWriterHelper.getFormatVersion(table); TableIdentity identity = pinTableIdentity(table, formatVersion); return new IcebergWriteSchemaContext( @@ -273,7 +272,7 @@ private static void validateWriterMetadataSources( } } - private static Schema resolveBranchSchema(Table table, String branchName, String tableName) { + private static void validateTargetBranch(Table table, String branchName, String tableName) { SnapshotRef ref = table.refs().get(branchName); if (ref == null) { throw new DorisConnectorException(branchName + " is not founded in " + tableName); @@ -282,51 +281,13 @@ private static Schema resolveBranchSchema(Table table, String branchName, String throw new DorisConnectorException(branchName + " is a tag, not a branch. Tags cannot be targets for producing snapshots"); } - return SnapshotUtil.schemaFor(table, ref.snapshotId()); - } - - private static void validateBranchWriterSchema( - Schema branchSchema, Schema currentSchema, String branchName, String tableName) { - Map branchFields = TypeUtil.indexById(branchSchema.asStruct()); - Map currentFields = TypeUtil.indexById(currentSchema.asStruct()); - Map currentParents = TypeUtil.indexParents(currentSchema.asStruct()); - for (Types.NestedField currentField : currentFields.values()) { - Types.NestedField branchField = branchFields.get(currentField.fieldId()); - if (branchField != null) { - if (currentField.isRequired() && branchField.isOptional()) { - throw incompatibleBranchSchema(branchSchema, currentSchema, branchName, tableName, - currentField, "is optional in the pinned branch schema and can contain explicit nulls"); - } - continue; - } - Types.NestedField highestMissingField = currentField; - Integer parentId = currentParents.get(currentField.fieldId()); - while (parentId != null && !branchFields.containsKey(parentId)) { - highestMissingField = Preconditions.checkNotNull(currentFields.get(parentId), - "Iceberg parent field %s is absent from current schema", parentId); - parentId = currentParents.get(parentId); - } - if (highestMissingField.isRequired() && highestMissingField.initialDefault() == null) { - throw incompatibleBranchSchema(branchSchema, currentSchema, branchName, tableName, - highestMissingField, - "is absent from the pinned branch schema and has no initial default"); - } - } - } - - private static DorisConnectorException incompatibleBranchSchema( - Schema branchSchema, Schema currentSchema, String branchName, String tableName, - Types.NestedField field, String incompatibility) { - return new DorisConnectorException("Iceberg table current schema " + currentSchema.schemaId() - + " cannot label files written with pinned branch " + branchName + " schema " - + branchSchema.schemaId() + " for table " + tableName + ": required field " - + field.name() + " (id " + field.fieldId() + ") " + incompatibility - + "; retry after updating the branch schema"); } void validateCurrentSchema(Table table, boolean requireCurrentPartitionSpec) { - Schema currentSchema = branchName.isPresent() - ? resolveBranchSchema(table, branchName.get(), tableName) : table.schema(); + if (branchName.isPresent()) { + validateTargetBranch(table, branchName.get(), tableName); + } + Schema currentSchema = table.schema(); int currentFormatVersion = IcebergWriterHelper.getFormatVersion(table); validateTableIdentity(table, currentFormatVersion); if (currentSchema.schemaId() != schema.schemaId() || currentFormatVersion != formatVersion) { @@ -335,9 +296,6 @@ void validateCurrentSchema(Table table, boolean requireCurrentPartitionSpec) { + ", current schema " + currentSchema.schemaId() + "/format " + currentFormatVersion + "; retry the statement"); } - if (branchName.isPresent()) { - validateBranchWriterSchema(schema, table.schema(), branchName.get(), tableName); - } PartitionSpec retainedSpec = table.specs().get(partitionSpec.specId()); if (retainedSpec == null || !partitionSpecJson.equals(PartitionSpecParser.toJson(retainedSpec))) { diff --git a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java index e93355a4e9b6ff..b0eb9bee16956a 100644 --- a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java +++ b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java @@ -101,6 +101,7 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; +import java.util.stream.Collectors; /** * Pins {@link IcebergWritePlanProvider#planWrite} for INSERT/OVERWRITE against legacy @@ -559,7 +560,9 @@ public void variantWriteDefaultsMustBeNull() { } @Test - public void branchWritePinsBranchHeadSchema() { + public void branchWriteUsesTableCurrentSchema() { + // A branch only selects the parent snapshot. Iceberg stamps a branch commit with the table-current + // schema and Spark binds branch writes to it, so a branch created before a rename writes the new name. InMemoryCatalog catalog = freshCatalog(); Table table = unpartitionedUnsortedTable(catalog); table.newAppend().commit(); @@ -569,14 +572,14 @@ public void branchWritePinsBranchHeadSchema() { IcebergWriteSchemaContext writeContext = IcebergWriteSchemaContext.create( table, "db1.t2", Optional.of("old_schema"), false, false); - Assertions.assertNotEquals(table.schema().schemaId(), writeContext.getSchema().schemaId()); - Assertions.assertNotNull(writeContext.getSchema().findField("name")); - Assertions.assertNull(writeContext.getSchema().findField("renamed_name")); + Assertions.assertEquals(table.schema().schemaId(), writeContext.getSchema().schemaId()); + Assertions.assertNotNull(writeContext.getSchema().findField("renamed_name")); + Assertions.assertNull(writeContext.getSchema().findField("name")); Assertions.assertDoesNotThrow(() -> writeContext.validateCurrentSchema(table, false)); } @Test - public void planWriteValidatesBoundColumnsAgainstPinnedBranchSchema() { + public void planWriteBindsBranchColumnsToTableCurrentSchema() { InMemoryCatalog catalog = freshCatalog(); Table table = unpartitionedUnsortedTable(catalog); table.newAppend().commit(); @@ -589,11 +592,13 @@ public void planWriteValidatesBoundColumnsAgainstPinnedBranchSchema() { List branchColumns = provider.getWriteColumns( session, tableHandle, Optional.of("old_schema")).orElseThrow(AssertionError::new); + Assertions.assertEquals(Arrays.asList("id", "renamed_name"), + branchColumns.stream().map(ConnectorColumn::getName).collect(Collectors.toList())); ConnectorSinkPlan plan = Assertions.assertDoesNotThrow(() -> provider.planWrite(session, new WriteHandle(tableHandle).branch("old_schema").boundTargetColumns(branchColumns))); - Assertions.assertTrue(plan.getDataSink().getIcebergTableSink().getSchemaJson().contains("name")); - Assertions.assertFalse(plan.getDataSink().getIcebergTableSink().getSchemaJson().contains("renamed_name")); + Assertions.assertEquals(SchemaParser.toJson(table.schema()), + plan.getDataSink().getIcebergTableSink().getSchemaJson()); } @Test @@ -673,7 +678,27 @@ public void writeRoutingSkipsV1VoidPartitionFields() { } @Test - public void branchWriteRejectsCurrentRequiredFieldAbsentWithoutDefault() { + public void branchWriteRejectsTagOrMissingRef() { + InMemoryCatalog catalog = freshCatalog(); + Table table = unpartitionedUnsortedTable(catalog); + table.newAppend().commit(); + table.manageSnapshots().createTag("old_tag", table.currentSnapshot().snapshotId()).commit(); + + DorisConnectorException tagException = Assertions.assertThrows(DorisConnectorException.class, + () -> IcebergWriteSchemaContext.create(table, "db1.t2", Optional.of("old_tag"), false, false)); + Assertions.assertTrue(tagException.getMessage().contains( + "old_tag is a tag, not a branch. Tags cannot be targets for producing snapshots"), + tagException.getMessage()); + DorisConnectorException missingException = Assertions.assertThrows(DorisConnectorException.class, + () -> IcebergWriteSchemaContext.create(table, "db1.t2", Optional.of("no_such_ref"), false, false)); + Assertions.assertTrue(missingException.getMessage().contains("no_such_ref is not founded in db1.t2"), + missingException.getMessage()); + } + + @Test + public void branchWriteBindsCurrentRequiredFieldWithoutDefault() { + // A required column added on main after the branch is part of the branch write. It has no default, so + // omitting it fails at analysis exactly as it does for a write to main. InMemoryCatalog catalog = freshCatalog(); Table table = unpartitionedUnsortedTable(catalog); table.newAppend().commit(); @@ -681,47 +706,78 @@ public void branchWriteRejectsCurrentRequiredFieldAbsentWithoutDefault() { table.updateSchema().allowIncompatibleChanges() .addRequiredColumn("required_value", Types.IntegerType.get()).commit(); - DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class, - () -> IcebergWriteSchemaContext.create( - table, "db1.t2", Optional.of("old_schema"), false, false)); + IcebergWriteSchemaContext writeContext = IcebergWriteSchemaContext.create( + table, "db1.t2", Optional.of("old_schema"), false, false); - Assertions.assertTrue(exception.getMessage().contains("required field required_value")); - Assertions.assertTrue(exception.getMessage().contains( - "is absent from the pinned branch schema and has no initial default")); + ConnectorColumn requiredValue = writeColumn(writeContext, "required_value"); + Assertions.assertFalse(requiredValue.isNullable()); + Assertions.assertNull(requiredValue.getDefaultValueSql()); + Assertions.assertDoesNotThrow(() -> writeContext.validateCurrentSchema(table, false)); + } + + @Test + public void branchWriteUsesTableCurrentWriteDefaults() { + InMemoryCatalog catalog = freshCatalog(); + Table table = formatVersionThreeTable(catalog); + table.updateSchema().addColumn("default_int", Types.IntegerType.get(), Literal.of(35)).commit(); + table.newAppend().commit(); + table.manageSnapshots().createBranch("old_defaults", table.currentSnapshot().snapshotId()).commit(); + table.updateSchema().updateColumnDefault("default_int", Literal.of(36)) + .addRequiredColumn("required_value", Types.IntegerType.get(), Literal.of(7)).commit(); + + IcebergWriteSchemaContext writeContext = IcebergWriteSchemaContext.create( + table, "db1.tv3", Optional.of("old_defaults"), false, false); + + Assertions.assertEquals("36", writeColumn(writeContext, "default_int").getDefaultValueSql()); + Assertions.assertEquals("7", writeColumn(writeContext, "required_value").getDefaultValueSql()); + Assertions.assertDoesNotThrow(() -> writeContext.validateCurrentSchema(table, false)); } @Test - public void branchWriteRevalidatesFieldMadeRequiredAfterPlanning() { + public void branchWriteBindsPartitionSpecToTableCurrentSchema() { + // The current spec partitions by a column the branch head schema never had. InMemoryCatalog catalog = freshCatalog(); Table table = unpartitionedUnsortedTable(catalog); table.newAppend().commit(); table.manageSnapshots().createBranch("old_schema", table.currentSnapshot().snapshotId()).commit(); - IcebergWriteSchemaContext writeContext = IcebergWriteSchemaContext.create( - table, "db1.t2", Optional.of("old_schema"), false, false); - table.updateSchema().allowIncompatibleChanges().requireColumn("name").commit(); + table.updateSchema().addColumn("region", Types.StringType.get()).commit(); + table.updateSpec().addField("region").commit(); - DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class, - () -> writeContext.validateCurrentSchema(table, false)); + IcebergWriteSchemaContext writeContext = Assertions.assertDoesNotThrow(() -> IcebergWriteSchemaContext.create( + table, "db1.t2", Optional.of("old_schema"), false, false)); - Assertions.assertTrue(exception.getMessage().contains("required field name")); - Assertions.assertTrue(exception.getMessage().contains( - "is optional in the pinned branch schema and can contain explicit nulls")); + Assertions.assertEquals(table.spec().specId(), writeContext.getPartitionSpec().specId()); + Assertions.assertEquals(table.schema().findField("region").fieldId(), + writeContext.getPartitionSpec().fields().get(0).sourceId()); } @Test - public void branchWriteAllowsCurrentRequiredFieldWithInitialDefault() { + public void branchWriteRejectsTableSchemaChangeBeforeCommit() { + // The branch head schema stays the same, but the files were written with the pinned current schema, so + // any later change to the table-current schema must fail the commit, as it does for a write to main. InMemoryCatalog catalog = freshCatalog(); - Table table = formatVersionThreeTable(catalog); + Table table = unpartitionedUnsortedTable(catalog); table.newAppend().commit(); table.manageSnapshots().createBranch("old_schema", table.currentSnapshot().snapshotId()).commit(); - table.updateSchema().addRequiredColumn( - "required_value", Types.IntegerType.get(), Literal.of(7)).commit(); - IcebergWriteSchemaContext writeContext = IcebergWriteSchemaContext.create( - table, "db1.tv3", Optional.of("old_schema"), false, false); + table, "db1.t2", Optional.of("old_schema"), false, false); + int pinnedSchemaId = table.schema().schemaId(); + table.updateSchema().addColumn("late_value", Types.IntegerType.get()).commit(); - Assertions.assertNull(writeContext.getSchema().findField("required_value")); - Assertions.assertDoesNotThrow(() -> writeContext.validateCurrentSchema(table, false)); + DorisConnectorException exception = Assertions.assertThrows(DorisConnectorException.class, + () -> writeContext.validateCurrentSchema(table, false)); + + Assertions.assertTrue(exception.getMessage().contains("pinned schema " + pinnedSchemaId + "/format"), + exception.getMessage()); + Assertions.assertTrue(exception.getMessage().contains( + "current schema " + table.schema().schemaId() + "/format"), exception.getMessage()); + } + + private static ConnectorColumn writeColumn(IcebergWriteSchemaContext writeContext, String name) { + return writeContext.getColumns().stream() + .filter(column -> column.getName().equals(name)) + .findFirst() + .orElseThrow(() -> new AssertionError("missing write column " + name)); } @Test diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java index cfdd3533232c24..af34ac1c3cfd77 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java @@ -462,6 +462,7 @@ private BuildInsertExecutorResult initPlanOnce(ConnectContext ctx, ); InsertUtils.pinConnectorWriteSchema( ctx.getStatementContext(), targetTableIf, originLogicalQuery, branchName); + InsertUtils.checkNormalizedWriteGeneration(ctx.getStatementContext(), targetTableIf, insertCtx); if (!(this instanceof InsertIntoDictionaryCommand)) { // process inline table (default values, empty values) if (needNormalizePlan) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java index a3c3123635e166..dce1771dcc3e10 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java @@ -600,10 +600,14 @@ private InsertCommandContext insertIntoPartitions(ConnectContext ctx, StmtExecut // reuses this ctx) so the connector points the overwrite commit at the branch. The guard above // already rejected @branch for connectors without supportsWriteBranch(). branchName.ifPresent(notUsed -> pluginCtx.setBranchName(branchName)); + TableIf targetTable = InsertUtils.getTargetTable(logicalQuery, ctx); if (sink.hasStaticPartition()) { pluginCtx.setStaticPartitionSpecFromExpressions(sink.getStaticPartitionKeyValues(), - InsertUtils.connectorWriteSchema(InsertUtils.getTargetTable(logicalQuery, ctx), true)); + InsertUtils.connectorWriteSchema(targetTable, true)); } + // The child of the copied sink was normalized against the writer schema pinned in run(); the inner + // insert does not normalize it again, so it checks that it pins the same generation. + InsertUtils.recordNormalizedWriteGeneration(ctx.getStatementContext(), targetTable, pluginCtx); insertCtx = pluginCtx; } else { throw new UserException("Current catalog does not support insert overwrite yet."); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java index c374abe4ef3db2..36ee0bcddc8765 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtils.java @@ -101,6 +101,7 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -501,6 +502,59 @@ static void pinConnectorWriteSchema(StatementContext statementContext, TableIf t targetTableIf.getId(), columns)); } + /** + * Records on the context of an INSERT OVERWRITE's inner insert the writer generation the plan of the overwrite + * was normalized against. + */ + static void recordNormalizedWriteGeneration(StatementContext statementContext, TableIf targetTableIf, + PluginDrivenInsertCommandContext insertCtx) { + statementContext.getConnectorWriteSchema(targetTableIf.getId()) + .ifPresent(writeSchema -> insertCtx.setNormalizedWriteGeneration(writeSchema, + statementContext.getConnectorWriteMetadataIdentity(targetTableIf.getId()))); + } + + /** + * INSERT OVERWRITE normalizes its plan against the writer schema it pinned, expanding DEFAULT and bare rows by + * position and casting values to the column types, and then runs an inner insert that pins the writer schema + * again. Without a column list that insert binds the normalized values by position, so values normalized for + * one generation would be written into other columns of another. Fail the statement instead, as the commit + * fences do for a schema that changes later. + */ + static void checkNormalizedWriteGeneration(StatementContext statementContext, TableIf targetTableIf, + Optional insertCtx) { + if (!insertCtx.isPresent() || !(insertCtx.get() instanceof PluginDrivenInsertCommandContext)) { + return; + } + PluginDrivenInsertCommandContext pluginCtx = (PluginDrivenInsertCommandContext) insertCtx.get(); + if (!pluginCtx.getNormalizedWriteSchema().isPresent()) { + return; + } + Optional> writeSchema = statementContext.getConnectorWriteSchema(targetTableIf.getId()); + if (!writeSchema.isPresent() + || !sameWriteGeneration(pluginCtx.getNormalizedWriteSchema().get(), writeSchema.get()) + || !pluginCtx.getNormalizedWriteMetadataIdentity().equals( + statementContext.getConnectorWriteMetadataIdentity(targetTableIf.getId()))) { + throw new AnalysisException("Table " + targetTableIf.getName() + + " changed while INSERT OVERWRITE was planning its write; retry the statement"); + } + } + + private static boolean sameWriteGeneration(List normalized, List pinned) { + if (normalized.size() != pinned.size()) { + return false; + } + for (int i = 0; i < normalized.size(); i++) { + Column expected = normalized.get(i); + Column actual = pinned.get(i); + if (!expected.getName().equals(actual.getName()) || expected.getUniqueId() != actual.getUniqueId() + || !expected.getType().equals(actual.getType()) || expected.isAllowNull() != actual.isAllowNull() + || !Objects.equals(expected.getDefaultValueSql(), actual.getDefaultValueSql())) { + return false; + } + } + return true; + } + static Column resolveExplicitConnectorWriteColumn(TableIf table, String name) { Column column = connectorWriteSchema(table, true).stream() .filter(candidate -> candidate.getName().equalsIgnoreCase(name)) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java index 7d96149086a444..e29279aa631332 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/PluginDrivenInsertCommandContext.java @@ -63,6 +63,12 @@ public class PluginDrivenInsertCommandContext extends BaseExternalTableInsertCom // The target schema the sink was bound to, whose column types the static partition values are cast to. private List boundTargetSchema = Collections.emptyList(); private Optional branchName = Optional.empty(); + // The writer schema and write metadata identity an INSERT OVERWRITE normalized its plan against; the inner + // insert that runs the plan pins them again and must find the same generation (see + // InsertUtils#checkNormalizedWriteGeneration). Empty for any other insert, and when the connector pins no + // writer schema. + private Optional> normalizedWriteSchema = Optional.empty(); + private Optional normalizedWriteMetadataIdentity = Optional.empty(); public Map getStaticPartitionSpec() { return staticPartitionSpec; @@ -163,4 +169,17 @@ public Optional getBranchName() { public void setBranchName(Optional branchName) { this.branchName = branchName == null ? Optional.empty() : branchName; } + + public Optional> getNormalizedWriteSchema() { + return normalizedWriteSchema; + } + + public Optional getNormalizedWriteMetadataIdentity() { + return normalizedWriteMetadataIdentity; + } + + public void setNormalizedWriteGeneration(List writeSchema, Optional writeMetadataIdentity) { + this.normalizedWriteSchema = Optional.of(writeSchema); + this.normalizedWriteMetadataIdentity = writeMetadataIdentity; + } } diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java index 2d42256d0c29ba..e624fecc4b0eb5 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertUtilsTest.java @@ -306,6 +306,84 @@ public void pinConnectorWriteSchemaUsesTargetBranchAndKeepsPinnedSchema() { Mockito.verify(table, Mockito.times(1)).resolveWriteColumns(branchName); } + @Test + public void overwriteInnerInsertRejectsAWriterSchemaReorderedAfterNormalization() { + // INSERT OVERWRITE t VALUES (DEFAULT, DEFAULT) expands the defaults by position against [a, b]. When another + // client moves b first before the inner insert pins the writer schema again, binding the expanded row by + // position would write the default of a into b and the default of b into a. + StatementContext statementContext = new StatementContext(); + PluginDrivenExternalTable table = overwriteTarget(); + Column a = writeColumn("a", 1, "1"); + Column b = writeColumn("b", 2, "2"); + Mockito.when(table.resolveWriteColumns(Optional.empty())) + .thenReturn(Optional.of(ImmutableList.of(a, b)), Optional.of(ImmutableList.of(b, a))); + PluginDrivenInsertCommandContext insertCtx = normalizeThenRepin(statementContext, table, "g1", "g1"); + + AnalysisException e = Assertions.assertThrows(AnalysisException.class, + () -> InsertUtils.checkNormalizedWriteGeneration(statementContext, table, Optional.of(insertCtx))); + Assertions.assertTrue(e.getMessage().contains( + "Table t changed while INSERT OVERWRITE was planning its write; retry the statement"), e.getMessage()); + } + + @Test + public void overwriteInnerInsertRejectsAnotherWriteMetadataIdentity() { + // The columns are the same, but the generation the connector fences writes with (for Iceberg the schema + // ID, partition spec and sort order) moved on between the two pins. + StatementContext statementContext = new StatementContext(); + PluginDrivenExternalTable table = overwriteTarget(); + Mockito.when(table.resolveWriteColumns(Optional.empty())) + .thenReturn(Optional.of(ImmutableList.of(writeColumn("a", 1, "1"), writeColumn("b", 2, "2")))); + PluginDrivenInsertCommandContext insertCtx = normalizeThenRepin(statementContext, table, "g1", "g2"); + + Assertions.assertThrows(AnalysisException.class, + () -> InsertUtils.checkNormalizedWriteGeneration(statementContext, table, Optional.of(insertCtx))); + } + + @Test + public void overwriteInnerInsertAcceptsTheGenerationItWasNormalizedAgainst() { + StatementContext statementContext = new StatementContext(); + PluginDrivenExternalTable table = overwriteTarget(); + Mockito.when(table.resolveWriteColumns(Optional.empty())) + .thenReturn(Optional.of(ImmutableList.of(writeColumn("a", 1, "1"), writeColumn("b", 2, "2"))), + Optional.of(ImmutableList.of(writeColumn("a", 1, "1"), writeColumn("b", 2, "2")))); + PluginDrivenInsertCommandContext insertCtx = normalizeThenRepin(statementContext, table, "g1", "g1"); + + Assertions.assertDoesNotThrow( + () -> InsertUtils.checkNormalizedWriteGeneration(statementContext, table, Optional.of(insertCtx))); + // Any other insert normalizes its own plan against its own pin. + Assertions.assertDoesNotThrow(() -> InsertUtils.checkNormalizedWriteGeneration( + statementContext, table, Optional.of(new PluginDrivenInsertCommandContext()))); + } + + private static PluginDrivenInsertCommandContext normalizeThenRepin(StatementContext statementContext, + PluginDrivenExternalTable table, String normalizedIdentity, String repinnedIdentity) { + UnboundConnectorTableSink sink = Mockito.mock(UnboundConnectorTableSink.class); + // The overwrite pins the writer schema, normalizes its plan against it and records that generation. + InsertUtils.pinConnectorWriteSchema(statementContext, table, sink, Optional.empty()); + statementContext.setConnectorWriteMetadataIdentity(table.getId(), normalizedIdentity); + PluginDrivenInsertCommandContext insertCtx = new PluginDrivenInsertCommandContext(); + InsertUtils.recordNormalizedWriteGeneration(statementContext, table, insertCtx); + // Each attempt of the inner insert drops the pins of the statement and pins the writer schema again. + statementContext.resetConnectorStatementScope(); + InsertUtils.pinConnectorWriteSchema(statementContext, table, sink, Optional.empty()); + statementContext.setConnectorWriteMetadataIdentity(table.getId(), repinnedIdentity); + return insertCtx; + } + + private static PluginDrivenExternalTable overwriteTarget() { + PluginDrivenExternalTable table = Mockito.mock(PluginDrivenExternalTable.class); + Mockito.when(table.getId()).thenReturn(17L); + Mockito.when(table.getName()).thenReturn("t"); + return table; + } + + private static Column writeColumn(String name, int fieldId, String defaultSql) { + Column column = new Column(name, PrimitiveType.INT, true); + column.setUniqueId(fieldId); + column.setConnectorDefaultValueSql(defaultSql); + return column; + } + @Test public void nativeDefaultStillExpandsExplicitDefault() { Column column = new Column("v", Type.INT, false, null, true, "7", ""); diff --git a/regression-test/data/external_table_p0/iceberg/test_iceberg_initial_defaults.out b/regression-test/data/external_table_p0/iceberg/test_iceberg_initial_defaults.out index bae87ce9b5fa15..d2945f45cf8df0 100644 --- a/regression-test/data/external_table_p0/iceberg/test_iceberg_initial_defaults.out +++ b/regression-test/data/external_table_p0/iceberg/test_iceberg_initial_defaults.out @@ -388,20 +388,20 @@ missing_map_col map> Yes true \N 4 123e4567-e89b-12d3-a456-426614174000 0A0B0C0D 2A2B2C 2024-12-17 23:59:59.123456+00:00 -- !parquet_branch_overwrite_default -- -14 35 +14 36 -- !parquet_main_overwrite_default -- 15 36 -- !parquet_branch_after_main_overwrite -- -14 35 +14 36 -- !orc_branch_overwrite_default -- -14 35 +14 36 -- !orc_main_overwrite_default -- 15 36 -- !orc_branch_after_main_overwrite -- -14 35 +14 36 diff --git a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.out b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.out index ca7df7a0a1c2a3..aa4ad7c868de43 100644 --- a/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.out +++ b/regression-test/data/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.out @@ -38,12 +38,13 @@ 4 CN-east new-spec 5 DE-west \N 6 \N null-zone + -- !branch_after_insert -- 1 CN base-cn \N 2 US base-us \N 3 \N \N \N -7 JP-east branch-insert \N -8 FR-west branch-overwrite-seed \N +7 JP-east branch branch-insert +8 FR-west branch-seed branch-overwrite-seed -- !main_unchanged_after_branch_insert -- 1 @@ -57,7 +58,7 @@ 1 CN base-cn \N 2 US base-us \N 3 \N \N \N -7 JP-east branch-insert \N +7 JP-east branch branch-insert 8 FR-west branch-overwrite branch-overwrite -- !base_tag_after_branch_overwrite -- @@ -72,3 +73,4 @@ 4 CN-east new-spec 5 DE-west \N 6 \N null-zone + diff --git a/regression-test/suites/external_table_p0/iceberg/branch_tag/iceberg_branch_tag_schema_change_extended.groovy b/regression-test/suites/external_table_p0/iceberg/branch_tag/iceberg_branch_tag_schema_change_extended.groovy index 6bcde949f75a1d..a97448ffe4e6da 100644 --- a/regression-test/suites/external_table_p0/iceberg/branch_tag/iceberg_branch_tag_schema_change_extended.groovy +++ b/regression-test/suites/external_table_p0/iceberg/branch_tag/iceberg_branch_tag_schema_change_extended.groovy @@ -70,14 +70,9 @@ suite("iceberg_branch_tag_schema_change_extended", "p0,external") { sql """ alter table ${table_name} modify column id bigint """ qt_b3_new_type """ select * from ${table_name}@branch(b3_schema) where id = 1 """ // Should use new type - // Test 3.1.4: Branch writes bind to the branch-head schema + // Test 3.1.4: Branch writes use the shared latest table schema sql """ alter table ${table_name} add column new_col string """ - test { - sql """ insert into ${table_name}@branch(b3_schema)(id, value, new_col) values (3, 30, 'test') """ - exception "Unknown column 'new_col'" - } - // A successful old-schema write advances the branch to a snapshot carrying the current table schema. - sql """ insert into ${table_name}@branch(b3_schema)(id, value) values (4, 40) """ + // Iceberg branch commits advance the branch to a snapshot written with current table metadata. sql """ insert into ${table_name}@branch(b3_schema)(id, value, new_col) values (3, 30, 'test') """ qt_b3_with_new_col """ select * from ${table_name}@branch(b3_schema) where id = 3 """ diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_initial_defaults.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_initial_defaults.groovy index f270b0b423dea6..2ddeb9891a9a7c 100644 --- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_initial_defaults.groovy +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_initial_defaults.groovy @@ -599,10 +599,10 @@ suite("test_iceberg_initial_defaults", "p0,external") { sparkUpdateMergeVerification) } - // Pin a branch while default_int still has write-default 35, then evolve only the main schema - // to 36. INSERT OVERWRITE performs its first VALUES(DEFAULT) normalization before constructing - // the inner INSERT command, so both that first pass and the writer must use the same target-ref - // schema. The branch check also proves the inner command does not silently fall back to main. + // Pin a branch while default_int still has write-default 35, then evolve the table-current + // schema to 36. Iceberg selects the branch as the parent snapshot but writes the new snapshot + // with the table-current schema, so VALUES(DEFAULT) normalization and the writer must both use + // the current value 36 even when the target is the older branch. String oldDefaultsBranch = "before_default_int_36" tableNames.each { String format, String currentTable -> sql """ALTER TABLE ${currentTable} CREATE BRANCH ${oldDefaultsBranch}""" diff --git a/regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_ref_actions_matrix.groovy b/regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_ref_actions_matrix.groovy index 4005f04ca6c2d5..03fa86835d16a1 100644 --- a/regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_ref_actions_matrix.groovy +++ b/regression-test/suites/external_table_p0/iceberg/test_iceberg_schema_ref_actions_matrix.groovy @@ -209,23 +209,19 @@ suite("test_iceberg_schema_ref_actions_matrix", order by id """ - // Scenario T09: writes bind to the target branch's head schema before fast-forward. - sql """ - explain insert into ${fastForwardTable}@branch(pre_rename_branch) - (id, old_name, metric) values (3, 'branch-3', 30) - """ + // Scenario T09: writes use the table's latest schema even when targeting an old branch. test { sql """ insert into ${fastForwardTable}@branch(pre_rename_branch) - (id, new_name, metric) values (3, 'branch-3', 30) + (id, old_name, metric) values (3, 'branch-3', 30) """ - exception "Unknown column 'new_name'" + exception "Unknown column 'old_name'" } - // A historical source relation must not replace the branch-head schema used to bind the target. + // A historical source relation must not replace the latest schema used to bind the branch target. sql """ explain insert into ${fastForwardTable}@branch(pre_rename_branch) - (id, old_name, metric) + (id, new_name, metric) select id, old_name, metric from ${fastForwardTable} for version as of ${preRenameSnapshot} """ diff --git a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy index c87b982ee384df..99f41c1c5a5e8b 100644 --- a/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy +++ b/regression-test/suites/external_table_p0/iceberg/write/test_iceberg_write_evolution_refs.groovy @@ -201,30 +201,16 @@ suite("test_iceberg_write_evolution_refs", order by id """ - // W01-S05: A branch created before schema evolution initially writes with its branch-head - // schema. Columns added or renamed on main are unavailable until the first branch commit - // advances the branch to a new snapshot. - test { - sql """ - insert into evolution_refs@branch(base_branch) - (id, zone, bucket_key, event_time, amount, payload, note) - values - (7, 'JP-east', 'branch-a', '2026-04-01 12:00:00', 70.70, - struct(70, 'branch', 'current-schema'), 'not-written') - """ - exception "Unknown column 'zone' in target table" - } - - // Seed a current-spec partition with the old branch schema. The successful commit advances - // the branch, so subsequent reads and writes expose the current names and added fields. + // W01-S05: Seed the current-spec branch partition before overwriting it; + // this proves replacement semantics while main and the base tag stay isolated. sql """ insert into evolution_refs@branch(base_branch) - (id, region, bucket_key, event_time, amount, payload) + (id, zone, bucket_key, event_time, amount, payload, note) values (7, 'JP-east', 'branch-a', '2026-04-01 12:00:00', 70.70, - struct(70, 'branch-insert')), + struct(70, 'branch', 'current-schema'), 'branch-insert'), (8, 'FR-west', 'branch-b', '2026-05-01 13:00:00', 80.80, - struct(80, 'branch-overwrite-seed')) + struct(80, 'branch-seed', 'current-schema'), 'branch-overwrite-seed') """ order_qt_branch_after_insert """ select id, zone, payload.label, note @@ -245,10 +231,6 @@ suite("test_iceberg_write_evolution_refs", struct(cast(80 as bigint), 'branch-overwrite', 'current-schema'), 'branch-overwrite' """ - assertEquals(0L, (sql """ - select count(*) from evolution_refs@branch(base_branch) - where payload.label = 'branch-overwrite-seed' - """)[0][0] as long) order_qt_branch_after_overwrite """ select id, zone, payload.label, note from evolution_refs@branch(base_branch)