jerryshao commented on code in PR #14281: URL: https://github.com/apache/gravitino/pull/14281#discussion_r4412705058
########## core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.gravitino.storage.relational.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.MetadataObject; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.meta.DomainEntity; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper; +import org.apache.gravitino.storage.relational.po.DomainPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.utils.ExceptionUtils; +import org.apache.gravitino.storage.relational.utils.SessionUtils; + +/** The service that stores domains in the {@code domain_meta} table. */ +public class DomainMetaService { + + private static final DomainMetaService INSTANCE = new DomainMetaService(); + + private DomainMetaService() {} + + /** + * Returns the singleton service instance. + * + * @return The domain meta service. + */ + public static DomainMetaService getInstance() { + return INSTANCE; + } + + /** + * Lists the domains of a metalake. + * + * @param ns The domain namespace of the metalake. + * @return The domains of the metalake. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "listDomainsByNamespace") + public List<DomainEntity> listDomainsByNamespace(Namespace ns) { + String metalakeName = ns.level(0); + List<DomainPO> domainPOs = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, mapper -> mapper.listDomainPOsByMetalake(metalakeName)); + return DomainPO.fromDomainPOs(domainPOs, ns); + } + + /** + * Gets a domain by its identifier. + * + * @param ident The domain identifier. + * @return The domain. + * @throws NoSuchEntityException If the domain does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getDomainByIdentifier") + public DomainEntity getDomainByIdentifier(NameIdentifier ident) { + String metalakeName = ident.namespace().level(0); + DomainPO domainPO = getDomainPOByMetalakeAndName(metalakeName, ident.name()); + return DomainPO.fromDomainPO(domainPO, ident.namespace()); + } + + /** + * Gets the domains with the given identifiers. Identifiers of domains that do not exist are + * skipped. + * + * @param identifiers The domain identifiers, all in the same metalake. + * @return The domains that exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "batchGetDomainByIdentifier") + public List<DomainEntity> batchGetDomainByIdentifier(List<NameIdentifier> identifiers) { + if (identifiers.isEmpty()) { + return Collections.emptyList(); + } + NameIdentifier firstIdent = identifiers.get(0); + String metalakeName = firstIdent.namespace().level(0); + List<String> domainNames = + identifiers.stream().map(NameIdentifier::name).collect(Collectors.toList()); + + List<DomainPO> domainPOs = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, + mapper -> mapper.listDomainPOsByMetalakeAndDomainNames(metalakeName, domainNames)); + return DomainPO.fromDomainPOs(domainPOs, firstIdent.namespace()); + } + + /** + * Inserts a domain. + * + * @param domainEntity The domain to insert. + * @param overwritten Whether to replace an active domain with the same name or ID. + * @throws IOException If the domain cannot be stored. + * @throws EntityAlreadyExistsException If a domain with the same name exists and {@code + * overwritten} is false. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertDomain") + public void insertDomain(DomainEntity domainEntity, boolean overwritten) throws IOException { + Namespace ns = domainEntity.namespace(); + String metalakeName = ns.level(0); + + try { + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeName); + } + + DomainPO domainPO = DomainPO.initializeDomainPO(domainEntity, metalakePO.getMetalakeId()); + + SessionUtils.doMultipleWithCommit( + () -> + lockMetalakeForDomainWrite(metalakePO.getMetalakeName(), metalakePO.getMetalakeId()), + () -> insertDomainWithoutCommit(domainEntity, domainPO, overwritten)); + } catch (RuntimeException e) { + try { + ExceptionUtils.checkSQLException( + e, Entity.EntityType.DOMAIN, domainEntity.nameIdentifier().toString()); + } catch (EntityAlreadyExistsException duplicate) { + if (overwritten) { + // A missing-row locking read does not fence a concurrent insert at READ_COMMITTED. + // Propagate the conflict so the whole transaction is rolled back before retrying. + throw ExceptionUtils.concurrentModification( + Entity.EntityType.DOMAIN, domainEntity.nameIdentifier()); + } + throw duplicate; + } + throw e; + } + } + + /** + * Updates a domain. The domain row is locked before the updater runs, so the updater is given the + * current domain and concurrent alters apply one after another instead of failing on each other. + * The write still compares the locked version as a backstop. + * + * <p>The returned domain is read back in the same transaction, so its backing tag name comes from + * the tag it is now bound to rather than from the updater's entity. + * + * @param identifier The domain identifier. + * @param updater The function that returns the updated domain. + * @param <E> The entity type. + * @return The updated domain, as stored. + * @throws IOException If the domain cannot be stored. + * @throws NoSuchEntityException If the domain does not exist. + * @throws EntityAlreadyExistsException If the domain is renamed to a name that is in use. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "updateDomain") + public <E extends Entity & HasIdentifier> DomainEntity updateDomain( + NameIdentifier identifier, Function<E, E> updater) throws IOException { + String metalakeName = identifier.namespace().level(0); + + try { + DomainPO observedDomainPO = getDomainPOByMetalakeAndName(metalakeName, identifier.name()); + DomainPO[] storedDomainPO = new DomainPO[1]; + SessionUtils.doWithCommit( + DomainMetaMapper.class, + mapper -> { + lockAndGetDomainPO(identifier, observedDomainPO); Review Comment: Fixed in e7ac948a60. `updateDomain` now takes the metalake fence before the domain row, like insert and delete. Its Javadoc says that the updater runs inside the transaction while the domain row is locked, and that any lock it takes must follow metalake, then domains, then tags. I added `testAlterHoldsMetalakeFenceAgainstCascadeDrop`: the updater locks the tag row after a latch, while a cascade drop is started. The drop now waits at the fence, and both succeed. Without the fence the test fails on H2. On the earlier thread: by "lock the tag row first" I meant before the entity is built, not before the domain. The order is domain, then tag, as in the design table. ########## core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.gravitino.storage.relational.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.MetadataObject; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.meta.DomainEntity; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper; +import org.apache.gravitino.storage.relational.po.DomainPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.utils.ExceptionUtils; +import org.apache.gravitino.storage.relational.utils.SessionUtils; + +/** The service that stores domains in the {@code domain_meta} table. */ +public class DomainMetaService { + + private static final DomainMetaService INSTANCE = new DomainMetaService(); + + private DomainMetaService() {} + + /** + * Returns the singleton service instance. + * + * @return The domain meta service. + */ + public static DomainMetaService getInstance() { + return INSTANCE; + } + + /** + * Lists the domains of a metalake. + * + * @param ns The domain namespace of the metalake. + * @return The domains of the metalake. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "listDomainsByNamespace") + public List<DomainEntity> listDomainsByNamespace(Namespace ns) { + String metalakeName = ns.level(0); + List<DomainPO> domainPOs = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, mapper -> mapper.listDomainPOsByMetalake(metalakeName)); + return DomainPO.fromDomainPOs(domainPOs, ns); + } + + /** + * Gets a domain by its identifier. + * + * @param ident The domain identifier. + * @return The domain. + * @throws NoSuchEntityException If the domain does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getDomainByIdentifier") + public DomainEntity getDomainByIdentifier(NameIdentifier ident) { + String metalakeName = ident.namespace().level(0); + DomainPO domainPO = getDomainPOByMetalakeAndName(metalakeName, ident.name()); + return DomainPO.fromDomainPO(domainPO, ident.namespace()); + } + + /** + * Gets the domains with the given identifiers. Identifiers of domains that do not exist are + * skipped. + * + * @param identifiers The domain identifiers, all in the same metalake. + * @return The domains that exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "batchGetDomainByIdentifier") + public List<DomainEntity> batchGetDomainByIdentifier(List<NameIdentifier> identifiers) { + if (identifiers.isEmpty()) { + return Collections.emptyList(); + } + NameIdentifier firstIdent = identifiers.get(0); + String metalakeName = firstIdent.namespace().level(0); + List<String> domainNames = + identifiers.stream().map(NameIdentifier::name).collect(Collectors.toList()); + + List<DomainPO> domainPOs = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, + mapper -> mapper.listDomainPOsByMetalakeAndDomainNames(metalakeName, domainNames)); + return DomainPO.fromDomainPOs(domainPOs, firstIdent.namespace()); + } + + /** + * Inserts a domain. + * + * @param domainEntity The domain to insert. + * @param overwritten Whether to replace an active domain with the same name or ID. + * @throws IOException If the domain cannot be stored. + * @throws EntityAlreadyExistsException If a domain with the same name exists and {@code + * overwritten} is false. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertDomain") + public void insertDomain(DomainEntity domainEntity, boolean overwritten) throws IOException { + Namespace ns = domainEntity.namespace(); + String metalakeName = ns.level(0); + + try { + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeName); + } + + DomainPO domainPO = DomainPO.initializeDomainPO(domainEntity, metalakePO.getMetalakeId()); + + SessionUtils.doMultipleWithCommit( + () -> + lockMetalakeForDomainWrite(metalakePO.getMetalakeName(), metalakePO.getMetalakeId()), + () -> insertDomainWithoutCommit(domainEntity, domainPO, overwritten)); + } catch (RuntimeException e) { + try { + ExceptionUtils.checkSQLException( + e, Entity.EntityType.DOMAIN, domainEntity.nameIdentifier().toString()); + } catch (EntityAlreadyExistsException duplicate) { + if (overwritten) { + // A missing-row locking read does not fence a concurrent insert at READ_COMMITTED. + // Propagate the conflict so the whole transaction is rolled back before retrying. + throw ExceptionUtils.concurrentModification( + Entity.EntityType.DOMAIN, domainEntity.nameIdentifier()); + } + throw duplicate; + } + throw e; + } + } + + /** + * Updates a domain. The domain row is locked before the updater runs, so the updater is given the + * current domain and concurrent alters apply one after another instead of failing on each other. + * The write still compares the locked version as a backstop. + * + * <p>The returned domain is read back in the same transaction, so its backing tag name comes from + * the tag it is now bound to rather than from the updater's entity. + * + * @param identifier The domain identifier. + * @param updater The function that returns the updated domain. + * @param <E> The entity type. + * @return The updated domain, as stored. + * @throws IOException If the domain cannot be stored. + * @throws NoSuchEntityException If the domain does not exist. + * @throws EntityAlreadyExistsException If the domain is renamed to a name that is in use. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "updateDomain") + public <E extends Entity & HasIdentifier> DomainEntity updateDomain( + NameIdentifier identifier, Function<E, E> updater) throws IOException { + String metalakeName = identifier.namespace().level(0); + + try { + DomainPO observedDomainPO = getDomainPOByMetalakeAndName(metalakeName, identifier.name()); + DomainPO[] storedDomainPO = new DomainPO[1]; + SessionUtils.doWithCommit( + DomainMetaMapper.class, + mapper -> { + lockAndGetDomainPO(identifier, observedDomainPO); + // The locking read cannot join the tag table, so the locked row is read again with the + // name of its backing tag. + DomainPO lockedDomainPO = mapper.selectDomainMetaById(observedDomainPO.domainId()); + DomainEntity oldDomainEntity = + DomainPO.fromDomainPO(lockedDomainPO, identifier.namespace()); + DomainEntity updatedDomainEntity = (DomainEntity) updater.apply((E) oldDomainEntity); + Preconditions.checkArgument( + Objects.equals(oldDomainEntity.id(), updatedDomainEntity.id()), + "The updated domain entity id: %s must have the same id as the old entity id %s", + updatedDomainEntity.id(), + oldDomainEntity.id()); + + DomainPO newDomainPO = DomainPO.updateDomainPO(lockedDomainPO, updatedDomainEntity); + updateDomainWithVersion(identifier, mapper, lockedDomainPO, newDomainPO); + storedDomainPO[0] = mapper.selectDomainMetaById(lockedDomainPO.domainId()); + }); + + return DomainPO.fromDomainPO(storedDomainPO[0], identifier.namespace()); + } catch (RuntimeException e) { + ExceptionUtils.checkSQLException(e, Entity.EntityType.DOMAIN, identifier.toString()); + throw e; + } + } + + /** + * Deletes a domain, together with its owner and the privileges granted on it. The backing tag and + * its assignments are kept. + * + * @param identifier The domain identifier. + * @return True if the domain is deleted, false if it does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "deleteDomain") + public boolean deleteDomain(NameIdentifier identifier) { + DomainPO domainPO; + try { + domainPO = getDomainPOByMetalakeAndName(identifier.namespace().level(0), identifier.name()); + } catch (NoSuchEntityException e) { + return false; + } + return deleteDomain(identifier, domainPO); + } + + /** + * Hard-deletes the domains soft-deleted before the given time. + * + * @param legacyTimeline The time before which soft-deleted domains are removed. + * @param limit The maximum number of domains to remove. + * @return The number of removed domains. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "deleteDomainMetasByLegacyTimeline") + public int deleteDomainMetasByLegacyTimeline(long legacyTimeline, int limit) { + return SessionUtils.doWithCommitAndFetchResult( + DomainMetaMapper.class, + mapper -> mapper.deleteDomainMetasByLegacyTimeline(legacyTimeline, limit)); + } + + /** + * Gets the ID of a domain. + * + * @param metalakeId The metalake ID. + * @param domainName The domain name. + * @return The domain ID. + * @throws NoSuchEntityException If the domain does not exist. + */ + public Long getDomainIdByMetalakeIdAndName(Long metalakeId, String domainName) { + DomainPO domainPO = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, + mapper -> mapper.selectDomainMetaByMetalakeIdAndName(metalakeId, domainName)); + + if (domainPO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.DOMAIN.name().toLowerCase(), + domainName); + } + return domainPO.domainId(); + } + + /** + * Deletes the domain the caller observed. The domain row is locked first, so a domain that was + * only altered since it was read is still deleted, and the dependent rows are cleaned up in the + * same transaction as the domain row itself. If another transaction deletes or renames the + * observed domain first, this method returns {@code false}. + * + * <p>The observed row is a parameter so that a test can hand in a stale one; production callers + * use {@link #deleteDomain(NameIdentifier)}, which reads it first. + */ + @VisibleForTesting + boolean deleteDomain(NameIdentifier identifier, DomainPO domainPO) { + long domainId = domainPO.domainId(); + + // The metalake lock comes first. Dropping the metalake soft-deletes the privileges before the + // domains, while this delete changes the domain row before its privileges; without the lock + // the two could deadlock on a domain with privileges. The domain row is locked next, and a row + // that is gone or renamed makes the delete report the domain as missing before any dependent + // row is touched. The version-checked delete keeps the version as a backstop. The cleanups that + // follow are blanket soft deletes for this domain ID, so their row counts carry no information + // and are not checked. + try { + SessionUtils.doMultipleWithCommit( + () -> lockMetalakeForDomainWrite(identifier.namespace().level(0), domainPO.metalakeId()), + () -> deleteDomainWithVersion(identifier, lockAndGetDomainPO(identifier, domainPO)), + () -> + SessionUtils.doWithoutCommit( + OwnerMetaMapper.class, + mapper -> + mapper.softDeleteOwnerRelByMetadataObjectIdAndType( + domainId, MetadataObject.Type.DOMAIN.name())), + () -> + SessionUtils.doWithoutCommit( + SecurableObjectMapper.class, + mapper -> + mapper.softDeleteObjectRelsByMetadataObject( + domainId, MetadataObject.Type.DOMAIN.name()))); + } catch (NoSuchEntityException e) { + return false; + } + + return true; + } + + /** + * Holds the parent metalake row for the rest of the transaction, so a domain cannot be created or + * deleted while the metalake is being dropped. The lock is shared, so domains under the same + * metalake can be written at the same time, while dropping the metalake takes an exclusive lock + * on the row. A metalake that is gone or renamed is reported as missing. + */ + private void lockMetalakeForDomainWrite(String metalakeName, Long metalakeId) { + OccWriteSupport.lockParentForChildWrite( + metalakeName, + Entity.EntityType.METALAKE, + () -> + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, + mapper -> mapper.selectMetalakeMetaByIdForShare(metalakeId)), + null, + current -> Objects.equals(current.getMetalakeName(), metalakeName)); + } + + /** + * Writes a new domain or replaces the active domain selected by name or stable ID. The overwrite + * path uses locking reads instead of a database-specific upsert, which keeps the stored domain ID + * and version stable across all databases. + */ + private void insertDomainWithoutCommit( + DomainEntity domainEntity, DomainPO initializedDomainPO, boolean overwritten) { + if (!overwritten) { + insertNewDomainWithoutCommit(initializedDomainPO); + return; + } + + DomainPO existingDomainPO = findAndLockDomainForOverwrite(initializedDomainPO); + if (existingDomainPO == null) { + if (SessionUtils.getWithoutCommit( + DomainMetaMapper.class, + mapper -> mapper.countDeletedDomainMetasById(initializedDomainPO.domainId())) + > 0) { + throw new EntityAlreadyExistsException( Review Comment: Added in e7ac948a60: - `testOverwriteRejectsIdOfDeletedDomain` deletes a domain, then overwrites with the same ID. It asserts `EntityAlreadyExistsException`, and that the message contains the ID. - `testOverwriteRacingInsertOfSameName` holds an uncommitted insert of the same name at a latch while an overwrite insert runs. On H2 the overwrite's locking read misses the uncommitted row, so the insert hits the unique key and is reported as `OptimisticLockException`. A database whose locking read waits for that row replaces it instead, so the test accepts either outcome and checks that exactly one domain remains. ########## core/src/main/java/org/apache/gravitino/storage/relational/service/DomainMetaService.java: ########## @@ -0,0 +1,473 @@ +/* + * 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.gravitino.storage.relational.service; + +import static org.apache.gravitino.metrics.source.MetricsSource.GRAVITINO_RELATIONAL_STORE_METRIC_NAME; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityAlreadyExistsException; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.MetadataObject; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.exceptions.NoSuchEntityException; +import org.apache.gravitino.meta.DomainEntity; +import org.apache.gravitino.metrics.Monitored; +import org.apache.gravitino.storage.relational.mapper.DomainMetaMapper; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.OwnerMetaMapper; +import org.apache.gravitino.storage.relational.mapper.SecurableObjectMapper; +import org.apache.gravitino.storage.relational.po.DomainPO; +import org.apache.gravitino.storage.relational.po.MetalakePO; +import org.apache.gravitino.storage.relational.utils.ExceptionUtils; +import org.apache.gravitino.storage.relational.utils.SessionUtils; + +/** The service that stores domains in the {@code domain_meta} table. */ +public class DomainMetaService { + + private static final DomainMetaService INSTANCE = new DomainMetaService(); + + private DomainMetaService() {} + + /** + * Returns the singleton service instance. + * + * @return The domain meta service. + */ + public static DomainMetaService getInstance() { + return INSTANCE; + } + + /** + * Lists the domains of a metalake. + * + * @param ns The domain namespace of the metalake. + * @return The domains of the metalake. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "listDomainsByNamespace") + public List<DomainEntity> listDomainsByNamespace(Namespace ns) { + String metalakeName = ns.level(0); + List<DomainPO> domainPOs = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, mapper -> mapper.listDomainPOsByMetalake(metalakeName)); + return DomainPO.fromDomainPOs(domainPOs, ns); + } + + /** + * Gets a domain by its identifier. + * + * @param ident The domain identifier. + * @return The domain. + * @throws NoSuchEntityException If the domain does not exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "getDomainByIdentifier") + public DomainEntity getDomainByIdentifier(NameIdentifier ident) { + String metalakeName = ident.namespace().level(0); + DomainPO domainPO = getDomainPOByMetalakeAndName(metalakeName, ident.name()); + return DomainPO.fromDomainPO(domainPO, ident.namespace()); + } + + /** + * Gets the domains with the given identifiers. Identifiers of domains that do not exist are + * skipped. + * + * @param identifiers The domain identifiers, all in the same metalake. + * @return The domains that exist. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "batchGetDomainByIdentifier") + public List<DomainEntity> batchGetDomainByIdentifier(List<NameIdentifier> identifiers) { + if (identifiers.isEmpty()) { + return Collections.emptyList(); + } + NameIdentifier firstIdent = identifiers.get(0); + String metalakeName = firstIdent.namespace().level(0); + List<String> domainNames = + identifiers.stream().map(NameIdentifier::name).collect(Collectors.toList()); + + List<DomainPO> domainPOs = + SessionUtils.getWithoutCommit( + DomainMetaMapper.class, + mapper -> mapper.listDomainPOsByMetalakeAndDomainNames(metalakeName, domainNames)); + return DomainPO.fromDomainPOs(domainPOs, firstIdent.namespace()); + } + + /** + * Inserts a domain. + * + * @param domainEntity The domain to insert. + * @param overwritten Whether to replace an active domain with the same name or ID. + * @throws IOException If the domain cannot be stored. + * @throws EntityAlreadyExistsException If a domain with the same name exists and {@code + * overwritten} is false. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "insertDomain") + public void insertDomain(DomainEntity domainEntity, boolean overwritten) throws IOException { + Namespace ns = domainEntity.namespace(); + String metalakeName = ns.level(0); + + try { + MetalakePO metalakePO = + SessionUtils.getWithoutCommit( + MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(metalakeName)); + if (metalakePO == null) { + throw new NoSuchEntityException( + NoSuchEntityException.NO_SUCH_ENTITY_MESSAGE, + Entity.EntityType.METALAKE.name().toLowerCase(), + metalakeName); + } + + DomainPO domainPO = DomainPO.initializeDomainPO(domainEntity, metalakePO.getMetalakeId()); + + SessionUtils.doMultipleWithCommit( + () -> + lockMetalakeForDomainWrite(metalakePO.getMetalakeName(), metalakePO.getMetalakeId()), + () -> insertDomainWithoutCommit(domainEntity, domainPO, overwritten)); + } catch (RuntimeException e) { + try { + ExceptionUtils.checkSQLException( + e, Entity.EntityType.DOMAIN, domainEntity.nameIdentifier().toString()); + } catch (EntityAlreadyExistsException duplicate) { + if (overwritten) { + // A missing-row locking read does not fence a concurrent insert at READ_COMMITTED. + // Propagate the conflict so the whole transaction is rolled back before retrying. + throw ExceptionUtils.concurrentModification( + Entity.EntityType.DOMAIN, domainEntity.nameIdentifier()); + } + throw duplicate; + } + throw e; + } + } + + /** + * Updates a domain. The domain row is locked before the updater runs, so the updater is given the + * current domain and concurrent alters apply one after another instead of failing on each other. + * The write still compares the locked version as a backstop. + * + * <p>The returned domain is read back in the same transaction, so its backing tag name comes from + * the tag it is now bound to rather than from the updater's entity. + * + * @param identifier The domain identifier. + * @param updater The function that returns the updated domain. + * @param <E> The entity type. + * @return The updated domain, as stored. + * @throws IOException If the domain cannot be stored. + * @throws NoSuchEntityException If the domain does not exist. + * @throws EntityAlreadyExistsException If the domain is renamed to a name that is in use. + */ + @Monitored( + metricsSource = GRAVITINO_RELATIONAL_STORE_METRIC_NAME, + baseMetricName = "updateDomain") + public <E extends Entity & HasIdentifier> DomainEntity updateDomain( + NameIdentifier identifier, Function<E, E> updater) throws IOException { + String metalakeName = identifier.namespace().level(0); + + try { + DomainPO observedDomainPO = getDomainPOByMetalakeAndName(metalakeName, identifier.name()); Review Comment: Yes, that is the intent: the in-flight call reports the domain as missing, and a later call with the name acts on the new domain. Documented on `updateDomain` and `deleteDomain` in e7ac948a60. ########## core/src/main/java/org/apache/gravitino/meta/DomainEntity.java: ########## @@ -0,0 +1,200 @@ +/* + * 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.gravitino.meta; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Maps; +import java.util.Collections; +import java.util.Map; +import java.util.Optional; +import javax.annotation.Nullable; +import lombok.AccessLevel; +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.ToString; +import lombok.experimental.Accessors; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Audit; +import org.apache.gravitino.Auditable; +import org.apache.gravitino.Entity; +import org.apache.gravitino.EntityFieldLimits; +import org.apache.gravitino.Field; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.domain.Domain; +import org.apache.gravitino.tag.TagValue; + +/** + * The stored form of a {@link Domain}. It holds the domain's data only; the operations on {@link + * Domain} are served by the domain manager. + * + * <p>The backing tag is stored with the domain as the ID of the tag and an optional value; the tag + * name is read from the tag when the domain is read. If the tag is deleted, the ID stays on the + * domain but has no name, so the domain has no backing tag. The parent is not stored with the + * domain: it comes from the sub-domain relation and is set on the entity when it is read. + */ +@Getter +@Accessors(fluent = true) +@EqualsAndHashCode +@ToString +public class DomainEntity implements Entity, Auditable, HasIdentifier { + + /** The unique id of the domain entity. */ + public static final Field ID = + Field.required("id", Long.class, "The unique id of the domain entity."); + + /** The name of the domain entity. */ + public static final Field NAME = + Field.required("name", "The name of the domain entity.", EntityFieldLimits.MAX_NAME_LENGTH); + + /** The comment of the domain entity. */ + public static final Field COMMENT = + Field.optional( + "comment", "The comment of the domain entity.", EntityFieldLimits.MAX_COMMENT_LENGTH); + + /** The properties of the domain entity. */ + public static final Field PROPERTIES = + Field.optional("properties", Map.class, "The properties of the domain entity."); + + /** The status of the domain entity. */ + public static final Field STATUS = + Field.required("status", Domain.Status.class, "The status of the domain entity."); + + /** The ID of the backing tag of the domain entity. */ + public static final Field BACKING_TAG_ID = + Field.optional( + "backing_tag_id", Long.class, "The ID of the backing tag of the domain entity."); + + /** The value of the backing tag of the domain entity. */ + public static final Field BACKING_TAG_VALUE = + Field.optional( + "backing_tag_value", + "The value of the backing tag of the domain entity.", + EntityFieldLimits.MAX_TAG_VALUE_LENGTH); + + /** The audit details of the domain entity. */ + public static final Field AUDIT_INFO = + Field.required("audit_info", Audit.class, "The audit details of the domain entity."); + + private final Long id; + private final String name; + private final Namespace namespace; + @Nullable private final String comment; + private final Map<String, String> properties; + private final Domain.Status status; + + @Getter(AccessLevel.NONE) + @Nullable + private final String parent; + + /** The ID of the backing tag, or null if the domain has none. */ + @Nullable private final Long backingTagId; + + /** The name of the backing tag, or null if the domain has none or the tag is deleted. */ + @Getter(AccessLevel.NONE) + @Nullable + private final String backingTagName; + + /** The value of the backing tag, or null if the tag is bound without a value. */ + @Nullable private final String backingTagValue; + + private final Audit auditInfo; + + @lombok.Builder(setterPrefix = "with", toBuilder = true) + private DomainEntity( + Long id, + String name, + Namespace namespace, + @Nullable String comment, + Map<String, String> properties, + Domain.Status status, + @Nullable String parent, + @Nullable Long backingTagId, + @Nullable String backingTagName, + @Nullable String backingTagValue, + Audit auditInfo) { + this.id = id; + this.name = name; + this.namespace = namespace; + this.comment = comment; + this.properties = properties; + this.status = status; + this.parent = parent; + this.backingTagId = backingTagId; + // The name is read from the bound tag, so it follows the ID: clearing the ID unbinds the tag. + this.backingTagName = backingTagId == null ? null : backingTagName; + this.backingTagValue = backingTagValue; + this.auditInfo = auditInfo; + validate(); + Preconditions.checkArgument( + backingTagId != null || backingTagValue == null, + "The backing tag value of domain %s needs a backing tag ID", + name); + // A tag bound without a value is stored as NULL, so the value is never blank. + Preconditions.checkArgument( + backingTagValue == null || StringUtils.isNotBlank(backingTagValue), + "The backing tag value of domain %s must not be blank", + name); + } + Review Comment: `validate()` is the default method of the `Entity` interface (`Entity.java:98`), which validates the entity's `fields()`. The class compiles and runs it as is, so no change. ########## core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/DomainMetaBaseSQLProvider.java: ########## @@ -0,0 +1,302 @@ +/* + * 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.gravitino.storage.relational.mapper.provider.base; + +import static org.apache.gravitino.storage.relational.mapper.DomainMetaMapper.DOMAIN_TABLE_NAME; + +import java.util.List; +import org.apache.gravitino.storage.relational.mapper.MetalakeMetaMapper; +import org.apache.gravitino.storage.relational.mapper.TagMetaMapper; +import org.apache.gravitino.storage.relational.mapper.provider.DatabaseTimeSQL; +import org.apache.gravitino.storage.relational.po.DomainPO; +import org.apache.ibatis.annotations.Param; + +/** The SQL of the {@code domain_meta} table shared by all backends. */ +public class DomainMetaBaseSQLProvider { + + private static final String DOMAIN_COLUMNS = + "dm.domain_id as domainId, dm.domain_name as domainName," + + " dm.metalake_id as metalakeId," + + " dm.domain_comment as comment," + + " dm.domain_status as status," + + " dm.backing_tag_id as backingTagId," + + " dm.backing_tag_value as backingTagValue," + + " dm.properties as properties," + + " dm.audit_info as auditInfo," + + " dm.current_version as currentVersion," + + " dm.last_version as lastVersion," + + " dm.deleted_at as deletedAt"; + + // Reads take the backing tag name from the live tag, so a deleted tag reads as no backing tag. + // Locking reads do not join: they must lock the domain row only, and PostgreSQL rejects FOR + // UPDATE on the nullable side of an outer join. + private static final String DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME = + DOMAIN_COLUMNS + ", bt.tag_name as backingTagName"; + + private static final String BACKING_TAG_JOIN = + " LEFT JOIN " + + TagMetaMapper.TAG_TABLE_NAME + + " bt ON dm.backing_tag_id = bt.tag_id AND dm.metalake_id = bt.metalake_id" + + " AND bt.deleted_at = 0"; + + /** + * Returns SQL that lists the active domains of a metalake. + * + * @param metalakeName The metalake name. + * @return The SQL. + */ + public String listDomainPOsByMetalake(@Param("metalakeName") String metalakeName) { + return "SELECT " + + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME + + " FROM " + + DOMAIN_TABLE_NAME + + " dm JOIN " + + MetalakeMetaMapper.TABLE_NAME + + " mm ON dm.metalake_id = mm.metalake_id" + + BACKING_TAG_JOIN + + " WHERE mm.metalake_name = #{metalakeName} AND dm.deleted_at = 0 AND mm.deleted_at = 0"; + } + + /** + * Returns SQL that selects an active domain by metalake name and domain name. + * + * @param metalakeName The metalake name. + * @param domainName The domain name. + * @return The SQL. + */ + public String selectDomainMetaByMetalakeAndName( + @Param("metalakeName") String metalakeName, @Param("domainName") String domainName) { + return "SELECT " + + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME + + " FROM " + + DOMAIN_TABLE_NAME + + " dm JOIN " + + MetalakeMetaMapper.TABLE_NAME + + " mm ON dm.metalake_id = mm.metalake_id" + + BACKING_TAG_JOIN + + " WHERE mm.metalake_name = #{metalakeName} AND dm.domain_name = #{domainName}" + + " AND dm.deleted_at = 0 AND mm.deleted_at = 0"; + } + + /** + * Returns SQL that selects an active domain by metalake ID and domain name. + * + * @param metalakeId The metalake ID. + * @param domainName The domain name. + * @return The SQL. + */ + public String selectDomainMetaByMetalakeIdAndName( + @Param("metalakeId") Long metalakeId, @Param("domainName") String domainName) { + return "SELECT " + + DOMAIN_COLUMNS + + " FROM " + + DOMAIN_TABLE_NAME + + " dm WHERE dm.metalake_id = #{metalakeId} AND dm.domain_name = #{domainName}" + + " AND dm.deleted_at = 0"; + } + + /** + * Returns SQL that selects and exclusively locks an active domain by its natural key. + * + * @param metalakeId The metalake ID. + * @param domainName The domain name. + * @return The SQL. + */ + public String selectDomainMetaByMetalakeIdAndNameForUpdate( + @Param("metalakeId") Long metalakeId, @Param("domainName") String domainName) { + return selectDomainMetaByMetalakeIdAndName(metalakeId, domainName) + " FOR UPDATE"; + } + + /** + * Returns SQL that selects an active domain by ID. + * + * @param domainId The domain ID. + * @return The SQL. + */ + public String selectDomainMetaById(@Param("domainId") Long domainId) { + return "SELECT " + + DOMAIN_COLUMNS_WITH_BACKING_TAG_NAME + + " FROM " + + DOMAIN_TABLE_NAME + + " dm" + + BACKING_TAG_JOIN + + " WHERE dm.domain_id = #{domainId} AND dm.deleted_at = 0"; + } + + /** + * Returns SQL that selects and exclusively locks an active domain by ID. + * + * @param domainId The domain ID. + * @return The SQL. + */ + public String selectDomainMetaByIdForUpdate(@Param("domainId") Long domainId) { + return "SELECT " + + DOMAIN_COLUMNS + + " FROM " + + DOMAIN_TABLE_NAME + + " dm WHERE dm.domain_id = #{domainId} AND dm.deleted_at = 0 FOR UPDATE"; + } + + /** + * Returns SQL that lists the active domains with the given IDs. + * + * @param domainIds The domain IDs. + * @return The SQL. + */ + public String listDomainPOsByDomainIds(@Param("domainIds") List<Long> domainIds) { + return "<script>" + + "SELECT " + + DOMAIN_COLUMNS + + " FROM " + + DOMAIN_TABLE_NAME + + " dm WHERE dm.deleted_at = 0" + + " AND dm.domain_id IN (" + + "<foreach collection='domainIds' item='domainId' separator=','>" + + "#{domainId}" + + "</foreach>" + + ")" Review Comment: The only caller, `MetadataObjectService.getDomainObjectsFullName`, returns early for an empty list, which is the same contract as the other `IN (...)` providers such as `TagMetaBaseSQLProvider`. No change. -- 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]
