This is an automated email from the ASF dual-hosted git repository.
singhpk234 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new ae2efce58e Fix: Group transaction changes by table (#3360)
ae2efce58e is described below
commit ae2efce58ea1103c3a28d988ce67ee3f9062e9e4
Author: Prashant Singh <[email protected]>
AuthorDate: Fri Mar 6 11:10:59 2026 -0800
Fix: Group transaction changes by table (#3360)
Co-authored-by: Prashant Kumar Singh <[email protected]>
---
.../it/test/PolarisRestCatalogIntegrationBase.java | 95 ++++++++++++++++
.../catalog/iceberg/IcebergCatalogHandler.java | 123 ++++++++++++---------
2 files changed, 168 insertions(+), 50 deletions(-)
diff --git
a/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java
b/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java
index 765c22b5b3..ab209302ab 100644
---
a/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java
+++
b/integration-tests/src/main/java/org/apache/polaris/service/it/test/PolarisRestCatalogIntegrationBase.java
@@ -1189,6 +1189,101 @@ public abstract class PolarisRestCatalogIntegrationBase
extends CatalogTests<RES
assertThat(latestCommittedSchema.asStruct()).isEqualTo(originalSchema.asStruct());
}
+ @Test
+ public void testCoalescedConflictOnOneTableRollsBackEntireTransaction() {
+ Namespace namespace = Namespace.of("coalescingAtomicNs");
+ TableIdentifier goodId = TableIdentifier.of(namespace, "goodTable");
+ TableIdentifier conflictId = TableIdentifier.of(namespace,
"conflictTable");
+
+ if (requiresNamespaceCreate()) {
+ catalog().createNamespace(namespace);
+ }
+
+ catalog().createTable(goodId, SCHEMA);
+ catalog().createTable(conflictId, SCHEMA);
+
+ Schema originalGoodSchema = catalog().loadTable(goodId).schema();
+ Schema originalConflictSchema = catalog().loadTable(conflictId).schema();
+
+ // goodTable: a single non-conflicting schema change
+ Transaction goodTx = catalog().loadTable(goodId).newTransaction();
+ goodTx.updateSchema().addColumn("new_col", Types.LongType.get()).commit();
+
+ // conflictTable: two independent transactions that both rename the same
column,
+ // producing two TableCommits for the same table. The first rename
succeeds,
+ // but the second conflicts because the schema has already changed.
+ Table conflictTable = catalog().loadTable(conflictId);
+ Transaction conflictTx1 = conflictTable.newTransaction();
+ Transaction conflictTx2 = conflictTable.newTransaction();
+ conflictTx1.updateSchema().renameColumn("data", "renamed-col1").commit();
+ conflictTx2.updateSchema().renameColumn("data", "renamed-col2").commit();
+
+ TableCommit goodCommit =
+ TableCommit.create(
+ goodId,
+ ((BaseTransaction) goodTx).startMetadata(),
+ ((BaseTransaction) goodTx).currentMetadata());
+ TableCommit conflictCommit1 =
+ TableCommit.create(
+ conflictId,
+ ((BaseTransaction) conflictTx1).startMetadata(),
+ ((BaseTransaction) conflictTx1).currentMetadata());
+ TableCommit conflictCommit2 =
+ TableCommit.create(
+ conflictId,
+ ((BaseTransaction) conflictTx2).startMetadata(),
+ ((BaseTransaction) conflictTx2).currentMetadata());
+
+ // The coalescing logic groups conflictCommit1 and conflictCommit2
together.
+ // The first rename succeeds, but the second fails requirement validation
+ // (schema ID changed). This should fail the entire transaction atomically.
+ assertThatThrownBy(
+ () -> restCatalog.commitTransaction(goodCommit, conflictCommit1,
conflictCommit2))
+ .isInstanceOf(CommitFailedException.class);
+
+ // Verify atomicity: neither table should have changed.
+ assertThat(catalog().loadTable(goodId).schema().asStruct())
+ .isEqualTo(originalGoodSchema.asStruct());
+ assertThat(catalog().loadTable(conflictId).schema().asStruct())
+ .isEqualTo(originalConflictSchema.asStruct());
+ }
+
+ @Test
+ public void
testMultipleNonConflictingUpdatesToSameTableWithSchemaAndProperties() {
+ Namespace namespace = Namespace.of("coalescingMixedNs");
+ TableIdentifier identifier = TableIdentifier.of(namespace,
"coalescingMixedTable");
+
+ if (requiresNamespaceCreate()) {
+ catalog().createNamespace(namespace);
+ }
+
+ // Use a single Iceberg transaction that performs both a schema change and
a property update.
+ // This produces a single TableCommit containing multiple metadata
updates, verifying
+ // that the server correctly handles multiple update types within a single
commit request.
+ Table table = catalog().buildTable(identifier, SCHEMA).create();
+ Transaction transaction = table.newTransaction();
+
+ UpdateSchema updateSchema =
+ transaction.updateSchema().addColumn("new_col", Types.LongType.get());
+ Schema expectedSchema = updateSchema.apply();
+ updateSchema.commit();
+
+ transaction.updateProperties().set("prop-key", "prop-val").commit();
+
+ TableCommit tableCommit =
+ TableCommit.create(
+ identifier,
+ ((BaseTransaction) transaction).startMetadata(),
+ ((BaseTransaction) transaction).currentMetadata());
+
+ restCatalog.commitTransaction(tableCommit);
+
+ // Verify both schema and property updates were applied.
+ Table loaded = catalog().loadTable(identifier);
+
assertThat(loaded.schema().asStruct()).isEqualTo(expectedSchema.asStruct());
+ assertThat(loaded.properties()).containsEntry("prop-key", "prop-val");
+ }
+
@Test
public void testTableExistsStatus() {
String tableName = "tbl1";
diff --git
a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java
b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java
index 88bcb595b0..2d8664f204 100644
---
a/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java
+++
b/runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java
@@ -41,6 +41,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -1044,60 +1045,82 @@ public abstract class IcebergCatalogHandler extends
CatalogHandler implements Au
new TransactionWorkspaceMetaStoreManager(diagnostics(),
metaStoreManager());
((IcebergCatalog)
baseCatalog).setMetaStoreManager(transactionMetaStoreManager);
+ // Group all changes by table identifier to handle them atomically.
+ // This prevents conflicts when multiple changes target the same table
entity.
+ // LinkedHashMap preserves insertion order for deterministic processing.
+ Map<TableIdentifier, List<UpdateTableRequest>> changesByTable = new
LinkedHashMap<>();
+ for (UpdateTableRequest change : commitTransactionRequest.tableChanges()) {
+ if (CatalogHandlerUtils.isCreate(change)) {
+ throw new BadRequestException(
+ "Unsupported operation: commitTranaction with
updateForStagedCreate: %s", change);
+ }
+ changesByTable.computeIfAbsent(change.identifier(), k -> new
ArrayList<>()).add(change);
+ }
+
+ // Process each table's changes in order.
+ // Note: All UpdateTableRequests for a given table are coalesced into a
single metadata
+ // update and a single tableOps.commit(), which results in one Polaris
entity update per
+ // table. This is subtly different from applying each UpdateTableRequest
as an independent
+ // commit (as if each were under a lock). Requirements are still validated
sequentially
+ // against the evolving metadata, so conflicts are detected correctly.
+ // See also the TODO in TransactionWorkspaceMetaStoreManager for a more
general (but more
+ // complex) alternative that would intercept at the MetaStoreManager layer.
List<TableMetadata> tableMetadataObjs = new ArrayList<>();
- commitTransactionRequest.tableChanges().stream()
- .forEach(
- change -> {
- Table table = baseCatalog.loadTable(change.identifier());
- if (!(table instanceof BaseTable baseTable)) {
- throw new IllegalStateException(
- "Cannot wrap catalog that does not produce BaseTable");
- }
- if (CatalogHandlerUtils.isCreate(change)) {
- throw new BadRequestException(
- "Unsupported operation: commitTranaction with
updateForStagedCreate: %s",
- change);
+ changesByTable.forEach(
+ (tableIdentifier, changes) -> {
+ Table table = baseCatalog.loadTable(tableIdentifier);
+ if (!(table instanceof BaseTable baseTable)) {
+ throw new IllegalStateException("Cannot wrap catalog that does not
produce BaseTable");
+ }
+
+ TableOperations tableOps = baseTable.operations();
+ TableMetadata baseMetadata = tableOps.current();
+
+ // Apply each change sequentially: validate requirements against
current state,
+ // then apply updates. This ensures conflicts are detected (e.g., if
two changes
+ // both expect schema ID 0, the second will fail after the first
increments it).
+ TableMetadata currentMetadata = baseMetadata;
+ for (UpdateTableRequest change : changes) {
+ // Validate requirements against the current metadata state
+ final TableMetadata metadataForValidation = currentMetadata;
+ change
+ .requirements()
+ .forEach(requirement ->
requirement.validate(metadataForValidation));
+
+ // TODO: Refactor to share/reconcile the update-application logic
below with
+ // CatalogHandlerUtils to avoid divergence as complexity grows.
+ TableMetadata.Builder metadataBuilder =
TableMetadata.buildFrom(currentMetadata);
+ for (MetadataUpdate singleUpdate : change.updates()) {
+ // Note: If location-overlap checking is refactored to be
atomic, we could
+ // support validation within a single multi-table transaction as
well, but
+ // will need to update the TransactionWorkspaceMetaStoreManager
to better
+ // expose the concept of being able to read uncommitted updates.
+ if (singleUpdate instanceof MetadataUpdate.SetLocation
setLocation) {
+ if (!currentMetadata.location().equals(setLocation.location())
+ && !realmConfig()
+
.getConfig(FeatureConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
+ throw new BadRequestException(
+ "Unsupported operation: commitTransaction containing
SetLocation"
+ + " for table '%s' and new location '%s'",
+ change.identifier(), ((MetadataUpdate.SetLocation)
singleUpdate).location());
+ }
}
- TableOperations tableOps = baseTable.operations();
- TableMetadata currentMetadata = tableOps.current();
-
- // Validate requirements; any CommitFailedExceptions will fail
the overall request
- change.requirements().forEach(requirement ->
requirement.validate(currentMetadata));
-
- // Apply changes
- TableMetadata.Builder metadataBuilder =
TableMetadata.buildFrom(currentMetadata);
- change.updates().stream()
- .forEach(
- singleUpdate -> {
- // Note: If location-overlap checking is refactored to
be atomic, we could
- // support validation within a single multi-table
transaction as well, but
- // will need to update the
TransactionWorkspaceMetaStoreManager to better
- // expose the concept of being able to read
uncommitted updates.
- if (singleUpdate instanceof MetadataUpdate.SetLocation
setLocation) {
- if
(!currentMetadata.location().equals(setLocation.location())
- && !realmConfig()
- .getConfig(
-
FeatureConfiguration.ALLOW_NAMESPACE_LOCATION_OVERLAP)) {
- throw new BadRequestException(
- "Unsupported operation: commitTransaction
containing SetLocation"
- + " for table '%s' and new location '%s'",
- change.identifier(), setLocation.location());
- }
- }
-
- // Apply updates to builder
- singleUpdate.applyTo(metadataBuilder);
- });
-
- // Commit into transaction workspace we swapped the baseCatalog
to use
- TableMetadata updatedMetadata = metadataBuilder.build();
- if (!updatedMetadata.changes().isEmpty()) {
- tableOps.commit(currentMetadata, updatedMetadata);
- }
+ // Apply updates to builder
+ singleUpdate.applyTo(metadataBuilder);
+ }
+
+ // Update currentMetadata to reflect this change for subsequent
requirement validation
+ currentMetadata = metadataBuilder.build();
+ }
+
+ // Commit all accumulated changes for this table in a single atomic
operation
+ if (!currentMetadata.changes().isEmpty()) {
+ tableOps.commit(baseMetadata, currentMetadata);
+ }
- tableMetadataObjs.add(updatedMetadata);
- });
+ tableMetadataObjs.add(currentMetadata);
+ });
// Commit the collected updates in a single atomic operation
List<EntityWithPath> pendingUpdates =
transactionMetaStoreManager.getPendingUpdates();
