snazy commented on code in PR #3237: URL: https://github.com/apache/polaris/pull/3237#discussion_r2610043793
########## persistence/nosql/persistence/metastore/src/main/java/org/apache/polaris/persistence/nosql/metastore/NoSqlMetaStore.java: ########## @@ -0,0 +1,1347 @@ +/* + * 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.persistence.nosql.metastore; + +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkState; +import static java.lang.String.format; +import static java.util.Objects.requireNonNull; +import static org.apache.polaris.core.entity.PolarisEntityConstants.ENTITY_BASE_LOCATION; +import static org.apache.polaris.core.persistence.dao.entity.BaseResult.ReturnStatus.ENTITY_ALREADY_EXISTS; +import static org.apache.polaris.core.persistence.dao.entity.BaseResult.ReturnStatus.ENTITY_NOT_FOUND; +import static org.apache.polaris.persistence.nosql.api.index.IndexContainer.newUpdatableIndex; +import static org.apache.polaris.persistence.nosql.api.index.IndexKey.INDEX_KEY_SERIALIZER; +import static org.apache.polaris.persistence.nosql.api.obj.ObjRef.OBJ_REF_SERIALIZER; +import static org.apache.polaris.persistence.nosql.api.obj.ObjRef.objRef; +import static org.apache.polaris.persistence.nosql.coretypes.catalog.CatalogStateObj.CATALOG_STATE_REF_NAME_PATTERN; +import static org.apache.polaris.persistence.nosql.coretypes.catalog.EntityIdSet.ENTITY_ID_SET_SERIALIZER; +import static org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings.containerTypeForEntityType; +import static org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings.filterIsEntityType; +import static org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings.isCatalogContent; +import static org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings.mapToEntity; +import static org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings.objTypeForPolarisTypeForFiltering; +import static org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings.referenceName; +import static org.apache.polaris.persistence.nosql.coretypes.realm.PolicyMapping.POLICY_MAPPING_SERIALIZER; +import static org.apache.polaris.persistence.nosql.coretypes.realm.PolicyMappingsObj.POLICY_MAPPINGS_REF_NAME; +import static org.apache.polaris.persistence.nosql.coretypes.realm.PolicyMappingsObj.PolicyMappingKey.fromIndexKey; +import static org.apache.polaris.persistence.nosql.coretypes.realm.RootObj.ROOT_REF_NAME; +import static org.apache.polaris.persistence.nosql.coretypes.refs.References.catalogReferenceNames; +import static org.apache.polaris.persistence.nosql.metastore.ContentIdentifier.identifierFromLocationString; +import static org.apache.polaris.persistence.nosql.metastore.ContentIdentifier.indexKeyToIdentifier; +import static org.apache.polaris.persistence.nosql.metastore.indexaccess.MemoizedIndexedAccess.newMemoizedIndexedAccess; +import static org.apache.polaris.persistence.nosql.metastore.mutation.EntityUpdate.Operation.CREATE; +import static org.apache.polaris.persistence.nosql.metastore.mutation.EntityUpdate.Operation.UPDATE; +import static org.apache.polaris.persistence.nosql.metastore.mutation.UpdateKeyForCatalogAndEntityType.updateKeyForCatalogAndEntityType; + +import com.google.common.collect.Streams; +import jakarta.annotation.Nonnull; +import jakarta.annotation.Nullable; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.function.Function; +import java.util.function.Predicate; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.polaris.core.PolarisDiagnostics; +import org.apache.polaris.core.entity.AsyncTaskType; +import org.apache.polaris.core.entity.LocationBasedEntity; +import org.apache.polaris.core.entity.PolarisBaseEntity; +import org.apache.polaris.core.entity.PolarisEntity; +import org.apache.polaris.core.entity.PolarisEntityConstants; +import org.apache.polaris.core.entity.PolarisEntityCore; +import org.apache.polaris.core.entity.PolarisEntitySubType; +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.entity.PolarisPrivilege; +import org.apache.polaris.core.entity.PolarisTaskConstants; +import org.apache.polaris.core.persistence.BaseMetaStoreManager; +import org.apache.polaris.core.persistence.PolarisObjectMapperUtil; +import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet; +import org.apache.polaris.core.persistence.dao.entity.CreateCatalogResult; +import org.apache.polaris.core.persistence.dao.entity.CreatePrincipalResult; +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.LoadGrantsResult; +import org.apache.polaris.core.persistence.dao.entity.LoadPolicyMappingsResult; +import org.apache.polaris.core.persistence.dao.entity.PolicyAttachmentResult; +import org.apache.polaris.core.persistence.pagination.Page; +import org.apache.polaris.core.persistence.pagination.PageToken; +import org.apache.polaris.core.policy.PolarisPolicyMappingRecord; +import org.apache.polaris.core.policy.PolicyMappingUtil; +import org.apache.polaris.core.policy.PolicyType; +import org.apache.polaris.core.storage.PolarisStorageConfigurationInfo; +import org.apache.polaris.core.storage.PolarisStorageIntegration; +import org.apache.polaris.core.storage.PolarisStorageIntegrationProvider; +import org.apache.polaris.core.storage.StorageLocation; +import org.apache.polaris.persistence.nosql.api.Persistence; +import org.apache.polaris.persistence.nosql.api.index.Index; +import org.apache.polaris.persistence.nosql.api.index.IndexKey; +import org.apache.polaris.persistence.nosql.api.obj.ObjRef; +import org.apache.polaris.persistence.nosql.api.obj.ObjTypes; +import org.apache.polaris.persistence.nosql.authz.api.Privilege; +import org.apache.polaris.persistence.nosql.authz.api.PrivilegeSet; +import org.apache.polaris.persistence.nosql.authz.api.Privileges; +import org.apache.polaris.persistence.nosql.coretypes.ContainerObj; +import org.apache.polaris.persistence.nosql.coretypes.ObjBase; +import org.apache.polaris.persistence.nosql.coretypes.acl.AclObj; +import org.apache.polaris.persistence.nosql.coretypes.catalog.CatalogObj; +import org.apache.polaris.persistence.nosql.coretypes.catalog.CatalogRoleObj; +import org.apache.polaris.persistence.nosql.coretypes.catalog.CatalogRolesObj; +import org.apache.polaris.persistence.nosql.coretypes.catalog.CatalogStateObj; +import org.apache.polaris.persistence.nosql.coretypes.catalog.CatalogsObj; +import org.apache.polaris.persistence.nosql.coretypes.content.ContentObj; +import org.apache.polaris.persistence.nosql.coretypes.mapping.EntityObjMappings; +import org.apache.polaris.persistence.nosql.coretypes.principals.PrincipalObj; +import org.apache.polaris.persistence.nosql.coretypes.principals.PrincipalsObj; +import org.apache.polaris.persistence.nosql.coretypes.realm.PolicyMapping; +import org.apache.polaris.persistence.nosql.coretypes.realm.PolicyMappingsObj; +import org.apache.polaris.persistence.nosql.coretypes.realm.RootObj; +import org.apache.polaris.persistence.nosql.metastore.committers.CatalogChangeCommitterWrapper; +import org.apache.polaris.persistence.nosql.metastore.committers.ChangeCommitter; +import org.apache.polaris.persistence.nosql.metastore.committers.ChangeCommitterWrapper; +import org.apache.polaris.persistence.nosql.metastore.committers.ChangeResult; +import org.apache.polaris.persistence.nosql.metastore.indexaccess.IndexedContainerAccess; +import org.apache.polaris.persistence.nosql.metastore.indexaccess.MemoizedIndexedAccess; +import org.apache.polaris.persistence.nosql.metastore.mutation.EntityUpdate; +import org.apache.polaris.persistence.nosql.metastore.mutation.GrantsMutation; +import org.apache.polaris.persistence.nosql.metastore.mutation.MutationAttempt; +import org.apache.polaris.persistence.nosql.metastore.mutation.MutationAttemptRoot; +import org.apache.polaris.persistence.nosql.metastore.mutation.MutationResults; +import org.apache.polaris.persistence.nosql.metastore.mutation.PolicyMutation; +import org.apache.polaris.persistence.nosql.metastore.mutation.PrincipalMutations; +import org.apache.polaris.persistence.nosql.metastore.mutation.PrincipalMutations.UpdateSecrets.SecretsUpdater; +import org.apache.polaris.persistence.nosql.metastore.mutation.UpdateKeyForCatalogAndEntityType; +import org.apache.polaris.persistence.nosql.metastore.privs.GrantTriplet; +import org.apache.polaris.persistence.nosql.metastore.privs.SecurableAndGrantee; +import org.apache.polaris.persistence.nosql.metastore.privs.SecurableGranteePrivilegeTuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +class NoSqlMetaStore extends NonFunctionalBasePersistence { + private static final Logger LOGGER = LoggerFactory.getLogger(NoSqlMetaStore.class); + + private final Persistence persistence; + private final Privileges privileges; + private final PolarisStorageIntegrationProvider storageIntegrationProvider; + private final MemoizedIndexedAccess memoizedIndexedAccess; + private final PolarisDiagnostics diagnostics; + + NoSqlMetaStore( + Persistence persistence, + Privileges privileges, + PolarisStorageIntegrationProvider storageIntegrationProvider, + PolarisDiagnostics diagnostics) { + this.persistence = persistence; + this.privileges = privileges; + this.storageIntegrationProvider = storageIntegrationProvider; + this.memoizedIndexedAccess = newMemoizedIndexedAccess(persistence); + this.diagnostics = diagnostics; + } + + <REF_OBJ extends ContainerObj, RESULT> RESULT performChange( + @Nonnull PolarisEntityType entityType, + @Nonnull Class<REF_OBJ> referencedObjType, + @Nonnull Class<RESULT> resultType, + long catalogStableId, + @Nonnull ChangeCommitter<REF_OBJ, RESULT> changeCommitter) { + try { + var committer = + persistence + .createCommitter( + referenceName(entityType, catalogStableId), referencedObjType, resultType) + .synchronizingLocally(); + var commitRetryable = new ChangeCommitterWrapper<>(changeCommitter, entityType); + return committer.commitRuntimeException(commitRetryable).orElseThrow(); + } finally { + memoizedIndexedAccess.invalidateIndexedAccess(catalogStableId, entityType.getCode()); + } + } + + long generateNewId() { + return persistence.generateId(); + } + + void initializeCatalogsIfNecessary() { + memoizedIndexedAccess + .indexedAccess(0, PolarisEntityType.CATALOG.getCode()) + .nameIndex() + .ifPresent( + names -> + persistence + .bucketizedBulkFetches( + Streams.stream(names).filter(Objects::nonNull).map(Map.Entry::getValue), + CatalogObj.class) + .filter(Objects::nonNull) + .forEach( + catalogObj -> { + LOGGER.debug("Initializing catalog {} if necessary", catalogObj.name()); + initializeCatalogIfNecessary(persistence, catalogObj); + })); + } + + CreateCatalogResult createCatalog( + PolarisBaseEntity catalog, List<PolarisBaseEntity> principalRoles) { + checkArgument(catalog != null && catalog.getType() == PolarisEntityType.CATALOG); + + LOGGER.debug("create catalog #{} '{}'", catalog.getId(), catalog.getName()); + + return performChange( + PolarisEntityType.CATALOG, + CatalogsObj.class, + CreateCatalogResult.class, + 0L, + ((state, ref, byName, byId) -> { + var nameKey = IndexKey.key(catalog.getName()); + var idKey = IndexKey.key(catalog.getId()); + + // check if that catalog has already been created + var existing = byName.get(nameKey); + var persistence = state.persistence(); + var catalogObj = existing != null ? persistence.fetch(existing, CatalogObj.class) : null; + + // if found, probably a retry, simply return the previously created catalog + // TODO not sure how a "retry" could happen with the same ID though (see + // PolarisMetaStoreManagerImpl.createCatalog())... + if (catalogObj != null && catalogObj.stableId() != catalog.getId()) { + // A catalog with the same name already exists (different ID) + return new ChangeResult.NoChange<>( + new CreateCatalogResult(ENTITY_ALREADY_EXISTS, null)); + } + if (catalogObj == null) { + catalogObj = + EntityObjMappings.<CatalogObj, CatalogObj.Builder>mapToObj( + catalog, Optional.empty()) + .id(persistence.generateId()) + .build(); + state.writeOrReplace("catalog", catalogObj); + } + + initializeCatalogIfNecessary(persistence, catalogObj); + + checkState(!byId.contains(idKey), "Catalog ID %s already used", catalog.getId()); + + // 'persistStorageIntegrationIfNeeded' is a no-op in all implementations ?!?!? + // persistStorageIntegrationIfNeeded(callCtx, catalog, integration); + + var catalogAdminRoleObj = + createCatalogRoleIdempotent( + catalogObj, + persistence.generateId(), + PolarisEntityConstants.getNameOfCatalogAdminRole()); + + var catalogAdminRole = mapToEntity(catalogAdminRoleObj, catalogObj.stableId()); + + var grants = new ArrayList<SecurableGranteePrivilegeTuple>(); + + // grant the catalog admin role access-management on the catalog + grants.add( + new SecurableGranteePrivilegeTuple( + catalog, catalogAdminRole, PolarisPrivilege.CATALOG_MANAGE_ACCESS)); + // grant the catalog admin role metadata-management on the catalog; this one is revocable + grants.add( + new SecurableGranteePrivilegeTuple( + catalog, catalogAdminRole, PolarisPrivilege.CATALOG_MANAGE_METADATA)); + + var effRoles = + principalRoles.isEmpty() + ? List.of( + requireNonNull( + lookupEntityByName( + 0L, + 0L, + PolarisEntityType.PRINCIPAL_ROLE.getCode(), + PolarisEntityConstants.getNameOfPrincipalServiceAdminRole()))) + : principalRoles; + + for (PolarisBaseEntity effRole : effRoles) { + grants.add( + new SecurableGranteePrivilegeTuple( + catalogAdminRole, effRole, PolarisPrivilege.CATALOG_ROLE_USAGE)); + } + + persistGrantsOrRevokes(true, grants.toArray(SecurableGranteePrivilegeTuple[]::new)); + + byName.put(nameKey, objRef(catalogObj)); + byId.put(idKey, nameKey); + + if (existing == null) { + // created + return new ChangeResult.CommitChange<>( + new CreateCatalogResult(catalog, catalogAdminRole)); + } + // retry + return new ChangeResult.NoChange<>(new CreateCatalogResult(ENTITY_ALREADY_EXISTS, null)); Review Comment: Both (index and persistence state) are always consistent. This particular one is literally an impossible case, so I removed it. This (and other) functions have to expect cases when call sites (management API handlers) pass already existing names and/or already existing entity IDs. For example, a call site can pass a non-existing catalog name with an already used entity ID. Would be easier if the IDs were generated exclusively by the implementations, not the call sites. > I think the checkState(!byId.contains(idKey on line 246 means that this line wouldn't actually represent stableId == id though, in which case reaching this line should maybe be an IllegalStateException indicating some kind of corrupted state? Actually not a corrupted state, but the API call sites passing an already used catalog ID into the "createCatalog" call. When I wrote this function, I thought about how a retry could actually happen, given that the existing implementations assume entity-ID-equality (the reason for the TODO at the beginning of the function). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
