diff --git a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java index 940d7fa05ec6..e93353cac85b 100644 --- a/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java +++ b/core/src/main/java/org/apache/iceberg/BaseMetastoreCatalog.java @@ -21,6 +21,7 @@ import java.io.Closeable; import java.io.IOException; import java.util.Map; +import java.util.function.UnaryOperator; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.AlreadyExistsException; @@ -237,12 +238,19 @@ private Transaction newReplaceTableTransaction(boolean orCreate) { throw new NoSuchTableException("Table does not exist: %s", identifier); } - TableMetadata metadata; tableProperties.putAll(tableOverrideProperties()); + UnaryOperator replacement = + base -> + base.buildReplacement( + schema, + spec, + sortOrder, + location != null ? location : base.location(), + tableProperties); + + TableMetadata metadata; if (ops.current() != null) { - String baseLocation = location != null ? location : ops.current().location(); - metadata = - ops.current().buildReplacement(schema, spec, sortOrder, baseLocation, tableProperties); + metadata = replacement.apply(ops.current()); } else { String baseLocation = location != null ? location : defaultWarehouseLocation(identifier); metadata = @@ -251,10 +259,10 @@ private Transaction newReplaceTableTransaction(boolean orCreate) { if (orCreate) { return Transactions.createOrReplaceTableTransaction( - identifier.toString(), ops, metadata, metricsReporter()); + identifier.toString(), ops, metadata, replacement, metricsReporter()); } else { return Transactions.replaceTableTransaction( - identifier.toString(), ops, metadata, metricsReporter()); + identifier.toString(), ops, metadata, replacement, metricsReporter()); } } diff --git a/core/src/main/java/org/apache/iceberg/BaseTransaction.java b/core/src/main/java/org/apache/iceberg/BaseTransaction.java index 46eda9e0c92e..baaf42a55a82 100644 --- a/core/src/main/java/org/apache/iceberg/BaseTransaction.java +++ b/core/src/main/java/org/apache/iceberg/BaseTransaction.java @@ -33,6 +33,7 @@ import java.util.Set; import java.util.UUID; import java.util.function.Consumer; +import java.util.function.UnaryOperator; import java.util.stream.Collectors; import org.apache.iceberg.encryption.EncryptionManager; import org.apache.iceberg.exceptions.CleanableFailure; @@ -71,6 +72,9 @@ enum TransactionType { Sets.newHashSet(); // keep track of files deleted in the most recent commit private final Consumer enqueueDelete = deletedFiles::add; private final TransactionType type; + // for replace transactions, rebuilds the replacement metadata from the latest table metadata so a + // concurrent writer's snapshots are preserved when the commit is retried; null otherwise + private final UnaryOperator replacement; private TableMetadata base; private TableMetadata current; private boolean hasLastOpCommitted; @@ -78,7 +82,7 @@ enum TransactionType { BaseTransaction( String tableName, TableOperations ops, TransactionType type, TableMetadata start) { - this(tableName, ops, type, start, LoggingMetricsReporter.instance()); + this(tableName, ops, type, start, null, LoggingMetricsReporter.instance()); } BaseTransaction( @@ -87,6 +91,25 @@ enum TransactionType { TransactionType type, TableMetadata start, MetricsReporter reporter) { + this(tableName, ops, type, start, null, reporter); + } + + BaseTransaction( + String tableName, + TableOperations ops, + TransactionType type, + TableMetadata start, + UnaryOperator replacement) { + this(tableName, ops, type, start, replacement, LoggingMetricsReporter.instance()); + } + + BaseTransaction( + String tableName, + TableOperations ops, + TransactionType type, + TableMetadata start, + UnaryOperator replacement, + MetricsReporter reporter) { this.tableName = tableName; this.ops = ops; this.transactionTable = new TransactionTable(); @@ -95,6 +118,7 @@ enum TransactionType { this.updates = Lists.newArrayList(); this.base = ops.current(); this.type = type; + this.replacement = replacement; this.hasLastOpCommitted = true; this.reporter = reporter; } @@ -311,20 +335,7 @@ private void commitReplaceTransaction(boolean orCreate) { .onlyRetryOn(CommitFailedException.class) .run( underlyingOps -> { - try { - underlyingOps.refresh(); - } catch (NoSuchTableException e) { - if (!orCreate) { - throw e; - } - } - - // 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 - } - + refreshReplacement(underlyingOps, orCreate); underlyingOps.commit(base, current); }); @@ -347,6 +358,42 @@ private void commitReplaceTransaction(boolean orCreate) { } } + // refreshes the underlying table before a replace commit. if a concurrent writer changed the + // table, rebuilds the replacement metadata on top of the refreshed table so the concurrent + // writer's snapshots are preserved in history (the replacement still becomes the current state), + // then re-applies the pending updates to recreate this transaction's snapshot on top. catalogs + // that merge changes server-side (e.g. REST) pass no replacement builder and keep their existing + // behavior of completely replacing the table metadata. + private void refreshReplacement(TableOperations underlyingOps, boolean orCreate) { + try { + underlyingOps.refresh(); + } catch (NoSuchTableException e) { + if (!orCreate) { + throw e; + } + } + + if (base == underlyingOps.current()) { + return; + } + + this.base = underlyingOps.current(); // just refreshed + if (replacement == null || base == null) { + return; + } + + TableMetadata rebuilt = replacement.apply(base); + // only rebuild when the concurrent change preserves this replacement's schema and spec. + // otherwise rebuilding would reassign field ids and break data this transaction already wrote, + // so fall back to replacing the table outright (last writer wins). + if (rebuilt.schema().sameSchema(current.schema()) && rebuilt.spec().equals(current.spec())) { + this.current = rebuilt; + for (PendingUpdate update : updates) { + update.commit(); + } + } + } + private void commitSimpleTransaction() { // if there were no changes, don't try to commit if (base == current) { diff --git a/core/src/main/java/org/apache/iceberg/Transactions.java b/core/src/main/java/org/apache/iceberg/Transactions.java index a8ea40a6b90b..07071a9329fb 100644 --- a/core/src/main/java/org/apache/iceberg/Transactions.java +++ b/core/src/main/java/org/apache/iceberg/Transactions.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg; +import java.util.function.UnaryOperator; import org.apache.iceberg.BaseTransaction.TransactionType; import org.apache.iceberg.metrics.MetricsReporter; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; @@ -36,6 +37,32 @@ public static Transaction createOrReplaceTableTransaction( tableName, ops, TransactionType.CREATE_OR_REPLACE_TABLE, start, reporter); } + /** + * Start a create-or-replace transaction that rebuilds its replacement metadata if a concurrent + * writer changes the table before the transaction commits. + * + * @param replacement builds the replacement metadata from the latest table metadata; used to + * rebuild the transaction's metadata on commit retry so concurrent snapshots are preserved + */ + public static Transaction createOrReplaceTableTransaction( + String tableName, + TableOperations ops, + TableMetadata start, + UnaryOperator replacement) { + return new BaseTransaction( + tableName, ops, TransactionType.CREATE_OR_REPLACE_TABLE, start, replacement); + } + + public static Transaction createOrReplaceTableTransaction( + String tableName, + TableOperations ops, + TableMetadata start, + UnaryOperator replacement, + MetricsReporter reporter) { + return new BaseTransaction( + tableName, ops, TransactionType.CREATE_OR_REPLACE_TABLE, start, replacement, reporter); + } + public static Transaction replaceTableTransaction( String tableName, TableOperations ops, TableMetadata start) { return new BaseTransaction(tableName, ops, TransactionType.REPLACE_TABLE, start); @@ -46,6 +73,31 @@ public static Transaction replaceTableTransaction( return new BaseTransaction(tableName, ops, TransactionType.REPLACE_TABLE, start, reporter); } + /** + * Start a replace transaction that rebuilds its replacement metadata if a concurrent writer + * changes the table before the transaction commits. + * + * @param replacement builds the replacement metadata from the latest table metadata; used to + * rebuild the transaction's metadata on commit retry so concurrent snapshots are preserved + */ + public static Transaction replaceTableTransaction( + String tableName, + TableOperations ops, + TableMetadata start, + UnaryOperator replacement) { + return new BaseTransaction(tableName, ops, TransactionType.REPLACE_TABLE, start, replacement); + } + + public static Transaction replaceTableTransaction( + String tableName, + TableOperations ops, + TableMetadata start, + UnaryOperator replacement, + MetricsReporter reporter) { + return new BaseTransaction( + tableName, ops, TransactionType.REPLACE_TABLE, start, replacement, reporter); + } + public static Transaction createTableTransaction( String tableName, TableOperations ops, TableMetadata start) { Preconditions.checkArgument( diff --git a/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java b/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java index 8d980b7ba176..c8a63b6daf61 100644 --- a/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java +++ b/core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.util.Map; +import java.util.function.UnaryOperator; import org.apache.hadoop.conf.Configurable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; @@ -369,17 +370,19 @@ private Transaction newReplaceTableTransaction(boolean orCreate) { } Map properties = propertiesBuilder.build(); + UnaryOperator replacement = + base -> base.buildReplacement(schema, spec, sortOrder, location, properties); TableMetadata metadata; if (ops.current() != null) { - metadata = ops.current().buildReplacement(schema, spec, sortOrder, location, properties); + metadata = replacement.apply(ops.current()); } else { metadata = tableMetadata(schema, spec, sortOrder, properties, location); } if (orCreate) { - return Transactions.createOrReplaceTableTransaction(location, ops, metadata); + return Transactions.createOrReplaceTableTransaction(location, ops, metadata, replacement); } else { - return Transactions.replaceTableTransaction(location, ops, metadata); + return Transactions.replaceTableTransaction(location, ops, metadata, replacement); } } } diff --git a/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java b/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java index 79196c0a7517..97443f995fe5 100644 --- a/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java +++ b/core/src/test/java/org/apache/iceberg/TestReplaceTransaction.java @@ -315,6 +315,44 @@ public void testReplaceTransactionConflict() { assertThat(listManifestFiles()).containsExactlyElementsOf(manifests); } + @TestTemplate + public void testReplaceTransactionConcurrentCommitRetainsHistory() { + // use random snapshot ids (like real catalogs) so the replace's snapshot and the concurrent + // writer's snapshot do not collide on the sequential ids that TestTables assigns by default + table.updateProperties().set("random-snapshot-ids", "true").commit(); + + table.newAppend().appendFile(FILE_A).commit(); + + validateSnapshot(null, table.currentSnapshot(), FILE_A); + long firstSnapshotId = table.currentSnapshot().snapshotId(); + + // start a replace that will make FILE_B the new current data + Transaction replace = TestTables.beginReplace(tableDir, "test", table.schema(), table.spec()); + replace.newAppend().appendFile(FILE_B).commit(); + + // a concurrent writer commits FILE_C out-of-band, forcing the replace commit to retry + table.newAppend().appendFile(FILE_C).commit(); + long concurrentSnapshotId = table.currentSnapshot().snapshotId(); + + replace.commitTransaction(); + + table.refresh(); + + // the replace wins for the current state + assertThat(table.currentSnapshot()).isNotNull(); + validateSnapshot(null, table.currentSnapshot(), FILE_B); + + // regression for #16942: a concurrent writer's snapshot committed during the replace must + // remain in the table history rather than being silently dropped on commit retry + assertThat(table.snapshot(concurrentSnapshotId)) + .as("Concurrent writer's snapshot should be preserved in history") + .isNotNull(); + assertThat(table.snapshot(firstSnapshotId)) + .as("Original snapshot should be preserved in history") + .isNotNull(); + assertThat(table.snapshots()).hasSize(3); + } + @TestTemplate public void testReplaceToCreateAndAppend() throws IOException { // this table doesn't exist. diff --git a/core/src/test/java/org/apache/iceberg/TestTables.java b/core/src/test/java/org/apache/iceberg/TestTables.java index fdf730a22074..9e0339c206b6 100644 --- a/core/src/test/java/org/apache/iceberg/TestTables.java +++ b/core/src/test/java/org/apache/iceberg/TestTables.java @@ -22,6 +22,7 @@ import java.io.File; import java.util.Map; +import java.util.function.UnaryOperator; import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.CommitStateUnknownException; @@ -204,8 +205,10 @@ public static Transaction beginReplace( TableMetadata current = ops.current(); TableMetadata metadata; if (current != null) { - metadata = current.buildReplacement(schema, spec, sortOrder, current.location(), properties); - return Transactions.replaceTableTransaction(name, ops, metadata); + UnaryOperator replacement = + base -> base.buildReplacement(schema, spec, sortOrder, base.location(), properties); + metadata = replacement.apply(current); + return Transactions.replaceTableTransaction(name, ops, metadata, replacement); } else { metadata = newTableMetadata(schema, spec, sortOrder, temp.toURI().toString(), properties); return Transactions.createTableTransaction(name, ops, metadata); diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCreateReplaceTable.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCreateReplaceTable.java index c3f66c0286d0..646375e0a405 100644 --- a/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCreateReplaceTable.java +++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/TestHiveCreateReplaceTable.java @@ -235,11 +235,13 @@ public void testReplaceTableTxnTableModifiedConcurrently() { txn.updateProperties().set("prop", "value").commit(); txn.commitTransaction(); - // the replace should still succeed + // the replace should still succeed, and the property the concurrent writer committed during + // the replace transaction must be preserved on retry. this is the same rebuild that preserves + // concurrent snapshots and matches REST delta semantics - see #16942 table = catalog.loadTable(TABLE_IDENTIFIER); assertThat(table.properties()) - .as("Table props should be updated") - .doesNotContainKey("another-prop") + .as("Replace retry should preserve concurrent property updates") + .containsEntry("another-prop", "another-value") .containsEntry("prop", "value"); }