Skip to content

[fix](fe) Bind Iceberg branch writes to the table's current schema #69499

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
morningman merged 2 commits into from
Oct 25, 2026
+280 −136
Merged

[fix](fe) Bind Iceberg branch writes to the table's current schema #69499

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2937f34
[fix](fe) Bind Iceberg branch writes to the table's current schema
morningman Oct 23, 2026
4999011
[fix](fe) Fail an INSERT OVERWRITE whose writer schema changes betwee…
morningman Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions ...or-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWritePlanProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<NestedField> currentColumns = writeSchema.columns();
boolean rowLevelWrite = writeOperation == WriteOperation.UPDATE || writeOperation == WriteOperation.MERGE;
Expand Down
60 changes: 9 additions & 51 deletions ...r-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergWriteSchemaContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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();

@github-actions github-actions Bot Oct 23, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve the writer generation across overwrite replanning

For a v3 table ordered [a, b] with write-defaults 1 and 2, run INSERT OVERWRITE t@branch(br) VALUES (DEFAULT, DEFAULT). InsertOverwriteTableCommand.run first pins that schema and normalizePlan produces InlineTable(1, 2). If another Iceberg client commits updateSchema().moveFirst("b").commit() before the inner INSERT starts, InsertIntoTableCommand.initPlan calls resetConnectorStatementScope() and discards both the original schema and metadata identity. It then pins [b, a] while needNormalizePlan=false retains those already-expanded literals. BindSink.getColumnToChildOutput maps by ordinal, producing Sink[b <- 1, a <- 2]: the defaults are silently written into the wrong columns, and all schema checks pass against the second pin. Before this change, a table-only reorder left both branch-head schema pins unchanged. Preserve the outer writer context/identity across this handoff, or restart normalization from the original DEFAULT expressions after repinning; cover the inter-pass reorder with a controlled concurrency test.

@morningman morningman Oct 23, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in 4999011. The interleaving is real, and it is not specific to branch writes: main-branch INSERT OVERWRITE on master has had the same exposure since #65851 (the outer pass pins the writer schema and then expands DEFAULT and bare rows by position against it) and #66529 (every attempt of the inner insert resets the statement scope and pins again). Before this PR a branch overwrite was exposed only when a concurrent commit moved the branch head onto a snapshot with another schema; binding branch writes to the table's current schema widened that to the same table-level changes as main.

The overwrite now records on the inner insert's PluginDrivenInsertCommandContext the writer columns and write metadata identity it normalized the plan against (InsertUtils.recordNormalizedWriteGeneration). After the inner insert pins again, InsertUtils.checkNormalizedWriteGeneration compares them (columns by position: name, field ID, type, nullability and default) and fails the statement with "retry the statement" when either differs, instead of binding the expanded values by position to the new schema. This covers main and branch overwrites alike, and an insert that normalizes its own plan is unaffected. branch-4.2 avoids the hole differently: it carries the pinned IcebergWriteSchemaContext on the unbound sink into the inner insert, so both passes share one generation, and its commit fence rejects a later change with "retry the statement". InsertUtilsTest replays the two pins with a reordered schema and with a changed identity (both fail without the check) and with an unchanged generation.

int formatVersion = IcebergWriterHelper.getFormatVersion(table);
TableIdentity identity = pinTableIdentity(table, formatVersion);
return new IcebergWriteSchemaContext(
Expand Down Expand Up @@ -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);
Expand All @@ -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<Integer, Types.NestedField> branchFields = TypeUtil.indexById(branchSchema.asStruct());
Map<Integer, Types.NestedField> currentFields = TypeUtil.indexById(currentSchema.asStruct());
Map<Integer, Integer> 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) {
Expand All @@ -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))) {
Expand Down
118 changes: 87 additions & 31 deletions ...ceberg/src/test/java/org/apache/doris/connector/iceberg/IcebergWritePlanProviderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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();
Expand All @@ -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();
Expand All @@ -589,11 +592,13 @@ public void planWriteValidatesBoundColumnsAgainstPinnedBranchSchema() {
List<ConnectorColumn> 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
Expand Down Expand Up @@ -673,55 +678,106 @@ 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();
table.manageSnapshots().createBranch("old_schema", table.currentSnapshot().snapshotId()).commit();
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
Expand Down
1 change: 1 addition & 0 deletions ...ain/java/org/apache/doris/nereids/trees/plans/commands/insert/InsertIntoTableCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
6 changes: 5 additions & 1 deletion ...ava/org/apache/doris/nereids/trees/plans/commands/insert/InsertOverwriteTableCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.");
Expand Down
Loading
Loading