yuqi1129 commented on code in PR #13106: URL: https://github.com/apache/gravitino/pull/13106#discussion_r4001847093
########## core/src/main/java/org/apache/gravitino/cache/RedisEntityCache.java: ########## @@ -0,0 +1,630 @@ +/* + * 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.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; +import redis.clients.jedis.commands.KeyCommands; +import redis.clients.jedis.commands.SortedSetCommands; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * An {@link EntityCache} that keeps one copy of every cached entity in Redis, shared by all nodes + * of a Gravitino cluster. It is selected with {@code gravitino.cache.implementation=redis} and + * reports {@link Coherence#SHARED}: a write on any node invalidates the single shared copy, so no + * per-node propagation is needed. + * + * <p>This is a shared cache, not a strongly consistent database and cache pair. The cache is only + * touched after the entity store has committed, and an invalidation deletes the entry rather than + * updating it in place, so a Redis failure degrades to a cache miss and never to a stale hit. What + * remains is the window between a store commit and the invalidation that follows it: a node that + * dies inside that window leaves the entry readable until its TTL expires. Callers that cannot + * tolerate that must read the store directly. + * + * <p><b>Keyspace.</b> See {@link RedisKeyspace}. All keys of one metalake share a Redis Cluster + * hash slot, so every script here is single-slot at any depth; the cost is that one metalake's + * entries are bounded by one cluster node. + * + * <p><b>Container drop.</b> {@link #invalidate(NameIdentifier, Entity.EntityType)} runs one Lua + * script that bumps the version fence of the dropped identifier, deletes its value, and walks the + * lexicographic index range of its descendants deleting each one. The script is atomic, so a + * concurrent reader sees the subtree either complete or gone, never half dropped. + * + * <p><b>Stale-write guard.</b> A read miss records the current fence of the identifier and of each + * of its ancestors for the calling thread. The write that fills the entry afterwards is a Lua + * script that compares those fences again and refuses to write if any has moved. A load that began + * before a drop of the entity, or of any container above it, therefore cannot refill the key once + * the drop has committed, whether or not the entity was indexed at the time. Fences outlive the + * value TTL (see {@code gravitino.cache.redis.fenceTtlMs}), so a slow reader cannot win by + * outliving the value. A write with no recorded fences, such as caching a freshly inserted entity, + * is written unconditionally. + * + * <p><b>Failure policy.</b> Reads and fills are optimizations: a Redis error or timeout makes them + * a miss or a no-op. An invalidation is a correctness obligation: a failure is propagated as a + * {@link RuntimeException} so it is never silently dropped, and the entry expires by TTL at the + * latest. An unreachable Redis at startup fails fast. + */ +public class RedisEntityCache extends BaseEntityCache { + + private static final Logger LOG = LoggerFactory.getLogger(RedisEntityCache.class); + + /** Per-thread bound on recorded fences awaiting the write that consumes them. */ + private static final int MAX_PENDING_FENCES = 1024; + + private static final long FAILURE_LOG_INTERVAL_MS = 30_000L; + private static final int SCAN_BATCH = 500; + + /** + * Reads a value, or on a miss the fences guarding a later fill. {@code KEYS[1]} is the value key; + * {@code ARGV} lists the fence keys. Returns {@code {1, value}} on a hit and {@code {0, + * fence...}} on a miss, with an absent fence reported as {@code "0"}. + */ + @VisibleForTesting + static final String READ_SCRIPT = + "local v = redis.call('GET', KEYS[1])\n" + + "if v then return {1, v} end\n" + + "local r = {0}\n" + + "for i = 1, #ARGV do r[#r + 1] = redis.call('GET', ARGV[i]) or '0' end\n" + + "return r\n"; + + /** + * Writes a value unless a guarding fence moved. {@code KEYS[1]} is the index key and {@code + * KEYS[2]} the value key; {@code ARGV[1]} is the index member, {@code ARGV[2]} the value, {@code + * ARGV[3]} the TTL in milliseconds (0 for none), followed by (fence key, expected value) pairs. + * Returns 1 if written and 0 if rejected. + */ + @VisibleForTesting + static final String PUT_SCRIPT = + "for i = 4, #ARGV, 2 do\n" + + " if (redis.call('GET', ARGV[i]) or '0') ~= ARGV[i + 1] then return 0 end\n" + + "end\n" + + "if tonumber(ARGV[3]) > 0 then\n" + + " redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])\n" + + "else\n" + + " redis.call('SET', KEYS[2], ARGV[2])\n" + + "end\n" + + "redis.call('ZADD', KEYS[1], 0, ARGV[1])\n" + + "return 1\n"; + + /** + * Drops an entry and its indexed descendants. {@code KEYS[1]} is the index key; {@code ARGV[1]} + * is the slot prefix, {@code ARGV[2]} the index member, {@code ARGV[3]} the identifier whose + * fence to bump, {@code ARGV[4]} the fence TTL in milliseconds (0 for none), followed by the + * descendant prefixes to range over. The upper bound {@code prefix\xff} is above every UTF-8 + * member starting with the prefix. Returns the number of values deleted. + */ + @VisibleForTesting + static final String DROP_SCRIPT = + "local idx = KEYS[1]\n" + + "local prefix = ARGV[1]\n" + + "local fence = prefix .. 'F:' .. ARGV[3]\n" + + "redis.call('INCR', fence)\n" + + "if tonumber(ARGV[4]) > 0 then redis.call('PEXPIRE', fence, ARGV[4]) end\n" Review Comment: [P1] Expiring counters can reuse a fence version and admit stale fills `fenceTtlMs > valueTtlMs` does not prevent ABA. A reader can observe fence `1` just before expiry and load the old entity; the fence expires; another node commits an update and successfully invalidates, recreating the fence as `1`; the old reader then passes `1 == 1` and caches stale data. The read only needs to cross the expiration boundary, not last for the whole fence TTL. I reproduced this with value TTL 100 ms and fence TTL 200 ms. An initially absent fence also admits the `0 -> invalidation -> expiry -> 0` variant. Please prevent generations from disappearing or being reused while a fill can still be accepted. Keeping non-expiring generations is the simplest correctness fix; bounded retention needs an explicit safe lifetime/reclamation protocol. Increasing the TTL alone is insufficient. Add tests for both ABA sequences. ########## core/src/main/java/org/apache/gravitino/cache/RedisEntityCache.java: ########## @@ -0,0 +1,630 @@ +/* + * 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.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; +import redis.clients.jedis.commands.KeyCommands; +import redis.clients.jedis.commands.SortedSetCommands; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * An {@link EntityCache} that keeps one copy of every cached entity in Redis, shared by all nodes + * of a Gravitino cluster. It is selected with {@code gravitino.cache.implementation=redis} and + * reports {@link Coherence#SHARED}: a write on any node invalidates the single shared copy, so no + * per-node propagation is needed. + * + * <p>This is a shared cache, not a strongly consistent database and cache pair. The cache is only + * touched after the entity store has committed, and an invalidation deletes the entry rather than + * updating it in place, so a Redis failure degrades to a cache miss and never to a stale hit. What + * remains is the window between a store commit and the invalidation that follows it: a node that + * dies inside that window leaves the entry readable until its TTL expires. Callers that cannot + * tolerate that must read the store directly. + * + * <p><b>Keyspace.</b> See {@link RedisKeyspace}. All keys of one metalake share a Redis Cluster + * hash slot, so every script here is single-slot at any depth; the cost is that one metalake's + * entries are bounded by one cluster node. + * + * <p><b>Container drop.</b> {@link #invalidate(NameIdentifier, Entity.EntityType)} runs one Lua + * script that bumps the version fence of the dropped identifier, deletes its value, and walks the + * lexicographic index range of its descendants deleting each one. The script is atomic, so a + * concurrent reader sees the subtree either complete or gone, never half dropped. + * + * <p><b>Stale-write guard.</b> A read miss records the current fence of the identifier and of each + * of its ancestors for the calling thread. The write that fills the entry afterwards is a Lua + * script that compares those fences again and refuses to write if any has moved. A load that began + * before a drop of the entity, or of any container above it, therefore cannot refill the key once + * the drop has committed, whether or not the entity was indexed at the time. Fences outlive the + * value TTL (see {@code gravitino.cache.redis.fenceTtlMs}), so a slow reader cannot win by + * outliving the value. A write with no recorded fences, such as caching a freshly inserted entity, + * is written unconditionally. + * + * <p><b>Failure policy.</b> Reads and fills are optimizations: a Redis error or timeout makes them + * a miss or a no-op. An invalidation is a correctness obligation: a failure is propagated as a + * {@link RuntimeException} so it is never silently dropped, and the entry expires by TTL at the + * latest. An unreachable Redis at startup fails fast. + */ +public class RedisEntityCache extends BaseEntityCache { + + private static final Logger LOG = LoggerFactory.getLogger(RedisEntityCache.class); + + /** Per-thread bound on recorded fences awaiting the write that consumes them. */ + private static final int MAX_PENDING_FENCES = 1024; + + private static final long FAILURE_LOG_INTERVAL_MS = 30_000L; + private static final int SCAN_BATCH = 500; + + /** + * Reads a value, or on a miss the fences guarding a later fill. {@code KEYS[1]} is the value key; + * {@code ARGV} lists the fence keys. Returns {@code {1, value}} on a hit and {@code {0, + * fence...}} on a miss, with an absent fence reported as {@code "0"}. + */ + @VisibleForTesting + static final String READ_SCRIPT = + "local v = redis.call('GET', KEYS[1])\n" + + "if v then return {1, v} end\n" + + "local r = {0}\n" + + "for i = 1, #ARGV do r[#r + 1] = redis.call('GET', ARGV[i]) or '0' end\n" + + "return r\n"; + + /** + * Writes a value unless a guarding fence moved. {@code KEYS[1]} is the index key and {@code + * KEYS[2]} the value key; {@code ARGV[1]} is the index member, {@code ARGV[2]} the value, {@code + * ARGV[3]} the TTL in milliseconds (0 for none), followed by (fence key, expected value) pairs. + * Returns 1 if written and 0 if rejected. + */ + @VisibleForTesting + static final String PUT_SCRIPT = + "for i = 4, #ARGV, 2 do\n" + + " if (redis.call('GET', ARGV[i]) or '0') ~= ARGV[i + 1] then return 0 end\n" + + "end\n" + + "if tonumber(ARGV[3]) > 0 then\n" + + " redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])\n" + + "else\n" + + " redis.call('SET', KEYS[2], ARGV[2])\n" + + "end\n" + + "redis.call('ZADD', KEYS[1], 0, ARGV[1])\n" + + "return 1\n"; + + /** + * Drops an entry and its indexed descendants. {@code KEYS[1]} is the index key; {@code ARGV[1]} + * is the slot prefix, {@code ARGV[2]} the index member, {@code ARGV[3]} the identifier whose + * fence to bump, {@code ARGV[4]} the fence TTL in milliseconds (0 for none), followed by the + * descendant prefixes to range over. The upper bound {@code prefix\xff} is above every UTF-8 + * member starting with the prefix. Returns the number of values deleted. + */ + @VisibleForTesting + static final String DROP_SCRIPT = + "local idx = KEYS[1]\n" + + "local prefix = ARGV[1]\n" + + "local fence = prefix .. 'F:' .. ARGV[3]\n" + + "redis.call('INCR', fence)\n" + + "if tonumber(ARGV[4]) > 0 then redis.call('PEXPIRE', fence, ARGV[4]) end\n" + + "local removed = redis.call('DEL', prefix .. 'D:' .. ARGV[2])\n" + + "redis.call('ZREM', idx, ARGV[2])\n" + + "for i = 5, #ARGV do\n" + + " local members = redis.call('ZRANGEBYLEX', idx, '[' .. ARGV[i], " + + "'[' .. ARGV[i] .. '\\255')\n" + + " for _, m in ipairs(members) do\n" + + " removed = removed + redis.call('DEL', prefix .. 'D:' .. m)\n" + + " redis.call('ZREM', idx, m)\n" + + " end\n" + + "end\n" + + "return removed\n"; + + private final UnifiedJedis jedis; + private final RedisKeyspace keyspace; + private final KryoEntitySerializer serializer; + private final SegmentedLock segmentedLock; + private final long valueTtlMs; + private final long fenceTtlMs; + private final byte[] readScript; + private final byte[] putScript; + private final byte[] dropScript; + private final AtomicLong lastFailureLogMs = new AtomicLong(); + + /** + * Fences observed by a read miss on this thread, keyed by index member, consumed by the write + * that fills the entry. Bounded and access-ordered so an entry whose write never happens (the + * store had no such entity) is eventually evicted; a leftover entry can only make a later write + * more conservative, never less. + */ + private final ThreadLocal<Map<String, FenceSnapshot>> pendingFences = + ThreadLocal.withInitial( + () -> + new LinkedHashMap<String, FenceSnapshot>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, FenceSnapshot> eldest) { + return size() > MAX_PENDING_FENCES; + } + }); + + /** + * Constructs a new {@link RedisEntityCache} connected to the Redis deployment described by the + * {@code gravitino.cache.redis.*} configuration. + * + * @param cacheConfig the cache configuration + */ + public RedisEntityCache(Config cacheConfig) { + this(cacheConfig, createClient(cacheConfig)); + probe(cacheConfig.get(Configs.CACHE_REDIS_ADDRESS)); + } + + /** + * Constructs a new {@link RedisEntityCache} over an existing client, without probing it. + * + * @param cacheConfig the cache configuration + * @param jedis the Redis client, standalone or cluster + */ + @VisibleForTesting + RedisEntityCache(Config cacheConfig, UnifiedJedis jedis) { + super(cacheConfig); + Preconditions.checkArgument(jedis != null, "jedis must not be null"); + this.jedis = jedis; + this.keyspace = new RedisKeyspace(cacheConfig.get(Configs.CACHE_REDIS_NAMESPACE)); + // Only the Kryo serializer exists; the config entry validates the name. + cacheConfig.get(Configs.CACHE_REDIS_SERIALIZER); + this.serializer = new KryoEntitySerializer(); + this.segmentedLock = new SegmentedLock(cacheConfig.get(Configs.CACHE_LOCK_SEGMENTS)); + this.valueTtlMs = cacheConfig.get(Configs.CACHE_EXPIRATION_TIME); + long configuredFenceTtl = cacheConfig.get(Configs.CACHE_REDIS_FENCE_TTL_MS); + this.fenceTtlMs = configuredFenceTtl == 0 ? 2 * valueTtlMs : configuredFenceTtl; + Preconditions.checkArgument( + valueTtlMs == 0 || fenceTtlMs > valueTtlMs, + "%s (%s ms) must exceed %s (%s ms)", + Configs.CACHE_REDIS_FENCE_TTL_MS.getKey(), + fenceTtlMs, + Configs.CACHE_EXPIRATION_TIME.getKey(), + valueTtlMs); + this.readScript = utf8(READ_SCRIPT); + this.putScript = utf8(PUT_SCRIPT); + this.dropScript = utf8(DROP_SCRIPT); + } + + /** {@inheritDoc} */ + @Override + public Coherence coherence() { + return Coherence.SHARED; + } + + /** + * {@inheritDoc} + * + * <p>On a miss, records the fences guarding this key for the calling thread so that the write + * which follows can be rejected if a drop commits in between. A Redis failure is a miss. + */ + @Override + public <E extends Entity & HasIdentifier> Optional<E> getIfPresent( + NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<String> fencePaths = RedisKeyspace.fencePaths(ident, schemaSeparator()); + List<byte[]> fenceKeys = Lists.newArrayListWithCapacity(fencePaths.size()); + for (String path : fencePaths) { + fenceKeys.add(utf8(keyspace.fenceKey(ident, path))); + } + + List<?> reply; + try { + reply = + (List<?>) + jedis.eval(readScript, ImmutableList.of(utf8(keyspace.valueKey(key))), fenceKeys); + } catch (JedisException e) { + logFailure("read", key, e); + return Optional.empty(); + } + + if (((Long) reply.get(0)) == 1L) { + byte[] bytes = (byte[]) reply.get(1); + try { + return Optional.of(convertEntity(serializer.deserialize(bytes))); + } catch (RuntimeException e) { + LOG.warn("Discarding cache entry {} that could not be deserialized", key, e); + discardQuietly(key); + return Optional.empty(); + } + } + + List<byte[]> epochs = Lists.newArrayListWithCapacity(fenceKeys.size()); + for (int i = 1; i < reply.size(); i++) { + epochs.add((byte[]) reply.get(i)); + } + pendingFences.get().put(RedisKeyspace.member(key), new FenceSnapshot(fenceKeys, epochs)); + return Optional.empty(); + } + + /** + * {@inheritDoc} + * + * <p>Bumps the fence of the identifier and deletes the entry together with every indexed + * descendant in one atomic script. A Redis failure is propagated, because a dropped invalidation + * would leave a stale value readable by every node. + */ + @Override + public boolean invalidate(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(keyspace.slotPrefix(ident))); + args.add(utf8(RedisKeyspace.member(key))); + args.add(utf8(ident.toString())); + args.add(utf8(Long.toString(fenceTtlMs))); + for (String prefix : RedisKeyspace.descendantPrefixes(key, schemaSeparator())) { + args.add(utf8(prefix)); + } + List<byte[]> keys = ImmutableList.of(utf8(keyspace.indexKey(ident))); + + return segmentedLock.withLock( + key, + () -> { + try { + jedis.eval(dropScript, keys, args); + return true; + } catch (JedisException e) { + throw new RuntimeException( + "Failed to invalidate entity cache entry " + + key + + " in Redis; a stale entry may remain readable until it expires", + e); + } + }); + } + + /** {@inheritDoc} A Redis failure is reported as absent. */ + @Override + public boolean contains(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + try { + return jedis.exists(utf8(keyspace.valueKey(key))); + } catch (JedisException e) { + logFailure("contains", key, e); + return false; + } + } + + /** + * {@inheritDoc} + * + * <p>Sums the index of every metalake, scanning each cluster node, so the result is a + * point-in-time estimate that may briefly exceed the number of live values: an index member is + * removed by the next drop that ranges over it, not when its value expires. + */ + @Override + public long size() { + AtomicLong total = new AtomicLong(); + forEachNode( + node -> + scan( + node, + keyspace.allIndexKeysPattern(), + indexKey -> total.addAndGet(node.zcard(indexKey)))); + return total.get(); + } + + /** + * {@inheritDoc} + * + * <p>Removes every value and index key of this namespace on every node. Fences are left in place + * so that a load in flight during the clear is still rejected; they expire on their own. + */ + @Override + public void clear() { + segmentedLock.withGlobalLock( + () -> { + pendingFences.get().clear(); + forEachNode( + node -> + scan( + node, + keyspace.allKeysPattern(), + key -> { + // One key per command: on a cluster node the keys scanned belong to many + // slots, and a multi-key UNLINK across slots is rejected. + if (!RedisKeyspace.isFenceKey(key)) { + node.unlink(key); + } + })); + }); + } + + /** + * {@inheritDoc} + * + * <p>Guarded by the fences recorded by the miss that preceded it on this thread, if any. A Redis + * failure discards the write; the next read loads again. + */ + @Override + protected <E extends Entity & HasIdentifier> void doPut(E entity) { + NameIdentifier ident = getIdentFromEntity(entity); + EntityCacheKey key = EntityCacheKey.of(ident, entity.type()); + String member = RedisKeyspace.member(key); + FenceSnapshot snapshot = pendingFences.get().remove(member); + + byte[] value; + try { + value = serializer.serialize(entity); + } catch (RuntimeException e) { + LOG.warn("Not caching entity {}: serialization failed", key, e); + return; + } + + List<byte[]> keys = + ImmutableList.of(utf8(keyspace.indexKey(ident)), utf8(keyspace.valueKey(key))); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(member)); + args.add(value); + args.add(utf8(Long.toString(valueTtlMs))); + if (snapshot != null) { Review Comment: [P1] Missing fence snapshots must not authorize unconditional writes `snapshot == null` skips every fence check, but this occurs on real store paths, not just deliberate cache warming: - `RelationalEntityStore.put(e, false)` commits the insert and then calls `cache.put(e)`. Another node can delete the entity and successfully invalidate it between these steps; the delayed put resurrects it. - `batchGet` records all misses before loading from the backend. With 1,025 misses, the first snapshot is evicted by the 1,024-entry bound, allowing its stale result to refill after invalidation. - A Redis read error or decoding failure returns a miss without obtaining a snapshot, so the subsequent fill is also unguarded. I reproduced the insert, batch-boundary, and decoding-failure cases against Redis. Please skip fills without a valid snapshot, including insert-time warming, and ensure failed reads cannot reuse an unrelated pending snapshot. Missing protection should fail closed for cache population. These paths need regression coverage in this PR. ########## core/src/main/java/org/apache/gravitino/cache/RedisEntityCache.java: ########## @@ -0,0 +1,630 @@ +/* + * 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.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; +import redis.clients.jedis.commands.KeyCommands; +import redis.clients.jedis.commands.SortedSetCommands; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * An {@link EntityCache} that keeps one copy of every cached entity in Redis, shared by all nodes + * of a Gravitino cluster. It is selected with {@code gravitino.cache.implementation=redis} and + * reports {@link Coherence#SHARED}: a write on any node invalidates the single shared copy, so no + * per-node propagation is needed. + * + * <p>This is a shared cache, not a strongly consistent database and cache pair. The cache is only + * touched after the entity store has committed, and an invalidation deletes the entry rather than + * updating it in place, so a Redis failure degrades to a cache miss and never to a stale hit. What + * remains is the window between a store commit and the invalidation that follows it: a node that + * dies inside that window leaves the entry readable until its TTL expires. Callers that cannot + * tolerate that must read the store directly. + * + * <p><b>Keyspace.</b> See {@link RedisKeyspace}. All keys of one metalake share a Redis Cluster + * hash slot, so every script here is single-slot at any depth; the cost is that one metalake's + * entries are bounded by one cluster node. + * + * <p><b>Container drop.</b> {@link #invalidate(NameIdentifier, Entity.EntityType)} runs one Lua + * script that bumps the version fence of the dropped identifier, deletes its value, and walks the + * lexicographic index range of its descendants deleting each one. The script is atomic, so a + * concurrent reader sees the subtree either complete or gone, never half dropped. + * + * <p><b>Stale-write guard.</b> A read miss records the current fence of the identifier and of each + * of its ancestors for the calling thread. The write that fills the entry afterwards is a Lua + * script that compares those fences again and refuses to write if any has moved. A load that began + * before a drop of the entity, or of any container above it, therefore cannot refill the key once + * the drop has committed, whether or not the entity was indexed at the time. Fences outlive the + * value TTL (see {@code gravitino.cache.redis.fenceTtlMs}), so a slow reader cannot win by + * outliving the value. A write with no recorded fences, such as caching a freshly inserted entity, + * is written unconditionally. + * + * <p><b>Failure policy.</b> Reads and fills are optimizations: a Redis error or timeout makes them + * a miss or a no-op. An invalidation is a correctness obligation: a failure is propagated as a + * {@link RuntimeException} so it is never silently dropped, and the entry expires by TTL at the + * latest. An unreachable Redis at startup fails fast. + */ +public class RedisEntityCache extends BaseEntityCache { + + private static final Logger LOG = LoggerFactory.getLogger(RedisEntityCache.class); + + /** Per-thread bound on recorded fences awaiting the write that consumes them. */ + private static final int MAX_PENDING_FENCES = 1024; + + private static final long FAILURE_LOG_INTERVAL_MS = 30_000L; + private static final int SCAN_BATCH = 500; + + /** + * Reads a value, or on a miss the fences guarding a later fill. {@code KEYS[1]} is the value key; + * {@code ARGV} lists the fence keys. Returns {@code {1, value}} on a hit and {@code {0, + * fence...}} on a miss, with an absent fence reported as {@code "0"}. + */ + @VisibleForTesting + static final String READ_SCRIPT = + "local v = redis.call('GET', KEYS[1])\n" + + "if v then return {1, v} end\n" + + "local r = {0}\n" + + "for i = 1, #ARGV do r[#r + 1] = redis.call('GET', ARGV[i]) or '0' end\n" + + "return r\n"; + + /** + * Writes a value unless a guarding fence moved. {@code KEYS[1]} is the index key and {@code + * KEYS[2]} the value key; {@code ARGV[1]} is the index member, {@code ARGV[2]} the value, {@code + * ARGV[3]} the TTL in milliseconds (0 for none), followed by (fence key, expected value) pairs. + * Returns 1 if written and 0 if rejected. + */ + @VisibleForTesting + static final String PUT_SCRIPT = + "for i = 4, #ARGV, 2 do\n" + + " if (redis.call('GET', ARGV[i]) or '0') ~= ARGV[i + 1] then return 0 end\n" + + "end\n" + + "if tonumber(ARGV[3]) > 0 then\n" + + " redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])\n" + + "else\n" + + " redis.call('SET', KEYS[2], ARGV[2])\n" + + "end\n" + + "redis.call('ZADD', KEYS[1], 0, ARGV[1])\n" Review Comment: [P2] Expired values leave index members indefinitely Only the value receives a TTL. Redis value expiration does not invoke `invalidateExpiredItem()`, and this implementation has no other caller or cleanup mechanism for that hook. Consequently, IDX retains every distinct cached identifier until an explicit matching invalidation or clear. I verified that after the value expires, `contains()` is false while `size()` remains 1. For a long-lived metalake with continually created job identifiers, this becomes unbounded historical index growth, rather than a brief size overestimate. It also increases the work done by the synchronous container-drop script. Please add bounded index reclamation, with an atomic absence check before removing a member so cleanup cannot unindex a concurrent refill. Expiring the entire IDX independently is unsafe because live values would lose their invalidation index. Add a test covering expired members and concurrent refill. ########## core/src/main/java/org/apache/gravitino/cache/RedisEntityCache.java: ########## @@ -0,0 +1,630 @@ +/* + * 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.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; +import redis.clients.jedis.commands.KeyCommands; +import redis.clients.jedis.commands.SortedSetCommands; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * An {@link EntityCache} that keeps one copy of every cached entity in Redis, shared by all nodes + * of a Gravitino cluster. It is selected with {@code gravitino.cache.implementation=redis} and + * reports {@link Coherence#SHARED}: a write on any node invalidates the single shared copy, so no + * per-node propagation is needed. + * + * <p>This is a shared cache, not a strongly consistent database and cache pair. The cache is only + * touched after the entity store has committed, and an invalidation deletes the entry rather than + * updating it in place, so a Redis failure degrades to a cache miss and never to a stale hit. What + * remains is the window between a store commit and the invalidation that follows it: a node that + * dies inside that window leaves the entry readable until its TTL expires. Callers that cannot + * tolerate that must read the store directly. + * + * <p><b>Keyspace.</b> See {@link RedisKeyspace}. All keys of one metalake share a Redis Cluster + * hash slot, so every script here is single-slot at any depth; the cost is that one metalake's + * entries are bounded by one cluster node. + * + * <p><b>Container drop.</b> {@link #invalidate(NameIdentifier, Entity.EntityType)} runs one Lua + * script that bumps the version fence of the dropped identifier, deletes its value, and walks the + * lexicographic index range of its descendants deleting each one. The script is atomic, so a + * concurrent reader sees the subtree either complete or gone, never half dropped. + * + * <p><b>Stale-write guard.</b> A read miss records the current fence of the identifier and of each + * of its ancestors for the calling thread. The write that fills the entry afterwards is a Lua + * script that compares those fences again and refuses to write if any has moved. A load that began + * before a drop of the entity, or of any container above it, therefore cannot refill the key once + * the drop has committed, whether or not the entity was indexed at the time. Fences outlive the + * value TTL (see {@code gravitino.cache.redis.fenceTtlMs}), so a slow reader cannot win by + * outliving the value. A write with no recorded fences, such as caching a freshly inserted entity, + * is written unconditionally. + * + * <p><b>Failure policy.</b> Reads and fills are optimizations: a Redis error or timeout makes them + * a miss or a no-op. An invalidation is a correctness obligation: a failure is propagated as a + * {@link RuntimeException} so it is never silently dropped, and the entry expires by TTL at the + * latest. An unreachable Redis at startup fails fast. + */ +public class RedisEntityCache extends BaseEntityCache { + + private static final Logger LOG = LoggerFactory.getLogger(RedisEntityCache.class); + + /** Per-thread bound on recorded fences awaiting the write that consumes them. */ + private static final int MAX_PENDING_FENCES = 1024; + + private static final long FAILURE_LOG_INTERVAL_MS = 30_000L; + private static final int SCAN_BATCH = 500; + + /** + * Reads a value, or on a miss the fences guarding a later fill. {@code KEYS[1]} is the value key; + * {@code ARGV} lists the fence keys. Returns {@code {1, value}} on a hit and {@code {0, + * fence...}} on a miss, with an absent fence reported as {@code "0"}. + */ + @VisibleForTesting + static final String READ_SCRIPT = + "local v = redis.call('GET', KEYS[1])\n" + + "if v then return {1, v} end\n" + + "local r = {0}\n" + + "for i = 1, #ARGV do r[#r + 1] = redis.call('GET', ARGV[i]) or '0' end\n" + + "return r\n"; + + /** + * Writes a value unless a guarding fence moved. {@code KEYS[1]} is the index key and {@code + * KEYS[2]} the value key; {@code ARGV[1]} is the index member, {@code ARGV[2]} the value, {@code + * ARGV[3]} the TTL in milliseconds (0 for none), followed by (fence key, expected value) pairs. + * Returns 1 if written and 0 if rejected. + */ + @VisibleForTesting + static final String PUT_SCRIPT = + "for i = 4, #ARGV, 2 do\n" + + " if (redis.call('GET', ARGV[i]) or '0') ~= ARGV[i + 1] then return 0 end\n" + + "end\n" + + "if tonumber(ARGV[3]) > 0 then\n" + + " redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])\n" + + "else\n" + + " redis.call('SET', KEYS[2], ARGV[2])\n" + + "end\n" + + "redis.call('ZADD', KEYS[1], 0, ARGV[1])\n" + + "return 1\n"; + + /** + * Drops an entry and its indexed descendants. {@code KEYS[1]} is the index key; {@code ARGV[1]} + * is the slot prefix, {@code ARGV[2]} the index member, {@code ARGV[3]} the identifier whose + * fence to bump, {@code ARGV[4]} the fence TTL in milliseconds (0 for none), followed by the + * descendant prefixes to range over. The upper bound {@code prefix\xff} is above every UTF-8 + * member starting with the prefix. Returns the number of values deleted. + */ + @VisibleForTesting + static final String DROP_SCRIPT = + "local idx = KEYS[1]\n" + + "local prefix = ARGV[1]\n" + + "local fence = prefix .. 'F:' .. ARGV[3]\n" + + "redis.call('INCR', fence)\n" + + "if tonumber(ARGV[4]) > 0 then redis.call('PEXPIRE', fence, ARGV[4]) end\n" + + "local removed = redis.call('DEL', prefix .. 'D:' .. ARGV[2])\n" + + "redis.call('ZREM', idx, ARGV[2])\n" + + "for i = 5, #ARGV do\n" + + " local members = redis.call('ZRANGEBYLEX', idx, '[' .. ARGV[i], " + + "'[' .. ARGV[i] .. '\\255')\n" + + " for _, m in ipairs(members) do\n" + + " removed = removed + redis.call('DEL', prefix .. 'D:' .. m)\n" + + " redis.call('ZREM', idx, m)\n" + + " end\n" + + "end\n" + + "return removed\n"; + + private final UnifiedJedis jedis; + private final RedisKeyspace keyspace; + private final KryoEntitySerializer serializer; + private final SegmentedLock segmentedLock; + private final long valueTtlMs; + private final long fenceTtlMs; + private final byte[] readScript; + private final byte[] putScript; + private final byte[] dropScript; + private final AtomicLong lastFailureLogMs = new AtomicLong(); + + /** + * Fences observed by a read miss on this thread, keyed by index member, consumed by the write + * that fills the entry. Bounded and access-ordered so an entry whose write never happens (the + * store had no such entity) is eventually evicted; a leftover entry can only make a later write + * more conservative, never less. + */ + private final ThreadLocal<Map<String, FenceSnapshot>> pendingFences = + ThreadLocal.withInitial( + () -> + new LinkedHashMap<String, FenceSnapshot>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, FenceSnapshot> eldest) { + return size() > MAX_PENDING_FENCES; + } + }); + + /** + * Constructs a new {@link RedisEntityCache} connected to the Redis deployment described by the + * {@code gravitino.cache.redis.*} configuration. + * + * @param cacheConfig the cache configuration + */ + public RedisEntityCache(Config cacheConfig) { + this(cacheConfig, createClient(cacheConfig)); + probe(cacheConfig.get(Configs.CACHE_REDIS_ADDRESS)); + } + + /** + * Constructs a new {@link RedisEntityCache} over an existing client, without probing it. + * + * @param cacheConfig the cache configuration + * @param jedis the Redis client, standalone or cluster + */ + @VisibleForTesting + RedisEntityCache(Config cacheConfig, UnifiedJedis jedis) { + super(cacheConfig); + Preconditions.checkArgument(jedis != null, "jedis must not be null"); + this.jedis = jedis; + this.keyspace = new RedisKeyspace(cacheConfig.get(Configs.CACHE_REDIS_NAMESPACE)); + // Only the Kryo serializer exists; the config entry validates the name. + cacheConfig.get(Configs.CACHE_REDIS_SERIALIZER); + this.serializer = new KryoEntitySerializer(); + this.segmentedLock = new SegmentedLock(cacheConfig.get(Configs.CACHE_LOCK_SEGMENTS)); + this.valueTtlMs = cacheConfig.get(Configs.CACHE_EXPIRATION_TIME); + long configuredFenceTtl = cacheConfig.get(Configs.CACHE_REDIS_FENCE_TTL_MS); + this.fenceTtlMs = configuredFenceTtl == 0 ? 2 * valueTtlMs : configuredFenceTtl; + Preconditions.checkArgument( + valueTtlMs == 0 || fenceTtlMs > valueTtlMs, + "%s (%s ms) must exceed %s (%s ms)", + Configs.CACHE_REDIS_FENCE_TTL_MS.getKey(), + fenceTtlMs, + Configs.CACHE_EXPIRATION_TIME.getKey(), + valueTtlMs); + this.readScript = utf8(READ_SCRIPT); + this.putScript = utf8(PUT_SCRIPT); + this.dropScript = utf8(DROP_SCRIPT); + } + + /** {@inheritDoc} */ + @Override + public Coherence coherence() { + return Coherence.SHARED; + } + + /** + * {@inheritDoc} + * + * <p>On a miss, records the fences guarding this key for the calling thread so that the write + * which follows can be rejected if a drop commits in between. A Redis failure is a miss. + */ + @Override + public <E extends Entity & HasIdentifier> Optional<E> getIfPresent( + NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<String> fencePaths = RedisKeyspace.fencePaths(ident, schemaSeparator()); + List<byte[]> fenceKeys = Lists.newArrayListWithCapacity(fencePaths.size()); + for (String path : fencePaths) { + fenceKeys.add(utf8(keyspace.fenceKey(ident, path))); + } + + List<?> reply; + try { + reply = + (List<?>) + jedis.eval(readScript, ImmutableList.of(utf8(keyspace.valueKey(key))), fenceKeys); + } catch (JedisException e) { + logFailure("read", key, e); + return Optional.empty(); + } + + if (((Long) reply.get(0)) == 1L) { + byte[] bytes = (byte[]) reply.get(1); + try { + return Optional.of(convertEntity(serializer.deserialize(bytes))); + } catch (RuntimeException e) { + LOG.warn("Discarding cache entry {} that could not be deserialized", key, e); + discardQuietly(key); + return Optional.empty(); + } + } + + List<byte[]> epochs = Lists.newArrayListWithCapacity(fenceKeys.size()); + for (int i = 1; i < reply.size(); i++) { + epochs.add((byte[]) reply.get(i)); + } + pendingFences.get().put(RedisKeyspace.member(key), new FenceSnapshot(fenceKeys, epochs)); + return Optional.empty(); + } + + /** + * {@inheritDoc} + * + * <p>Bumps the fence of the identifier and deletes the entry together with every indexed + * descendant in one atomic script. A Redis failure is propagated, because a dropped invalidation + * would leave a stale value readable by every node. + */ + @Override + public boolean invalidate(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(keyspace.slotPrefix(ident))); + args.add(utf8(RedisKeyspace.member(key))); + args.add(utf8(ident.toString())); + args.add(utf8(Long.toString(fenceTtlMs))); + for (String prefix : RedisKeyspace.descendantPrefixes(key, schemaSeparator())) { + args.add(utf8(prefix)); + } + List<byte[]> keys = ImmutableList.of(utf8(keyspace.indexKey(ident))); + + return segmentedLock.withLock( + key, + () -> { + try { + jedis.eval(dropScript, keys, args); + return true; + } catch (JedisException e) { + throw new RuntimeException( + "Failed to invalidate entity cache entry " + + key + + " in Redis; a stale entry may remain readable until it expires", + e); + } + }); + } + + /** {@inheritDoc} A Redis failure is reported as absent. */ + @Override + public boolean contains(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + try { + return jedis.exists(utf8(keyspace.valueKey(key))); + } catch (JedisException e) { + logFailure("contains", key, e); + return false; + } + } + + /** + * {@inheritDoc} + * + * <p>Sums the index of every metalake, scanning each cluster node, so the result is a + * point-in-time estimate that may briefly exceed the number of live values: an index member is + * removed by the next drop that ranges over it, not when its value expires. + */ + @Override + public long size() { + AtomicLong total = new AtomicLong(); + forEachNode( + node -> + scan( + node, + keyspace.allIndexKeysPattern(), + indexKey -> total.addAndGet(node.zcard(indexKey)))); + return total.get(); + } + + /** + * {@inheritDoc} + * + * <p>Removes every value and index key of this namespace on every node. Fences are left in place + * so that a load in flight during the clear is still rejected; they expire on their own. + */ + @Override + public void clear() { + segmentedLock.withGlobalLock( + () -> { + pendingFences.get().clear(); + forEachNode( + node -> + scan( + node, + keyspace.allKeysPattern(), + key -> { + // One key per command: on a cluster node the keys scanned belong to many + // slots, and a multi-key UNLINK across slots is rejected. + if (!RedisKeyspace.isFenceKey(key)) { + node.unlink(key); + } + })); + }); + } + + /** + * {@inheritDoc} + * + * <p>Guarded by the fences recorded by the miss that preceded it on this thread, if any. A Redis + * failure discards the write; the next read loads again. + */ + @Override + protected <E extends Entity & HasIdentifier> void doPut(E entity) { + NameIdentifier ident = getIdentFromEntity(entity); + EntityCacheKey key = EntityCacheKey.of(ident, entity.type()); + String member = RedisKeyspace.member(key); + FenceSnapshot snapshot = pendingFences.get().remove(member); + + byte[] value; + try { + value = serializer.serialize(entity); + } catch (RuntimeException e) { + LOG.warn("Not caching entity {}: serialization failed", key, e); + return; + } + + List<byte[]> keys = + ImmutableList.of(utf8(keyspace.indexKey(ident)), utf8(keyspace.valueKey(key))); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(member)); + args.add(value); + args.add(utf8(Long.toString(valueTtlMs))); + if (snapshot != null) { + for (int i = 0; i < snapshot.fenceKeys.size(); i++) { + args.add(snapshot.fenceKeys.get(i)); + args.add(snapshot.epochs.get(i)); + } + } + + try { + Object written = jedis.eval(putScript, keys, args); + if (Long.valueOf(0L).equals(written)) { + LOG.debug("Rejected write of {}: the entry was invalidated while it was loading", key); + } + } catch (JedisException e) { + logFailure("write", key, e); + } + } + + /** {@inheritDoc} */ + @Override + public <E extends Entity & HasIdentifier> void invalidateOnKeyChange(E entity) { + // Every cacheable entity is self-contained (see BaseEntityCache#isCacheable), so inserting one + // never requires invalidating a different key. + } + + /** {@inheritDoc} */ + @Override + public <E extends Exception> void withCacheLock(EntityCacheKey key, ThrowingRunnable<E> action) + throws E { + Preconditions.checkArgument(key != null, "Key cannot be null"); + Preconditions.checkArgument(action != null, "Action cannot be null"); + try { + segmentedLock.withLockAndThrow(key, action); + } finally { + pendingFences.get().remove(RedisKeyspace.member(key)); + } + } + + /** {@inheritDoc} */ + @Override + public <T, E extends Exception> T withCacheLock(EntityCacheKey key, ThrowingSupplier<T, E> action) + throws E { + Preconditions.checkArgument(key != null, "Key cannot be null"); + Preconditions.checkArgument(action != null, "Action cannot be null"); + try { + return segmentedLock.withLockAndThrow(key, action); + } finally { + pendingFences.get().remove(RedisKeyspace.member(key)); + } + } + + /** + * {@inheritDoc} + * + * <p>Redis expires values on its own; this only drops the index member, best effort. + */ + @Override + protected void invalidateExpiredItem(EntityCacheKey key) { + try { + jedis.zrem(keyspace.indexKey(key.identifier()), RedisKeyspace.member(key)); + } catch (JedisException e) { + logFailure("unindex", key, e); + } + } + + /** + * Closes the underlying Redis client. Not part of the {@link EntityCache} SPI: the entity store + * keeps its cache for the life of the process, so this exists for tests and embedded use. + */ + public void close() { Review Comment: [P2] Wire Redis client closure into the entity-store lifecycle `RelationalEntityStore.close()` currently calls `cache::clear`, not this method. With the new shared backend, stopping one store therefore clears the cache used by every other node, while leaving its own Redis client/pools open. I reproduced both effects: the peer's entry disappeared after `store.close()`, and the supposedly closed store's Redis client remained usable. This is also a resource leak when stores are closed/reinitialized within a process. Please provide a compatible cache close contract, have the store invoke it, and let Redis close only its own client. Local caches can retain clear-on-close behavior; clearing shared data should be an explicit operation. Cover resource release and preservation of another node's cached entries in a lifecycle test. ########## core/src/main/java/org/apache/gravitino/cache/RedisEntityCache.java: ########## @@ -0,0 +1,630 @@ +/* + * 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.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; +import redis.clients.jedis.commands.KeyCommands; +import redis.clients.jedis.commands.SortedSetCommands; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * An {@link EntityCache} that keeps one copy of every cached entity in Redis, shared by all nodes + * of a Gravitino cluster. It is selected with {@code gravitino.cache.implementation=redis} and + * reports {@link Coherence#SHARED}: a write on any node invalidates the single shared copy, so no + * per-node propagation is needed. + * + * <p>This is a shared cache, not a strongly consistent database and cache pair. The cache is only + * touched after the entity store has committed, and an invalidation deletes the entry rather than + * updating it in place, so a Redis failure degrades to a cache miss and never to a stale hit. What + * remains is the window between a store commit and the invalidation that follows it: a node that + * dies inside that window leaves the entry readable until its TTL expires. Callers that cannot + * tolerate that must read the store directly. + * + * <p><b>Keyspace.</b> See {@link RedisKeyspace}. All keys of one metalake share a Redis Cluster + * hash slot, so every script here is single-slot at any depth; the cost is that one metalake's + * entries are bounded by one cluster node. + * + * <p><b>Container drop.</b> {@link #invalidate(NameIdentifier, Entity.EntityType)} runs one Lua + * script that bumps the version fence of the dropped identifier, deletes its value, and walks the + * lexicographic index range of its descendants deleting each one. The script is atomic, so a + * concurrent reader sees the subtree either complete or gone, never half dropped. + * + * <p><b>Stale-write guard.</b> A read miss records the current fence of the identifier and of each + * of its ancestors for the calling thread. The write that fills the entry afterwards is a Lua + * script that compares those fences again and refuses to write if any has moved. A load that began + * before a drop of the entity, or of any container above it, therefore cannot refill the key once + * the drop has committed, whether or not the entity was indexed at the time. Fences outlive the + * value TTL (see {@code gravitino.cache.redis.fenceTtlMs}), so a slow reader cannot win by + * outliving the value. A write with no recorded fences, such as caching a freshly inserted entity, + * is written unconditionally. + * + * <p><b>Failure policy.</b> Reads and fills are optimizations: a Redis error or timeout makes them + * a miss or a no-op. An invalidation is a correctness obligation: a failure is propagated as a + * {@link RuntimeException} so it is never silently dropped, and the entry expires by TTL at the + * latest. An unreachable Redis at startup fails fast. + */ +public class RedisEntityCache extends BaseEntityCache { + + private static final Logger LOG = LoggerFactory.getLogger(RedisEntityCache.class); + + /** Per-thread bound on recorded fences awaiting the write that consumes them. */ + private static final int MAX_PENDING_FENCES = 1024; + + private static final long FAILURE_LOG_INTERVAL_MS = 30_000L; + private static final int SCAN_BATCH = 500; + + /** + * Reads a value, or on a miss the fences guarding a later fill. {@code KEYS[1]} is the value key; + * {@code ARGV} lists the fence keys. Returns {@code {1, value}} on a hit and {@code {0, + * fence...}} on a miss, with an absent fence reported as {@code "0"}. + */ + @VisibleForTesting + static final String READ_SCRIPT = + "local v = redis.call('GET', KEYS[1])\n" + + "if v then return {1, v} end\n" + + "local r = {0}\n" + + "for i = 1, #ARGV do r[#r + 1] = redis.call('GET', ARGV[i]) or '0' end\n" + + "return r\n"; + + /** + * Writes a value unless a guarding fence moved. {@code KEYS[1]} is the index key and {@code + * KEYS[2]} the value key; {@code ARGV[1]} is the index member, {@code ARGV[2]} the value, {@code + * ARGV[3]} the TTL in milliseconds (0 for none), followed by (fence key, expected value) pairs. + * Returns 1 if written and 0 if rejected. + */ + @VisibleForTesting + static final String PUT_SCRIPT = + "for i = 4, #ARGV, 2 do\n" + + " if (redis.call('GET', ARGV[i]) or '0') ~= ARGV[i + 1] then return 0 end\n" + + "end\n" + + "if tonumber(ARGV[3]) > 0 then\n" + + " redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])\n" + + "else\n" + + " redis.call('SET', KEYS[2], ARGV[2])\n" + + "end\n" + + "redis.call('ZADD', KEYS[1], 0, ARGV[1])\n" + + "return 1\n"; + + /** + * Drops an entry and its indexed descendants. {@code KEYS[1]} is the index key; {@code ARGV[1]} + * is the slot prefix, {@code ARGV[2]} the index member, {@code ARGV[3]} the identifier whose + * fence to bump, {@code ARGV[4]} the fence TTL in milliseconds (0 for none), followed by the + * descendant prefixes to range over. The upper bound {@code prefix\xff} is above every UTF-8 + * member starting with the prefix. Returns the number of values deleted. + */ + @VisibleForTesting + static final String DROP_SCRIPT = + "local idx = KEYS[1]\n" + + "local prefix = ARGV[1]\n" + + "local fence = prefix .. 'F:' .. ARGV[3]\n" + + "redis.call('INCR', fence)\n" + + "if tonumber(ARGV[4]) > 0 then redis.call('PEXPIRE', fence, ARGV[4]) end\n" + + "local removed = redis.call('DEL', prefix .. 'D:' .. ARGV[2])\n" + + "redis.call('ZREM', idx, ARGV[2])\n" + + "for i = 5, #ARGV do\n" + + " local members = redis.call('ZRANGEBYLEX', idx, '[' .. ARGV[i], " + + "'[' .. ARGV[i] .. '\\255')\n" + + " for _, m in ipairs(members) do\n" + + " removed = removed + redis.call('DEL', prefix .. 'D:' .. m)\n" + + " redis.call('ZREM', idx, m)\n" + + " end\n" + + "end\n" + + "return removed\n"; + + private final UnifiedJedis jedis; + private final RedisKeyspace keyspace; + private final KryoEntitySerializer serializer; + private final SegmentedLock segmentedLock; + private final long valueTtlMs; + private final long fenceTtlMs; + private final byte[] readScript; + private final byte[] putScript; + private final byte[] dropScript; + private final AtomicLong lastFailureLogMs = new AtomicLong(); + + /** + * Fences observed by a read miss on this thread, keyed by index member, consumed by the write + * that fills the entry. Bounded and access-ordered so an entry whose write never happens (the + * store had no such entity) is eventually evicted; a leftover entry can only make a later write + * more conservative, never less. + */ + private final ThreadLocal<Map<String, FenceSnapshot>> pendingFences = + ThreadLocal.withInitial( + () -> + new LinkedHashMap<String, FenceSnapshot>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, FenceSnapshot> eldest) { + return size() > MAX_PENDING_FENCES; + } + }); + + /** + * Constructs a new {@link RedisEntityCache} connected to the Redis deployment described by the + * {@code gravitino.cache.redis.*} configuration. + * + * @param cacheConfig the cache configuration + */ + public RedisEntityCache(Config cacheConfig) { + this(cacheConfig, createClient(cacheConfig)); + probe(cacheConfig.get(Configs.CACHE_REDIS_ADDRESS)); + } + + /** + * Constructs a new {@link RedisEntityCache} over an existing client, without probing it. + * + * @param cacheConfig the cache configuration + * @param jedis the Redis client, standalone or cluster + */ + @VisibleForTesting + RedisEntityCache(Config cacheConfig, UnifiedJedis jedis) { + super(cacheConfig); + Preconditions.checkArgument(jedis != null, "jedis must not be null"); + this.jedis = jedis; + this.keyspace = new RedisKeyspace(cacheConfig.get(Configs.CACHE_REDIS_NAMESPACE)); + // Only the Kryo serializer exists; the config entry validates the name. + cacheConfig.get(Configs.CACHE_REDIS_SERIALIZER); + this.serializer = new KryoEntitySerializer(); + this.segmentedLock = new SegmentedLock(cacheConfig.get(Configs.CACHE_LOCK_SEGMENTS)); + this.valueTtlMs = cacheConfig.get(Configs.CACHE_EXPIRATION_TIME); + long configuredFenceTtl = cacheConfig.get(Configs.CACHE_REDIS_FENCE_TTL_MS); + this.fenceTtlMs = configuredFenceTtl == 0 ? 2 * valueTtlMs : configuredFenceTtl; + Preconditions.checkArgument( + valueTtlMs == 0 || fenceTtlMs > valueTtlMs, + "%s (%s ms) must exceed %s (%s ms)", + Configs.CACHE_REDIS_FENCE_TTL_MS.getKey(), + fenceTtlMs, + Configs.CACHE_EXPIRATION_TIME.getKey(), + valueTtlMs); + this.readScript = utf8(READ_SCRIPT); + this.putScript = utf8(PUT_SCRIPT); + this.dropScript = utf8(DROP_SCRIPT); + } + + /** {@inheritDoc} */ + @Override + public Coherence coherence() { + return Coherence.SHARED; + } + + /** + * {@inheritDoc} + * + * <p>On a miss, records the fences guarding this key for the calling thread so that the write + * which follows can be rejected if a drop commits in between. A Redis failure is a miss. + */ + @Override + public <E extends Entity & HasIdentifier> Optional<E> getIfPresent( + NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<String> fencePaths = RedisKeyspace.fencePaths(ident, schemaSeparator()); + List<byte[]> fenceKeys = Lists.newArrayListWithCapacity(fencePaths.size()); + for (String path : fencePaths) { + fenceKeys.add(utf8(keyspace.fenceKey(ident, path))); + } + + List<?> reply; + try { + reply = + (List<?>) + jedis.eval(readScript, ImmutableList.of(utf8(keyspace.valueKey(key))), fenceKeys); + } catch (JedisException e) { + logFailure("read", key, e); + return Optional.empty(); + } + + if (((Long) reply.get(0)) == 1L) { + byte[] bytes = (byte[]) reply.get(1); + try { + return Optional.of(convertEntity(serializer.deserialize(bytes))); + } catch (RuntimeException e) { + LOG.warn("Discarding cache entry {} that could not be deserialized", key, e); + discardQuietly(key); + return Optional.empty(); + } + } + + List<byte[]> epochs = Lists.newArrayListWithCapacity(fenceKeys.size()); + for (int i = 1; i < reply.size(); i++) { + epochs.add((byte[]) reply.get(i)); + } + pendingFences.get().put(RedisKeyspace.member(key), new FenceSnapshot(fenceKeys, epochs)); + return Optional.empty(); + } + + /** + * {@inheritDoc} + * + * <p>Bumps the fence of the identifier and deletes the entry together with every indexed + * descendant in one atomic script. A Redis failure is propagated, because a dropped invalidation + * would leave a stale value readable by every node. + */ + @Override + public boolean invalidate(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(keyspace.slotPrefix(ident))); + args.add(utf8(RedisKeyspace.member(key))); + args.add(utf8(ident.toString())); + args.add(utf8(Long.toString(fenceTtlMs))); + for (String prefix : RedisKeyspace.descendantPrefixes(key, schemaSeparator())) { + args.add(utf8(prefix)); + } + List<byte[]> keys = ImmutableList.of(utf8(keyspace.indexKey(ident))); + + return segmentedLock.withLock( + key, + () -> { + try { + jedis.eval(dropScript, keys, args); + return true; + } catch (JedisException e) { + throw new RuntimeException( + "Failed to invalidate entity cache entry " + + key + + " in Redis; a stale entry may remain readable until it expires", + e); + } + }); + } + + /** {@inheritDoc} A Redis failure is reported as absent. */ + @Override + public boolean contains(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + try { + return jedis.exists(utf8(keyspace.valueKey(key))); + } catch (JedisException e) { + logFailure("contains", key, e); + return false; + } + } + + /** + * {@inheritDoc} + * + * <p>Sums the index of every metalake, scanning each cluster node, so the result is a + * point-in-time estimate that may briefly exceed the number of live values: an index member is + * removed by the next drop that ranges over it, not when its value expires. + */ + @Override + public long size() { + AtomicLong total = new AtomicLong(); + forEachNode( + node -> + scan( + node, + keyspace.allIndexKeysPattern(), + indexKey -> total.addAndGet(node.zcard(indexKey)))); + return total.get(); + } + + /** + * {@inheritDoc} + * + * <p>Removes every value and index key of this namespace on every node. Fences are left in place + * so that a load in flight during the clear is still rejected; they expire on their own. + */ + @Override + public void clear() { + segmentedLock.withGlobalLock( + () -> { + pendingFences.get().clear(); + forEachNode( + node -> + scan( + node, + keyspace.allKeysPattern(), + key -> { + // One key per command: on a cluster node the keys scanned belong to many + // slots, and a multi-key UNLINK across slots is rejected. + if (!RedisKeyspace.isFenceKey(key)) { + node.unlink(key); + } + })); + }); + } + + /** + * {@inheritDoc} + * + * <p>Guarded by the fences recorded by the miss that preceded it on this thread, if any. A Redis + * failure discards the write; the next read loads again. + */ + @Override + protected <E extends Entity & HasIdentifier> void doPut(E entity) { + NameIdentifier ident = getIdentFromEntity(entity); + EntityCacheKey key = EntityCacheKey.of(ident, entity.type()); + String member = RedisKeyspace.member(key); + FenceSnapshot snapshot = pendingFences.get().remove(member); + + byte[] value; + try { + value = serializer.serialize(entity); + } catch (RuntimeException e) { + LOG.warn("Not caching entity {}: serialization failed", key, e); + return; + } + + List<byte[]> keys = + ImmutableList.of(utf8(keyspace.indexKey(ident)), utf8(keyspace.valueKey(key))); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(member)); + args.add(value); + args.add(utf8(Long.toString(valueTtlMs))); + if (snapshot != null) { + for (int i = 0; i < snapshot.fenceKeys.size(); i++) { + args.add(snapshot.fenceKeys.get(i)); + args.add(snapshot.epochs.get(i)); + } + } + + try { + Object written = jedis.eval(putScript, keys, args); + if (Long.valueOf(0L).equals(written)) { + LOG.debug("Rejected write of {}: the entry was invalidated while it was loading", key); + } + } catch (JedisException e) { + logFailure("write", key, e); + } + } + + /** {@inheritDoc} */ + @Override + public <E extends Entity & HasIdentifier> void invalidateOnKeyChange(E entity) { + // Every cacheable entity is self-contained (see BaseEntityCache#isCacheable), so inserting one + // never requires invalidating a different key. + } + + /** {@inheritDoc} */ + @Override + public <E extends Exception> void withCacheLock(EntityCacheKey key, ThrowingRunnable<E> action) + throws E { + Preconditions.checkArgument(key != null, "Key cannot be null"); + Preconditions.checkArgument(action != null, "Action cannot be null"); + try { + segmentedLock.withLockAndThrow(key, action); + } finally { + pendingFences.get().remove(RedisKeyspace.member(key)); + } + } + + /** {@inheritDoc} */ + @Override + public <T, E extends Exception> T withCacheLock(EntityCacheKey key, ThrowingSupplier<T, E> action) + throws E { + Preconditions.checkArgument(key != null, "Key cannot be null"); + Preconditions.checkArgument(action != null, "Action cannot be null"); + try { + return segmentedLock.withLockAndThrow(key, action); + } finally { + pendingFences.get().remove(RedisKeyspace.member(key)); + } + } + + /** + * {@inheritDoc} + * + * <p>Redis expires values on its own; this only drops the index member, best effort. + */ + @Override + protected void invalidateExpiredItem(EntityCacheKey key) { + try { + jedis.zrem(keyspace.indexKey(key.identifier()), RedisKeyspace.member(key)); + } catch (JedisException e) { + logFailure("unindex", key, e); + } + } + + /** + * Closes the underlying Redis client. Not part of the {@link EntityCache} SPI: the entity store + * keeps its cache for the life of the process, so this exists for tests and embedded use. + */ + public void close() { + jedis.close(); + } + + @VisibleForTesting + UnifiedJedis client() { + return jedis; + } + + @VisibleForTesting + RedisKeyspace keyspace() { + return keyspace; + } + + private static UnifiedJedis createClient(Config config) { + String address = config.get(Configs.CACHE_REDIS_ADDRESS); + Preconditions.checkArgument( + StringUtils.isNotBlank(address), + "%s must be set when %s is 'redis'", + Configs.CACHE_REDIS_ADDRESS.getKey(), + Configs.CACHE_IMPLEMENTATION.getKey()); + + Set<HostAndPort> nodes = new LinkedHashSet<>(); + for (String node : address.split(",")) { + if (StringUtils.isNotBlank(node)) { + nodes.add(HostAndPort.from(node.trim())); + } + } + Preconditions.checkArgument( + !nodes.isEmpty(), "%s lists no address", Configs.CACHE_REDIS_ADDRESS.getKey()); + + int timeoutMs = config.get(Configs.CACHE_REDIS_TIMEOUT_MS); + DefaultJedisClientConfig.Builder clientConfig = + DefaultJedisClientConfig.builder().timeoutMillis(timeoutMs); + String username = config.get(Configs.CACHE_REDIS_USERNAME); + if (StringUtils.isNotBlank(username)) { + clientConfig.user(username); + } + String password = config.get(Configs.CACHE_REDIS_PASSWORD); + if (StringUtils.isNotBlank(password)) { + clientConfig.password(password); + } + JedisClientConfig built = clientConfig.build(); + + if (config.get(Configs.CACHE_REDIS_CLUSTER)) { + return new JedisCluster(nodes, built); + } + Preconditions.checkArgument( + nodes.size() == 1, + "%s lists %s addresses; a standalone Redis takes exactly one, set %s=true for a cluster", + Configs.CACHE_REDIS_ADDRESS.getKey(), + nodes.size(), + Configs.CACHE_REDIS_CLUSTER.getKey()); + return new JedisPooled(nodes.iterator().next(), built); + } + + /** Fails fast if Redis cannot be reached, so a misconfiguration never degrades silently. */ + private void probe(String address) { + try { + jedis.exists(keyspace.allKeysPattern()); + } catch (JedisException e) { + jedis.close(); + throw new IllegalStateException("Cannot reach the Redis entity cache at " + address, e); + } + } + + private void discardQuietly(EntityCacheKey key) { + try { + jedis.unlink(keyspace.valueKey(key)); + jedis.zrem(keyspace.indexKey(key.identifier()), RedisKeyspace.member(key)); + } catch (JedisException e) { + logFailure("discard", key, e); + } + } + + /** Runs an action against the standalone server, or against every node of the cluster. */ + private void forEachNode(Consumer<NodeCommands> action) { + if (jedis instanceof JedisCluster) { + for (ConnectionPool pool : ((JedisCluster) jedis).getClusterNodes().values()) { Review Comment: [P2] Iterate cluster primaries rather than every discovered node `getClusterNodes()` includes replicas. `SCAN` can return an IDX key from a replica, but the subsequent `ZCARD` uses a direct `Jedis` connection without cluster redirection handling. In a real three-primary/three-replica cluster, I reproduced `size()` failing with `JedisMovedDataException: MOVED 6916 127.0.0.1:17001` after discovering all six nodes. Please enumerate current primaries and handle topology changes/redirections for node-wide operations. `clear()` uses the same traversal and needs the same treatment. The integration setup currently sets `SLAVES_PER_MASTER=0`, so add coverage with replicas rather than relying solely on a primary-only cluster. ########## core/src/main/java/org/apache/gravitino/cache/RedisEntityCache.java: ########## @@ -0,0 +1,630 @@ +/* + * 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.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; +import com.google.common.collect.Lists; +import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import redis.clients.jedis.ConnectionPool; +import redis.clients.jedis.DefaultJedisClientConfig; +import redis.clients.jedis.HostAndPort; +import redis.clients.jedis.Jedis; +import redis.clients.jedis.JedisClientConfig; +import redis.clients.jedis.JedisCluster; +import redis.clients.jedis.JedisPooled; +import redis.clients.jedis.UnifiedJedis; +import redis.clients.jedis.commands.KeyCommands; +import redis.clients.jedis.commands.SortedSetCommands; +import redis.clients.jedis.exceptions.JedisException; +import redis.clients.jedis.params.ScanParams; +import redis.clients.jedis.resps.ScanResult; + +/** + * An {@link EntityCache} that keeps one copy of every cached entity in Redis, shared by all nodes + * of a Gravitino cluster. It is selected with {@code gravitino.cache.implementation=redis} and + * reports {@link Coherence#SHARED}: a write on any node invalidates the single shared copy, so no + * per-node propagation is needed. + * + * <p>This is a shared cache, not a strongly consistent database and cache pair. The cache is only + * touched after the entity store has committed, and an invalidation deletes the entry rather than + * updating it in place, so a Redis failure degrades to a cache miss and never to a stale hit. What + * remains is the window between a store commit and the invalidation that follows it: a node that + * dies inside that window leaves the entry readable until its TTL expires. Callers that cannot + * tolerate that must read the store directly. + * + * <p><b>Keyspace.</b> See {@link RedisKeyspace}. All keys of one metalake share a Redis Cluster + * hash slot, so every script here is single-slot at any depth; the cost is that one metalake's + * entries are bounded by one cluster node. + * + * <p><b>Container drop.</b> {@link #invalidate(NameIdentifier, Entity.EntityType)} runs one Lua + * script that bumps the version fence of the dropped identifier, deletes its value, and walks the + * lexicographic index range of its descendants deleting each one. The script is atomic, so a + * concurrent reader sees the subtree either complete or gone, never half dropped. + * + * <p><b>Stale-write guard.</b> A read miss records the current fence of the identifier and of each + * of its ancestors for the calling thread. The write that fills the entry afterwards is a Lua + * script that compares those fences again and refuses to write if any has moved. A load that began + * before a drop of the entity, or of any container above it, therefore cannot refill the key once + * the drop has committed, whether or not the entity was indexed at the time. Fences outlive the + * value TTL (see {@code gravitino.cache.redis.fenceTtlMs}), so a slow reader cannot win by + * outliving the value. A write with no recorded fences, such as caching a freshly inserted entity, + * is written unconditionally. + * + * <p><b>Failure policy.</b> Reads and fills are optimizations: a Redis error or timeout makes them + * a miss or a no-op. An invalidation is a correctness obligation: a failure is propagated as a + * {@link RuntimeException} so it is never silently dropped, and the entry expires by TTL at the + * latest. An unreachable Redis at startup fails fast. + */ +public class RedisEntityCache extends BaseEntityCache { + + private static final Logger LOG = LoggerFactory.getLogger(RedisEntityCache.class); + + /** Per-thread bound on recorded fences awaiting the write that consumes them. */ + private static final int MAX_PENDING_FENCES = 1024; + + private static final long FAILURE_LOG_INTERVAL_MS = 30_000L; + private static final int SCAN_BATCH = 500; + + /** + * Reads a value, or on a miss the fences guarding a later fill. {@code KEYS[1]} is the value key; + * {@code ARGV} lists the fence keys. Returns {@code {1, value}} on a hit and {@code {0, + * fence...}} on a miss, with an absent fence reported as {@code "0"}. + */ + @VisibleForTesting + static final String READ_SCRIPT = + "local v = redis.call('GET', KEYS[1])\n" + + "if v then return {1, v} end\n" + + "local r = {0}\n" + + "for i = 1, #ARGV do r[#r + 1] = redis.call('GET', ARGV[i]) or '0' end\n" + + "return r\n"; + + /** + * Writes a value unless a guarding fence moved. {@code KEYS[1]} is the index key and {@code + * KEYS[2]} the value key; {@code ARGV[1]} is the index member, {@code ARGV[2]} the value, {@code + * ARGV[3]} the TTL in milliseconds (0 for none), followed by (fence key, expected value) pairs. + * Returns 1 if written and 0 if rejected. + */ + @VisibleForTesting + static final String PUT_SCRIPT = + "for i = 4, #ARGV, 2 do\n" + + " if (redis.call('GET', ARGV[i]) or '0') ~= ARGV[i + 1] then return 0 end\n" + + "end\n" + + "if tonumber(ARGV[3]) > 0 then\n" + + " redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])\n" + + "else\n" + + " redis.call('SET', KEYS[2], ARGV[2])\n" + + "end\n" + + "redis.call('ZADD', KEYS[1], 0, ARGV[1])\n" + + "return 1\n"; + + /** + * Drops an entry and its indexed descendants. {@code KEYS[1]} is the index key; {@code ARGV[1]} + * is the slot prefix, {@code ARGV[2]} the index member, {@code ARGV[3]} the identifier whose + * fence to bump, {@code ARGV[4]} the fence TTL in milliseconds (0 for none), followed by the + * descendant prefixes to range over. The upper bound {@code prefix\xff} is above every UTF-8 + * member starting with the prefix. Returns the number of values deleted. + */ + @VisibleForTesting + static final String DROP_SCRIPT = + "local idx = KEYS[1]\n" + + "local prefix = ARGV[1]\n" + + "local fence = prefix .. 'F:' .. ARGV[3]\n" + + "redis.call('INCR', fence)\n" + + "if tonumber(ARGV[4]) > 0 then redis.call('PEXPIRE', fence, ARGV[4]) end\n" + + "local removed = redis.call('DEL', prefix .. 'D:' .. ARGV[2])\n" + + "redis.call('ZREM', idx, ARGV[2])\n" + + "for i = 5, #ARGV do\n" + + " local members = redis.call('ZRANGEBYLEX', idx, '[' .. ARGV[i], " + + "'[' .. ARGV[i] .. '\\255')\n" + + " for _, m in ipairs(members) do\n" + + " removed = removed + redis.call('DEL', prefix .. 'D:' .. m)\n" + + " redis.call('ZREM', idx, m)\n" + + " end\n" + + "end\n" + + "return removed\n"; + + private final UnifiedJedis jedis; + private final RedisKeyspace keyspace; + private final KryoEntitySerializer serializer; + private final SegmentedLock segmentedLock; + private final long valueTtlMs; + private final long fenceTtlMs; + private final byte[] readScript; + private final byte[] putScript; + private final byte[] dropScript; + private final AtomicLong lastFailureLogMs = new AtomicLong(); + + /** + * Fences observed by a read miss on this thread, keyed by index member, consumed by the write + * that fills the entry. Bounded and access-ordered so an entry whose write never happens (the + * store had no such entity) is eventually evicted; a leftover entry can only make a later write + * more conservative, never less. + */ + private final ThreadLocal<Map<String, FenceSnapshot>> pendingFences = + ThreadLocal.withInitial( + () -> + new LinkedHashMap<String, FenceSnapshot>(16, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, FenceSnapshot> eldest) { + return size() > MAX_PENDING_FENCES; + } + }); + + /** + * Constructs a new {@link RedisEntityCache} connected to the Redis deployment described by the + * {@code gravitino.cache.redis.*} configuration. + * + * @param cacheConfig the cache configuration + */ + public RedisEntityCache(Config cacheConfig) { + this(cacheConfig, createClient(cacheConfig)); + probe(cacheConfig.get(Configs.CACHE_REDIS_ADDRESS)); + } + + /** + * Constructs a new {@link RedisEntityCache} over an existing client, without probing it. + * + * @param cacheConfig the cache configuration + * @param jedis the Redis client, standalone or cluster + */ + @VisibleForTesting + RedisEntityCache(Config cacheConfig, UnifiedJedis jedis) { + super(cacheConfig); + Preconditions.checkArgument(jedis != null, "jedis must not be null"); + this.jedis = jedis; + this.keyspace = new RedisKeyspace(cacheConfig.get(Configs.CACHE_REDIS_NAMESPACE)); + // Only the Kryo serializer exists; the config entry validates the name. + cacheConfig.get(Configs.CACHE_REDIS_SERIALIZER); + this.serializer = new KryoEntitySerializer(); + this.segmentedLock = new SegmentedLock(cacheConfig.get(Configs.CACHE_LOCK_SEGMENTS)); + this.valueTtlMs = cacheConfig.get(Configs.CACHE_EXPIRATION_TIME); + long configuredFenceTtl = cacheConfig.get(Configs.CACHE_REDIS_FENCE_TTL_MS); + this.fenceTtlMs = configuredFenceTtl == 0 ? 2 * valueTtlMs : configuredFenceTtl; + Preconditions.checkArgument( + valueTtlMs == 0 || fenceTtlMs > valueTtlMs, + "%s (%s ms) must exceed %s (%s ms)", + Configs.CACHE_REDIS_FENCE_TTL_MS.getKey(), + fenceTtlMs, + Configs.CACHE_EXPIRATION_TIME.getKey(), + valueTtlMs); + this.readScript = utf8(READ_SCRIPT); + this.putScript = utf8(PUT_SCRIPT); + this.dropScript = utf8(DROP_SCRIPT); + } + + /** {@inheritDoc} */ + @Override + public Coherence coherence() { + return Coherence.SHARED; + } + + /** + * {@inheritDoc} + * + * <p>On a miss, records the fences guarding this key for the calling thread so that the write + * which follows can be rejected if a drop commits in between. A Redis failure is a miss. + */ + @Override + public <E extends Entity & HasIdentifier> Optional<E> getIfPresent( + NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<String> fencePaths = RedisKeyspace.fencePaths(ident, schemaSeparator()); + List<byte[]> fenceKeys = Lists.newArrayListWithCapacity(fencePaths.size()); + for (String path : fencePaths) { + fenceKeys.add(utf8(keyspace.fenceKey(ident, path))); + } + + List<?> reply; + try { + reply = + (List<?>) + jedis.eval(readScript, ImmutableList.of(utf8(keyspace.valueKey(key))), fenceKeys); + } catch (JedisException e) { + logFailure("read", key, e); + return Optional.empty(); + } + + if (((Long) reply.get(0)) == 1L) { + byte[] bytes = (byte[]) reply.get(1); + try { + return Optional.of(convertEntity(serializer.deserialize(bytes))); + } catch (RuntimeException e) { + LOG.warn("Discarding cache entry {} that could not be deserialized", key, e); + discardQuietly(key); + return Optional.empty(); + } + } + + List<byte[]> epochs = Lists.newArrayListWithCapacity(fenceKeys.size()); + for (int i = 1; i < reply.size(); i++) { + epochs.add((byte[]) reply.get(i)); + } + pendingFences.get().put(RedisKeyspace.member(key), new FenceSnapshot(fenceKeys, epochs)); + return Optional.empty(); + } + + /** + * {@inheritDoc} + * + * <p>Bumps the fence of the identifier and deletes the entry together with every indexed + * descendant in one atomic script. A Redis failure is propagated, because a dropped invalidation + * would leave a stale value readable by every node. + */ + @Override + public boolean invalidate(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + List<byte[]> args = Lists.newArrayList(); + args.add(utf8(keyspace.slotPrefix(ident))); + args.add(utf8(RedisKeyspace.member(key))); + args.add(utf8(ident.toString())); + args.add(utf8(Long.toString(fenceTtlMs))); + for (String prefix : RedisKeyspace.descendantPrefixes(key, schemaSeparator())) { + args.add(utf8(prefix)); + } + List<byte[]> keys = ImmutableList.of(utf8(keyspace.indexKey(ident))); + + return segmentedLock.withLock( + key, + () -> { + try { + jedis.eval(dropScript, keys, args); + return true; + } catch (JedisException e) { + throw new RuntimeException( + "Failed to invalidate entity cache entry " + + key + + " in Redis; a stale entry may remain readable until it expires", + e); + } + }); + } + + /** {@inheritDoc} A Redis failure is reported as absent. */ + @Override + public boolean contains(NameIdentifier ident, Entity.EntityType type) { + checkArguments(ident, type); + EntityCacheKey key = EntityCacheKey.of(ident, type); + try { + return jedis.exists(utf8(keyspace.valueKey(key))); + } catch (JedisException e) { + logFailure("contains", key, e); + return false; + } + } + + /** + * {@inheritDoc} + * + * <p>Sums the index of every metalake, scanning each cluster node, so the result is a + * point-in-time estimate that may briefly exceed the number of live values: an index member is + * removed by the next drop that ranges over it, not when its value expires. + */ + @Override + public long size() { + AtomicLong total = new AtomicLong(); + forEachNode( + node -> + scan( + node, + keyspace.allIndexKeysPattern(), + indexKey -> total.addAndGet(node.zcard(indexKey)))); + return total.get(); + } + + /** + * {@inheritDoc} + * + * <p>Removes every value and index key of this namespace on every node. Fences are left in place + * so that a load in flight during the clear is still rejected; they expire on their own. + */ + @Override + public void clear() { + segmentedLock.withGlobalLock( + () -> { + pendingFences.get().clear(); + forEachNode( + node -> + scan( + node, + keyspace.allKeysPattern(), + key -> { + // One key per command: on a cluster node the keys scanned belong to many + // slots, and a multi-key UNLINK across slots is rejected. + if (!RedisKeyspace.isFenceKey(key)) { Review Comment: [P1] Clearing values and the index separately can break later hierarchical invalidation The global lock is local to this cache instance. This interleaving is possible across nodes: 1. A's `clear()` unlinks a value. 2. B misses and atomically fills its value and IDX member. 3. A unlinks IDX. The value is now live but unindexed. A subsequent successful metalake/catalog invalidation cannot find it, and a cache hit does not validate ancestor fences. I reproduced this ordering against Redis and confirmed that the value survives a later metalake invalidation. Keeping existing fences unchanged also does not reject in-flight fills as the method comment claims. Please make clearing atomic with fills per metalake, including advancing a guarding generation and maintaining value/index consistency. Also check the separate `UNLINK`/`ZREM` in `discardQuietly()`, which can remove a concurrent fill's index member. Add a deterministic clear/fill/parent-invalidation regression test. ########## core/src/test/java/org/apache/gravitino/cache/integration/test/RedisEntityCacheTestBase.java: ########## @@ -0,0 +1,524 @@ +/* + * 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.integration.test; + +import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.gravitino.Config; +import org.apache.gravitino.Configs; +import org.apache.gravitino.Entity; +import org.apache.gravitino.HasIdentifier; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; +import org.apache.gravitino.cache.CacheFactory; +import org.apache.gravitino.cache.Coherence; +import org.apache.gravitino.cache.EntityCache; +import org.apache.gravitino.cache.EntityCacheKey; +import org.apache.gravitino.cache.RedisEntityCache; +import org.apache.gravitino.meta.AuditInfo; +import org.apache.gravitino.meta.BaseMetalake; +import org.apache.gravitino.meta.CatalogEntity; +import org.apache.gravitino.meta.ColumnEntity; +import org.apache.gravitino.meta.SchemaEntity; +import org.apache.gravitino.meta.SchemaVersion; +import org.apache.gravitino.meta.TableEntity; +import org.apache.gravitino.rel.types.Types; +import org.apache.gravitino.utils.HierarchicalSchemaUtil; +import org.apache.gravitino.utils.TestUtil; +import org.awaitility.Awaitility; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import redis.clients.jedis.UnifiedJedis; + +/** + * Behavior of {@link RedisEntityCache} against a real Redis, shared by the standalone and cluster + * suites. Each test uses its own key namespace, and "nodes" are separate cache instances over the + * same Redis, which is exactly what two Gravitino servers sharing one cache are. + */ +public abstract class RedisEntityCacheTestBase { + + protected static final String SEP = HierarchicalSchemaUtil.schemaSeparator(); + private static final long DEFAULT_TTL_MS = 60_000L; + + private final List<RedisEntityCache> caches = new ArrayList<>(); + protected String namespace; + + /** The {@code gravitino.cache.redis.address} value for the Redis under test. */ + protected abstract String address(); + + /** Whether the Redis under test is a cluster. */ + protected abstract boolean cluster(); + + /** A raw client for inspecting keys directly. */ + protected abstract UnifiedJedis rawClient(); + + @BeforeEach + void newNamespace() { + namespace = "it-" + UUID.randomUUID().toString().substring(0, 8); + } + + @AfterEach + void closeNodes() { + for (RedisEntityCache cache : caches) { + try { + cache.clear(); + } finally { + cache.close(); + } + } + caches.clear(); + } + + protected Config config(long ttlMs) { + Config config = new Config(false) {}; + config.set(Configs.CACHE_IMPLEMENTATION, "redis"); + config.set(Configs.CACHE_REDIS_ADDRESS, address()); + config.set(Configs.CACHE_REDIS_CLUSTER, cluster()); + config.set(Configs.CACHE_REDIS_NAMESPACE, namespace); + config.set(Configs.CACHE_EXPIRATION_TIME, ttlMs); + return config; + } + + /** A cache instance standing in for one Gravitino node. */ + protected RedisEntityCache newNode() { + return newNode(DEFAULT_TTL_MS); + } + + protected RedisEntityCache newNode(long ttlMs) { + EntityCache cache = CacheFactory.getEntityCache(config(ttlMs)); + Assertions.assertInstanceOf(RedisEntityCache.class, cache); + caches.add((RedisEntityCache) cache); + return (RedisEntityCache) cache; + } + + protected String valueKey(NameIdentifier ident, Entity.EntityType type) { + String metalake = ident.hasNamespace() ? ident.namespace().level(0) : ident.name(); + return namespace + ":{" + metalake + "}:D:" + ident + ":" + type; + } + + protected String indexKey(String metalake) { + return namespace + ":{" + metalake + "}:IDX"; + } + + protected static BaseMetalake metalake(String name) { + return BaseMetalake.builder() + .withId(1L) + .withName(name) + .withVersion(SchemaVersion.V_0_1) + .withComment("c") + .withProperties(ImmutableMap.of()) + .withAuditInfo(audit()) + .build(); + } + + protected static CatalogEntity catalog(String metalake, String name) { + return TestUtil.getTestCatalogEntity(2L, name, Namespace.of(metalake), "hive", "c"); + } + + protected static SchemaEntity schema(String metalake, String catalog, String name) { + return TestUtil.getTestSchemaEntity(3L, name, Namespace.of(metalake, catalog), "c"); + } + + protected static TableEntity table( + String metalake, String catalog, String schema, String name, String comment) { + ColumnEntity column = + ColumnEntity.builder() + .withId(10L) + .withName("id") + .withPosition(0) + .withDataType(Types.LongType.get()) + .withNullable(false) + .withAutoIncrement(false) + .withAuditInfo(audit()) + .build(); + return TableEntity.builder() + .withId(4L) + .withName(name) + .withNamespace(Namespace.of(metalake, catalog, schema)) + .withComment(comment) + .withColumns(ImmutableList.of(column)) + .withAuditInfo(audit()) + .build(); + } + + protected static AuditInfo audit() { + return AuditInfo.builder() + .withCreator("tester") + .withCreateTime(Instant.ofEpochMilli(1_700_000_000_000L)) + .build(); + } + + private static <E extends Entity & HasIdentifier> Optional<E> get( + EntityCache cache, NameIdentifier ident, Entity.EntityType type) { + return cache.getIfPresent(ident, type); + } + + @Test + void testFactoryBuildsASharedCache() { + RedisEntityCache cache = newNode(); + Assertions.assertEquals(Coherence.SHARED, cache.coherence()); + Assertions.assertEquals(0, cache.size()); + } + + @Test + void testCrossNodeFreshness() { + RedisEntityCache nodeA = newNode(); + RedisEntityCache nodeB = newNode(); + CatalogEntity catalog = catalog("m1", "c1"); + + nodeA.put(catalog); + Assertions.assertEquals( + Optional.of(catalog), get(nodeB, catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + Assertions.assertTrue(nodeB.contains(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + + nodeA.invalidate(catalog.nameIdentifier(), Entity.EntityType.CATALOG); + Assertions.assertEquals( + Optional.empty(), get(nodeB, catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + Assertions.assertFalse(nodeB.contains(catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + } + + @Test + void testCatalogDropRemovesDescendantsAndKeepsSiblings() { + RedisEntityCache cache = newNode(); + BaseMetalake metalake = metalake("m1"); + CatalogEntity catalog1 = catalog("m1", "catalog1"); + CatalogEntity catalog10 = catalog("m1", "catalog10"); + SchemaEntity schema = schema("m1", "catalog1", "s1"); + TableEntity table = table("m1", "catalog1", "s1", "t1", "v1"); + + cache.put(metalake); + cache.put(catalog1); + cache.put(catalog10); + cache.put(schema); + cache.put(table); + Assertions.assertEquals(5, cache.size()); + + cache.invalidate(catalog1.nameIdentifier(), Entity.EntityType.CATALOG); + + Assertions.assertTrue(cache.contains(metalake.nameIdentifier(), Entity.EntityType.METALAKE)); + Assertions.assertTrue(cache.contains(catalog10.nameIdentifier(), Entity.EntityType.CATALOG)); + Assertions.assertFalse(cache.contains(catalog1.nameIdentifier(), Entity.EntityType.CATALOG)); + Assertions.assertFalse(cache.contains(schema.nameIdentifier(), Entity.EntityType.SCHEMA)); + Assertions.assertFalse(cache.contains(table.nameIdentifier(), Entity.EntityType.TABLE)); + Assertions.assertEquals(2, cache.size()); + } + + @Test + void testMetalakeDropRemovesEverythingUnderIt() { + RedisEntityCache cache = newNode(); + cache.put(metalake("m1")); + cache.put(catalog("m1", "c1")); + cache.put(schema("m1", "c1", "s1")); + cache.put(table("m1", "c1", "s1", "t1", "v1")); + cache.put(metalake("m2")); + cache.put(catalog("m2", "c1")); + + cache.invalidate(NameIdentifier.of("m1"), Entity.EntityType.METALAKE); + + Assertions.assertEquals(2, cache.size()); + Assertions.assertTrue(cache.contains(NameIdentifier.of("m2"), Entity.EntityType.METALAKE)); + Assertions.assertTrue(cache.contains(NameIdentifier.of("m2", "c1"), Entity.EntityType.CATALOG)); + Assertions.assertFalse( + cache.contains(NameIdentifier.of("m1", "c1", "s1", "t1"), Entity.EntityType.TABLE)); + } + + @Test + void testSchemaDropRemovesNestedSchemasAndKeepsPrefixSiblings() { + RedisEntityCache cache = newNode(); + SchemaEntity raw = schema("m1", "c1", "raw"); + SchemaEntity rawEvents = schema("m1", "c1", "raw" + SEP + "events"); + TableEntity nestedTable = table("m1", "c1", "raw" + SEP + "events", "t1", "v1"); + SchemaEntity raw2 = schema("m1", "c1", "raw2"); + TableEntity raw2Table = table("m1", "c1", "raw2", "t1", "v1"); + + cache.put(raw); + cache.put(rawEvents); + cache.put(nestedTable); + cache.put(raw2); + cache.put(raw2Table); + + cache.invalidate(raw.nameIdentifier(), Entity.EntityType.SCHEMA); + + Assertions.assertFalse(cache.contains(raw.nameIdentifier(), Entity.EntityType.SCHEMA)); + Assertions.assertFalse(cache.contains(rawEvents.nameIdentifier(), Entity.EntityType.SCHEMA)); + Assertions.assertFalse(cache.contains(nestedTable.nameIdentifier(), Entity.EntityType.TABLE)); + Assertions.assertTrue(cache.contains(raw2.nameIdentifier(), Entity.EntityType.SCHEMA)); + Assertions.assertTrue(cache.contains(raw2Table.nameIdentifier(), Entity.EntityType.TABLE)); + } + + @Test + void testStaleWriteIsRejectedAfterTheKeyWasInvalidated() { + RedisEntityCache nodeA = newNode(); + RedisEntityCache nodeB = newNode(); + TableEntity oldTable = table("m1", "c1", "s1", "t1", "old"); + TableEntity newTable = table("m1", "c1", "s1", "t1", "new"); + NameIdentifier ident = oldTable.nameIdentifier(); + + // Node B misses and starts loading the old row from the store... + Assertions.assertEquals(Optional.empty(), get(nodeB, ident, Entity.EntityType.TABLE)); + // ...while node A commits an update and invalidates. + nodeA.put(newTable); + nodeA.invalidate(ident, Entity.EntityType.TABLE); + // Node B's late fill must not resurrect the old row. + nodeB.put(oldTable); + + Assertions.assertFalse(nodeA.contains(ident, Entity.EntityType.TABLE)); + Assertions.assertEquals(Optional.empty(), get(nodeA, ident, Entity.EntityType.TABLE)); + } + + @Test + void testStaleWriteIsRejectedAfterAnAncestorWasDroppedEvenIfNeverIndexed() { + RedisEntityCache nodeA = newNode(); + RedisEntityCache nodeB = newNode(); + TableEntity table = table("m1", "c1", "s1", "t1", "v1"); + + // The table was never cached, so a drop of its catalog finds nothing in the index for it. + Assertions.assertEquals( + Optional.empty(), get(nodeB, table.nameIdentifier(), Entity.EntityType.TABLE)); + nodeA.invalidate(NameIdentifier.of("m1", "c1"), Entity.EntityType.CATALOG); + nodeB.put(table); + + Assertions.assertFalse(nodeA.contains(table.nameIdentifier(), Entity.EntityType.TABLE)); + } + + @Test + void testStaleWriteIsRejectedAfterANestedSchemaParentWasDropped() { + RedisEntityCache nodeA = newNode(); + RedisEntityCache nodeB = newNode(); + TableEntity table = table("m1", "c1", "raw" + SEP + "events", "t1", "v1"); + + Assertions.assertEquals( + Optional.empty(), get(nodeB, table.nameIdentifier(), Entity.EntityType.TABLE)); + nodeA.invalidate(NameIdentifier.of("m1", "c1", "raw"), Entity.EntityType.SCHEMA); + nodeB.put(table); + + Assertions.assertFalse(nodeA.contains(table.nameIdentifier(), Entity.EntityType.TABLE)); + } + + @Test + void testReloadAfterInvalidationIsAccepted() { + RedisEntityCache nodeA = newNode(); + RedisEntityCache nodeB = newNode(); + TableEntity table = table("m1", "c1", "s1", "t1", "v2"); + NameIdentifier ident = table.nameIdentifier(); + + Assertions.assertEquals(Optional.empty(), get(nodeB, ident, Entity.EntityType.TABLE)); + nodeA.invalidate(ident, Entity.EntityType.TABLE); + // A fresh miss observes the new fence, so the load that follows it is current. + Assertions.assertEquals(Optional.empty(), get(nodeB, ident, Entity.EntityType.TABLE)); + nodeB.put(table); + + Assertions.assertEquals(Optional.of(table), get(nodeA, ident, Entity.EntityType.TABLE)); + } + + @Test + void testWriteWithoutAPrecedingMissIsUnconditional() { + RedisEntityCache cache = newNode(); + CatalogEntity catalog = catalog("m1", "c1"); + + cache.invalidate(catalog.nameIdentifier(), Entity.EntityType.CATALOG); + // Caching a freshly inserted entity has no load window to guard. + cache.put(catalog); + + Assertions.assertEquals( + Optional.of(catalog), get(cache, catalog.nameIdentifier(), Entity.EntityType.CATALOG)); + } + + @Test + void testStoreReadPatternUnderCacheLock() { + RedisEntityCache nodeA = newNode(); + RedisEntityCache nodeB = newNode(); + TableEntity table = table("m1", "c1", "s1", "t1", "v1"); + EntityCacheKey key = EntityCacheKey.of(table.nameIdentifier(), Entity.EntityType.TABLE); + + // The RelationalEntityStore#get shape: lock, miss, load, put. + TableEntity loaded = + nodeA.withCacheLock( + key, + () -> { + Optional<TableEntity> cached = + get(nodeA, table.nameIdentifier(), Entity.EntityType.TABLE); + if (cached.isPresent()) { + return cached.get(); + } + nodeA.put(table); + return table; + }); + Assertions.assertEquals(table, loaded); + Assertions.assertEquals( + Optional.of(table), get(nodeB, table.nameIdentifier(), Entity.EntityType.TABLE)); + + // The same shape with another node invalidating between the miss and the put. + nodeA.invalidate(table.nameIdentifier(), Entity.EntityType.TABLE); + nodeA.withCacheLock( + key, + () -> { + Assertions.assertEquals( + Optional.empty(), get(nodeA, table.nameIdentifier(), Entity.EntityType.TABLE)); + nodeB.invalidate(table.nameIdentifier(), Entity.EntityType.TABLE); + nodeA.put(table); + return null; + }); + Assertions.assertFalse(nodeB.contains(table.nameIdentifier(), Entity.EntityType.TABLE)); + } + + @Test + void testConcurrentReadersAndWriters() throws Exception { Review Comment: Please extend the regression coverage to exercise store behavior and distinguish old/new versions This concurrent test repeatedly inserts the same `table` object, so observing stale content still satisfies `assertEquals(table, seen)`. The existing 39 new unit tests pass, but the failure cases in the inline comments are not covered. The most valuable additions for this PR are: - Drive the actual `RelationalEntityStore` insert/get/batchGet paths with two cache instances, distinct v1/v2 contents, and latches controlling DB-load/commit/invalidation/fill ordering. Assert that a successful invalidation cannot be undone by an older fill. - Exercise 1,025 cache misses before batch population; Redis read failure followed by recovery; and decoding failure followed by concurrent invalidation. - Test both absent-fence expiry and `1 -> expiry -> 1` ABA near the expiration boundary. - Interleave clear and another node's fill, then invalidate the parent and verify no live orphan remains. - Verify eventual removal of expired index members and safe cleanup during refill. - Close one store and verify its Redis resources are released while another node's entries remain. - Run cluster size/clear tests with replicas, and namespace isolation tests for `ns` versus `ns:other` and glob characters. These should use deterministic ordering/assertions where possible; a stress loop with identical values cannot establish the stale-write guarantee. ########## core/src/main/java/org/apache/gravitino/cache/RedisKeyspace.java: ########## @@ -0,0 +1,176 @@ +/* + * 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.google.common.base.Preconditions; +import com.google.common.collect.Lists; +import java.util.List; +import java.util.regex.Pattern; +import org.apache.commons.lang3.StringUtils; +import org.apache.gravitino.Entity; +import org.apache.gravitino.NameIdentifier; +import org.apache.gravitino.Namespace; + +/** + * Key layout of {@link RedisEntityCache}. Every key derives from {@link EntityCacheKey#toString()}, + * which is {@code <identifier>:<TYPE>}, for example {@code m1.c1.s1.t1:TABLE}. + * + * <table> + * <caption>Keys written under one namespace</caption> + * <tr><th>Purpose</th><th>Key</th><th>Redis type</th></tr> + * <tr><td>Value</td><td>{@code <ns>:{<metalake>}:D:<identifier>:<TYPE>}</td><td>string</td></tr> + * <tr><td>Fence</td><td>{@code <ns>:{<metalake>}:F:<identifier>}</td><td>string (counter)</td></tr> + * <tr><td>Index</td><td>{@code <ns>:{<metalake>}:IDX}</td><td>sorted set, all scores 0</td></tr> + * </table> + * + * <p>The {@code {<metalake>}} segment is a Redis Cluster hash tag holding the first level of the + * identifier, so every key of one metalake lives in one hash slot and the multi-key scripts that + * drop a container and its descendants never span slots. The index holds one member per cached + * value, {@code <identifier>:<TYPE>}, so the descendants of a container are exactly the members in + * the lexicographic range starting with {@code <identifier>.} (and, for a schema, {@code + * <identifier><schema separator>}). + * + * <p>A fence is a counter keyed by the identifier alone, without the type. A container drop + * increments the fence of the dropped identifier; a write compares the fences of the identifier and + * of every ancestor against the values it observed before it loaded the entity, so a load that + * began before a drop anywhere above it cannot refill the key afterwards. + */ +final class RedisKeyspace { + + /** Separates {@link NameIdentifier} levels inside a key. */ + static final String NAME_LEVEL_BOUNDARY = "."; + + /** Index of the schema level in a fully qualified identifier: metalake, catalog, schema. */ + private static final int SCHEMA_LEVEL = 2; + + private static final String VALUE_MARKER = "D:"; + private static final String FENCE_MARKER = "F:"; + private static final String INDEX_NAME = "IDX"; + + private final String namespace; + + RedisKeyspace(String namespace) { + Preconditions.checkArgument(StringUtils.isNotBlank(namespace), "namespace must not be blank"); + Preconditions.checkArgument( + !namespace.contains("{") && !namespace.contains("}"), + "namespace must not contain '{' or '}'"); + this.namespace = namespace; + } + + /** The hash tag of an identifier: the metalake, which is its first level. */ + static String hashTag(NameIdentifier ident) { + return ident.hasNamespace() ? ident.namespace().level(0) : ident.name(); + } + + /** The index member for a key, {@code <identifier>:<TYPE>}. */ + static String member(EntityCacheKey key) { + return key.toString(); + } + + /** Prefix shared by every key of the identifier's metalake: {@code <ns>:{<metalake>}:}. */ + String slotPrefix(NameIdentifier ident) { + return namespace + ":{" + hashTag(ident) + "}:"; + } + + String indexKey(NameIdentifier ident) { + return slotPrefix(ident) + INDEX_NAME; + } + + String valueKey(EntityCacheKey key) { + return slotPrefix(key.identifier()) + VALUE_MARKER + member(key); + } + + /** + * The fence key of one identifier path, see {@link #fencePaths(NameIdentifier, String)}. + * + * @param ident The identifier whose metalake selects the hash slot + * @param identifierPath The fenced identifier, the entity's own or an ancestor's + */ + String fenceKey(NameIdentifier ident, String identifierPath) { + return slotPrefix(ident) + FENCE_MARKER + identifierPath; + } + + /** Glob pattern matching every key this namespace writes, for {@code SCAN}. */ + String allKeysPattern() { + return namespace + ":*"; Review Comment: [P2] Namespace scanning can delete another deployment's cache Both `isolated` and `isolated:other` are valid namespaces, but clearing the former scans `isolated:*`, which also matches the latter's values and indexes. I reproduced the second deployment's entry disappearing after the first called `clear()`. This can also happen during store shutdown. In addition, namespace validation permits Redis glob metacharacters such as `*`, `?`, and `[]`, which are interpolated without escaping. Please escape the literal namespace in scan patterns and match the complete key boundary (the namespace must be followed immediately by the hash-tag segment), optionally verifying exact ownership before deletion. Add prefix-related and glob-containing namespace isolation tests. -- 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]
