github-actions[bot] commented on code in PR #69831:
URL: https://github.com/apache/doris/pull/69831#discussion_r4422394915
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java:
##########
@@ -43,30 +44,28 @@
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
import java.util.function.Predicate;
public class OlapTableStream extends BaseTableStream {
- @SerializedName("po")
- private Map<Long, Long> partitionOffset;
+ // partition id -> what the stream holds for the partition. A state is
replaced as a whole under the stream write
+ // lock, never changed in place, and it is also read without the lock: an
IVM refresh reads the stream through a
+ // scan it creates after the statement has locked its tables
(OlapTableStreamWrapper). So it is a concurrent map,
+ // and such a read sees one whole state of a partition.
Review Comment:
[P1] Preserve table-stream partition state across image upgrades
`OlapTableStream` used to persist consumed offsets, historical TSOs, and
consumption times in `po`, `pct`, and `hpt`. This head reads only the new `ps`
field and has no `alternate` annotations or migration, so an image/checkpoint
written by the parent commit deserializes every existing stream with an empty
`partitionStates` map. After an FE restart or rolling upgrade, a user stream
can replay retained changes and an IVM stream is treated as unconsumed; the
next refresh can reread current rows into an already populated MV, while
expired history is no longer represented. Replay only fills partitions that
consume after the restart. Preserve the old fields through Gson migration
before exposing the stream state.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java:
##########
@@ -281,16 +288,364 @@ 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, and MetaService
is fenced at it before any of the
+ * offsets is read. Every commit at or below the watermark is unregistered
from the TSO service by then, which
+ * does not mean it is over: one the FE gave up on after a response it may
outlive, such as a MetaService
+ * transaction that timed out once its commit was sent, can still land.
The fence settles every such commit --
+ * it has landed before the fence, and a read made afterwards shows it, or
it can no longer land -- 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();
+ // One fence a round, before any of its reads. The commits MetaService
is applying at that moment conflict
+ // on it once and are retried, as on any commit conflict.
+ long fenceTso;
+ try {
+ fenceTso =
Env.getCurrentEnv().getGlobalTransactionMgr().advanceTsoFence(keepAliveTso);
Review Comment:
[P2] Retry failed cloud keep-alive rounds
`lastRoundMs` is advanced before the cloud keep-alive work, but this path
treats several failures as a completed round: an initialized TSO service with
`committedTso <= 0`, a failed `advanceTsoFence`, and a `getReadStates`
`UserException` all return without recording a keep-alive. The daemon then
waits the full `ivm_stream_keepalive_interval_second` before trying again. With
the documented interval up to half the table TTL, an offset that entered the
renewal margin just before a transient MetaService failure can expire before
the retry, making strict INCREMENTAL fail or AUTO fall back to COMPLETE.
Propagate round success/failure and do not consume the interval on an
unsuccessful cloud round (with suitable backoff).
--
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]
