This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 640ac60ae8 [#11032] feat(authz): add cache support infrastructure
(#11033)
640ac60ae8 is described below
commit 640ac60ae898af32fe50fab03adab6e2bde8e0d5
Author: Qi Yu <[email protected]>
AuthorDate: Thu May 14 15:41:43 2026 +0800
[#11032] feat(authz): add cache support infrastructure (#11033)
### What changes were proposed in this pull request?
This PR extracts the infrastructure part from #10996 to support the
follow-up JcasbinAuthorizer cache refactor.
It includes:
- Add a generic `GravitinoCache` abstraction with Caffeine and no-op
implementations.
- Add per-request lookup caches in `AuthorizationRequestContext` for
user, group, metadata id, and owner lookup deduplication.
- Thread `AuthorizationRequestContext` through authorization
filter/executor paths.
- Update `isMetalakeUser` to accept `AuthorizationRequestContext`.
- Switch entity/owner change polling mapper APIs to id-based cursors and
expose latest id lookup helpers.
This PR intentionally does not rewrite `JcasbinAuthorizer` cache
behavior. The JcasbinAuthorizer version-validated cache refactor will
stay in the follow-up PR.
### Why are the changes needed?
This reduces the size of #10996 and provides the reusable support layer
needed by the JcasbinAuthorizer cache refactor.
Fix: #11032
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `./gradlew :core:spotlessApply :server-common:spotlessApply
:server:spotlessApply`
- `./gradlew :core:test --tests
org.apache.gravitino.authorization.TestAuthorizationRequestContext
--tests org.apache.gravitino.cache.TestGravitinoCache --tests
org.apache.gravitino.storage.relational.mapper.provider.base.TestAuthMappers
--tests
org.apache.gravitino.storage.relational.mapper.provider.base.TestEntityChangeLogMapper
--tests
org.apache.gravitino.storage.relational.service.TestEntityChangeLogService
--tests
org.apache.gravitino.storage.relational.service.TestTableMetaService
:server-common:test --tests
org.apache.gravitino.server.authorization.TestPassThroughAuthorizer
:server:test --tests
org.apache.gravitino.server.web.filter.TestGravitinoInterceptionService`
---
.../authorization/AuthorizationRequestContext.java | 160 ++++++-----
.../authorization/AuthorizationUtils.java | 7 +-
.../authorization/GravitinoAuthorizer.java | 6 +-
.../gravitino/cache/CaffeineGravitinoCache.java | 101 +++++++
.../org/apache/gravitino/cache/GravitinoCache.java | 74 +++++
.../gravitino/cache/NoOpsGravitinoCache.java | 66 +++++
.../relational/mapper/EntityChangeLogMapper.java | 5 +-
.../mapper/EntityChangeLogSQLProviderFactory.java | 8 +-
.../storage/relational/mapper/OwnerMetaMapper.java | 5 +-
.../mapper/OwnerMetaSQLProviderFactory.java | 8 +-
.../base/EntityChangeLogBaseSQLProvider.java | 27 +-
.../provider/base/OwnerMetaBaseSQLProvider.java | 21 +-
.../relational/po/auth/ChangedOwnerInfo.java | 1 +
.../TestAuthorizationRequestContext.java | 226 +++++++++++++++
.../apache/gravitino/cache/TestGravitinoCache.java | 319 +++++++++++++++++++++
.../mapper/provider/base/TestAuthMappers.java | 9 +-
.../provider/base/TestEntityChangeLogMapper.java | 3 +-
.../service/TestEntityChangeLogService.java | 81 +++---
.../relational/service/TestTableMetaService.java | 21 +-
.../authorization/PassThroughAuthorizer.java | 2 +-
.../AuthorizationExpressionConstants.java | 3 +-
.../AuthorizationExpressionConverter.java | 4 +-
.../authorization/jcasbin/JcasbinAuthorizer.java | 2 +-
.../authorization/MockGravitinoAuthorizer.java | 2 +-
.../authorization/TestPassThroughAuthorizer.java | 3 +-
.../web/filter/GravitinoInterceptionService.java | 8 +-
.../AssociatePolicyAuthorizationExecutor.java | 3 +-
.../AssociateTagAuthorizationExecutor.java | 3 +-
.../authorization/AuthorizationExecutor.java | 4 +-
.../authorization/CommonAuthorizerExecutor.java | 5 +-
.../authorization/RunJobAuthorizationExecutor.java | 6 +-
.../filter/TestGravitinoInterceptionService.java | 4 +-
32 files changed, 1039 insertions(+), 158 deletions(-)
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/AuthorizationRequestContext.java
b/core/src/main/java/org/apache/gravitino/authorization/AuthorizationRequestContext.java
index 392eb13c27..554fed8b52 100644
---
a/core/src/main/java/org/apache/gravitino/authorization/AuthorizationRequestContext.java
+++
b/core/src/main/java/org/apache/gravitino/authorization/AuthorizationRequestContext.java
@@ -6,7 +6,9 @@
* 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
@@ -20,11 +22,35 @@ package org.apache.gravitino.authorization;
import java.security.Principal;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
import org.apache.gravitino.MetadataObject;
-
+import org.apache.gravitino.storage.relational.po.auth.GroupUpdatedAt;
+import org.apache.gravitino.storage.relational.po.auth.OwnerInfo;
+import org.apache.gravitino.storage.relational.po.auth.UserUpdatedAt;
+
+/**
+ * Per-HTTP-request scratchpad shared by {@link GravitinoAuthorizer} calls. A
fresh instance is
+ * created for each request by the authorization filter and threaded through
{@code authorize},
+ * {@code isOwner}, {@code isMetalakeUser} etc., so that:
+ *
+ * <ul>
+ * <li>repeated authorization decisions for the same {@code (principal,
metalake, object,
+ * privilege)} short-circuit via {@link #allowAuthorizerCache} / {@link
#denyAuthorizerCache};
+ * <li>user identity, name→id and metadataId→owner lookups are de-duplicated
within the request
+ * (see the {@code computeXxxIfAbsent} helpers) so each underlying DB
query runs at most once;
+ * <li>per-request role loading happens at most once via {@link
#loadRole(Runnable)}.
+ * </ul>
+ *
+ * <p>Instances are not intended to outlive a request and are not reusable
across threads beyond the
+ * request handling thread; the internal maps are {@link ConcurrentHashMap}
purely to tolerate any
+ * incidental fan-out (e.g. async listeners) within the same request scope.
+ */
public class AuthorizationRequestContext {
/** Used to cache the results of metadata authorization. */
@@ -36,6 +62,18 @@ public class AuthorizationRequestContext {
/** Used to determine whether the role has already been loaded. */
private final AtomicBoolean hasLoadRole = new AtomicBoolean();
+ /** Per-request user identity cache. Key: {@code metalake::userName}. */
+ private final Map<String, Optional<UserUpdatedAt>> userInfoCache = new
ConcurrentHashMap<>();
+
+ /** Per-request group identity cache. Key: {@code metalake::groupName}. */
+ private final Map<String, Optional<GroupUpdatedAt>> groupInfoCache = new
ConcurrentHashMap<>();
+
+ /** Per-request name→id cache. Deduplicates resolveMetadataId within a
single request. */
+ private final Map<String, Long> metadataIdCache = new ConcurrentHashMap<>();
+
+ /** Per-request metadataId→owner cache. Deduplicates isOwner within a single
request. */
+ private final Map<Long, Optional<OwnerInfo>> ownerCache = new
ConcurrentHashMap<>();
+
private volatile String originalAuthorizationExpression;
/**
@@ -78,6 +116,11 @@ public class AuthorizationRequestContext {
return denyAuthorizerCache.computeIfAbsent(context, authorizer);
}
+ /**
+ * Runs {@code runnable} at most once per request. The double-checked guard
plus {@code
+ * synchronized(this)} prevents two concurrent authorize calls in the same
request from both
+ * triggering the (potentially expensive) role load.
+ */
public void loadRole(Runnable runnable) {
if (hasLoadRole.get()) {
return;
@@ -95,6 +138,45 @@ public class AuthorizationRequestContext {
}
}
+ /**
+ * Per-request {@link UserUpdatedAt} dedup. Loader may return {@link
Optional#empty()} to cache
+ * the "user not found" outcome and avoid repeated DB lookups within a
single request.
+ */
+ public Optional<UserUpdatedAt> computeUserInfoIfAbsent(
+ String key, Function<String, Optional<UserUpdatedAt>> loader) {
+ return userInfoCache.computeIfAbsent(
+ key, k -> Objects.requireNonNull(loader.apply(k), "User info loader
must not return null"));
+ }
+
+ /**
+ * Per-request {@link GroupUpdatedAt} dedup. Loader may return {@link
Optional#empty()} to cache
+ * the "group not found" outcome and avoid repeated DB lookups within a
single request.
+ */
+ public Optional<GroupUpdatedAt> computeGroupInfoIfAbsent(
+ String key, Function<String, Optional<GroupUpdatedAt>> loader) {
+ return groupInfoCache.computeIfAbsent(
+ key,
+ k -> Objects.requireNonNull(loader.apply(k), "Group info loader must
not return null"));
+ }
+
+ /** Per-request name→id dedup. Loader must return a non-null id or throw. */
+ public Long computeMetadataIdIfAbsent(String key, Function<String, Long>
loader) {
+ return metadataIdCache.computeIfAbsent(
+ key,
+ k -> Objects.requireNonNull(loader.apply(k), "Metadata id loader must
not return null"));
+ }
+
+ /**
+ * Per-request metadataId→owner dedup. Loader returns {@link
Optional#empty()} when the object has
+ * no owner; the absent result is cached as well.
+ */
+ public Optional<OwnerInfo> computeOwnerIfAbsent(
+ Long metadataId, Function<Long, Optional<OwnerInfo>> loader) {
+ return ownerCache.computeIfAbsent(
+ metadataId,
+ id -> Objects.requireNonNull(loader.apply(id), "Owner loader must not
return null"));
+ }
+
public String getOriginalAuthorizationExpression() {
return originalAuthorizationExpression;
}
@@ -103,70 +185,18 @@ public class AuthorizationRequestContext {
this.originalAuthorizationExpression = originalAuthorizationExpression;
}
+ /**
+ * Composite key for {@link #allowAuthorizerCache} / {@link
#denyAuthorizerCache}. Immutable —
+ * mutating any field after construction would silently corrupt the {@link
+ * java.util.Objects#hashCode} used by the backing {@link ConcurrentHashMap}.
+ */
+ @Getter
+ @AllArgsConstructor
+ @EqualsAndHashCode
public static class AuthorizationKey {
- private Principal principal;
- private String metalake;
- private MetadataObject metadataObject;
- private Privilege.Name privilege;
-
- public AuthorizationKey(
- Principal principal,
- String metalake,
- MetadataObject metadataObject,
- Privilege.Name privilege) {
- this.principal = principal;
- this.metalake = metalake;
- this.metadataObject = metadataObject;
- this.privilege = privilege;
- }
-
- @Override
- public boolean equals(Object o) {
- if (!(o instanceof AuthorizationKey)) {
- return false;
- }
- AuthorizationKey that = (AuthorizationKey) o;
- return Objects.equals(principal, that.principal)
- && Objects.equals(metalake, that.metalake)
- && Objects.equals(metadataObject, that.metadataObject)
- && Objects.equals(privilege, that.privilege);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(principal, metalake, metadataObject, privilege);
- }
-
- public Principal getPrincipal() {
- return principal;
- }
-
- public void setPrincipal(Principal principal) {
- this.principal = principal;
- }
-
- public String getMetalake() {
- return metalake;
- }
-
- public void setMetalake(String metalake) {
- this.metalake = metalake;
- }
-
- public MetadataObject getMetadataObject() {
- return metadataObject;
- }
-
- public void setMetadataObject(MetadataObject metadataObject) {
- this.metadataObject = metadataObject;
- }
-
- public Privilege.Name getPrivilege() {
- return privilege;
- }
-
- public void setPrivilege(Privilege.Name privilege) {
- this.privilege = privilege;
- }
+ private final Principal principal;
+ private final String metalake;
+ private final MetadataObject metadataObject;
+ private final Privilege.Name privilege;
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
b/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
index 4938ff6034..16642a5855 100644
---
a/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
+++
b/core/src/main/java/org/apache/gravitino/authorization/AuthorizationUtils.java
@@ -117,8 +117,13 @@ public class AuthorizationUtils {
private AuthorizationUtils() {}
public static void checkCurrentUser(String metalake, String user) {
+ checkCurrentUser(metalake, user, new AuthorizationRequestContext());
+ }
+
+ public static void checkCurrentUser(
+ String metalake, String user, AuthorizationRequestContext
requestContext) {
GravitinoAuthorizer authorizer =
GravitinoEnv.getInstance().gravitinoAuthorizer();
- if (authorizer != null && !authorizer.isMetalakeUser(metalake)) {
+ if (authorizer != null && !authorizer.isMetalakeUser(metalake,
requestContext)) {
throw new ForbiddenException(
"Current user %s doesn't exist in the metalake %s, you should add
the user to the metalake first",
user, metalake);
diff --git
a/core/src/main/java/org/apache/gravitino/authorization/GravitinoAuthorizer.java
b/core/src/main/java/org/apache/gravitino/authorization/GravitinoAuthorizer.java
index 7965f173ad..6ab74382ad 100644
---
a/core/src/main/java/org/apache/gravitino/authorization/GravitinoAuthorizer.java
+++
b/core/src/main/java/org/apache/gravitino/authorization/GravitinoAuthorizer.java
@@ -92,12 +92,14 @@ public interface GravitinoAuthorizer extends Closeable {
boolean isSelf(Entity.EntityType type, NameIdentifier nameIdentifier);
/**
- * Determine whether the user is the metalake user
+ * Determine whether the user is the metalake user.
*
* @param metalake metalake
+ * @param requestContext authorization request context; enables per-request
dedup with other
+ * authorizer calls (e.g. {@code authorize}/{@code isOwner}) that look
up the same user.
* @return authorization result
*/
- boolean isMetalakeUser(String metalake);
+ boolean isMetalakeUser(String metalake, AuthorizationRequestContext
requestContext);
/**
* Determine whether the user can set owner
diff --git
a/core/src/main/java/org/apache/gravitino/cache/CaffeineGravitinoCache.java
b/core/src/main/java/org/apache/gravitino/cache/CaffeineGravitinoCache.java
new file mode 100644
index 0000000000..f1ed54df75
--- /dev/null
+++ b/core/src/main/java/org/apache/gravitino/cache/CaffeineGravitinoCache.java
@@ -0,0 +1,101 @@
+/*
+ * 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 com.github.benmanes.caffeine.cache.Ticker;
+import com.google.common.annotations.VisibleForTesting;
+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(ttlMs, maxSize, Ticker.systemTicker());
+ }
+
+ CaffeineGravitinoCache(long ttlMs, long maxSize, Ticker ticker) {
+ this.cache =
+ Caffeine.newBuilder()
+ .expireAfterWrite(ttlMs, TimeUnit.MILLISECONDS)
+ .maximumSize(maxSize)
+ .ticker(ticker)
+ .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) {
+ // Prefix invalidation scans all keys. It is intended for infrequent
structural invalidations
+ // such as dropping or renaming an entity hierarchy, not for per-request
hot paths.
+ cache.asMap().keySet().removeIf(k -> k instanceof String && ((String)
k).startsWith(prefix));
+ }
+
+ @Override
+ public long size() {
+ return cache.estimatedSize();
+ }
+
+ @VisibleForTesting
+ void cleanUp() {
+ cache.cleanUp();
+ }
+
+ @Override
+ public void close() {
+ cache.invalidateAll();
+ cache.cleanUp();
+ }
+}
diff --git a/core/src/main/java/org/apache/gravitino/cache/GravitinoCache.java
b/core/src/main/java/org/apache/gravitino/cache/GravitinoCache.java
new file mode 100644
index 0000000000..3771608e3c
--- /dev/null
+++ b/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 is a String and 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();
+}
diff --git
a/core/src/main/java/org/apache/gravitino/cache/NoOpsGravitinoCache.java
b/core/src/main/java/org/apache/gravitino/cache/NoOpsGravitinoCache.java
new file mode 100644
index 0000000000..8c41eabeef
--- /dev/null
+++ b/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
+ }
+}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
index ada263b359..d82754947c 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogMapper.java
@@ -38,7 +38,10 @@ public interface EntityChangeLogMapper {
@SelectProvider(type = EntityChangeLogSQLProviderFactory.class, method =
"selectEntityChanges")
List<EntityChangeRecord> selectEntityChanges(
- @Param("createdAtFrom") long createdAtFrom, @Param("maxRows") int
maxRows);
+ @Param("lastConsumedId") long lastConsumedId, @Param("maxRows") int
maxRows);
+
+ @SelectProvider(type = EntityChangeLogSQLProviderFactory.class, method =
"selectMaxChangeId")
+ Long selectMaxChangeId();
@InsertProvider(type = EntityChangeLogSQLProviderFactory.class, method =
"insertEntityChange")
void insertEntityChange(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
index 4ee8e43156..c48f2c0a1a 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/EntityChangeLogSQLProviderFactory.java
@@ -51,8 +51,12 @@ public class EntityChangeLogSQLProviderFactory {
static class EntityChangeLogH2Provider extends
EntityChangeLogBaseSQLProvider {}
public static String selectEntityChanges(
- @Param("createdAtFrom") long createdAtFrom, @Param("maxRows") int
maxRows) {
- return getProvider().selectEntityChanges(createdAtFrom, maxRows);
+ @Param("lastConsumedId") long lastConsumedId, @Param("maxRows") int
maxRows) {
+ return getProvider().selectEntityChanges(lastConsumedId, maxRows);
+ }
+
+ public static String selectMaxChangeId() {
+ return getProvider().selectMaxChangeId();
}
public static String insertEntityChange(
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
index b2872ac3b2..7a74310184 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaMapper.java
@@ -116,5 +116,8 @@ public interface OwnerMetaMapper {
@Param("metadataObjectType") String metadataObjectType);
@SelectProvider(type = OwnerMetaSQLProviderFactory.class, method =
"selectChangedOwners")
- List<ChangedOwnerInfo> selectChangedOwners(@Param("updatedAtFrom") long
updatedAtFrom);
+ List<ChangedOwnerInfo> selectChangedOwners(@Param("lastConsumedId") long
lastConsumedId);
+
+ @SelectProvider(type = OwnerMetaSQLProviderFactory.class, method =
"selectMaxChangeId")
+ Long selectMaxChangeId();
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
index 3d805459c8..bcf2795789 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/OwnerMetaSQLProviderFactory.java
@@ -121,7 +121,11 @@ public class OwnerMetaSQLProviderFactory {
return
getProvider().selectOwnerByMetadataObjectIdAndType(metadataObjectId,
metadataObjectType);
}
- public static String selectChangedOwners(@Param("updatedAtFrom") long
updatedAtFrom) {
- return getProvider().selectChangedOwners(updatedAtFrom);
+ public static String selectChangedOwners(@Param("lastConsumedId") long
lastConsumedId) {
+ return getProvider().selectChangedOwners(lastConsumedId);
+ }
+
+ public static String selectMaxChangeId() {
+ return getProvider().selectMaxChangeId();
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
index 6193263c5e..78695830cc 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/EntityChangeLogBaseSQLProvider.java
@@ -26,23 +26,27 @@ import org.apache.ibatis.annotations.Param;
public class EntityChangeLogBaseSQLProvider {
/**
- * Cursor-advance contract for the entity change poller:
+ * Cursor-advance contract for the entity change poller: {@code id} is
monotonic and unique, so
+ * callers only need to remember the last consumed id.
*
- * <p>The {@code created_at >= #{createdAtFrom}} predicate is
<b>inclusive</b>. Combined with
- * {@code ORDER BY created_at, id}, callers must remember the {@code
(lastCreatedAt, lastId)} of
- * the last consumed row and on the next poll: pass {@code createdAtFrom =
lastCreatedAt} and
- * client-side skip rows whose {@code id <= lastId} until they encounter a
row with {@code
- * created_at > lastCreatedAt}. Naively advancing by {@code lastCreatedAt +
1} would miss rows
- * sharing the same millisecond boundary; advancing by {@code lastCreatedAt}
re-reads the boundary
- * row and relies on the client-side id filter.
+ * <p>This table is a short-lived broadcast log for local cache
invalidation, not a queue. In a
+ * multi-node deployment every server instance has its own local cache and
should independently
+ * consume the same change rows. A new instance may initialize its cursor
from {@link
+ * #selectMaxChangeId()} because its cache starts empty and it does not need
historical
+ * invalidations. Re-consuming a row on an existing instance is acceptable:
entity DROP/ALTER
+ * handling only invalidates cache keys, and invalidation is idempotent.
*/
public String selectEntityChanges(
- @Param("createdAtFrom") long createdAtFrom, @Param("maxRows") int
maxRows) {
+ @Param("lastConsumedId") long lastConsumedId, @Param("maxRows") int
maxRows) {
return "SELECT id, metalake_name as metalakeName, entity_type as
entityType,"
+ " entity_full_name as fullName, operate_type as operateType,
created_at as createdAt"
+ " FROM "
+ ENTITY_CHANGE_LOG_TABLE_NAME
- + " WHERE created_at >= #{createdAtFrom} ORDER BY created_at, id LIMIT
#{maxRows}";
+ + " WHERE id > #{lastConsumedId} ORDER BY id LIMIT #{maxRows}";
+ }
+
+ public String selectMaxChangeId() {
+ return "SELECT COALESCE(MAX(id), 0) FROM " + ENTITY_CHANGE_LOG_TABLE_NAME;
}
/**
@@ -68,6 +72,9 @@ public class EntityChangeLogBaseSQLProvider {
}
public String pruneOldEntityChanges(@Param("before") long before) {
+ // Keep the retention window conservative. A running server can be delayed
by long GC pauses,
+ // network isolation, scheduler stalls, or clock skew between nodes;
pruning too aggressively
+ // can let that server miss an invalidation while its local cache is still
warm.
return "DELETE FROM "
+ ENTITY_CHANGE_LOG_TABLE_NAME
+ " WHERE created_at < #{before} LIMIT 1000";
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
index 5ec2fbdd89..2ac79dbab7 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/base/OwnerMetaBaseSQLProvider.java
@@ -274,6 +274,9 @@ public class OwnerMetaBaseSQLProvider {
public String deleteOwnerMetasByLegacyTimeline(
@Param("legacyTimeline") Long legacyTimeline, @Param("limit") int limit)
{
+ // Keep this cutoff comfortably behind the present time. These deleted
owner rows are also used
+ // as short-lived cache-invalidation signals; a running server that is
delayed by long GC,
+ // network isolation, scheduler stalls, or clock skew must still have time
to consume them.
return "DELETE FROM "
+ OWNER_TABLE_NAME
+ " WHERE deleted_at > 0 AND deleted_at < #{legacyTimeline} LIMIT
#{limit}";
@@ -290,13 +293,23 @@ public class OwnerMetaBaseSQLProvider {
+ " ORDER BY updated_at DESC, id DESC LIMIT 1";
}
- public String selectChangedOwners(@Param("updatedAtFrom") long
updatedAtFrom) {
- return "SELECT metadata_object_id as metadataObjectId,"
+ public String selectChangedOwners(@Param("lastConsumedId") long
lastConsumedId) {
+ // Owner changes are broadcast to every server instance because owner
caches are local. Each
+ // instance tracks its own last consumed id; re-reading a row is harmless
because cache
+ // invalidation is idempotent.
+ return "SELECT id,"
+ + " metadata_object_id as metadataObjectId,"
+ " metadata_object_type as metadataObjectType,"
+ " updated_at as updatedAt"
+ " FROM "
+ OWNER_TABLE_NAME
- + " WHERE deleted_at = 0 AND updated_at >= #{updatedAtFrom}"
- + " ORDER BY updated_at, id LIMIT 1000";
+ + " WHERE deleted_at = 0 AND id > #{lastConsumedId}"
+ + " ORDER BY id LIMIT 1000";
+ }
+
+ public String selectMaxChangeId() {
+ // A newly started server has an empty local owner cache. It can start
from the current max id
+ // and consume only owner changes that happen after startup.
+ return "SELECT COALESCE(MAX(id), 0) FROM " + OWNER_TABLE_NAME + " WHERE
deleted_at = 0";
}
}
diff --git
a/core/src/main/java/org/apache/gravitino/storage/relational/po/auth/ChangedOwnerInfo.java
b/core/src/main/java/org/apache/gravitino/storage/relational/po/auth/ChangedOwnerInfo.java
index 16f93d5fd8..04b9b32204 100644
---
a/core/src/main/java/org/apache/gravitino/storage/relational/po/auth/ChangedOwnerInfo.java
+++
b/core/src/main/java/org/apache/gravitino/storage/relational/po/auth/ChangedOwnerInfo.java
@@ -29,6 +29,7 @@ import lombok.Setter;
@NoArgsConstructor
@AllArgsConstructor
public class ChangedOwnerInfo {
+ private long id;
private long metadataObjectId;
private String metadataObjectType;
private long updatedAt;
diff --git
a/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationRequestContext.java
b/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationRequestContext.java
index 2216a7dcfe..e4ef672bcc 100644
---
a/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationRequestContext.java
+++
b/core/src/test/java/org/apache/gravitino/authorization/TestAuthorizationRequestContext.java
@@ -18,12 +18,18 @@
package org.apache.gravitino.authorization;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.gravitino.storage.relational.po.auth.GroupUpdatedAt;
+import org.apache.gravitino.storage.relational.po.auth.OwnerInfo;
+import org.apache.gravitino.storage.relational.po.auth.UserUpdatedAt;
import org.junit.jupiter.api.Test;
public class TestAuthorizationRequestContext {
@@ -101,4 +107,224 @@ public class TestAuthorizationRequestContext {
context.loadRole(counter::incrementAndGet);
assertEquals(2, counter.get(), "After a successful loadRole, further calls
must be ignored.");
}
+
+ @Test
+ public void testComputeUserInfoIfAbsentDedupesLoaderInvocation() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ AtomicInteger loaderCalls = new AtomicInteger();
+ UserUpdatedAt expected = new UserUpdatedAt(42L, 1234L);
+
+ Optional<UserUpdatedAt> first =
+ context.computeUserInfoIfAbsent(
+ "ml::alice",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(expected);
+ });
+ Optional<UserUpdatedAt> second =
+ context.computeUserInfoIfAbsent(
+ "ml::alice",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.empty();
+ });
+
+ assertTrue(first.isPresent());
+ assertEquals(expected, first.get());
+ assertTrue(second.isPresent());
+ assertEquals(expected, second.get());
+ assertEquals(1, loaderCalls.get(), "Loader must run only once for the same
key");
+ }
+
+ @Test
+ public void testComputeUserInfoIfAbsentCachesEmptyResult() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ AtomicInteger loaderCalls = new AtomicInteger();
+
+ Optional<UserUpdatedAt> first =
+ context.computeUserInfoIfAbsent(
+ "ml::ghost",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.empty();
+ });
+ Optional<UserUpdatedAt> second =
+ context.computeUserInfoIfAbsent(
+ "ml::ghost",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(new UserUpdatedAt(1L, 1L));
+ });
+
+ assertFalse(first.isPresent());
+ assertFalse(second.isPresent(), "Empty optional should be cached and
reused");
+ assertEquals(1, loaderCalls.get());
+ }
+
+ @Test
+ public void testComputeUserInfoIfAbsentDifferentKeysRunLoaderEachTime() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ AtomicInteger loaderCalls = new AtomicInteger();
+
+ context.computeUserInfoIfAbsent(
+ "ml::alice",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(new UserUpdatedAt(1L, 1L));
+ });
+ context.computeUserInfoIfAbsent(
+ "ml::bob",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(new UserUpdatedAt(2L, 2L));
+ });
+
+ assertEquals(2, loaderCalls.get());
+ }
+
+ @Test
+ public void testComputeGroupInfoIfAbsentCachesPresentAndAbsentResults() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ AtomicInteger loaderCalls = new AtomicInteger();
+
+ GroupUpdatedAt groupInfo = new GroupUpdatedAt(42L, 1234L);
+ Optional<GroupUpdatedAt> presentFirst =
+ context.computeGroupInfoIfAbsent(
+ "ml::group1",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(groupInfo);
+ });
+ Optional<GroupUpdatedAt> presentSecond =
+ context.computeGroupInfoIfAbsent(
+ "ml::group1",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.empty();
+ });
+
+ Optional<GroupUpdatedAt> absentFirst =
+ context.computeGroupInfoIfAbsent(
+ "ml::missing-group",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.empty();
+ });
+ Optional<GroupUpdatedAt> absentSecond =
+ context.computeGroupInfoIfAbsent(
+ "ml::missing-group",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(new GroupUpdatedAt(99L, 9999L));
+ });
+
+ assertEquals(Optional.of(groupInfo), presentFirst);
+ assertEquals(Optional.of(groupInfo), presentSecond);
+ assertFalse(absentFirst.isPresent());
+ assertFalse(absentSecond.isPresent(), "Absent group result must also be
cached");
+ assertEquals(2, loaderCalls.get(), "Loader must fire once per distinct
group key");
+ }
+
+ @Test
+ public void testComputeMetadataIdIfAbsentDedupesLoaderInvocation() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ AtomicInteger loaderCalls = new AtomicInteger();
+
+ Long first =
+ context.computeMetadataIdIfAbsent(
+ "ml::cat::TABLE",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return 1001L;
+ });
+ Long second =
+ context.computeMetadataIdIfAbsent(
+ "ml::cat::TABLE",
+ k -> {
+ loaderCalls.incrementAndGet();
+ return 9999L;
+ });
+
+ assertEquals(1001L, first);
+ assertEquals(1001L, second);
+ assertEquals(1, loaderCalls.get());
+ }
+
+ @Test
+ public void testComputeOwnerIfAbsentCachesPresentAndAbsentResults() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ AtomicInteger loaderCalls = new AtomicInteger();
+
+ OwnerInfo ownerInfo = new OwnerInfo(99L, "USER");
+ Optional<OwnerInfo> presentFirst =
+ context.computeOwnerIfAbsent(
+ 10L,
+ id -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(ownerInfo);
+ });
+ Optional<OwnerInfo> presentSecond =
+ context.computeOwnerIfAbsent(
+ 10L,
+ id -> {
+ loaderCalls.incrementAndGet();
+ return Optional.empty();
+ });
+
+ Optional<OwnerInfo> absentFirst =
+ context.computeOwnerIfAbsent(
+ 20L,
+ id -> {
+ loaderCalls.incrementAndGet();
+ return Optional.empty();
+ });
+ Optional<OwnerInfo> absentSecond =
+ context.computeOwnerIfAbsent(
+ 20L,
+ id -> {
+ loaderCalls.incrementAndGet();
+ return Optional.of(new OwnerInfo(123L, "USER"));
+ });
+
+ assertEquals(Optional.of(ownerInfo), presentFirst);
+ assertEquals(Optional.of(ownerInfo), presentSecond);
+ assertFalse(absentFirst.isPresent());
+ assertFalse(absentSecond.isPresent(), "Absent owner result must also be
cached");
+ assertEquals(2, loaderCalls.get(), "Loader must fire once per distinct
metadataId");
+ }
+
+ @Test
+ public void testComputeHelpersRejectNullLoaderResults() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+
+ NullPointerException userError =
+ assertThrows(
+ NullPointerException.class,
+ () -> context.computeUserInfoIfAbsent("ml::user", key -> null));
+ assertTrue(userError.getMessage().contains("User info loader must not
return null"));
+
+ NullPointerException groupError =
+ assertThrows(
+ NullPointerException.class,
+ () -> context.computeGroupInfoIfAbsent("ml::group", key -> null));
+ assertTrue(groupError.getMessage().contains("Group info loader must not
return null"));
+
+ NullPointerException metadataError =
+ assertThrows(
+ NullPointerException.class,
+ () -> context.computeMetadataIdIfAbsent("ml::catalog", key ->
null));
+ assertTrue(metadataError.getMessage().contains("Metadata id loader must
not return null"));
+
+ NullPointerException ownerError =
+ assertThrows(
+ NullPointerException.class, () -> context.computeOwnerIfAbsent(1L,
id -> null));
+ assertTrue(ownerError.getMessage().contains("Owner loader must not return
null"));
+ }
+
+ @Test
+ public void testOriginalAuthorizationExpressionRoundTrip() {
+ AuthorizationRequestContext context = new AuthorizationRequestContext();
+ context.setOriginalAuthorizationExpression("OWNER && HAS_PRIVILEGE");
+ assertEquals("OWNER && HAS_PRIVILEGE",
context.getOriginalAuthorizationExpression());
+ }
}
diff --git
a/core/src/test/java/org/apache/gravitino/cache/TestGravitinoCache.java
b/core/src/test/java/org/apache/gravitino/cache/TestGravitinoCache.java
new file mode 100644
index 0000000000..132e142002
--- /dev/null
+++ b/core/src/test/java/org/apache/gravitino/cache/TestGravitinoCache.java
@@ -0,0 +1,319 @@
+/*
+ * 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.Ticker;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Tests for {@link CaffeineGravitinoCache} and {@link NoOpsGravitinoCache}.
*/
+public class TestGravitinoCache {
+
+ @Test
+ void testCaffeinePutAndGet() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("key1", 100L);
+ cache.put("key2", 200L);
+
+ Optional<Long> val1 = cache.getIfPresent("key1");
+ Assertions.assertTrue(val1.isPresent());
+ Assertions.assertEquals(100L, val1.get());
+
+ Optional<Long> val2 = cache.getIfPresent("key2");
+ Assertions.assertTrue(val2.isPresent());
+ Assertions.assertEquals(200L, val2.get());
+
+ Optional<Long> missing = cache.getIfPresent("nonexistent");
+ Assertions.assertFalse(missing.isPresent());
+
+ Assertions.assertEquals(2, cache.size());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidate() {
+ CaffeineGravitinoCache<String, String> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("a", "val-a");
+ cache.put("b", "val-b");
+
+ cache.invalidate("a");
+ Assertions.assertFalse(cache.getIfPresent("a").isPresent());
+ Assertions.assertTrue(cache.getIfPresent("b").isPresent());
+
+ Assertions.assertEquals(1, cache.size());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateAll() {
+ CaffeineGravitinoCache<String, Integer> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("x", 1);
+ cache.put("y", 2);
+ cache.put("z", 3);
+
+ cache.invalidateAll();
+ Assertions.assertEquals(0, cache.size());
+ Assertions.assertFalse(cache.getIfPresent("x").isPresent());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateByPrefix() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ // Simulate hierarchical keys: metalake::catalog::schema::
+ cache.put("lake1::cat1::", 1L);
+ cache.put("lake1::cat1::s1::", 2L);
+ cache.put("lake1::cat1::s1::t1::TABLE", 3L);
+ cache.put("lake1::cat1::s1::t2::TABLE", 4L);
+ cache.put("lake1::cat1::s2::", 5L);
+ cache.put("lake1::cat2::", 6L);
+ cache.put("lake2::cat3::", 7L);
+
+ Assertions.assertEquals(7, cache.size());
+
+ // Drop catalog cat1 — should invalidate cat1 and all children
+ cache.invalidateByPrefix("lake1::cat1::");
+
+ Assertions.assertEquals(2, cache.size());
+ Assertions.assertFalse(cache.getIfPresent("lake1::cat1::").isPresent());
+
Assertions.assertFalse(cache.getIfPresent("lake1::cat1::s1::").isPresent());
+
Assertions.assertFalse(cache.getIfPresent("lake1::cat1::s1::t1::TABLE").isPresent());
+
Assertions.assertFalse(cache.getIfPresent("lake1::cat1::s1::t2::TABLE").isPresent());
+
Assertions.assertFalse(cache.getIfPresent("lake1::cat1::s2::").isPresent());
+
+ // cat2 and lake2 should be unaffected
+ Assertions.assertTrue(cache.getIfPresent("lake1::cat2::").isPresent());
+ Assertions.assertTrue(cache.getIfPresent("lake2::cat3::").isPresent());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateByPrefixLeaf() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("lake1::cat1::s1::t1::TABLE", 1L);
+ cache.put("lake1::cat1::s1::t2::TABLE", 2L);
+ cache.put("lake1::cat1::s1::f1::FILESET", 3L);
+
+ // Drop specific table — only t1 should be invalidated
+ cache.invalidateByPrefix("lake1::cat1::s1::t1::TABLE");
+
+ Assertions.assertEquals(2, cache.size());
+
Assertions.assertFalse(cache.getIfPresent("lake1::cat1::s1::t1::TABLE").isPresent());
+
Assertions.assertTrue(cache.getIfPresent("lake1::cat1::s1::t2::TABLE").isPresent());
+
Assertions.assertTrue(cache.getIfPresent("lake1::cat1::s1::f1::FILESET").isPresent());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineOverwrite() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("k", 1L);
+ Assertions.assertEquals(1L, cache.getIfPresent("k").get());
+
+ cache.put("k", 2L);
+ Assertions.assertEquals(2L, cache.getIfPresent("k").get());
+
+ Assertions.assertEquals(1, cache.size());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testNoOpsCache() {
+ NoOpsGravitinoCache<String, Long> cache = new NoOpsGravitinoCache<>();
+ try {
+ cache.put("key1", 100L);
+ Assertions.assertFalse(cache.getIfPresent("key1").isPresent());
+ Assertions.assertEquals(0, cache.size());
+
+ // All operations are no-ops, should not throw
+ cache.invalidate("key1");
+ cache.invalidateAll();
+ cache.invalidateByPrefix("any");
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineWithNonStringKeys() {
+ CaffeineGravitinoCache<Long, String> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put(1L, "role1");
+ cache.put(2L, "role2");
+ cache.put(3L, "role3");
+
+ Assertions.assertEquals("role1", cache.getIfPresent(1L).get());
+ Assertions.assertEquals(3, cache.size());
+
+ cache.invalidate(2L);
+ Assertions.assertFalse(cache.getIfPresent(2L).isPresent());
+ Assertions.assertEquals(2, cache.size());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateByPrefixIgnoresNonStringKeys() {
+ CaffeineGravitinoCache<Long, String> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put(10L, "role10");
+ cache.put(11L, "role11");
+
+ cache.invalidateByPrefix("1");
+
+ Assertions.assertEquals(2, cache.size());
+ Assertions.assertEquals("role10", cache.getIfPresent(10L).get());
+ Assertions.assertEquals("role11", cache.getIfPresent(11L).get());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineExpiresAfterWriteTtl() {
+ ManualTicker ticker = new ManualTicker();
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(50L, 1000L, ticker);
+ try {
+ cache.put("k", 1L);
+ Assertions.assertTrue(cache.getIfPresent("k").isPresent());
+
+ ticker.advance(51L, TimeUnit.MILLISECONDS);
+ Optional<Long> afterTtl = cache.getIfPresent("k");
+ Assertions.assertFalse(afterTtl.isPresent(), "Entry should have expired
after write TTL");
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineEvictsBeyondMaxSize() {
+ CaffeineGravitinoCache<Long, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 5L);
+ try {
+ for (long i = 0; i < 50L; i++) {
+ cache.put(i, i);
+ }
+
+ Awaitility.await()
+ .atMost(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ cache.cleanUp();
+ Assertions.assertTrue(
+ cache.size() <= 10L,
+ "Eviction should trim entries close to maxSize=5;
observed: " + cache.size());
+ });
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateByPrefixWithEmptyPrefixDropsAll() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("a", 1L);
+ cache.put("b", 2L);
+ cache.put("c", 3L);
+
+ cache.invalidateByPrefix("");
+ Assertions.assertEquals(0, cache.size());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateByPrefixNoMatch() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("lake1::cat1::", 1L);
+ cache.put("lake2::cat2::", 2L);
+
+ cache.invalidateByPrefix("missing-prefix::");
+
+ Assertions.assertEquals(
+ 2, cache.size(), "No keys should be removed when prefix has no
match");
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testCaffeineInvalidateNonExistentKey() {
+ CaffeineGravitinoCache<String, Long> cache = new
CaffeineGravitinoCache<>(60_000L, 1000L);
+ try {
+ cache.put("present", 1L);
+ // Should not throw or affect existing entries
+ cache.invalidate("absent");
+ Assertions.assertEquals(1, cache.size());
+ Assertions.assertTrue(cache.getIfPresent("present").isPresent());
+ } finally {
+ cache.close();
+ }
+ }
+
+ @Test
+ void testNoOpsCacheSizeAlwaysZero() {
+ NoOpsGravitinoCache<String, Long> cache = new NoOpsGravitinoCache<>();
+ try {
+ for (long i = 0; i < 100; i++) {
+ cache.put("k" + i, i);
+ }
+ Assertions.assertEquals(0, cache.size());
+ } finally {
+ cache.close();
+ }
+ }
+
+ private static class ManualTicker implements Ticker {
+ private final AtomicLong nanos = new AtomicLong();
+
+ @Override
+ public long read() {
+ return nanos.get();
+ }
+
+ private void advance(long time, TimeUnit unit) {
+ nanos.addAndGet(unit.toNanos(time));
+ }
+ }
+}
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java
index 439a5523fd..cb02de4aa1 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestAuthMappers.java
@@ -351,15 +351,16 @@ public class TestAuthMappers {
throw new RuntimeException("Update failed", e);
}
- List<ChangedOwnerInfo> changed = ownerMetaMapper.selectChangedOwners(50L);
+ List<ChangedOwnerInfo> changed = ownerMetaMapper.selectChangedOwners(0L);
Assertions.assertEquals(1, changed.size());
Assertions.assertEquals(200L, changed.get(0).getMetadataObjectId());
Assertions.assertEquals("SCHEMA", changed.get(0).getMetadataObjectType());
Assertions.assertEquals(100L, changed.get(0).getUpdatedAt());
- // With the same timestamp, the row is returned again for timestamp-only
polling.
- List<ChangedOwnerInfo> sameTimestamp =
ownerMetaMapper.selectChangedOwners(100L);
- Assertions.assertEquals(1, sameTimestamp.size());
+ // Polling after the last seen id should not return the same row again.
+ List<ChangedOwnerInfo> sameTimestamp =
+ ownerMetaMapper.selectChangedOwners(changed.get(0).getId());
+ Assertions.assertTrue(sameTimestamp.isEmpty());
}
private AuditInfo buildAuditInfo() {
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
index 8b4a3be8c2..8893a74116 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/mapper/provider/base/TestEntityChangeLogMapper.java
@@ -106,8 +106,7 @@ public class TestEntityChangeLogMapper {
"metalake1", "TABLE", "metalake1.cat.schema.tbl", OperateType.ALTER);
long jvmAfter = System.currentTimeMillis();
- List<EntityChangeRecord> records =
- entityChangeLogMapper.selectEntityChanges(jvmBefore - 1000L, 10);
+ List<EntityChangeRecord> records =
entityChangeLogMapper.selectEntityChanges(0L, 10);
Assertions.assertEquals(1, records.size());
EntityChangeRecord record = records.get(0);
Assertions.assertEquals("metalake1", record.getMetalakeName());
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
index d605400f41..7d85ea7f99 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestEntityChangeLogService.java
@@ -47,19 +47,24 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
private static final String CATALOG_NAME =
"catalog_for_entity_change_log_test";
private static final String SCHEMA_NAME =
"schema_for_entity_change_log_test";
- private List<EntityChangeRecord> listEntityChanges(long createdAtFrom) {
+ private long maxEntityChangeId() {
return SessionUtils.doWithCommitAndFetchResult(
- EntityChangeLogMapper.class, mapper ->
mapper.selectEntityChanges(createdAtFrom, 100));
+ EntityChangeLogMapper.class, EntityChangeLogMapper::selectMaxChangeId);
+ }
+
+ private List<EntityChangeRecord> listEntityChanges(long lastConsumedId) {
+ return SessionUtils.doWithCommitAndFetchResult(
+ EntityChangeLogMapper.class, mapper ->
mapper.selectEntityChanges(lastConsumedId, 100));
}
private void assertEntityChange(
- long createdAtFrom,
+ long lastConsumedId,
String metalakeName,
Entity.EntityType entityType,
String fullName,
OperateType operateType) {
Assertions.assertTrue(
- listEntityChanges(createdAtFrom).stream()
+ listEntityChanges(lastConsumedId).stream()
.anyMatch(
record ->
record.getMetalakeName().equals(metalakeName)
@@ -73,7 +78,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
void testMetalakeChangeLogOnRenameAndDrop() throws IOException {
BaseMetalake metalake = createAndInsertMakeLake(METALAKE_NAME);
- long beforeRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeRename = maxEntityChangeId();
BaseMetalake renamedMetalake =
backend.update(
metalake.nameIdentifier(),
@@ -82,13 +87,17 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
createBaseMakeLake(
metalake.id(), METALAKE_NAME + "_renamed",
metalake.auditInfo()));
assertEntityChange(
- beforeRename, METALAKE_NAME, Entity.EntityType.METALAKE,
METALAKE_NAME, OperateType.ALTER);
+ maxIdBeforeRename,
+ METALAKE_NAME,
+ Entity.EntityType.METALAKE,
+ METALAKE_NAME,
+ OperateType.ALTER);
- long beforeDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeDrop = maxEntityChangeId();
Assertions.assertTrue(
MetalakeMetaService.getInstance().deleteMetalake(renamedMetalake.nameIdentifier(),
false));
assertEntityChange(
- beforeDrop,
+ maxIdBeforeDrop,
renamedMetalake.name(),
Entity.EntityType.METALAKE,
renamedMetalake.name(),
@@ -100,7 +109,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
createAndInsertMakeLake(METALAKE_NAME);
CatalogEntity catalog = createAndInsertCatalog(METALAKE_NAME,
CATALOG_NAME);
- long beforeCatalogRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeCatalogRename = maxEntityChangeId();
CatalogEntity renamedCatalog =
backend.update(
catalog.nameIdentifier(),
@@ -109,17 +118,17 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
createCatalog(
catalog.id(), catalog.namespace(), CATALOG_NAME +
"_renamed", AUDIT_INFO));
assertEntityChange(
- beforeCatalogRename,
+ maxIdBeforeCatalogRename,
METALAKE_NAME,
Entity.EntityType.CATALOG,
NameIdentifierUtil.ofCatalog(METALAKE_NAME, CATALOG_NAME).toString(),
OperateType.ALTER);
- long beforeCatalogDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeCatalogDrop = maxEntityChangeId();
Assertions.assertTrue(
CatalogMetaService.getInstance().deleteCatalog(renamedCatalog.nameIdentifier(),
false));
assertEntityChange(
- beforeCatalogDrop,
+ maxIdBeforeCatalogDrop,
METALAKE_NAME,
Entity.EntityType.CATALOG,
NameIdentifierUtil.ofCatalog(METALAKE_NAME,
renamedCatalog.name()).toString(),
@@ -127,7 +136,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
CatalogEntity schemaCatalog = createAndInsertCatalog(METALAKE_NAME,
CATALOG_NAME + "_schema");
SchemaEntity schema = createAndInsertSchema(METALAKE_NAME,
schemaCatalog.name(), SCHEMA_NAME);
- long beforeSchemaRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeSchemaRename = maxEntityChangeId();
SchemaEntity renamedSchema =
backend.update(
schema.nameIdentifier(),
@@ -136,17 +145,17 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
createSchemaEntity(
schema.id(), schema.namespace(), SCHEMA_NAME + "_renamed",
AUDIT_INFO));
assertEntityChange(
- beforeSchemaRename,
+ maxIdBeforeSchemaRename,
METALAKE_NAME,
Entity.EntityType.SCHEMA,
NameIdentifierUtil.ofSchema(METALAKE_NAME, schemaCatalog.name(),
SCHEMA_NAME).toString(),
OperateType.ALTER);
- long beforeSchemaDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeSchemaDrop = maxEntityChangeId();
Assertions.assertTrue(
SchemaMetaService.getInstance().deleteSchema(renamedSchema.nameIdentifier(),
false));
assertEntityChange(
- beforeSchemaDrop,
+ maxIdBeforeSchemaDrop,
METALAKE_NAME,
Entity.EntityType.SCHEMA,
NameIdentifierUtil.ofSchema(METALAKE_NAME, schemaCatalog.name(),
renamedSchema.name())
@@ -162,7 +171,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
TableEntity table =
createTableEntity(RandomIdGenerator.INSTANCE.nextId(), namespace,
"table1", AUDIT_INFO);
backend.insert(table, false);
- long beforeTableRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeTableRename = maxEntityChangeId();
TableEntity renamedTable =
TableMetaService.getInstance()
.updateTable(
@@ -170,17 +179,17 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
entity ->
createTableEntity(table.id(), table.namespace(), "table2",
table.auditInfo()));
assertEntityChange(
- beforeTableRename,
+ maxIdBeforeTableRename,
METALAKE_NAME,
Entity.EntityType.TABLE,
NameIdentifierUtil.ofTable(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"table1").toString(),
OperateType.ALTER);
- long beforeTableDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeTableDrop = maxEntityChangeId();
Assertions.assertTrue(
TableMetaService.getInstance().deleteTable(renamedTable.nameIdentifier()));
assertEntityChange(
- beforeTableDrop,
+ maxIdBeforeTableDrop,
METALAKE_NAME,
Entity.EntityType.TABLE,
NameIdentifierUtil.ofTable(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"table2").toString(),
@@ -193,24 +202,24 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
"topic1",
AUDIT_INFO);
backend.insert(topic, false);
- long beforeTopicRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeTopicRename = maxEntityChangeId();
TopicEntity renamedTopic =
backend.update(
topic.nameIdentifier(),
Entity.EntityType.TOPIC,
entity -> createTopicEntity(topic.id(), topic.namespace(),
"topic2", AUDIT_INFO));
assertEntityChange(
- beforeTopicRename,
+ maxIdBeforeTopicRename,
METALAKE_NAME,
Entity.EntityType.TOPIC,
NameIdentifierUtil.ofTopic(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"topic1").toString(),
OperateType.ALTER);
- long beforeTopicDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeTopicDrop = maxEntityChangeId();
Assertions.assertTrue(
TopicMetaService.getInstance().deleteTopic(renamedTopic.nameIdentifier()));
assertEntityChange(
- beforeTopicDrop,
+ maxIdBeforeTopicDrop,
METALAKE_NAME,
Entity.EntityType.TOPIC,
NameIdentifierUtil.ofTopic(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"topic2").toString(),
@@ -222,7 +231,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
NamespaceUtil.ofView(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME),
"view1");
ViewMetaService.getInstance().insertView(view, false);
- long beforeViewRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeViewRename = maxEntityChangeId();
ViewEntity renamedView =
ViewMetaService.getInstance()
.updateView(
@@ -237,16 +246,16 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
.withAuditInfo(view.auditInfo())
.build());
assertEntityChange(
- beforeViewRename,
+ maxIdBeforeViewRename,
METALAKE_NAME,
Entity.EntityType.VIEW,
NameIdentifierUtil.ofView(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"view1").toString(),
OperateType.ALTER);
- long beforeViewDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeViewDrop = maxEntityChangeId();
Assertions.assertTrue(ViewMetaService.getInstance().deleteView(renamedView.nameIdentifier()));
assertEntityChange(
- beforeViewDrop,
+ maxIdBeforeViewDrop,
METALAKE_NAME,
Entity.EntityType.VIEW,
NameIdentifierUtil.ofView(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"view2").toString(),
@@ -259,7 +268,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
"fileset1",
AUDIT_INFO);
FilesetMetaService.getInstance().insertFileset(fileset, false);
- long beforeFilesetRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeFilesetRename = maxEntityChangeId();
FilesetEntity renamedFileset =
FilesetMetaService.getInstance()
.updateFileset(
@@ -267,18 +276,18 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
entity ->
createFilesetEntity(fileset.id(), fileset.namespace(),
"fileset2", AUDIT_INFO));
assertEntityChange(
- beforeFilesetRename,
+ maxIdBeforeFilesetRename,
METALAKE_NAME,
Entity.EntityType.FILESET,
NameIdentifierUtil.ofFileset(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"fileset1")
.toString(),
OperateType.ALTER);
- long beforeFilesetDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeFilesetDrop = maxEntityChangeId();
Assertions.assertTrue(
FilesetMetaService.getInstance().deleteFileset(renamedFileset.nameIdentifier()));
assertEntityChange(
- beforeFilesetDrop,
+ maxIdBeforeFilesetDrop,
METALAKE_NAME,
Entity.EntityType.FILESET,
NameIdentifierUtil.ofFileset(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"fileset2")
@@ -295,7 +304,7 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
Map.of("k1", "v1"),
AUDIT_INFO);
ModelMetaService.getInstance().insertModel(model, false);
- long beforeModelRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeModelRename = maxEntityChangeId();
ModelEntity renamedModel =
ModelMetaService.getInstance()
.updateModel(
@@ -310,17 +319,17 @@ public class TestEntityChangeLogService extends
TestJDBCBackend {
model.properties(),
AUDIT_INFO));
assertEntityChange(
- beforeModelRename,
+ maxIdBeforeModelRename,
METALAKE_NAME,
Entity.EntityType.MODEL,
NameIdentifierUtil.ofModel(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"model1").toString(),
OperateType.ALTER);
- long beforeModelDrop = System.currentTimeMillis() - 1;
+ long maxIdBeforeModelDrop = maxEntityChangeId();
Assertions.assertTrue(
ModelMetaService.getInstance().deleteModel(renamedModel.nameIdentifier()));
assertEntityChange(
- beforeModelDrop,
+ maxIdBeforeModelDrop,
METALAKE_NAME,
Entity.EntityType.MODEL,
NameIdentifierUtil.ofModel(METALAKE_NAME, CATALOG_NAME, SCHEMA_NAME,
"model2").toString(),
diff --git
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
index 6596b4af2c..08d874f295 100644
---
a/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
+++
b/core/src/test/java/org/apache/gravitino/storage/relational/service/TestTableMetaService.java
@@ -69,9 +69,14 @@ public class TestTableMetaService extends TestJDBCBackend {
private final String catalogName = "catalog_for_table_test";
private final String schemaName = "schema_for_table_test";
- private List<EntityChangeRecord> listEntityChanges(long createdAtFrom) {
+ private long maxEntityChangeId() {
return SessionUtils.doWithCommitAndFetchResult(
- EntityChangeLogMapper.class, mapper ->
mapper.selectEntityChanges(createdAtFrom, 100));
+ EntityChangeLogMapper.class, EntityChangeLogMapper::selectMaxChangeId);
+ }
+
+ private List<EntityChangeRecord> listEntityChanges(long lastConsumedId) {
+ return SessionUtils.doWithCommitAndFetchResult(
+ EntityChangeLogMapper.class, mapper ->
mapper.selectEntityChanges(lastConsumedId, 100));
}
@TestTemplate
@@ -201,7 +206,7 @@ public class TestTableMetaService extends TestJDBCBackend {
TableMetaService.getInstance().insertTable(createdTable, false);
// test update table without changing schema name
- long beforeRename = System.currentTimeMillis() - 1;
+ long maxIdBeforeRename = maxEntityChangeId();
TableEntity updatedTable =
TableEntity.builder()
.withId(createdTable.id())
@@ -222,7 +227,7 @@ public class TestTableMetaService extends TestJDBCBackend {
compareTwoColumns(updatedTable.columns(), retrievedTable.columns());
compareTwoColumns(updatedTable.columns(), retrievedTable.columns());
Assertions.assertTrue(
- listEntityChanges(beforeRename).stream()
+ listEntityChanges(maxIdBeforeRename).stream()
.anyMatch(
record ->
record.getMetalakeName().equals(metalakeName)
@@ -263,7 +268,7 @@ public class TestTableMetaService extends TestJDBCBackend {
AUDIT_INFO);
backend.insert(newSchema, false);
- long beforeSchemaMove = System.currentTimeMillis() - 1;
+ long maxIdBeforeSchemaMove = maxEntityChangeId();
TableEntity movedTable =
TableEntity.builder()
.withId(updatedTable.id())
@@ -275,7 +280,7 @@ public class TestTableMetaService extends TestJDBCBackend {
TableMetaService.getInstance()
.updateTable(updatedTable.nameIdentifier(), oldTable -> movedTable);
Assertions.assertTrue(
- listEntityChanges(beforeSchemaMove).stream()
+ listEntityChanges(maxIdBeforeSchemaMove).stream()
.anyMatch(
record ->
record.getMetalakeName().equals(metalakeName)
@@ -298,11 +303,11 @@ public class TestTableMetaService extends TestJDBCBackend
{
Assertions.assertEquals(updatedTable2.auditInfo(),
retrievedTable2.auditInfo());
compareTwoColumns(updatedTable2.columns(), retrievedTable2.columns());
- long beforeDelete = System.currentTimeMillis() - 1;
+ long maxIdBeforeDelete = maxEntityChangeId();
Assertions.assertTrue(
TableMetaService.getInstance().deleteTable(updatedTable2.nameIdentifier()));
Assertions.assertTrue(
- listEntityChanges(beforeDelete).stream()
+ listEntityChanges(maxIdBeforeDelete).stream()
.anyMatch(
record ->
record.getMetalakeName().equals(metalakeName)
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/PassThroughAuthorizer.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/PassThroughAuthorizer.java
index 06e20b8ada..76de95b0d3 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/PassThroughAuthorizer.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/PassThroughAuthorizer.java
@@ -79,7 +79,7 @@ public class PassThroughAuthorizer implements
GravitinoAuthorizer {
}
@Override
- public boolean isMetalakeUser(String metalake) {
+ public boolean isMetalakeUser(String metalake, AuthorizationRequestContext
requestContext) {
AccessControlDispatcher dispatcher =
GravitinoEnv.getInstance().accessControlDispatcher();
if (dispatcher != null) {
try {
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConstants.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConstants.java
index 6e72401720..bdb13b8ba8 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConstants.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConstants.java
@@ -177,7 +177,8 @@ public class AuthorizationExpressionConstants {
* Special case: "METALAKE_USER" is used here as a unique authorization
token, not a logical
* expression like other constants. This is intentional and required for
metalake-level user
* authorization checks {@link
- *
org.apache.gravitino.authorization.GravitinoAuthorizer#isMetalakeUser(String)}.
+ *
org.apache.gravitino.authorization.GravitinoAuthorizer#isMetalakeUser(String,
+ * org.apache.gravitino.authorization.AuthorizationRequestContext)}.
*/
public static final String LOAD_METALAKE_AUTHORIZATION_EXPRESSION =
"METALAKE_USER";
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConverter.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConverter.java
index 99123a162f..4507ba0a50 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConverter.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/expression/AuthorizationExpressionConverter.java
@@ -212,7 +212,9 @@ public class AuthorizationExpressionConverter {
*/
public static String replaceAnyPrivilege(String expression) {
expression = expression.replaceAll("SERVICE_ADMIN",
"authorizer.isServiceAdmin()");
- expression = expression.replaceAll("METALAKE_USER",
"authorizer.isMetalakeUser(METALAKE_NAME)");
+ expression =
+ expression.replaceAll(
+ "METALAKE_USER",
"authorizer.isMetalakeUser(METALAKE_NAME,authorizationContext)");
// A single privilege (e.g., SELECT_TABLE) can be granted or denied at
multiple namespace
// levels: metalake, catalog, schema, and table.
diff --git
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
index 2e1f669b8f..3778f8b19f 100644
---
a/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
+++
b/server-common/src/main/java/org/apache/gravitino/server/authorization/jcasbin/JcasbinAuthorizer.java
@@ -252,7 +252,7 @@ public class JcasbinAuthorizer implements
GravitinoAuthorizer {
}
@Override
- public boolean isMetalakeUser(String metalake) {
+ public boolean isMetalakeUser(String metalake, AuthorizationRequestContext
requestContext) {
String currentUserName = PrincipalUtils.getCurrentUserName();
if (StringUtils.isBlank(currentUserName)) {
return false;
diff --git
a/server-common/src/test/java/org/apache/gravitino/server/authorization/MockGravitinoAuthorizer.java
b/server-common/src/test/java/org/apache/gravitino/server/authorization/MockGravitinoAuthorizer.java
index f9ef64bab0..51c55ed797 100644
---
a/server-common/src/test/java/org/apache/gravitino/server/authorization/MockGravitinoAuthorizer.java
+++
b/server-common/src/test/java/org/apache/gravitino/server/authorization/MockGravitinoAuthorizer.java
@@ -93,7 +93,7 @@ public class MockGravitinoAuthorizer implements
GravitinoAuthorizer {
}
@Override
- public boolean isMetalakeUser(String metalake) {
+ public boolean isMetalakeUser(String metalake, AuthorizationRequestContext
requestContext) {
return true;
}
diff --git
a/server-common/src/test/java/org/apache/gravitino/server/authorization/TestPassThroughAuthorizer.java
b/server-common/src/test/java/org/apache/gravitino/server/authorization/TestPassThroughAuthorizer.java
index f5290d01fa..e173491d6a 100644
---
a/server-common/src/test/java/org/apache/gravitino/server/authorization/TestPassThroughAuthorizer.java
+++
b/server-common/src/test/java/org/apache/gravitino/server/authorization/TestPassThroughAuthorizer.java
@@ -60,7 +60,8 @@ public class TestPassThroughAuthorizer {
passThroughAuthorizer.isOwner(
principal, "metalake", metadataObject, new
AuthorizationRequestContext()));
Assertions.assertTrue(passThroughAuthorizer.isServiceAdmin());
- Assertions.assertTrue(passThroughAuthorizer.isMetalakeUser("metalake"));
+ Assertions.assertTrue(
+ passThroughAuthorizer.isMetalakeUser("metalake", new
AuthorizationRequestContext()));
Assertions.assertTrue(passThroughAuthorizer.isSelf(Entity.EntityType.USER,
null));
Assertions.assertTrue(
passThroughAuthorizer.hasSetOwnerPermission(
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
index 5dc0057fcd..0f542bdeb9 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/GravitinoInterceptionService.java
@@ -40,6 +40,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.gravitino.Entity;
import org.apache.gravitino.MetadataObject;
import org.apache.gravitino.NameIdentifier;
+import org.apache.gravitino.authorization.AuthorizationRequestContext;
import org.apache.gravitino.authorization.AuthorizationUtils;
import org.apache.gravitino.exceptions.ForbiddenException;
import org.apache.gravitino.exceptions.NoSuchMetalakeException;
@@ -153,13 +154,16 @@ public class GravitinoInterceptionService implements
InterceptionService {
extractNameIdentifierFromParameters(parameters, args);
Map<String, Object> pathParams =
Utils.extractPathParamsFromParameters(parameters, args);
+ AuthorizationRequestContext authorizationRequestContext =
+ new AuthorizationRequestContext();
// Check metalake and user existence before authorization
NameIdentifier metalakeIdent =
metadataContext.get(Entity.EntityType.METALAKE);
if (metalakeIdent != null) {
String currentUser = PrincipalUtils.getCurrentUserName();
try {
- AuthorizationUtils.checkCurrentUser(metalakeIdent.name(),
currentUser);
+ AuthorizationUtils.checkCurrentUser(
+ metalakeIdent.name(), currentUser,
authorizationRequestContext);
} catch (NoSuchMetalakeException e) {
LOG.warn(
"Metalake {} does not exist when validating user {}",
metalakeIdent, currentUser);
@@ -199,7 +203,7 @@ public class GravitinoInterceptionService implements
InterceptionService {
args,
secondaryExpression,
secondaryExpressionCondition);
- boolean authorizeResult = executor.execute();
+ boolean authorizeResult =
executor.execute(authorizationRequestContext);
if (!authorizeResult) {
return buildNoAuthResponse(expressionAnnotation,
metadataContext, method, expression);
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociatePolicyAuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociatePolicyAuthorizationExecutor.java
index d9f4de8805..59b2e83305 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociatePolicyAuthorizationExecutor.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociatePolicyAuthorizationExecutor.java
@@ -54,13 +54,12 @@ public class AssociatePolicyAuthorizationExecutor extends
CommonAuthorizerExecut
}
@Override
- public boolean execute() throws Exception {
+ public boolean execute(AuthorizationRequestContext context) throws Exception
{
Object request = extractFromParameters(parameters, args);
if (request == null) {
return false;
}
- AuthorizationRequestContext context = new AuthorizationRequestContext();
context.setOriginalAuthorizationExpression(expression);
Entity.EntityType targetType =
Entity.EntityType.POLICY; // policies are the only supported batch
target here
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociateTagAuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociateTagAuthorizationExecutor.java
index 8d5080e800..2fef4e9ee3 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociateTagAuthorizationExecutor.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AssociateTagAuthorizationExecutor.java
@@ -54,13 +54,12 @@ public class AssociateTagAuthorizationExecutor extends
CommonAuthorizerExecutor
}
@Override
- public boolean execute() throws Exception {
+ public boolean execute(AuthorizationRequestContext context) throws Exception
{
Object request = extractFromParameters(parameters, args);
if (request == null) {
return false;
}
- AuthorizationRequestContext context = new AuthorizationRequestContext();
context.setOriginalAuthorizationExpression(expression);
Entity.EntityType targetType =
Entity.EntityType.TAG; // Tags are the only supported batch target here
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
index be9c78f96e..d98d448a10 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/AuthorizationExecutor.java
@@ -17,7 +17,9 @@
package org.apache.gravitino.server.web.filter.authorization;
+import org.apache.gravitino.authorization.AuthorizationRequestContext;
+
public interface AuthorizationExecutor {
- boolean execute() throws Exception;
+ boolean execute(AuthorizationRequestContext authorizationRequestContext)
throws Exception;
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CommonAuthorizerExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CommonAuthorizerExecutor.java
index b03bdeaa58..59d9bcc5eb 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CommonAuthorizerExecutor.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/CommonAuthorizerExecutor.java
@@ -45,10 +45,9 @@ public class CommonAuthorizerExecutor implements
AuthorizationExecutor {
}
@Override
- public boolean execute() throws Exception {
- AuthorizationRequestContext authorizationRequestContext = new
AuthorizationRequestContext();
+ public boolean execute(AuthorizationRequestContext
authorizationRequestContext) throws Exception {
authorizationRequestContext.setOriginalAuthorizationExpression(expression);
return authorizationExpressionEvaluator.evaluate(
- metadataContext, pathParams, new AuthorizationRequestContext(),
entityType);
+ metadataContext, pathParams, authorizationRequestContext, entityType);
}
}
diff --git
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/RunJobAuthorizationExecutor.java
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/RunJobAuthorizationExecutor.java
index 55a15835d0..7e592408ed 100644
---
a/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/RunJobAuthorizationExecutor.java
+++
b/server/src/main/java/org/apache/gravitino/server/web/filter/authorization/RunJobAuthorizationExecutor.java
@@ -36,6 +36,7 @@ public class RunJobAuthorizationExecutor implements
AuthorizationExecutor {
private final AuthorizationExpressionEvaluator
authorizationExpressionEvaluator;
private final Map<String, Object> pathParams;
private final Optional<String> entityType;
+ private final String expression;
public RunJobAuthorizationExecutor(
Parameter[] parameters,
@@ -46,6 +47,7 @@ public class RunJobAuthorizationExecutor implements
AuthorizationExecutor {
Optional<String> entityType) {
this.parameters = parameters;
this.args = args;
+ this.expression = expression;
this.metadataContext = metadataContext;
this.authorizationExpressionEvaluator = new
AuthorizationExpressionEvaluator(expression);
this.pathParams = pathParams;
@@ -53,13 +55,13 @@ public class RunJobAuthorizationExecutor implements
AuthorizationExecutor {
}
@Override
- public boolean execute() throws Exception {
+ public boolean execute(AuthorizationRequestContext context) throws Exception
{
Object request = extractFromParameters(parameters, args);
if (request == null) {
return false;
}
- AuthorizationRequestContext context = new AuthorizationRequestContext();
+ context.setOriginalAuthorizationExpression(expression);
Preconditions.checkArgument(
request instanceof JobRunRequest,
"Expected JobRunRequest but found %s",
diff --git
a/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
b/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
index a96160da84..02d6048c95 100644
---
a/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
+++
b/server/src/test/java/org/apache/gravitino/server/web/filter/TestGravitinoInterceptionService.java
@@ -170,7 +170,7 @@ public class TestGravitinoInterceptionService {
.when(
() ->
AuthorizationUtils.checkCurrentUser(
- ArgumentMatchers.any(), ArgumentMatchers.any()))
+ ArgumentMatchers.any(), ArgumentMatchers.any(),
ArgumentMatchers.any()))
.thenThrow(new NoSuchMetalakeException("Metalake nonExistentMetalake
does not exist"));
GravitinoInterceptionService gravitinoInterceptionService =
@@ -318,7 +318,7 @@ public class TestGravitinoInterceptionService {
}
@Override
- public boolean isMetalakeUser(String metalake) {
+ public boolean isMetalakeUser(String metalake, AuthorizationRequestContext
requestContext) {
return true;
}
