morningman commented on code in PR #69831:
URL: https://github.com/apache/doris/pull/69831#discussion_r4423900660
##########
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:
The format change is intended. The PR description says so under Persistence:
"Table Stream is not released, so the old fields are not read: a stream from an
image written by an earlier build reads back with no partition state and has to
be dropped and recreated."
No release carries Table Stream: `OlapTableStream` is on none of branch-4.0,
branch-4.1 and branch-4.2, and `enable_table_stream` is experimental, off by
default and not mutable. So there is no rolling upgrade from a release to
cover. A cluster that ran an earlier master build with the switch on drops and
recreates its streams and IVM MVs; we do not carry a migration for a format
that was never released.
The consequence you describe is right, and it is why such a stream has to be
recreated rather than kept. In shared-nothing mode, where the offsets live only
in FE metadata, an IVM refresh takes a partition without an offset for one the
MV has not read yet: it reads the partition from the base table and appends it,
so the MV gets those rows a second time and the refresh still succeeds. A user
stream reads its retained changelog again.
##########
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:
The mechanism is right: `lastRoundMs` is set before the round, the three
paths you list record nothing (the first two for the whole round, a failed
`getReadStates` for its batch), and the next attempt comes one
`ivm_stream_keepalive_interval_second` later. Shared-nothing mode has the same
shape (a `getTSO()` failure, a busy lock).
Leaving a failed round to the next one is intended. Such a round records
nothing (`testCloudKeepAliveSkipsRoundWhenTheFenceFails`,
`testCloudKeepAliveSkipsRoundWhenReadStatesFail`), and the next round that
succeeds renews a partition that is still idle even if its K has expired by
then: neither mode's condition asks whether K is within the TTL, and "nothing
was committed after the offset" holds however old K is. What a failed round can
cost is the refresh that reads a write made after it, which falls back to
COMPLETE as before this PR (a strict INCREMENTAL refresh fails); no row is read
wrong.
How much that costs is a matter of margin. When an idle partition is
written, the newer of its offset and K is at most TTL/2 + one interval old, and
one interval older for each round that could not renew it. After the write the
partition is not renewed any more, and the refresh that reads the write has to
start before that TSO is a TTL old. So that refresh is incremental if it starts
within TTL/2 - interval after the write, one interval less per failed round.
With the defaults (a TTL of a day, 600 seconds) that is 11 h 50 min, and a
refresh right after the write still reads incrementally after 71 failed rounds
in a row.
What this comment does show is that the bound in the config description was
wrong: "at most half the shortest changelog.ttl_seconds". At half, no time is
left even when every round succeeds, and a refresh shortly after the write
falls back; retrying failed rounds sooner would not change that. Fixed in
00befb02e5e: the description of `ivm_stream_keepalive_interval_second` now
states the time a refresh has after a write, less one interval for each check
that could not renew the partition, and asks for an interval well under half
the TTL. The PR description says the same.
A faster retry is not added. A cloud round advances MetaService's TSO fence
before it reads, so retrying sooner while MetaService is failing writes to it
more often, and the commits being applied at that moment conflict on each fence
once. The interval is the retry period; more margin comes from a shorter
interval, which costs more checks but no more edit logs.
--
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]
