From 2937f3442328f6ab64215bcaea1aa079c57af64c Mon Sep 17 00:00:00 2001 From: morningman Date: Fri, 23 Oct 2026 10:21:46 +0800 Subject: [PATCH 1/2] [fix](fe) Bind Iceberg branch writes to the table's current schema ### What problem does this PR solve? Issue Number: None Related PR: #66538, #65851 Problem Summary: A write to an Iceberg branch (INSERT [OVERWRITE] INTO t@branch(b)) pins the schema of the branch head snapshot as its writer schema, while a read of the branch uses the table-current schema. Iceberg validates a branch commit against the table-current schema and stamps that schema on the new snapshot, Spark binds branch writes to the current schema, and the Iceberg docs ("Schema selection with branches and tags") say writes to a branch use the table schema. So once the table schema evolves, a write to an older branch rejects the current column names and accepts the old ones, writes the branch head's write-default instead of the current one, and fails outright when the current partition spec has a source column the branch head schema does not have. Master also needs an extra check (validateBranchWriterSchema) because the files it writes carry a different schema than the one Iceberg records for the new snapshot. branch-4.1 and branch-4.2 (#66538) bind branch writes to table.schema(). This change pins table.schema() for branch writes, keeps the check that the target is an existing branch and not a tag, and fences the commit on the table-current schema id for branch writes as for writes to main. The branch head checks are removed. ### Release note Fix: Writes to an Iceberg branch now use the table's current schema (column names, write-defaults and partition spec), as Spark does, instead of the schema of the branch's head snapshot. ### Check List (For Author) - Test: Regression test / Unit Test - test_iceberg_initial_defaults (a branch write gets the current write-default 36), write/test_iceberg_write_evolution_refs, branch_tag/iceberg_branch_tag_schema_change_extended and test_iceberg_schema_ref_actions_matrix (an older branch takes the current column names and rejects the old ones) now expect what branch-4.2 expects. They pass locally; without this change each of them fails at the new expectation - IcebergWritePlanProviderTest: a branch created before a rename writes the new name; the branch write uses the current write-defaults, binds a current required column without a default, and binds the current partition spec whose source column the branch head lacks; a schema change after pinning fails the commit; a tag or a missing ref is rejected - Behavior changed: Yes. See the release note. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 (1M context) --- .../iceberg/IcebergWritePlanProvider.java | 5 +- .../iceberg/IcebergWriteSchemaContext.java | 60 ++------- .../iceberg/IcebergWritePlanProviderTest.java | 118 +++++++++++++----- .../iceberg/test_iceberg_initial_defaults.out | 8 +- .../test_iceberg_write_evolution_refs.out | 8 +- ...g_branch_tag_schema_change_extended.groovy | 9 +- .../test_iceberg_initial_defaults.groovy | 8 +- ...t_iceberg_schema_ref_actions_matrix.groovy | 14 +-- .../test_iceberg_write_evolution_refs.groovy | 28 +---- 9 files changed, 123 insertions(+), 135 deletions(-) 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/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) From 49990114c9502537c6c096da0dd1eeeb8c833b6e Mon Sep 17 00:00:00 2001 From: morningman Date: Fri, 23 Oct 2026 22:57:43 +0800 Subject: [PATCH 2/2] [fix](fe) Fail an INSERT OVERWRITE whose writer schema changes between its planning passes ### What problem does this PR solve? Issue Number: None Related PR: #66538, #65851 Problem Summary: Review follow-up. INSERT OVERWRITE into a connector table plans twice. InsertOverwriteTableCommand.run pins the writer schema and normalizes the plan against it: bare DEFAULT and empty rows are expanded by position and values are cast to the pinned column types. The inner InsertIntoTableCommand that runs the plan is built with needNormalizePlan=false, and every attempt of its initPlan resets the statement's connector scope and pins the writer schema again. Without a column list BindSink then maps the normalized values by position onto the second pin, and every later fence (the bound write metadata identity, validateBoundWriteColumns, the begin and commit schema checks) compares against that second pin. So when another client reorders the columns between the passes, e.g. updateSchema().moveFirst("b") on [a, b] during INSERT OVERWRITE t VALUES (DEFAULT, DEFAULT) the defaults are silently written as b <- 1, a <- 2. Main-branch overwrites have had this exposure since #65851 (pin, then expand defaults) and #66529 (repin on every attempt); the previous commit gives branch overwrites the same one, because a branch write now pins the table's current schema. branch-4.2 carries the pinned writer context from the outer pass into the inner insert, so both passes share one generation. The overwrite now records on the inner insert's context the writer columns and write metadata identity it normalized against, and the inner insert, after pinning again, fails the statement with "retry the statement" when either differs. Nothing changes when the table did not change between the passes or for any other insert. ### Release note None ### Check List (For Author) - Test: Unit Test - InsertUtilsTest: a reordered writer schema and a changed write metadata identity between the passes fail; the same generation, and an insert that normalizes its own plan, pass - InsertOverwriteTableCommandTest, InsertOverwriteDroppedTargetTest, InsertOverwriteSkipAuthResetTest, IcebergStaticPartitionLiteralTest, InsertIntoTableCommandTest, InsertIntoTableCommandTableStreamTest, CloudInsertProfileRetryTest, DictionaryInsertTargetDropRaceTest, PluginDrivenTableSinkBindingTest, PluginDrivenTableSinkTest, ReadLockTest, RowLevelDmlCommandTest - Behavior changed: Yes. An INSERT OVERWRITE into a connector table whose writer schema or write metadata changes while the statement plans now fails with "retry the statement" instead of binding its values by position to the new schema. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 (1M context) --- .../insert/InsertIntoTableCommand.java | 1 + .../insert/InsertOverwriteTableCommand.java | 6 +- .../plans/commands/insert/InsertUtils.java | 54 +++++++++++++ .../PluginDrivenInsertCommandContext.java | 19 +++++ .../commands/insert/InsertUtilsTest.java | 78 +++++++++++++++++++ 5 files changed, 157 insertions(+), 1 deletion(-) 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", "");