You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sshivampeta (via GitHub)" <gi...@apache.org> on 2026/05/11 17:02:59 UTC
[PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
sshivampeta opened a new pull request, #16289:
URL: https://github.com/apache/iceberg/pull/16289
Summary
Fixes https://github.com/apache/iceberg/issues/16232.
This change prevents REPLACE TABLE transactions from silently overwriting concurrent committed table changes. If table metadata changes after a replace transaction starts, the replace transaction now fails with CommitFailedException instead of committing stale metadata.
Problem
Before this change, replaceTransaction() could overwrite concurrent commits without conflict detection. This affected concurrent:
schema updates
table property updates
data appends
snapshot expiration
another replace transaction
The root problem was that replace transactions refreshed the underlying table state but still committed the stale replacement metadata built when the transaction started. In addition, replace-table update requirements skipped several optimistic concurrency checks.
This allowed committed changes to disappear silently.
Changes
Updated BaseTransaction.commitReplaceTransaction to fail if the table metadata changed since the replace transaction started.
Updated UpdateRequirements.forReplaceTable to use the same optimistic concurrency checks as ordinary table updates.
Added regression coverage in TestReplaceTableSafety.
Updated catalog tests to expect concurrent replace conflicts instead of last-writer-wins behavior.
Updated Hive replace transaction tests to preserve concurrent changes and expect conflict failures.
Updated Spark encryption replace test expectations for concurrent replace behavior.
Why this fixes the issue
A replace transaction now compares the original base metadata with the latest table metadata after refresh. If another writer committed in between, the replace transaction fails before committing stale metadata.
This prevents stale replace metadata from overwriting committed schema, property, snapshot, or data changes.
For REST-style commits, UpdateRequirements.forReplaceTable now emits normal optimistic concurrency requirements, so server-side validation can also reject conflicting replace commits.
Testing
Ran:
./gradlew :iceberg-core:test --tests "org.apache.iceberg.TestReplaceTableSafety" --tests "org.apache.iceberg.TestUpdateRequirements"
./gradlew :iceberg-hive-metastore:test --tests "org.apache.iceberg.hive.TestHiveCreateReplaceTable"
./gradlew :iceberg-hive-metastore:test
./gradlew :iceberg-spark:iceberg-spark-4.1_2.13:checkstyleTest
./gradlew :iceberg-azure:integrationTest :iceberg-aws:integrationTest --continue
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "rdblue (via GitHub)" <gi...@apache.org>.
rdblue commented on PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#issuecomment-4461369585
The contract of REPLACE TABLE is that the table's data, schema, and partitioning are replaced. There are no assumptions to validate, so checking for conflicts between concurrent changes and those changes is not required. If this is what you're trying to change, then I'm -1 for this commit because it is changing behavior that is already considered and chosen. It looks like this is the intent of this PR, but please reply if I'm wrong.
That said, REPLACE TABLE is also intended to keep table history (it is not a drop/re-create). So there are, potentially, valid problems with concurrency. For example, if a table's schema is updated from schema-id 1 to schema-id 2 and then the REPLACE TABLE operation removes the new schema-id 2 -- as though the concurrent commit never happened -- then that's a bug that we should fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "sshivampeta (via GitHub)" <gi...@apache.org>.
sshivampeta commented on PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#issuecomment-4613358963
@nssalian / @rdblue could you review the PR ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#issuecomment-5173219686
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "sshivampeta (via GitHub)" <gi...@apache.org>.
sshivampeta commented on PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#issuecomment-4504322499
> The contract of REPLACE TABLE is that the table's data, schema, and partitioning are replaced. There are no assumptions to validate, so checking for conflicts between concurrent changes and those changes is not required. If this is what you're trying to change, then I'm -1 for this commit because it is changing behavior that is already considered and chosen. It looks like this is the intent of this PR, but please reply if I'm wrong.
>
> That said, REPLACE TABLE is also intended to keep table history (it is not a drop/re-create). So there are, potentially, valid problems with concurrency. For example, if a table's schema is updated from schema-id 1 to schema-id 2 and then the REPLACE TABLE operation removes the new schema-id 2 -- as though the concurrent commit never happened -- then that's a bug that we should fix.
Thanks for the detailed feedback, @rdblue. You're right that REPLACE TABLE by contract replaces the table's data, schema, and partitioning — and that's intentional. This PR is not trying to change that contract.
The concern is specifically about table history preservation. As you noted, REPLACE TABLE should keep table history (it is not a drop/re-create). The bug is that commitReplaceTransaction refreshes base to detect version conflicts but then commits the original stale current metadata wholesale — silently erasing concurrently-committed schemas, expired snapshots, and other history changes. For example:
A concurrent expireSnapshots successfully removes snapshot-1, but the replace brings it back because its stale metadata still references it — this is table corruption (the snapshot files may have been physically deleted).
A concurrent updateSchema adds schema-id 2, but the replace drops it — as you described, "as though the concurrent commit never happened."
The current approach (fail-fast on any concurrent metadata change) is the simplest safe starting point to prevent this class of bugs. It's conservative — it will reject even non-conflicting concurrent changes — but it ensures no committed work is silently lost. A more refined approach that allows non-conflicting changes to survive (e.g., rebasing replace metadata on top of the refreshed base) could follow as a separate PR once the immediate safety issue is addressed.
Would this framing address your concern? Happy to discuss further or adjust the approach.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #16289: Core: Fix the concurrent REPLACE TABLE transaction commits
URL: https://github.com/apache/iceberg/pull/16289
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "sshivampeta (via GitHub)" <gi...@apache.org>.
sshivampeta commented on code in PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#discussion_r3278348846
##########
core/src/main/java/org/apache/iceberg/BaseTransaction.java:
##########
@@ -320,10 +320,13 @@ private void commitReplaceTransaction(boolean orCreate) {
}
}
- // because this is a replace table, it will always completely replace the table
- // metadata. even if it was just updated.
- if (base != underlyingOps.current()) {
- this.base = underlyingOps.current(); // just refreshed
+ // Replace transactions must not silently overwrite concurrent commits. If the table
+ // metadata has changed since the transaction started, fail instead of rebasing and
+ // merging staged updates.
+ if (base != null && underlyingOps.current() != base) {
Review Comment:
Good catch. Since onlyRetryOn(CommitFailedException.class) retries on CommitFailedException, throwing it directly inside the lambda would cause the retry loop to keep hitting the same concurrent-modification check (since base is never updated) until retries are exhausted — wasteful.
Fixed: now wrapping the CommitFailedException in PendingUpdateFailedException (the same pattern used by applyUpdates in commitSimpleTransaction) to immediately break the retry loop. Added a matching catch (PendingUpdateFailedException e) block in commitReplaceTransaction that cleans up and re-throws the unwrapped CommitFailedException
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "nssalian (via GitHub)" <gi...@apache.org>.
nssalian commented on code in PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#discussion_r3238987445
##########
core/src/main/java/org/apache/iceberg/BaseTransaction.java:
##########
@@ -320,10 +320,13 @@ private void commitReplaceTransaction(boolean orCreate) {
}
}
- // because this is a replace table, it will always completely replace the table
- // metadata. even if it was just updated.
- if (base != underlyingOps.current()) {
- this.base = underlyingOps.current(); // just refreshed
+ // Replace transactions must not silently overwrite concurrent commits. If the table
+ // metadata has changed since the transaction started, fail instead of rebasing and
+ // merging staged updates.
+ if (base != null && underlyingOps.current() != base) {
Review Comment:
Since the base assignment was removed, this would cause the base to not be updated and keep retrying. `applyUpdates` has a better approach here with `PendingUpdateFailedException` to break the retry loop.
##########
core/src/main/java/org/apache/iceberg/UpdateRequirements.java:
##########
@@ -41,7 +41,9 @@ public static List<UpdateRequirement> forReplaceTable(
TableMetadata base, List<MetadataUpdate> metadataUpdates) {
Preconditions.checkArgument(null != base, "Invalid table metadata: null");
Preconditions.checkArgument(null != metadataUpdates, "Invalid metadata updates: null");
- Builder builder = new Builder(base, true);
+ // use the same optimistic concurrency checks as ordinary commits; replace commits must not
+ // silently drop concurrent schema, partition spec, sort order, or ref changes.
+ Builder builder = new Builder(base, false);
Review Comment:
Can you add a test for this change?
##########
core/src/test/java/org/apache/iceberg/TestReplaceTableSafety.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.nio.file.Path;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.types.Types.NestedField;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Regression tests for GH-16232.
+ *
+ * <p>Replace transactions must fail on concurrent table metadata changes rather than silently
+ * overwriting committed updates.
+ */
+public class TestReplaceTableSafety {
+
+ private static final Namespace NS = Namespace.of("db");
+ private static final TableIdentifier TABLE = TableIdentifier.of(NS, "tbl");
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()), required(2, "data", Types.StringType.get()));
+
+ private static final Schema SCHEMA_WITH_EXTRA_COL =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ required(2, "data", Types.StringType.get()),
+ NestedField.optional(3, "extra", Types.StringType.get()));
+
+ private static final DataFile FILE_A =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/path/to/data-a.parquet")
+ .withFileSizeInBytes(10)
+ .withRecordCount(2)
+ .build();
+
+ private static final DataFile FILE_B =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/path/to/data-b.parquet")
+ .withFileSizeInBytes(10)
+ .withRecordCount(2)
+ .build();
+
+ private InMemoryCatalog catalog;
+
+ @BeforeEach
+ public void before(@TempDir Path temp) {
+ catalog = new InMemoryCatalog();
+ catalog.initialize(
+ "in-memory",
+ ImmutableMap.of(
+ CatalogProperties.WAREHOUSE_LOCATION,
+ temp.toAbsolutePath().toString(),
+ CatalogProperties.FILE_IO_IMPL,
+ "org.apache.iceberg.inmemory.InMemoryFileIO"));
+ catalog.createNamespace(NS);
+ }
+
+ @Test
+ public void replaceVsSchemaUpdateFailsAndPreservesSchema() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent schema update
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.schema().asStruct()).isEqualTo(SCHEMA_WITH_EXTRA_COL.asStruct());
+ }
+
+ @Test
+ public void replaceVsPropertyWriteFailsAndPreservesProperty() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent property update
+ table.updateProperties().set("k1", "v1").commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.properties()).containsEntry("k1", "v1");
+ }
+
+ @Test
+ public void replaceVsAppendFailsAndPreservesCommittedData() {
+ catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent append
+ catalog.loadTable(TABLE).newFastAppend().appendFile(FILE_B).commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ // concurrent data remains reachable
+ assertThat(after.currentSnapshot()).isNotNull();
+ assertThat(after.newScan().planFiles()).hasSize(1);
+ }
+
+ @Test
+ public void replaceVsReplaceFailsSecondCommitAndPreservesFirst() {
+ catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction secondReplace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ secondReplace.newFastAppend().appendFile(FILE_A).commit();
+
+ Transaction firstReplace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ firstReplace.newFastAppend().appendFile(FILE_B).commit();
+ firstReplace.commitTransaction();
+
+ assertThatThrownBy(secondReplace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+ }
+
+ @Test
+ public void replaceVsExpireSnapshotsFailsAndDoesNotResurrectExpiredSnapshot() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_B).commit();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_B).commit();
+
+ // concurrent expire
+ table.expireSnapshots().expireSnapshotId(snapshotId).commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.snapshot(snapshotId)).isNull();
+ }
+
+ @Test
+ public void replaceVsMultipleConcurrentChangesFailsAndPreservesAll() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent changes
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ table.updateProperties().set("k1", "v1").commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.schema().asStruct()).isEqualTo(SCHEMA_WITH_EXTRA_COL.asStruct());
+ assertThat(after.properties()).containsEntry("k1", "v1");
+ }
+}
Review Comment:
Few more tests would help.
1. For the `createOrReplaceTransaction (orCreate=true)`
2. v3 row-lineage tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "sshivampeta (via GitHub)" <gi...@apache.org>.
sshivampeta commented on code in PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#discussion_r3278352010
##########
core/src/test/java/org/apache/iceberg/TestReplaceTableSafety.java:
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.nio.file.Path;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.inmemory.InMemoryCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.types.Types.NestedField;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Regression tests for GH-16232.
+ *
+ * <p>Replace transactions must fail on concurrent table metadata changes rather than silently
+ * overwriting committed updates.
+ */
+public class TestReplaceTableSafety {
+
+ private static final Namespace NS = Namespace.of("db");
+ private static final TableIdentifier TABLE = TableIdentifier.of(NS, "tbl");
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()), required(2, "data", Types.StringType.get()));
+
+ private static final Schema SCHEMA_WITH_EXTRA_COL =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ required(2, "data", Types.StringType.get()),
+ NestedField.optional(3, "extra", Types.StringType.get()));
+
+ private static final DataFile FILE_A =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/path/to/data-a.parquet")
+ .withFileSizeInBytes(10)
+ .withRecordCount(2)
+ .build();
+
+ private static final DataFile FILE_B =
+ DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("/path/to/data-b.parquet")
+ .withFileSizeInBytes(10)
+ .withRecordCount(2)
+ .build();
+
+ private InMemoryCatalog catalog;
+
+ @BeforeEach
+ public void before(@TempDir Path temp) {
+ catalog = new InMemoryCatalog();
+ catalog.initialize(
+ "in-memory",
+ ImmutableMap.of(
+ CatalogProperties.WAREHOUSE_LOCATION,
+ temp.toAbsolutePath().toString(),
+ CatalogProperties.FILE_IO_IMPL,
+ "org.apache.iceberg.inmemory.InMemoryFileIO"));
+ catalog.createNamespace(NS);
+ }
+
+ @Test
+ public void replaceVsSchemaUpdateFailsAndPreservesSchema() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent schema update
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.schema().asStruct()).isEqualTo(SCHEMA_WITH_EXTRA_COL.asStruct());
+ }
+
+ @Test
+ public void replaceVsPropertyWriteFailsAndPreservesProperty() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent property update
+ table.updateProperties().set("k1", "v1").commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.properties()).containsEntry("k1", "v1");
+ }
+
+ @Test
+ public void replaceVsAppendFailsAndPreservesCommittedData() {
+ catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent append
+ catalog.loadTable(TABLE).newFastAppend().appendFile(FILE_B).commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ // concurrent data remains reachable
+ assertThat(after.currentSnapshot()).isNotNull();
+ assertThat(after.newScan().planFiles()).hasSize(1);
+ }
+
+ @Test
+ public void replaceVsReplaceFailsSecondCommitAndPreservesFirst() {
+ catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction secondReplace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ secondReplace.newFastAppend().appendFile(FILE_A).commit();
+
+ Transaction firstReplace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ firstReplace.newFastAppend().appendFile(FILE_B).commit();
+ firstReplace.commitTransaction();
+
+ assertThatThrownBy(secondReplace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+ }
+
+ @Test
+ public void replaceVsExpireSnapshotsFailsAndDoesNotResurrectExpiredSnapshot() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+ table.newFastAppend().appendFile(FILE_A).commit();
+ long snapshotId = table.currentSnapshot().snapshotId();
+ table.newFastAppend().appendFile(FILE_B).commit();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_B).commit();
+
+ // concurrent expire
+ table.expireSnapshots().expireSnapshotId(snapshotId).commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.snapshot(snapshotId)).isNull();
+ }
+
+ @Test
+ public void replaceVsMultipleConcurrentChangesFailsAndPreservesAll() {
+ Table table = catalog.buildTable(TABLE, SCHEMA).create();
+
+ Transaction replace = catalog.buildTable(TABLE, SCHEMA).replaceTransaction();
+ replace.newFastAppend().appendFile(FILE_A).commit();
+
+ // concurrent changes
+ table.updateSchema().addColumn("extra", Types.StringType.get()).commit();
+ table.updateProperties().set("k1", "v1").commit();
+
+ assertThatThrownBy(replace::commitTransaction)
+ .isInstanceOf(org.apache.iceberg.exceptions.CommitFailedException.class)
+ .hasMessageContaining("replace transaction");
+
+ Table after = catalog.loadTable(TABLE);
+ assertThat(after.schema().asStruct()).isEqualTo(SCHEMA_WITH_EXTRA_COL.asStruct());
+ assertThat(after.properties()).containsEntry("k1", "v1");
+ }
+}
Review Comment:
```
Added 4 additional tests:
createOrReplaceVsSchemaUpdateFailsAndPreservesSchema — createOrReplaceTransaction (orCreate=true) on an existing table detects concurrent schema update and fails, preserving the concurrently-added column.
createOrReplaceNewTableSucceeds — createOrReplaceTransaction for a table that doesn't exist yet succeeds normally (happy-path / no conflict).
createOrReplaceVsPropertyWriteFailsAndPreservesProperty — createOrReplaceTransaction on an existing table detects concurrent property write and fails.
replaceV3TableVsConcurrentAppendFails — creates a v3 format table, starts a replace transaction, makes a concurrent append (which would advance next-row-id), and verifies the replace fails instead of committing stale row lineage metadata.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "sshivampeta (via GitHub)" <gi...@apache.org>.
sshivampeta commented on code in PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#discussion_r3278350256
##########
core/src/main/java/org/apache/iceberg/UpdateRequirements.java:
##########
@@ -41,7 +41,9 @@ public static List<UpdateRequirement> forReplaceTable(
TableMetadata base, List<MetadataUpdate> metadataUpdates) {
Preconditions.checkArgument(null != base, "Invalid table metadata: null");
Preconditions.checkArgument(null != metadataUpdates, "Invalid metadata updates: null");
- Builder builder = new Builder(base, true);
+ // use the same optimistic concurrency checks as ordinary commits; replace commits must not
+ // silently drop concurrent schema, partition spec, sort order, or ref changes.
+ Builder builder = new Builder(base, false);
Review Comment:
```
Added 5 tests in TestUpdateRequirements:
replaceTableAddSchemaProducesFieldIdRequirement — verifies forReplaceTable now emits AssertLastAssignedFieldId
replaceTableAddSchemaFailsOnConcurrentFieldIdChange — verifies field-id conflict detection fails as expected
replaceTableSetCurrentSchemaProducesSchemaIdRequirement — verifies AssertCurrentSchemaID is emitted
replaceTableAddPartitionSpecProducesPartitionIdRequirement — verifies AssertLastAssignedPartitionId is emitted
replaceTableAddSortOrderProducesDefaultSortOrderRequirement — verifies AssertDefaultSortOrderID is emitted
These confirm that forReplaceTable with Builder(base, false) now produces the same OCC requirements as forUpdateTable, rather than skipping them as it did with Builder(base, true).
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix the concurrent REPLACE TABLE transaction commits [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16289:
URL: https://github.com/apache/iceberg/pull/16289#issuecomment-5098428805
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org