jerryshao commented on code in PR #13831:
URL: https://github.com/apache/gravitino/pull/13831#discussion_r4291811437
##########
core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java:
##########
@@ -226,6 +228,7 @@ public boolean exists(NameIdentifier ident,
Entity.EntityType entityType) throws
@Override
public <E extends Entity & HasIdentifier> void put(E e, boolean overwritten)
throws IOException, EntityAlreadyExistsException {
+ long epochBeforeWrite = cacheInvalidationEpoch.get();
Review Comment:
[Important] This samples the fence before `backend.insert`, so the
publication of `e` is discarded if the epoch moved at any point during the
whole write round trip. `cacheInvalidationEpoch` is a single process-wide
counter (`:100`) incremented by every `invalidateCache` (`:693`) -- once per
local put/update/delete, and once per change-log row delivered from another
node, because the poller's target routes through this same method
(`newCacheChangeLogListener`, `:173-188`). So an invalidation of a completely
unrelated entity suppresses the caching of a value this node just wrote and
knows to be current, and a poller batch of N rows bumps the counter N times, so
any write (or point read, `:273-276`) overlapping a delivery burst loses its
cache fill. Before this PR both `put` and `get` published unconditionally, so
this is a new cost on cache fill rate, paid under exactly the write load where
the cache matters most.
The same-key race the fence targets is already excluded here by the key lock
taken in `cacheIfEpochUnchanged` plus the post-`put` recheck in
`publishIfEpochUnchanged` (`:680-686`): a same-key delete that commits before
this insert has already bumped the epoch before the sample, and one that
commits after will invalidate the entry once it gets the key lock. So the
pre-write sample buys nothing for `put` that a sample taken after
`backend.insert` would not.
Two options worth considering: sample the epoch after the insert for `put`,
or make the fence per key (a per-key invalidation version, or a per-segment
counter) so unrelated invalidations stop suppressing publication on all three
paths. I have not measured the hit-rate impact, so the magnitude is open -- but
the mechanism is unconditional.
Verified by: reading `put` (`:228-245`), the epoch field (`:100`),
`invalidateCache` (`:690-695`) and `newCacheChangeLogListener` (`:173-188`) at
this head.
##########
core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java:
##########
@@ -296,14 +369,205 @@ boolean pollChanges() {
private synchronized boolean doPollChanges() {
BatchDelivery delivery = fetchNextDelivery();
- if (delivery == null) {
+ if (replayWindowMs > 0) {
+ expireRememberedIds();
+ }
+ boolean incrementalBacklog = delivery != null && delivery.changes.size()
>= batchSize;
+ if (delivery != null) {
+ deliver(delivery);
+ remember(delivery.changes);
+ }
+ // Replay failure must not discard or delay delivery of a successfully
fetched new batch.
+ boolean replayBacklog = false;
+ if (replayWindowMs > 0
+ && entityPollHighWaterId > 0
+ && (replayUpperId != 0 || replaySweepDue())
+ && (!replayFailed
+ || nanoTime() - replayFailureNanos >=
TimeUnit.SECONDS.toNanos(pollIntervalSecs))) {
+ try {
+ replayBacklog = replayRecentChanges();
+ replayFailed = false;
+ } catch (RuntimeException e) {
+ if (handleInterruptIfAny(e, "Entity change log replay")) {
+ return false;
+ }
+ replayFailed = true;
+ metrics.replayFailed();
+ replayFailureNanos = nanoTime();
+ LOG.warn(
+ "Entity change log replay failed cursor={} replayAfterCreatedAt={}
replayAfterId={} "
+ + "replayUpperId={} windowMs={}; incremental delivery
continues; replay retries after the poll interval",
+ entityPollHighWaterId,
+ replayAfterCreatedAt,
+ replayAfterId,
+ replayUpperId,
+ replayWindowMs,
+ e);
+ }
+ }
+ return incrementalBacklog || replayBacklog;
Review Comment:
[Important] A full replay page makes this return `true`, so `runPollCycle`
sets `nextDelayMillis = 0` (`:322`) and the next cycle starts at once. But
`fetchNextDelivery` samples the database tail whenever `changes.size() <
batchSize` (`:581-583`), and during a replay-only drain the incremental query
is empty, so that first disjunct is always true. The once-per-interval guard
whose comment at `:584-585` says \"Sample at most once per interval during a
drain\" therefore never applies to a replay-driven drain: every zero-delay
cycle issues three queries back to back (incremental, tail, replay). A sweep
that needs more than one page -- window rows above `pollBatchSize`, 2000 by
default, or any cold-start sweep over a busy window -- runs that triple at full
speed with no delay between cycles, which is exactly the query amplification
the guard was added to bound.
The comment at `:320-322` and metrics.md both note the three queries, so
this is clearly known; noting the cost is not the same as wanting it. Suggest
extending the existing rate limit to cover immediately-rescheduled cycles, e.g.
skip the tail sample when the previous cycle returned a backlog and `nowNanos -
lastTailSampleNanos < pollIntervalSecs`, regardless of which of the two queries
was full. The tail is observability-only (`:577-578`), so a 15s-stale gauge
during a drain costs nothing.
Verified by: reading `doPollChanges` (`:372-408`), `runPollCycle`
(`:313-325`) and `fetchNextDelivery` (`:574-600`) at this head, and confirming
no test asserts tail-query count for a replay drain.
##########
core/src/main/java/org/apache/gravitino/storage/relational/mapper/provider/postgresql/EntityChangeLogPostgreSQLProvider.java:
##########
@@ -60,4 +60,10 @@ public String pruneOldEntityChanges(@Param("retentionMs")
long retentionMs) {
+ ENTITY_CHANGE_LOG_PRUNE_BATCH_SIZE
+ ")";
}
+
+ /** {@inheritDoc} */
+ @Override
+ protected String currentTimeMillisSql() {
+ return CURRENT_TIME_MILLIS_SQL;
Review Comment:
[Question] This override points the replay window filter at
`CAST(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP) * 1000 AS BIGINT)`
(`DatabaseTimeSQL.java:43-44`), and `insertEntityChange` stamps `created_at`
with the same expression (`:35`, `:45-50`). On PostgreSQL `CURRENT_TIMESTAMP`
is the *transaction start* time, so a transaction that begins, writes its
change-log row, and commits 51s later records `created_at` at its start: by the
time the row becomes visible its recorded age already exceeds `replayWindowSecs
- sweepInterval` (45s with defaults) and replay never sees it. That is
precisely the long-transaction late-commit case this PR exists to fix, and
PostgreSQL is the only backend with this semantics -- MySQL/H2
`CURRENT_TIMESTAMP` is statement time, so there the window behaves as
documented.
The PR description and `docs/gravitino-server-config.md` both call this out,
so it is a known limitation rather than an oversight. Was `clock_timestamp()`
(statement time) considered for the PostgreSQL insert path? It would restore
the full window for long transactions. If it was rejected because
`pruneOldEntityChanges` (`:51-62`) and `retentionSecs` would then be measured
from statement time instead, saying so in the doc paragraph would help
operators who are sizing the window against their transaction durations.
Verified by: reading the PostgreSQL provider in full, `DatabaseTimeSQL`
(`:42-45`), and the new `selectRecentEntityChanges` filter
(`EntityChangeLogBaseSQLProvider.java:72-88`).
##########
core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java:
##########
@@ -303,18 +306,7 @@ public <E extends Entity & HasIdentifier> List<E> batchGet(
allEntities.add(entity);
continue;
}
- cache.withCacheLock(
- EntityCacheKey.of(entity.nameIdentifier(), entity.type()),
- () -> {
- if (cacheInvalidationEpoch.get() == epochBeforeRead) {
- cache.put(entity);
- // Invalidation of another key or an ancestor can advance the
epoch while this key
- // lock is held. Remove the value if that happened during the
put.
- if (cacheInvalidationEpoch.get() != epochBeforeRead) {
- cache.invalidate(entity.nameIdentifier(), entity.type());
- }
- }
- });
+ cacheIfEpochUnchanged(entity, epochBeforeRead);
Review Comment:
[Nit] This call reaches `cacheIfEpochUnchanged`, whose own first branch
(`:663-668`) repeats the non-cacheable check that the inline branch just above
(`:300-307`) already handled with a `continue`. So for `batchGet` only the
inline copy ever runs, and the two copies disagree on policy: the inline one
gates `cache.put` on the epoch, the helper one does not. Since `put` (`:243`)
reaches the helper's branch, the ungated version is live for that path.
Deleting the inline branch and letting the helper decide would remove the
duplication and force one answer to the real question: should the
related-invalidation hook call be epoch-gated for an entity that is never
cached? (It looks harmless either way -- an extra hook invocation only
invalidates more -- which is an argument for keeping just the ungated helper.)
Verified by: reading `batchGet` (`:281-311`) and the two helpers
(`:662-688`) at this head.
##########
core/src/main/java/org/apache/gravitino/storage/relational/EntityChangeLogPoller.java:
##########
@@ -296,14 +369,205 @@ boolean pollChanges() {
private synchronized boolean doPollChanges() {
BatchDelivery delivery = fetchNextDelivery();
- if (delivery == null) {
+ if (replayWindowMs > 0) {
+ expireRememberedIds();
+ }
+ boolean incrementalBacklog = delivery != null && delivery.changes.size()
>= batchSize;
+ if (delivery != null) {
+ deliver(delivery);
+ remember(delivery.changes);
+ }
+ // Replay failure must not discard or delay delivery of a successfully
fetched new batch.
+ boolean replayBacklog = false;
+ if (replayWindowMs > 0
+ && entityPollHighWaterId > 0
+ && (replayUpperId != 0 || replaySweepDue())
+ && (!replayFailed
+ || nanoTime() - replayFailureNanos >=
TimeUnit.SECONDS.toNanos(pollIntervalSecs))) {
+ try {
+ replayBacklog = replayRecentChanges();
+ replayFailed = false;
+ } catch (RuntimeException e) {
+ if (handleInterruptIfAny(e, "Entity change log replay")) {
+ return false;
+ }
+ replayFailed = true;
+ metrics.replayFailed();
+ replayFailureNanos = nanoTime();
+ LOG.warn(
+ "Entity change log replay failed cursor={} replayAfterCreatedAt={}
replayAfterId={} "
+ + "replayUpperId={} windowMs={}; incremental delivery
continues; replay retries after the poll interval",
+ entityPollHighWaterId,
+ replayAfterCreatedAt,
+ replayAfterId,
+ replayUpperId,
+ replayWindowMs,
+ e);
+ }
+ }
+ return incrementalBacklog || replayBacklog;
+ }
+
+ private boolean replayRecentChanges() {
+ if (replayUpperId == 0) {
+ replaySweepStarted = true;
+ lastReplaySweepStartNanos = nanoTime();
+ replayUpperId = entityPollHighWaterId;
+ }
+ List<EntityChangeRecord> page =
+ SessionUtils.getWithoutCommit(
+ EntityChangeLogMapper.class,
+ m ->
+ m.selectRecentEntityChanges(
+ replayAfterCreatedAt, replayAfterId, replayUpperId,
replayWindowMs, batchSize));
+ List<EntityChangeRecord> unseen = new ArrayList<>();
+ for (EntityChangeRecord change : page) {
+ if (!rememberedIds.containsKey(change.getId())) {
+ unseen.add(change);
+ // Absence from this bounded set is evidence of a possible late
commit, not proof:
+ // startup history and capacity eviction can also produce previously
unseen IDs.
+ LOG.debug(
+ "Replaying previously unseen entity change behind cursor:
changeId={} cursor={} "
+ + "startupCursor={} entityType={} fullName={} operation={}
createdAt={} approximateAgeMs={} "
+ + "windowMs={}; possible late commit, startup history, or
dedup eviction; "
+ + "age uses JVM time and is not transaction duration",
+ change.getId(),
+ entityPollHighWaterId,
+ startupHighWaterId,
+ change.getEntityType(),
+ change.getFullName(),
+ change.getOperateType(),
+ change.getCreatedAt(),
+ System.currentTimeMillis() - change.getCreatedAt(),
+ replayWindowMs);
+ }
+ }
+ if (!unseen.isEmpty()) {
+ metrics.replayRecordsFound(unseen.size());
+ unseen.sort(Comparator.comparingLong(EntityChangeRecord::getId));
+ LOG.warn(
Review Comment:
[Nit] On the first sweep after `start()` the dedup set is necessarily empty
-- `start()` only sets the cursor and logs the config (`:236-250`) -- so every
row inside the window is \"previously unseen\" and this WARN fires with up to
20 record summaries (`MAX_SUMMARIZED_RECORDS`, `:98`) on every single restart,
with nothing wrong. That trains operators to ignore the one log line that is
meant to flag a real late commit.
`replaySweepStarted` is still `false` when the first sweep is entered
(`:412-415`), so the cold-start case is cheap to distinguish: log at INFO with
an explicit \"startup history\" wording for the first sweep of the process and
keep WARN for later ones, where an unseen row really does mean a late commit or
a dedup eviction. `testReplayStartupIncludesLowerIdsAndCapacityWarning`
(`core/src/test/java/org/apache/gravitino/storage/relational/TestEntityChangeLogPoller.java:396`)
already covers the behaviour and could pin the level.
Verified by: reading `start()` (`:233-255`), `replayRecentChanges`
(`:410-500`) and the dedup set's initial state at this head.
##########
core/src/main/java/org/apache/gravitino/storage/relational/RelationalEntityStore.java:
##########
@@ -667,6 +659,34 @@ private static Entity.EntityType
relationUpdateTargetType(Type relType) {
}
}
+ private <E extends Entity & HasIdentifier> void cacheIfEpochUnchanged(E
entity, long epoch) {
+ if (cache instanceof BaseEntityCache &&
!BaseEntityCache.isCacheable(entity.type())) {
+ // Uncached entities still invoke the cache's related-invalidation hook.
There is no stale
+ // value to publish, and taking this entity's key lock can nest locks
inside that hook.
+ cache.put(entity);
+ return;
+ }
+ cache.withCacheLock(
+ EntityCacheKey.of(entity.nameIdentifier(), entity.type()),
+ () -> publishIfEpochUnchanged(entity, epoch));
Review Comment:
[Question] This takes the entity's own key lock and then calls
`cache.put(entity)` inside it, which runs `BaseEntityCache.put` ->
`invalidateOnKeyChange(entity)` while that lock is held.
`BaseEntityCache.put`'s own comment
(`core/src/main/java/org/apache/gravitino/cache/BaseEntityCache.java:109-113`)
states the hook is called deliberately \"before any subclass takes this
entity's lock\" because \"it may take another key's lock (e.g. the model key
when inserting a model version), and nesting two entity locks could deadlock\".
The non-cacheable branch immediately above avoids exactly this and says so
(`:663-667`), but the cacheable path still nests.
Nothing deadlocks today: every in-tree implementation of the hook is a no-op
(`CaffeineEntityCache.java:266-271`, `NoOpsCache.java:110`), and
`get`/`batchGet` already called `cache.put` inside the key lock before this PR.
What is new is `put` (`:243`), which previously published with no lock at all.
So the question is about the contract rather than current behaviour: is the
intent now that an implementation overriding `invalidateOnKeyChange` must never
acquire another entity's key lock? If so, that constraint would be worth
stating on `SupportsEntityStoreCache.invalidateOnKeyChange` (`:77-95`), since
the base-class comment currently promises the opposite guarantee.
Verified by: reading `cacheIfEpochUnchanged`/`publishIfEpochUnchanged`
(`:662-688`), `BaseEntityCache.put` (`:105-120`),
`CaffeineEntityCache.doPut`/`invalidateOnKeyChange` (`:243-271`) and
`SegmentedLock` (`:32-140`) at this head.
--
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]
