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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
