You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@polaris.apache.org by dh...@apache.org on 2025/03/11 16:39:14 UTC
(polaris) branch main updated: Core persistence refactor phase 3 (diffbase on phase 2) - Add AtomicOperationMetaStoreManager which passes all tests (#1139)
This is an automated email from the ASF dual-hosted git repository.
dhuo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git
The following commit(s) were added to refs/heads/main by this push:
new b6ebbafe3 Core persistence refactor phase 3 (diffbase on phase 2) - Add AtomicOperationMetaStoreManager which passes all tests (#1139)
b6ebbafe3 is described below
commit b6ebbafe30c4caab4be29fbeb4d31fefd2aa7a20
Author: Dennis Huo <74...@users.noreply.github.com>
AuthorDate: Tue Mar 11 09:39:07 2025 -0700
Core persistence refactor phase 3 (diffbase on phase 2) - Add AtomicOperationMetaStoreManager which passes all tests (#1139)
* Initial sketch of how shared code can be lifted from PolarisMetaStoreManagerImpl into BaseMetaStoreManager
* Make all TransactionalPersistence methods explicit in terms of whether they are
one-shot atomic or if they are intended to be run as part of a caller-managed
transaction, in which case the action will not be flushed/durable until the
transaction is committed.
Delegate all the BasePersistence methods to subclass variations of *InCurrentTxn
so that the TransactionalPersistence implementations also happen to fully
implement the real semantics of BasePersistence.
* Invert the method naming semantics; "doFooAtomically" is now the set of "new" methods defined in `BasePersistence`
while the original "doFoo" will mean that it expects to be run in a current transaction. Removed all the InCurrentTxn
suffixes.
* Copy/paste PolarisMetaStoreManagerImpl.java to a new AtomicOperationMetaStoreManager
as a diffbase.
* Add AtomicOperationMetaStoreManager class which starts as just a full copy/paste
of PolarisMetaStoreManagerImpl, but remove all its usage of runInTransaction
instead only using the "doFooAtomically" BasePersistence methods.
Implement the intended compare-and-swap semantics in writeEntity/writeEntities
in AbstractTransactionalPersistence so that the TreeMap and EclipseLink impls
can both actually successfully be used in "single-durable-operation" mode
as plain BasePersistence implementations (e.g. without using their
runInTransaction capabilities at all).
* Update all concurrency TODO comments with more thorough explanation of implications
of failure modes in the absence of additional bulk methods,.
Reorder the internals of dropEntity to drop the entity first, instead of grants,
so that the drop attempt can't put the entity into a partially-broken but still
existing state; instead, server failures just might cause garbage grant records
which are safe/expected but need to be cleaned up in a background task somewhere.
Fix the atomicity of name-collision-checks in createCatalog and createPrincipal.
With these changes, the behavior of AtomicOperationMetaStoreManager should be
fully "safe"/correct, with only the following suboptimal scenarios:
-If the server fails during CreateCatalog, it's possible for a
partially-initialized catalog (e.g. lacking catalog_admin and grants) to exist;
this should just be deleted and the creation should be retried in such a case.
-If grants on parents of an entity (e.g. namespaces, catalogs) are changed
in the middle of an operation on the child entity, it's possible for
happens-before semantics to be violated up to the duration of a single
in-flight request if grants are not cached, and up to the duration of
cache expiration in the worse case if the server crashes after updating
grants but before updating grantRecordVersions on affected entities.
-If a parent entity (e.g. namespace, catalog) is deleted in the middle
of creating or updating a child entity, the create/update may still
return "success" despite the parent being deleted, which effectively
deletes the child entity as well.
-If a new entity is being created under an empty namespace/catalog at
the same time the parent namespace/catalog is being deleted, despite
the semantic of preventing deletion of non-empty containers, the
creation may succeed *and* the deletion of the parent may succeed,
effectively deleting the child entity as well; this can only happen
within the concurrency window of a single request (e.g. happening
within milliseconds of each other).
-In general, in the face of server failures or concurrency collisions,
grantRecords may be stale up to the lifetime of EntityCache elements.
-In general, whether or not there are server failures or concurrency
collisions, the grantsOnSecurable and grantsToGrantee of a given
entity may not be evaluated at the same effective SNAPSHOT_READ
point in time, but this shouldn't be a problem because the securable
side of grants can already be modified independently from the
grantee side of grants.
* Step 1 of inverting the semantics again so that we only define "InCurrentTxn" versions of methods
in TransactionalPersistence.
The fact that this unit compiles before we've renamed doFooAtomically back to doFoo helps
prove that we didn't leave anything accidentally behind to call the atomic versions inside
a transaction, because as of this moment there is not unqualified method name doFoo anymore;
only either doFooInCurrentTxn or doFooAtomically. Committing this as a separate unit to
have the intermediate proof-of-correctness.
* Step 2 of inverting the method names; remove the "Atomically" suffix from all method
names; now all the unqualified methods in BasePersistence are implicitly supposed
to behave "atomically". All the methods that expect to be run inside a transaction
that's managed by the outer caller will have the "InCurrentTxn" suffix.
* Update javadoc for writeEntities per suggestion from offline discussion
* From offline discussion, make PolarisCallContext just return a BasePersistence
instead of a TransactionalPersistence. Just use blind casts for now in
PolarisMetaStoreManagerImpl; refactoring can be done later to better define
how the Persistence impl is injected in a more typesafe way for compatible
PolarisMetaStoreManager impls.
* Restore application[-it].properties to default in-memory impl for now instead
of using the atomic one.
---
.../PolarisEclipseLinkMetaStoreSessionImpl.java | 78 +-
.../apache/polaris/core/PolarisCallContext.java | 10 +-
...l.java => AtomicOperationMetaStoreManager.java} | 1434 ++++++--------------
.../core/persistence/BaseMetaStoreManager.java | 186 +++
.../polaris/core/persistence/BasePersistence.java | 31 +-
.../persistence/EntityAlreadyExistsException.java | 41 +
.../LocalPolarisMetaStoreManagerFactory.java | 10 +-
.../AbstractTransactionalPersistence.java | 662 +++++++--
.../transactional/PolarisEntityResolver.java | 2 +-
.../transactional/PolarisMetaStoreManagerImpl.java | 423 +++---
.../PolarisTreeMapMetaStoreSessionImpl.java | 79 +-
.../transactional/PolarisTreeMapStore.java | 9 +-
.../transactional/TransactionalPersistence.java | 229 +++-
...TreeMapAtomicOperationMetaStoreManagerTest.java | 47 +
.../src/main/resources/application-it.properties | 2 +
.../src/main/resources/application.properties | 2 +
.../quarkus/catalog/BasePolarisCatalogTest.java | 2 +-
...moryAtomicOperationMetaStoreManagerFactory.java | 52 +
18 files changed, 1845 insertions(+), 1454 deletions(-)
diff --git a/extension/persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java b/extension/persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java
index 5637a9327..7005d5cfe 100644
--- a/extension/persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java
+++ b/extension/persistence/eclipselink/src/main/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreSessionImpl.java
@@ -247,33 +247,34 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
* @return new unique entity identifier
*/
@Override
- public long generateNewId(@Nonnull PolarisCallContext callCtx) {
+ public long generateNewIdInCurrentTxn(@Nonnull PolarisCallContext callCtx) {
// This function can be called within a transaction or out of transaction.
// If called out of transaction, create a new transaction, otherwise run in current transaction
return localSession.get() != null
? this.store.getNextSequence(localSession.get())
- : runInReadTransaction(callCtx, () -> generateNewId(callCtx));
+ : runInReadTransaction(callCtx, () -> generateNewIdInCurrentTxn(callCtx));
}
/** {@inheritDoc} */
@Override
- public void writeToEntities(
+ public void writeToEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
this.store.writeToEntities(localSession.get(), entity);
}
/** {@inheritDoc} */
@Override
- public <T extends PolarisStorageConfigurationInfo> void persistStorageIntegrationIfNeeded(
- @Nonnull PolarisCallContext callContext,
- @Nonnull PolarisBaseEntity entity,
- @Nullable PolarisStorageIntegration<T> storageIntegration) {
+ public <T extends PolarisStorageConfigurationInfo>
+ void persistStorageIntegrationIfNeededInCurrentTxn(
+ @Nonnull PolarisCallContext callContext,
+ @Nonnull PolarisBaseEntity entity,
+ @Nullable PolarisStorageIntegration<T> storageIntegration) {
// not implemented for eclipselink store
}
/** {@inheritDoc} */
@Override
- public void writeToEntitiesActive(
+ public void writeToEntitiesActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
// write it
this.store.writeToEntitiesActive(localSession.get(), entity);
@@ -281,7 +282,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void writeToEntitiesChangeTracking(
+ public void writeToEntitiesChangeTrackingInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
// write it
this.store.writeToEntitiesChangeTracking(localSession.get(), entity);
@@ -289,7 +290,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void writeToGrantRecords(
+ public void writeToGrantRecordsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {
// write it
this.store.writeToGrantRecords(localSession.get(), grantRec);
@@ -297,7 +298,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void deleteFromEntities(
+ public void deleteFromEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
// delete it
@@ -306,7 +307,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void deleteFromEntitiesActive(
+ public void deleteFromEntitiesActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
// delete it
this.store.deleteFromEntitiesActive(localSession.get(), new PolarisEntitiesActiveKey(entity));
@@ -319,7 +320,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
* @param entity entity record to delete
*/
@Override
- public void deleteFromEntitiesChangeTracking(
+ public void deleteFromEntitiesChangeTrackingInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
// delete it
this.store.deleteFromEntitiesChangeTracking(localSession.get(), entity);
@@ -327,14 +328,14 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void deleteFromGrantRecords(
+ public void deleteFromGrantRecordsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {
this.store.deleteFromGrantRecords(localSession.get(), grantRec);
}
/** {@inheritDoc} */
@Override
- public void deleteAllEntityGrantRecords(
+ public void deleteAllEntityGrantRecordsInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
@Nonnull PolarisEntityCore entity,
@Nonnull List<PolarisGrantRecord> grantsOnGrantee,
@@ -344,19 +345,19 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void deleteAll(@Nonnull PolarisCallContext callCtx) {
+ public void deleteAllInCurrentTxn(@Nonnull PolarisCallContext callCtx) {
this.store.deleteAll(localSession.get());
}
/** {@inheritDoc} */
@Override
- public @Nullable PolarisBaseEntity lookupEntity(
+ public @Nullable PolarisBaseEntity lookupEntityInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long catalogId, long entityId, int typeCode) {
return ModelEntity.toEntity(this.store.lookupEntity(localSession.get(), catalogId, entityId));
}
@Override
- public @Nonnull List<PolarisBaseEntity> lookupEntities(
+ public @Nonnull List<PolarisBaseEntity> lookupEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds) {
return this.store.lookupEntities(localSession.get(), entityIds).stream()
.map(ModelEntity::toEntity)
@@ -365,7 +366,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisChangeTrackingVersions> lookupEntityVersions(
+ public @Nonnull List<PolarisChangeTrackingVersions> lookupEntityVersionsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds) {
Map<PolarisEntityId, ModelEntity> idToEntityMap =
this.store.lookupEntities(localSession.get(), entityIds).stream()
@@ -388,7 +389,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
@Nullable
- public EntityNameLookupRecord lookupEntityActive(
+ public EntityNameLookupRecord lookupEntityActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntitiesActiveKey entityActiveKey) {
// lookup the active entity slice
return ModelEntityActive.toEntityActive(
@@ -398,34 +399,35 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
@Nonnull
- public List<EntityNameLookupRecord> lookupEntityActiveBatch(
+ public List<EntityNameLookupRecord> lookupEntityActiveBatchInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
@Nonnull List<PolarisEntitiesActiveKey> entityActiveKeys) {
// now build a list to quickly verify that nothing has changed
return entityActiveKeys.stream()
- .map(entityActiveKey -> this.lookupEntityActive(callCtx, entityActiveKey))
+ .map(entityActiveKey -> this.lookupEntityActiveInCurrentTxn(callCtx, entityActiveKey))
.collect(Collectors.toList());
}
/** {@inheritDoc} */
@Override
- public @Nonnull List<EntityNameLookupRecord> listEntities(
+ public @Nonnull List<EntityNameLookupRecord> listEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long parentId,
@Nonnull PolarisEntityType entityType) {
- return listEntities(callCtx, catalogId, parentId, entityType, Predicates.alwaysTrue());
+ return this.listEntitiesInCurrentTxn(
+ callCtx, catalogId, parentId, entityType, Predicates.alwaysTrue());
}
@Override
- public @Nonnull List<EntityNameLookupRecord> listEntities(
+ public @Nonnull List<EntityNameLookupRecord> listEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long parentId,
@Nonnull PolarisEntityType entityType,
@Nonnull Predicate<PolarisBaseEntity> entityFilter) {
// full range scan under the parent for that type
- return listEntities(
+ return this.listEntitiesInCurrentTxn(
callCtx,
catalogId,
parentId,
@@ -443,7 +445,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
}
@Override
- public @Nonnull <T> List<T> listEntities(
+ public @Nonnull <T> List<T> listEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long parentId,
@@ -464,7 +466,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public boolean hasChildren(
+ public boolean hasChildrenInCurrentTxn(
@Nonnull PolarisCallContext callContext,
@Nullable PolarisEntityType entityType,
long catalogId,
@@ -476,7 +478,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public int lookupEntityGrantRecordsVersion(
+ public int lookupEntityGrantRecordsVersionInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long catalogId, long entityId) {
ModelEntityChangeTracking entity =
this.store.lookupEntityChangeTracking(localSession.get(), catalogId, entityId);
@@ -487,7 +489,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nullable PolarisGrantRecord lookupGrantRecord(
+ public @Nullable PolarisGrantRecord lookupGrantRecordInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long securableCatalogId,
long securableId,
@@ -507,7 +509,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnSecurable(
+ public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnSecurableInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId) {
// now fetch all grants for this securable
return this.store
@@ -519,7 +521,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnGrantee(
+ public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnGranteeInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId) {
// now fetch all grants assigned to this grantee
return this.store
@@ -531,7 +533,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nullable PolarisPrincipalSecrets loadPrincipalSecrets(
+ public @Nullable PolarisPrincipalSecrets loadPrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull String clientId) {
return ModelPrincipalSecrets.toPrincipalSecrets(
this.store.lookupPrincipalSecrets(localSession.get(), clientId));
@@ -539,7 +541,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nonnull PolarisPrincipalSecrets generateNewPrincipalSecrets(
+ public @Nonnull PolarisPrincipalSecrets generateNewPrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull String principalName, long principalId) {
// ensure principal client id is unique
PolarisPrincipalSecrets principalSecrets;
@@ -563,7 +565,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public @Nonnull PolarisPrincipalSecrets rotatePrincipalSecrets(
+ public @Nonnull PolarisPrincipalSecrets rotatePrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
@Nonnull String clientId,
long principalId,
@@ -610,7 +612,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
- public void deletePrincipalSecrets(
+ public void deletePrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull String clientId, long principalId) {
// load the existing secrets
ModelPrincipalSecrets principalSecrets =
@@ -643,7 +645,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
public @Nullable <T extends PolarisStorageConfigurationInfo>
- PolarisStorageIntegration<T> createStorageIntegration(
+ PolarisStorageIntegration<T> createStorageIntegrationInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long entityId,
@@ -655,7 +657,7 @@ public class PolarisEclipseLinkMetaStoreSessionImpl extends AbstractTransactiona
/** {@inheritDoc} */
@Override
public @Nullable <T extends PolarisStorageConfigurationInfo>
- PolarisStorageIntegration<T> loadPolarisStorageIntegration(
+ PolarisStorageIntegration<T> loadPolarisStorageIntegrationInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
PolarisStorageConfigurationInfo storageConfig =
BaseMetaStoreManager.extractStorageConfiguration(callCtx, entity);
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java b/polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java
index c584dde55..3b5e96251 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java
@@ -21,7 +21,7 @@ package org.apache.polaris.core;
import jakarta.annotation.Nonnull;
import java.time.Clock;
import java.time.ZoneId;
-import org.apache.polaris.core.persistence.transactional.TransactionalPersistence;
+import org.apache.polaris.core.persistence.BasePersistence;
/**
* The Call context is allocated each time a new REST request is processed. It contains instances of
@@ -30,7 +30,7 @@ import org.apache.polaris.core.persistence.transactional.TransactionalPersistenc
public class PolarisCallContext {
// meta store which is used to persist Polaris entity metadata
- private final TransactionalPersistence metaStore;
+ private final BasePersistence metaStore;
// diag services
private final PolarisDiagnostics diagServices;
@@ -40,7 +40,7 @@ public class PolarisCallContext {
private final Clock clock;
public PolarisCallContext(
- @Nonnull TransactionalPersistence metaStore,
+ @Nonnull BasePersistence metaStore,
@Nonnull PolarisDiagnostics diagServices,
@Nonnull PolarisConfigurationStore configurationStore,
@Nonnull Clock clock) {
@@ -51,7 +51,7 @@ public class PolarisCallContext {
}
public PolarisCallContext(
- @Nonnull TransactionalPersistence metaStore, @Nonnull PolarisDiagnostics diagServices) {
+ @Nonnull BasePersistence metaStore, @Nonnull PolarisDiagnostics diagServices) {
this.metaStore = metaStore;
this.diagServices = diagServices;
this.configurationStore = new PolarisConfigurationStore() {};
@@ -66,7 +66,7 @@ public class PolarisCallContext {
original.getClock());
}
- public TransactionalPersistence getMetaStore() {
+ public BasePersistence getMetaStore() {
return metaStore;
}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java
similarity index 59%
copy from polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java
copy to polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java
index da09a6799..df3192c59 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManager.java
@@ -16,13 +16,8 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.polaris.core.persistence.transactional;
+package org.apache.polaris.core.persistence;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Nullable;
import java.util.ArrayList;
@@ -33,6 +28,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.stream.Collectors;
import org.apache.polaris.core.PolarisCallContext;
@@ -50,7 +46,6 @@ import org.apache.polaris.core.entity.PolarisGrantRecord;
import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
import org.apache.polaris.core.entity.PolarisPrivilege;
import org.apache.polaris.core.entity.PolarisTaskConstants;
-import org.apache.polaris.core.persistence.*;
import org.apache.polaris.core.persistence.dao.entity.BaseResult;
import org.apache.polaris.core.persistence.dao.entity.ChangeTrackingResult;
import org.apache.polaris.core.persistence.dao.entity.CreateCatalogResult;
@@ -59,7 +54,6 @@ import org.apache.polaris.core.persistence.dao.entity.DropEntityResult;
import org.apache.polaris.core.persistence.dao.entity.EntitiesResult;
import org.apache.polaris.core.persistence.dao.entity.EntityResult;
import org.apache.polaris.core.persistence.dao.entity.EntityWithPath;
-import org.apache.polaris.core.persistence.dao.entity.GenerateEntityIdResult;
import org.apache.polaris.core.persistence.dao.entity.ListEntitiesResult;
import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
import org.apache.polaris.core.storage.PolarisCredentialProperty;
@@ -70,84 +64,52 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Default implementation of the Polaris Meta Store Manager. Uses the underlying meta store to store
- * and retrieve all Polaris metadata
+ * Implementation of PolarisMetaStoreManager which only relies on one-shot atomic operations into a
+ * BasePersistence implementation without any kind of open-ended multi-statement transactions.
*/
-@SuppressFBWarnings("NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE")
-public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
- private static final Logger LOGGER = LoggerFactory.getLogger(PolarisMetaStoreManagerImpl.class);
-
- /** mapper, allows to serialize/deserialize properties to/from JSON */
- private static final ObjectMapper MAPPER = new ObjectMapper();
+public class AtomicOperationMetaStoreManager extends BaseMetaStoreManager {
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(AtomicOperationMetaStoreManager.class);
/**
- * Persist the specified new entity. Persist will write this entity in the ENTITIES, in the
- * ENTITIES_ACTIVE and finally in the ENTITIES_CHANGE_TRACKING tables
+ * Persist the specified new entity.
*
* @param callCtx call context
* @param ms meta store in read/write mode
- * @param entity entity we need a DPO for
+ * @param entity entity we need a new persisted record for
*/
- private void persistNewEntity(
+ private EntityResult persistNewEntity(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
+ @Nonnull BasePersistence ms,
@Nonnull PolarisBaseEntity entity) {
+ // Invoke shared logic for validation and filling out remaining fields.
+ entity = prepareToPersistNewEntity(callCtx, ms, entity);
- // validate the entity type and subtype
- callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
- callCtx
- .getDiagServices()
- .checkNotNull(entity.getName(), "unexpected_null_name", "entity={}", entity);
- PolarisEntityType type = PolarisEntityType.fromCode(entity.getTypeCode());
- callCtx.getDiagServices().checkNotNull(type, "unknown_type", "entity={}", entity);
- PolarisEntitySubType subType = PolarisEntitySubType.fromCode(entity.getSubTypeCode());
- callCtx.getDiagServices().checkNotNull(subType, "unexpected_null_subType", "entity={}", entity);
- callCtx
- .getDiagServices()
- .check(
- subType.getParentType() == null || subType.getParentType() == type,
- "invalid_subtype",
- "type={} subType={}",
- type,
- subType);
-
- // if top-level entity, its parent should be the account
- callCtx
- .getDiagServices()
- .check(
- !type.isTopLevel() || entity.getParentId() == PolarisEntityConstants.getRootEntityId(),
- "top_level_parent_should_be_account",
- "entity={}",
- entity);
-
- // id should not be null
- callCtx
- .getDiagServices()
- .check(
- entity.getId() != 0 || type == PolarisEntityType.ROOT,
- "id_not_set",
- "entity={}",
- entity);
-
- // creation timestamp must be filled
- callCtx.getDiagServices().check(entity.getCreateTimestamp() != 0, "null_create_timestamp");
-
- // this is the first change
- entity.setLastUpdateTimestamp(entity.getCreateTimestamp());
-
- // set all other timestamps to 0
- entity.setDropTimestamp(0);
- entity.setPurgeTimestamp(0);
- entity.setToPurgeTimestamp(0);
-
- // write it
- ms.writeEntity(callCtx, entity, true, null);
+ try {
+ // write it
+ ms.writeEntity(callCtx, entity, true, null);
+ } catch (EntityAlreadyExistsException e) {
+ if (e.getExistingEntity().getId() == entity.getId()) {
+ // Since ids are unique and reserved when generated, a matching id means a low-level retry
+ // which should behave idempotently; simply return the entity we were trying to write.
+ // Note: We return the "entity" instead of the "existingEntity" in case the entity has
+ // also since been updated; our idempotent creation is perfectly allowed to finish with
+ // a later timestamp than some update that may have inserted itself after creation but
+ // before the retry, and so the caller of the creation should see the original entity we
+ // were trying to create, not the updated one.
+ return new EntityResult(entity);
+ } else {
+ // If ids don't match it means we hit a name collision.
+ return new EntityResult(
+ BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS, e.getExistingEntity().getSubTypeCode());
+ }
+ }
+ return new EntityResult(entity);
}
/**
* Persist the specified entity after it has been changed. We will update the last changed time,
- * increment the entity version and persist it back to the ENTITIES and ENTITIES_CHANGE_TRACKING
- * tables
+ * increment the entity version and persist it in one atomic operation.
*
* @param callCtx call context
* @param ms meta store
@@ -158,50 +120,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
*/
private @Nonnull PolarisBaseEntity persistEntityAfterChange(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
+ @Nonnull BasePersistence ms,
@Nonnull PolarisBaseEntity entity,
boolean nameOrParentChanged,
@Nonnull PolarisBaseEntity originalEntity) {
-
- // validate the entity type and subtype
- callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
- callCtx
- .getDiagServices()
- .checkNotNull(entity.getName(), "unexpected_null_name", "entity={}", entity);
- PolarisEntityType type = entity.getType();
- callCtx.getDiagServices().checkNotNull(type, "unexpected_null_type", "entity={}", entity);
- PolarisEntitySubType subType = entity.getSubType();
- callCtx.getDiagServices().checkNotNull(subType, "unexpected_null_subType", "entity={}", entity);
- callCtx
- .getDiagServices()
- .check(
- subType.getParentType() == null || subType.getParentType() == type,
- "invalid_subtype",
- "type={} subType={} entity={}",
- type,
- subType,
- entity);
-
- // entity should not have been dropped
- callCtx
- .getDiagServices()
- .check(entity.getDropTimestamp() == 0, "entity_dropped", "entity={}", entity);
-
- // creation timestamp must be filled
- long createTimestamp = entity.getCreateTimestamp();
- callCtx
- .getDiagServices()
- .check(createTimestamp != 0, "null_create_timestamp", "entity={}", entity);
-
- // ensure time is not moving backward...
- long now = System.currentTimeMillis();
- if (now < entity.getCreateTimestamp()) {
- now = entity.getCreateTimestamp() + 1;
- }
-
- // update last update timestamp and increment entity version
- entity.setLastUpdateTimestamp(now);
- entity.setEntityVersion(entity.getEntityVersion() + 1);
+ // Invoke shared logic for validation and updating expected fields.
+ entity =
+ prepareToPersistEntityAfterChange(callCtx, ms, entity, nameOrParentChanged, originalEntity);
// persist it to the various slices
ms.writeEntity(callCtx, entity, nameOrParentChanged, originalEntity);
@@ -226,7 +151,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
*/
private void dropEntity(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
+ @Nonnull BasePersistence ms,
@Nonnull PolarisBaseEntity entity) {
// validate the entity type and subtype
@@ -236,7 +161,19 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// creation timestamp must be filled
callCtx.getDiagServices().check(entity.getDropTimestamp() == 0, "already_dropped");
- // for now drop all associated grants, etc. synchronously
+ // Remove the main entity itself first-thing; once its id no longer resolves successfully
+ // it will be pruned out of any grant-record lookups anyways.
+ ms.deleteEntity(callCtx, entity);
+
+ // Best-effort cleanup - drop grant records, update grantRecordVersions for affected
+ // other entities.
+ // TODO: Support some more formal garbage-collection mechanism so that a garbage-collector
+ // doesn't have to "guess" at whether orphaned id links are garbage vs "newly allocated" ids
+ // that just haven't been fully committed yet. Potentially pre-populate a tombstone
+ // record that only behaves as a tombstone if the main entity is also verified to be
+ // gone; then garbage-collection looks for cleanup grants based on the tombstone and only
+ // removes the tombstone once cleanup is done; cleanup can occur incrementally.
+
// delete ALL grant records to (if the entity is a grantee) and from that entity
final List<PolarisGrantRecord> grantsOnGrantee =
(entity.getType().isGrantee())
@@ -268,9 +205,6 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
ms.writeEntity(callCtx, entityGrantChanged, false, originalEntity);
}
- // remove the entity being dropped now
- ms.deleteEntity(callCtx, entity);
-
// if it is a principal, we also need to drop the secrets
if (entity.getType() == PolarisEntityType.PRINCIPAL) {
// get internal properties
@@ -281,7 +215,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
String clientId = properties.get(PolarisEntityConstants.getClientIdPropertyName());
// delete it from the secret slice
- ms.deletePrincipalSecrets(callCtx, clientId, entity.getId());
+ ((IntegrationPersistence) ms).deletePrincipalSecrets(callCtx, clientId, entity.getId());
}
}
@@ -298,7 +232,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
*/
private @Nonnull PolarisGrantRecord persistNewGrantRecord(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
+ @Nonnull BasePersistence ms,
@Nonnull PolarisEntityCore securable,
@Nonnull PolarisEntityCore grantee,
@Nonnull PolarisPrivilege priv) {
@@ -352,6 +286,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
securableEntity.setGrantRecordsVersion(securableEntity.getGrantRecordsVersion() + 1);
ms.writeEntity(callCtx, securableEntity, false, originalSecurableEntity);
+ // TODO: Reorder and/or expose bulk update of both grantRecordsVersions and grant records. In
+ // the meantime, cache can be disabled or configured with a short enough expiry time to
+ // define an "eventual consistency" timeframe.
+ // TODO: Figure out if it's actually necessary to separately validate whether the entities have
+ // not changed, if we plan to include the compare-and-swap in the helper method that updates the
+ // grantRecordsVersions already.
+
// done, return the new grant record
return grantRecord;
}
@@ -368,7 +309,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
*/
private void revokeGrantRecord(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
+ @Nonnull BasePersistence ms,
@Nonnull PolarisEntityCore securable,
@Nonnull PolarisEntityCore grantee,
@Nonnull PolarisGrantRecord grantRecord) {
@@ -434,33 +375,49 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// grants have changed, we need to bump-up the grants version
refreshSecurable.setGrantRecordsVersion(refreshSecurable.getGrantRecordsVersion() + 1);
ms.writeEntity(callCtx, refreshSecurable, false, originalRefreshSecurable);
+
+ // TODO: Reorder and/or expose bulk update of both grantRecordsVersions and grant records. In
+ // the meantime, cache can be disabled or configured with a short enough expiry time to
+ // define an "eventual consistency" timeframe.
}
- /**
- * Create a new catalog. This not only creates the new catalog entity but also the initial admin
- * role required to admin this catalog.
- *
- * @param callCtx call context
- * @param ms meta store in read/write mode
- * @param catalog the catalog entity to create
- * @param integration the storage integration that should be attached to the catalog. If null, do
- * nothing, otherwise persist the integration.
- * @param principalRoles once the catalog has been created, list of principal roles to grant its
- * catalog_admin role to. If no principal role is specified, we will grant the catalog_admin
- * role of the newly created catalog to the service admin role.
- * @return the catalog we just created and its associated admin catalog role or error if we failed
- * to
- */
- private @Nonnull CreateCatalogResult createCatalog(
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull CreateCatalogResult createCatalog(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nonnull PolarisBaseEntity catalog,
- @Nullable PolarisStorageIntegration<?> integration,
@Nonnull List<PolarisEntityCore> principalRoles) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
+
// validate input
callCtx.getDiagServices().checkNotNull(catalog, "unexpected_null_catalog");
+ Map<String, String> internalProp = getInternalPropertyMap(callCtx, catalog);
+ String integrationIdentifierOrId =
+ internalProp.get(PolarisEntityConstants.getStorageIntegrationIdentifierPropertyName());
+ String storageConfigInfoStr =
+ internalProp.get(PolarisEntityConstants.getStorageConfigInfoPropertyName());
+ PolarisStorageIntegration<?> integration;
+ // storageConfigInfo's presence is needed to create a storage integration
+ // and the catalog should not have an internal property of storage identifier or id yet
+ if (storageConfigInfoStr != null && integrationIdentifierOrId == null) {
+ integration =
+ ((IntegrationPersistence) ms)
+ .createStorageIntegration(
+ callCtx,
+ catalog.getCatalogId(),
+ catalog.getId(),
+ PolarisStorageConfigurationInfo.deserialize(
+ callCtx.getDiagServices(), storageConfigInfoStr));
+ } else {
+ integration = null;
+ }
+
// check if that catalog has already been created
+ // This can be done safely as a separate atomic operation before trying to create the catalog
+ // because same-id idempotent-retry collisions of this sort are necessarily sequential, so
+ // there is no concurrency conflict for something else creating a catalog of this same id.
PolarisBaseEntity refreshCatalog =
ms.lookupEntity(callCtx, catalog.getCatalogId(), catalog.getId(), catalog.getTypeCode());
@@ -493,24 +450,19 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// done, return the existing catalog
return new CreateCatalogResult(refreshCatalog, catalogAdminRole);
}
+ ((IntegrationPersistence) ms).persistStorageIntegrationIfNeeded(callCtx, catalog, integration);
- // check that a catalog with the same name does not exist already
- // if it exists, this is an error, the client should retry
- if (ms.lookupEntityIdAndSubTypeByName(
- callCtx,
- PolarisEntityConstants.getNullId(),
- PolarisEntityConstants.getRootEntityId(),
- PolarisEntityType.CATALOG.getCode(),
- catalog.getName())
- != null) {
+ // now create and persist new catalog entity
+ EntityResult lowLevelResult = this.persistNewEntity(callCtx, ms, catalog);
+ if (lowLevelResult.getReturnStatus() == BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS) {
+ // TODO: Garbage-collection should include integrations, and anything else created before
+ // this if the server crashes before being able to do this cleanup.
+ // TODO: Perform best-effort cleanup. For now, none of the codebase apparently cleans
+ // up storage integrations "if needed", but also the default impls don't create any
+ // storage integrations "if needed" either.
return new CreateCatalogResult(BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS, null);
}
- ms.persistStorageIntegrationIfNeeded(callCtx, catalog, integration);
-
- // now create and persist new catalog entity
- this.persistNewEntity(callCtx, ms, catalog);
-
// create the catalog admin role for this new catalog
long adminRoleId = ms.generateNewId(callCtx);
PolarisBaseEntity adminRole =
@@ -564,18 +516,24 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
}
}
+ // TODO: Reorder and/or expose bulk update/create of new entities and grant records. In the
+ // meantime, if a server crashes halfway through catalog creation, it might be in a partially
+ // initialized state. In such a case, the correct action is to simply delete the catalog
+ // and recreate it -- SERVICE_MANAGE_ACCESS already possesses CATALOG_DROP at the
+ // root-container level even if it requires the grants in here to have CATALOG_MANAGE_ACCESS
+ // or CATALOG_MANAGE_METADATA, so no one can use the catalog if this initialization is
+ // incomplete, but the won't be "stuck" orphaned since the service admin can still delete the
+ // catalog.
+
// success, return the two entities
return new CreateCatalogResult(catalog, adminRole);
}
- /**
- * Bootstrap Polaris catalog service
- *
- * @param callCtx call context
- * @param ms meta store in read/write mode
- */
- private void bootstrapPolarisService(
- @Nonnull PolarisCallContext callCtx, @Nonnull TransactionalPersistence ms) {
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull BaseResult bootstrapPolarisService(@Nonnull PolarisCallContext callCtx) {
+ // get meta store we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// Create a root container entity that can represent the securable for any top-level grants.
PolarisBaseEntity rootContainer =
@@ -602,7 +560,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
PolarisEntityConstants.getRootPrincipalName());
// create this principal
- this.createPrincipal(callCtx, ms, rootPrincipal);
+ this.createPrincipal(callCtx, rootPrincipal);
// now create the account admin principal role
long serviceAdminPrincipalRoleId = ms.generateNewId(callCtx);
@@ -631,16 +589,10 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
rootContainer,
serviceAdminPrincipalRole,
PolarisPrivilege.SERVICE_MANAGE_ACCESS);
- }
- /** {@inheritDoc} */
- @Override
- public @Nonnull BaseResult bootstrapPolarisService(@Nonnull PolarisCallContext callCtx) {
- // get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // run operation in a read/write transaction
- ms.runActionInTransaction(callCtx, () -> this.bootstrapPolarisService(callCtx, ms));
+ // TODO: Make idempotent by being able to continue where it left off for the context's realm.
+ // In the meantime, if a realm was only partially initialized before the server crashed,
+ // it's fine to purge the realm and retry the bootstrap.
// all good
return new BaseResult(BaseResult.ReturnStatus.SUCCESS);
@@ -649,44 +601,36 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Override
public @Nonnull BaseResult purge(@Nonnull PolarisCallContext callCtx) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
- // run operation in a read/write transaction
LOGGER.warn("Deleting all metadata in the metastore...");
- ms.runActionInTransaction(callCtx, () -> ms.deleteAll(callCtx));
+ ms.deleteAll(callCtx);
LOGGER.warn("Finished deleting all metadata in the metastore");
// all good
return new BaseResult(BaseResult.ReturnStatus.SUCCESS);
}
- /**
- * See {@link #readEntityByName(PolarisCallContext, List, PolarisEntityType, PolarisEntitySubType,
- * String)}
- */
- private @Nonnull EntityResult readEntityByName(
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull EntityResult readEntityByName(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisEntityType entityType,
@Nonnull PolarisEntitySubType entitySubType,
@Nonnull String name) {
- // first resolve again the catalogPath to that entity
- PolarisEntityResolver resolver = new PolarisEntityResolver(callCtx, ms, catalogPath);
-
- // return if we failed to resolve
- if (resolver.isFailure()) {
- return new EntityResult(BaseResult.ReturnStatus.CATALOG_PATH_CANNOT_BE_RESOLVED, null);
- }
+ BasePersistence ms = callCtx.getMetaStore();
// now looking the entity by name
+ // TODO: Clean up shared logic for catalogId/parentId
+ long catalogId =
+ catalogPath == null || catalogPath.size() == 0 ? 0l : catalogPath.get(0).getId();
+ long parentId =
+ catalogPath == null || catalogPath.size() == 0
+ ? 0l
+ : catalogPath.get(catalogPath.size() - 1).getId();
PolarisBaseEntity entity =
- ms.lookupEntityByName(
- callCtx,
- resolver.getCatalogIdOrNull(),
- resolver.getParentId(),
- entityType.getCode(),
- name);
+ ms.lookupEntityByName(callCtx, catalogId, parentId, entityType.getCode(), name);
// if found, check if subType really matches
if (entity != null
@@ -695,6 +639,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
entity = null;
}
+ // TODO: Use post-validation to enforce consistent view against catalogPath. In the
+ // meantime, happens-before ordering semantics aren't guaranteed during high-concurrency
+ // race conditions, such as first revoking a grant on a namespace before adding sensitive
+ // data to a table; but the window of inconsistency is only the duration of a single
+ // in-flight request (the cache-based resolution follows a different path entirely).
+
// success, return what we found
return (entity == null)
? new EntityResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null)
@@ -703,40 +653,24 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull EntityResult readEntityByName(
- @Nonnull PolarisCallContext callCtx,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityType entityType,
- @Nonnull PolarisEntitySubType entitySubType,
- @Nonnull String name) {
- // get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // run operation in a read/write transaction
- return ms.runInReadTransaction(
- callCtx, () -> readEntityByName(callCtx, ms, catalogPath, entityType, entitySubType, name));
- }
-
- /**
- * See {@link #listEntities(PolarisCallContext, List, PolarisEntityType, PolarisEntitySubType)}
- */
- private @Nonnull ListEntitiesResult listEntities(
+ public @Nonnull ListEntitiesResult listEntities(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisEntityType entityType,
@Nonnull PolarisEntitySubType entitySubType) {
- // first resolve again the catalogPath to that entity
- PolarisEntityResolver resolver = new PolarisEntityResolver(callCtx, ms, catalogPath);
-
- // return if we failed to resolve
- if (resolver.isFailure()) {
- return new ListEntitiesResult(BaseResult.ReturnStatus.CATALOG_PATH_CANNOT_BE_RESOLVED, null);
- }
+ // get meta store we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// return list of active entities
+ // TODO: Clean up shared logic for catalogId/parentId
+ long catalogId =
+ catalogPath == null || catalogPath.size() == 0 ? 0l : catalogPath.get(0).getId();
+ long parentId =
+ catalogPath == null || catalogPath.size() == 0
+ ? 0l
+ : catalogPath.get(catalogPath.size() - 1).getId();
List<EntityNameLookupRecord> toreturnList =
- ms.listEntities(callCtx, resolver.getCatalogIdOrNull(), resolver.getParentId(), entityType);
+ ms.listEntities(callCtx, catalogId, parentId, entityType);
// prune the returned list with only entities matching the entity subtype
if (entitySubType != PolarisEntitySubType.ANY_SUBTYPE) {
@@ -746,79 +680,23 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
.collect(Collectors.toList());
}
+ // TODO: Use post-validation to enforce consistent view against catalogPath. In the
+ // meantime, happens-before ordering semantics aren't guaranteed during high-concurrency
+ // race conditions, such as first revoking a grant on a namespace before adding a table
+ // with sensitive data; but the window of inconsistency is only the duration of a single
+ // in-flight request (the cache-based resolution follows a different path entirely).
+
// done
return new ListEntitiesResult(toreturnList);
}
/** {@inheritDoc} */
@Override
- public @Nonnull ListEntitiesResult listEntities(
- @Nonnull PolarisCallContext callCtx,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityType entityType,
- @Nonnull PolarisEntitySubType entitySubType) {
- // get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // run operation in a read transaction
- return ms.runInReadTransaction(
- callCtx, () -> listEntities(callCtx, ms, catalogPath, entityType, entitySubType));
- }
-
- /** {@inheritDoc} */
- @Override
- public @Nonnull GenerateEntityIdResult generateNewEntityId(@Nonnull PolarisCallContext callCtx) {
- // get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- return new GenerateEntityIdResult(ms.generateNewId(callCtx));
- }
-
- /**
- * Given the internal property as a map of key/value pairs, serialize it to a String
- *
- * @param properties a map of key/value pairs
- * @return a String, the JSON representation of the map
- */
- public String serializeProperties(PolarisCallContext callCtx, Map<String, String> properties) {
-
- String jsonString = null;
- try {
- // Deserialize the JSON string to a Map<String, String>
- jsonString = MAPPER.writeValueAsString(properties);
- } catch (JsonProcessingException ex) {
- callCtx.getDiagServices().fail("got_json_processing_exception", "ex={}", ex);
- }
-
- return jsonString;
- }
-
- /**
- * Given the serialized properties, deserialize those to a {@code Map<String, String>}
- *
- * @param properties a JSON string representing the set of properties
- * @return a Map of string
- */
- public Map<String, String> deserializeProperties(PolarisCallContext callCtx, String properties) {
-
- Map<String, String> retProperties = null;
- try {
- // Deserialize the JSON string to a Map<String, String>
- retProperties = MAPPER.readValue(properties, new TypeReference<>() {});
- } catch (JsonMappingException ex) {
- callCtx.getDiagServices().fail("got_json_mapping_exception", "ex={}", ex);
- } catch (JsonProcessingException ex) {
- callCtx.getDiagServices().fail("got_json_processing_exception", "ex={}", ex);
- }
-
- return retProperties;
- }
+ public @Nonnull CreatePrincipalResult createPrincipal(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity principal) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
- /** {@link #createPrincipal(PolarisCallContext, PolarisBaseEntity)} */
- private @Nonnull CreatePrincipalResult createPrincipal(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nonnull PolarisBaseEntity principal) {
// validate input
callCtx.getDiagServices().checkNotNull(principal, "unexpected_null_principal");
@@ -828,6 +706,9 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
callCtx, principal.getCatalogId(), principal.getId(), principal.getTypeCode());
// if found, probably a retry, simply return the previously created principal
+ // This can be done safely as a separate atomic operation before trying to create the principal
+ // because same-id idempotent-retry collisions of this sort are necessarily sequential, so
+ // there is no concurrency conflict for something else creating a principal of this same id.
if (refreshPrincipal != null) {
// if found, ensure it is indeed a principal
callCtx
@@ -863,7 +744,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
refreshPrincipal.getInternalProperties());
// get the main and secondary secrets for that client
- PolarisPrincipalSecrets principalSecrets = ms.loadPrincipalSecrets(callCtx, clientId);
+ PolarisPrincipalSecrets principalSecrets =
+ ((IntegrationPersistence) ms).loadPrincipalSecrets(callCtx, clientId);
// should not be null
callCtx
@@ -879,21 +761,10 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
return new CreatePrincipalResult(refreshPrincipal, principalSecrets);
}
- // check that a principal with the same name does not exist already
- // if it exists, this is an error, the client should retry
- if (ms.lookupEntityIdAndSubTypeByName(
- callCtx,
- PolarisEntityConstants.getNullId(),
- PolarisEntityConstants.getRootEntityId(),
- PolarisEntityType.PRINCIPAL.getCode(),
- principal.getName())
- != null) {
- return new CreatePrincipalResult(BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS, null);
- }
-
// generate new secrets for this principal
PolarisPrincipalSecrets principalSecrets =
- ms.generateNewPrincipalSecrets(callCtx, principal.getName(), principal.getId());
+ ((IntegrationPersistence) ms)
+ .generateNewPrincipalSecrets(callCtx, principal.getName(), principal.getId());
// generate properties
Map<String, String> internalProperties = getInternalPropertyMap(callCtx, principal);
@@ -904,63 +775,53 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
principal.setInternalProperties(this.serializeProperties(callCtx, internalProperties));
// now create and persist new catalog entity
- this.persistNewEntity(callCtx, ms, principal);
+ EntityResult lowLevelResult = this.persistNewEntity(callCtx, ms, principal);
+ if (lowLevelResult.getReturnStatus() == BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS) {
+ // Name collision; do best-effort cleanup of the new principalSecrets we created
+ // TODO: Garbage-collection should include principal secrets if the server crashes
+ // before being able to do this cleanup.
+ ((IntegrationPersistence) ms)
+ .deletePrincipalSecrets(
+ callCtx, principalSecrets.getPrincipalClientId(), principal.getId());
+ return new CreatePrincipalResult(BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS, null);
+ }
// success, return the two entities
return new CreatePrincipalResult(principal, principalSecrets);
}
- /** {@inheritDoc} */
- @Override
- public @Nonnull CreatePrincipalResult createPrincipal(
- @Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity principal) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(callCtx, () -> this.createPrincipal(callCtx, ms, principal));
- }
-
- /** See {@link #loadPrincipalSecrets(PolarisCallContext, String)} */
- private @Nullable PolarisPrincipalSecrets loadPrincipalSecrets(
- @Nonnull PolarisCallContext callCtx, TransactionalPersistence ms, @Nonnull String clientId) {
- return ms.loadPrincipalSecrets(callCtx, clientId);
- }
-
/** {@inheritDoc} */
@Override
public @Nonnull PrincipalSecretsResult loadPrincipalSecrets(
@Nonnull PolarisCallContext callCtx, @Nonnull String clientId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
- // need to run inside a read/write transaction
PolarisPrincipalSecrets secrets =
- ms.runInTransaction(callCtx, () -> this.loadPrincipalSecrets(callCtx, ms, clientId));
+ ((IntegrationPersistence) ms).loadPrincipalSecrets(callCtx, clientId);
return (secrets == null)
? new PrincipalSecretsResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null)
: new PrincipalSecretsResult(secrets);
}
- /** See {@link #} */
- private @Nullable PolarisPrincipalSecrets rotatePrincipalSecrets(
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull PrincipalSecretsResult rotatePrincipalSecrets(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nonnull String clientId,
long principalId,
boolean reset,
@Nonnull String oldSecretHash) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
+
// if not found, the principal must have been dropped
EntityResult loadEntityResult =
loadEntity(
- callCtx,
- ms,
- PolarisEntityConstants.getNullId(),
- principalId,
- PolarisEntityType.PRINCIPAL.getCode());
+ callCtx, PolarisEntityConstants.getNullId(), principalId, PolarisEntityType.PRINCIPAL);
if (loadEntityResult.getReturnStatus() != BaseResult.ReturnStatus.SUCCESS) {
- return null;
+ return new PrincipalSecretsResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null);
}
PolarisBaseEntity principal = loadEntityResult.getEntity();
@@ -976,7 +837,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)
!= null;
PolarisPrincipalSecrets secrets =
- ms.rotatePrincipalSecrets(callCtx, clientId, principalId, doReset, oldSecretHash);
+ ((IntegrationPersistence) ms)
+ .rotatePrincipalSecrets(callCtx, clientId, principalId, doReset, oldSecretHash);
if (reset
&& !internalProps.containsKey(
@@ -995,27 +857,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
principal.setEntityVersion(principal.getEntityVersion() + 1);
ms.writeEntity(callCtx, principal, true, originalPrincipal);
}
- return secrets;
- }
- /** {@inheritDoc} */
- @Override
- public @Nonnull PrincipalSecretsResult rotatePrincipalSecrets(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull String clientId,
- long principalId,
- boolean reset,
- @Nonnull String oldSecretHash) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- PolarisPrincipalSecrets secrets =
- ms.runInTransaction(
- callCtx,
- () ->
- this.rotatePrincipalSecrets(
- callCtx, ms, clientId, principalId, reset, oldSecretHash));
+ // TODO: Rethink the atomicity of the relationship between principalSecrets and principal
return (secrets == null)
? new PrincipalSecretsResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null)
@@ -1024,43 +867,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull CreateCatalogResult createCatalog(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull PolarisBaseEntity catalog,
- @Nonnull List<PolarisEntityCore> principalRoles) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- Map<String, String> internalProp = getInternalPropertyMap(callCtx, catalog);
- String integrationIdentifierOrId =
- internalProp.get(PolarisEntityConstants.getStorageIntegrationIdentifierPropertyName());
- String storageConfigInfoStr =
- internalProp.get(PolarisEntityConstants.getStorageConfigInfoPropertyName());
- PolarisStorageIntegration<?> integration;
- // storageConfigInfo's presence is needed to create a storage integration
- // and the catalog should not have an internal property of storage identifier or id yet
- if (storageConfigInfoStr != null && integrationIdentifierOrId == null) {
- integration =
- ms.createStorageIntegration(
- callCtx,
- catalog.getCatalogId(),
- catalog.getId(),
- PolarisStorageConfigurationInfo.deserialize(
- callCtx.getDiagServices(), storageConfigInfoStr));
- } else {
- integration = null;
- }
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx, () -> this.createCatalog(callCtx, ms, catalog, integration, principalRoles));
- }
-
- /** {@link #createEntityIfNotExists(PolarisCallContext, List, PolarisBaseEntity)} */
- private @Nonnull EntityResult createEntityIfNotExists(
+ public @Nonnull EntityResult createEntityIfNotExists(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisBaseEntity entity) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// entity cannot be null
callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
@@ -1068,54 +880,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// entity name must be specified
callCtx.getDiagServices().checkNotNull(entity.getName(), "unexpected_null_entity_name");
- // first, check if the entity has already been created, in which case we will simply return it
- PolarisBaseEntity entityFound =
- ms.lookupEntity(callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
- if (entityFound != null) {
- // probably the client retried, simply return it
- return new EntityResult(entityFound);
- }
-
- // first resolve again the catalogPath
- PolarisEntityResolver resolver = new PolarisEntityResolver(callCtx, ms, catalogPath);
-
- // return if we failed to resolve
- if (resolver.isFailure()) {
- return new EntityResult(BaseResult.ReturnStatus.CATALOG_PATH_CANNOT_BE_RESOLVED, null);
- }
-
- // check if an entity does not already exist with the same name. If true, this is an error
- EntityNameLookupRecord entityActiveRecord =
- ms.lookupEntityIdAndSubTypeByName(
- callCtx,
- entity.getCatalogId(),
- entity.getParentId(),
- entity.getType().getCode(),
- entity.getName());
- if (entityActiveRecord != null) {
- return new EntityResult(
- BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS, entityActiveRecord.getSubTypeCode());
- }
-
- // persist that new entity
- this.persistNewEntity(callCtx, ms, entity);
-
- // done, return that newly created entity
- return new EntityResult(entity);
- }
-
- /** {@inheritDoc} */
- @Override
- public @Nonnull EntityResult createEntityIfNotExists(
- @Nonnull PolarisCallContext callCtx,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisBaseEntity entity) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx, () -> this.createEntityIfNotExists(callCtx, ms, catalogPath, entity));
+ // TODO: Use post-validation to enforce consistent view against catalogPath. In the
+ // meantime, happens-before ordering semantics aren't guaranteed during high-concurrency
+ // race conditions, such as first revoking a grant on a namespace before adding sensitive
+ // data to a table; but the window of inconsistency is only the duration of a single
+ // in-flight request (the cache-based resolution follows a different path entirely).
+ return this.persistNewEntity(callCtx, ms, entity);
}
@Override
@@ -1124,113 +894,99 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull List<? extends PolarisBaseEntity> entities) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
+ List<PolarisBaseEntity> createdEntities = new ArrayList<>(entities.size());
+ for (PolarisBaseEntity entity : entities) {
+ PolarisBaseEntity entityToCreate =
+ prepareToPersistNewEntity(callCtx, ms, new PolarisBaseEntity(entity));
+ createdEntities.add(entityToCreate);
+ }
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx,
- () -> {
- List<PolarisBaseEntity> createdEntities = new ArrayList<>(entities.size());
- for (PolarisBaseEntity entity : entities) {
- EntityResult entityCreateResult =
- createEntityIfNotExists(callCtx, ms, catalogPath, entity);
- // abort everything if error
- if (entityCreateResult.getReturnStatus() != BaseResult.ReturnStatus.SUCCESS) {
- ms.rollback();
- return new EntitiesResult(
- entityCreateResult.getReturnStatus(), entityCreateResult.getExtraInformation());
- }
- createdEntities.add(entityCreateResult.getEntity());
- }
- return new EntitiesResult(createdEntities);
- });
+ try {
+ ms.writeEntities(callCtx, createdEntities, null);
+ // TODO: Use post-validation to enforce consistent view against catalogPath. In the
+ // meantime, happens-before ordering semantics aren't guaranteed during high-concurrency
+ // race conditions, such as first revoking a grant on a namespace before adding sensitive
+ // data to a table; but the window of inconsistency is only the duration of a single
+ // in-flight request (the cache-based resolution follows a different path entirely).
+ } catch (EntityAlreadyExistsException e) {
+ return new EntitiesResult(
+ BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS,
+ String.format(
+ "Existing entity id: '%s', type %s subtype %s",
+ e.getExistingEntity().getId(),
+ e.getExistingEntity().getTypeCode(),
+ e.getExistingEntity().getSubTypeCode()));
+ }
+
+ return new EntitiesResult(createdEntities);
}
- /**
- * See {@link #updateEntityPropertiesIfNotChanged(PolarisCallContext, List, PolarisBaseEntity)}
- */
- private @Nonnull EntityResult updateEntityPropertiesIfNotChanged(
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull EntityResult updateEntityPropertiesIfNotChanged(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisBaseEntity entity) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
+
// entity cannot be null
callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
- // re-resolve everything including that entity
- PolarisEntityResolver resolver = new PolarisEntityResolver(callCtx, ms, catalogPath, entity);
-
- // if resolution failed, return false
- if (resolver.isFailure()) {
- return new EntityResult(BaseResult.ReturnStatus.CATALOG_PATH_CANNOT_BE_RESOLVED, null);
- }
-
- // lookup the entity, cannot be null
- PolarisBaseEntity entityRefreshed =
- ms.lookupEntity(callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
- callCtx
- .getDiagServices()
- .checkNotNull(entityRefreshed, "unexpected_entity_not_found", "entity={}", entity);
-
- // check that the version of the entity has not changed at all to avoid concurrent updates
- if (entityRefreshed.getEntityVersion() != entity.getEntityVersion()) {
- return new EntityResult(BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, null);
- }
-
- PolarisBaseEntity originalEntity = new PolarisBaseEntity(entityRefreshed);
-
- // update the two properties
- entityRefreshed.setInternalProperties(entity.getInternalProperties());
- entityRefreshed.setProperties(entity.getProperties());
+ // Create a copy of the input to use as the "original".
+ // TODO: No longer make a copy once we have immutable PolarisBaseEntity everywhere.
+ PolarisBaseEntity originalEntity = new PolarisBaseEntity(entity);
// persist this entity after changing it. This will update the version and update the last
// updated time. Because the entity version is changed, we will update the change tracking table
- PolarisBaseEntity persistedEntity =
- this.persistEntityAfterChange(callCtx, ms, entityRefreshed, false, originalEntity);
- return new EntityResult(persistedEntity);
+ try {
+ PolarisBaseEntity persistedEntity =
+ this.persistEntityAfterChange(
+ callCtx, ms, new PolarisBaseEntity(entity), false, originalEntity);
+
+ // TODO: Revalidate parent-path *after* performing update to fulfill the semantic of returning
+ // a NotFoundException if some element of the parent path was concurrently deleted.
+ return new EntityResult(persistedEntity);
+ } catch (RetryOnConcurrencyException e) {
+ return new EntityResult(
+ BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, e.getMessage());
+ }
}
/** {@inheritDoc} */
@Override
- public @Nonnull EntityResult updateEntityPropertiesIfNotChanged(
- @Nonnull PolarisCallContext callCtx,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisBaseEntity entity) {
+ public @Nonnull EntitiesResult updateEntitiesPropertiesIfNotChanged(
+ @Nonnull PolarisCallContext callCtx, @Nonnull List<EntityWithPath> entities) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx, () -> this.updateEntityPropertiesIfNotChanged(callCtx, ms, catalogPath, entity));
- }
+ BasePersistence ms = callCtx.getMetaStore();
- /** See {@link #updateEntitiesPropertiesIfNotChanged(PolarisCallContext, List)} */
- private @Nonnull EntitiesResult updateEntitiesPropertiesIfNotChanged(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nonnull List<EntityWithPath> entities) {
// ensure that the entities list is not null
callCtx.getDiagServices().checkNotNull(entities, "unexpected_null_entities");
- // list of all updated entities
List<PolarisBaseEntity> updatedEntities = new ArrayList<>(entities.size());
+ List<PolarisBaseEntity> originalEntities = new ArrayList<>(entities.size());
- // iterate over the list and update each, one at a time
+ // Collect list of updatedEntities
for (EntityWithPath entityWithPath : entities) {
- // update that entity, abort if it fails
- EntityResult updatedEntityResult =
- this.updateEntityPropertiesIfNotChanged(
- callCtx, ms, entityWithPath.getCatalogPath(), entityWithPath.getEntity());
-
- // if failed, rollback and return the last error
- if (updatedEntityResult.getReturnStatus() != BaseResult.ReturnStatus.SUCCESS) {
- ms.rollback();
- return new EntitiesResult(
- updatedEntityResult.getReturnStatus(), updatedEntityResult.getExtraInformation());
- }
+ // Create a copy of the input to use as the "original".
+ PolarisBaseEntity originalEntity = new PolarisBaseEntity(entityWithPath.getEntity());
+ PolarisBaseEntity updatedEntity =
+ prepareToPersistEntityAfterChange(
+ callCtx,
+ ms,
+ new PolarisBaseEntity(entityWithPath.getEntity()),
+ false,
+ originalEntity);
+ originalEntities.add(originalEntity);
+ updatedEntities.add(updatedEntity);
+ }
- // one more was updated
- updatedEntities.add(updatedEntityResult.getEntity());
+ try {
+ ms.writeEntities(callCtx, updatedEntities, originalEntities);
+ } catch (RetryOnConcurrencyException e) {
+ return new EntitiesResult(
+ BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, e.getMessage());
}
// good, all success
@@ -1239,27 +995,14 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull EntitiesResult updateEntitiesPropertiesIfNotChanged(
- @Nonnull PolarisCallContext callCtx, @Nonnull List<EntityWithPath> entities) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx, () -> this.updateEntitiesPropertiesIfNotChanged(callCtx, ms, entities));
- }
-
- /**
- * See {@link PolarisMetaStoreManager#renameEntity(PolarisCallContext, List, PolarisEntityCore,
- * List, PolarisEntity)}
- */
- private @Nonnull EntityResult renameEntity(
+ public @Nonnull EntityResult renameEntity(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisEntityCore entityToRename,
@Nullable List<PolarisEntityCore> newCatalogPath,
- @Nonnull PolarisBaseEntity renamedEntity) {
+ @Nonnull PolarisEntity renamedEntity) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// entity and new name cannot be null
callCtx.getDiagServices().checkNotNull(entityToRename, "unexpected_null_entityToRename");
@@ -1278,15 +1021,6 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
newCatalogPath = catalogPath;
}
- // re-resolve everything including that entity
- PolarisEntityResolver resolver =
- new PolarisEntityResolver(callCtx, ms, catalogPath, entityToRename);
-
- // if resolution failed, return false
- if (resolver.isFailure()) {
- return new EntityResult(BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, null);
- }
-
// find the entity to rename
PolarisBaseEntity refreshEntityToRename =
ms.lookupEntity(
@@ -1311,23 +1045,19 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
return new EntityResult(BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RENAMED, null);
}
- // re-resolve the new catalog path if this entity is going to be moved
- if (newCatalogPath != null) {
- resolver = new PolarisEntityResolver(callCtx, ms, newCatalogPath);
-
- // if resolution failed, return false
- if (resolver.isFailure()) {
- return new EntityResult(BaseResult.ReturnStatus.CATALOG_PATH_CANNOT_BE_RESOLVED, null);
- }
- }
-
// ensure that nothing exists where we create that entity
// if this entity already exists, this is an error
+ long catalogId =
+ newCatalogPath == null || newCatalogPath.size() == 0 ? 0l : newCatalogPath.get(0).getId();
+ long parentId =
+ newCatalogPath == null || newCatalogPath.size() == 0
+ ? 0l
+ : newCatalogPath.get(newCatalogPath.size() - 1).getId();
EntityNameLookupRecord entityActiveRecord =
ms.lookupEntityIdAndSubTypeByName(
callCtx,
- resolver.getCatalogIdOrNull(),
- resolver.getParentId(),
+ catalogId,
+ parentId,
refreshEntityToRename.getTypeCode(),
renamedEntity.getName());
if (entityActiveRecord != null) {
@@ -1346,7 +1076,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// re-parent if a new catalog path was specified
if (newCatalogPath != null) {
- refreshEntityToRename.setParentId(resolver.getParentId());
+ refreshEntityToRename.setParentId(parentId);
}
// persist the entity after change. This will update the lastUpdateTimestamp and bump up the
@@ -1354,51 +1084,31 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// lookups if applicable
PolarisBaseEntity renamedEntityToReturn =
this.persistEntityAfterChange(callCtx, ms, refreshEntityToRename, true, originalEntity);
+
+ // TODO: Use post-validation of source and destination parent paths and/or update the
+ // BasePersistence interface to support conditions on entities *other* than the main
+ // entity/entities being changed ina a single atomic operation
return new EntityResult(renamedEntityToReturn);
}
/** {@inheritDoc} */
@Override
- public @Nonnull EntityResult renameEntity(
- @Nonnull PolarisCallContext callCtx,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityCore entityToRename,
- @Nullable List<PolarisEntityCore> newCatalogPath,
- @Nonnull PolarisEntity renamedEntity) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx,
- () ->
- this.renameEntity(
- callCtx, ms, catalogPath, entityToRename, newCatalogPath, renamedEntity));
- }
-
- /**
- * See
- *
- * <p>{@link #dropEntityIfExists(PolarisCallContext, List, PolarisEntityCore, Map, boolean)}
- */
- private @Nonnull DropEntityResult dropEntityIfExists(
+ public @Nonnull DropEntityResult dropEntityIfExists(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisEntityCore entityToDrop,
@Nullable Map<String, String> cleanupProperties,
boolean cleanup) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
+
// entity cannot be null
callCtx.getDiagServices().checkNotNull(entityToDrop, "unexpected_null_entity");
- // re-resolve everything including that entity
- PolarisEntityResolver resolver =
- new PolarisEntityResolver(callCtx, ms, catalogPath, entityToDrop);
-
- // if resolution failed, return false
- if (resolver.isFailure()) {
- return new DropEntityResult(BaseResult.ReturnStatus.CATALOG_PATH_CANNOT_BE_RESOLVED, null);
- }
+ // TODO: Either document allowance of dropping entity concurrently with potentially-impacting
+ // changes in the parent path (e.g. race-condition revocation of grants on parent) or
+ // add a way for BasePersistence methods to validate conditions on catalogPath entities
+ // atomically with the drop.
// first find the entity to drop
PolarisBaseEntity refreshEntityToDrop =
@@ -1422,6 +1132,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long catalogId = refreshEntityToDrop.getId();
// if not all namespaces are dropped, we cannot drop this catalog
+ // TODO: Come up with atomic solution to blocking dropping of container entities that
+ // have children; one option is reference-counting if all child creation/drop operations
+ // become two-entity bulk conditional updates that also update the refcount on the parent
+ // if not changed concurrently (else retry). In the meantime, there's a window of time
+ // where dropping a namespace or container is effectively "recursive" in deleting its
+ // children as well if those child entities were created within the short window of
+ // the race condition.
if (ms.hasChildren(callCtx, PolarisEntityType.NAMESPACE, catalogId, catalogId)) {
return new DropEntityResult(BaseResult.ReturnStatus.NAMESPACE_NOT_EMPTY, null);
}
@@ -1464,7 +1181,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
if (cleanup) {
PolarisBaseEntity taskEntity =
new PolarisEntity.Builder()
- .setId(generateNewEntityId(callCtx).getId())
+ .setId(ms.generateNewId(callCtx))
.setCatalogId(0L)
.setName("entityCleanup_" + entityToDrop.getId())
.setType(PolarisEntityType.TASK)
@@ -1482,7 +1199,11 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
taskEntity.setInternalProperties(
PolarisObjectMapperUtil.serializeProperties(callCtx, cleanupProperties));
}
- createEntityIfNotExists(callCtx, ms, null, taskEntity);
+ // TODO: Add a way to create the task entities atomically with dropping the entity;
+ // in the meantime, if the server fails partway through a dropEntity, it's possible that
+ // the entity is dropped but we don't have any persisted task records that will carry
+ // out the cleanup.
+ createEntityIfNotExists(callCtx, null, taskEntity);
return new DropEntityResult(taskEntity.getId());
}
@@ -1492,127 +1213,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull DropEntityResult dropEntityIfExists(
- @Nonnull PolarisCallContext callCtx,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityCore entityToDrop,
- @Nullable Map<String, String> cleanupProperties,
- boolean cleanup) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx,
- () ->
- this.dropEntityIfExists(
- callCtx, ms, catalogPath, entityToDrop, cleanupProperties, cleanup));
- }
-
- /**
- * Resolve the arguments of granting/revoking a usage grant between a role (catalog or principal
- * role) and a grantee (either a principal role or a principal)
- *
- * @param callCtx call context
- * @param ms meta store in read/write mode
- * @param catalog if the role is a catalog role, the caller needs to pass-in the catalog entity
- * which was used to resolve that role. Else null.
- * @param role the role, either a catalog or principal role
- * @param grantee the grantee
- * @return resolver for the specified entities
- */
- private @Nonnull PolarisEntityResolver resolveRoleToGranteeUsageGrant(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nullable PolarisEntityCore catalog,
- @Nonnull PolarisEntityCore role,
- @Nonnull PolarisEntityCore grantee) {
-
- // validate the grantee input
- callCtx.getDiagServices().checkNotNull(grantee, "unexpected_null_grantee");
- callCtx
- .getDiagServices()
- .check(grantee.getType().isGrantee(), "not_a_grantee", "grantee={}", grantee);
-
- // validate role
- callCtx.getDiagServices().checkNotNull(role, "unexpected_null_role");
-
- // role should be a catalog or a principal role
- boolean isCatalogRole = role.getTypeCode() == PolarisEntityType.CATALOG_ROLE.getCode();
- boolean isPrincipalRole = role.getTypeCode() == PolarisEntityType.PRINCIPAL_ROLE.getCode();
- callCtx.getDiagServices().check(isCatalogRole || isPrincipalRole, "not_a_role");
-
- // if the role is a catalog role, ensure a catalog is specified and
- // vice-versa, catalog should be null if the role is a principal role
- callCtx
- .getDiagServices()
- .check(
- (catalog == null && isPrincipalRole) || (catalog != null && isCatalogRole),
- "catalog_mismatch",
- "catalog={} role={}",
- catalog,
- role);
-
- // re-resolve now all these entities
- List<PolarisEntityCore> otherTopLevelEntities = new ArrayList<>(2);
- otherTopLevelEntities.add(role);
- otherTopLevelEntities.add(grantee);
-
- // ensure these entities have not changed
- return new PolarisEntityResolver(
- callCtx, ms, catalog != null ? List.of(catalog) : null, null, otherTopLevelEntities);
- }
-
- /**
- * Helper function to resolve the securable to role grant privilege
- *
- * @param grantee resolved grantee
- * @param catalogPath path to that entity, cannot be null or empty if securable has a catalogId
- * @param securable securable entity, must have been resolved by the client
- * @return a resolver for the role, the catalog path and the securable
- */
- private PolarisEntityResolver resolveSecurableToRoleGrant(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nonnull PolarisEntityCore grantee,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityCore securable) {
- // validate role input
- callCtx.getDiagServices().checkNotNull(grantee, "unexpected_null_grantee");
- callCtx
- .getDiagServices()
- .check(grantee.getType().isGrantee(), "not_grantee_type", "grantee={}", grantee);
-
- // securable must be supplied
- callCtx.getDiagServices().checkNotNull(securable, "unexpected_null_securable");
- if (securable.getCatalogId() > 0) {
- // catalogPath must be supplied if the securable has a catalogId
- callCtx.getDiagServices().checkNotNull(catalogPath, "unexpected_null_catalogPath");
- }
-
- // re-resolve now all these entities
- return new PolarisEntityResolver(callCtx, ms, catalogPath, securable, List.of(grantee));
- }
-
- /**
- * See {@link #grantUsageOnRoleToGrantee(PolarisCallContext, PolarisEntityCore, PolarisEntityCore,
- * PolarisEntityCore)}
- */
- private @Nonnull PrivilegeResult grantUsageOnRoleToGrantee(
+ public @Nonnull PrivilegeResult grantUsageOnRoleToGrantee(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nullable PolarisEntityCore catalog,
@Nonnull PolarisEntityCore role,
@Nonnull PolarisEntityCore grantee) {
-
- // ensure these entities have not changed
- PolarisEntityResolver resolver =
- this.resolveRoleToGranteeUsageGrant(callCtx, ms, catalog, role, grantee);
-
- // if failure to resolve, let the caller know
- if (resolver.isFailure()) {
- return new PrivilegeResult(BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, null);
- }
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// the usage privilege to grant
PolarisPrivilege usagePriv =
@@ -1631,38 +1238,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull PrivilegeResult grantUsageOnRoleToGrantee(
+ public @Nonnull PrivilegeResult revokeUsageOnRoleFromGrantee(
@Nonnull PolarisCallContext callCtx,
@Nullable PolarisEntityCore catalog,
@Nonnull PolarisEntityCore role,
@Nonnull PolarisEntityCore grantee) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx, () -> this.grantUsageOnRoleToGrantee(callCtx, ms, catalog, role, grantee));
- }
-
- /**
- * See {@link #revokeUsageOnRoleFromGrantee(PolarisCallContext, PolarisEntityCore,
- * PolarisEntityCore, PolarisEntityCore)}
- */
- private @Nonnull PrivilegeResult revokeUsageOnRoleFromGrantee(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nullable PolarisEntityCore catalog,
- @Nonnull PolarisEntityCore role,
- @Nonnull PolarisEntityCore grantee) {
-
- // ensure these entities have not changed
- PolarisEntityResolver resolver =
- this.resolveRoleToGranteeUsageGrant(callCtx, ms, catalog, role, grantee);
-
- // if failure to resolve, let the caller know
- if (resolver.isFailure()) {
- return new PrivilegeResult(BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, null);
- }
+ BasePersistence ms = callCtx.getMetaStore();
// the usage privilege to revoke
PolarisPrivilege usagePriv =
@@ -1693,85 +1275,31 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull PrivilegeResult revokeUsageOnRoleFromGrantee(
- @Nonnull PolarisCallContext callCtx,
- @Nullable PolarisEntityCore catalog,
- @Nonnull PolarisEntityCore role,
- @Nonnull PolarisEntityCore grantee) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx, () -> this.revokeUsageOnRoleFromGrantee(callCtx, ms, catalog, role, grantee));
- }
-
- /**
- * See {@link #grantPrivilegeOnSecurableToRole(PolarisCallContext, PolarisEntityCore, List,
- * PolarisEntityCore, PolarisPrivilege)}
- */
- private @Nonnull PrivilegeResult grantPrivilegeOnSecurableToRole(
+ public @Nonnull PrivilegeResult grantPrivilegeOnSecurableToRole(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
@Nonnull PolarisEntityCore grantee,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisEntityCore securable,
- @Nonnull PolarisPrivilege priv) {
-
- // re-resolve now all these entities
- PolarisEntityResolver resolver =
- this.resolveSecurableToRoleGrant(callCtx, ms, grantee, catalogPath, securable);
-
- // if failure to resolve, let the caller know
- if (resolver.isFailure()) {
- return new PrivilegeResult(BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, null);
- }
+ @Nonnull PolarisPrivilege privilege) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// grant specified privilege on this securable to this role and return the grant
PolarisGrantRecord grantRecord =
- this.persistNewGrantRecord(callCtx, ms, securable, grantee, priv);
+ this.persistNewGrantRecord(callCtx, ms, securable, grantee, privilege);
return new PrivilegeResult(grantRecord);
}
/** {@inheritDoc} */
@Override
- public @Nonnull PrivilegeResult grantPrivilegeOnSecurableToRole(
+ public @Nonnull PrivilegeResult revokePrivilegeOnSecurableFromRole(
@Nonnull PolarisCallContext callCtx,
@Nonnull PolarisEntityCore grantee,
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisEntityCore securable,
@Nonnull PolarisPrivilege privilege) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx,
- () ->
- this.grantPrivilegeOnSecurableToRole(
- callCtx, ms, grantee, catalogPath, securable, privilege));
- }
-
- /**
- * See {@link #revokePrivilegeOnSecurableFromRole(PolarisCallContext, PolarisEntityCore, List,
- * PolarisEntityCore, PolarisPrivilege)}
- */
- private @Nonnull PrivilegeResult revokePrivilegeOnSecurableFromRole(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nonnull PolarisEntityCore grantee,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityCore securable,
- @Nonnull PolarisPrivilege priv) {
-
- // re-resolve now all these entities
- PolarisEntityResolver resolver =
- this.resolveSecurableToRoleGrant(callCtx, ms, grantee, catalogPath, securable);
-
- // if failure to resolve, let the caller know
- if (resolver.isFailure()) {
- return new PrivilegeResult(BaseResult.ReturnStatus.ENTITY_CANNOT_BE_RESOLVED, null);
- }
+ BasePersistence ms = callCtx.getMetaStore();
// lookup the grants records to find this grant
PolarisGrantRecord grantRecord =
@@ -1781,7 +1309,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
securable.getId(),
grantee.getCatalogId(),
grantee.getId(),
- priv.getCode());
+ privilege.getCode());
// the grant does not exist, nothing to do really
if (grantRecord == null) {
@@ -1797,29 +1325,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull PrivilegeResult revokePrivilegeOnSecurableFromRole(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull PolarisEntityCore grantee,
- @Nullable List<PolarisEntityCore> catalogPath,
- @Nonnull PolarisEntityCore securable,
- @Nonnull PolarisPrivilege privilege) {
+ public @Nonnull LoadGrantsResult loadGrantsOnSecurable(
+ @Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
- // need to run inside a read/write transaction
- return ms.runInTransaction(
- callCtx,
- () ->
- this.revokePrivilegeOnSecurableFromRole(
- callCtx, ms, grantee, catalogPath, securable, privilege));
- }
-
- /** {@link #loadGrantsOnSecurable(PolarisCallContext, long, long)} */
- private @Nonnull LoadGrantsResult loadGrantsOnSecurable(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- long securableCatalogId,
- long securableId) {
+ // TODO: Consider whether it's necessary to look up both ends of the grant records atomically
+ // or if it's actually safe as two independent lookups.
// lookup grants version for this securable entity
int grantsVersion =
@@ -1854,22 +1366,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull LoadGrantsResult loadGrantsOnSecurable(
- @Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId) {
+ public @Nonnull LoadGrantsResult loadGrantsToGrantee(
+ @Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
- // need to run inside a read transaction
- return ms.runInReadTransaction(
- callCtx, () -> this.loadGrantsOnSecurable(callCtx, ms, securableCatalogId, securableId));
- }
-
- /** {@link #loadGrantsToGrantee(PolarisCallContext, long, long)} */
- public @Nonnull LoadGrantsResult loadGrantsToGrantee(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- long granteeCatalogId,
- long granteeId) {
+ // TODO: Consider whether it's necessary to look up both ends of the grant records atomically
+ // or if it's actually safe as two independent lookups.
// lookup grants version for this grantee entity
int grantsVersion = ms.lookupEntityGrantRecordsVersion(callCtx, granteeCatalogId, granteeId);
@@ -1903,21 +1406,11 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull LoadGrantsResult loadGrantsToGrantee(
- @Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId) {
+ public @Nonnull ChangeTrackingResult loadEntitiesChangeTracking(
+ @Nonnull PolarisCallContext callCtx, @Nonnull List<PolarisEntityId> entityIds) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
- // need to run inside a read transaction
- return ms.runInReadTransaction(
- callCtx, () -> this.loadGrantsToGrantee(callCtx, ms, granteeCatalogId, granteeId));
- }
-
- /** {@link PolarisMetaStoreManager#loadEntitiesChangeTracking(PolarisCallContext, List)} */
- private @Nonnull ChangeTrackingResult loadEntitiesChangeTracking(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- @Nonnull List<PolarisEntityId> entityIds) {
List<PolarisChangeTrackingVersions> changeTracking =
ms.lookupEntityVersions(callCtx, entityIds);
return new ChangeTrackingResult(changeTracking);
@@ -1925,52 +1418,26 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull ChangeTrackingResult loadEntitiesChangeTracking(
- @Nonnull PolarisCallContext callCtx, @Nonnull List<PolarisEntityId> entityIds) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read transaction
- return ms.runInReadTransaction(
- callCtx, () -> this.loadEntitiesChangeTracking(callCtx, ms, entityIds));
- }
-
- /** Refer to {@link #loadEntity(PolarisCallContext, long, long, PolarisEntityType)} */
- private @Nonnull EntityResult loadEntity(
+ public @Nonnull EntityResult loadEntity(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
long entityCatalogId,
long entityId,
- int entityTypeCode) {
+ @Nonnull PolarisEntityType entityType) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
+
// this is an easy one
- PolarisBaseEntity entity = ms.lookupEntity(callCtx, entityCatalogId, entityId, entityTypeCode);
+ PolarisBaseEntity entity =
+ ms.lookupEntity(callCtx, entityCatalogId, entityId, entityType.getCode());
return (entity != null)
? new EntityResult(entity)
: new EntityResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null);
}
- /** {@inheritDoc} */
@Override
- public @Nonnull EntityResult loadEntity(
- @Nonnull PolarisCallContext callCtx,
- long entityCatalogId,
- long entityId,
- @Nonnull PolarisEntityType entityType) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read transaction
- return ms.runInReadTransaction(
- callCtx,
- () -> this.loadEntity(callCtx, ms, entityCatalogId, entityId, entityType.getCode()));
- }
-
- /** Refer to {@link #loadTasks(PolarisCallContext, String, int)} */
- private @Nonnull EntitiesResult loadTasks(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
- String executorId,
- int limit) {
+ public @Nonnull EntitiesResult loadTasks(
+ @Nonnull PolarisCallContext callCtx, String executorId, int limit) {
+ BasePersistence ms = callCtx.getMetaStore();
// find all available tasks
List<PolarisBaseEntity> availableTasks =
@@ -1997,10 +1464,9 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
Function.identity());
List<PolarisBaseEntity> loadedTasks = new ArrayList<>();
+ final AtomicInteger failedLeaseCount = new AtomicInteger(0);
availableTasks.forEach(
task -> {
- // Make a copy to avoid mutating someone else's reference.
- // TODO: Refactor into immutable/Builder pattern.
PolarisBaseEntity updatedTask = new PolarisBaseEntity(task);
Map<String, String> properties =
PolarisObjectMapperUtil.deserializeProperties(callCtx, task.getProperties());
@@ -2015,27 +1481,28 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
+ 1));
updatedTask.setProperties(
PolarisObjectMapperUtil.serializeProperties(callCtx, properties));
- EntityResult result = updateEntityPropertiesIfNotChanged(callCtx, ms, null, updatedTask);
+ EntityResult result = updateEntityPropertiesIfNotChanged(callCtx, null, updatedTask);
if (result.getReturnStatus() == BaseResult.ReturnStatus.SUCCESS) {
loadedTasks.add(result.getEntity());
} else {
- // TODO: Consider performing incremental leasing of individual tasks one at a time
- // instead of requiring all-or-none semantics for all the tasks we think we listed,
- // or else contention could be very bad.
- ms.rollback();
- throw new RetryOnConcurrencyException(
- "Failed to lease available task with status %s, info: %s",
- result.getReturnStatus(), result.getExtraInformation());
+ failedLeaseCount.getAndIncrement();
}
});
- return new EntitiesResult(loadedTasks);
- }
- @Override
- public @Nonnull EntitiesResult loadTasks(
- @Nonnull PolarisCallContext callCtx, String executorId, int limit) {
- TransactionalPersistence ms = callCtx.getMetaStore();
- return ms.runInTransaction(callCtx, () -> this.loadTasks(callCtx, ms, executorId, limit));
+ // Since the contract of this method is to only return an empty list once no available tasks
+ // are found anymore, if we happen to fail to lease any tasks at all due to all of them
+ // hitting concurrency errors, we need to just throw a concurrency exception rather than
+ // returning an empty list.
+ if (loadedTasks.isEmpty() && failedLeaseCount.get() > 0) {
+ // TODO: Currently the contract defined by BasePolarisMetaStoreManagerTest expects either
+ // a thrown RetryOnConcurrencyException or else a successful EntitiesResult in the face
+ // of concurrent loadTasks calls; this is inconsistent with the rest of the
+ // PolarisMetaStoreManager interface where everything is supposed to encapsulate errors
+ // in the Result type instead of throwing. But for now, we'll throw.
+ throw new RetryOnConcurrencyException(
+ "Failed to lease any of %s tasks due to concurrent leases", failedLeaseCount.get());
+ }
+ return new EntitiesResult(loadedTasks);
}
/** {@inheritDoc} */
@@ -2050,7 +1517,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull Set<String> allowedWriteLocations) {
// get meta store session we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
callCtx
.getDiagServices()
.check(
@@ -2064,9 +1531,14 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
reloadedEntity.getReturnStatus(), reloadedEntity.getExtraInformation());
}
+ // TODO: Consider whether this independent lookup is safe for the model already or whether
+ // we need better atomicity semantics between the base entity and the embedded storage
+ // integration.
+
// get storage integration
PolarisStorageIntegration<PolarisStorageConfigurationInfo> storageIntegration =
- ms.loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
+ ((IntegrationPersistence) ms)
+ .loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
// cannot be null
callCtx
@@ -2105,7 +1577,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull Set<PolarisStorageActions> actions,
@Nonnull Set<String> locations) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ BasePersistence ms = callCtx.getMetaStore();
callCtx
.getDiagServices()
.check(
@@ -2120,7 +1592,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// get storage integration, expect not null
PolarisStorageIntegration<PolarisStorageConfigurationInfo> storageIntegration =
- ms.loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
+ ((IntegrationPersistence) ms)
+ .loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
callCtx
.getDiagServices()
.checkNotNull(
@@ -2164,22 +1637,28 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
return deserializeProperties(callCtx, internalPropStr);
}
- /** {@link #loadResolvedEntityById(PolarisCallContext, long, long, PolarisEntityType)} */
- private @Nonnull ResolvedEntityResult loadResolvedEntityById(
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull ResolvedEntityResult loadResolvedEntityById(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
long entityCatalogId,
long entityId,
- int typeCode) {
+ PolarisEntityType entityType) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// load that entity
- PolarisBaseEntity entity = ms.lookupEntity(callCtx, entityCatalogId, entityId, typeCode);
+ PolarisBaseEntity entity =
+ ms.lookupEntity(callCtx, entityCatalogId, entityId, entityType.getCode());
// if entity not found, return null
if (entity == null) {
return new ResolvedEntityResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null);
}
+ // TODO: Consider whether it's necessary to look up grant records atomically with the entity
+ // or if it's actually safe as independent lookups. Document the nuances of independent lookups.
+
// load the grant records
final List<PolarisGrantRecord> grantRecords;
if (entity.getType().isGrantee()) {
@@ -2196,30 +1675,14 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** {@inheritDoc} */
@Override
- public @Nonnull ResolvedEntityResult loadResolvedEntityById(
- @Nonnull PolarisCallContext callCtx,
- long entityCatalogId,
- long entityId,
- PolarisEntityType entityType) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read transaction
- return ms.runInReadTransaction(
- callCtx,
- () ->
- this.loadResolvedEntityById(
- callCtx, ms, entityCatalogId, entityId, entityType.getCode()));
- }
-
- /** {@link #loadResolvedEntityById(PolarisCallContext, long, long, PolarisEntityType)} */
- private @Nonnull ResolvedEntityResult loadResolvedEntityByName(
+ public @Nonnull ResolvedEntityResult loadResolvedEntityByName(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
long entityCatalogId,
long parentId,
@Nonnull PolarisEntityType entityType,
@Nonnull String entityName) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// load that entity
PolarisBaseEntity entity =
@@ -2230,6 +1693,9 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
return new ResolvedEntityResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null);
}
+ // TODO: Consider whether it's necessary to look up grant records atomically with the entity
+ // or if it's actually safe as independent lookups. Document the nuances of independent lookups.
+
// load the grant records
final List<PolarisGrantRecord> grantRecords;
if (entity.getType().isGrantee()) {
@@ -2242,84 +1708,59 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
grantRecords = ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entity.getId());
}
- // return the result
- return new ResolvedEntityResult(entity, entity.getGrantRecordsVersion(), grantRecords);
- }
-
- /** {@inheritDoc} */
- @Override
- public @Nonnull ResolvedEntityResult loadResolvedEntityByName(
- @Nonnull PolarisCallContext callCtx,
- long entityCatalogId,
- long parentId,
- @Nonnull PolarisEntityType entityType,
- @Nonnull String entityName) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read transaction
ResolvedEntityResult result =
- ms.runInReadTransaction(
- callCtx,
- () ->
- this.loadResolvedEntityByName(
- callCtx, ms, entityCatalogId, parentId, entityType, entityName));
+ new ResolvedEntityResult(entity, entity.getGrantRecordsVersion(), grantRecords);
if (PolarisEntityConstants.getRootContainerName().equals(entityName)
&& entityType == PolarisEntityType.ROOT
&& !result.isSuccess()) {
// Backfill rootContainer if needed.
- ms.runActionInTransaction(
- callCtx,
- () -> {
- PolarisBaseEntity rootContainer =
- new PolarisBaseEntity(
- PolarisEntityConstants.getNullId(),
- PolarisEntityConstants.getRootEntityId(),
- PolarisEntityType.ROOT,
- PolarisEntitySubType.NULL_SUBTYPE,
- PolarisEntityConstants.getRootEntityId(),
- PolarisEntityConstants.getRootContainerName());
- EntityResult backfillResult =
- this.createEntityIfNotExists(callCtx, ms, null, rootContainer);
- if (backfillResult.isSuccess()) {
- PolarisBaseEntity serviceAdminRole =
- ms.lookupEntityByName(
- callCtx,
- 0L,
- 0L,
- PolarisEntityType.PRINCIPAL_ROLE.getCode(),
- PolarisEntityConstants.getNameOfPrincipalServiceAdminRole());
- if (serviceAdminRole != null) {
- this.persistNewGrantRecord(
- callCtx,
- ms,
- rootContainer,
- serviceAdminRole,
- PolarisPrivilege.SERVICE_MANAGE_ACCESS);
- }
- }
- });
+ PolarisBaseEntity rootContainer =
+ new PolarisBaseEntity(
+ PolarisEntityConstants.getNullId(),
+ PolarisEntityConstants.getRootEntityId(),
+ PolarisEntityType.ROOT,
+ PolarisEntitySubType.NULL_SUBTYPE,
+ PolarisEntityConstants.getRootEntityId(),
+ PolarisEntityConstants.getRootContainerName());
+ EntityResult backfillResult = this.createEntityIfNotExists(callCtx, null, rootContainer);
+ if (backfillResult.isSuccess()) {
+ PolarisBaseEntity serviceAdminRole =
+ ms.lookupEntityByName(
+ callCtx,
+ 0L,
+ 0L,
+ PolarisEntityType.PRINCIPAL_ROLE.getCode(),
+ PolarisEntityConstants.getNameOfPrincipalServiceAdminRole());
+ if (serviceAdminRole != null) {
+ // TODO: If the original backfill attempt crashes after persisting the root entity
+ // but before persisting new grant record, there can be a stuck root-container that
+ // lacks the grant to the serviceAdmin. However, nowadays this backfill code might
+ // already be obsolete since it was just a holdover when bootstrap didn't create
+ // the root container, so consider removing this backfill entirely or else fix
+ // the backfill of this serviceAdminRole grant.
+ this.persistNewGrantRecord(
+ callCtx, ms, rootContainer, serviceAdminRole, PolarisPrivilege.SERVICE_MANAGE_ACCESS);
+ }
+ }
// Redo the lookup in a separate read transaction.
result =
- ms.runInReadTransaction(
- callCtx,
- () ->
- this.loadResolvedEntityByName(
- callCtx, ms, entityCatalogId, parentId, entityType, entityName));
+ this.loadResolvedEntityByName(callCtx, entityCatalogId, parentId, entityType, entityName);
}
return result;
}
/** {@inheritDoc} */
- private @Nonnull ResolvedEntityResult refreshResolvedEntity(
+ @Override
+ public @Nonnull ResolvedEntityResult refreshResolvedEntity(
@Nonnull PolarisCallContext callCtx,
- @Nonnull TransactionalPersistence ms,
int entityVersion,
int entityGrantRecordsVersion,
@Nonnull PolarisEntityType entityType,
long entityCatalogId,
long entityId) {
+ // get metastore we should be using
+ BasePersistence ms = callCtx.getMetaStore();
// load version information
PolarisChangeTrackingVersions entityVersions =
@@ -2345,6 +1786,9 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
entity = null;
}
+ // TODO: Consider whether it's necessary to look up grant records atomically with the entity
+ // or if it's actually safe as independent lookups. Document the nuances of independent lookups.
+
// load the grant records if required
final List<PolarisGrantRecord> grantRecords;
if (entityVersions.getGrantRecordsVersion() != entityGrantRecordsVersion) {
@@ -2362,30 +1806,4 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// return the result
return new ResolvedEntityResult(entity, entityVersions.getGrantRecordsVersion(), grantRecords);
}
-
- /** {@inheritDoc} */
- @Override
- public @Nonnull ResolvedEntityResult refreshResolvedEntity(
- @Nonnull PolarisCallContext callCtx,
- int entityVersion,
- int entityGrantRecordsVersion,
- @Nonnull PolarisEntityType entityType,
- long entityCatalogId,
- long entityId) {
- // get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- // need to run inside a read transaction
- return ms.runInReadTransaction(
- callCtx,
- () ->
- this.refreshResolvedEntity(
- callCtx,
- ms,
- entityVersion,
- entityGrantRecordsVersion,
- entityType,
- entityCatalogId,
- entityId));
- }
}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/BaseMetaStoreManager.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/BaseMetaStoreManager.java
index c27ee1b37..e0bd49a2f 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/BaseMetaStoreManager.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/BaseMetaStoreManager.java
@@ -18,14 +18,25 @@
*/
package org.apache.polaris.core.persistence;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Nonnull;
import java.util.Map;
import org.apache.polaris.core.PolarisCallContext;
import org.apache.polaris.core.entity.PolarisBaseEntity;
import org.apache.polaris.core.entity.PolarisEntityConstants;
+import org.apache.polaris.core.entity.PolarisEntitySubType;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.persistence.dao.entity.GenerateEntityIdResult;
import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
+/** Shared basic PolarisMetaStoreManager logic for transactional and non-transactional impls. */
public abstract class BaseMetaStoreManager implements PolarisMetaStoreManager {
+ /** mapper, allows to serialize/deserialize properties to/from JSON */
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
public static PolarisStorageConfigurationInfo extractStorageConfiguration(
@Nonnull PolarisCallContext callCtx, PolarisBaseEntity reloadedEntity) {
Map<String, String> propMap =
@@ -45,4 +56,179 @@ public abstract class BaseMetaStoreManager implements PolarisMetaStoreManager {
return PolarisStorageConfigurationInfo.deserialize(
callCtx.getDiagServices(), storageConfigInfoStr);
}
+
+ /**
+ * Given the internal property as a map of key/value pairs, serialize it to a String
+ *
+ * @param properties a map of key/value pairs
+ * @return a String, the JSON representation of the map
+ */
+ public String serializeProperties(PolarisCallContext callCtx, Map<String, String> properties) {
+
+ String jsonString = null;
+ try {
+ // Deserialize the JSON string to a Map<String, String>
+ jsonString = MAPPER.writeValueAsString(properties);
+ } catch (JsonProcessingException ex) {
+ callCtx.getDiagServices().fail("got_json_processing_exception", "ex={}", ex);
+ }
+
+ return jsonString;
+ }
+
+ /**
+ * Given the serialized properties, deserialize those to a {@code Map<String, String>}
+ *
+ * @param properties a JSON string representing the set of properties
+ * @return a Map of string
+ */
+ public Map<String, String> deserializeProperties(PolarisCallContext callCtx, String properties) {
+
+ Map<String, String> retProperties = null;
+ try {
+ // Deserialize the JSON string to a Map<String, String>
+ retProperties = MAPPER.readValue(properties, new TypeReference<>() {});
+ } catch (JsonMappingException ex) {
+ callCtx.getDiagServices().fail("got_json_mapping_exception", "ex={}", ex);
+ } catch (JsonProcessingException ex) {
+ callCtx.getDiagServices().fail("got_json_processing_exception", "ex={}", ex);
+ }
+
+ return retProperties;
+ }
+
+ /**
+ * Performs basic validation of expected invariants on a new entity, then returns the entity with
+ * fields filled out for which the persistence layer is responsible.
+ *
+ * @param callCtx call context
+ * @param ms meta store in read/write mode
+ * @param entity entity we need a new persisted record for
+ */
+ protected PolarisBaseEntity prepareToPersistNewEntity(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull BasePersistence ms,
+ @Nonnull PolarisBaseEntity entity) {
+
+ // validate the entity type and subtype
+ callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
+ callCtx
+ .getDiagServices()
+ .checkNotNull(entity.getName(), "unexpected_null_name", "entity={}", entity);
+ PolarisEntityType type = PolarisEntityType.fromCode(entity.getTypeCode());
+ callCtx.getDiagServices().checkNotNull(type, "unknown_type", "entity={}", entity);
+ PolarisEntitySubType subType = PolarisEntitySubType.fromCode(entity.getSubTypeCode());
+ callCtx.getDiagServices().checkNotNull(subType, "unexpected_null_subType", "entity={}", entity);
+ callCtx
+ .getDiagServices()
+ .check(
+ subType.getParentType() == null || subType.getParentType() == type,
+ "invalid_subtype",
+ "type={} subType={}",
+ type,
+ subType);
+
+ // if top-level entity, its parent should be the account
+ callCtx
+ .getDiagServices()
+ .check(
+ !type.isTopLevel() || entity.getParentId() == PolarisEntityConstants.getRootEntityId(),
+ "top_level_parent_should_be_account",
+ "entity={}",
+ entity);
+
+ // id should not be null
+ callCtx
+ .getDiagServices()
+ .check(
+ entity.getId() != 0 || type == PolarisEntityType.ROOT,
+ "id_not_set",
+ "entity={}",
+ entity);
+
+ // creation timestamp must be filled
+ callCtx.getDiagServices().check(entity.getCreateTimestamp() != 0, "null_create_timestamp");
+
+ // this is the first change
+ // TODO: Make immutable; make sure no caller depends on the input entity actually
+ // being changed.
+ entity.setLastUpdateTimestamp(entity.getCreateTimestamp());
+
+ // set all other timestamps to 0
+ entity.setDropTimestamp(0);
+ entity.setPurgeTimestamp(0);
+ entity.setToPurgeTimestamp(0);
+ return entity;
+ }
+
+ /**
+ * Performs basic validation of expected invariants on a changed entity, then returns the entity
+ * with fields filled out for which the persistence layer is responsible.
+ *
+ * @param callCtx call context
+ * @param ms meta store
+ * @param entity the entity which has been changed
+ * @param nameOrParentChanged indicates if parent or name changed
+ * @param originalEntity the original state of the entity before changes
+ * @return the entity with its version and lastUpdateTimestamp updated
+ */
+ protected @Nonnull PolarisBaseEntity prepareToPersistEntityAfterChange(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull BasePersistence ms,
+ @Nonnull PolarisBaseEntity entity,
+ boolean nameOrParentChanged,
+ @Nonnull PolarisBaseEntity originalEntity) {
+
+ // validate the entity type and subtype
+ callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
+ callCtx
+ .getDiagServices()
+ .checkNotNull(entity.getName(), "unexpected_null_name", "entity={}", entity);
+ PolarisEntityType type = entity.getType();
+ callCtx.getDiagServices().checkNotNull(type, "unexpected_null_type", "entity={}", entity);
+ PolarisEntitySubType subType = entity.getSubType();
+ callCtx.getDiagServices().checkNotNull(subType, "unexpected_null_subType", "entity={}", entity);
+ callCtx
+ .getDiagServices()
+ .check(
+ subType.getParentType() == null || subType.getParentType() == type,
+ "invalid_subtype",
+ "type={} subType={} entity={}",
+ type,
+ subType,
+ entity);
+
+ // entity should not have been dropped
+ callCtx
+ .getDiagServices()
+ .check(entity.getDropTimestamp() == 0, "entity_dropped", "entity={}", entity);
+
+ // creation timestamp must be filled
+ long createTimestamp = entity.getCreateTimestamp();
+ callCtx
+ .getDiagServices()
+ .check(createTimestamp != 0, "null_create_timestamp", "entity={}", entity);
+
+ // ensure time is not moving backward...
+ long now = System.currentTimeMillis();
+ if (now < entity.getCreateTimestamp()) {
+ now = entity.getCreateTimestamp() + 1;
+ }
+
+ // update last update timestamp and increment entity version
+ // TODO: Make immutable; make sure no caller depends on the input entity actually
+ // being changed.
+ entity.setLastUpdateTimestamp(now);
+ entity.setEntityVersion(entity.getEntityVersion() + 1);
+ return entity;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public @Nonnull GenerateEntityIdResult generateNewEntityId(@Nonnull PolarisCallContext callCtx) {
+ // get meta store we should be using
+ BasePersistence ms = callCtx.getMetaStore();
+
+ return new GenerateEntityIdResult(ms.generateNewId(callCtx));
+ }
}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/BasePersistence.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/BasePersistence.java
index 75c30d430..dccb54684 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/BasePersistence.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/BasePersistence.java
@@ -60,6 +60,11 @@ public interface BasePersistence {
* or push down the return status enums from PolarisMetaStoreManager into this layer and document
* accordingly.
*
+ * <p>TODO: Change originalEntity to be just the set of members taht participate in conditions,
+ * similar to PolarisEntityCore, and make the callsites in BasePolarisCatalog actually plumb
+ * through correctly, in particular for values the PolarisMetaStoreManagerImpl doesn't have access
+ * to such as the original name and parentId in renames.
+ *
* @param callCtx call context
* @param entity entity to persist
* @param nameOrParentChanged if true, also write it to by-name lookups if applicable
@@ -73,19 +78,36 @@ public interface BasePersistence {
@Nullable PolarisBaseEntity originalEntity);
/**
- * Atomically write a batch of entities to the persistence backend conditional on *every* member
- * of originalEntities matching the existing persistent state. After this commit, *every* member
- * of entities must be committed durably.
+ * write a batch of entities to the persistence backend conditional on *every* member of
+ * originalEntities matching the existing persistent state. After this commit, *every* member of
+ * entities must be committed durably.
+ *
+ * <p>Important: For now, the following constraints apply:
+ *
+ * <ul>
+ * <li>Either all entities are CREATE xor all entities are UPDATE; no mixing and matching of
+ * CREATE and UPDATE in a single batch here
+ * <li>Either all entities are catalog-scoped child entities (Namespace/Table/View) or all
+ * entities are root-scoped entities (Catalogs/Principals/PrincipalRoles/Tasks). TODO:
+ * Document whether CatalogRole is considered catalog-scoped, root-scoped, or "other" for
+ * these purposes.
+ * </ul>
*
* <p>TODO: Push down the multi-entity commit from PolarisMetaStoreManagerImpl to use this instead
* of running single writeEntity actions within a transaction.
*
+ * <p>TODO: Change originalEntity to be just the set of members taht participate in conditions,
+ * similar to PolarisEntityCore, and make the callsites in BasePolarisCatalog actually plumb
+ * through correctly, in particular for values the PolarisMetaStoreManagerImpl doesn't have access
+ * to such as the original name and parentId in renames.
+ *
* @param callCtx call context
* @param entities entities to persist
* @param originalEntities original states of the entity to use for compare-and-swap purposes, or
* null if this is expected to be a brand-new entity; must contain exactly as many elements as
* {@code entities} where each item corresponds to the element of {@code entities} in the same
- * index as this list.
+ * index as this list. If non-null, we expect all elements of originalEntities to be non-null;
+ * there is no mix-and-match "create" and "update" in a single batch.
*/
void writeEntities(
@Nonnull PolarisCallContext callCtx,
@@ -198,6 +220,7 @@ public interface BasePersistence {
* @param name the name of the entity
* @return null if the specified entity does not exist
*/
+ @Nullable
default EntityNameLookupRecord lookupEntityIdAndSubTypeByName(
@Nonnull PolarisCallContext callCtx,
long catalogId,
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/EntityAlreadyExistsException.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/EntityAlreadyExistsException.java
new file mode 100644
index 000000000..e31e2ae02
--- /dev/null
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/EntityAlreadyExistsException.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.persistence;
+
+import org.apache.polaris.core.entity.PolarisBaseEntity;
+
+/**
+ * Exception raised when some kind of conclit in the persistence layer prevents the attempted
+ * creation of a new entity; provides a member holding the conflicting entity.
+ */
+public class EntityAlreadyExistsException extends RuntimeException {
+ private PolarisBaseEntity existingEntity;
+
+ /**
+ * @param existingEntity The conflicting entity that caused creation to fail.
+ */
+ public EntityAlreadyExistsException(PolarisBaseEntity existingEntity) {
+ super(existingEntity.getName() + ":" + existingEntity.getId());
+ this.existingEntity = existingEntity;
+ }
+
+ public PolarisBaseEntity getExistingEntity() {
+ return this.existingEntity;
+ }
+}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java
index ff2eaaf13..9af4e1763 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java
@@ -87,6 +87,14 @@ public abstract class LocalPolarisMetaStoreManagerFactory<StoreType>
}
}
+ /**
+ * Subclasses can override this to inject different implementations of PolarisMetaStoreManager
+ * into the existing realm-based setup flow.
+ */
+ protected PolarisMetaStoreManager createNewMetaStoreManager() {
+ return new PolarisMetaStoreManagerImpl();
+ }
+
private void initializeForRealm(
RealmContext realmContext, RootCredentialsSet rootCredentialsSet) {
final StoreType backingStore = createBackingStore(diagnostics);
@@ -94,7 +102,7 @@ public abstract class LocalPolarisMetaStoreManagerFactory<StoreType>
realmContext.getRealmIdentifier(),
() -> createMetaStoreSession(backingStore, realmContext, rootCredentialsSet, diagnostics));
- PolarisMetaStoreManager metaStoreManager = new PolarisMetaStoreManagerImpl();
+ PolarisMetaStoreManager metaStoreManager = createNewMetaStoreManager();
metaStoreManagerMap.put(realmContext.getRealmIdentifier(), metaStoreManager);
}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/AbstractTransactionalPersistence.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/AbstractTransactionalPersistence.java
index 6ed613ef3..c673c4849 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/AbstractTransactionalPersistence.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/AbstractTransactionalPersistence.java
@@ -21,11 +21,22 @@ package org.apache.polaris.core.persistence.transactional;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Nullable;
import java.util.List;
+import java.util.function.Function;
+import java.util.function.Predicate;
import org.apache.polaris.core.PolarisCallContext;
import org.apache.polaris.core.entity.EntityNameLookupRecord;
import org.apache.polaris.core.entity.PolarisBaseEntity;
+import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
import org.apache.polaris.core.entity.PolarisEntitiesActiveKey;
import org.apache.polaris.core.entity.PolarisEntityCore;
+import org.apache.polaris.core.entity.PolarisEntityId;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.PolarisGrantRecord;
+import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
+import org.apache.polaris.core.persistence.EntityAlreadyExistsException;
+import org.apache.polaris.core.persistence.RetryOnConcurrencyException;
+import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
/**
* Extends BasePersistence to express a more "transaction-oriented" control flow for backing stores
@@ -33,67 +44,10 @@ import org.apache.polaris.core.entity.PolarisEntityCore;
* the BasePersistence methods in terms of lower-level methods that subclasses must implement.
*/
public abstract class AbstractTransactionalPersistence implements TransactionalPersistence {
-
- /** {@inheritDoc} */
- @Override
- public void writeEntities(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull List<PolarisBaseEntity> entities,
- @Nullable List<PolarisBaseEntity> originalEntities) {
- throw new UnsupportedOperationException("Not yet implemented");
- }
-
- /** {@inheritDoc} */
- @Override
- public PolarisBaseEntity lookupEntityByName(
- @Nonnull PolarisCallContext callCtx,
- long catalogId,
- long parentId,
- int typeCode,
- @Nonnull String name) {
- // TODO: Consistently pull down the runInTransaction logic without running into conflicting
- // nested transactions into here so that instead of having the caller be responsible for
- // initiating the runInReadTransaction, we make this method call inherently safe to do
- // the two-phase lookup.
- PolarisEntitiesActiveKey entityActiveKey =
- new PolarisEntitiesActiveKey(catalogId, parentId, typeCode, name);
-
- // ensure that the entity exists
- EntityNameLookupRecord entityActiveRecord = lookupEntityActive(callCtx, entityActiveKey);
-
- // if not found, return null
- if (entityActiveRecord == null) {
- return null;
- }
-
- // lookup the entity, should be there
- PolarisBaseEntity entity =
- lookupEntity(
- callCtx,
- entityActiveRecord.getCatalogId(),
- entityActiveRecord.getId(),
- entityActiveRecord.getTypeCode());
- callCtx
- .getDiagServices()
- .checkNotNull(
- entity, "unexpected_not_found_entity", "entityActiveRecord={}", entityActiveRecord);
-
- // return it now
- return entity;
- }
-
- /** {@inheritDoc} */
- @Override
- public EntityNameLookupRecord lookupEntityIdAndSubTypeByName(
- @Nonnull PolarisCallContext callCtx,
- long catalogId,
- long parentId,
- int typeCode,
- @Nonnull String name) {
- PolarisEntitiesActiveKey entityActiveKey =
- new PolarisEntitiesActiveKey(catalogId, parentId, typeCode, name);
- return lookupEntityActive(callCtx, entityActiveKey);
- }
+ //
+ // New abstract methods specific to this slice-based transactional persistence that subclasses
+ // must implement to inherit implementations of lookup/write/delete
+ //
/**
* Lookup an entity by entityActiveKey
@@ -103,32 +57,9 @@ public abstract class AbstractTransactionalPersistence implements TransactionalP
* @return null if the specified entity does not exist or has been dropped.
*/
@Nullable
- protected abstract EntityNameLookupRecord lookupEntityActive(
+ protected abstract EntityNameLookupRecord lookupEntityActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntitiesActiveKey entityActiveKey);
- /** {@inheritDoc} */
- @Override
- public void writeEntity(
- @Nonnull PolarisCallContext callCtx,
- @Nonnull PolarisBaseEntity entity,
- boolean nameOrParentChanged,
- @Nullable PolarisBaseEntity originalEntity) {
- // TODO: Pull down relevant compare-and-swap semantics from PolarisMetaStoreManagerImpl
- // into this layer.
- writeToEntities(callCtx, entity);
- writeToEntitiesChangeTracking(callCtx, entity);
-
- if (nameOrParentChanged) {
- if (originalEntity != null) {
- // In our case, rename isn't automatically handled when the main "entities" slice
- // is updated; instead we must explicitly remove from the old entitiesActive
- // key as well.
- deleteFromEntitiesActive(callCtx, originalEntity);
- }
- writeToEntitiesActive(callCtx, entity);
- }
- }
-
/**
* Write the base entity to the entities table. If there is a conflict (existing record with the
* same id), all attributes of the new record will replace the existing one.
@@ -137,7 +68,7 @@ public abstract class AbstractTransactionalPersistence implements TransactionalP
* @param entity entity record to write, potentially replacing an existing entity record with the
* same key
*/
- protected abstract void writeToEntities(
+ protected abstract void writeToEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity);
/**
@@ -148,7 +79,7 @@ public abstract class AbstractTransactionalPersistence implements TransactionalP
* @param entity entity record to write, potentially replacing an existing entity record with the
* same key
*/
- protected abstract void writeToEntitiesActive(
+ protected abstract void writeToEntitiesActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity);
/**
@@ -159,24 +90,16 @@ public abstract class AbstractTransactionalPersistence implements TransactionalP
* @param entity entity record to write, potentially replacing an existing entity record with the
* same key
*/
- protected abstract void writeToEntitiesChangeTracking(
+ protected abstract void writeToEntitiesChangeTrackingInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity);
- /** {@inheritDoc} */
- @Override
- public void deleteEntity(@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
- deleteFromEntitiesActive(callCtx, entity);
- deleteFromEntities(callCtx, entity);
- deleteFromEntitiesChangeTracking(callCtx, entity);
- }
-
/**
* Delete the base entity from the entities table.
*
* @param callCtx call context
* @param entity entity record to delete
*/
- protected abstract void deleteFromEntities(
+ protected abstract void deleteFromEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity);
/**
@@ -185,7 +108,7 @@ public abstract class AbstractTransactionalPersistence implements TransactionalP
* @param callCtx call context
* @param entity entity record to delete
*/
- protected abstract void deleteFromEntitiesActive(
+ protected abstract void deleteFromEntitiesActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity);
/**
@@ -194,6 +117,547 @@ public abstract class AbstractTransactionalPersistence implements TransactionalP
* @param callCtx call context
* @param entity entity record to delete
*/
- protected abstract void deleteFromEntitiesChangeTracking(
+ protected abstract void deleteFromEntitiesChangeTrackingInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity);
+
+ //
+ // Implementations of the one-shot atomic BasePersistence methods which explicitly run
+ // the in-transaction variants of methods in a new transaction.
+ //
+
+ /** {@inheritDoc} */
+ @Override
+ public long generateNewId(@Nonnull PolarisCallContext callCtx) {
+ return runInTransaction(callCtx, () -> this.generateNewIdInCurrentTxn(callCtx));
+ }
+
+ /** Helper to perform the compare-and-swap semantics of a single writeEntity call. */
+ private void checkConditionsForWriteEntityInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisBaseEntity entity,
+ @Nullable PolarisBaseEntity originalEntity) {
+ PolarisBaseEntity refreshedEntity =
+ this.lookupEntityInCurrentTxn(
+ callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
+
+ if (originalEntity == null) {
+ if (refreshedEntity != null) {
+ // If this is a "create", and we manage to look up an existing entity with already
+ // the same id, where ids are uniquely reserved when generated, it means it's a
+ // low-level retry possibly in the face of a transient connectivity failure to
+ // the backend database.
+ throw new EntityAlreadyExistsException(refreshedEntity);
+ } else {
+ // Successfully verified the entity doesn't already exist by-id, but for a "create"
+ // we must also check for name-collection now.
+ refreshedEntity =
+ this.lookupEntityByNameInCurrentTxn(
+ callCtx,
+ entity.getCatalogId(),
+ entity.getParentId(),
+ entity.getType().getCode(),
+ entity.getName());
+ if (refreshedEntity != null) {
+ // Name-collision conflict.
+ throw new EntityAlreadyExistsException(refreshedEntity);
+ }
+ }
+ } else {
+ // This is an "update".
+ if (refreshedEntity == null
+ || refreshedEntity.getEntityVersion() != originalEntity.getEntityVersion()
+ || refreshedEntity.getGrantRecordsVersion() != originalEntity.getGrantRecordsVersion()) {
+ // TODO: Better standardization of exception types, possibly make the ones that are
+ // really part of the persistence contract be CheckedExceptions.
+ throw new RetryOnConcurrencyException(
+ "Entity '%s' id '%s' concurrently modified; expected version %s/%s got %s/%s",
+ entity.getName(),
+ entity.getId(),
+ originalEntity.getEntityVersion(),
+ originalEntity.getGrantRecordsVersion(),
+ refreshedEntity != null ? refreshedEntity.getEntityVersion() : -1,
+ refreshedEntity != null ? refreshedEntity.getGrantRecordsVersion() : -1);
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void writeEntity(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisBaseEntity entity,
+ boolean nameOrParentChanged,
+ @Nullable PolarisBaseEntity originalEntity) {
+ runActionInTransaction(
+ callCtx,
+ () -> {
+ this.checkConditionsForWriteEntityInCurrentTxn(callCtx, entity, originalEntity);
+ this.writeEntityInCurrentTxn(callCtx, entity, nameOrParentChanged, originalEntity);
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void writeEntities(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull List<PolarisBaseEntity> entities,
+ @Nullable List<PolarisBaseEntity> originalEntities) {
+ if (originalEntities != null) {
+ callCtx
+ .getDiagServices()
+ .check(
+ entities.size() == originalEntities.size(),
+ "mismatched_entities_and_original_entities_size",
+ "entities.size()={}, originalEntities.size()={}",
+ entities.size(),
+ originalEntities.size());
+ }
+ runActionInTransaction(
+ callCtx,
+ () -> {
+ // Validate and write each one independently so that we can also detect conflicting
+ // writes to the same entity id within a given batch (so that previously written
+ // ones will be seen during validation of the later item).
+ for (int i = 0; i < entities.size(); ++i) {
+ PolarisBaseEntity entity = entities.get(i);
+ PolarisBaseEntity originalEntity =
+ originalEntities != null ? originalEntities.get(i) : null;
+ // TODO: This isn't quite correct right now, because originalEntity is only actually
+ // safe to use for entityVersion and grantRecordsVersion right now. Once we refactor
+ // the writeEntities[] methods to take something like PolarisEntityCore
+ // for originalEntity and force the callsites such as BasePolarisCatalog to actually
+ // provide the original values, this will be correct. For now, the API does't support
+ // bulk renames anyways.
+ boolean nameOrParentChanged =
+ originalEntity == null
+ || !entity.getName().equals(originalEntity.getName())
+ || entity.getParentId() != originalEntity.getParentId();
+ try {
+ this.checkConditionsForWriteEntityInCurrentTxn(callCtx, entity, originalEntity);
+ } catch (EntityAlreadyExistsException e) {
+ // If the ids are equal then it is an idempotent-create-retry error, which counts
+ // as a "success" for multi-entity commit purposes; name-collisions on different
+ // ids counts as a true error that we rethrow.
+ if (e.getExistingEntity().getId() != entity.getId()) {
+ throw e;
+ }
+ // Else silently swallow the apparent create-retry
+ }
+ this.writeEntityInCurrentTxn(callCtx, entity, nameOrParentChanged, originalEntity);
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void writeToGrantRecords(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {
+ runActionInTransaction(callCtx, () -> this.writeToGrantRecordsInCurrentTxn(callCtx, grantRec));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void deleteEntity(@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
+ runActionInTransaction(callCtx, () -> this.deleteEntityInCurrentTxn(callCtx, entity));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void deleteFromGrantRecords(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {
+ runActionInTransaction(
+ callCtx, () -> this.deleteFromGrantRecordsInCurrentTxn(callCtx, grantRec));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void deleteAllEntityGrantRecords(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisEntityCore entity,
+ @Nonnull List<PolarisGrantRecord> grantsOnGrantee,
+ @Nonnull List<PolarisGrantRecord> grantsOnSecurable) {
+ runActionInTransaction(
+ callCtx,
+ () ->
+ this.deleteAllEntityGrantRecordsInCurrentTxn(
+ callCtx, entity, grantsOnGrantee, grantsOnSecurable));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void deleteAll(@Nonnull PolarisCallContext callCtx) {
+ runActionInTransaction(callCtx, () -> this.deleteAllInCurrentTxn(callCtx));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public PolarisBaseEntity lookupEntity(
+ @Nonnull PolarisCallContext callCtx, long catalogId, long entityId, int typeCode) {
+ return runInReadTransaction(
+ callCtx, () -> this.lookupEntityInCurrentTxn(callCtx, catalogId, entityId, typeCode));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public PolarisBaseEntity lookupEntityByName(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ int typeCode,
+ @Nonnull String name) {
+ return runInReadTransaction(
+ callCtx,
+ () -> this.lookupEntityByNameInCurrentTxn(callCtx, catalogId, parentId, typeCode, name));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public EntityNameLookupRecord lookupEntityIdAndSubTypeByName(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ int typeCode,
+ @Nonnull String name) {
+ return runInReadTransaction(
+ callCtx,
+ () ->
+ this.lookupEntityIdAndSubTypeByNameInCurrentTxn(
+ callCtx, catalogId, parentId, typeCode, name));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public List<PolarisBaseEntity> lookupEntities(
+ @Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds) {
+ return runInReadTransaction(callCtx, () -> this.lookupEntitiesInCurrentTxn(callCtx, entityIds));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public List<PolarisChangeTrackingVersions> lookupEntityVersions(
+ @Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds) {
+ return runInReadTransaction(
+ callCtx, () -> this.lookupEntityVersionsInCurrentTxn(callCtx, entityIds));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public List<EntityNameLookupRecord> listEntities(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ @Nonnull PolarisEntityType entityType) {
+ return runInReadTransaction(
+ callCtx, () -> this.listEntitiesInCurrentTxn(callCtx, catalogId, parentId, entityType));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public List<EntityNameLookupRecord> listEntities(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ @Nonnull PolarisEntityType entityType,
+ @Nonnull Predicate<PolarisBaseEntity> entityFilter) {
+ return runInReadTransaction(
+ callCtx,
+ () ->
+ this.listEntitiesInCurrentTxn(callCtx, catalogId, parentId, entityType, entityFilter));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public <T> List<T> listEntities(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ @Nonnull PolarisEntityType entityType,
+ int limit,
+ @Nonnull Predicate<PolarisBaseEntity> entityFilter,
+ @Nonnull Function<PolarisBaseEntity, T> transformer) {
+ return runInReadTransaction(
+ callCtx,
+ () ->
+ this.listEntitiesInCurrentTxn(
+ callCtx, catalogId, parentId, entityType, limit, entityFilter, transformer));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public int lookupEntityGrantRecordsVersion(
+ @Nonnull PolarisCallContext callCtx, long catalogId, long entityId) {
+ return runInReadTransaction(
+ callCtx,
+ () -> this.lookupEntityGrantRecordsVersionInCurrentTxn(callCtx, catalogId, entityId));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public PolarisGrantRecord lookupGrantRecord(
+ @Nonnull PolarisCallContext callCtx,
+ long securableCatalogId,
+ long securableId,
+ long granteeCatalogId,
+ long granteeId,
+ int privilegeCode) {
+ return runInReadTransaction(
+ callCtx,
+ () ->
+ this.lookupGrantRecordInCurrentTxn(
+ callCtx,
+ securableCatalogId,
+ securableId,
+ granteeCatalogId,
+ granteeId,
+ privilegeCode));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public List<PolarisGrantRecord> loadAllGrantRecordsOnSecurable(
+ @Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId) {
+ return runInReadTransaction(
+ callCtx,
+ () ->
+ this.loadAllGrantRecordsOnSecurableInCurrentTxn(
+ callCtx, securableCatalogId, securableId));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public List<PolarisGrantRecord> loadAllGrantRecordsOnGrantee(
+ @Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId) {
+ return runInReadTransaction(
+ callCtx,
+ () -> this.loadAllGrantRecordsOnGranteeInCurrentTxn(callCtx, granteeCatalogId, granteeId));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean hasChildren(
+ @Nonnull PolarisCallContext callCtx,
+ @Nullable PolarisEntityType optionalEntityType,
+ long catalogId,
+ long parentId) {
+ return runInReadTransaction(
+ callCtx,
+ () -> this.hasChildrenInCurrentTxn(callCtx, optionalEntityType, catalogId, parentId));
+ }
+
+ //
+ // Implementations of the one-shot atomic IntegrationPersistence methods which explicitly run
+ // the * variants of methods in a new transaction.
+ //
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public PolarisPrincipalSecrets loadPrincipalSecrets(
+ @Nonnull PolarisCallContext callCtx, @Nonnull String clientId) {
+ return runInReadTransaction(
+ callCtx, () -> this.loadPrincipalSecretsInCurrentTxn(callCtx, clientId));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nonnull
+ public PolarisPrincipalSecrets generateNewPrincipalSecrets(
+ @Nonnull PolarisCallContext callCtx, @Nonnull String principalName, long principalId) {
+ return runInTransaction(
+ callCtx,
+ () -> this.generateNewPrincipalSecretsInCurrentTxn(callCtx, principalName, principalId));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public PolarisPrincipalSecrets rotatePrincipalSecrets(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull String clientId,
+ long principalId,
+ boolean reset,
+ @Nonnull String oldSecretHash) {
+ return runInTransaction(
+ callCtx,
+ () ->
+ this.rotatePrincipalSecretsInCurrentTxn(
+ callCtx, clientId, principalId, reset, oldSecretHash));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void deletePrincipalSecrets(
+ @Nonnull PolarisCallContext callCtx, @Nonnull String clientId, long principalId) {
+ runActionInTransaction(
+ callCtx, () -> this.deletePrincipalSecretsInCurrentTxn(callCtx, clientId, principalId));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public <T extends PolarisStorageConfigurationInfo>
+ PolarisStorageIntegration<T> createStorageIntegration(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long entityId,
+ PolarisStorageConfigurationInfo polarisStorageConfigurationInfo) {
+ return runInTransaction(
+ callCtx,
+ () ->
+ this.createStorageIntegrationInCurrentTxn(
+ callCtx, catalogId, entityId, polarisStorageConfigurationInfo));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public <T extends PolarisStorageConfigurationInfo> void persistStorageIntegrationIfNeeded(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisBaseEntity entity,
+ @Nullable PolarisStorageIntegration<T> storageIntegration) {
+ runActionInTransaction(
+ callCtx,
+ () ->
+ this.persistStorageIntegrationIfNeededInCurrentTxn(
+ callCtx, entity, storageIntegration));
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public <T extends PolarisStorageConfigurationInfo>
+ PolarisStorageIntegration<T> loadPolarisStorageIntegration(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
+ return runInReadTransaction(
+ callCtx, () -> this.loadPolarisStorageIntegrationInCurrentTxn(callCtx, entity));
+ }
+
+ //
+ // Implementations of the in-transaction versions for basic write/delete/lookup using the
+ // slice-based model supported by this class.
+ //
+
+ /** {@inheritDoc} */
+ @Override
+ public void writeEntityInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisBaseEntity entity,
+ boolean nameOrParentChanged,
+ @Nullable PolarisBaseEntity originalEntity) {
+ this.writeToEntitiesInCurrentTxn(callCtx, entity);
+ this.writeToEntitiesChangeTrackingInCurrentTxn(callCtx, entity);
+
+ if (nameOrParentChanged) {
+ if (originalEntity != null) {
+ // In our case, rename isn't automatically handled when the main "entities" slice
+ // is updated; instead we must explicitly remove from the old entitiesActive
+ // key as well.
+ this.deleteFromEntitiesActiveInCurrentTxn(callCtx, originalEntity);
+ }
+ this.writeToEntitiesActiveInCurrentTxn(callCtx, entity);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void writeEntitiesInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull List<PolarisBaseEntity> entities,
+ @Nullable List<PolarisBaseEntity> originalEntities) {
+ if (originalEntities != null) {
+ callCtx
+ .getDiagServices()
+ .check(
+ entities.size() == originalEntities.size(),
+ "mismatched_entities_and_original_entities_size",
+ "entities.size()={}, originalEntities.size()={}",
+ entities.size(),
+ originalEntities.size());
+ }
+ for (int i = 0; i < entities.size(); ++i) {
+ PolarisBaseEntity entity = entities.get(i);
+ PolarisBaseEntity originalEntity = originalEntities != null ? originalEntities.get(i) : null;
+ // TODO: This isn't quite correct right now, because originalEntity is only actually
+ // safe to use for entityVersion and grantRecordsVersion right now. Once we refactor
+ // the writeEntities[] methods to take something like PolarisEntityCore
+ // for originalEntity and force the callsites such as BasePolarisCatalog to actually
+ // provide the original values, this will be correct. For now, the API does't support
+ // bulk renames anyways.
+ boolean nameOrParentChanged =
+ originalEntity == null
+ || !entity.getName().equals(originalEntity.getName())
+ || entity.getParentId() != originalEntity.getParentId();
+ this.writeEntityInCurrentTxn(callCtx, entity, nameOrParentChanged, originalEntity);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public void deleteEntityInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
+ this.deleteFromEntitiesActiveInCurrentTxn(callCtx, entity);
+ this.deleteFromEntitiesInCurrentTxn(callCtx, entity);
+ this.deleteFromEntitiesChangeTrackingInCurrentTxn(callCtx, entity);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public PolarisBaseEntity lookupEntityByNameInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ int typeCode,
+ @Nonnull String name) {
+ PolarisEntitiesActiveKey entityActiveKey =
+ new PolarisEntitiesActiveKey(catalogId, parentId, typeCode, name);
+
+ // ensure that the entity exists
+ EntityNameLookupRecord entityActiveRecord =
+ this.lookupEntityActiveInCurrentTxn(callCtx, entityActiveKey);
+
+ // if not found, return null
+ if (entityActiveRecord == null) {
+ return null;
+ }
+
+ // lookup the entity, should be there
+ PolarisBaseEntity entity =
+ this.lookupEntityInCurrentTxn(
+ callCtx,
+ entityActiveRecord.getCatalogId(),
+ entityActiveRecord.getId(),
+ entityActiveRecord.getTypeCode());
+ callCtx
+ .getDiagServices()
+ .checkNotNull(
+ entity, "unexpected_not_found_entity", "entityActiveRecord={}", entityActiveRecord);
+
+ // return it now
+ return entity;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ @Nullable
+ public EntityNameLookupRecord lookupEntityIdAndSubTypeByNameInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ int typeCode,
+ @Nonnull String name) {
+ PolarisEntitiesActiveKey entityActiveKey =
+ new PolarisEntitiesActiveKey(catalogId, parentId, typeCode, name);
+ return this.lookupEntityActiveInCurrentTxn(callCtx, entityActiveKey);
+ }
}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisEntityResolver.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisEntityResolver.java
index bb4e39591..83e8123df 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisEntityResolver.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisEntityResolver.java
@@ -283,7 +283,7 @@ public class PolarisEntityResolver {
// now lookup all these entities by name
Iterator<EntityNameLookupRecord> activeRecordIt =
- ms.lookupEntityActiveBatch(callCtx, entityActiveKeys).iterator();
+ ms.lookupEntityActiveBatchInCurrentTxn(callCtx, entityActiveKeys).iterator();
// now validate if there was a change and if yes, re-resolve again
for (PolarisEntityCore resolveEntity : toResolve) {
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java
index da09a6799..e89bc3a45 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisMetaStoreManagerImpl.java
@@ -18,10 +18,6 @@
*/
package org.apache.polaris.core.persistence.transactional;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.JsonMappingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Nullable;
@@ -59,7 +55,6 @@ import org.apache.polaris.core.persistence.dao.entity.DropEntityResult;
import org.apache.polaris.core.persistence.dao.entity.EntitiesResult;
import org.apache.polaris.core.persistence.dao.entity.EntityResult;
import org.apache.polaris.core.persistence.dao.entity.EntityWithPath;
-import org.apache.polaris.core.persistence.dao.entity.GenerateEntityIdResult;
import org.apache.polaris.core.persistence.dao.entity.ListEntitiesResult;
import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
import org.apache.polaris.core.storage.PolarisCredentialProperty;
@@ -77,77 +72,32 @@ import org.slf4j.LoggerFactory;
public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
private static final Logger LOGGER = LoggerFactory.getLogger(PolarisMetaStoreManagerImpl.class);
- /** mapper, allows to serialize/deserialize properties to/from JSON */
- private static final ObjectMapper MAPPER = new ObjectMapper();
-
/**
- * Persist the specified new entity. Persist will write this entity in the ENTITIES, in the
- * ENTITIES_ACTIVE and finally in the ENTITIES_CHANGE_TRACKING tables
+ * A version of BaseMetaStoreManager::persistNewEntity but instead of calling the one-shot
+ * immediate-peristence APIs of BasePersistence, expects to be run under an outer
+ * runInTransaction, and calls through to analogous versions of * methods of
+ * TransactionalPersistence.
*
* @param callCtx call context
* @param ms meta store in read/write mode
- * @param entity entity we need a DPO for
+ * @param entity entity we need a new persisted record for
*/
- private void persistNewEntity(
+ protected void persistNewEntity(
@Nonnull PolarisCallContext callCtx,
@Nonnull TransactionalPersistence ms,
@Nonnull PolarisBaseEntity entity) {
-
- // validate the entity type and subtype
- callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
- callCtx
- .getDiagServices()
- .checkNotNull(entity.getName(), "unexpected_null_name", "entity={}", entity);
- PolarisEntityType type = PolarisEntityType.fromCode(entity.getTypeCode());
- callCtx.getDiagServices().checkNotNull(type, "unknown_type", "entity={}", entity);
- PolarisEntitySubType subType = PolarisEntitySubType.fromCode(entity.getSubTypeCode());
- callCtx.getDiagServices().checkNotNull(subType, "unexpected_null_subType", "entity={}", entity);
- callCtx
- .getDiagServices()
- .check(
- subType.getParentType() == null || subType.getParentType() == type,
- "invalid_subtype",
- "type={} subType={}",
- type,
- subType);
-
- // if top-level entity, its parent should be the account
- callCtx
- .getDiagServices()
- .check(
- !type.isTopLevel() || entity.getParentId() == PolarisEntityConstants.getRootEntityId(),
- "top_level_parent_should_be_account",
- "entity={}",
- entity);
-
- // id should not be null
- callCtx
- .getDiagServices()
- .check(
- entity.getId() != 0 || type == PolarisEntityType.ROOT,
- "id_not_set",
- "entity={}",
- entity);
-
- // creation timestamp must be filled
- callCtx.getDiagServices().check(entity.getCreateTimestamp() != 0, "null_create_timestamp");
-
- // this is the first change
- entity.setLastUpdateTimestamp(entity.getCreateTimestamp());
-
- // set all other timestamps to 0
- entity.setDropTimestamp(0);
- entity.setPurgeTimestamp(0);
- entity.setToPurgeTimestamp(0);
+ // Invoke shared logic for validation and filling out remaining fields.
+ entity = prepareToPersistNewEntity(callCtx, ms, entity);
// write it
- ms.writeEntity(callCtx, entity, true, null);
+ ms.writeEntityInCurrentTxn(callCtx, entity, true, null);
}
/**
- * Persist the specified entity after it has been changed. We will update the last changed time,
- * increment the entity version and persist it back to the ENTITIES and ENTITIES_CHANGE_TRACKING
- * tables
+ * A version of BaseMetaStoreManager::persistEntityAfterChange but instead of calling the one-shot
+ * immediate-peristence APIs of BasePersistence, expects to be run under an outer
+ * runInTransaction, and calls through to analogous versions of * methods of
+ * TransactionalPersistence.
*
* @param callCtx call context
* @param ms meta store
@@ -162,49 +112,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisBaseEntity entity,
boolean nameOrParentChanged,
@Nonnull PolarisBaseEntity originalEntity) {
+ // Invoke shared logic for validation and updating expected fields.
+ entity =
+ prepareToPersistEntityAfterChange(callCtx, ms, entity, nameOrParentChanged, originalEntity);
- // validate the entity type and subtype
- callCtx.getDiagServices().checkNotNull(entity, "unexpected_null_entity");
- callCtx
- .getDiagServices()
- .checkNotNull(entity.getName(), "unexpected_null_name", "entity={}", entity);
- PolarisEntityType type = entity.getType();
- callCtx.getDiagServices().checkNotNull(type, "unexpected_null_type", "entity={}", entity);
- PolarisEntitySubType subType = entity.getSubType();
- callCtx.getDiagServices().checkNotNull(subType, "unexpected_null_subType", "entity={}", entity);
- callCtx
- .getDiagServices()
- .check(
- subType.getParentType() == null || subType.getParentType() == type,
- "invalid_subtype",
- "type={} subType={} entity={}",
- type,
- subType,
- entity);
-
- // entity should not have been dropped
- callCtx
- .getDiagServices()
- .check(entity.getDropTimestamp() == 0, "entity_dropped", "entity={}", entity);
-
- // creation timestamp must be filled
- long createTimestamp = entity.getCreateTimestamp();
- callCtx
- .getDiagServices()
- .check(createTimestamp != 0, "null_create_timestamp", "entity={}", entity);
-
- // ensure time is not moving backward...
- long now = System.currentTimeMillis();
- if (now < entity.getCreateTimestamp()) {
- now = entity.getCreateTimestamp() + 1;
- }
-
- // update last update timestamp and increment entity version
- entity.setLastUpdateTimestamp(now);
- entity.setEntityVersion(entity.getEntityVersion() + 1);
-
- // persist it to the various slices
- ms.writeEntity(callCtx, entity, nameOrParentChanged, originalEntity);
+ // Use the write method defined in TransactionalPersistence which expects an
+ // existing runInTransaction to already be in-place.
+ ms.writeEntityInCurrentTxn(callCtx, entity, nameOrParentChanged, originalEntity);
// return it
return entity;
@@ -240,11 +154,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// delete ALL grant records to (if the entity is a grantee) and from that entity
final List<PolarisGrantRecord> grantsOnGrantee =
(entity.getType().isGrantee())
- ? ms.loadAllGrantRecordsOnGrantee(callCtx, entity.getCatalogId(), entity.getId())
+ ? ms.loadAllGrantRecordsOnGranteeInCurrentTxn(
+ callCtx, entity.getCatalogId(), entity.getId())
: List.of();
final List<PolarisGrantRecord> grantsOnSecurable =
- ms.loadAllGrantRecordsOnSecurable(callCtx, entity.getCatalogId(), entity.getId());
- ms.deleteAllEntityGrantRecords(callCtx, entity, grantsOnGrantee, grantsOnSecurable);
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(
+ callCtx, entity.getCatalogId(), entity.getId());
+ ms.deleteAllEntityGrantRecordsInCurrentTxn(callCtx, entity, grantsOnGrantee, grantsOnSecurable);
// Now determine the set of entities on the other side of the grants we just removed. Grants
// from/to these entities has been removed, hence we need to update the grant version of
@@ -261,15 +177,15 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// Bump up the grant version of these entities
List<PolarisBaseEntity> entities =
- ms.lookupEntities(callCtx, new ArrayList<>(entityIdsGrantChanged));
+ ms.lookupEntitiesInCurrentTxn(callCtx, new ArrayList<>(entityIdsGrantChanged));
for (PolarisBaseEntity entityGrantChanged : entities) {
PolarisBaseEntity originalEntity = new PolarisBaseEntity(entityGrantChanged);
entityGrantChanged.setGrantRecordsVersion(entityGrantChanged.getGrantRecordsVersion() + 1);
- ms.writeEntity(callCtx, entityGrantChanged, false, originalEntity);
+ ms.writeEntityInCurrentTxn(callCtx, entityGrantChanged, false, originalEntity);
}
// remove the entity being dropped now
- ms.deleteEntity(callCtx, entity);
+ ms.deleteEntityInCurrentTxn(callCtx, entity);
// if it is a principal, we also need to drop the secrets
if (entity.getType() == PolarisEntityType.PRINCIPAL) {
@@ -281,8 +197,10 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
String clientId = properties.get(PolarisEntityConstants.getClientIdPropertyName());
// delete it from the secret slice
- ms.deletePrincipalSecrets(callCtx, clientId, entity.getId());
+ ms.deletePrincipalSecretsInCurrentTxn(callCtx, clientId, entity.getId());
}
+ // TODO: Also, if an entity contains a storage integration, delete the storage integration
+ // and other things of that nature.
}
/**
@@ -323,12 +241,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
priv.getCode());
// persist the new grant
- ms.writeToGrantRecords(callCtx, grantRecord);
+ ms.writeToGrantRecordsInCurrentTxn(callCtx, grantRecord);
// load the grantee (either a catalog/principal role or a principal) and increment its grants
// version
PolarisBaseEntity granteeEntity =
- ms.lookupEntity(callCtx, grantee.getCatalogId(), grantee.getId(), grantee.getTypeCode());
+ ms.lookupEntityInCurrentTxn(
+ callCtx, grantee.getCatalogId(), grantee.getId(), grantee.getTypeCode());
callCtx
.getDiagServices()
.checkNotNull(granteeEntity, "grantee_not_found", "grantee={}", grantee);
@@ -336,12 +255,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// grants have changed, we need to bump-up the grants version
granteeEntity.setGrantRecordsVersion(granteeEntity.getGrantRecordsVersion() + 1);
- ms.writeEntity(callCtx, granteeEntity, false, originalGranteeEntity);
+ ms.writeEntityInCurrentTxn(callCtx, granteeEntity, false, originalGranteeEntity);
// we also need to invalidate the grants on that securable so that we can reload them.
// load the securable and increment its grants version
PolarisBaseEntity securableEntity =
- ms.lookupEntity(
+ ms.lookupEntityInCurrentTxn(
callCtx, securable.getCatalogId(), securable.getId(), securable.getTypeCode());
callCtx
.getDiagServices()
@@ -350,7 +269,10 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// grants have changed, we need to bump-up the grants version
securableEntity.setGrantRecordsVersion(securableEntity.getGrantRecordsVersion() + 1);
- ms.writeEntity(callCtx, securableEntity, false, originalSecurableEntity);
+ ms.writeEntityInCurrentTxn(callCtx, securableEntity, false, originalSecurableEntity);
+
+ // TODO: Update this to be an atomic bulk-update of the grantee/securable, ideally along
+ // with adding the grant record in the same bulk-update.
// done, return the new grant record
return grantRecord;
@@ -401,11 +323,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
.check(grantee.getType().isGrantee(), "not_a_grantee", "grantee={}", grantee);
// remove that grant
- ms.deleteFromGrantRecords(callCtx, grantRecord);
+ ms.deleteFromGrantRecordsInCurrentTxn(callCtx, grantRecord);
// load the grantee and increment its grants version
PolarisBaseEntity refreshGrantee =
- ms.lookupEntity(callCtx, grantee.getCatalogId(), grantee.getId(), grantee.getTypeCode());
+ ms.lookupEntityInCurrentTxn(
+ callCtx, grantee.getCatalogId(), grantee.getId(), grantee.getTypeCode());
callCtx
.getDiagServices()
.checkNotNull(
@@ -414,12 +337,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// grants have changed, we need to bump-up the grants version
refreshGrantee.setGrantRecordsVersion(refreshGrantee.getGrantRecordsVersion() + 1);
- ms.writeEntity(callCtx, refreshGrantee, false, originalRefreshGrantee);
+ ms.writeEntityInCurrentTxn(callCtx, refreshGrantee, false, originalRefreshGrantee);
// we also need to invalidate the grants on that securable so that we can reload them.
// load the securable and increment its grants version
PolarisBaseEntity refreshSecurable =
- ms.lookupEntity(
+ ms.lookupEntityInCurrentTxn(
callCtx, securable.getCatalogId(), securable.getId(), securable.getTypeCode());
callCtx
.getDiagServices()
@@ -433,7 +356,10 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// grants have changed, we need to bump-up the grants version
refreshSecurable.setGrantRecordsVersion(refreshSecurable.getGrantRecordsVersion() + 1);
- ms.writeEntity(callCtx, refreshSecurable, false, originalRefreshSecurable);
+ ms.writeEntityInCurrentTxn(callCtx, refreshSecurable, false, originalRefreshSecurable);
+
+ // TODO: Update this to be an atomic bulk-update of the grantee/securable, ideally along
+ // with removing the grant record in the same bulk-update.
}
/**
@@ -462,7 +388,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// check if that catalog has already been created
PolarisBaseEntity refreshCatalog =
- ms.lookupEntity(callCtx, catalog.getCatalogId(), catalog.getId(), catalog.getTypeCode());
+ ms.lookupEntityInCurrentTxn(
+ callCtx, catalog.getCatalogId(), catalog.getId(), catalog.getTypeCode());
// if found, probably a retry, simply return the previously created catalog
if (refreshCatalog != null) {
@@ -477,7 +404,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// lookup catalog admin role, should exist
PolarisBaseEntity catalogAdminRole =
- ms.lookupEntityByName(
+ ms.lookupEntityByNameInCurrentTxn(
callCtx,
refreshCatalog.getId(),
refreshCatalog.getId(),
@@ -496,7 +423,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// check that a catalog with the same name does not exist already
// if it exists, this is an error, the client should retry
- if (ms.lookupEntityIdAndSubTypeByName(
+ if (ms.lookupEntityIdAndSubTypeByNameInCurrentTxn(
callCtx,
PolarisEntityConstants.getNullId(),
PolarisEntityConstants.getRootEntityId(),
@@ -506,13 +433,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
return new CreateCatalogResult(BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS, null);
}
- ms.persistStorageIntegrationIfNeeded(callCtx, catalog, integration);
+ ms.persistStorageIntegrationIfNeededInCurrentTxn(callCtx, catalog, integration);
// now create and persist new catalog entity
this.persistNewEntity(callCtx, ms, catalog);
// create the catalog admin role for this new catalog
- long adminRoleId = ms.generateNewId(callCtx);
+ long adminRoleId = ms.generateNewIdInCurrentTxn(callCtx);
PolarisBaseEntity adminRole =
new PolarisBaseEntity(
catalog.getId(),
@@ -536,7 +463,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
if (principalRoles.isEmpty()) {
// lookup service admin role, should exist
PolarisBaseEntity serviceAdminRole =
- ms.lookupEntityByName(
+ ms.lookupEntityByNameInCurrentTxn(
callCtx,
PolarisEntityConstants.getNullId(),
PolarisEntityConstants.getRootEntityId(),
@@ -591,7 +518,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// Now bootstrap the service by creating the root principal and the service_admin principal
// role. The principal role will be granted to that root principal and the root catalog admin
// of the root catalog will be granted to that principal role.
- long rootPrincipalId = ms.generateNewId(callCtx);
+ long rootPrincipalId = ms.generateNewIdInCurrentTxn(callCtx);
PolarisBaseEntity rootPrincipal =
new PolarisBaseEntity(
PolarisEntityConstants.getNullId(),
@@ -605,7 +532,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
this.createPrincipal(callCtx, ms, rootPrincipal);
// now create the account admin principal role
- long serviceAdminPrincipalRoleId = ms.generateNewId(callCtx);
+ long serviceAdminPrincipalRoleId = ms.generateNewIdInCurrentTxn(callCtx);
PolarisBaseEntity serviceAdminPrincipalRole =
new PolarisBaseEntity(
PolarisEntityConstants.getNullId(),
@@ -637,7 +564,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Override
public @Nonnull BaseResult bootstrapPolarisService(@Nonnull PolarisCallContext callCtx) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// run operation in a read/write transaction
ms.runActionInTransaction(callCtx, () -> this.bootstrapPolarisService(callCtx, ms));
@@ -649,11 +576,11 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Override
public @Nonnull BaseResult purge(@Nonnull PolarisCallContext callCtx) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// run operation in a read/write transaction
LOGGER.warn("Deleting all metadata in the metastore...");
- ms.runActionInTransaction(callCtx, () -> ms.deleteAll(callCtx));
+ ms.runActionInTransaction(callCtx, () -> ms.deleteAllInCurrentTxn(callCtx));
LOGGER.warn("Finished deleting all metadata in the metastore");
// all good
@@ -681,7 +608,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// now looking the entity by name
PolarisBaseEntity entity =
- ms.lookupEntityByName(
+ ms.lookupEntityByNameInCurrentTxn(
callCtx,
resolver.getCatalogIdOrNull(),
resolver.getParentId(),
@@ -710,7 +637,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntitySubType entitySubType,
@Nonnull String name) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// run operation in a read/write transaction
return ms.runInReadTransaction(
@@ -736,7 +663,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// return list of active entities
List<EntityNameLookupRecord> toreturnList =
- ms.listEntities(callCtx, resolver.getCatalogIdOrNull(), resolver.getParentId(), entityType);
+ ms.listEntitiesInCurrentTxn(
+ callCtx, resolver.getCatalogIdOrNull(), resolver.getParentId(), entityType);
// prune the returned list with only entities matching the entity subtype
if (entitySubType != PolarisEntitySubType.ANY_SUBTYPE) {
@@ -758,62 +686,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntityType entityType,
@Nonnull PolarisEntitySubType entitySubType) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// run operation in a read transaction
return ms.runInReadTransaction(
callCtx, () -> listEntities(callCtx, ms, catalogPath, entityType, entitySubType));
}
- /** {@inheritDoc} */
- @Override
- public @Nonnull GenerateEntityIdResult generateNewEntityId(@Nonnull PolarisCallContext callCtx) {
- // get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
-
- return new GenerateEntityIdResult(ms.generateNewId(callCtx));
- }
-
- /**
- * Given the internal property as a map of key/value pairs, serialize it to a String
- *
- * @param properties a map of key/value pairs
- * @return a String, the JSON representation of the map
- */
- public String serializeProperties(PolarisCallContext callCtx, Map<String, String> properties) {
-
- String jsonString = null;
- try {
- // Deserialize the JSON string to a Map<String, String>
- jsonString = MAPPER.writeValueAsString(properties);
- } catch (JsonProcessingException ex) {
- callCtx.getDiagServices().fail("got_json_processing_exception", "ex={}", ex);
- }
-
- return jsonString;
- }
-
- /**
- * Given the serialized properties, deserialize those to a {@code Map<String, String>}
- *
- * @param properties a JSON string representing the set of properties
- * @return a Map of string
- */
- public Map<String, String> deserializeProperties(PolarisCallContext callCtx, String properties) {
-
- Map<String, String> retProperties = null;
- try {
- // Deserialize the JSON string to a Map<String, String>
- retProperties = MAPPER.readValue(properties, new TypeReference<>() {});
- } catch (JsonMappingException ex) {
- callCtx.getDiagServices().fail("got_json_mapping_exception", "ex={}", ex);
- } catch (JsonProcessingException ex) {
- callCtx.getDiagServices().fail("got_json_processing_exception", "ex={}", ex);
- }
-
- return retProperties;
- }
-
/** {@link #createPrincipal(PolarisCallContext, PolarisBaseEntity)} */
private @Nonnull CreatePrincipalResult createPrincipal(
@Nonnull PolarisCallContext callCtx,
@@ -824,7 +703,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// check if that catalog has already been created
PolarisBaseEntity refreshPrincipal =
- ms.lookupEntity(
+ ms.lookupEntityInCurrentTxn(
callCtx, principal.getCatalogId(), principal.getId(), principal.getTypeCode());
// if found, probably a retry, simply return the previously created principal
@@ -863,7 +742,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
refreshPrincipal.getInternalProperties());
// get the main and secondary secrets for that client
- PolarisPrincipalSecrets principalSecrets = ms.loadPrincipalSecrets(callCtx, clientId);
+ PolarisPrincipalSecrets principalSecrets =
+ ms.loadPrincipalSecretsInCurrentTxn(callCtx, clientId);
// should not be null
callCtx
@@ -881,7 +761,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// check that a principal with the same name does not exist already
// if it exists, this is an error, the client should retry
- if (ms.lookupEntityIdAndSubTypeByName(
+ if (ms.lookupEntityIdAndSubTypeByNameInCurrentTxn(
callCtx,
PolarisEntityConstants.getNullId(),
PolarisEntityConstants.getRootEntityId(),
@@ -893,7 +773,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// generate new secrets for this principal
PolarisPrincipalSecrets principalSecrets =
- ms.generateNewPrincipalSecrets(callCtx, principal.getName(), principal.getId());
+ ms.generateNewPrincipalSecretsInCurrentTxn(callCtx, principal.getName(), principal.getId());
// generate properties
Map<String, String> internalProperties = getInternalPropertyMap(callCtx, principal);
@@ -915,7 +795,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
public @Nonnull CreatePrincipalResult createPrincipal(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity principal) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(callCtx, () -> this.createPrincipal(callCtx, ms, principal));
@@ -924,7 +804,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
/** See {@link #loadPrincipalSecrets(PolarisCallContext, String)} */
private @Nullable PolarisPrincipalSecrets loadPrincipalSecrets(
@Nonnull PolarisCallContext callCtx, TransactionalPersistence ms, @Nonnull String clientId) {
- return ms.loadPrincipalSecrets(callCtx, clientId);
+ return ms.loadPrincipalSecretsInCurrentTxn(callCtx, clientId);
}
/** {@inheritDoc} */
@@ -932,7 +812,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
public @Nonnull PrincipalSecretsResult loadPrincipalSecrets(
@Nonnull PolarisCallContext callCtx, @Nonnull String clientId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
PolarisPrincipalSecrets secrets =
@@ -976,7 +856,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)
!= null;
PolarisPrincipalSecrets secrets =
- ms.rotatePrincipalSecrets(callCtx, clientId, principalId, doReset, oldSecretHash);
+ ms.rotatePrincipalSecretsInCurrentTxn(
+ callCtx, clientId, principalId, doReset, oldSecretHash);
if (reset
&& !internalProps.containsKey(
@@ -986,14 +867,14 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
principal.setInternalProperties(
PolarisObjectMapperUtil.serializeProperties(callCtx, internalProps));
principal.setEntityVersion(principal.getEntityVersion() + 1);
- ms.writeEntity(callCtx, principal, true, originalPrincipal);
+ ms.writeEntityInCurrentTxn(callCtx, principal, true, originalPrincipal);
} else if (internalProps.containsKey(
PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE)) {
internalProps.remove(PolarisEntityConstants.PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_STATE);
principal.setInternalProperties(
PolarisObjectMapperUtil.serializeProperties(callCtx, internalProps));
principal.setEntityVersion(principal.getEntityVersion() + 1);
- ms.writeEntity(callCtx, principal, true, originalPrincipal);
+ ms.writeEntityInCurrentTxn(callCtx, principal, true, originalPrincipal);
}
return secrets;
}
@@ -1007,7 +888,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
boolean reset,
@Nonnull String oldSecretHash) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
PolarisPrincipalSecrets secrets =
@@ -1029,7 +910,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisBaseEntity catalog,
@Nonnull List<PolarisEntityCore> principalRoles) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
Map<String, String> internalProp = getInternalPropertyMap(callCtx, catalog);
String integrationIdentifierOrId =
@@ -1041,7 +922,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// and the catalog should not have an internal property of storage identifier or id yet
if (storageConfigInfoStr != null && integrationIdentifierOrId == null) {
integration =
- ms.createStorageIntegration(
+ ms.createStorageIntegrationInCurrentTxn(
callCtx,
catalog.getCatalogId(),
catalog.getId(),
@@ -1070,9 +951,12 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// first, check if the entity has already been created, in which case we will simply return it
PolarisBaseEntity entityFound =
- ms.lookupEntity(callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
+ ms.lookupEntityInCurrentTxn(
+ callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
if (entityFound != null) {
// probably the client retried, simply return it
+ // TODO: Check correctness of returning entityFound vs entity here. It may have already
+ // been updated after the creation.
return new EntityResult(entityFound);
}
@@ -1086,7 +970,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// check if an entity does not already exist with the same name. If true, this is an error
EntityNameLookupRecord entityActiveRecord =
- ms.lookupEntityIdAndSubTypeByName(
+ ms.lookupEntityIdAndSubTypeByNameInCurrentTxn(
callCtx,
entity.getCatalogId(),
entity.getParentId(),
@@ -1111,7 +995,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisBaseEntity entity) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1124,7 +1008,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull List<? extends PolarisBaseEntity> entities) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1167,13 +1051,26 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// lookup the entity, cannot be null
PolarisBaseEntity entityRefreshed =
- ms.lookupEntity(callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
+ ms.lookupEntityInCurrentTxn(
+ callCtx, entity.getCatalogId(), entity.getId(), entity.getTypeCode());
callCtx
.getDiagServices()
.checkNotNull(entityRefreshed, "unexpected_entity_not_found", "entity={}", entity);
// check that the version of the entity has not changed at all to avoid concurrent updates
if (entityRefreshed.getEntityVersion() != entity.getEntityVersion()) {
+ // TODO: Check if this is a bug that might cause grantRecordVersions to go backwards. The
+ // behavior for grantRecordVersions updates could be either to still allow the update
+ // while taking the higher grantRecordsVersion, or to hard-fail. If we still allow the
+ // update, then we're only guaranteeing that the caller was authorized for the update
+ // *at some point in time* in the vicinity of an updated grant. If we don't have retries
+ // that re-fetch the entity properties and try to re-apply updates server-side this is
+ // probably fine, but if we do have server-side retries *after* grantRecordVersion
+ // resolution, this could mean a bug where a grant is revoked before something else
+ // already modifies the table in a way that is only safe with that grant revoked,
+ // while the retry will allow the update to go through. If we consider grantRecordsVersion
+ // a strict part of the version comparison here then we have well-ordering guarantees.
+ // TODO: Give useful extraInformation
return new EntityResult(BaseResult.ReturnStatus.TARGET_ENTITY_CONCURRENTLY_MODIFIED, null);
}
@@ -1197,7 +1094,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nullable List<PolarisEntityCore> catalogPath,
@Nonnull PolarisBaseEntity entity) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1242,7 +1139,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
public @Nonnull EntitiesResult updateEntitiesPropertiesIfNotChanged(
@Nonnull PolarisCallContext callCtx, @Nonnull List<EntityWithPath> entities) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1289,7 +1186,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// find the entity to rename
PolarisBaseEntity refreshEntityToRename =
- ms.lookupEntity(
+ ms.lookupEntityInCurrentTxn(
callCtx,
entityToRename.getCatalogId(),
entityToRename.getId(),
@@ -1324,7 +1221,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// ensure that nothing exists where we create that entity
// if this entity already exists, this is an error
EntityNameLookupRecord entityActiveRecord =
- ms.lookupEntityIdAndSubTypeByName(
+ ms.lookupEntityIdAndSubTypeByNameInCurrentTxn(
callCtx,
resolver.getCatalogIdOrNull(),
resolver.getParentId(),
@@ -1366,7 +1263,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nullable List<PolarisEntityCore> newCatalogPath,
@Nonnull PolarisEntity renamedEntity) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1402,7 +1299,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// first find the entity to drop
PolarisBaseEntity refreshEntityToDrop =
- ms.lookupEntity(
+ ms.lookupEntityInCurrentTxn(
callCtx, entityToDrop.getCatalogId(), entityToDrop.getId(), entityToDrop.getTypeCode());
// if this entity was not found, return failure
@@ -1422,13 +1319,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long catalogId = refreshEntityToDrop.getId();
// if not all namespaces are dropped, we cannot drop this catalog
- if (ms.hasChildren(callCtx, PolarisEntityType.NAMESPACE, catalogId, catalogId)) {
+ if (ms.hasChildrenInCurrentTxn(callCtx, PolarisEntityType.NAMESPACE, catalogId, catalogId)) {
return new DropEntityResult(BaseResult.ReturnStatus.NAMESPACE_NOT_EMPTY, null);
}
// get the list of catalog roles, at most 2
List<PolarisBaseEntity> catalogRoles =
- ms.listEntities(
+ ms.listEntitiesInCurrentTxn(
callCtx,
catalogId,
catalogId,
@@ -1448,7 +1345,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
this.dropEntity(callCtx, ms, catalogRoles.get(0));
}
} else if (refreshEntityToDrop.getType() == PolarisEntityType.NAMESPACE) {
- if (ms.hasChildren(
+ if (ms.hasChildrenInCurrentTxn(
callCtx, null, refreshEntityToDrop.getCatalogId(), refreshEntityToDrop.getId())) {
return new DropEntityResult(BaseResult.ReturnStatus.NAMESPACE_NOT_EMPTY, null);
}
@@ -1464,7 +1361,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
if (cleanup) {
PolarisBaseEntity taskEntity =
new PolarisEntity.Builder()
- .setId(generateNewEntityId(callCtx).getId())
+ .setId(ms.generateNewIdInCurrentTxn(callCtx))
.setCatalogId(0L)
.setName("entityCleanup_" + entityToDrop.getId())
.setType(PolarisEntityType.TASK)
@@ -1499,7 +1396,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nullable Map<String, String> cleanupProperties,
boolean cleanup) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1637,7 +1534,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntityCore role,
@Nonnull PolarisEntityCore grantee) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1672,7 +1569,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// first, ensure that this privilege has been granted
PolarisGrantRecord grantRecord =
- ms.lookupGrantRecord(
+ ms.lookupGrantRecordInCurrentTxn(
callCtx,
role.getCatalogId(),
role.getId(),
@@ -1699,7 +1596,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntityCore role,
@Nonnull PolarisEntityCore grantee) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1742,7 +1639,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntityCore securable,
@Nonnull PolarisPrivilege privilege) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1775,7 +1672,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// lookup the grants records to find this grant
PolarisGrantRecord grantRecord =
- ms.lookupGrantRecord(
+ ms.lookupGrantRecordInCurrentTxn(
callCtx,
securable.getCatalogId(),
securable.getId(),
@@ -1804,7 +1701,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntityCore securable,
@Nonnull PolarisPrivilege privilege) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read/write transaction
return ms.runInTransaction(
@@ -1823,7 +1720,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// lookup grants version for this securable entity
int grantsVersion =
- ms.lookupEntityGrantRecordsVersion(callCtx, securableCatalogId, securableId);
+ ms.lookupEntityGrantRecordsVersionInCurrentTxn(callCtx, securableCatalogId, securableId);
// return null if securable does not exists
if (grantsVersion == 0) {
@@ -1832,7 +1729,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// now fetch all grants for this securable
final List<PolarisGrantRecord> returnGrantRecords =
- ms.loadAllGrantRecordsOnSecurable(callCtx, securableCatalogId, securableId);
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, securableCatalogId, securableId);
// find all unique grantees
List<PolarisEntityId> entityIds =
@@ -1843,7 +1740,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
grantRecord.getGranteeCatalogId(), grantRecord.getGranteeId()))
.distinct()
.collect(Collectors.toList());
- List<PolarisBaseEntity> entities = ms.lookupEntities(callCtx, entityIds);
+ List<PolarisBaseEntity> entities = ms.lookupEntitiesInCurrentTxn(callCtx, entityIds);
// done, return the list of grants and their version
return new LoadGrantsResult(
@@ -1857,7 +1754,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
public @Nonnull LoadGrantsResult loadGrantsOnSecurable(
@Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
return ms.runInReadTransaction(
@@ -1872,7 +1769,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long granteeId) {
// lookup grants version for this grantee entity
- int grantsVersion = ms.lookupEntityGrantRecordsVersion(callCtx, granteeCatalogId, granteeId);
+ int grantsVersion =
+ ms.lookupEntityGrantRecordsVersionInCurrentTxn(callCtx, granteeCatalogId, granteeId);
// return null if grantee does not exists
if (grantsVersion == 0) {
@@ -1881,7 +1779,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// now fetch all grants for this grantee
final List<PolarisGrantRecord> returnGrantRecords =
- ms.loadAllGrantRecordsOnGrantee(callCtx, granteeCatalogId, granteeId);
+ ms.loadAllGrantRecordsOnGranteeInCurrentTxn(callCtx, granteeCatalogId, granteeId);
// find all unique securables
List<PolarisEntityId> entityIds =
@@ -1892,7 +1790,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
grantRecord.getSecurableCatalogId(), grantRecord.getSecurableId()))
.distinct()
.collect(Collectors.toList());
- List<PolarisBaseEntity> entities = ms.lookupEntities(callCtx, entityIds);
+ List<PolarisBaseEntity> entities = ms.lookupEntitiesInCurrentTxn(callCtx, entityIds);
// done, return the list of grants and their version
return new LoadGrantsResult(
@@ -1906,7 +1804,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
public @Nonnull LoadGrantsResult loadGrantsToGrantee(
@Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
return ms.runInReadTransaction(
@@ -1919,7 +1817,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull TransactionalPersistence ms,
@Nonnull List<PolarisEntityId> entityIds) {
List<PolarisChangeTrackingVersions> changeTracking =
- ms.lookupEntityVersions(callCtx, entityIds);
+ ms.lookupEntityVersionsInCurrentTxn(callCtx, entityIds);
return new ChangeTrackingResult(changeTracking);
}
@@ -1928,7 +1826,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
public @Nonnull ChangeTrackingResult loadEntitiesChangeTracking(
@Nonnull PolarisCallContext callCtx, @Nonnull List<PolarisEntityId> entityIds) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
return ms.runInReadTransaction(
@@ -1943,7 +1841,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long entityId,
int entityTypeCode) {
// this is an easy one
- PolarisBaseEntity entity = ms.lookupEntity(callCtx, entityCatalogId, entityId, entityTypeCode);
+ PolarisBaseEntity entity =
+ ms.lookupEntityInCurrentTxn(callCtx, entityCatalogId, entityId, entityTypeCode);
return (entity != null)
? new EntityResult(entity)
: new EntityResult(BaseResult.ReturnStatus.ENTITY_NOT_FOUND, null);
@@ -1957,7 +1856,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long entityId,
@Nonnull PolarisEntityType entityType) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
return ms.runInReadTransaction(
@@ -1974,7 +1873,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// find all available tasks
List<PolarisBaseEntity> availableTasks =
- ms.listEntities(
+ ms.listEntitiesInCurrentTxn(
callCtx,
PolarisEntityConstants.getRootEntityId(),
PolarisEntityConstants.getRootEntityId(),
@@ -2034,7 +1933,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Override
public @Nonnull EntitiesResult loadTasks(
@Nonnull PolarisCallContext callCtx, String executorId, int limit) {
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
return ms.runInTransaction(callCtx, () -> this.loadTasks(callCtx, ms, executorId, limit));
}
@@ -2050,7 +1949,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull Set<String> allowedWriteLocations) {
// get meta store session we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
callCtx
.getDiagServices()
.check(
@@ -2066,7 +1965,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// get storage integration
PolarisStorageIntegration<PolarisStorageConfigurationInfo> storageIntegration =
- ms.loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
+ ms.loadPolarisStorageIntegrationInCurrentTxn(callCtx, reloadedEntity.getEntity());
// cannot be null
callCtx
@@ -2105,7 +2004,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull Set<PolarisStorageActions> actions,
@Nonnull Set<String> locations) {
// get meta store we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
callCtx
.getDiagServices()
.check(
@@ -2120,7 +2019,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// get storage integration, expect not null
PolarisStorageIntegration<PolarisStorageConfigurationInfo> storageIntegration =
- ms.loadPolarisStorageIntegration(callCtx, reloadedEntity.getEntity());
+ ms.loadPolarisStorageIntegrationInCurrentTxn(callCtx, reloadedEntity.getEntity());
callCtx
.getDiagServices()
.checkNotNull(
@@ -2173,7 +2072,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
int typeCode) {
// load that entity
- PolarisBaseEntity entity = ms.lookupEntity(callCtx, entityCatalogId, entityId, typeCode);
+ PolarisBaseEntity entity =
+ ms.lookupEntityInCurrentTxn(callCtx, entityCatalogId, entityId, typeCode);
// if entity not found, return null
if (entity == null) {
@@ -2184,10 +2084,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
final List<PolarisGrantRecord> grantRecords;
if (entity.getType().isGrantee()) {
grantRecords =
- new ArrayList<>(ms.loadAllGrantRecordsOnGrantee(callCtx, entityCatalogId, entityId));
- grantRecords.addAll(ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entityId));
+ new ArrayList<>(
+ ms.loadAllGrantRecordsOnGranteeInCurrentTxn(callCtx, entityCatalogId, entityId));
+ grantRecords.addAll(
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, entityCatalogId, entityId));
} else {
- grantRecords = ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entityId);
+ grantRecords =
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, entityCatalogId, entityId);
}
// return the result
@@ -2202,7 +2105,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long entityId,
PolarisEntityType entityType) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
return ms.runInReadTransaction(
@@ -2223,7 +2126,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// load that entity
PolarisBaseEntity entity =
- ms.lookupEntityByName(callCtx, entityCatalogId, parentId, entityType.getCode(), entityName);
+ ms.lookupEntityByNameInCurrentTxn(
+ callCtx, entityCatalogId, parentId, entityType.getCode(), entityName);
// null if entity not found
if (entity == null) {
@@ -2235,11 +2139,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
if (entity.getType().isGrantee()) {
grantRecords =
new ArrayList<>(
- ms.loadAllGrantRecordsOnGrantee(callCtx, entityCatalogId, entity.getId()));
+ ms.loadAllGrantRecordsOnGranteeInCurrentTxn(
+ callCtx, entityCatalogId, entity.getId()));
grantRecords.addAll(
- ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entity.getId()));
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, entityCatalogId, entity.getId()));
} else {
- grantRecords = ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entity.getId());
+ grantRecords =
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, entityCatalogId, entity.getId());
}
// return the result
@@ -2255,7 +2161,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
@Nonnull PolarisEntityType entityType,
@Nonnull String entityName) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
ResolvedEntityResult result =
@@ -2283,7 +2189,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
this.createEntityIfNotExists(callCtx, ms, null, rootContainer);
if (backfillResult.isSuccess()) {
PolarisBaseEntity serviceAdminRole =
- ms.lookupEntityByName(
+ ms.lookupEntityByNameInCurrentTxn(
callCtx,
0L,
0L,
@@ -2323,7 +2229,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// load version information
PolarisChangeTrackingVersions entityVersions =
- ms.lookupEntityVersions(callCtx, List.of(new PolarisEntityId(entityCatalogId, entityId)))
+ ms.lookupEntityVersionsInCurrentTxn(
+ callCtx, List.of(new PolarisEntityId(entityCatalogId, entityId)))
.get(0);
// if null, the entity has been purged
@@ -2334,7 +2241,8 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
// load the entity if something changed
final PolarisBaseEntity entity;
if (entityVersion != entityVersions.getEntityVersion()) {
- entity = ms.lookupEntity(callCtx, entityCatalogId, entityId, entityType.getCode());
+ entity =
+ ms.lookupEntityInCurrentTxn(callCtx, entityCatalogId, entityId, entityType.getCode());
// if not found, return null
if (entity == null) {
@@ -2350,10 +2258,13 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
if (entityVersions.getGrantRecordsVersion() != entityGrantRecordsVersion) {
if (entityType.isGrantee()) {
grantRecords =
- new ArrayList<>(ms.loadAllGrantRecordsOnGrantee(callCtx, entityCatalogId, entityId));
- grantRecords.addAll(ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entityId));
+ new ArrayList<>(
+ ms.loadAllGrantRecordsOnGranteeInCurrentTxn(callCtx, entityCatalogId, entityId));
+ grantRecords.addAll(
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, entityCatalogId, entityId));
} else {
- grantRecords = ms.loadAllGrantRecordsOnSecurable(callCtx, entityCatalogId, entityId);
+ grantRecords =
+ ms.loadAllGrantRecordsOnSecurableInCurrentTxn(callCtx, entityCatalogId, entityId);
}
} else {
grantRecords = null;
@@ -2373,7 +2284,7 @@ public class PolarisMetaStoreManagerImpl extends BaseMetaStoreManager {
long entityCatalogId,
long entityId) {
// get metastore we should be using
- TransactionalPersistence ms = callCtx.getMetaStore();
+ TransactionalPersistence ms = ((TransactionalPersistence) callCtx.getMetaStore());
// need to run inside a read transaction
return ms.runInReadTransaction(
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapMetaStoreSessionImpl.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapMetaStoreSessionImpl.java
index 181aee2b0..521c5dfbb 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapMetaStoreSessionImpl.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapMetaStoreSessionImpl.java
@@ -99,13 +99,13 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
* @return new unique entity identifier
*/
@Override
- public long generateNewId(@Nonnull PolarisCallContext callCtx) {
+ public long generateNewIdInCurrentTxn(@Nonnull PolarisCallContext callCtx) {
return this.store.getNextSequence();
}
/** {@inheritDoc} */
@Override
- public void writeToEntities(
+ public void writeToEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
// write it
this.store.getSliceEntities().write(entity);
@@ -113,16 +113,17 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public <T extends PolarisStorageConfigurationInfo> void persistStorageIntegrationIfNeeded(
- @Nonnull PolarisCallContext callContext,
- @Nonnull PolarisBaseEntity entity,
- @Nullable PolarisStorageIntegration<T> storageIntegration) {
+ public <T extends PolarisStorageConfigurationInfo>
+ void persistStorageIntegrationIfNeededInCurrentTxn(
+ @Nonnull PolarisCallContext callContext,
+ @Nonnull PolarisBaseEntity entity,
+ @Nullable PolarisStorageIntegration<T> storageIntegration) {
// not implemented for in-memory store
}
/** {@inheritDoc} */
@Override
- public void writeToEntitiesActive(
+ public void writeToEntitiesActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
// write it
this.store.getSliceEntitiesActive().write(entity);
@@ -130,7 +131,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void writeToEntitiesChangeTracking(
+ public void writeToEntitiesChangeTrackingInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
// write it
this.store.getSliceEntitiesChangeTracking().write(entity);
@@ -138,7 +139,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void writeToGrantRecords(
+ public void writeToGrantRecordsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {
// write it
this.store.getSliceGrantRecords().write(grantRec);
@@ -147,7 +148,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void deleteFromEntities(
+ public void deleteFromEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
// delete it
@@ -156,7 +157,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void deleteFromEntitiesActive(
+ public void deleteFromEntitiesActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
// delete it
this.store.getSliceEntitiesActive().delete(this.store.buildEntitiesActiveKey(entity));
@@ -169,7 +170,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
* @param entity entity record to delete
*/
@Override
- public void deleteFromEntitiesChangeTracking(
+ public void deleteFromEntitiesChangeTrackingInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntityCore entity) {
// delete it
this.store.getSliceEntitiesChangeTracking().delete(this.store.buildEntitiesKey(entity));
@@ -177,7 +178,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void deleteFromGrantRecords(
+ public void deleteFromGrantRecordsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec) {
// delete it
@@ -187,7 +188,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void deleteAllEntityGrantRecords(
+ public void deleteAllEntityGrantRecordsInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
@Nonnull PolarisEntityCore entity,
@Nonnull List<PolarisGrantRecord> grantsOnGrantee,
@@ -206,21 +207,21 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void deleteAll(@Nonnull PolarisCallContext callCtx) {
+ public void deleteAllInCurrentTxn(@Nonnull PolarisCallContext callCtx) {
// clear all slices
this.store.deleteAll();
}
/** {@inheritDoc} */
@Override
- public @Nullable PolarisBaseEntity lookupEntity(
+ public @Nullable PolarisBaseEntity lookupEntityInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long catalogId, long entityId, int typeCode) {
return this.store.getSliceEntities().read(this.store.buildKeyComposite(catalogId, entityId));
}
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisBaseEntity> lookupEntities(
+ public @Nonnull List<PolarisBaseEntity> lookupEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds) {
// allocate return list
return entityIds.stream()
@@ -234,7 +235,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisChangeTrackingVersions> lookupEntityVersions(
+ public @Nonnull List<PolarisChangeTrackingVersions> lookupEntityVersionsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds) {
// allocate return list
return entityIds.stream()
@@ -255,7 +256,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
@Nullable
- public EntityNameLookupRecord lookupEntityActive(
+ public EntityNameLookupRecord lookupEntityActiveInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisEntitiesActiveKey entityActiveKey) {
// lookup the active entity slice
PolarisBaseEntity entity =
@@ -283,34 +284,35 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
@Nonnull
- public List<EntityNameLookupRecord> lookupEntityActiveBatch(
+ public List<EntityNameLookupRecord> lookupEntityActiveBatchInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
@Nonnull List<PolarisEntitiesActiveKey> entityActiveKeys) {
// now build a list to quickly verify that nothing has changed
return entityActiveKeys.stream()
- .map(entityActiveKey -> this.lookupEntityActive(callCtx, entityActiveKey))
+ .map(entityActiveKey -> this.lookupEntityActiveInCurrentTxn(callCtx, entityActiveKey))
.collect(Collectors.toList());
}
/** {@inheritDoc} */
@Override
- public @Nonnull List<EntityNameLookupRecord> listEntities(
+ public @Nonnull List<EntityNameLookupRecord> listEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long parentId,
@Nonnull PolarisEntityType entityType) {
- return listEntities(callCtx, catalogId, parentId, entityType, Predicates.alwaysTrue());
+ return this.listEntitiesInCurrentTxn(
+ callCtx, catalogId, parentId, entityType, Predicates.alwaysTrue());
}
@Override
- public @Nonnull List<EntityNameLookupRecord> listEntities(
+ public @Nonnull List<EntityNameLookupRecord> listEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long parentId,
@Nonnull PolarisEntityType entityType,
@Nonnull Predicate<PolarisBaseEntity> entityFilter) {
// full range scan under the parent for that type
- return listEntities(
+ return this.listEntitiesInCurrentTxn(
callCtx,
catalogId,
parentId,
@@ -328,7 +330,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
}
@Override
- public @Nonnull <T> List<T> listEntities(
+ public @Nonnull <T> List<T> listEntitiesInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long parentId,
@@ -343,7 +345,8 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
.stream()
.map(
nameRecord ->
- this.lookupEntity(callCtx, catalogId, nameRecord.getId(), entityType.getCode()))
+ this.lookupEntityInCurrentTxn(
+ callCtx, catalogId, nameRecord.getId(), entityType.getCode()))
.filter(entityFilter)
.limit(limit)
.map(transformer)
@@ -352,7 +355,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public boolean hasChildren(
+ public boolean hasChildrenInCurrentTxn(
@Nonnull PolarisCallContext callContext,
@Nullable PolarisEntityType entityType,
long catalogId,
@@ -368,7 +371,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public int lookupEntityGrantRecordsVersion(
+ public int lookupEntityGrantRecordsVersionInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long catalogId, long entityId) {
PolarisBaseEntity entity =
this.store
@@ -381,7 +384,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public @Nullable PolarisGrantRecord lookupGrantRecord(
+ public @Nullable PolarisGrantRecord lookupGrantRecordInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long securableCatalogId,
long securableId,
@@ -398,7 +401,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnSecurable(
+ public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnSecurableInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId) {
// now fetch all grants for this securable
return this.store
@@ -408,7 +411,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnGrantee(
+ public @Nonnull List<PolarisGrantRecord> loadAllGrantRecordsOnGranteeInCurrentTxn(
@Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId) {
// now fetch all grants assigned to this grantee
return this.store
@@ -418,14 +421,14 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public @Nullable PolarisPrincipalSecrets loadPrincipalSecrets(
+ public @Nullable PolarisPrincipalSecrets loadPrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull String clientId) {
return this.store.getSlicePrincipalSecrets().read(clientId);
}
/** {@inheritDoc} */
@Override
- public @Nonnull PolarisPrincipalSecrets generateNewPrincipalSecrets(
+ public @Nonnull PolarisPrincipalSecrets generateNewPrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull String principalName, long principalId) {
// ensure principal client id is unique
PolarisPrincipalSecrets principalSecrets;
@@ -448,7 +451,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public @Nonnull PolarisPrincipalSecrets rotatePrincipalSecrets(
+ public @Nonnull PolarisPrincipalSecrets rotatePrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
@Nonnull String clientId,
long principalId,
@@ -493,7 +496,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
- public void deletePrincipalSecrets(
+ public void deletePrincipalSecretsInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull String clientId, long principalId) {
// load the existing secrets
PolarisPrincipalSecrets principalSecrets = this.store.getSlicePrincipalSecrets().read(clientId);
@@ -525,7 +528,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
public @Nullable <T extends PolarisStorageConfigurationInfo>
- PolarisStorageIntegration<T> createStorageIntegration(
+ PolarisStorageIntegration<T> createStorageIntegrationInCurrentTxn(
@Nonnull PolarisCallContext callCtx,
long catalogId,
long entityId,
@@ -537,7 +540,7 @@ public class PolarisTreeMapMetaStoreSessionImpl extends AbstractTransactionalPer
/** {@inheritDoc} */
@Override
public @Nullable <T extends PolarisStorageConfigurationInfo>
- PolarisStorageIntegration<T> loadPolarisStorageIntegration(
+ PolarisStorageIntegration<T> loadPolarisStorageIntegrationInCurrentTxn(
@Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity) {
PolarisStorageConfigurationInfo storageConfig =
BaseMetaStoreManager.extractStorageConfiguration(callCtx, entity);
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapStore.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapStore.java
index fe80bdf23..89784fd88 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapStore.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/PolarisTreeMapStore.java
@@ -386,7 +386,9 @@ public class PolarisTreeMapStore {
this.startWriteTransaction();
return transactionCode.get();
} catch (Throwable e) {
- this.rollback();
+ if (this.tr != null) {
+ this.rollback();
+ }
throw e;
} finally {
this.tr = null;
@@ -413,7 +415,9 @@ public class PolarisTreeMapStore {
this.startWriteTransaction();
transactionCode.run();
} catch (Throwable e) {
- this.rollback();
+ if (this.tr != null) {
+ this.rollback();
+ }
throw e;
} finally {
this.tr = null;
@@ -499,6 +503,7 @@ public class PolarisTreeMapStore {
* @return next id, must be in a read/write transaction
*/
public long getNextSequence() {
+ this.ensureReadWriteTr();
return this.nextId.incrementAndGet();
}
diff --git a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalPersistence.java b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalPersistence.java
index b594a7f15..f3d3842bb 100644
--- a/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalPersistence.java
+++ b/polaris-core/src/main/java/org/apache/polaris/core/persistence/transactional/TransactionalPersistence.java
@@ -19,13 +19,25 @@
package org.apache.polaris.core.persistence.transactional;
import jakarta.annotation.Nonnull;
+import jakarta.annotation.Nullable;
import java.util.List;
+import java.util.function.Function;
+import java.util.function.Predicate;
import java.util.function.Supplier;
import org.apache.polaris.core.PolarisCallContext;
import org.apache.polaris.core.entity.EntityNameLookupRecord;
+import org.apache.polaris.core.entity.PolarisBaseEntity;
+import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
import org.apache.polaris.core.entity.PolarisEntitiesActiveKey;
+import org.apache.polaris.core.entity.PolarisEntityCore;
+import org.apache.polaris.core.entity.PolarisEntityId;
+import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.entity.PolarisGrantRecord;
+import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
import org.apache.polaris.core.persistence.BasePersistence;
import org.apache.polaris.core.persistence.IntegrationPersistence;
+import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo;
+import org.apache.polaris.core.storage.PolarisStorageIntegration;
/**
* Extends BasePersistence to express a more "transaction-oriented" control flow for backing stores
@@ -88,9 +100,224 @@ public interface TransactionalPersistence extends BasePersistence, IntegrationPe
* @return the list of entityActiveKeys for the specified lookup operation
*/
@Nonnull
- List<EntityNameLookupRecord> lookupEntityActiveBatch(
+ List<EntityNameLookupRecord> lookupEntityActiveBatchInCurrentTxn(
@Nonnull PolarisCallContext callCtx, List<PolarisEntitiesActiveKey> entityActiveKeys);
/** Rollback the current transaction */
void rollback();
+
+ //
+ // Every method of BasePersistence will have a related * method here; the semantics
+ // being that transactional implementations of a PolarisMetaStoreManager may choose to
+ // self-manage outer transactions to perform all the persistence calls within that provided
+ // transaction, while the basic implementation will only use the "durable in a single-shot"
+ // methods from BasePersistence. Condition-checks for atomic compare-and-swap behaviors are *not*
+ // expected to occur within these *InCurrentTxn methods.
+ //
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#generateNewId} */
+ long generateNewIdInCurrentTxn(@Nonnull PolarisCallContext callCtx);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.BasePersistence#writeEntity}
+ *
+ * <p>NOTE: By virtue of the way callers of these *InCurrentTxn methods organize entity-state
+ * checks interspersed between different persistence actions, the basic compare-and-swap
+ * conditions are *not* expected to be enforced within these methods, in contrast to the analogous
+ * methods in BasePersistence. For example, BasePersistence::writeEntity is expected to use the
+ * entityVersion of originalEntity as part of an atomic conditional check before writing the new
+ * entity, but TransactionalPersistence::writeEntityInCurrentTxn is *not* expected to do the same.
+ */
+ void writeEntityInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisBaseEntity entity,
+ boolean nameOrParentChanged,
+ @Nullable PolarisBaseEntity originalEntity);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#writeEntities} */
+ void writeEntitiesInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull List<PolarisBaseEntity> entities,
+ @Nullable List<PolarisBaseEntity> originalEntities);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#writeToGrantRecords} */
+ void writeToGrantRecordsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#deleteEntity} */
+ void deleteEntityInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisBaseEntity entity);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#deleteFromGrantRecords} */
+ void deleteFromGrantRecordsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull PolarisGrantRecord grantRec);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#deleteAllEntityGrantRecords} */
+ void deleteAllEntityGrantRecordsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull PolarisEntityCore entity,
+ @Nonnull List<PolarisGrantRecord> grantsOnGrantee,
+ @Nonnull List<PolarisGrantRecord> grantsOnSecurable);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#deleteAll} */
+ void deleteAllInCurrentTxn(@Nonnull PolarisCallContext callCtx);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#lookupEntity} */
+ @Nullable
+ PolarisBaseEntity lookupEntityInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, long catalogId, long entityId, int typeCode);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#lookupEntityByName} */
+ @Nullable
+ PolarisBaseEntity lookupEntityByNameInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ int typeCode,
+ @Nonnull String name);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.BasePersistence#lookupEntityIdAndSubTypeByName}
+ */
+ @Nullable
+ EntityNameLookupRecord lookupEntityIdAndSubTypeByNameInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ int typeCode,
+ @Nonnull String name);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#lookupEntities} */
+ @Nonnull
+ List<PolarisBaseEntity> lookupEntitiesInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#lookupEntityVersions} */
+ @Nonnull
+ List<PolarisChangeTrackingVersions> lookupEntityVersionsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, List<PolarisEntityId> entityIds);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#listEntities} */
+ @Nonnull
+ List<EntityNameLookupRecord> listEntitiesInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ @Nonnull PolarisEntityType entityType);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#listEntities} */
+ @Nonnull
+ List<EntityNameLookupRecord> listEntitiesInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ @Nonnull PolarisEntityType entityType,
+ @Nonnull Predicate<PolarisBaseEntity> entityFilter);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#listEntities} */
+ @Nonnull
+ <T> List<T> listEntitiesInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long parentId,
+ @Nonnull PolarisEntityType entityType,
+ int limit,
+ @Nonnull Predicate<PolarisBaseEntity> entityFilter,
+ @Nonnull Function<PolarisBaseEntity, T> transformer);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.BasePersistence#lookupEntityGrantRecordsVersion}
+ */
+ int lookupEntityGrantRecordsVersionInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, long catalogId, long entityId);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#lookupGrantRecord} */
+ @Nullable
+ PolarisGrantRecord lookupGrantRecordInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long securableCatalogId,
+ long securableId,
+ long granteeCatalogId,
+ long granteeId,
+ int privilegeCode);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.BasePersistence#loadAllGrantRecordsOnSecurable}
+ */
+ @Nonnull
+ List<PolarisGrantRecord> loadAllGrantRecordsOnSecurableInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, long securableCatalogId, long securableId);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.BasePersistence#loadAllGrantRecordsOnGrantee}
+ */
+ @Nonnull
+ List<PolarisGrantRecord> loadAllGrantRecordsOnGranteeInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, long granteeCatalogId, long granteeId);
+
+ /** See {@link org.apache.polaris.core.persistence.BasePersistence#hasChildren} */
+ boolean hasChildrenInCurrentTxn(
+ @Nonnull PolarisCallContext callContext,
+ @Nullable PolarisEntityType optionalEntityType,
+ long catalogId,
+ long parentId);
+
+ /** See {@link org.apache.polaris.core.persistence.IntegrationPersistence#loadPrincipalSecrets} */
+ @Nullable
+ PolarisPrincipalSecrets loadPrincipalSecretsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull String clientId);
+
+ /**
+ * See {@link
+ * org.apache.polaris.core.persistence.IntegrationPersistence#generateNewPrincipalSecrets}
+ */
+ @Nonnull
+ PolarisPrincipalSecrets generateNewPrincipalSecretsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull String principalName, long principalId);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.IntegrationPersistence#rotatePrincipalSecrets}
+ */
+ @Nullable
+ PolarisPrincipalSecrets rotatePrincipalSecretsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ @Nonnull String clientId,
+ long principalId,
+ boolean reset,
+ @Nonnull String oldSecretHash);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.IntegrationPersistence#deletePrincipalSecrets}
+ */
+ void deletePrincipalSecretsInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx, @Nonnull String clientId, long principalId);
+
+ /**
+ * See {@link org.apache.polaris.core.persistence.IntegrationPersistence#createStorageIntegration}
+ */
+ @Nullable
+ <T extends PolarisStorageConfigurationInfo>
+ PolarisStorageIntegration<T> createStorageIntegrationInCurrentTxn(
+ @Nonnull PolarisCallContext callCtx,
+ long catalogId,
+ long entityId,
+ PolarisStorageConfigurationInfo polarisStorageConfigurationInfo);
+
+ /**
+ * See {@link
+ * org.apache.polaris.core.persistence.IntegrationPersistence#persistStorageIntegrationIfNeeded}
+ */
+ <T extends PolarisStorageConfigurationInfo> void persistStorageIntegrationIfNeededInCurrentTxn(
+ @Nonnull PolarisCallContext callContext,
+ @Nonnull PolarisBaseEntity entity,
+ @Nullable PolarisStorageIntegration<T> storageIntegration);
+
+ /**
+ * See {@link
+ * org.apache.polaris.core.persistence.IntegrationPersistence#loadPolarisStorageIntegration}
+ */
+ @Nullable
+ <T extends PolarisStorageConfigurationInfo>
+ PolarisStorageIntegration<T> loadPolarisStorageIntegrationInCurrentTxn(
+ @Nonnull PolarisCallContext callContext, @Nonnull PolarisBaseEntity entity);
}
diff --git a/polaris-core/src/test/java/org/apache/polaris/core/persistence/PolarisTreeMapAtomicOperationMetaStoreManagerTest.java b/polaris-core/src/test/java/org/apache/polaris/core/persistence/PolarisTreeMapAtomicOperationMetaStoreManagerTest.java
new file mode 100644
index 000000000..42e61f7d4
--- /dev/null
+++ b/polaris-core/src/test/java/org/apache/polaris/core/persistence/PolarisTreeMapAtomicOperationMetaStoreManagerTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.core.persistence;
+
+import static org.apache.polaris.core.persistence.PrincipalSecretsGenerator.RANDOM_SECRETS;
+
+import java.time.ZoneId;
+import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.PolarisConfigurationStore;
+import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
+import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.persistence.transactional.PolarisTreeMapMetaStoreSessionImpl;
+import org.apache.polaris.core.persistence.transactional.PolarisTreeMapStore;
+import org.mockito.Mockito;
+
+public class PolarisTreeMapAtomicOperationMetaStoreManagerTest
+ extends BasePolarisMetaStoreManagerTest {
+ @Override
+ public PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
+ PolarisDiagnostics diagServices = new PolarisDefaultDiagServiceImpl();
+ PolarisTreeMapStore store = new PolarisTreeMapStore(diagServices);
+ PolarisCallContext callCtx =
+ new PolarisCallContext(
+ new PolarisTreeMapMetaStoreSessionImpl(store, Mockito.mock(), RANDOM_SECRETS),
+ diagServices,
+ new PolarisConfigurationStore() {},
+ timeSource.withZone(ZoneId.systemDefault()));
+
+ return new PolarisTestMetaStoreManager(new AtomicOperationMetaStoreManager(), callCtx);
+ }
+}
diff --git a/quarkus/defaults/src/main/resources/application-it.properties b/quarkus/defaults/src/main/resources/application-it.properties
index 5f46d203f..5e110071d 100644
--- a/quarkus/defaults/src/main/resources/application-it.properties
+++ b/quarkus/defaults/src/main/resources/application-it.properties
@@ -25,6 +25,8 @@ quarkus.http.port=0
quarkus.management.port=0
+# polaris.persistence.type=eclipse-link
+# polaris.persistence.type=in-memory-atomic
polaris.persistence.type=in-memory
polaris.features.defaults."ALLOW_EXTERNAL_CATALOG_CREDENTIAL_VENDING"=false
diff --git a/quarkus/defaults/src/main/resources/application.properties b/quarkus/defaults/src/main/resources/application.properties
index a523b15f9..76ab4e82d 100644
--- a/quarkus/defaults/src/main/resources/application.properties
+++ b/quarkus/defaults/src/main/resources/application.properties
@@ -95,6 +95,8 @@ polaris.features.defaults."SUPPORTED_CATALOG_STORAGE_TYPES"=["S3","GCS","AZURE",
# polaris.features.realm-overrides."my-realm"."INITIALIZE_DEFAULT_CATALOG_FILEIO_FOR_TEST"=true
# polaris.features.realm-overrides."my-realm"."SKIP_CREDENTIAL_SUBSCOPING_INDIRECTION"=true
+# polaris.persistence.type=eclipse-link
+# polaris.persistence.type=in-memory-atomic
polaris.persistence.type=in-memory
polaris.file-io.type=default
diff --git a/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/BasePolarisCatalogTest.java b/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/BasePolarisCatalogTest.java
index 7f0b73e9c..e51ded52f 100644
--- a/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/BasePolarisCatalogTest.java
+++ b/quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/BasePolarisCatalogTest.java
@@ -344,7 +344,7 @@ public abstract class BasePolarisCatalogTest extends CatalogTests<BasePolarisCat
@Override
public Supplier<TransactionalPersistence> getOrCreateSessionSupplier(
RealmContext realmContext) {
- return () -> polarisContext.getMetaStore();
+ return () -> ((TransactionalPersistence) polarisContext.getMetaStore());
}
@Override
diff --git a/service/common/src/main/java/org/apache/polaris/service/persistence/InMemoryAtomicOperationMetaStoreManagerFactory.java b/service/common/src/main/java/org/apache/polaris/service/persistence/InMemoryAtomicOperationMetaStoreManagerFactory.java
new file mode 100644
index 000000000..3f0b1a355
--- /dev/null
+++ b/service/common/src/main/java/org/apache/polaris/service/persistence/InMemoryAtomicOperationMetaStoreManagerFactory.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.persistence;
+
+import io.smallrye.common.annotation.Identifier;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.persistence.AtomicOperationMetaStoreManager;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
+import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider;
+
+/**
+ * Uses a PolarisTreeMapStore for the underlying persistence layer but uses it to initialize an
+ * AtomicOperationMetaStoreManager instead of the transactional version.
+ */
+@ApplicationScoped
+@Identifier("in-memory-atomic")
+public class InMemoryAtomicOperationMetaStoreManagerFactory
+ extends InMemoryPolarisMetaStoreManagerFactory {
+
+ public InMemoryAtomicOperationMetaStoreManagerFactory() {
+ super(null, null);
+ }
+
+ @Inject
+ public InMemoryAtomicOperationMetaStoreManagerFactory(
+ PolarisStorageIntegrationProvider storageIntegration, PolarisDiagnostics diagnostics) {
+ super(storageIntegration, diagnostics);
+ }
+
+ @Override
+ protected PolarisMetaStoreManager createNewMetaStoreManager() {
+ return new AtomicOperationMetaStoreManager();
+ }
+}