Core: REPLACE TABLE transaction silently overwrites concurrent committed changes

### Apache Iceberg version

main (also affects 1.5.x, 1.6.x, 1.7.x)

### Query engine

Any (tested with Java SDK directly via InMemoryCatalog)

### Please describe the bug 🐞

## Summary

`CREATE OR REPLACE TABLE` could be implemented either as dropping existing table followed by creating a new table or by overwriting existing table (via `replaceTransaction()`). The latter implementation silently overwrites concurrent committed changes to the same table. Any modification that commits between the start and end of a replace transaction — schema updates, property writes, DML appends, snapshot expiration — is lost without error or conflict detection.

This is specific to REPLACE transactions. Regular operations (`updateProperties`, `updateSchema`, `expireSnapshots`, DML) are immune because they use OCC with retry and additive re-apply on refreshed metadata.

## Root Cause

Two issues in `BaseTransaction.commitReplaceTransaction`:

1. **Stale metadata committed**: The method refreshes `base` (to detect version conflicts) but commits the original `current` metadata snapshot built at transaction creation time. It never calls `applyUpdates()` to incorporate changes that happened since the transaction started. In contrast, `commitSimpleTransaction` does call `applyUpdates()`, which re-applies pending updates on the refreshed base.

2. **Conflict detection disabled**: `UpdateRequirements.forReplaceTable` sets `isReplace=true`, which skips 5 categories of conflict assertions (schema, partition spec, sort order, snapshot refs, default branch). This means the server-side OCC check cannot detect that concurrent changes were overwritten.

The combination means: a REPLACE transaction will always succeed (no `CommitFailedException`) regardless of what happened to the table concurrently, and it will silently overwrite all concurrent changes.

## Scenarios That Break

I wrote a self-contained JUnit 5 test class (below) that demonstrates 6 distinct failure scenarios. All tests **pass** on current `main`, proving the bugs exist. Each test:
1. Creates a table with known initial state
2. Starts a `replaceTransaction()` (captures base metadata)
3. Performs a concurrent modification that commits successfully
4. Commits the replace transaction
5. Asserts the bug: the concurrent committed change was silently lost

### 1. Replace vs Snapshot Expiration — expired snapshot comes back

A snapshot that was successfully expired reappears after the replace commits. This is table corruption: the metadata now references snapshot files that may have been physically deleted by the expiration process.

### 2. Replace vs Schema Update — column addition silently lost

A concurrently-added column disappears after the replace commits. The schema reverts to what it was when the replace transaction started.

### 3. Replace vs Property Write — property silently clobbered

A concurrently-set table property vanishes after the replace commits. The properties map reverts to the replace transaction's snapshot.

### 4. Replace vs DML Append — committed data snapshot lost

A concurrent data append creates a new snapshot that disappears after the replace commits. This is **silent data loss** — the data was committed successfully but is no longer reachable.

### 5. Replace vs Replace — no conflict detection between two replaces

Two concurrent replace transactions both succeed without `CommitFailedException`. The second one silently overwrites the first with no awareness of the conflict. For regular operations (e.g., two concurrent `updateProperties` setting different keys), OCC detects the conflict and the second writer retries additively, preserving both changes.

### 6. Compound — multiple concurrent changes all lost simultaneously

Three concurrent modifications (expire snapshot + set property + add column) are all committed successfully, then all silently overwritten when the replace commits. Demonstrates that the bug affects all metadata dimensions simultaneously.

## Additional Context for V3 Tables

On V3 tables with row lineage, this bug is even more severe. `snapshot.first-row-id` is set from `base.nextRowId()` at transaction creation time. If a concurrent commit advances the table's `next-row-id`, the replace's snapshot will have a stale `first-row-id`, causing:

```
CommitFailedException: Cannot add a snapshot, first-row-id is behind table next-row-id: X < Y
```

Since `commitReplaceTransaction` retries with the same stale `current` (never re-applying updates), this failure is **permanent** — the replace can never succeed after any concurrent snapshot change on a V3 table.

## Reproducing Test

TestReplaceTableSafety.java (self-contained, no external dependencies)

```java
/*
* 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 java.nio.file.Path;
import java.util.Set;
import java.util.stream.Collectors;
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.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;

/**
* Tests demonstrating safety bugs in Iceberg's REPLACE TABLE transaction.
*
*

Each test PASSES on current Iceberg code, proving the bug exists. The bugs are specific to
* REPLACE and cannot occur with regular {@code updateTable} operations, which use OCC with retry:
* on conflict, base is refreshed and updates are re-applied additively on the latest metadata.
*
*

REPLACE bypasses this: {@code commitReplaceTransaction} commits a stale {@code current}
* metadata snapshot built at transaction creation time, silently overwriting any concurrent
* committed changes.
*
*

Root causes:
*
*

*
• {@code BaseTransaction.commitReplaceTransaction} refreshes {@code base} but never calls
  * {@code applyUpdates()}, so the stale {@code current} metadata from transaction creation
  * time is committed wholesale.
  *
• {@code UpdateRequirements.forReplaceTable} uses {@code isReplace=true}, skipping conflict
  * assertions (schema, spec, sort order, snapshot refs, branches).
  *

*
* @see #13651 - Snapshot Revival
* @see #12738 - Stale Partition Spec
*/
public class TestReplaceTableSafety {

private static final Namespace NS = Namespace.of("db");
private static final TableIdentifier TABLE = TableIdentifier.of(NS, "t");

private static final Schema SCHEMA_ID_DATA =
new Schema(
required(1, "id", Types.IntegerType.get()),
required(2, "data", Types.StringType.get()));

private static final Schema SCHEMA_ID_DATA_EXTRA =
new Schema(
required(1, "id", Types.IntegerType.get()),
required(2, "data", Types.StringType.get()),
required(3, "extra", Types.StringType.get()));

@TempDir private Path tempDir;
private InMemoryCatalog catalog;

@BeforeEach
public void before() {
catalog = new InMemoryCatalog();
catalog.initialize("test", ImmutableMap.of("warehouse", tempDir.toString()));
catalog.createNamespace(NS);
}

private DataFile dataFile(PartitionSpec spec, String path) {
return DataFiles.builder(spec)
.withPath(path)
.withFileSizeInBytes(10)
.withRecordCount(1)
.build();
}

// ---------------------------------------------------------------------------
// Bug 1: Replace vs snapshot expiration -- expired snapshot revived
//
// commitReplaceTransaction commits stale metadata that still contains the
// expired snapshot. The expiration commit is silently overwritten.
// ---------------------------------------------------------------------------

@Test
public void testReplaceVsExpiration_SnapshotRevival() {
Table table = catalog.buildTable(TABLE, SCHEMA_ID_DATA).create();

table.newFastAppend().appendFile(dataFile(table.spec(), "/data/a.parquet")).commit();
long snapshot1 = table.currentSnapshot().snapshotId();

table.newFastAppend().appendFile(dataFile(table.spec(), "/data/b.parquet")).commit();

// Start replace transaction (captures metadata with both snapshots)
Transaction replace = catalog.buildTable(TABLE, SCHEMA_ID_DATA).replaceTransaction();

// Concurrently expire snapshot 1
table = catalog.loadTable(TABLE);
table.expireSnapshots().expireSnapshotId(snapshot1).commit();

// Verify snapshot 1 is gone after expiration
table = catalog.loadTable(TABLE);
assertThat(collectSnapshotIds(table))
.as("Snapshot 1 should be gone after expiration")
.doesNotContain(snapshot1);

// Commit the replace
replace.commitTransaction();

// BUG: snapshot 1 is back -- revived after expiration
table = catalog.loadTable(TABLE);
assertThat(collectSnapshotIds(table))
.as("Expired snapshot should be revived by replace (BUG: stale snapshot revival)")
.contains(snapshot1);
}

// ---------------------------------------------------------------------------
// Bug 2: Replace vs concurrent schema update -- change silently lost
//
// The concurrent addColumn commit is silently overwritten because REPLACE
// commits the stale schema from transaction creation time.
// ---------------------------------------------------------------------------

@Test
public void testReplaceVsSchemaUpdate_SchemaChangeLost() {
catalog.buildTable(TABLE, SCHEMA_ID_DATA).create();

// Start replace (captures metadata with original schema)
Transaction replace = catalog.buildTable(TABLE, SCHEMA_ID_DATA).replaceTransaction();

// Concurrently add a column
Table table = catalog.loadTable(TABLE);
table.updateSchema().addColumn("replacement", Types.StringType.get()).commit();

// Commit the replace -- overwrites the schema update
replace.commitTransaction();

// BUG: the concurrent schema evolution is silently overwritten
Table result = catalog.loadTable(TABLE);
assertThat(result.schema().findField("replacement"))
.as("Column added by concurrent schema update should be gone (BUG: silent overwrite)")
.isNull();
}

// ---------------------------------------------------------------------------
// Bug 3: Replace vs concurrent property write -- property clobbered
//
// This bug is REPLACE-specific. With regular updateProperties, two concurrent
// writers setting DIFFERENT keys both survive via OCC retry + additive re-apply.
// With REPLACE, the entire stale properties map is committed wholesale.
// ---------------------------------------------------------------------------

@Test
public void testReplaceVsPropertyWrite_PropertyClobbered() {
catalog.buildTable(TABLE, SCHEMA_ID_DATA).withProperty("key1", "original").create();

// Start replace with new properties
Transaction replace =
catalog
.buildTable(TABLE, SCHEMA_ID_DATA)
.withProperty("key1", "replaced")
.replaceTransaction();

// Concurrently set a different property
Table table = catalog.loadTable(TABLE);
table.updateProperties().set("key2", "concurrent_value").commit();

// Verify concurrent property exists before replace commit
table = catalog.loadTable(TABLE);
assertThat(table.properties()).containsEntry("key2", "concurrent_value");

// Commit the replace
replace.commitTransaction();

// BUG: the concurrent property write is silently lost
Table result = catalog.loadTable(TABLE);
assertThat(result.properties())
.as("Concurrent property 'key2' should be gone (BUG: property clobbered by replace)")
.doesNotContainKey("key2");
assertThat(result.properties()).containsEntry("key1", "replaced");
}

// ---------------------------------------------------------------------------
// Bug 4: Replace vs concurrent DML -- snapshot lost
//
// A concurrent data append creates a new snapshot, but REPLACE commits
// stale metadata from before the append, so the DML snapshot vanishes.
// This is silent data loss.
// ---------------------------------------------------------------------------

@Test
public void testReplaceVsDataCommit_SnapshotLost() {
Table table = catalog.buildTable(TABLE, SCHEMA_ID_DATA).create();

table.newFastAppend().appendFile(dataFile(table.spec(), "/data/a.parquet")).commit();

// Start replace (captures metadata with 1 snapshot)
Transaction replace = catalog.buildTable(TABLE, SCHEMA_ID_DATA).replaceTransaction();

// Concurrently append data (creates snapshot 2)
table = catalog.loadTable(TABLE);
table.newFastAppend().appendFile(dataFile(table.spec(), "/data/b.parquet")).commit();
table = catalog.loadTable(TABLE);
long dmlSnapshotId = table.currentSnapshot().snapshotId();

// Commit the replace
replace.commitTransaction();

// BUG: the DML snapshot is lost -- overwritten by stale replace metadata
Table result = catalog.loadTable(TABLE);
assertThat(collectSnapshotIds(result))
.as("DML snapshot should be lost (BUG: replace overwrites concurrent DML)")
.doesNotContain(dmlSnapshotId);
}

// ---------------------------------------------------------------------------
// Bug 5: Replace vs concurrent replace -- no conflict detection
//
// Two concurrent REPLACE transactions both succeed without
// CommitFailedException. The second silently overwrites the first.
// For regular operations, OCC detects the conflict and triggers retry.
// ---------------------------------------------------------------------------

@Test
public void testReplaceVsReplace_LastWriterWinsNoConflict() {
catalog.buildTable(TABLE, SCHEMA_ID_DATA).withProperty("key1", "v1").create();

// Start two concurrent replace transactions
Transaction replaceA =
catalog
.buildTable(TABLE, SCHEMA_ID_DATA)
.withProperty("key1", "from_A")
.replaceTransaction();

Transaction replaceB =
catalog
.buildTable(TABLE, SCHEMA_ID_DATA)
.withProperty("key1", "from_B")
.replaceTransaction();

// Commit A first
replaceA.commitTransaction();
Table afterA = catalog.loadTable(TABLE);
assertThat(afterA.properties()).containsEntry("key1", "from_A");

// Commit B -- succeeds without CommitFailedException (no conflict detection)
replaceB.commitTransaction();

// BUG: B silently overwrites A with no conflict detection
Table result = catalog.loadTable(TABLE);
assertThat(result.properties())
.as("Replace B should silently overwrite A (BUG: no conflict detection between replaces)")
.containsEntry("key1", "from_B");
}

// ---------------------------------------------------------------------------
// Bug 6: Compound -- replace vs expiration + property + schema update
//
// Three concurrent committed changes are all silently lost when REPLACE
// commits stale metadata. Demonstrates the bug affects all metadata
// dimensions simultaneously.
// ---------------------------------------------------------------------------

@Test
public void testReplaceCompound_MultipleViolations() {
PartitionSpec specOnData =
PartitionSpec.builderFor(SCHEMA_ID_DATA_EXTRA).identity("data").build();
Table table =
catalog
.buildTable(TABLE, SCHEMA_ID_DATA_EXTRA)
.withPartitionSpec(specOnData)
.withProperty("marker", "original")
.create();

table.newFastAppend().appendFile(dataFile(table.spec(), "/data/a.parquet")).commit();
long snapshot1 = table.currentSnapshot().snapshotId();
table.newFastAppend().appendFile(dataFile(table.spec(), "/data/b.parquet")).commit();

// Start replace with new spec and property
PartitionSpec specOnExtra =
PartitionSpec.builderFor(SCHEMA_ID_DATA_EXTRA).identity("extra").build();
Transaction replace =
catalog
.buildTable(TABLE, SCHEMA_ID_DATA_EXTRA)
.withPartitionSpec(specOnExtra)
.withProperty("marker", "from_replace")
.replaceTransaction();

// Concurrent modification 1: expire snapshot 1
table = catalog.loadTable(TABLE);
table.expireSnapshots().expireSnapshotId(snapshot1).commit();

// Concurrent modification 2: set property
table = catalog.loadTable(TABLE);
table.updateProperties().set("marker", "from_concurrent").commit();

// Concurrent modification 3: add a column
table = catalog.loadTable(TABLE);
table.updateSchema().addColumn("added_col", Types.StringType.get()).commit();

// Commit replace -- overwrites all three concurrent changes
replace.commitTransaction();

Table result = catalog.loadTable(TABLE);

// Violation 1: expired snapshot is back
assertThat(collectSnapshotIds(result))
.as("Expired snapshot should be revived (BUG)")
.contains(snapshot1);

// Violation 2: concurrent property change is lost
assertThat(result.properties().get("marker"))
.as("Concurrent property update should be lost (BUG)")
.isEqualTo("from_replace");

// Violation 3: concurrent schema evolution is lost
assertThat(result.schema().findField("added_col"))
.as("Concurrent column addition should be lost (BUG)")
.isNull();
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

private Set collectSnapshotIds(Table table) {
return org.apache.iceberg.relocated.com.google.common.collect.Streams.stream(table.snapshots())
.map(Snapshot::snapshotId)
.collect(Collectors.toSet());
}
}

```

### Running the test

```bash
# Copy into an Iceberg clone and run:
cp TestReplaceTableSafety.java /path/to/iceberg/core/src/test/java/org/apache/iceberg/
cd /path/to/iceberg
./gradlew :iceberg-core:test --tests "org.apache.iceberg.TestReplaceTableSafety"
```

All 6 tests should **PASS** (proving the bugs exist). When the fix lands, these tests will start failing.

## Related Issues and Discussions

- **Mailing list**: [DISCUSS: Replace table transaction in REST Catalog](https://lists.apache.org/thread/d4hzd4cfvopvckcfw50orqksjzymd4lm)
- **Mailing list**: [DISCUSS: Allow Commit Conflicts in REPLACE TABLE transactions](https://lists.apache.org/thread/f7zd1dvlf6l1gw6wmnh9g404trcrkg06)
- #13651 — Corrupted table history due to CREATE OR REPLACE TABLE & Snapshot Expiration at the same time
- #12738 — Unexpected behavior with CREATE OR REPLACE AS SELECT (stale partition spec)
- #11524 — Unable to roll back to previous version after CREATE OR REPLACE with REST Catalog
- https://github.com/apache/iceberg/pull/15904 — [WIP] Replace transactions rebase onto refreshed metadata (active fix)
- https://github.com/apache/iceberg/pull/15905 — Repro: Concurrent replace V3 failure
- https://github.com/apache/iceberg/pull/15914 — Repro: REST vs non-REST concurrent replace behaviour

## Why This Is REPLACE-Specific

Regular `updateTable` operations cannot silently overwrite concurrent changes because they have three layers of defense:

1. **OCC conflict detection**: `base != current()` catches any concurrent commit
2. **Retry with refresh**: on conflict, base is refreshed to the latest metadata
3. **Additive re-apply**: the update is re-applied on refreshed state, preserving concurrent changes to other fields

For example, two concurrent `updateProperties` calls setting *different* keys both survive: the second writer detects the conflict, refreshes, and re-applies its key additively. REPLACE bypasses all three layers.

### Willingness to contribute

- [X] I can contribute a fix for this bug independently
- [X] I would be willing to contribute a fix for this bug with guidance from the Iceberg community
- [ ] I cannot contribute a fix for this bug at this time