This is an automated email from the ASF dual-hosted git repository.
luwei16 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new efc929aa7af [fix](table stream) fix table stream TSO boundary
semantics (#67480)
efc929aa7af is described below
commit efc929aa7af6ced10d8a4c7764cfc22fc6bee543
Author: TsukiokaKogane <[email protected]>
AuthorDate: Thu Sep 10 20:11:13 2026 +0800
[fix](table stream) fix table stream TSO boundary semantics (#67480)
### What problem does this PR solve?
Issue Number: close #67093
Related PR: #xxx
Problem Summary:
Timestamp-based `@incr` reads are intended to use a left-closed,
right-open interval: `[startTimestamp, endTimestamp)`.
Previously, FE converted both timestamps to the last logical TSO of
their respective milliseconds, while BE applied `GT/LE` predicates. This
produced `(endOfStartMs, endOfEndMs]`.
Consequently, the scan excluded changes in the requested start
millisecond and included changes in the requested end millisecond. For
example, a query for `[10:00:00, 10:00:01)` excluded changes at
`10:00:00.000` and included changes at `10:00:01.000`, assigning
boundary events to the wrong window.
This PR changes BE TSO predicates to `GE/LT` and updates FE bound
construction across incremental reads, Table Streams, snapshots, time
travel, and transaction visibility waiting. Timestamp-based reads now
follow the intended half-open interval, while Table Stream consumption
and historical snapshots retain their existing inclusive committed-TSO
endpoints.
#### Scan bounds and persisted offsets
Table Stream metadata continues to store actual committed TSO values.
Only scan-facing bounds are converted:
```text
(lastConsumed, currentTso]
==
[nextTso(lastConsumed), nextTso(currentTso))
```
The conversion happens when constructing the scan view. The underlying
stream update map and offset commit paths retain the original committed
TSOs, preventing scan-bound conversion from advancing persisted offsets
or being applied again on the next consumption.
For the table below:
- `startOf(ms)` means `composeEmptyCounterTSO(ms)`, with logical counter
zero.
- `nextTso(t)` means the successor of a valid real TSO.
- `L`, `C`, and `H` are the last consumed, current committed, and
historical committed TSOs.
| Read path | Scan-facing bounds | Resulting semantics |
| --- | --- | --- |
| `@incr(startTimestamp, endTimestamp)` | `[startOf(startMs),
startOf(endMs))` | Includes the entire start millisecond and excludes
the entire end millisecond. |
| `@incr` without a start timestamp | Starts at `startOf(0)` | Reads
from the beginning up to the applicable exclusive end. |
| Incremental scan without an explicit end | Uses `nextTso(C)` when a
current committed TSO exists | Includes changes at the current committed
TSO. |
| Local and Cloud Table Stream consumption | `[nextTso(L), nextTso(C))`
| Preserves `(L, C]`. |
| First incremental consumption without a recorded lower bound | Lower
bound unset; end is `nextTso(C)` | Includes changes through `C`. |
| DUP snapshot/history scan | Upper bound `nextTso(H)` | Includes rows
committed at `H`. |
| MOW snapshot reconstruction | Base rows below `nextTso(H)`; binlog
changes starting at `nextTso(H)` | Preserves the snapshot at `H`, with
subsequent changes used to reconstruct before-images. |
| `FOR VERSION AS OF v` | Exclusive snapshot boundary `nextTso(v)` |
Keeps the requested version inclusive. |
| `FOR TIME AS OF ms` | Exclusive snapshot boundary
`nextTso(startOf(ms))` | Includes every logical TSO within the requested
millisecond. |
For MOW time travel, the base branch uses `commit_tso < boundary`, and
the binlog reconstruction branch starts at `commit_tso >= boundary`.
Both branches therefore use the same snapshot boundary.
#### Missing offsets and empty snapshot partitions
A missing bound or a negative sentinel such as `-1` is not a real
committed TSO. `toExclusiveBound` converts these values to `null`; valid
committed TSOs are converted to their successors. When a scan has no
explicit end, the planner converts the current partition TSO in the same
way and leaves the end unset if the partition has never received a real
TSO.
Snapshot partition selection also distinguishes a real consumption
baseline from an empty-partition marker. A partition that was empty when
the stream was created can have a recorded offset of `-1`; in Cloud
mode, it can additionally be marked `CONSUMED`. Neither condition alone
means that the partition contained data at the snapshot boundary.
This PR requires a positive recorded offset for such partitions to
participate in the consumed-partition snapshot path. Cloud mode also
requires the offset field to be present. This prevents snapshot
reconstruction from falling back to the partition's current TSO and
exposing rows inserted after the snapshot into a partition that was
originally empty. Incremental reads can still consume those later
changes through their normal path.
#### Visibility waiting
The non-cloud visibility waiter now uses `commitTSO < endTSO`, matching
the exclusive upper-bound representation.
- An explicit positive end timestamp `E` maps to `startOf(E)`, excluding
transactions in millisecond `E`.
- When the end is omitted, or the existing non-positive fallback
applies, query-start millisecond `M` maps to `startOf(M + 1)`. This
includes all logical counters within `M`, so relevant transactions
committed in the same millisecond are still awaited.
The Cloud visibility waiter is unchanged and continues to use its
transaction-ID watermark.
#### TSO range and sentinel handling
`Long.MAX_VALUE` is reserved as `UNBOUNDED_TSO`, representing the
latest/unbounded version. Real allocated TSOs are limited to
`MAX_REAL_TSO = Long.MAX_VALUE - 1`.
`nextTso` validates its input and returns `tso + 1`. It does not
saturate: passing `Long.MAX_VALUE` is rejected. `FOR VERSION AS OF
Long.MAX_VALUE` handles the sentinel explicitly and keeps it as the
exclusive upper bound above all real TSOs.
The TSO generator now uses `composeRealTso` to validate physical and
logical components and reject values outside the real-TSO range,
enforcing the invariant required by successor conversion.
These changes preserve the stored offset representation. They change the
interpretation of scan bounds exchanged between FE and BE, so unchanged
persisted metadata does not imply compatibility between old and new
FE/BE scan semantics.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [x] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
be/src/exec/scan/olap_scanner.cpp | 5 +-
.../doris/catalog/stream/OlapTableStream.java | 8 ++-
.../catalog/stream/OlapTableStreamWrapper.java | 28 ++++++--
.../doris/nereids/rules/analysis/BindRelation.java | 42 ++++++++----
.../org/apache/doris/planner/OlapScanNode.java | 24 +++----
.../main/java/org/apache/doris/tso/TSOService.java | 2 +-
.../java/org/apache/doris/tso/TSOTimestamp.java | 74 +++++++++++++++++++---
.../translator/PhysicalPlanTranslatorTest.java | 19 ++++--
.../trees/plans/ExplainTableStreamPlanTest.java | 72 +++++++++++++++++++--
.../org/apache/doris/planner/OlapScanNodeTest.java | 26 ++++++++
.../org/apache/doris/tso/TSOTimestampTest.java | 15 +++++
11 files changed, 261 insertions(+), 54 deletions(-)
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index 4fafd8ffb30..d0d3a03bee9 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -332,13 +332,14 @@ Status OlapScanner::_init_tso_predicates() {
const auto* tso_column = read_schema->column(tso_ordinal);
const auto& tso_data_type = read_schema->data_type(tso_ordinal);
+ // The TSO scan range is left-closed right-open [start_tso, end_tso).
if (_start_tso.has_value()) {
-
_tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::GT>(
+
_tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::GE>(
tso_ordinal, tso_column->name(), tso_data_type,
Field::create_field<TYPE_BIGINT>(*_start_tso), false));
}
if (_end_tso.has_value()) {
-
_tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::LE>(
+
_tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::LT>(
tso_ordinal, tso_column->name(), tso_data_type,
Field::create_field<TYPE_BIGINT>(*_end_tso), false));
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
index 20f6bf915e8..ec6fde1a97f 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java
@@ -230,7 +230,13 @@ public class OlapTableStream extends BaseTableStream {
}
public boolean hasConsumedData(long partitionId) {
- return partitionOffset.containsKey(partitionId);
+ // A partition that was empty at stream creation is recorded with the
sentinel offset -1
+ // (see initializeLocalOffsets); a real committed TSO is always
positive (its physical part
+ // is non-zero). So only a positive recorded offset counts as a real
consumption baseline.
+ // This keeps empty partitions out of the snapshot scan instead of
letting them fall back to
+ // the live partition TSO and leak post-snapshot rows.
+ Long offset = partitionOffset.get(partitionId);
+ return offset != null && offset > 0;
}
public Pair<Long, Long> getStreamUpdate(Long partitionId) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java
index 3fc9ca11a6a..d6ed6f2411a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStreamWrapper.java
@@ -27,6 +27,7 @@ import org.apache.doris.common.Pair;
import org.apache.doris.common.util.Util;
import org.apache.doris.thrift.TColumn;
import org.apache.doris.thrift.TPrimitiveType;
+import org.apache.doris.tso.TSOTimestamp;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -306,8 +307,17 @@ public class OlapTableStreamWrapper extends OlapTable {
public List<Long> filterConsumedPartitionIds(List<Long> partitionIds) {
if (hasCloudReadStates()) {
return partitionIds.stream()
- .filter(id -> cloudReadStates.get(id).getOffsetState()
- ==
Cloud.TableStreamOffsetStatePB.TABLE_STREAM_OFFSET_CONSUMED)
+ .filter(id -> {
+ Cloud.TableStreamPartitionReadStatePB state =
cloudReadStates.get(id);
+ // A partition empty at stream creation is recorded as
CONSUMED with the
+ // sentinel offset -1 (CloudInternalCatalog:
emptyPartition -> commit_tso=-1).
+ // Such a partition has no real consumption baseline;
exclude it so snapshot
+ // rebuild does not fall back to the live TSO and leak
post-snapshot rows.
+ // Mirrors the non-cloud hasConsumedData() offset > 0
guard.
+ return state.getOffsetState()
+ ==
Cloud.TableStreamOffsetStatePB.TABLE_STREAM_OFFSET_CONSUMED
+ && state.hasOffsetTso() &&
state.getOffsetTso() > 0;
+ })
.collect(ImmutableList.toImmutableList());
}
return partitionIds.stream()
@@ -333,14 +343,24 @@ public class OlapTableStreamWrapper extends OlapTable {
public Map<Long, Pair<Long, Long>> getPartitionOffsets(List<Long>
selectedPartitionIds) {
return outputUpdateMap.entrySet().stream()
.filter(s -> selectedPartitionIds.contains(s.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey,
Map.Entry::getValue));
+ .collect(Collectors.toMap(Map.Entry::getKey, s -> {
+ // Storage keeps the real committed TSO points
(closed-interval semantics).
+ // BE scans a left-closed right-open range [startTso,
endTso), so convert the
+ // bounds only in this scan-facing read view.
outputUpdateMap and the offset
+ // commit path (toOlapTableStreamUpdate) stay on the
real-TSO coordinate system.
+ Pair<Long, Long> v = s.getValue();
+ return Pair.of(TSOTimestamp.toExclusiveBound(v.first),
+ TSOTimestamp.toExclusiveBound(v.second));
+ }));
}
// get history partition offsets partitionId -> (null,
historicalTimestampOffset)
public Map<Long, Pair<Long, Long>> getHistoryPartitionOffsets(List<Long>
selectedPartitionIds) {
return outputUpdateMap.entrySet().stream()
.filter(s -> selectedPartitionIds.contains(s.getKey()))
- .collect(Collectors.toMap(Map.Entry::getKey, s ->
Pair.of(null, s.getValue().first)));
+ // historicalTso is an inclusive upper bound; shift to the
half-open exclusive end.
+ .collect(Collectors.toMap(Map.Entry::getKey,
+ s -> Pair.of(null,
TSOTimestamp.toExclusiveBound(s.getValue().first))));
}
public List<Long> filterNormalSnapshotPartitionIds(List<Long>
partitionIds) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
index 7cafa1bc918..60479e23a8b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
@@ -74,7 +74,7 @@ import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
import org.apache.doris.nereids.trees.expressions.InPredicate;
-import org.apache.doris.nereids.trees.expressions.LessThanEqual;
+import org.apache.doris.nereids.trees.expressions.LessThan;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
@@ -502,8 +502,8 @@ public class BindRelation extends OneAnalysisRuleFactory {
/**
* Build the time-travel (FOR VERSION/TIME AS OF) plan for an olap scan.
- * dup: Filter(__DORIS_COMMIT_TSO_COL__ <= targetTso).
- * mow: base(survived rows, tso<=t1) UNION ALL binlog(before-image of
UPDATE_BEFORE/DELETE).
+ * dup: Filter(__DORIS_COMMIT_TSO_COL__ < targetTso), targetTso being
the exclusive upper bound.
+ * mow: base(survived rows, tso < targetTso) UNION ALL
binlog(before-image of UPDATE_BEFORE/DELETE).
*/
private LogicalPlan buildTimeTravelPlan(LogicalOlapScan scan, OlapTable
olapTable,
TableSnapshot snapshot, UnboundRelation unboundRelation,
List<String> qualifier,
@@ -541,8 +541,9 @@ public class BindRelation extends OneAnalysisRuleFactory {
}
/**
- * Add Filter(__DORIS_COMMIT_TSO_COL__ <= targetTso) on top of {@code
child}. The tso slot is
- * resolved from {@code child} output; {@code child} must pass through the
scan output slots.
+ * Add Filter(__DORIS_COMMIT_TSO_COL__ < targetTso) on top of {@code
child}, where targetTso is
+ * the right-open (exclusive) upper bound from resolveSnapshotTso. The tso
slot is resolved from
+ * {@code child} output; {@code child} must pass through the scan output
slots.
*/
private LogicalPlan addCommitTsoFilter(LogicalPlan child, long targetTso,
OlapTable olapTable) {
Slot tsoSlot = null;
@@ -554,25 +555,35 @@ public class BindRelation extends OneAnalysisRuleFactory {
}
Preconditions.checkArgument(tsoSlot != null,
"%s not found on table %s", Column.COMMIT_TSO_COL,
olapTable.getQualifiedName());
- Expression conjunct = new LessThanEqual(tsoSlot, new
BigIntLiteral(targetTso));
+ Expression conjunct = new LessThan(tsoSlot, new
BigIntLiteral(targetTso));
return new LogicalFilter<>(ImmutableSet.of(conjunct), child);
}
/**
- * Resolve a TableSnapshot to a target commit tso (inclusive upper bound).
- * VERSION: the literal is the tso itself. TIME: wall-clock string ->
ms -> tso upper bound.
+ * Resolve a TableSnapshot to the right-open (exclusive) commit-tso upper
bound: the scan keeps
+ * rows with commit_tso < the returned value. VERSION: literal tso + 1
(so the literal itself is
+ * included). TIME: successor of (requested millisecond, logical counter
0), so that exact TSO
+ * is included but larger logical counters in the same millisecond are
excluded. Used uniformly
+ * by the dup filter, the mow union left filter and the mow union
right-branch lower bound.
*/
private long resolveSnapshotTso(TableSnapshot snapshot) {
if (snapshot.getType() == TableSnapshot.VersionType.VERSION) {
+ long version;
try {
- return Long.parseLong(snapshot.getValue().trim());
+ version = Long.parseLong(snapshot.getValue().trim());
} catch (NumberFormatException e) {
throw new AnalysisException(
"Invalid version in FOR VERSION AS OF: " +
snapshot.getValue());
}
+ // UNBOUNDED_TSO (Long.MAX_VALUE) is the "latest / unbounded"
sentinel: keep it as the
+ // exclusive upper bound above every real TSO. A real version is
converted to its
+ // right-open successor so that commit_tso < result includes the
requested version.
+ return version == TSOTimestamp.UNBOUNDED_TSO
+ ? TSOTimestamp.UNBOUNDED_TSO
+ : TSOTimestamp.nextTso(version);
}
long ms = OlapScanNode.parseChangeTimestamp(snapshot.getValue());
- return TSOTimestamp.composeFullTimestamp(ms);
+ return TSOTimestamp.nextTso(TSOTimestamp.composePhysicalTimestamp(ms));
}
/**
@@ -589,14 +600,16 @@ public class BindRelation extends OneAnalysisRuleFactory {
|| ((SlotReference) slot).isVisible())
.collect(Collectors.toList());
- // left: base survived rows at t1 = delete_sign=0 AND commit_tso<=t1,
projected to visible.
+ // left: base survived rows at t1 = delete_sign=0 AND commit_tso <
targetTso, projected to visible.
LogicalPlan left = checkAndAddDeleteSignFilter(baseScan,
ConnectContext.get(), olapTable, true);
left = projectFromOriginSlots(addCommitTsoFilter(left, targetTso,
olapTable), visibleOutput);
- // right: binlog MIN_DELTA over tso>t1, keep UPDATE_BEFORE/DELETE rows
(before image),
+ // right: binlog MIN_DELTA over tso >= targetTso, keep
UPDATE_BEFORE/DELETE rows (before image),
// projected to the same visible schema. BE splits each change so
UPDATE_BEFORE/DELETE rows
// already carry the pre-change value in the (same-named) value
columns.
RowBinlogTableWrapper binlogTable = new
RowBinlogTableWrapper(olapTable, CollectionUtils.isEmpty(partIds)
+ // targetTso is the exclusive upper bound; the right branch
reads tso >= targetTso,
+ // seamlessly meeting the left branch's commit_tso < targetTso.
? makeUniformedTimestampRangeMap(olapTable.getPartitionIds(),
Pair.of(targetTso, null)) :
makeUniformedTimestampRangeMap(partIds, Pair.of(targetTso,
null)));
RelationId binlogRelationId =
cascadesContext.getStatementContext().getNextRelationId();
@@ -709,10 +722,11 @@ public class BindRelation extends OneAnalysisRuleFactory {
private Pair<Long, Long> parseTimestampRange(TableScanParams scanParams) {
Map<String, String> params = scanParams.getMapParams();
+ // @incr reads a left-closed right-open range [startTso, endTso): BE
applies GE/LT directly.
+ // composePhysicalTimestamp maps a millisecond to its start (logical
counter 0), so GE includes
+ // the whole startMs and LT excludes the whole endMs. No +1 shift is
needed here.
Long startTimestamp = OlapScanNode.parseChangeTimestamp(
params.getOrDefault(OlapScanNode.OLAP_START_TIMESTAMP, "0"));
- // BE applies start_tso < commit_tso <= end_tso. Logical zero on both
boundaries
- // therefore gives the user-facing physical interval [startTimestamp,
endTimestamp).
startTimestamp = TSOTimestamp.composePhysicalTimestamp(startTimestamp);
Long endTimestamp = null;
if (params.containsKey((OlapScanNode.OLAP_END_TIMESTAMP))) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 3418137d57c..fcbd9151002 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -100,6 +100,7 @@ import org.apache.doris.thrift.TScanRange;
import org.apache.doris.thrift.TScanRangeLocation;
import org.apache.doris.thrift.TScanRangeLocations;
import org.apache.doris.thrift.TSortInfo;
+import org.apache.doris.tso.TSOTimestamp;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
@@ -581,13 +582,17 @@ public class OlapScanNode extends ScanNode {
parseBinlogScanType(scanParams, ((OlapTableWrapper)
olapTable).getOriginTable());
Pair<Long, Long> update =
getPartitionOffset(partition.getId());
if (update != null) {
+ // push down tso range as half-open [startTso, endTso)
bounds
if (update.first != null) {
paloRange.setStartTso(update.first);
}
- if (update.second != null) {
- paloRange.setEndTso(update.second);
- } else {
- paloRange.setEndTso(partition.getTso());
+ // No end recorded: fall back to the current committed
TSO. toExclusiveBound
+ // returns null for a partition that never got a real TSO
(getTso() == -1), in
+ // which case we leave endTso unset (no upper bound)
instead of using -1.
+ Long endTso = update.second != null
+ ? update.second :
TSOTimestamp.toExclusiveBound(partition.getTso());
+ if (endTso != null) {
+ paloRange.setEndTso(endTso);
}
}
if (binlogScanType != TBinlogScanType.NONE) {
@@ -1964,14 +1969,6 @@ public class OlapScanNode extends ScanNode {
return scanParams;
}
- public long getIncrementalScanEndTime() {
- if (scanParams != null && scanParams.incrementalRead()
- && scanParams.getMapParams().containsKey(OLAP_END_TIMESTAMP)) {
- return
parseChangeTimestamp(scanParams.getMapParams().get(OLAP_END_TIMESTAMP));
- }
- return 0;
- }
-
public static long parseChangeTimestamp(String ts) {
if (ts != null) {
long changeTimestamp;
@@ -1983,6 +1980,9 @@ public class OlapScanNode extends ScanNode {
if (changeTimestamp < 0) {
throw new ParseException("Invalid TIMESTAMP format in incr
clause: " + ts);
}
+ if (changeTimestamp > TSOTimestamp.MAX_PHYSICAL_TIMESTAMP) {
+ throw new ParseException("Timestamp exceeds supported TSO
range: " + ts);
+ }
return changeTimestamp;
}
throw new ParseException("Invalid timestamp:" + ts);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
b/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
index 2acfff15564..bb4d3e0e949 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tso/TSOService.java
@@ -243,7 +243,7 @@ public class TSOService extends MasterDaemon {
if (MetricRepo.isInit) {
MetricRepo.COUNTER_TSO_CLOCK_GET_SUCCESS.increase(1L);
}
- return TSOTimestamp.composeTimestamp(physical, logical);
+ return TSOTimestamp.composeRealTso(physical, logical);
}
throw new RuntimeException("Failed to get TSO after " +
maxGetTSORetryCount + " retries", lastFailure);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tso/TSOTimestamp.java
b/fe/fe-core/src/main/java/org/apache/doris/tso/TSOTimestamp.java
index 191bd6f71ea..0afc9eeb11a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tso/TSOTimestamp.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tso/TSOTimestamp.java
@@ -22,6 +22,7 @@ import org.apache.doris.common.io.Writable;
import org.apache.doris.persist.gson.GsonUtils;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import com.google.gson.annotations.SerializedName;
import java.io.DataInput;
@@ -67,6 +68,18 @@ public final class TSOTimestamp implements Writable,
Comparable<TSOTimestamp> {
// Maximum logical counter value
public static final long MAX_LOGICAL_COUNTER = (1L << LOGICAL_BITS) - 1L;
+ // Sentinel meaning "no upper bound" / "latest" (e.g. FOR VERSION AS OF
9223372036854775807).
+ // It is intentionally NOT a real allocated TSO; it sorts above every real
TSO so that a
+ // right-open predicate {@code x < UNBOUNDED_TSO} still selects all rows.
+ public static final long UNBOUNDED_TSO = Long.MAX_VALUE;
+
+ // The largest legal real (allocated) TSO. Real TSOs never reach
UNBOUNDED_TSO, which leaves
+ // room for nextTso() to compute a successor without overflow.
+ public static final long MAX_REAL_TSO = Long.MAX_VALUE - 1;
+
+ // Largest physical millisecond that can be represented by a real TSO.
+ public static final long MAX_PHYSICAL_TIMESTAMP =
extractPhysicalTime(MAX_REAL_TSO);
+
/**
* Constructor with specific physical time and logical counter
*
@@ -104,21 +117,44 @@ public final class TSOTimestamp implements Writable,
Comparable<TSOTimestamp> {
}
/**
- * Compose 64-bit TSO timestamp from physical time
+ * Compose the lower boundary of a physical millisecond.
*
- * @return 64-bit TSO timestamp with full counter
+ * @return 64-bit TSO timestamp with a zero logical counter
*/
- public static long composeFullTimestamp(long physicalTimestamp) {
- return composeTimestamp(physicalTimestamp, LOGICAL_MASK);
+ public static long composePhysicalTimestamp(long physicalTimestamp) {
+ return composeRealTso(physicalTimestamp, 0L);
}
/**
- * Compose the lower boundary of a physical millisecond.
+ * The next discrete TSO after a real {@code tso}. TSO values are dense
integers, so this converts
+ * an inclusive bound into the equivalent right-open (exclusive) bound:
{@code x <= tso} is the
+ * same row set as {@code x < nextTso(tso)}, and a lower bound that
excludes {@code tso} itself is
+ * {@code x >= nextTso(tso)}. Callers should use this instead of a bare
{@code + 1} so the TSO
+ * interval arithmetic stays in one place.
*
- * @return 64-bit TSO timestamp with a zero logical counter
+ * <p>This is a pure successor over real TSOs only. The {@link
#UNBOUNDED_TSO} sentinel is not a
+ * real TSO and must be handled by callers before reaching here; requiring
a real input keeps the
+ * successor free of overflow and makes the "real TSOs never reach the
sentinel" assumption an
+ * enforced invariant rather than a comment.
*/
- public static long composePhysicalTimestamp(long physicalTimestamp) {
- return composeTimestamp(physicalTimestamp, 0L);
+ public static long nextTso(long tso) {
+ Preconditions.checkArgument(tso >= 0 && tso <= MAX_REAL_TSO,
+ "nextTso expects a real TSO in [0, %s], got %s", MAX_REAL_TSO,
tso);
+ return tso + 1;
+ }
+
+ /**
+ * Convert an inclusive stored TSO bound into the half-open (exclusive)
bound the scan pushes
+ * down, tolerating "no bound" inputs. Returns {@code null} when the input
is {@code null} or a
+ * negative sentinel (e.g. a partition that never got a real TSO stores
-1, meaning no committed
+ * change): a {@code null} result tells the caller to leave that bound
unset rather than feeding
+ * a non-real value into {@link #nextTso}. A real TSO is mapped to its
successor.
+ */
+ public static Long toExclusiveBound(Long storedTso) {
+ if (storedTso == null || storedTso < 0) {
+ return null;
+ }
+ return nextTso(storedTso);
}
/**
@@ -215,6 +251,28 @@ public final class TSOTimestamp implements Writable,
Comparable<TSOTimestamp> {
| (logical);
}
+ /**
+ * Compose a real (allocated) TSO from physical time and logical counter,
validating that the
+ * inputs and the result stay within the legal real-TSO range instead of
silently masking. This
+ * is the single construction entry for TSOs produced by the generator, so
the invariant
+ * "a real TSO is non-negative and never reaches {@link #UNBOUNDED_TSO}"
is enforced here once
+ * rather than assumed at every call site.
+ *
+ * @throws IllegalArgumentException if the components are out of range or
the composed value
+ * would exceed {@link #MAX_REAL_TSO}
+ */
+ public static long composeRealTso(long physicalTime, long logicalCounter) {
+ Preconditions.checkArgument(physicalTime >= 0 && physicalTime <=
RAW_PHYSICAL_MASK,
+ "physicalTime out of range [0, %s]: %s", RAW_PHYSICAL_MASK,
physicalTime);
+ Preconditions.checkArgument(logicalCounter >= 0 && logicalCounter <=
MAX_LOGICAL_COUNTER,
+ "logicalCounter out of range [0, %s]: %s",
MAX_LOGICAL_COUNTER, logicalCounter);
+ long tso = Math.addExact(
+ Math.multiplyExact(physicalTime, 1L << PHYSICAL_SHIFT),
logicalCounter);
+ Preconditions.checkArgument(tso <= MAX_REAL_TSO,
+ "composed TSO exceeds MAX_REAL_TSO (%s): %s", MAX_REAL_TSO,
tso);
+ return tso;
+ }
+
public static long extractTimestamp(long tso) {
// extract physical time from TSO timestamp by remove Lower 18 bits
logical counter bits
return (tso >> PHYSICAL_SHIFT);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
index be631b212d3..cf2f92487bd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
@@ -144,11 +144,20 @@ public class PhysicalPlanTranslatorTest extends
TestWithFeService {
+ "'enable_unique_key_merge_on_write' = 'false',"
+ "'sequence_mapping.s1' = 'v1',"
+ "'sequence_mapping.s2' = 'v2');");
+ // Bump the base table with a real (positive) tso before creating the
stream so that the
+ // stream records a positive consumption offset at creation time;
hasConsumedData() only
+ // treats a positive offset as a real baseline, otherwise every
partition would be pruned out
+ // of the snapshot scan.
+ Database database = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_db");
+ OlapTable binlogScanSchemaTable =
+ (OlapTable)
database.getTableOrMetaException("binlog_scan_schema_t");
+ bumpPartitionsAndReplicas(binlogScanSchemaTable, 2L, 100L);
createTable("create stream test_db.binlog_scan_schema_stream "
+ "on table test_db.binlog_scan_schema_t properties('type' =
'append_only')");
- Database database = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_db");
- bumpPartitionsAndReplicas(
- (OlapTable)
database.getTableOrMetaException("binlog_scan_schema_t"), 2L);
+ // Advance the base table tso again after the stream is created so the
partition has new data
+ // beyond the recorded consumption offset, driving the snapshot scan
down the rebuild path
+ // (base scan wrapped in OlapTableWrapper unioned with the binlog
before-image).
+ bumpPartitionsAndReplicas(binlogScanSchemaTable, 3L, 200L);
createTable("create table test_db.t_topn_lazy(c1 int, c2 int, c3 int) "
+ "duplicate key(c1) distributed by hash(c1) buckets 1 "
+ "properties('replication_num' = '1', 'light_schema_change' =
'true');");
@@ -544,10 +553,10 @@ public class PhysicalPlanTranslatorTest extends
TestWithFeService {
return scanNodes;
}
- private static void bumpPartitionsAndReplicas(OlapTable table, long
newVersion) {
+ private static void bumpPartitionsAndReplicas(OlapTable table, long
newVersion, long tso) {
for (Partition partition : table.getPartitions()) {
long timestamp = System.currentTimeMillis();
- partition.setVisibleVersionAndTime(newVersion, timestamp,
timestamp);
+ partition.setVisibleVersionAndTime(newVersion, timestamp, tso);
partition.setNextVersion(newVersion + 1);
for (MaterializedIndex index :
partition.getMaterializedIndices(IndexExtState.VISIBLE, true)) {
for (Tablet tablet : index.getTablets()) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
index fc4f3f77f0f..e139ab4ab5e 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
@@ -25,7 +25,6 @@ import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.MaterializedIndex;
import org.apache.doris.catalog.MaterializedIndex.IndexExtState;
import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.OlapTableWrapper;
import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.Replica;
import org.apache.doris.catalog.RowBinlogTableWrapper;
@@ -36,6 +35,7 @@ import org.apache.doris.catalog.stream.OlapTableStreamUpdate;
import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.Pair;
+import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.nereids.NereidsPlanner;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.glue.translator.PhysicalPlanTranslator;
@@ -43,12 +43,17 @@ import
org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.LessThan;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
import org.apache.doris.nereids.trees.plans.commands.ExplainCommand;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableStreamScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
@@ -71,6 +76,7 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.Set;
/**
* UTs for table stream query plan, including
@@ -359,11 +365,12 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
TPaloScanRange range = loc.getScanRange().getPaloScanRange();
long tabletId = range.getTabletId();
long pid = tabletIdToPartitionId.get(tabletId);
- long expectedStart = stream.getStreamUpdate(pid).first;
+ // BE reads [startTso, endTso), so the recorded offset is
shifted to its next TSO.
+ long expectedStart =
TSOTimestamp.nextTso(stream.getStreamUpdate(pid).first);
Assertions.assertEquals(expectedScanType,
range.getBinlogScanType(),
"binlog scan type should match stream consume type");
Assertions.assertEquals(expectedStart, range.getStartTso(),
- "startTSO should equal stream partitionOffset (last
committed binlog TSO)");
+ "startTSO should equal stream partitionOffset (last
committed binlog TSO) + 1");
assertedAtLeastOne = true;
}
}
@@ -396,11 +403,14 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
}
}
Assertions.assertNotNull(incrementalScan1);
- OlapTableWrapper wrapper = (OlapTableWrapper)
incrementalScan1.getOlapTable();
Map<Long, Long> prevOffsets = new java.util.HashMap<>();
Map<Long, Long> nextOffsets = new java.util.HashMap<>();
for (Long pid : incrementalScan1.getSelectedPartitionIds()) {
- Pair<Long, Long> off = wrapper.getPartitionOffset(pid);
+ // Use the raw (un-shifted) stream offsets, mirroring what the
production
+ // StreamConsumptionInfoExtractor commits. Reading them back from
the scan node's
+ // RowBinlogTableWrapper would return the already +1-shifted
scan-range bounds and
+ // introduce a spurious double shift into this closed-loop check.
+ Pair<Long, Long> off = stream.getStreamUpdate(pid);
if (off.first != null) {
prevOffsets.put(pid, off.first);
}
@@ -440,8 +450,9 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
for (TScanRangeLocations loc : locations) {
TPaloScanRange range = loc.getScanRange().getPaloScanRange();
long pid = tabletIdToPartitionId.get(range.getTabletId());
- Assertions.assertEquals(nextOffsets.get(pid),
range.getStartTso(),
- "after offset commit, new startTSO must equal the
previously committed next TSO");
+ // BE reads [startTso, endTso), so the stream wrapper shifts
the recorded offset by +1.
+
Assertions.assertEquals(TSOTimestamp.nextTso(nextOffsets.get(pid)),
range.getStartTso(),
+ "after offset commit, new startTSO must equal the
previously committed next TSO + 1");
assertedAtLeastOne = true;
}
}
@@ -586,6 +597,7 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
// asserting every incremental scan range carries the composed
start/end TSO for its partition.
String startTs = "2026-05-25 20:51:28";
String endTs = "2026-05-25 21:51:28";
+ // @incr is left-closed right-open [start, end): BE uses GE/LT on the
composed bounds directly.
long expectedStartTso =
TSOTimestamp.composePhysicalTimestamp(OlapScanNode.parseChangeTimestamp(startTs));
long expectedEndTso =
TSOTimestamp.composePhysicalTimestamp(OlapScanNode.parseChangeTimestamp(endTs));
@@ -652,6 +664,52 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
}
}
+ @Test
+ public void testDupTimeTravelIncludesExactTimestamp() {
+ assertTimeTravelBoundary("tbl_dup_stream_base", "time as of '0'", 1L,
false);
+ String timestamp = TimeUtils.longToTimeString(1700000000000L);
+ // (1700000000000 ms, logical 0) is inclusive; logical 1 is the
exclusive upper bound.
+ assertTimeTravelBoundary("tbl_dup_stream_base", "time as of '" +
timestamp + "'",
+ 445644800000000001L, false);
+ }
+
+ @Test
+ public void testMowTimeTravelIncludesExactTimestamp() {
+ assertTimeTravelBoundary("tbl_stream_base", "time as of '0'", 1L,
true);
+ String timestamp = TimeUtils.longToTimeString(1700000000000L);
+ assertTimeTravelBoundary("tbl_stream_base", "time as of '" + timestamp
+ "'",
+ 445644800000000001L, true);
+ }
+
+ private void assertTimeTravelBoundary(String table, String snapshot, long
exclusiveBound, boolean mow) {
+ Plan plan = PlanChecker.from(connectContext)
+ .analyze("select * from test_stream." + table + " for " +
snapshot)
+ .getCascadesContext().getRewritePlan();
+ Set<LogicalFilter<?>> filters = plan.collect(node -> node instanceof
LogicalFilter);
+ List<Expression> commitPredicates = new ArrayList<>();
+ for (LogicalFilter<?> filter : filters) {
+ for (Expression conjunct : filter.getConjuncts()) {
+ if (conjunct instanceof LessThan && conjunct.child(0)
instanceof SlotReference
+ && ((SlotReference)
conjunct.child(0)).getName().equals(Column.COMMIT_TSO_COL)) {
+ commitPredicates.add(conjunct);
+ }
+ }
+ }
+ Assertions.assertEquals(1, commitPredicates.size());
+ Assertions.assertEquals(new BigIntLiteral(exclusiveBound),
commitPredicates.get(0).child(1));
+
+ Set<LogicalOlapScan> binlogScans = plan.collect(node -> node
instanceof LogicalOlapScan
+ && ((LogicalOlapScan) node).getTable() instanceof
RowBinlogTableWrapper);
+ Assertions.assertEquals(mow ? 1 : 0, binlogScans.size());
+ for (LogicalOlapScan scan : binlogScans) {
+ RowBinlogTableWrapper wrapper = (RowBinlogTableWrapper)
scan.getTable();
+ Assertions.assertFalse(wrapper.getPartitionIds().isEmpty());
+ for (Long partitionId : wrapper.getPartitionIds()) {
+ Assertions.assertEquals(Pair.of(exclusiveBound, null),
wrapper.getPartitionOffset(partitionId));
+ }
+ }
+ }
+
@Test
public void testMowTimeTravelQualifiedColumnCanBind() {
// MOW time-travel goes through a union whose outputs are rebuilt with
empty qualifiers.
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeTest.java
index 93a8ed6cbe3..0c3b524cd12 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/planner/OlapScanNodeTest.java
@@ -46,7 +46,9 @@ import org.apache.doris.cloud.catalog.CloudPartition;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.util.DebugPointUtil;
+import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.nereids.exceptions.ParseException;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TOlapScanNode;
import org.apache.doris.thrift.TPaloScanRange;
@@ -64,6 +66,8 @@ import org.junit.jupiter.api.Test;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -73,6 +77,28 @@ import java.util.Set;
import java.util.stream.Collectors;
public class OlapScanNodeTest {
+ @Test
+ public void testParseChangeTimestampRange() {
+ DateTimeFormatter format = TimeUtils.getDatetimeFormatWithTimeZone();
+ Assertions.assertEquals(0L, OlapScanNode.parseChangeTimestamp("0"));
+ Assertions.assertEquals(1700000000000L,
+
OlapScanNode.parseChangeTimestamp(format.format(Instant.ofEpochMilli(1700000000000L))));
+ // The parser accepts whole seconds; these straddle the physical limit
35184372088831 ms.
+ Assertions.assertEquals(35184372088000L,
+
OlapScanNode.parseChangeTimestamp(format.format(Instant.ofEpochMilli(35184372088000L))));
+ ParseException error = Assertions.assertThrows(ParseException.class,
+ () ->
OlapScanNode.parseChangeTimestamp(format.format(Instant.ofEpochMilli(35184372089000L))));
+ Assertions.assertTrue(error.getMessage().contains("Timestamp exceeds
supported TSO range"));
+ Assertions.assertThrows(ParseException.class,
+ () -> OlapScanNode.parseChangeTimestamp("4000-01-01
00:00:00"));
+ Assertions.assertThrows(ParseException.class,
+ () -> OlapScanNode.parseChangeTimestamp("9999-01-01
00:00:00"));
+ Assertions.assertThrows(ParseException.class,
+ () ->
OlapScanNode.parseChangeTimestamp(format.format(Instant.ofEpochMilli(-1000L))));
+ Assertions.assertThrows(ParseException.class, () ->
OlapScanNode.parseChangeTimestamp("invalid"));
+ Assertions.assertThrows(ParseException.class, () ->
OlapScanNode.parseChangeTimestamp(null));
+ }
+
private MaterializedIndex createMaterializedIndex(List<Long> tabletIds) {
MaterializedIndex index = new MaterializedIndex();
List<Tablet> tablets =
Lists.newArrayListWithExpectedSize(tabletIds.size());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/tso/TSOTimestampTest.java
b/fe/fe-core/src/test/java/org/apache/doris/tso/TSOTimestampTest.java
index 7f6ebb52e3a..c65dae049d6 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/tso/TSOTimestampTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/tso/TSOTimestampTest.java
@@ -68,6 +68,21 @@ public class TSOTimestampTest {
org.junit.jupiter.api.Assertions.assertEquals(0L,
TSOTimestamp.extractLogicalCounter(composed));
}
+ @Test
+ public void testComposePhysicalTimestampRange() {
+ // Signed TSOs reserve 18 low bits for the logical counter.
+ long maxPhysicalTime = (1L << 45) - 1;
+ Assertions.assertEquals(0L, TSOTimestamp.composePhysicalTimestamp(0L));
+ Assertions.assertEquals(Long.MAX_VALUE -
TSOTimestamp.MAX_LOGICAL_COUNTER,
+ TSOTimestamp.composePhysicalTimestamp(maxPhysicalTime));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> TSOTimestamp.composePhysicalTimestamp(-1L));
+ Assertions.assertThrows(ArithmeticException.class,
+ () -> TSOTimestamp.composePhysicalTimestamp(maxPhysicalTime +
1));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> TSOTimestamp.composePhysicalTimestamp(1L << 46));
+ }
+
@Test
public void testBitWidthLimitations() {
// Test that values are properly masked to fit in their respective bit
widths
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
