[#12151] improvement(core): complete cached entity change-log coverage (#12374)

Depends on #12111.

### What changes were proposed in this pull request?

Centralize cache invalidation change-log emission at the `JDBCBackend`
entity-store mutation boundary.

- Emit `ALTER` for cacheable entity overwrite and update operations.
- Emit `DROP` after a successful delete.
- Keep only the old identifier for rename invalidation.
- Do not emit an event for create because the entity cache has no
negative entries and list operations bypass the cache.
- Emit one root `DROP` for cascading deletion and let prefix
invalidation clear cached descendants.
- Commit the metadata mutation and its change-log row in the same
transaction.
- Remove duplicated change-log emission from type-specific MetaServices.
- Update the multi-node cache design and mutation coverage tests.

The covered cacheable types are metalake, catalog, schema, table, topic,
view, fileset, tag, policy, and job.

### Why are the changes needed?

Change-log emission was distributed across selected MetaService paths.
This missed ordinary non-rename updates and mutation paths such as
overwrite, status changes, enable/disable, import, and repair.

Centralizing emission ensures that every invalidating mutation of a
cacheable entity follows the same transaction and coverage policy.

Fix: #12151

### Does this PR introduce _any_ user-facing change?

No. This PR does not change public APIs or configuration properties.

### How was this patch tested?

- `./gradlew :core:spotlessApply -PskipDockerTests`
- `./gradlew :core:test --tests
"org.apache.gravitino.storage.relational.service.TestEntityChangeLogService"
-PskipITs -PskipDockerTests`
- `./gradlew :core:test --tests
"org.apache.gravitino.storage.relational.service.TestTableMetaService"
-PskipITs -PskipDockerTests`
- `./gradlew :core:test --tests
"org.apache.gravitino.storage.relational.TestEntityCacheCrossNodeInvalidation"
-PskipITs -PskipDockerTests`
- `./gradlew :core:test -PskipITs -PskipDockerTests`
- Added `TestEntityStorageChangeLog`, which runs the mutation lifecycle
and cascade-drop assertions against H2, MySQL, and PostgreSQL with cache
enabled and disabled.

Author: yuqi1129

core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java

@@ -45,35 +45,25 @@ public abstract class BaseEntityCache implements EntityCache {
    * SupportsEntityStoreCache#put(Entity)} for the contract. New entity types are excluded by
    * default until their invalidation behavior has been validated.
    *
-   * <p>{@code USER}, {@code GROUP} and {@code ROLE} are materialized with relation-derived data
-   * joined in at load time: a role carries its securable objects, and a user/group carries its role
-   * names. A mutation on the entity itself invalidates its own key through the write path, but this
-   * embedded data also goes stale through a mutation on a different entity. For example, deleting
-   * or renaming a securable object changes a role's materialized form, and deleting or renaming a
-   * role changes a user's/group's role names. Such a mutation touches neither this entity's own key
-   * nor any hierarchy ancestor of it, so neither the write-path invalidation nor a prefix cascade
-   * over the entity hierarchy would evict it. Caching them would therefore serve stale
-   * authorization data.
+   * <p>Only self-contained entities are cacheable: a stale copy of one is at worst cosmetically old
+   * (an old comment, property, or job status), never a wrong pointer, and each can be invalidated
+   * with a single one-to-one key drop. Every other type is read straight from the store.
+   * User/group/role embed relation-derived data that a per-node cache cannot invalidate;
+   * model/model version and function carry a load-bearing pointer that would be silently wrong if
+   * served stale.
    */
   private static final Set<Entity.EntityType> CACHEABLE_TYPES =
       Sets.immutableEnumSet(
           Entity.EntityType.METALAKE,
           Entity.EntityType.CATALOG,
           Entity.EntityType.SCHEMA,
           Entity.EntityType.TABLE,
+          Entity.EntityType.TOPIC,
           Entity.EntityType.VIEW,
-          Entity.EntityType.COLUMN,
           Entity.EntityType.FILESET,
-          Entity.EntityType.TOPIC,
           Entity.EntityType.TAG,
-          Entity.EntityType.MODEL,
-          Entity.EntityType.MODEL_VERSION,
           Entity.EntityType.POLICY,
-          Entity.EntityType.TABLE_STATISTIC,
-          Entity.EntityType.JOB_TEMPLATE,
-          Entity.EntityType.JOB,
-          Entity.EntityType.AUDIT,
-          Entity.EntityType.FUNCTION);
+          Entity.EntityType.JOB);
 
   protected final Config cacheConfig;
 
@@ -88,6 +78,18 @@ public BaseEntityCache(Config config) {
     this.cacheConfig = config;
   }
 
+  /**
+   * {@inheritDoc}
+   *
+   * <p>Defaults to {@link Coherence#LOCAL_PER_NODE}: an in-memory, per-node cache whose changes
+   * must be propagated to the other nodes. A shared implementation must override this to return
+   * {@link Coherence#SHARED}.
+   */
+  @Override
+  public Coherence coherence() {
+    return Coherence.LOCAL_PER_NODE;
+  }
+
   /**
    * Returns whether entities of the given type may be cached.
    *

core/src/main/java/org/apache/gravitino/cache/CaffeineEntityCache.java

@@ -30,7 +30,6 @@
 import com.googlecode.concurrenttrees.radix.RadixTree;
 import com.googlecode.concurrenttrees.radix.node.concrete.DefaultCharArrayNodeFactory;
 import java.util.List;
-import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutorService;
@@ -43,7 +42,6 @@
 import org.apache.gravitino.Entity;
 import org.apache.gravitino.HasIdentifier;
 import org.apache.gravitino.NameIdentifier;
-import org.apache.gravitino.meta.ModelVersionEntity;
 import org.apache.gravitino.utils.HierarchicalSchemaUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -57,10 +55,10 @@
  * schemas and tables under it).
  *
  * <p>Relation query results are NOT cached by this implementation; relation and list operations
- * always fall back to the {@code EntityStore}. Entity types whose materialized form embeds
- * relation-derived data ({@code USER}, {@code GROUP}, {@code ROLE}) are excluded from caching
- * entirely by {@link BaseEntityCache#put}, because without relation tracking their entries could
- * not be invalidated when the referenced entities change.
+ * always fall back to the {@code EntityStore}. Only the self-contained metadata objects listed in
+ * {@link BaseEntityCache#isCacheable} are cached; every other type (user/group/role, model/model
+ * version, function, job template, and any type not in that allowlist) is read straight from the
+ * {@code EntityStore}.
  */
 public class CaffeineEntityCache extends BaseEntityCache {
   private static final int CACHE_CLEANUP_CORE_THREADS = 1;
@@ -228,13 +226,9 @@ protected <E extends Entity & HasIdentifier> void doPut(E entity) {
   /** {@inheritDoc} */
   @Override
   public <E extends Entity & HasIdentifier> void invalidateOnKeyChange(E entity) {
-    // Invalidate the cache if inserting the entity may affect related cache keys.
-    // For example, inserting a model version changes the latest version of the model,
-    // so the corresponding model cache entry should be invalidated.
-    if (Objects.requireNonNull(entity.type()) == Entity.EntityType.MODEL_VERSION) {
-      NameIdentifier modelIdent = ((ModelVersionEntity) entity).modelIdentifier();
-      invalidate(modelIdent, Entity.EntityType.MODEL);
-    }
+    // Every cacheable entity is self-contained (see BaseEntityCache#isCacheable), so inserting one
+    // never requires invalidating a different key. Kept for the SPI contract; implementations that
+    // cache derived entries can override this.
   }
 
   /** {@inheritDoc} */

core/src/main/java/org/apache/gravitino/cache/Coherence.java

@@ -0,0 +1,45 @@
+/*
+ * 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.cache;
+
+/**
+ * Describes how an {@link EntityCache} implementation stays coherent across a multi-node Gravitino
+ * cluster. The write path reads this marker to decide whether a change made on one node must be
+ * propagated to the other nodes.
+ */
+public enum Coherence {
+  /** The implementation does not retain entries, so it needs no coherence mechanism. */
+  NONE,
+
+  /**
+   * Each node keeps its own copy of the cache, so a change made on one node does not clear the
+   * copies held by the other nodes. To stay correct across a cluster the write path must publish
+   * the change (via the {@code entity_change_log}) so every other node's poller invalidates its own
+   * copy. The default {@code caffeine} cache is {@code LOCAL_PER_NODE}.
+   */
+  LOCAL_PER_NODE,
+
+  /**
+   * A single copy is shared by the whole cluster, so a write clears it once and every node observes
+   * the change immediately. There is nothing per-node to propagate. The {@code redis} cache is
+   * {@code SHARED}.
+   */
+  SHARED
+}

core/src/main/java/org/apache/gravitino/cache/EntityCache.java

@@ -28,6 +28,14 @@
  * this SPI; relation and list operations always fall back to the {@code EntityStore}.
  */
 public interface EntityCache extends SupportsEntityStoreCache {
+  /**
+   * Returns how this cache stays coherent across a multi-node cluster. The write path uses this to
+   * decide whether a local change must be propagated to the other nodes (see {@link Coherence}).
+   *
+   * @return the {@link Coherence} model of this cache
+   */
+  Coherence coherence();
+
   /**
    * Clears all entries from the cache, including data and index, resetting it to an empty state.
    */

core/src/main/java/org/apache/gravitino/cache/NoOpsCache.java

@@ -37,6 +37,12 @@ public NoOpsCache(Config config) {
     super(config);
   }
 
+  /** {@inheritDoc} */
+  @Override
+  public Coherence coherence() {
+    return Coherence.NONE;
+  }
+
   /** {@inheritDoc} */
   @Override
   protected void invalidateExpiredItem(EntityCacheKey key) {

core/src/main/java/org/apache/gravitino/cache/SupportsEntityStoreCache.java

@@ -68,12 +68,11 @@ <E extends Entity & HasIdentifier> Optional<E> getIfPresent(
    * invalidation path covers, so it must remain non-cacheable until its invalidation behavior has
    * been validated. Implementations extending {@link BaseEntityCache} get this check for free.
    *
-   * <p>{@code USER}, {@code GROUP} and {@code ROLE} are deliberately excluded because they are
-   * materialized with relation-derived data joined in at load time (a role carries its securable
-   * objects, and a user/group carries its role names). That embedded data goes stale through
-   * mutations on other entities that touch neither this entity's own key nor any of its hierarchy
-   * ancestors. Since no invalidation path covers them, caching them serves stale authorization
-   * data.
+   * <p>Only self-contained entities are approved. User/group/role are deliberately excluded because
+   * they contain relation-derived data whose source can change through another entity. Model/model
+   * version and function are also excluded because they carry load-bearing pointers that must not
+   * be served stale. Other unapproved and newly introduced entity types go straight to the store
+   * until their invalidation behavior has been validated.
    *
    * <p>Implementations must also invoke {@link #invalidateOnKeyChange(Entity)} for every entity,
    * including non-cacheable ones, since a non-cacheable entity may still invalidate a cacheable

core/src/main/java/org/apache/gravitino/catalog/CatalogChangeLogListener.java

@@ -24,6 +24,7 @@
 import org.apache.gravitino.Entity.EntityType;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.storage.relational.EntityChangeLogListener;
+import org.apache.gravitino.storage.relational.EntityChangeLogNameIdentifierCodec;
 import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -116,11 +117,17 @@ private Optional<NameIdentifier> catalogIdentifier(EntityChangeRecord change) {
       return Optional.empty();
     }
 
-    String[] names = change.getFullName().split("\\.");
-    if (names.length != 2) {
+    NameIdentifier ident;
+    try {
+      ident = EntityChangeLogNameIdentifierCodec.decode(change.getFullName());
+    } catch (IllegalArgumentException e) {
       LOG.warn("Invalid catalog full name in entity change log: {}", change.getFullName());
       return Optional.empty();
     }
-    return Optional.of(NameIdentifier.of(names[0], names[1]));
+    if (ident.namespace().length() != 1) {
+      LOG.warn("Invalid catalog full name in entity change log: {}", change.getFullName());
+      return Optional.empty();
+    }
+    return Optional.of(ident);
   }
 }

core/src/main/java/org/apache/gravitino/storage/relational/EntityCacheChangeLogListener.java

@@ -0,0 +1,131 @@
+/*
+ * 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 com.google.common.base.Preconditions;
+import java.util.List;
+import java.util.Locale;
+import org.apache.gravitino.Entity.EntityType;
+import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.cache.EntityCache;
+import org.apache.gravitino.storage.relational.po.cache.EntityChangeRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Keeps a per-node {@link EntityCache} coherent across a multi-node cluster by replaying {@code
+ * entity_change_log} rows written by other nodes.
+ *
+ * <p>Every ALTER/DROP row is replayed as a direct {@link EntityCache#invalidate(NameIdentifier,
+ * EntityType)} for exactly the changed entity key. Because the cache indexes its keys by identifier
+ * prefix, invalidating a container (for example a schema) cascades to its cached children on the
+ * local node through that forward prefix scan; no reverse index is involved.
+ *
+ * <p>This listener is registered only for a {@link
+ * org.apache.gravitino.cache.Coherence#LOCAL_PER_NODE} cache: a shared cache has a single
+ * cluster-wide copy and nothing per-node to invalidate. It is called <em>synchronously</em> on the
+ * poller thread, so it performs only fast, in-memory, idempotent invalidations.
+ *
+ * <p>The two failure modes of a replayed row are handled separately, because they mean different
+ * things:
+ *
+ * <ul>
+ *   <li>A <b>malformed row</b> (unknown entity type, undecodable full name) names no entity, so
+ *       there is nothing to invalidate. It is logged and skipped, and the rest of the batch still
+ *       applies.
+ *   <li>A <b>failed invalidation</b> means this node may now serve stale metadata indefinitely. The
+ *       whole cache is cleared instead, which is strictly stronger than the invalidation that
+ *       failed and only costs a cold-cache penalty, since the cache is derived state. If even the
+ *       clear fails the exception propagates, and {@link EntityChangeLogPoller} retries the batch
+ *       and ultimately applies its configured listener failure action.
+ * </ul>
+ */
+public class EntityCacheChangeLogListener implements EntityChangeLogListener {
+
+  private static final Logger LOG = LoggerFactory.getLogger(EntityCacheChangeLogListener.class);
+
+  private final EntityCache cache;
+
+  /**
+   * Creates a listener that invalidates the given entity store cache.
+   *
+   * @param cache the per-node entity store cache to keep coherent
+   */
+  public EntityCacheChangeLogListener(EntityCache cache) {
+    Preconditions.checkArgument(cache != null, "cache cannot be null");
+    this.cache = cache;
+  }
+
+  @Override
+  public void onEntityChange(List<EntityChangeRecord> changes) {
+    for (EntityChangeRecord change : changes) {
+      EntityType type = entityType(change);
+      NameIdentifier ident = identifier(change);
+      if (type == null || ident == null) {
+        // Already logged by the parsing helpers. A row that names no entity cannot invalidate
+        // anything, so skipping it leaves no stale entry behind.
+        continue;
+      }
+
+      try {
+        LOG.debug("Invalidating entity cache due to entity change log: {} ({})", ident, type);
+        cache.invalidate(ident, type);
+      } catch (RuntimeException e) {
+        // Dropping a single invalidation would leave this node serving that entity stale until it
+        // expires. Clearing the whole cache is the safe superset, and it also covers the rest of
+        // this batch, so there is nothing left to replay.
+        LOG.error(
+            "Failed to invalidate {} ({}) from the entity change log, clearing the local entity "
+                + "cache to stay coherent",
+            ident,
+            type,
+            e);
+        cache.clear();
+        return;
+      }
+    }
+  }
+
+  private EntityType entityType(EntityChangeRecord change) {
+    if (change.getEntityType() == null) {
+      LOG.warn("Invalid entity type in entity change log: null");
+      return null;
+    }
+    try {
+      return EntityType.valueOf(change.getEntityType().toUpperCase(Locale.ROOT));
+    } catch (IllegalArgumentException e) {
+      LOG.warn("Unknown entity type in entity change log: {}", change.getEntityType());
+      return null;
+    }
+  }
+
+  private NameIdentifier identifier(EntityChangeRecord change) {
+    String fullName = change.getFullName();
+    if (fullName == null || fullName.isEmpty()) {
+      LOG.warn("Invalid full name in entity change log: {}", fullName);
+      return null;
+    }
+    try {
+      return EntityChangeLogNameIdentifierCodec.decode(fullName);
+    } catch (IllegalArgumentException e) {
+      LOG.warn("Undecodable full name in entity change log: {}", fullName, e);
+      return null;
+    }
+  }
+}