You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@polaris.apache.org by "ayushtkn (via GitHub)" <gi...@apache.org> on 2026/06/30 11:20:02 UTC
[PR] Support staged table creates in commitTransaction [polaris]
ayushtkn opened a new pull request, #4939:
URL: https://github.com/apache/polaris/pull/4939
Implement staged creates in the transaction workspace. Polaris previously rejected these and authorized
the whole batch with `TABLE_WRITE_PROPERTIES` + `TABLE_CREATE` on every
table identifier.
## Checklist
- [ ] π‘οΈ Don't disclose security issues! (contact security@apache.org)
- [ ] π Clearly explained why the changes are needed, or linked related issues: Fixes #
- [ ] π§ͺ Added/updated tests with good coverage, or manually tested (and explained how)
- [ ] π‘ Added comments for complex logic
- [ ] π§Ύ Updated `CHANGELOG.md` (if needed)
- [ ] π Updated documentation in `site/content/in-dev/unreleased` (if needed)
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "flyingImer (via GitHub)" <gi...@apache.org>.
flyingImer commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3867222787
##########
polaris-core/src/main/java/org/apache/polaris/core/storage/StorageUtil.java:
##########
@@ -83,6 +87,54 @@ public class StorageUtil {
return Set.of(viewMetadata.location());
}
+ /**
+ * Returns true if the set of locations used by {@code current} differs from the set used by
+ * {@code base}. Covers the table location as well as write.data.path and write.metadata.path
+ * property overrides.
+ */
+ public static boolean locationsChanged(TableMetadata base, TableMetadata current) {
+ return !getLocationsUsedByTable(base).equals(getLocationsUsedByTable(current));
+ }
+
+ /**
+ * Checks that no two distinct tables in {@code batch} use overlapping locations (pairwise
+ * containment check across all data locations returned by {@link #getLocationsUsedByTable}).
+ * Same-identifier pairs are skipped so a table's own write.data.path and write.metadata.path
+ * don't count as self-overlap.
+ *
+ * @throws BadRequestException if any two distinct tables have overlapping locations
+ */
+ public static void validateNoOverlapWithinBatch(
+ List<Map.Entry<TableIdentifier, TableMetadata>> batch) {
+ // Collect one (location, identifier) pair per data location across all batch entries, then
+ // sort alphabetically by location string. Overlapping paths will be alphabetically adjacent
+ // after sorting, so a single linear scan catches all conflicts in O(n log n) rather than
+ // the O(n^2) pairwise scan.
+ List<Map.Entry<TableIdentifier, StorageLocation>> entries =
+ batch.stream()
+ .flatMap(
+ e ->
+ getLocationsUsedByTable(e.getValue()).stream()
+ .map(loc -> Map.entry(e.getKey(), StorageLocation.of(loc))))
+ .sorted(Comparator.comparing(e -> e.getValue().toString()))
Review Comment:
I don't think lexicographic sorting gives us the adjacency property required by the check below.
For example, consider locations `s3://b/a`, `s3://b/a-archive`, and `s3://b/a/c` belonging to three different tables. They sort in that order, so `s3://b/a` and `s3://b/a/c` are never compared, even though `StorageLocation.isChildOf()` considers the latter a child of the former after normalizing the trailing slash.
Could we either keep the pairwise check, or sort/index by the same path semantics used by `isChildOf()`? A regression test with this shape would help pin the behavior.
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -212,6 +212,32 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull PolarisBaseEntity catalog,
@NonNull List<PolarisEntityCore> principalRoles);
+ /**
+ * Commits a {@link MetaStoreChangeSet} atomically. This is the primary mutation method; simpler
+ * per-entity operations are thin wrappers on top of it.
+ *
+ * <p>The default implementation falls back to calling individual per-entity methods in sequence
+ * and is therefore not atomic. Backends should override this method to provide the atomic
+ * guarantee.
+ *
+ * @param callCtx call context
+ * @param changeSet the set of entity creations and CAS-updates to commit
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx, @NonNull MetaStoreChangeSet changeSet) {
+ for (EntityWithPath create : changeSet.creates()) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
This sequential fallback makes an API documented as atomic partially commit for implementations that don't override it. NoSqlMetaStoreManager currently inherits this default, so if an earlier create succeeds and a later create/update fails, the earlier mutation remains durable.
That is especially dangerous for this caller because commitTransaction treats any batch failure as a transaction failure and cleans up the metadata files it wrote, potentially leaving the already-persisted table entity pointing at a deleted metadata file.
Could we make atomic commit a required capability here instead of falling back to sequential mutations like either implement the batch atomically for NoSQL or fail closed when the backend cannot provide that guarantee? WDYT?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on PR #4939:
URL: https://github.com/apache/polaris/pull/4939#issuecomment-4922975288
@snazy / @dimas-b any feedback / suggestions around this.
> It's a pretty big change. I think it deserved a dev ML discussion
It is here, seems quiet though
https://lists.apache.org/thread/2x25mj2pv981ocj83l8wfpr8yqvpplmg
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3575038695
##########
polaris-core/src/main/java/org/apache/polaris/core/storage/StorageUtil.java:
##########
@@ -83,6 +86,51 @@ public class StorageUtil {
return Set.of(viewMetadata.location());
}
+ /**
+ * Returns true if the set of locations used by {@code current} differs from the set used by
+ * {@code base}. Covers the table location as well as write.data.path and write.metadata.path
+ * property overrides.
+ */
+ public static boolean locationsChanged(TableMetadata base, TableMetadata current) {
+ return !getLocationsUsedByTable(base).equals(getLocationsUsedByTable(current));
+ }
+
+ /**
+ * Checks that no two distinct tables in {@code batch} use overlapping locations (pairwise
+ * containment check across all data locations returned by {@link #getLocationsUsedByTable}).
+ * Same-identifier pairs are skipped so a table's own write.data.path and write.metadata.path
+ * don't count as self-overlap.
+ *
+ * @throws BadRequestException if any two distinct tables have overlapping locations
+ */
+ public static void validateNoOverlapWithinBatch(
+ List<Map.Entry<TableIdentifier, TableMetadata>> batch) {
+ List<Map.Entry<TableIdentifier, StorageLocation>> entries =
+ batch.stream()
+ .flatMap(
+ e ->
+ getLocationsUsedByTable(e.getValue()).stream()
+ .map(loc -> Map.entry(e.getKey(), StorageLocation.of(loc))))
+ .toList();
+ for (int i = 0; i < entries.size(); i++) {
+ for (int j = i + 1; j < entries.size(); j++) {
+ TableIdentifier idA = entries.get(i).getKey();
+ TableIdentifier idB = entries.get(j).getKey();
+ if (idA.equals(idB)) {
+ continue;
+ }
+ StorageLocation a = entries.get(i).getValue();
+ StorageLocation b = entries.get(j).getValue();
+ if (a.isChildOf(b) || b.isChildOf(a)) {
Review Comment:
We can probably sort locations alphabetically and avoid `O(n^2)` performance. If a common prefix exists it's next entry will be the one that conflicts. WDYT?
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
`commitTransactionBatch()` can work, but I do not see a JDBC impl. for it. Are you planning to add one later?
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
Also, it feel that this method will become the focal point of a `PolarisMetaStoreManager`. It must be implemented correctly by all MetaStore managers.
With that in mind, I tend to think it might be best to reformulate all MetaStore changes in terms of this method.
Taking this one step further, `TransactionState` introduced in this PR looks very valuable to multi-object changes. Background (just FYI):
https://lists.apache.org/thread/rf5orxs815zs4h64p4rwp03q3pbgxb5r
https://lists.apache.org/thread/obvrrh0w36f87rwljptzwonpt89qkcm1
It feels like it may worth refactoring the MetaStore API in a more intrusive way and actually make `TransactionState` (or a similar object) a central piece of the API. Per-method "atomic" changes not longer look sufficient for the more advanced use cases we're trying to implement here. The multitude of change methods is hard to implement and those individual methods (like `PolarisMetaStoreManager.createEntityIfNotExists()` seem to be too specific, while being implementable through the "batch" change method.
What I'm trying to say is that using `TransactionState` in this PR just for the narrow use case of `commitTransation` does not feel right to me. It feels like we're making the most technically challenging aspect of the system a corner case. I'd rather make it the main case and other simpler cases can ride on top of it.
Sorry for increasing the scope of this PR, but I think it is time to have this discussion again :)
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on PR #4939:
URL: https://github.com/apache/polaris/pull/4939#issuecomment-5305263539
Not stale
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3571668467
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1297,137 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
Review Comment:
Thanks for confirming. I think it deserves a separate discussion. I think it's way beyond the scope of current PR :sweat_smile:
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3575672425
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
Thanx @dimas-b for the pointers.
> but I do not see a JDBC impl. for it.
I didn't catch it, can you explain a bit which class are you referring to. There isn't a separate JDBC-specific `commitTransactionBatch` class β JDBC deployments use `AtomicOperationMetaStoreManager` (via `JdbcMetaStoreManagerFactory`), and I added `commitTransactionBatch(MetaStoreChangeSet)` there. It merges all `creates` and `updates` into a single `writeEntities` call; `JdbcBasePersistenceImpl.writeEntities` already wraps that in `DatasourceOperations.runWithinTransaction` (auto-commit off, single connection, commit or rollback). So one batch commit = one RDBMS transaction.
Am I missing something here?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #4939:
URL: https://github.com/apache/polaris/pull/4939#issuecomment-5305241529
This PR is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3563057233
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
Thanx. Have added `AtomicOperationMetaStoreManager.commitTransactionBatch` which should take care, I have added tests around that as well in the `AtomicOperationMetaStoreManagerTest`
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java:
##########
@@ -1218,6 +1218,37 @@ void validateStagedTableCreate(TableIdentifier tableIdentifier, TableMetadata ta
catalogEntity, tableIdentifier, resolvedNamespace, location, storageLeafEntity));
}
+ /**
+ * Validates location overlap for an existing table whose locations changed. Unlike {@link
+ * #validateStagedTableCreate}, this uses the parent (namespace) path for resolution since the
+ * table already exists and its resolved path includes the table entity itself.
+ */
+ void validateTableLocationUpdate(TableIdentifier tableIdentifier, TableMetadata tableMetadata) {
+ PolarisResolvedPathWrapper resolvedTableEntities =
+ resolvedEntityView.getPassthroughResolvedPath(
Review Comment:
Added `validateNoOverlapWithinBatch` that runs first with a pairwise
`StorageLocation.isChildOf` check across all location-affecting entries (creates + location-changing updates) in the batch
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1297,137 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
Review Comment:
yes, what I could decode
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on PR #4939:
URL: https://github.com/apache/polaris/pull/4939#issuecomment-4856958831
@ayushtkn : Thanks for the contribution!
I only briefly looked at the PR, so I might have missed something :sweat_smile: I wonder what end-to-end flow is envisioned? Is it 1) do a staged create N times, 2) write table files, 3) commit all staged tables via the `commitTransaction` API?
It's a pretty big change. I think it deserved a `dev` ML discussion :slightly_smiling_face:
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3590815065
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
My preliminary preference would be towards the `Session` approach.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3597136313
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1297,137 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
Review Comment:
Related `dev` discussion: https://lists.apache.org/thread/vx0k8ow4k87m4y7cxpmojb0zy17t5ldy
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3575174400
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java:
##########
@@ -1218,6 +1218,37 @@ void validateStagedTableCreate(TableIdentifier tableIdentifier, TableMetadata ta
catalogEntity, tableIdentifier, resolvedNamespace, location, storageLeafEntity));
}
+ /**
+ * Validates location overlap for an existing table whose locations changed. Unlike {@link
+ * #validateStagedTableCreate}, this uses the parent (namespace) path for resolution since the
+ * table already exists and its resolved path includes the table entity itself.
+ */
+ void validateTableLocationUpdate(TableIdentifier tableIdentifier, TableMetadata tableMetadata) {
+ PolarisResolvedPathWrapper resolvedTableEntities =
+ resolvedEntityView.getPassthroughResolvedPath(
Review Comment:
Thx - this check LGTM, but let's keep discussing the batch commit method (separate thread).
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3575719158
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
Thanks for pointing me to the mailing list discussions. I read through the threads (including `Appendix A/B `and Dennis's Postgres notes) and think I have some context now.
**My understanding of the ask:** Move away from the `TransactionWorkspaceMetaStoreManager` pattern (intercepting individual calls, returning success before persistence, batching at the end). Instead, use a first-class `Session/ChangeSet` API where batching is the foundational contract, conflict detection is scoped to an explicit commit boundary, and backend isolation behavior is well-defined.
**What this PR does toward that:**
Introduced `MetaStoreChangeSet` as a first-class batch payload (instead of raw create/update lists).
`commitTransaction` now builds an explicit `MetaStoreChangeSet` and commits via `commitTransactionBatch(changeSet`).
**What is still interim:**
The API is not fully inverted yet β`commitTransactionBatch`(default) still calls the individual abstract methods, and backends are still built around single-entity ops.ofCreate/of Update are there so single ops can later ride on batch commit, but that wiring is follow-up.
`TransactionWorkspaceMetaStoreManager` remains in `IcebergCatalogHandler` as a temporary bridge to run Iceberg table ops and collect pending mutations before assembling the final change-set.
Does this `MetaStoreChangeSet `pivot provide a good enough architectural stepping stone to merge this and unblock the Iceberg `commitTransaction` endpoint today? Or do you strongly feel we need to rip out the `TransactionWorkspaceMetaStoreManager` and build the full `Session` API within this PR itself?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "ayushtkn (via GitHub)" <gi...@apache.org>.
ayushtkn commented on PR #4939:
URL: https://github.com/apache/polaris/pull/4939#issuecomment-4857432876
Thanx @dimas-b for taking a look. yep it is the same.
1. Client calls `catalog.buildTable(id, schema).stageCreate()` N times β builds table metadata locally, no server roundtrip
2. Client optionally writes data files to the staged locations
3. Client calls catalog.commitTransaction() β sends a single `CommitTransactionRequest` containing all staged creates (identified by `AssertTableDoesNotExist` requirement) and any regular updates
The server only sees step 3. This PR implements the server-side handling for that request when it includes staged creates alongside regular updates.
Sure, I will shoot a mail to dev@ for more feedback
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3561632341
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1312,151 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
+ List<Map.Entry<TableIdentifier, TableMetadata>> stagedCreateEntries = new ArrayList<>();
+ // Track regular updates where locations changed (location, write.data.path,
+ // write.metadata.path)
+ // for deferred overlap validation inside the pre-commit hook.
+ List<Map.Entry<TableIdentifier, TableMetadata>> locationChangedUpdates = new ArrayList<>();
+ try {
+ changesByTable.forEach(
+ (tableIdentifier, changes) -> {
+ boolean isStagedCreate = changes.stream().anyMatch(CatalogHandlerUtils::isCreate);
Review Comment:
Why are we calling it "staged"? How is it different from a normal create? As far as I can tell, there is no difference π€
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1312,151 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
+ List<Map.Entry<TableIdentifier, TableMetadata>> stagedCreateEntries = new ArrayList<>();
+ // Track regular updates where locations changed (location, write.data.path,
+ // write.metadata.path)
+ // for deferred overlap validation inside the pre-commit hook.
+ List<Map.Entry<TableIdentifier, TableMetadata>> locationChangedUpdates = new ArrayList<>();
+ try {
+ changesByTable.forEach(
+ (tableIdentifier, changes) -> {
+ 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(
Review Comment:
Is it normal for a create request to have more than one `UpdateTableRequest` element?
Should we simply check that there is one and it satisfies `isCreate()`?
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalMetaStoreManagerImpl.java:
##########
@@ -1172,6 +1172,41 @@ public void deletePrincipalSecrets(
callCtx, () -> this.updateEntitiesPropertiesIfNotChanged(callCtx, ms, entities));
}
+ /** {@inheritDoc} */
+ @Override
+ public @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
+
+ return ms.runInTransaction(
Review Comment:
@ayushtkn : I'm not sure whether you're aware of this, but this code will NOT be used for JDBC Persistence... Just FYI π€· Check usage paths to confirm, if you prefer.
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1312,151 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
+ List<Map.Entry<TableIdentifier, TableMetadata>> stagedCreateEntries = new ArrayList<>();
+ // Track regular updates where locations changed (location, write.data.path,
+ // write.metadata.path)
+ // for deferred overlap validation inside the pre-commit hook.
+ List<Map.Entry<TableIdentifier, TableMetadata>> locationChangedUpdates = new ArrayList<>();
+ try {
+ changesByTable.forEach(
+ (tableIdentifier, changes) -> {
+ 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);
+ }
+
+ 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 the AssertTableDoesNotExist requirement against null metadata
+ // (table does not exist yet). This is the only requirement allowed for staged
+ // creates β CatalogHandlerUtils.isCreate() rejects any others.
+ 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);
- });
+ // Track updates where locations changed for deferred overlap validation
+ if (locationsChanged(baseMetadata, currentMetadata)) {
+ locationChangedUpdates.add(Map.entry(tableIdentifier, currentMetadata));
+ }
+
+ tableMetadataObjs.add(currentMetadata);
+ }
+ });
+ } finally {
+ // Always restore the real metastore manager, even if processing throws.
+ ((LocalIcebergCatalog) baseCatalog).setMetaStoreManager(metaStoreManager());
+ }
- // Commit the collected updates in a single atomic operation
+ List<EntityWithPath> pendingCreations = transactionMetaStoreManager.getPendingCreations();
List<EntityWithPath> pendingUpdates = transactionMetaStoreManager.getPendingUpdates();
+
+ // Validate staged creates against the real metastore (now restored).
+ // Check intra-transaction overlap first (DB overlap check won't see uncommitted peers).
+ Set<String> claimedLocationsInTx = new HashSet<>();
+ for (Map.Entry<TableIdentifier, TableMetadata> entry : stagedCreateEntries) {
+ Set<String> dataLocations = StorageUtil.getLocationsUsedByTable(entry.getValue());
+ for (String location : dataLocations) {
+ if (!claimedLocationsInTx.add(location)) {
+ throw new BadRequestException(
+ "Transaction contains multiple creates pointing to the same location: %s", location);
+ }
+ }
+ ((LocalIcebergCatalog) baseCatalog)
Review Comment:
nit: this cast is done multiple times. It might be worth introducing a strongly typed variable for this.
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/LocalIcebergCatalog.java:
##########
@@ -1218,6 +1218,37 @@ void validateStagedTableCreate(TableIdentifier tableIdentifier, TableMetadata ta
catalogEntity, tableIdentifier, resolvedNamespace, location, storageLeafEntity));
}
+ /**
+ * Validates location overlap for an existing table whose locations changed. Unlike {@link
+ * #validateStagedTableCreate}, this uses the parent (namespace) path for resolution since the
+ * table already exists and its resolved path includes the table entity itself.
+ */
+ void validateTableLocationUpdate(TableIdentifier tableIdentifier, TableMetadata tableMetadata) {
+ PolarisResolvedPathWrapper resolvedTableEntities =
+ resolvedEntityView.getPassthroughResolvedPath(
Review Comment:
This is called before changes are committed. Can it detect overlaps within the multi-table commit batch?
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
In current JDBC persistence code each of those changes will run in a separate RDBMS Tx, AFAIK. It is based on `AtomicOperationMetaStoreManager` π€·
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1312,151 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
+ List<Map.Entry<TableIdentifier, TableMetadata>> stagedCreateEntries = new ArrayList<>();
+ // Track regular updates where locations changed (location, write.data.path,
+ // write.metadata.path)
+ // for deferred overlap validation inside the pre-commit hook.
+ List<Map.Entry<TableIdentifier, TableMetadata>> locationChangedUpdates = new ArrayList<>();
+ try {
+ changesByTable.forEach(
+ (tableIdentifier, changes) -> {
+ 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);
+ }
+
+ 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 the AssertTableDoesNotExist requirement against null metadata
+ // (table does not exist yet). This is the only requirement allowed for staged
+ // creates β CatalogHandlerUtils.isCreate() rejects any others.
+ 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);
Review Comment:
Should we delegate to `CatalogHandlerUtils.create()` perhaps? π€
##########
runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java:
##########
@@ -1205,84 +1297,137 @@ 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<>();
- 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());
+ // Track staged-creates for deferred location validation after the real metastore is restored.
Review Comment:
The TOCTOU problem wrt location checks exists in the per-table create/update flows too, does it not?
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3590809344
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
@ayushtkn : I hope to post a `dev` thread on this to start a broader discussion. This PR makes sense, but it feels like we need to broaden the scope. Let's discuss on `dev` (soon) and depending on feedback we can either do for a bigger change or go back to this PR.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3590809344
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
@ayushtkn : I hope to post a `dev` thread on this to start a broader discussion. This PR makes sense, but it feels like we need to broaden the scope. Let's discussion on `dev` (soon) and depending on feedback we can either do for a bigger change or go back to this PR.
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support staged table creates in commitTransaction [polaris]
Posted by "dimas-b (via GitHub)" <gi...@apache.org>.
dimas-b commented on code in PR #4939:
URL: https://github.com/apache/polaris/pull/4939#discussion_r3597137810
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
Cf .https://lists.apache.org/thread/vx0k8ow4k87m4y7cxpmojb0zy17t5ldy
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManager.java:
##########
@@ -284,6 +284,30 @@ default BaseResult bootstrapPolarisService(@NonNull PolarisCallContext callCtx)
@NonNull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@NonNull PolarisCallContext callCtx, @NonNull List<EntityWithPath> entities);
+ /**
+ * Commits a batch of entity creations and property updates within a single transaction.
+ *
+ * @param callCtx call context
+ * @param creates entities to create
+ * @param updates entities to update (compare-and-swap)
+ * @return result indicating success or failure
+ */
+ default @NonNull EntitiesResult commitTransactionBatch(
+ @NonNull PolarisCallContext callCtx,
+ @NonNull List<EntityWithPath> creates,
+ @NonNull List<EntityWithPath> updates) {
+ for (EntityWithPath create : creates) {
+ EntityResult result = createEntityIfNotExists(callCtx, create.catalogPath(), create.entity());
Review Comment:
Cf. https://lists.apache.org/thread/vx0k8ow4k87m4y7cxpmojb0zy17t5ldy
--
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: issues-unsubscribe@polaris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org