github-actions[bot] commented on code in PR #69831:
URL: https://github.com/apache/doris/pull/69831#discussion_r4410495902
##########
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:
[P1] Fence timed-out commits before using the cloud watermark
`getCommittedTso()` does not currently guarantee that every earlier commit
has stopped being in flight. A configured FDB transaction timeout (1031)
becomes `TXN_TIMEOUT` -> `KV_TXN_COMMIT_ERR`;
`CloudGlobalTransactionMgr.commitTxn()` removes that transaction with
`abandonCommitTso()` (lines 974-975), without the fence used for RPC
exceptions. FoundationDB explicitly allows a timed-out commit to [finish
later](https://apple.github.io/foundationdb/developer-guide.html#non-retryable-errors).
For an idle partition at C, commit T>C can time out, be unregistered, and
let the durable watermark advance to K>=T. If this round reads C=endTso before
that commit finishes, it persists (C,K). Once the delayed commit publishes T,
the next refresh still matches C but `applyChangelogTtl()` starts after K,
silently dropping that committed change; an expired C that previously forced
COMPLETE now passes the TTL check. Fence indeterminate commit responses before
unregistering them, or establish an equivalent MS fence before using K as a
permanent no-change proof. Please cover the timeout-response path, not only RPC
exceptions.
##########
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:
[P2] Snapshot keep-alives under the stream read lock
The IVM refresh starts with a plan such as:
```
LogicalOlapTableSink(mv)
LogicalAggregate(...)
LogicalOlapScan(base)
```
`CollectRelation`/`StatementContext.lock()` lock the base and target tables,
but the stream is looked up later by `IvmDeltaRewriteState` when it constructs
this wrapper. That path never acquires the stream read lock. Meanwhile the
keeper writes `OlapTableStream.partitionKeepAlive`, a plain `HashMap`, under
the stream write lock while holding a compatible base-table read lock.
A first keep-alive for another partition can resize that map during this
`get`, transiently hiding an existing fresh K. The wrapper permanently captures
its absence; when C is older than the TTL, `applyChangelogTtl()` rejects it and
AUTO needlessly rebuilds the whole MV (strict INCREMENTAL fails). Capture the
offsets and keep-alives under the stream read lock, or publish a
concurrency-safe snapshot; explicitly bound SQL streams are locked, but these
synthesized IVM readers are not.
--
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]
