morningman commented on code in PR #69831:
URL: https://github.com/apache/doris/pull/69831#discussion_r4413851390
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java:
##########
@@ -281,16 +288,350 @@ private Optional<EditLogItem>
cleanupStalePartitionOffsets(OlapTableStream strea
return Optional.of(editLogItem);
}
+ /**
+ * Records a keep-alive TSO for the partitions of an IVM stream whose
offsets stay where the partitions were
+ * last consumed while the partitions are not written, so that a read of
such a partition does not take its
+ * offset for expired once the partition is written again (F-2). A
partition qualifies when the base table
+ * applies changelog.ttl_seconds, the stream has a real offset for it past
the initial snapshot, nothing is
+ * committed to the partition after that offset, and the newer of the
offset and its keep-alive TSO is older
+ * than half the TTL.
+ *
+ * <p>The keep-alive TSO is taken from the TSO service while the base
table read lock is held. A commit takes
+ * its TSO inside the base table write lock, so every commit the partition
has not had yet gets a larger one,
+ * and the partition holds no change in (offset, keep-alive TSO]. Locks
are taken in the order
+ * {@link #cleanupStalePartitionOffsets(OlapTableStream)} takes them.
+ */
+ public Optional<EditLogItem> keepAliveIdlePartitions(OlapTableStream
stream) {
+ if (!stream.tryReadLock(Table.TRY_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("skip keeping stream {} alive because stream read
lock is busy", stream.getName());
+ }
+ return Optional.empty();
+ }
+ // stream read lock is held
+ OlapTable baseTable;
+ Map<Long, Long> offsets;
+ Map<Long, Long> effectiveOffsets = new HashMap<>();
+ try {
+ if (stream.isDisabled() || stream.isStale()) {
+ return Optional.empty();
+ }
+ baseTable = stream.getBaseTableNullable();
+ if (baseTable == null) {
+ return Optional.empty();
+ }
+ offsets = stream.getConsumedPartitionOffsets();
+ offsets.forEach((partitionId, offset) ->
effectiveOffsets.put(partitionId,
+ OlapTableStream.effectiveOffset(offset,
stream.getKeepAlive(partitionId))));
+ } finally {
+ stream.readUnlock();
+ }
+ // stream read lock is released
+ if (offsets.isEmpty()) {
+ return Optional.empty();
+ }
+ while (DebugPointUtil.getDebugParamOrDefault(
+
"TableStreamManager.keepAliveIdlePartitions.blockBeforeBaseTableLock", -1L) ==
stream.getId()) {
+ LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
+ }
+ if (!baseTable.tryReadLock(Table.TRY_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("skip keeping stream {} alive because base table {}
read lock is busy",
+ stream.getName(), baseTable.getName());
+ }
+ return Optional.empty();
+ }
+ Map<Long, Long> keptAlive = new HashMap<>();
+ long keepAliveTso;
+ EditLogItem editLogItem;
+ try {
+ if (baseTable.isDropped || !baseTable.hasChangelogTtl()) {
+ return Optional.empty();
+ }
+ long renewCutoffTso =
keepAliveRenewCutoffTso(baseTable.getChangelogConfig().getTtlSeconds());
+ Set<Long> idle = new HashSet<>();
+ for (Map.Entry<Long, Long> entry : offsets.entrySet()) {
+ Partition partition = baseTable.getPartition(entry.getKey());
+ // A commit after the offset shows in the partition TSO once
it is visible, and in the next
+ // version from the moment it is committed.
+ if (partition != null && effectiveOffsets.get(entry.getKey())
<= renewCutoffTso
+ && partition.getTso() <= entry.getValue()
+ && partition.getNextVersion() ==
partition.getVisibleVersion() + 1) {
+ idle.add(entry.getKey());
+ }
+ }
+ if (idle.isEmpty()) {
+ return Optional.empty();
+ }
+ keepAliveTso = Env.getCurrentEnv().getTSOService().getTSO();
+ if (!stream.tryWriteLockIfExist(Table.TRY_LOCK_TIMEOUT_MS,
TimeUnit.MILLISECONDS)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("skip keeping stream {} alive because it is busy
or dropped", stream.getName());
+ }
+ return Optional.empty();
+ }
+ try {
+ if (stream.isDisabled() || stream.isStale()) {
+ return Optional.empty();
+ }
+ for (Long partitionId : idle) {
+ // A consumption between the two stream locks moved the
offset, and the partition waits for
+ // the next round.
+ if
(offsets.get(partitionId).equals(stream.getPartitionOffset(partitionId))) {
+ keptAlive.put(partitionId, offsets.get(partitionId));
+ }
+ }
+ if (keptAlive.isEmpty()) {
+ return Optional.empty();
+ }
+ stream.unprotectedKeepAlive(keptAlive, keepAliveTso);
+ editLogItem =
Env.getCurrentEnv().getEditLog().logTableStreamCleanup(
+ TableStreamCleanupInfo.ofKeepAlive(new
TableStreamCleanupInfo.PartitionKeepAliveEntry(
+ stream.getDatabase().getId(), stream.getId(),
keptAlive, keepAliveTso)));
+ } finally {
+ stream.writeUnlock();
+ }
+ } finally {
+ baseTable.readUnlock();
+ }
+ LOG.info("kept {} idle partition offsets of stream {}.{} ({}) alive up
to tso {}",
+ keptAlive.size(), stream.getDatabase().getFullName(),
stream.getName(), stream.getId(),
+ keepAliveTso);
+ return Optional.of(editLogItem);
+ }
+
+ // Renewed once less than half the TTL is left: the newer of the offset
and the keep-alive TSO would have
+ // expired under half the TTL. A TTL longer than the clock has run never
expires anything, and this cutoff is
+ // 0 then.
+ private static long keepAliveRenewCutoffTso(long ttlSeconds) {
+ return
TSOTimestamp.calculateCutoff(TSOTimestamp.composePhysicalTimestamp(System.currentTimeMillis()),
+ ttlSeconds / 2);
+ }
+
+ /**
+ * The cloud mode counterpart of {@link
#keepAliveIdlePartitions(OlapTableStream)}, for the IVM streams of one
+ * round, whose offsets live in MetaService and are read from it in
batches.
+ *
+ * <p>The keep-alive TSO is the committed TSO watermark, taken before any
of the offsets is read. Every commit
+ * at or below the watermark has finished by then, so a read made
afterwards shows all of them, and a commit
+ * that lands later gets a larger TSO. A partition whose offset is at its
end in that read thus holds no change
+ * in (offset, watermark], for good. A partition qualifies on what
shared-nothing mode checks, as MetaService
+ * reports it -- a real offset past the initial snapshot, at the end of
the partition, whose newer of it and its
+ * keep-alive TSO is older than half the TTL -- when the watermark is past
that newer one. The keep-alive TSO is
+ * recorded in FE metadata with the offset it was taken against, as in
shared-nothing mode, and counts only for
+ * a read from that offset: an offset MetaService no longer reports, after
a consumption or a restore, leaves it
+ * out.
+ */
+ public List<EditLogItem>
keepAliveIdleCloudPartitions(List<OlapTableStream> streams) {
+ return keepAliveIdleCloudPartitions(streams,
CLOUD_KEEP_ALIVE_BATCH_PARTITIONS);
+ }
+
+ @VisibleForTesting
+ List<EditLogItem> keepAliveIdleCloudPartitions(List<OlapTableStream>
streams, int batchPartitions) {
+ List<CloudKeepAliveSnapshot> snapshots = new ArrayList<>();
+ for (OlapTableStream stream : streams) {
+ snapshotForCloudKeepAlive(stream).ifPresent(snapshots::add);
+ }
+ if (snapshots.isEmpty()) {
+ return Collections.emptyList();
+ }
+ TSOService.TSOStatusSnapshot tsoStatus =
Env.getCurrentEnv().getTSOService().getStatusSnapshot();
+ if (!tsoStatus.isInitialized() || tsoStatus.getCommittedTso() <= 0) {
+ LOG.info("skip keeping IVM stream offsets alive because the
committed TSO is not available");
+ return Collections.emptyList();
+ }
+ long keepAliveTso = tsoStatus.getCommittedTso();
Review Comment:
Confirmed. MetaService sets `fdb_txn_timeout_ms` on every FDB transaction, a
1031 during the commit comes back as `KV_TXN_COMMIT_ERR`, and `commitTxn()`
unregisters that attempt with `abandonCommitTso()`, so the watermark can pass a
commit that still lands. A round that reads the offsets in that window records
a K that hides the commit, and the refresh after it skips the commit for good.
Fixed in 19b8e521923: `keepAliveIdleCloudPartitions` now advances
MetaService's TSO fence to K (`advanceTsoFence`) before it reads any offset. A
commit reads the fence key in its own FDB transaction and is rejected at or
below the fence, so once the fence is advanced, every commit at or below K has
either landed, and the read that follows shows it, or can no longer land. A
round whose fence fails records nothing. The cost is one fence write a round:
commits being applied at that moment conflict on the key once and are retried
through the existing `KV_TXN_CONFLICT` loop. `TableStreamManagerKeepAliveTest`
now checks that the fence is advanced to the watermark once a round, before any
read, and that a round whose fence fails reads and records nothing; both cases
fail on the code before the fix. The javadoc and the PR description no longer
claim that every commit at or below the watermark has finished.
The commit path itself still treats that timeout as a definite failure. That
also affects the bounded cloud `@incr` reads, which admit an end at or below
the watermark, and it predates this PR (#67820), so it is left to the commit
path and will be followed up separately. The keep-alive round no longer depends
on it.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java:
##########
@@ -65,6 +68,13 @@ public OlapTableStreamWrapper(OlapTableStream stream,
OlapTable baseTable, List<
this.baseTable = baseTable;
this.keysType = baseTable.getKeysType();
this.outputUpdateMap = buildOutputUpdateMap(selectedPartitionIds);
+ this.keepAlives = Maps.newHashMap();
+ for (Long partitionId : selectedPartitionIds) {
+ OlapTableStream.PartitionKeepAlive keepAlive =
stream.getKeepAlive(partitionId);
Review Comment:
Confirmed. `IvmDeltaRewriteState` looks the IVM stream up and builds the
wrapper after `CollectRelation` has collected the statement's tables, so the
stream is never locked there, and the keeper's `put` can resize the `HashMap`
under that `get`. The effect is the one you describe: a needless COMPLETE
fallback, or a failed strict refresh, when the missed K is what kept an expired
offset readable. No row is read wrong.
Fixed in 6cb029cd6c7: `partitionKeepAlive` is now declared as a
`ConcurrentMap`, so a `get` never misses an existing entry while another one is
added, and the planner path takes no new lock. The field is declared as one,
not only initialized as one, so GSON also reads it back from an image as a
concurrent map, as it does `Database.name2Function`.
`testKeepAliveTsoReadsBackFromTheImage` checks that, and it fails with the
field declared as a plain `Map`. The race itself has no deterministic test.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
