-
Notifications
You must be signed in to change notification settings - Fork 319
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Jiebao Xiao
committed
Feb 26, 2024
1 parent
e5c25f2
commit 7760554
Showing
8 changed files
with
202 additions
and
148 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
133 changes: 133 additions & 0 deletions
133
...rc/main/java/com/datastrato/gravitino/storage/relational/service/MetalakeMetaService.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| /* | ||
| * Copyright 2024 Datastrato Pvt Ltd. | ||
| * This software is licensed under the Apache License version 2. | ||
| */ | ||
|
|
||
| package com.datastrato.gravitino.storage.relational.service; | ||
|
|
||
| import com.datastrato.gravitino.Entity; | ||
| import com.datastrato.gravitino.EntityAlreadyExistsException; | ||
| import com.datastrato.gravitino.HasIdentifier; | ||
| import com.datastrato.gravitino.NameIdentifier; | ||
| import com.datastrato.gravitino.exceptions.NoSuchEntityException; | ||
| import com.datastrato.gravitino.meta.BaseMetalake; | ||
| import com.datastrato.gravitino.storage.relational.mapper.MetalakeMetaMapper; | ||
| import com.datastrato.gravitino.storage.relational.po.MetalakePO; | ||
| import com.datastrato.gravitino.storage.relational.utils.POConverters; | ||
| import com.datastrato.gravitino.storage.relational.utils.SessionUtils; | ||
| import com.google.common.base.Preconditions; | ||
| import java.io.IOException; | ||
| import java.sql.SQLIntegrityConstraintViolationException; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
| import java.util.function.Function; | ||
|
|
||
| public class MetalakeMetaService { | ||
| private static final MetalakeMetaService INSTANCE = new MetalakeMetaService(); | ||
|
|
||
| public static MetalakeMetaService getInstance() { | ||
| return INSTANCE; | ||
| } | ||
|
|
||
| private MetalakeMetaService() {} | ||
|
|
||
| public List<BaseMetalake> listMetalakes() { | ||
| List<MetalakePO> metalakePOS = | ||
| SessionUtils.getWithoutCommit( | ||
| MetalakeMetaMapper.class, MetalakeMetaMapper::listMetalakePOs); | ||
| return POConverters.fromMetalakePOs(metalakePOS); | ||
| } | ||
|
|
||
| public BaseMetalake getMetalakeByIdent(NameIdentifier ident) { | ||
| MetalakePO metalakePO = | ||
| SessionUtils.getWithoutCommit( | ||
| MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(ident.name())); | ||
| if (metalakePO == null) { | ||
| throw new NoSuchEntityException("No such entity: %s", ident.toString()); | ||
| } | ||
| return POConverters.fromMetalakePO(metalakePO); | ||
| } | ||
|
|
||
| public void insertMetalake(BaseMetalake baseMetalake, boolean overwrite) { | ||
| try { | ||
| SessionUtils.doWithCommit( | ||
| MetalakeMetaMapper.class, | ||
| mapper -> { | ||
| MetalakePO po = POConverters.initializeMetalakePOWithVersion(baseMetalake); | ||
| if (overwrite) { | ||
| mapper.insertMetalakeMetaOnDuplicateKeyUpdate(po); | ||
| } else { | ||
| mapper.insertMetalakeMeta(po); | ||
| } | ||
| }); | ||
| } catch (RuntimeException re) { | ||
| if (re.getCause() != null | ||
| && re.getCause().getCause() != null | ||
| && re.getCause().getCause() instanceof SQLIntegrityConstraintViolationException) { | ||
| // TODO We should make more fine-grained exception judgments | ||
| // Usually throwing `SQLIntegrityConstraintViolationException` means that | ||
| // SQL violates the constraints of `primary key` and `unique key`. | ||
| // We simply think that the entity already exists at this time. | ||
| throw new EntityAlreadyExistsException( | ||
| String.format( | ||
| "Metalake entity: %s already exists", baseMetalake.nameIdentifier().name())); | ||
| } | ||
| throw re; | ||
| } | ||
| } | ||
|
|
||
| public <E extends Entity & HasIdentifier> BaseMetalake updateMetalake( | ||
| NameIdentifier ident, Function<E, E> updater) throws IOException { | ||
| MetalakePO oldMetalakePO = | ||
| SessionUtils.getWithoutCommit( | ||
| MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeMetaByName(ident.name())); | ||
| if (oldMetalakePO == null) { | ||
| throw new NoSuchEntityException("No such entity: %s", ident.toString()); | ||
| } | ||
|
|
||
| BaseMetalake oldMetalakeEntity = POConverters.fromMetalakePO(oldMetalakePO); | ||
| BaseMetalake newMetalakeEntity = (BaseMetalake) updater.apply((E) oldMetalakeEntity); | ||
| Preconditions.checkArgument( | ||
| Objects.equals(oldMetalakeEntity.id(), newMetalakeEntity.id()), | ||
| "The updated metalake entity id: %s should be same with the metalake entity id before: %s", | ||
| newMetalakeEntity.id(), | ||
| oldMetalakeEntity.id()); | ||
| MetalakePO newMetalakePO = | ||
| POConverters.updateMetalakePOWithVersion(oldMetalakePO, newMetalakeEntity); | ||
|
|
||
| Integer updateResult = | ||
| SessionUtils.doWithCommitAndFetchResult( | ||
| MetalakeMetaMapper.class, | ||
| mapper -> mapper.updateMetalakeMeta(newMetalakePO, oldMetalakePO)); | ||
| if (updateResult > 0) { | ||
| return newMetalakeEntity; | ||
| } else { | ||
| throw new IOException("Failed to update the entity: " + ident); | ||
| } | ||
| } | ||
|
|
||
| public boolean deleteMetalake(NameIdentifier ident, boolean cascade) { | ||
| Long metalakeId = | ||
| SessionUtils.getWithoutCommit( | ||
| MetalakeMetaMapper.class, mapper -> mapper.selectMetalakeIdMetaByName(ident.name())); | ||
| if (metalakeId != null) { | ||
| if (cascade) { | ||
| SessionUtils.doMultipleWithCommit( | ||
| () -> | ||
| SessionUtils.doWithoutCommit( | ||
| MetalakeMetaMapper.class, | ||
| mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId)), | ||
| () -> { | ||
| // TODO We will cascade delete the metadata of sub-resources under the metalake | ||
| }); | ||
| } else { | ||
| // TODO Check whether the sub-resources are empty. If the sub-resources are not empty, | ||
| // deletion is not allowed. | ||
| SessionUtils.doWithCommit( | ||
| MetalakeMetaMapper.class, | ||
| mapper -> mapper.softDeleteMetalakeMetaByMetalakeId(metalakeId)); | ||
| } | ||
| } | ||
| return true; | ||
| } | ||
| } |
Oops, something went wrong.