Copilot commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3498328006
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1297,131 @@ public void commitTransaction(CommitTransactionRequest
commitTransactionRequest)
// 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.
+ // Process each table's changes in order. Both staged-creates and regular
updates are
+ // processed within the transaction workspace — creates buffer into
pendingCreations,
+ // updates buffer into pendingUpdates.
List<TableMetadata> tableMetadataObjs = new ArrayList<>();
+ // Track staged-creates for deferred location validation after the real
metastore is restored.
+ List<Map.Entry<TableIdentifier, TableMetadata>> stagedCreateEntries = new
ArrayList<>();
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");
+ boolean isStagedCreate =
changes.stream().anyMatch(CatalogHandlerUtils::isCreate);
+
+ // Reject invalid groups: a table cannot have both staged-create and
regular
+ // update requests in the same transaction. If any request asserts
the table
+ // does not exist, ALL requests for that table must be
staged-creates.
+ if (isStagedCreate
+ && changes.stream().anyMatch(change ->
!CatalogHandlerUtils.isCreate(change))) {
+ throw new BadRequestException(
+ "Invalid transaction: table '%s' has both staged-create and"
+ + " regular update requests",
+ tableIdentifier);
}
- 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());
- }
+ if (isStagedCreate) {
+ // For staged-creates, the table does not yet exist in the
metastore.
+ // Use newTableOps directly to get TableOperations for creation.
+ // The workspace buffers the createEntityIfNotExists call.
+ TableOperations tableOps =
+ ((LocalIcebergCatalog)
baseCatalog).newTableOps(tableIdentifier);
+
+ // Build metadata from empty, applying all updates from the
staged-create.
+ // Apply update filters to enforce location transformations (e.g.,
preventing
+ // a user from staging a table to an unauthorized storage path).
+ // Validate requirements against null (table does not exist yet) —
this enforces
+ // AssertTableDoesNotExist and any other requirements the client
sends.
Review Comment:
This comment suggests staged-create requests may include arbitrary
additional `UpdateRequirement`s, but `CatalogHandlerUtils.isCreate(...)`
currently enforces that the *only* allowed create requirement is
`AssertTableDoesNotExist` (it throws if any others are present). Please align
the comment with the actual supported behavior (or relax the validation if
broader requirements are intended).
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1297,131 @@ public void commitTransaction(CommitTransactionRequest
commitTransactionRequest)
// 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.
+ // Process each table's changes in order. Both staged-creates and regular
updates are
+ // processed within the transaction workspace — creates buffer into
pendingCreations,
+ // updates buffer into pendingUpdates.
List<TableMetadata> tableMetadataObjs = new ArrayList<>();
+ // Track staged-creates for deferred location validation after the real
metastore is restored.
+ List<Map.Entry<TableIdentifier, TableMetadata>> stagedCreateEntries = new
ArrayList<>();
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");
+ boolean isStagedCreate =
changes.stream().anyMatch(CatalogHandlerUtils::isCreate);
+
+ // Reject invalid groups: a table cannot have both staged-create and
regular
+ // update requests in the same transaction. If any request asserts
the table
+ // does not exist, ALL requests for that table must be
staged-creates.
+ if (isStagedCreate
+ && changes.stream().anyMatch(change ->
!CatalogHandlerUtils.isCreate(change))) {
+ throw new BadRequestException(
+ "Invalid transaction: table '%s' has both staged-create and"
+ + " regular update requests",
+ tableIdentifier);
}
- 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());
- }
+ if (isStagedCreate) {
+ // For staged-creates, the table does not yet exist in the
metastore.
+ // Use newTableOps directly to get TableOperations for creation.
+ // The workspace buffers the createEntityIfNotExists call.
+ TableOperations tableOps =
+ ((LocalIcebergCatalog)
baseCatalog).newTableOps(tableIdentifier);
+
+ // Build metadata from empty, applying all updates from the
staged-create.
+ // Apply update filters to enforce location transformations (e.g.,
preventing
+ // a user from staging a table to an unauthorized storage path).
+ // Validate requirements against null (table does not exist yet) —
this enforces
+ // AssertTableDoesNotExist and any other requirements the client
sends.
+ TableMetadata.Builder metadataBuilder =
TableMetadata.buildFromEmpty();
+ for (UpdateTableRequest change : changes) {
+ UpdateTableRequest filteredChange = applyUpdateFilters(change);
+ filteredChange.requirements().forEach(req ->
req.validate((TableMetadata) null));
+ for (MetadataUpdate singleUpdate : filteredChange.updates()) {
+ singleUpdate.applyTo(metadataBuilder);
}
+ }
- // Apply updates to builder
- singleUpdate.applyTo(metadataBuilder);
+ // Commit with null base to create the table entity (buffered in
workspace)
+ tableOps.commit(null, metadataBuilder.build());
+ TableMetadata createdMetadata = tableOps.current();
+ tableMetadataObjs.add(createdMetadata);
+ stagedCreateEntries.add(Map.entry(tableIdentifier,
createdMetadata));
+
+ } else {
+ // Regular update: load existing table and apply changes.
+ // The workspace buffers the updateEntityPropertiesIfNotChanged
call.
+ Table table = baseCatalog.loadTable(tableIdentifier);
+ if (!(table instanceof BaseTable baseTable)) {
+ throw new IllegalStateException(
+ "Cannot wrap catalog that does not produce BaseTable");
}
- // Update currentMetadata to reflect this change for subsequent
requirement validation
- currentMetadata = metadataBuilder.build();
- }
+ 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).
+ // Apply update filters to enforce location transformations.
+ TableMetadata currentMetadata = baseMetadata;
+ for (UpdateTableRequest change : changes) {
+ UpdateTableRequest filteredChange = applyUpdateFilters(change);
+ final TableMetadata metadataForValidation = currentMetadata;
+ filteredChange.requirements().forEach(req ->
req.validate(metadataForValidation));
+
+ TableMetadata.Builder metadataBuilder =
TableMetadata.buildFrom(currentMetadata);
+ for (MetadataUpdate singleUpdate : filteredChange.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'",
+ filteredChange.identifier(), setLocation.location());
+ }
+ }
+ singleUpdate.applyTo(metadataBuilder);
+ }
+ currentMetadata = metadataBuilder.build();
+ }
- // Commit all accumulated changes for this table in a single atomic
operation
- if (!currentMetadata.changes().isEmpty()) {
- tableOps.commit(baseMetadata, currentMetadata);
- }
+ // Commit all accumulated changes for this table in a single
atomic operation
+ if (!currentMetadata.changes().isEmpty()) {
+ tableOps.commit(baseMetadata, currentMetadata);
+ }
- tableMetadataObjs.add(currentMetadata);
+ tableMetadataObjs.add(currentMetadata);
+ }
});
- // Commit the collected updates in a single atomic operation
+ // Restore the real metastore manager for deferred validation and atomic
commit.
+ ((LocalIcebergCatalog)
baseCatalog).setMetaStoreManager(metaStoreManager());
+
+ // Deferred validation: validate location overlap for pending creations
using the real
+ // metastore (which supports listEntities/hasOverlappingSiblings). During
the loop above,
+ // the workspace returned no-op results for these calls.
+ // Also check for intra-transaction overlaps: two creates in the same
batch must not
+ // claim the same location, since the DB check won't see uncommitted peers.
+ Set<String> claimedLocationsInTx = new HashSet<>();
+ for (Map.Entry<TableIdentifier, TableMetadata> entry :
stagedCreateEntries) {
+ String location = entry.getValue().location();
+ if (!claimedLocationsInTx.add(location)) {
+ throw new BadRequestException(
+ "Transaction contains multiple creates pointing to the same
location: %s", location);
+ }
Review Comment:
The intra-transaction overlap guard for staged creates only checks
`TableMetadata.location()`, but the real overlap validation
(`validateStagedTableCreate`) considers *all* data locations derived from table
metadata/properties (via `StorageUtil.getLocationsUsedByTable`). Two staged
creates in the same transaction could therefore still overlap on a derived
location without being caught (the DB overlap check also won’t see uncommitted
peers).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
