yuqi1129 commented on code in PR #14281: URL: https://github.com/apache/gravitino/pull/14281#discussion_r4411913271
########## 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: **[Important] F2: this branch has no test.** It was asked for in the last two review rounds, and `grep reserved` in `TestDomainMetaService` still finds nothing. Suggested test: insert a domain, delete it, then insert with `overwritten = true` and the same ID. Assert `EntityAlreadyExistsException`, and assert that the message contains the ID. The concurrent branch at L161-167 (overwrite insert hits a duplicate key → `OptimisticLockException`) has no test either. You can drive it with two overwrite inserts of the same name that are held at a latch. ########## 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: **[Question] F4: a domain dropped and recreated with the same name while the call is in flight.** The name is read here first. The row is then locked by that ID. If the domain is dropped and a new one with the same name is created between those two steps, the lock check fails. `updateDomain` then throws `NoSuchEntityException`, and `deleteDomain` returns `false`, even though a domain with that name exists. The design says "later calls with that name act on the new domain", so failing an in-flight call can be fine. If that is the intent, please note it in the Javadoc. If the manager should always act on the current holder of the name, it could resolve the name once more on this miss. ########## 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: **[Important] F1: missing metalake fence. A backing-tag change will deadlock with a cascade metalake drop.** The design (Locking) sets one global order: *metalake fence (shared) → domain rows → tag rows*. Its table lists "Set or remove the backing tag" as `domain (exclusive) → tag (exclusive)`. The manager can only lock the tag inside the updater: `EntityStore` has no transaction across calls, so a tag lock taken before `updateDomain` is already released. This method skips the metalake fence. A cascade metalake drop locks in this order: - metalake X; - `softDeleteTagMetasByMetalakeId`, which locks the tag rows; - `softDeleteDomainMetasByMetalakeId`, which waits for the domain rows (`MetalakeMetaService.java:289-295`; also `:399-405`). So: - A (set backing tag) holds domain X and waits for tag X; - B (drop metalake) holds tag X and waits for domain X. **Repro.** I added a probe test. A runs `updateDomain` with an updater that calls `TagMetaMapper.selectTagByTagIdForUpdate(tag.id())`. While A holds the domain lock, B runs `MetalakeMetaService.deleteMetalake(ident, true)`. - MySQL: `MySQLTransactionRollbackException: Deadlock found when trying to get lock`. A is rolled back. - PostgreSQL: `deadlock detected ... while locking tuple in relation "tag_meta"`. - H2: `Deadlock detected`. B is rolled back. Nothing calls this today, but this method is the entry point that #13884 will use to change the backing tag. `b435243cb3` added the same fence to `deleteDomain` for the same reason. The update path is the matching place and still lacks it. **Suggested fix.** Take the fence first: ```java mapper -> { lockMetalakeForDomainWrite(metalakeName, observedDomainPO.metalakeId()); lockAndGetDomainPO(identifier, observedDomainPO); ... ``` I applied this locally and reran the probe. Both transactions succeed on MySQL and PostgreSQL, and all `TestDomainMetaService` cases still pass. On H2 the drop only hits a lock-wait timeout, because the probe holds the lock for 2s on purpose. That is not a deadlock. Please also say in the Javadoc: - the updater runs inside this transaction while it holds the domain row lock; - any lock the updater takes must follow the order metalake → domain → tag. This also answers the earlier thread on `DomainEntity.java:141`. There, "lock the tag row first" cannot hold across store calls, and it is the opposite of the design table. A regression test with this probe, using latches, would pin the behavior. -- 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]
