pingtimeout commented on code in PR #1339:
URL: https://github.com/apache/polaris/pull/1339#discussion_r2073407737
##########
polaris-core/src/main/java/org/apache/polaris/core/persistence/cache/InMemoryEntityCache.java:
##########
@@ -20,194 +20,99 @@
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
-import com.github.benmanes.caffeine.cache.RemovalListener;
+import com.github.benmanes.caffeine.cache.RemovalCause;
+import com.google.common.collect.HashBiMap;
+import com.google.common.util.concurrent.MoreExecutors;
import jakarta.annotation.Nonnull;
import jakarta.annotation.Nullable;
-import java.util.AbstractMap;
-import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import net.jcip.annotations.GuardedBy;
import org.apache.polaris.core.PolarisCallContext;
import org.apache.polaris.core.config.BehaviorChangeConfiguration;
import org.apache.polaris.core.config.FeatureConfiguration;
import org.apache.polaris.core.config.PolarisConfiguration;
import org.apache.polaris.core.entity.PolarisBaseEntity;
import org.apache.polaris.core.entity.PolarisEntityType;
-import org.apache.polaris.core.entity.PolarisGrantRecord;
import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
-/** An in-memory entity cache with a limit of 100k entities and a 1h TTL. */
+/** The entity cache, can be private or shared */
public class InMemoryEntityCache implements EntityCache {
-
// cache mode
private EntityCacheMode cacheMode;
// the meta store manager
private final PolarisMetaStoreManager polarisMetaStoreManager;
- // Caffeine cache to keep entries by id
- private final Cache<Long, ResolvedPolarisEntity> byId;
+ // Caffeine cache to keep entries
+ @GuardedBy("lock")
+ private final Cache<Long, ResolvedPolarisEntity> cache;
+
+ // Map of entity names to entity ids
+ @GuardedBy("lock")
+ private final HashBiMap<Long, EntityCacheByNameKey> nameToIdMap;
- // index by name
- private final AbstractMap<EntityCacheByNameKey, ResolvedPolarisEntity>
byName;
+ // Locks to ensure that an entity can only be refreshed by one thread at a
time
+ private final ReentrantReadWriteLock lock;
/**
* Constructor. Cache can be private or shared
*
* @param polarisMetaStoreManager the meta store manager implementation
*/
public InMemoryEntityCache(@Nonnull PolarisMetaStoreManager
polarisMetaStoreManager) {
+ // remember the meta store manager
+ this.polarisMetaStoreManager = polarisMetaStoreManager;
- // by name cache
- this.byName = new ConcurrentHashMap<>();
-
- // When an entry is removed, we simply remove it from the byName map
- RemovalListener<Long, ResolvedPolarisEntity> removalListener =
- (key, value, cause) -> {
- if (value != null) {
- // compute name key
- EntityCacheByNameKey nameKey = new
EntityCacheByNameKey(value.getEntity());
+ // enabled by default
+ this.cacheMode = EntityCacheMode.ENABLE;
- // if it is still active, remove it from the name key
- this.byName.remove(nameKey, value);
- }
- };
+ this.nameToIdMap = HashBiMap.create();
+ this.lock = new ReentrantReadWriteLock();
long weigherTarget =
PolarisConfiguration.loadConfig(FeatureConfiguration.ENTITY_CACHE_WEIGHER_TARGET);
- Caffeine<Long, ResolvedPolarisEntity> byIdBuilder =
+ Caffeine<Long, ResolvedPolarisEntity> cacheBuilder =
Caffeine.newBuilder()
.maximumWeight(weigherTarget)
.weigher(EntityWeigher.asWeigher())
.expireAfterAccess(1, TimeUnit.HOURS) // Expire entries after 1
hour of no access
- .removalListener(removalListener); // Set the removal listener
+ // Add a removal listener so that the name-to-id mapping is also
purged after every
+ // cache
+ // replacement, invalidation or expiration . Configure a direct
executor so that the
+ // removal
+ // listener is invoked by the same thread that is updating the
cache.
Review Comment:
Not really, but only thanks to the direct executor usage:
1. The Caffeine cache entry removal happens while holding the write lock.
That is, during the `cache.invalidate(...)` method call.
2. The Caffeine cache does invoke the removal listener asynchronously. I.e.
during the call to replace/remove an entity, it schedules a call to the removal
listener on a separate executor. This will indeed cause the listener to be
executed on a different thread, which could lead to the race condition you
describe.
3. However, by instructing Caffeine to use a direct executor, the listener
is called synchronously by the calling thread before the replace/remove entity
call even returns. I.e. when Caffeine calls `executor.submit(...)`, the direct
executor transforms that into a synchronous method call.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
