morningman commented on code in PR #69831:
URL: https://github.com/apache/doris/pull/69831#discussion_r4437848252
##########
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:
Follow-up on the commit path: #69925 makes `commitTxn()` advance the TSO
fence before it gives up the TSO of a commit that MetaService answers with
`KV_TXN_COMMIT_ERR`, as it already does when the RPC fails, so the watermark no
longer passes a commit that can still land. The keep-alive round keeps its own
fence; it does not rely on that fix.
--
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]
