yuqi1129 commented on code in PR #13831:
URL: https://github.com/apache/gravitino/pull/13831#discussion_r4293056994
##########
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:
Thanks. You are right that the interval guard does not cover a replay-only
drain. I'm keeping it as-is for this PR, for two reasons:
- **Cost:** the extra query is `SELECT COALESCE(MAX(id), 0)`, which is a
single primary-key index lookup. That is negligible next to the replay page
query it runs beside.
- **Count:** it is bounded by the pages in one sweep, at most `windowRows /
pollBatchSize - 1` extra calls per sweep. For example, at 100 writes/s the
window holds about 6,000 rows, which means about 2 extra `MAX(id)` calls every
15 s.
Fixing it would add one more piece of "previous cycle had a backlog" state
to a poller that already carries a lot of state, and this PR aims to stay a
short-term fix. If the tail query ever shows up in profiles, the rate limit you
suggest is the right follow-up. The extra query is already documented
(`metrics.md` and the comment at `:320-322`).
##########
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:
Thanks for the careful analysis. The cost you describe is real. A fill is
skipped on any epoch change, and this was an accepted trade-off: a skipped fill
only means the next read goes to the database. The pre-insert sample is still
deliberate, though. A post-insert sample reopens the ancestor race.
`CaffeineEntityCache.invalidate` takes only the ancestor key's segment lock and
scans an index snapshot; it never takes the child's lock:
| step | T1: `put(table t)` | T2: `drop schema s` (cascade) |
|---|---|---|
| 1 | insert `t` holding `FOR SHARE` on `s`, commit | waits for `FOR UPDATE`
on `s` |
| 2 | | cascade-deletes `t`, commits |
| 3 | | `epoch++`, invalidates `s` and its descendants; `t` is not cached
yet |
| 4 | **post-insert sample** reads the new epoch | |
| 5 | takes `t`'s key lock, sees an unchanged epoch, publishes `t` | |
The dropped `t` then stays cached until TTL. That is the stale-publication
case #13828 P0-2 targets. With the sample taken before the insert, step 5 sees
a changed epoch and skips. The key lock and the post-put recheck only exclude a
same-key invalidation, not an ancestor invalidation that has already finished.
A per-key fence would remove the false skips. One option is striped counters
checked along the ancestor chain, with clears still bumping the global epoch.
That is a larger change than this P0 needs, so I'd rather measure the hit-rate
impact first, as part of the #13828 performance work (item 6). Hit rate is the
next step there.
##########
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:
Yes, it was considered. `clock_timestamp()` on the PostgreSQL insert would
make `created_at` statement time, matching MySQL/H2. Since
`EntityChangeLogWriter.append` runs right before commit, that would almost
remove the long-transaction gap.
I'm keeping the existing writer in this PR on purpose:
- **Scope:** the gap only matters for PostgreSQL transactions that take
longer than the window minus one sweep (45 s by default) from start to commit.
Normal metadata transactions take milliseconds to seconds, so the ordering
problem this PR fixes is covered at the default window.
- **Mitigation:** the limitation is documented (PR description and
`gravitino-server-config.md`), and `replayWindowSecs` can be raised, up to
86400.
- **Risk:** changing how `created_at` is stamped on PostgreSQL is a separate
behavior change that needs its own PostgreSQL validation.
If long transactions turn out to lose invalidations in practice, switching
the PostgreSQL insert to `clock_timestamp()` is a small follow-up, and
retention would not be affected.
##########
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:
Good catch on the contract. This is now documented in 5a1312fe35.
`get` has always published through `cache.put` while holding the entity's
key lock, so "the hook must not take another entity's lock" was already an
implicit constraint. The `BaseEntityCache` comment promised the opposite, which
was wrong. The commit:
- states the constraint on `SupportsEntityStoreCache#invalidateOnKeyChange`:
it may run under the inserted entity's own cache lock, so it must not acquire
another entity's lock;
- corrects the comment in `BaseEntityCache.put` to match.
No behavior change: every in-tree implementation is a no-op.
##########
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:
Agreed. Fixed in 5a1312fe35: the inline non-cacheable branch in `batchGet`
is removed, and every entity goes through `cacheIfEpochUnchanged`.
Non-cacheable entities now always invoke the hook without epoch gating. As you
note, an extra hook call can only invalidate more, so that is the safe
direction.
##########
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:
Agreed. Fixed in 5a1312fe35. Until the first replay sweep completes (it may
span several pages), unseen rows are logged at INFO as `Entity change log
startup replay delivered ...; expected after a restart because deduplication
starts empty`. Later sweeps keep the WARN, and its wording no longer lists
startup history as a cause.
`replay-records-total` is unchanged; it already documents that startup
history is included. Tests:
- `testReplayStartupIncludesLowerIdsAndCapacityWarning` now pins the INFO
level and the absence of the WARN. It fails if the INFO branch is removed.
- The new `testUnseenRowAfterFirstSweepLogsLateCommitWarning` pins the WARN
for a row found after the first sweep.
The operator docs are updated as well.
--
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]
