Core: Fix the concurrent REPLACE TABLE transaction commits

[sshivampeta]

Summary

Fixes #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

[nssalian]

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.

[sshivampeta]

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

[nssalian]

Can you add a test for this change?

[sshivampeta]

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).

[nssalian]

Few more tests would help.

1. For the createOrReplaceTransaction (orCreate=true)
2. v3 row-lineage tests.

[sshivampeta]

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.

[rdblue]

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.

[sshivampeta]

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.