[#6566] improvement(core): Add the cache mechanism for metalake and use cache to load in-use information.
#6569
+186
−67
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
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 |
|---|---|---|
|
|
@@ -20,13 +20,20 @@ | |
|
|
||
| import static org.apache.gravitino.Metalake.PROPERTY_IN_USE; | ||
|
|
||
| import com.github.benmanes.caffeine.cache.Cache; | ||
| import com.github.benmanes.caffeine.cache.Caffeine; | ||
| import com.github.benmanes.caffeine.cache.Scheduler; | ||
| import com.google.common.annotations.VisibleForTesting; | ||
| import com.google.common.collect.Maps; | ||
| import com.google.common.util.concurrent.ThreadFactoryBuilder; | ||
| import java.io.IOException; | ||
| import java.time.Instant; | ||
| import java.util.Arrays; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ScheduledThreadPoolExecutor; | ||
| import java.util.concurrent.TimeUnit; | ||
| import org.apache.gravitino.Entity.EntityType; | ||
| import org.apache.gravitino.EntityAlreadyExistsException; | ||
| import org.apache.gravitino.EntityStore; | ||
|
|
@@ -64,6 +71,21 @@ public class MetalakeManager implements MetalakeDispatcher { | |
|
|
||
| private final IdGenerator idGenerator; | ||
|
|
||
| @VisibleForTesting | ||
| static final Cache<NameIdentifier, BaseMetalake> METALAKE_CACHE = | ||
|
There was a problem hiding this comment. If our goal is to accelerate the acquisition of There was a problem hiding this comment. Cacheing There was a problem hiding this comment. I think caching metalake is better, because the amount of metalake is quite limited, with small memory size we can improve the performance a lot, it is worthy to cache the metalake. There was a problem hiding this comment. Why do you use uppercase for this variable? Typically, we only use uppercase letter for There was a problem hiding this comment. It indeed has |
||
| Caffeine.newBuilder() | ||
| .expireAfterAccess(24, TimeUnit.HOURS) | ||
| .removalListener((k, v, c) -> LOG.info("Closing metalake {}.", k)) | ||
| .scheduler( | ||
| Scheduler.forScheduledExecutorService( | ||
| new ScheduledThreadPoolExecutor( | ||
| 1, | ||
| new ThreadFactoryBuilder() | ||
| .setDaemon(true) | ||
| .setNameFormat("metalake-cleaner-%d") | ||
| .build()))) | ||
| .build(); | ||
|
|
||
| /** | ||
| * Constructs a MetalakeManager instance. | ||
| * | ||
|
|
@@ -73,6 +95,13 @@ public class MetalakeManager implements MetalakeDispatcher { | |
| public MetalakeManager(EntityStore store, IdGenerator idGenerator) { | ||
| this.store = store; | ||
| this.idGenerator = idGenerator; | ||
|
|
||
| // pre-load all metalakes and put them into cache, this is useful when user load schema/table | ||
| // directly without list/get metalake first. | ||
| BaseMetalake[] metalakes = listMetalakes(); | ||
| for (BaseMetalake metalake : metalakes) { | ||
| METALAKE_CACHE.put(metalake.nameIdentifier(), metalake); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -103,10 +132,12 @@ public static void checkMetalake(NameIdentifier ident, EntityStore store) | |
| public static boolean metalakeInUse(EntityStore store, NameIdentifier ident) | ||
| throws NoSuchMetalakeException { | ||
| try { | ||
| BaseMetalake metalake = METALAKE_CACHE.getIfPresent(ident); | ||
| if (metalake == null) { | ||
| metalake = store.get(ident, EntityType.METALAKE, BaseMetalake.class); | ||
| } | ||
|
Comment on lines
+135
to
+138
There was a problem hiding this comment. why not cache the result after getting it from the store? There was a problem hiding this comment. Please see #6569 (comment) There was a problem hiding this comment. after the user alters the metalake, then load schema/table directly without list/get the metalake, the cache never be hitting, right? There was a problem hiding this comment. Yeah. alter a metalake is not operated frequently, so I guess it's acceptable |
||
| return (boolean) | ||
| metalake.propertiesMetadata().getOrDefault(metalake.properties(), PROPERTY_IN_USE); | ||
| } catch (NoSuchEntityException e) { | ||
| LOG.warn("Metalake {} does not exist", ident, e); | ||
| throw new NoSuchMetalakeException(METALAKE_DOES_NOT_EXIST_MSG, ident); | ||
|
|
@@ -149,20 +180,24 @@ public BaseMetalake[] listMetalakes() { | |
| */ | ||
| @Override | ||
| public BaseMetalake loadMetalake(NameIdentifier ident) throws NoSuchMetalakeException { | ||
| return METALAKE_CACHE.get( | ||
| ident, | ||
| k -> { | ||
| try { | ||
| BaseMetalake baseMetalake = | ||
| TreeLockUtils.doWithTreeLock( | ||
| ident, | ||
| LockType.READ, | ||
| () -> store.get(ident, EntityType.METALAKE, BaseMetalake.class)); | ||
| return newMetalakeWithResolvedProperties(baseMetalake); | ||
| } catch (NoSuchEntityException e) { | ||
| LOG.warn("Metalake {} does not exist", ident, e); | ||
| throw new NoSuchMetalakeException(METALAKE_DOES_NOT_EXIST_MSG, ident); | ||
| } catch (IOException ioe) { | ||
| LOG.error("Loading Metalake {} failed due to storage issues", ident, ioe); | ||
| throw new RuntimeException(ioe); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private BaseMetalake newMetalakeWithResolvedProperties(BaseMetalake metalakeEntity) { | ||
|
|
@@ -222,6 +257,7 @@ public BaseMetalake createMetalake( | |
| () -> { | ||
| try { | ||
| store.put(metalake, false /* overwritten */); | ||
| METALAKE_CACHE.put(ident, newMetalakeWithResolvedProperties(metalake)); | ||
| return metalake; | ||
| } catch (EntityAlreadyExistsException | AlreadyExistsException e) { | ||
| LOG.warn("Metalake {} already exists", ident, e); | ||
|
|
@@ -253,22 +289,24 @@ public BaseMetalake alterMetalake(NameIdentifier ident, MetalakeChange... change | |
| throw new MetalakeNotInUseException( | ||
| "Metalake %s is not in use, please enable it first", ident); | ||
| } | ||
| METALAKE_CACHE.invalidate(ident); | ||
|
There was a problem hiding this comment. Can the result be cached after updating? There was a problem hiding this comment. Put it back to cache seems to be an optional if this optional is not frequently called. Anyway, this is an improvement, let me check if we can add it back. |
||
| BaseMetalake baseMetalake = | ||
| store.update( | ||
| ident, | ||
| BaseMetalake.class, | ||
| EntityType.METALAKE, | ||
| metalake -> { | ||
| BaseMetalake.Builder builder = newMetalakeBuilder(metalake); | ||
| Map<String, String> newProps = | ||
| metalake.properties() == null | ||
| ? Maps.newHashMap() | ||
| : Maps.newHashMap(metalake.properties()); | ||
| builder = updateEntity(builder, newProps, changes); | ||
|
|
||
| return builder.build(); | ||
| }); | ||
| METALAKE_CACHE.put(ident, newMetalakeWithResolvedProperties(baseMetalake)); | ||
| return baseMetalake; | ||
| } catch (NoSuchEntityException ne) { | ||
| LOG.warn("Metalake {} does not exist", ident, ne); | ||
| throw new NoSuchMetalakeException(METALAKE_DOES_NOT_EXIST_MSG, ident); | ||
|
|
@@ -305,6 +343,8 @@ public boolean dropMetalake(NameIdentifier ident, boolean force) | |
| "Metalake %s is in use, please disable it first or use force option", ident); | ||
| } | ||
|
|
||
| METALAKE_CACHE.invalidate(ident); | ||
|
|
||
| List<CatalogEntity> catalogEntities = | ||
| store.list(Namespace.of(ident.name()), CatalogEntity.class, EntityType.CATALOG); | ||
| if (!catalogEntities.isEmpty() && !force) { | ||
|
|
@@ -331,22 +371,25 @@ public void enableMetalake(NameIdentifier ident) throws NoSuchMetalakeException | |
| try { | ||
| boolean inUse = metalakeInUse(store, ident); | ||
| if (!inUse) { | ||
| METALAKE_CACHE.invalidate(ident); | ||
| BaseMetalake baseMetalake = | ||
| store.update( | ||
| ident, | ||
| BaseMetalake.class, | ||
| EntityType.METALAKE, | ||
| metalake -> { | ||
| BaseMetalake.Builder builder = newMetalakeBuilder(metalake); | ||
|
|
||
| Map<String, String> newProps = | ||
| metalake.properties() == null | ||
| ? Maps.newHashMap() | ||
| : Maps.newHashMap(metalake.properties()); | ||
| newProps.put(PROPERTY_IN_USE, "true"); | ||
| builder.withProperties(newProps); | ||
|
|
||
| return builder.build(); | ||
| }); | ||
| METALAKE_CACHE.put(ident, newMetalakeWithResolvedProperties(baseMetalake)); | ||
| } | ||
|
|
||
| return null; | ||
|
|
@@ -365,6 +408,7 @@ public void disableMetalake(NameIdentifier ident) throws NoSuchMetalakeException | |
| try { | ||
| boolean inUse = metalakeInUse(store, ident); | ||
| if (inUse) { | ||
| METALAKE_CACHE.invalidate(ident); | ||
| store.update( | ||
| ident, | ||
| BaseMetalake.class, | ||
|
|
||
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
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why do we make it static? I assume there will be only one
CatalogManager, so there should be only onecatalogCache, right?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The method check
catalogInUseandmetalakeInUseare all static. If we want to use cache for them, we need to change it tostaticYes, there will be only one cache and all catalogs shares the same instance, It's not a big problem I think.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
should named
CATALOG_CACHE?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
should it be
final?