[TEST]

common/src/main/java/org/apache/gravitino/config/ConfigConstants.java

@@ -93,5 +93,5 @@ private ConfigConstants() {}
   public static final String VERSION_1_3_0 = "1.3.0";
 
   /** The current version of backend storage initialization script. */
-  public static final String CURRENT_SCRIPT_VERSION = VERSION_1_2_0;
+  public static final String CURRENT_SCRIPT_VERSION = VERSION_1_3_0;
 }

core/src/main/java/org/apache/gravitino/Configs.java

@@ -329,6 +329,24 @@ private Configs() {}
           .longConf()
           .createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_OWNER_CACHE_SIZE);
 
+  public static final long DEFAULT_GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE = 100000L;
+
+  public static final ConfigEntry<Long> GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE =
+      new ConfigBuilder("gravitino.authorization.jcasbin.metadataIdCacheSize")
+          .doc("The maximum size of the metadata-id cache for authorization")
+          .version(ConfigConstants.VERSION_1_3_0)
+          .longConf()
+          .createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_METADATA_ID_CACHE_SIZE);
+
+  public static final long DEFAULT_GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS = 3L;
+
+  public static final ConfigEntry<Long> GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS =
+      new ConfigBuilder("gravitino.authorization.jcasbin.changePollIntervalSecs")
+          .doc("The interval in seconds for polling entity and owner changes")
+          .version(ConfigConstants.VERSION_1_3_0)
+          .longConf()
+          .createWithDefault(DEFAULT_GRAVITINO_AUTHORIZATION_CHANGE_POLL_INTERVAL_SECS);
+
   public static final ConfigEntry<List<String>> SERVICE_ADMINS =
       new ConfigBuilder("gravitino.authorization.serviceAdmins")
           .doc("The admins of Gravitino service")

core/src/main/java/org/apache/gravitino/authorization/GravitinoAuthorizer.java

@@ -164,4 +164,18 @@ default void handleRolePrivilegeChange(String metalake, String roleName) {
    */
   void handleMetadataOwnerChange(
       String metalake, Long oldOwnerId, NameIdentifier nameIdentifier, Entity.EntityType type);
+
+  /**
+   * Called when an entity undergoes a structural change (rename or drop) that invalidates cached
+   * name-to-id mappings in the metadataIdCache. The authorizer evicts the cache key for the given
+   * entity and all of its descendants (cascade invalidation).
+   *
+   * @param metalake the metalake name
+   * @param nameIdentifier the entity name identifier
+   * @param type the entity type
+   */
+  default void handleEntityStructuralChange(
+      String metalake, NameIdentifier nameIdentifier, Entity.EntityType type) {
+    // default no-op for backward compatibility
+  }
 }

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

@@ -0,0 +1,88 @@
+/*
+ * 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;
+
+import com.github.benmanes.caffeine.cache.Cache;
+import com.github.benmanes.caffeine.cache.Caffeine;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A Caffeine-backed implementation of {@link GravitinoCache}. Supports configurable TTL and maximum
+ * size.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ */
+public class CaffeineGravitinoCache<K, V> implements GravitinoCache<K, V> {
+
+  private final Cache<K, V> cache;
+
+  /**
+   * Creates a new CaffeineGravitinoCache with the given TTL and maximum size.
+   *
+   * @param ttlMs the time-to-live in milliseconds for cache entries (safety-net TTL)
+   * @param maxSize the maximum number of entries in the cache
+   */
+  public CaffeineGravitinoCache(long ttlMs, long maxSize) {
+    this.cache =
+        Caffeine.newBuilder()
+            .expireAfterWrite(ttlMs, TimeUnit.MILLISECONDS)
+            .maximumSize(maxSize)
+            .build();
+  }
+
+  @Override
+  public Optional<V> getIfPresent(K key) {
+    V value = cache.getIfPresent(key);
+    return Optional.ofNullable(value);
+  }
+
+  @Override
+  public void put(K key, V value) {
+    cache.put(key, value);
+  }
+
+  @Override
+  public void invalidate(K key) {
+    cache.invalidate(key);
+  }
+
+  @Override
+  public void invalidateAll() {
+    cache.invalidateAll();
+  }
+
+  @Override
+  public void invalidateByPrefix(String prefix) {
+    cache.asMap().keySet().removeIf(k -> k.toString().startsWith(prefix));
+  }
+
+  @Override
+  public long size() {
+    cache.cleanUp();
+    return cache.estimatedSize();
+  }
+
+  @Override
+  public void close() {
+    cache.invalidateAll();
+    cache.cleanUp();
+  }
+}

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

@@ -0,0 +1,74 @@
+/*
+ * 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;
+
+import java.io.Closeable;
+import java.util.Optional;
+
+/**
+ * A general-purpose cache interface used by the authorization subsystem. Implementations include a
+ * Caffeine-backed cache and a no-op cache for testing.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ */
+public interface GravitinoCache<K, V> extends Closeable {
+
+  /**
+   * Returns the value associated with the key, or empty if not present.
+   *
+   * @param key the cache key
+   * @return an Optional containing the cached value, or empty if absent
+   */
+  Optional<V> getIfPresent(K key);
+
+  /**
+   * Associates the value with the key in the cache.
+   *
+   * @param key the cache key
+   * @param value the value to cache
+   */
+  void put(K key, V value);
+
+  /**
+   * Removes the entry for the given key.
+   *
+   * @param key the cache key to invalidate
+   */
+  void invalidate(K key);
+
+  /** Removes all entries from the cache. */
+  void invalidateAll();
+
+  /**
+   * Evicts all entries whose key (as a String) starts with the given prefix. Only meaningful when K
+   * = String. Used by metadataIdCache for hierarchical cascade invalidation: dropping a catalog
+   * evicts the catalog entry plus all schema/table/fileset/... entries beneath it.
+   *
+   * @param prefix the prefix to match against key strings
+   */
+  void invalidateByPrefix(String prefix);
+
+  /**
+   * Returns the approximate number of entries in the cache.
+   *
+   * @return the cache size
+   */
+  long size();
+}

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

@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+import java.util.Optional;
+
+/**
+ * A no-op implementation of {@link GravitinoCache} that never caches anything. Useful for testing
+ * and for environments where caching is disabled.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ */
+public class NoOpsGravitinoCache<K, V> implements GravitinoCache<K, V> {
+
+  @Override
+  public Optional<V> getIfPresent(K key) {
+    return Optional.empty();
+  }
+
+  @Override
+  public void put(K key, V value) {
+    // no-op
+  }
+
+  @Override
+  public void invalidate(K key) {
+    // no-op
+  }
+
+  @Override
+  public void invalidateAll() {
+    // no-op
+  }
+
+  @Override
+  public void invalidateByPrefix(String prefix) {
+    // no-op
+  }
+
+  @Override
+  public long size() {
+    return 0;
+  }
+
+  @Override
+  public void close() {
+    // no-op
+  }
+}

core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java

@@ -0,0 +1,53 @@
+/*
+ * 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.mapper;
+
+import java.util.List;
+import org.apache.gravitino.storage.relational.po.auth.EntityChangeRecord;
+import org.apache.ibatis.annotations.DeleteProvider;
+import org.apache.ibatis.annotations.InsertProvider;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.SelectProvider;
+
+/**
+ * A MyBatis Mapper for entity_change_log table operations.
+ *
+ * <p>This append-only log tracks structural changes to entities (create, alter, drop) and is used
+ * by the entity change poller to drive targeted invalidation of the metadataIdCache on HA peer
+ * nodes.
+ */
+public interface EntityChangeLogMapper {
+
+  String ENTITY_CHANGE_LOG_TABLE_NAME = "entity_change_log";
+
+  @SelectProvider(type = EntityChangeLogSQLProviderFactory.class, method = "selectEntityChanges")
+  List<EntityChangeRecord> selectChanges(
+      @Param("createdAtAfter") long createdAtAfter, @Param("maxRows") int maxRows);
+
+  @InsertProvider(type = EntityChangeLogSQLProviderFactory.class, method = "insertEntityChange")
+  void insertChange(
+      @Param("metalakeName") String metalakeName,
+      @Param("entityType") String entityType,
+      @Param("fullName") String fullName,
+      @Param("operateType") String operateType,
+      @Param("createdAt") long createdAt);
+
+  @DeleteProvider(type = EntityChangeLogSQLProviderFactory.class, method = "pruneOldEntityChanges")
+  void pruneOldEntries(@Param("before") long before);
+}