This is an automated email from the ASF dual-hosted git repository.
yuqi1129 pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new bf5f30f411 [Cherry-pick to branch-1.3] [#12416] fix(core): Cascade
cache invalidation to hierarchical schema descendants (#12417) (#12435)
bf5f30f411 is described below
commit bf5f30f411b52dffb2159c2f50fa05f382170d6c
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Aug 12 19:57:08 2026 +0800
[Cherry-pick to branch-1.3] [#12416] fix(core): Cascade cache invalidation
to hierarchical schema descendants (#12417) (#12435)
**Cherry-pick Information:**
- Original commit: bb93bf923433096735abdca9f248a4b3adfdfbcf
- Target branch: `branch-1.3`
- Status: ⚠️ **Has conflicts - manual resolution required**
**Do not merge** until conflict markers are resolved and the
`cherry-pick-conflict` label is removed.
Please review and resolve the conflicts before merging.
---------
Signed-off-by: yuqi <[email protected]>
Co-authored-by: Qi Yu <[email protected]>
---
.../gravitino/cache/CaffeineEntityCache.java | 45 +++-
.../storage/relational/RelationalEntityStore.java | 18 +-
.../cache/TestCaffeineEntityCacheInvalidation.java | 182 ++++++++++++++++
.../relational/TestRelationalEntityStore.java | 64 ++----
...TestRelationalEntityStoreHierarchicalCache.java | 229 +++++++++++++++++++++
5 files changed, 481 insertions(+), 57 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
b/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
index fd785edf5a..ee186f4cc5 100644
--- a/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
+++ b/core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java
@@ -54,6 +54,7 @@ import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.meta.GenericEntity;
import org.apache.gravitino.meta.ModelVersionEntity;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -80,6 +81,12 @@ public class CaffeineEntityCache extends BaseEntityCache {
private static final Logger LOG =
LoggerFactory.getLogger(CaffeineEntityCache.class.getName());
+ /**
+ * Separates {@link NameIdentifier} levels in a cache key. See {@link
+ * #invalidationPrefixes(EntityCacheKey)} for why it is not the only child
boundary.
+ */
+ private static final String NAME_LEVEL_BOUNDARY = ".";
+
/** Segmented locking for better concurrency */
private final SegmentedLock segmentedLock;
@@ -486,17 +493,18 @@ public class CaffeineEntityCache extends BaseEntityCache {
cacheData.invalidate(currentKeyToRemove);
cacheIndex.remove(currentKeyToRemove.toString());
- // Remove related entity keys
- List<EntityCacheKey> relatedEntityKeysToRemove =
- Lists.newArrayList(
-
cacheIndex.getValuesForKeysStartingWith(currentKeyToRemove.identifier().toString()));
+ // Remove the current entity's relation entries and its descendants.
+ Set<EntityCacheKey> relatedEntityKeysToRemove = Sets.newHashSet();
+ for (String prefix : invalidationPrefixes(currentKeyToRemove)) {
+
cacheIndex.getValuesForKeysStartingWith(prefix).forEach(relatedEntityKeysToRemove::add);
+ }
queue.addAll(relatedEntityKeysToRemove);
// Look up from reverse index to go to next depth
- List<List<EntityCacheKey>> reverseKeysToRemove =
- Lists.newArrayList(
- reverseIndex.getValuesForKeysStartingWith(
- currentKeyToRemove.identifier().toString()));
+ List<List<EntityCacheKey>> reverseKeysToRemove = Lists.newArrayList();
+ for (String prefix : invalidationPrefixes(currentKeyToRemove)) {
+
reverseIndex.getValuesForKeysStartingWith(prefix).forEach(reverseKeysToRemove::add);
+ }
reverseKeysToRemove.forEach(
key -> {
@@ -520,6 +528,27 @@ public class CaffeineEntityCache extends BaseEntityCache {
return true;
}
+ /**
+ * Returns exact prefixes for the entity, its ordinary descendants, and
nested schemas.
+ *
+ * <p>The entity-type prefix includes all relation entries for the same
entity without matching an
+ * entity of another type. The name-level and schema-level prefixes use
explicit child boundaries,
+ * so identifiers that merely share a prefix remain cached.
+ *
+ * @param key the entity cache key
+ * @return prefixes identifying the entity and its descendants
+ */
+ private Set<String> invalidationPrefixes(EntityCacheKey key) {
+ String identifier = key.identifier().toString();
+ Set<String> prefixes = Sets.newLinkedHashSet();
+ prefixes.add(EntityCacheKey.of(key.identifier(),
key.entityType()).toString());
+ prefixes.add(identifier + NAME_LEVEL_BOUNDARY);
+ if (key.entityType() == Entity.EntityType.SCHEMA) {
+ prefixes.add(identifier + HierarchicalSchemaUtil.schemaSeparator());
+ }
+ return prefixes;
+ }
+
/** Starts the cache stats monitor. */
private void startCacheStatsMonitor() {
scheduler.scheduleAtFixedRate(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
index 6f21491173..c710b778b0 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java
@@ -395,13 +395,14 @@ public class RelationalEntityStore
backend.updateEntityRelations(
relType, srcEntityIdent, srcEntityType, destEntitiesToAdd,
destEntitiesToRemove);
+ Entity.EntityType destEntityType = relationDestinationEntityType(relType);
cache.invalidate(srcEntityIdent, srcEntityType, relType);
for (NameIdentifier destToAdd : destEntitiesToAdd) {
- cache.invalidate(destToAdd, srcEntityType, relType);
+ cache.invalidate(destToAdd, destEntityType, relType);
}
for (NameIdentifier destToRemove : destEntitiesToRemove) {
- cache.invalidate(destToRemove, srcEntityType, relType);
+ cache.invalidate(destToRemove, destEntityType, relType);
}
return result;
@@ -420,6 +421,19 @@ public class RelationalEntityStore
backend.batchPut(entities, overwritten);
}
+ private static Entity.EntityType relationDestinationEntityType(
+ SupportsRelationOperations.Type relType) {
+ switch (relType) {
+ case POLICY_METADATA_OBJECT_REL:
+ return Entity.EntityType.POLICY;
+ case TAG_METADATA_OBJECT_REL:
+ return Entity.EntityType.TAG;
+ default:
+ throw new IllegalArgumentException(
+ String.format("Doesn't support the relation type %s", relType));
+ }
+ }
+
private <E extends Entity & HasIdentifier>
Optional<List<RelationalEntity<?>>> getCachedRelations(
SupportsRelationOperations.Type relType,
NameIdentifier nameIdentifier,
diff --git
a/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheInvalidation.java
b/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheInvalidation.java
index 58c0450852..dcb66e5c7b 100644
---
a/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheInvalidation.java
+++
b/core/src/test/java/org/apache/gravitino/cache/TestCaffeineEntityCacheInvalidation.java
@@ -22,18 +22,28 @@ import com.google.common.collect.Lists;
import java.time.Instant;
import java.util.List;
import java.util.Optional;
+import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.Namespace;
import org.apache.gravitino.SupportsRelationOperations;
import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.authorization.Privileges;
import org.apache.gravitino.authorization.SecurableObject;
import org.apache.gravitino.authorization.SecurableObjects;
import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.CatalogEntity;
import org.apache.gravitino.meta.RoleEntity;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.meta.TopicEntity;
import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
import org.apache.gravitino.utils.NameIdentifierUtil;
+import org.apache.gravitino.utils.TestUtil;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -48,6 +58,8 @@ import org.junit.jupiter.api.Test;
*/
public class TestCaffeineEntityCacheInvalidation {
+ private static final Namespace CATALOG_NS = Namespace.of("metalake",
"catalog1");
+
private CaffeineEntityCache cache;
private AuditInfo auditInfo;
@@ -83,6 +95,18 @@ public class TestCaffeineEntityCacheInvalidation {
.build();
}
+ /**
+ * Joins nested schema levels the way they reach the cache. The cache sits
above the storage
+ * layer, so nested schema names still carry the configured external
separator.
+ */
+ private static String hierarchicalName(String... levels) {
+ return String.join(HierarchicalSchemaUtil.schemaSeparator(), levels);
+ }
+
+ private static Namespace schemaNamespace(String schemaName) {
+ return Namespace.of(CATALOG_NS.level(0), CATALOG_NS.level(1), schemaName);
+ }
+
/**
* Core scenario reproducing GitHub issue #11297: after caching the
METADATA_OBJECT_ROLE_REL
* relation for a schema, invalidating the role entity must also remove the
stale relation cache.
@@ -373,4 +397,162 @@ public class TestCaffeineEntityCacheInvalidation {
Assertions.assertNull(
roleReverseKeys, "Reverse index for role should be empty after
invalidation");
}
+
+ @Test
+ void testInvalidateHierarchicalSchemaCascadesToNestedSchemas() {
+ // A HierarchicalSchema nests inside a single name level, joined by the
schema separator, so
+ // "raw:events:2024" is a child of "raw:events" without adding a
NameIdentifier level.
+ String parentName = hierarchicalName("raw", "events");
+ String childName = hierarchicalName("raw", "events", "2024");
+
+ SchemaEntity parent = TestUtil.getTestSchemaEntity(2L, parentName,
CATALOG_NS, "cmt");
+ SchemaEntity child = TestUtil.getTestSchemaEntity(3L, childName,
CATALOG_NS, "cmt");
+ TableEntity tableInParent =
+ TestUtil.getTestTableEntity(4L, "t_parent",
schemaNamespace(parentName));
+ TableEntity tableInChild =
+ TestUtil.getTestTableEntity(5L, "t_child", schemaNamespace(childName));
+
+ cache.put(parent);
+ cache.put(child);
+ cache.put(tableInParent);
+ cache.put(tableInChild);
+ Assertions.assertEquals(4, cache.size());
+
+ cache.invalidate(parent.nameIdentifier(), Entity.EntityType.SCHEMA);
+
+ Assertions.assertFalse(cache.contains(parent.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(cache.contains(tableInParent.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertFalse(cache.contains(child.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(cache.contains(tableInChild.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertEquals(0, cache.size());
+ }
+
+ @Test
+ void testInvalidateHierarchicalSchemaCascadesToAnyDepth() {
+ String level1 = hierarchicalName("raw");
+ String level2 = hierarchicalName("raw", "events");
+ String level3 = hierarchicalName("raw", "events", "2024");
+ String level4 = hierarchicalName("raw", "events", "2024", "q1");
+
+ cache.put(TestUtil.getTestSchemaEntity(2L, level1, CATALOG_NS, "cmt"));
+ cache.put(TestUtil.getTestSchemaEntity(3L, level2, CATALOG_NS, "cmt"));
+ cache.put(TestUtil.getTestSchemaEntity(4L, level3, CATALOG_NS, "cmt"));
+ SchemaEntity deepest = TestUtil.getTestSchemaEntity(5L, level4,
CATALOG_NS, "cmt");
+ cache.put(deepest);
+ TableEntity deepestTable = TestUtil.getTestTableEntity(6L, "t_deep",
schemaNamespace(level4));
+ cache.put(deepestTable);
+ Assertions.assertEquals(5, cache.size());
+
+ cache.invalidate(
+ TestUtil.getTestSchemaEntity(2L, level1, CATALOG_NS,
"cmt").nameIdentifier(),
+ Entity.EntityType.SCHEMA);
+
+ Assertions.assertFalse(cache.contains(deepest.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(cache.contains(deepestTable.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertEquals(0, cache.size());
+ }
+
+ @Test
+ void testInvalidateHierarchicalSchemaDoesNotTouchSiblings() {
+ // Guards against over-matching: "raw:events2" is a sibling of
"raw:events", not a descendant,
+ // exactly like the catalog1 / catalog10 case the "." boundary already
protects against.
+ String target = hierarchicalName("raw", "events");
+ String sibling = hierarchicalName("raw", "events2");
+ String siblingOfParent = hierarchicalName("raw2", "events");
+
+ SchemaEntity targetSchema = TestUtil.getTestSchemaEntity(2L, target,
CATALOG_NS, "cmt");
+ SchemaEntity siblingSchema = TestUtil.getTestSchemaEntity(3L, sibling,
CATALOG_NS, "cmt");
+ SchemaEntity otherBranch = TestUtil.getTestSchemaEntity(4L,
siblingOfParent, CATALOG_NS, "cmt");
+ TableEntity siblingTable =
+ TestUtil.getTestTableEntity(5L, "t_sibling", schemaNamespace(sibling));
+
+ cache.put(targetSchema);
+ cache.put(siblingSchema);
+ cache.put(otherBranch);
+ cache.put(siblingTable);
+
+ cache.invalidate(targetSchema.nameIdentifier(), Entity.EntityType.SCHEMA);
+
+ Assertions.assertFalse(cache.contains(targetSchema.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(cache.contains(siblingSchema.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(cache.contains(otherBranch.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(cache.contains(siblingTable.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertEquals(3, cache.size());
+ }
+
+ @Test
+ void testInvalidateCatalogCascadesToHierarchicalSchemas() {
+ CatalogEntity catalog =
+ TestUtil.getTestCatalogEntity(1L, "catalog1",
Namespace.of("metalake"), "hive", "cmt");
+ String nested = hierarchicalName("raw", "events", "2024");
+ SchemaEntity schema = TestUtil.getTestSchemaEntity(2L, nested, CATALOG_NS,
"cmt");
+ TableEntity table = TestUtil.getTestTableEntity(3L, "t1",
schemaNamespace(nested));
+
+ cache.put(catalog);
+ cache.put(schema);
+ cache.put(table);
+ Assertions.assertEquals(3, cache.size());
+
+ cache.invalidate(catalog.nameIdentifier(), Entity.EntityType.CATALOG);
+
+ Assertions.assertFalse(cache.contains(schema.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(cache.contains(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertEquals(0, cache.size());
+ }
+
+ @Test
+ void testInvalidateHierarchicalSchemaCascadesWithNonDefaultSeparator()
throws Exception {
+ Config separatorConfig = new Config(false) {};
+ separatorConfig.set(Configs.SCHEMA_SEPARATOR, "|");
+ Object previousConfig = FieldUtils.readField(GravitinoEnv.getInstance(),
"config", true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
separatorConfig, true);
+
+ try {
+ Assertions.assertEquals("|", HierarchicalSchemaUtil.schemaSeparator());
+
+ String parentName = hierarchicalName("raw", "events");
+ String childName = hierarchicalName("raw", "events", "2024");
+ String siblingName = hierarchicalName("raw", "events2");
+
+ SchemaEntity parent = TestUtil.getTestSchemaEntity(2L, parentName,
CATALOG_NS, "cmt");
+ SchemaEntity child = TestUtil.getTestSchemaEntity(3L, childName,
CATALOG_NS, "cmt");
+ SchemaEntity sibling = TestUtil.getTestSchemaEntity(4L, siblingName,
CATALOG_NS, "cmt");
+ TableEntity tableInChild =
+ TestUtil.getTestTableEntity(5L, "t_child",
schemaNamespace(childName));
+
+ cache.put(parent);
+ cache.put(child);
+ cache.put(sibling);
+ cache.put(tableInChild);
+
+ cache.invalidate(parent.nameIdentifier(), Entity.EntityType.SCHEMA);
+
+ Assertions.assertFalse(cache.contains(parent.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(cache.contains(child.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(
+ cache.contains(tableInChild.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertTrue(cache.contains(sibling.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertEquals(1, cache.size());
+ } finally {
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
previousConfig, true);
+ }
+ }
+
+ @Test
+ void testInvalidateLeafDoesNotEvictSameNameEntityOfAnotherType() {
+ // A cache key is "<identifier>:<type>", and ":" is also the default
schema separator. Only a
+ // schema can nest, so the schema-separator scan must not run for other
types, otherwise
+ // invalidating a table would also drop the topic of the same name.
+ Namespace schemaNs = schemaNamespace("schema1");
+ TableEntity table = TestUtil.getTestTableEntity(2L, "shared_name",
schemaNs);
+ TopicEntity topic = TestUtil.getTestTopicEntity(3L, "shared_name",
schemaNs, "cmt");
+
+ cache.put(table);
+ cache.put(topic);
+
+ cache.invalidate(table.nameIdentifier(), Entity.EntityType.TABLE);
+
+ Assertions.assertFalse(cache.contains(table.nameIdentifier(),
Entity.EntityType.TABLE));
+ Assertions.assertTrue(cache.contains(topic.nameIdentifier(),
Entity.EntityType.TOPIC));
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
index 042c29681b..c260b618db 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStore.java
@@ -36,6 +36,8 @@ import org.apache.gravitino.exceptions.NoSuchEntityException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.mockito.InOrder;
import org.mockito.Mockito;
@@ -157,13 +159,15 @@ public class TestRelationalEntityStore {
dst, Entity.EntityType.TAG,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
}
- @Test
- void testUpdateEntityRelationsInvalidatesCacheAfterBackendUpdate()
+ @ParameterizedTest
+ @CsvSource({"TAG_METADATA_OBJECT_REL, TAG", "POLICY_METADATA_OBJECT_REL,
POLICY"})
+ void testUpdateEntityRelationsInvalidatesCacheAfterBackendUpdate(
+ SupportsRelationOperations.Type relationType, Entity.EntityType
destinationType)
throws IOException, NoSuchEntityException, EntityAlreadyExistsException,
IllegalAccessException {
NameIdentifier src = NameIdentifier.of("metalake", "catalog", "schema",
"table1");
- NameIdentifier destToAdd = NameIdentifier.of("metalake", "tag1");
- NameIdentifier destToRemove = NameIdentifier.of("metalake", "tag2");
+ NameIdentifier destToAdd = NameIdentifier.of("metalake", "destination1");
+ NameIdentifier destToRemove = NameIdentifier.of("metalake",
"destination2");
NameIdentifier[] destEntitiesToAdd = new NameIdentifier[] {destToAdd};
NameIdentifier[] destEntitiesToRemove = new NameIdentifier[]
{destToRemove};
NoOpsCache cache = (NoOpsCache) FieldUtils.readField(store, "cache", true);
@@ -171,61 +175,27 @@ public class TestRelationalEntityStore {
Mockito.doAnswer(
invocation -> {
Mockito.verify(cache, Mockito.never())
- .invalidate(
- src,
- Entity.EntityType.TABLE,
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ .invalidate(src, Entity.EntityType.TABLE, relationType);
Mockito.verify(cache, Mockito.never())
- .invalidate(
- destToAdd,
- Entity.EntityType.TABLE,
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ .invalidate(destToAdd, destinationType, relationType);
Mockito.verify(cache, Mockito.never())
- .invalidate(
- destToRemove,
- Entity.EntityType.TABLE,
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ .invalidate(destToRemove, destinationType, relationType);
return List.of();
})
.when(backend)
.updateEntityRelations(
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
- src,
- Entity.EntityType.TABLE,
- destEntitiesToAdd,
- destEntitiesToRemove);
+ relationType, src, Entity.EntityType.TABLE, destEntitiesToAdd,
destEntitiesToRemove);
store.updateEntityRelations(
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
- src,
- Entity.EntityType.TABLE,
- destEntitiesToAdd,
- destEntitiesToRemove);
+ relationType, src, Entity.EntityType.TABLE, destEntitiesToAdd,
destEntitiesToRemove);
InOrder inOrder = Mockito.inOrder(backend, cache);
inOrder
.verify(backend)
.updateEntityRelations(
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL,
- src,
- Entity.EntityType.TABLE,
- destEntitiesToAdd,
- destEntitiesToRemove);
- inOrder
- .verify(cache)
- .invalidate(
- src, Entity.EntityType.TABLE,
SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
- inOrder
- .verify(cache)
- .invalidate(
- destToAdd,
- Entity.EntityType.TABLE,
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
- inOrder
- .verify(cache)
- .invalidate(
- destToRemove,
- Entity.EntityType.TABLE,
- SupportsRelationOperations.Type.TAG_METADATA_OBJECT_REL);
+ relationType, src, Entity.EntityType.TABLE, destEntitiesToAdd,
destEntitiesToRemove);
+ inOrder.verify(cache).invalidate(src, Entity.EntityType.TABLE,
relationType);
+ inOrder.verify(cache).invalidate(destToAdd, destinationType, relationType);
+ inOrder.verify(cache).invalidate(destToRemove, destinationType,
relationType);
}
}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
new file mode 100644
index 0000000000..77a8fd419f
--- /dev/null
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/TestRelationalEntityStoreHierarchicalCache.java
@@ -0,0 +1,229 @@
+/*
+ * 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;
+
+import java.io.File;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.Config;
+import org.apache.gravitino.Configs;
+import org.apache.gravitino.Entity;
+import org.apache.gravitino.GravitinoEnv;
+import org.apache.gravitino.Namespace;
+import org.apache.gravitino.cache.CaffeineEntityCache;
+import org.apache.gravitino.meta.AuditInfo;
+import org.apache.gravitino.meta.BaseMetalake;
+import org.apache.gravitino.meta.CatalogEntity;
+import org.apache.gravitino.meta.SchemaEntity;
+import org.apache.gravitino.meta.SchemaVersion;
+import org.apache.gravitino.meta.TableEntity;
+import org.apache.gravitino.storage.RandomIdGenerator;
+import org.apache.gravitino.utils.HierarchicalSchemaUtil;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
+
+/**
+ * Verifies that dropping a hierarchical schema through a real {@link
RelationalEntityStore} also
+ * drops its nested descendants from the entity cache, for the default and a
non-default schema
+ * separator.
+ */
+public class TestRelationalEntityStoreHierarchicalCache {
+
+ private static final String METALAKE = "metalake_hs";
+ private static final String CATALOG = "catalog_hs";
+ private static final AuditInfo AUDIT_INFO =
+
AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+ private RelationalEntityStore store;
+ private String dbPath;
+ private Object previousConfig;
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (store != null) {
+ store.close();
+ store = null;
+ }
+ if (dbPath != null) {
+ FileUtils.deleteQuietly(new File(dbPath));
+ dbPath = null;
+ }
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config",
previousConfig, true);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {":", "|"})
+ void testDropHierarchicalSchemaEvictsNestedDescendantsFromCache(String
separator)
+ throws Exception {
+ initStore(separator);
+
+ String parentName = String.join(separator, "raw", "events");
+ String childName = String.join(separator, "raw", "events", "2024");
+ String siblingName = String.join(separator, "raw", "events2");
+
+ store.put(metalake(), false);
+ store.put(catalog(), false);
+ SchemaEntity parent = schema(parentName);
+ SchemaEntity child = schema(childName);
+ SchemaEntity sibling = schema(siblingName);
+ store.put(parent, false);
+ store.put(child, false);
+ store.put(sibling, false);
+ TableEntity tableInChild = table("t_child", childName);
+ store.put(tableInChild, false);
+
+ // Read everything back so the cache is populated with the names the store
actually returns.
+ store.get(parent.nameIdentifier(), Entity.EntityType.SCHEMA,
SchemaEntity.class);
+ store.get(child.nameIdentifier(), Entity.EntityType.SCHEMA,
SchemaEntity.class);
+ store.get(sibling.nameIdentifier(), Entity.EntityType.SCHEMA,
SchemaEntity.class);
+ store.get(tableInChild.nameIdentifier(), Entity.EntityType.TABLE,
TableEntity.class);
+ Assertions.assertTrue(
+ store.getCache().contains(child.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertTrue(
+ store.getCache().contains(tableInChild.nameIdentifier(),
Entity.EntityType.TABLE));
+
+ // The cache is keyed by the identifier that reaches the store, which
still carries the
+ // configured external separator; the physical separator only exists in
the backend rows.
+ Set<String> cacheKeys =
+ ((CaffeineEntityCache) store.getCache())
+ .getCacheData().asMap().keySet().stream()
+ .map(Object::toString)
+ .collect(Collectors.toSet());
+ Assertions.assertTrue(
+ cacheKeys.stream().anyMatch(key -> key.contains(childName)),
+ "nested schema must be cached under its logical name, keys: " +
cacheKeys);
+ Assertions.assertTrue(
+ cacheKeys.stream()
+ .noneMatch(key ->
key.contains(HierarchicalSchemaUtil.physicalSeparator())),
+ "no cache key may carry the physical separator, keys: " + cacheKeys);
+
+ store.delete(parent.nameIdentifier(), Entity.EntityType.SCHEMA, true);
+
+ Assertions.assertFalse(
+ store.getCache().contains(parent.nameIdentifier(),
Entity.EntityType.SCHEMA));
+ Assertions.assertFalse(
+ store.getCache().contains(child.nameIdentifier(),
Entity.EntityType.SCHEMA),
+ "nested schema must not survive the drop of its parent");
+ Assertions.assertFalse(
+ store.getCache().contains(tableInChild.nameIdentifier(),
Entity.EntityType.TABLE),
+ "table of a nested schema must not survive the drop of the parent
schema");
+ Assertions.assertTrue(
+ store.getCache().contains(sibling.nameIdentifier(),
Entity.EntityType.SCHEMA),
+ "a sibling sharing a name prefix must not be invalidated");
+ }
+
+ private void initStore(String separator) throws Exception {
+ dbPath = "/tmp/gravitino_hs_cache_test_" +
UUID.randomUUID().toString().replace("-", "");
+ File dir = new File(dbPath);
+ if (!dir.exists() && !dir.mkdirs()) {
+ throw new IOException("Failed to create test directory " + dbPath);
+ }
+
+ Config config = Mockito.mock(Config.class);
+
Mockito.when(config.get(Configs.ENTITY_STORE)).thenReturn(Configs.RELATIONAL_ENTITY_STORE);
+ Mockito.when(config.get(Configs.ENTITY_RELATIONAL_STORE))
+ .thenReturn(Configs.DEFAULT_ENTITY_RELATIONAL_STORE);
+ Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_URL))
+
.thenReturn(String.format("jdbc:h2:file:%s;DB_CLOSE_DELAY=-1;MODE=MYSQL",
dbPath));
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_USER)).thenReturn("root");
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_PASSWORD)).thenReturn("123456");
+ Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_DRIVER))
+ .thenReturn("org.h2.Driver");
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS))
+ .thenReturn(Configs.DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_CONNECTIONS);
+
Mockito.when(config.get(Configs.ENTITY_RELATIONAL_JDBC_BACKEND_WAIT_MILLISECONDS))
+
.thenReturn(Configs.DEFAULT_RELATIONAL_JDBC_BACKEND_MAX_WAIT_MILLISECONDS);
+ Mockito.when(config.get(Configs.STORE_DELETE_AFTER_TIME)).thenReturn(20 *
60 * 1000L);
+ Mockito.when(config.get(Configs.VERSION_RETENTION_COUNT)).thenReturn(1L);
+
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_POLL_INTERVAL_SECS)).thenReturn(3L);
+
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_RETENTION_SECS)).thenReturn(24
* 60 * 60L);
+
Mockito.when(config.get(Configs.ENTITY_CHANGE_LOG_CLEANUP_INTERVAL_SECS)).thenReturn(60
* 60L);
+ Mockito.when(config.get(Configs.CACHE_ENABLED)).thenReturn(true);
+
Mockito.when(config.get(Configs.CACHE_IMPLEMENTATION)).thenReturn("caffeine");
+ Mockito.when(config.get(Configs.CACHE_MAX_ENTRIES)).thenReturn(10_000);
+
Mockito.when(config.get(Configs.CACHE_EXPIRATION_TIME)).thenReturn(3_600_000L);
+ Mockito.when(config.get(Configs.CACHE_WEIGHER_ENABLED)).thenReturn(true);
+ Mockito.when(config.get(Configs.CACHE_STATS_ENABLED)).thenReturn(false);
+ Mockito.when(config.get(Configs.CACHE_LOCK_SEGMENTS)).thenReturn(16);
+ Mockito.when(config.get(Configs.SCHEMA_SEPARATOR)).thenReturn(separator);
+
+ previousConfig = FieldUtils.readField(GravitinoEnv.getInstance(),
"config", true);
+ FieldUtils.writeField(GravitinoEnv.getInstance(), "config", config, true);
+ FieldUtils.writeField(
+ GravitinoEnv.getInstance(), "idGenerator", RandomIdGenerator.INSTANCE,
true);
+ Assertions.assertEquals(separator,
HierarchicalSchemaUtil.schemaSeparator());
+
+ store = new RelationalEntityStore();
+ store.initialize(config);
+ }
+
+ private static BaseMetalake metalake() {
+ return BaseMetalake.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(METALAKE)
+ .withAuditInfo(AUDIT_INFO)
+ .withComment("")
+ .withProperties(null)
+ .withVersion(SchemaVersion.V_0_1)
+ .build();
+ }
+
+ private static CatalogEntity catalog() {
+ return CatalogEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(CATALOG)
+ .withNamespace(Namespace.of(METALAKE))
+ .withType(Catalog.Type.RELATIONAL)
+ .withProvider("test")
+ .withComment("")
+ .withProperties(null)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private static SchemaEntity schema(String name) {
+ return SchemaEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(Namespace.of(METALAKE, CATALOG))
+ .withComment("")
+ .withProperties(null)
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+
+ private static TableEntity table(String name, String schemaName) {
+ return TableEntity.builder()
+ .withId(RandomIdGenerator.INSTANCE.nextId())
+ .withName(name)
+ .withNamespace(Namespace.of(METALAKE, CATALOG, schemaName))
+ .withAuditInfo(AUDIT_INFO)
+ .build();
+ }
+}
