[fix](table stream) fix table stream TSO boundary semantics

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));
     }

cloud/src/meta-service/meta_service_partition.cpp

@@ -1214,7 +1214,9 @@ void MetaServiceImpl::commit_table_stream_partition_internal(
         auto source_version_it = stream_versions.find(partition_id);
         DCHECK(source_version_it != stream_versions.end());
         const VersionPB& source_version = source_version_it->second;
-        if (offset.offset_tso() > source_version.commit_tso()) {
+        // Water marks use next-point semantics (real commit_tso + 1); the initial offset may
+        // legally reach source commit_tso + 1. The empty-partition sentinel (-1) always passes.
+        if (offset.offset_tso() > source_version.commit_tso() + 1) {
             code = MetaServiceCode::INVALID_ARGUMENT;
             msg = fmt::format("initial offset exceeds source commit TSO for partition {}",
                               partition_id);

cloud/src/meta-service/meta_service_table_stream.cpp

@@ -96,7 +96,10 @@ TableStreamReadResult fill_partition_read_state(const TableStreamIdentityPB& ide
                                                 TableStreamPartitionReadStatePB* state) {
     state->set_partition_id(partition_id);
     state->set_visible_version(version.version());
-    state->set_end_tso(version.commit_tso());
+    // TSO water marks are stored as the next tso to read (real commit_tso + 1) so that the
+    // scan range is a unified half-open interval [offset_tso, end_tso). Expose the source's
+    // upper bound in the same next-point semantics.
+    state->set_end_tso(version.commit_tso() + 1);
     if (!offset) {
         state->set_offset_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_UNKNOWN);
         return {};

cloud/src/meta-service/meta_service_txn.cpp

@@ -262,7 +262,9 @@ class TableStreamUpdateTxnContext {
                                update.partition_id());
             return false;
         }
-        if (update.next_offset_tso() <= source_version.commit_tso()) {
+        // Water marks use next-point semantics (real commit_tso + 1), so the next offset may
+        // legally reach source commit_tso + 1 (meaning all committed rows have been consumed).
+        if (update.next_offset_tso() <= source_version.commit_tso() + 1) {
             return true;
         }
         code_ = MetaServiceCode::INVALID_ARGUMENT;

fe/fe-core/src/main/java/org/apache/doris/catalog/stream/OlapTableStream.java

@@ -133,17 +133,20 @@ private void initializeLocalOffsets() {
         if (baseTable == null) {
             return;
         }
+        // TSO water marks use next-point semantics: the stored offset is the next tso to read
+        // (real commit_tso + 1), so the scan range stays a unified half-open interval
+        // [offset, end). set offset according to baseTable
         if (!showInitialRows) {
             // set partition offset
             baseTable.getPartitions()
-                    .forEach(p -> partitionOffset.put(p.getId(), p.getTso()));
+                    .forEach(p -> partitionOffset.put(p.getId(), p.getTso() + 1));
         } else {
             baseTable.getPartitions()
                     .stream()
                     .filter(p -> p.getVisibleVersion() > Partition.PARTITION_INIT_VERSION)
                     .forEach(p -> {
-                                historicalPartitionTSO.put(p.getId(), p.getTso());
-                                }
+                                historicalPartitionTSO.put(p.getId(), p.getTso() + 1);
+                                    }
                     );
         }
     }
@@ -182,14 +185,16 @@ void fillTableStreamConsumptionInfo(List<TRow> dataBatch) {
                     // UNIT
                     trow.addToColumnValue(new TCell().setStringVal(entry.getValue().getName()));
                     if (partitionOffset.containsKey(entry.getKey())) {
+                        // partitionOffset stores a next-point (real last-consumed tso + 1);
+                        // subtract 1 to expose the last-consumed tso to users unchanged.
+                        long lastConsumedTso = partitionOffset.get(entry.getKey()) - 1;
                         // CONSUMPTION_STATUS
                         trow.addToColumnValue(new TCell()
-                                .setStringVal(String.valueOf(partitionOffset.get(entry.getKey()))));
+                                .setStringVal(String.valueOf(lastConsumedTso)));
                         // LAG
                         trow.addToColumnValue(new TCell()
                                 .setStringVal(String.valueOf(
-                                        entry.getValue().getTso()
-                                                - partitionOffset.get(entry.getKey()))));
+                                        entry.getValue().getTso() - lastConsumedTso)));
                         // LAST_CONSUMPTION_TIME
                         if (partitionConsumptionTime.containsKey(entry.getKey())) {
                             trow.addToColumnValue(new TCell()
@@ -219,9 +224,11 @@ void fillTableStreamConsumptionInfo(List<TRow> dataBatch) {
     }
 
     public boolean hasData(Partition partition) {
+        // partitionOffset stores a next-point (real last-consumed tso + 1); all visible data has
+        // been consumed once it reaches the partition's latest commit tso + 1.
         // if all available visible data has been consumed, return false
         return  (!partitionOffset.containsKey(partition.getId())
-                || !partitionOffset.get(partition.getId()).equals(partition.getTso()))
+                || !partitionOffset.get(partition.getId()).equals(partition.getTso() + 1))
                 && partition.hasData();
     }
 
@@ -234,13 +241,16 @@ public boolean hasConsumedData(long partitionId) {
     }
 
     public Pair<Long, Long> getStreamUpdate(Long partitionId) {
+        // Both bounds use next-point semantics (real commit_tso + 1). The left bound (start) is
+        // already stored as a next-point in partitionOffset/historicalPartitionTSO, so only the
+        // right bound (end), derived from the partition's latest commit tso, needs +1.
         // if partition has historical data, return <historical tso, current tso>
         // otherwise, return <current consumed tso, current tso>
         Long left = partitionOffset.get(partitionId);
         if (historicalPartitionTSO.containsKey(partitionId)) {
             left = historicalPartitionTSO.get(partitionId);
         }
-        return Pair.of(left, getBaseTableNullable().getPartition(partitionId).getTso());
+        return Pair.of(left, getBaseTableNullable().getPartition(partitionId).getTso() + 1);
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java

@@ -516,7 +516,10 @@ private void fillRows(Map<Long, Cloud.TableStreamPartitionReadStatePB> partition
                     if (!state.hasOffsetTso()) {
                         throw new UserException("MetaService returned a Cloud Table Stream state without Offset TSO");
                     }
-                    row.addToColumnValue(new TCell().setStringVal(String.valueOf(state.getOffsetTso())));
+                    // offset_tso stores a next-point (real last-consumed tso + 1); subtract 1 to
+                    // expose the last-consumed tso. LAG uses end - offset where both are next-point,
+                    // so the difference already equals the real lag.
+                    row.addToColumnValue(new TCell().setStringVal(String.valueOf(state.getOffsetTso() - 1)));
                     row.addToColumnValue(new TCell().setStringVal(
                             String.valueOf(state.getEndTso() - state.getOffsetTso())));
                     row.addToColumnValue(new TCell().setLongVal(

fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java

@@ -210,10 +210,14 @@ protected List<Cloud.TableStreamOffsetPB> captureTableStreamInitialOffsets(
             Cloud.TableStreamOffsetStatePB state = stream.isShowInitialRows() && !emptyPartition
                     ? Cloud.TableStreamOffsetStatePB.TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING
                     : Cloud.TableStreamOffsetStatePB.TABLE_STREAM_OFFSET_CONSUMED;
+            // Water marks use next-point semantics: store the next tso to read (real commit_tso
+            // + 1) so the scan range is a unified half-open interval [offset, end). The empty
+            // partition keeps its -1 sentinel.
+            long offsetTso = emptyPartition ? commitTso : commitTso + 1;
             offsets.add(Cloud.TableStreamOffsetPB.newBuilder()
                     .setPartitionId(partitionId)
                     .setState(state)
-                    .setOffsetTso(commitTso)
+                    .setOffsetTso(offsetTso)
                     .build());
         }
         return offsets;

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/StreamConsumptionInfoExtractor.java

@@ -120,7 +120,8 @@ private static OlapTableStreamUpdate toOlapTableStreamUpdate(OlapTableStreamWrap
             if (update.second != null) {
                 next.put(entry.getKey(), update.second);
             } else {
-                next.put(entry.getKey(), wrapper.getPartition(entry.getKey()).getTso());
+                // Water marks use next-point semantics (real commit_tso + 1); align the fallback.
+                next.put(entry.getKey(), wrapper.getPartition(entry.getKey()).getTso() + 1);
             }
         }
         return new OlapTableStreamUpdate(prev, next);

fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java

@@ -581,13 +581,15 @@ private void addScanRangeLocations(Partition partition,
                         parseBinlogScanType(scanParams, ((OlapTableWrapper) olapTable).getOriginTable());
                 Pair<Long, Long> update = getPartitionOffset(partition.getId());
                 if (update != null) {
+                    // TSO bounds already use next-point semantics (real commit_tso + 1), forming a
+                    // half-open scan range [startTso, endTso) at the BE. Pass them through as-is.
                     if (update.first != null) {
                         paloRange.setStartTso(update.first);
                     }
                     if (update.second != null) {
                         paloRange.setEndTso(update.second);
                     } else {
-                        paloRange.setEndTso(partition.getTso());
+                        paloRange.setEndTso(partition.getTso() + 1);
                     }
                 }
                 if (binlogScanType != TBinlogScanType.NONE) {
@@ -1964,14 +1966,6 @@ public TableScanParams getScanParams() {
         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;

fe/fe-core/src/test/java/org/apache/doris/catalog/stream/CloudTableStreamConsumptionTest.java

@@ -121,7 +121,9 @@ public void testConsumptionViewReadsAuthoritativeCloudOffsets() throws Exception
             rows.sort(Comparator.comparing(row -> row.getColumnValue().get(3).getStringVal()));
             Assertions.assertEquals(2, rows.size());
             Assertions.assertEquals("p1", rows.get(0).getColumnValue().get(3).getStringVal());
-            Assertions.assertEquals("100", rows.get(0).getColumnValue().get(4).getStringVal());
+            // offset_tso 100 is a next-point; the displayed last-consumed tso is 100 - 1 = 99,
+            // while LAG stays end(130) - offset(100) = 30.
+            Assertions.assertEquals("99", rows.get(0).getColumnValue().get(4).getStringVal());
             Assertions.assertEquals("30", rows.get(0).getColumnValue().get(5).getStringVal());
             Assertions.assertEquals(999, rows.get(0).getColumnValue().get(6).getLongVal());
             Assertions.assertEquals("p2", rows.get(1).getColumnValue().get(3).getStringVal());

fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java

@@ -390,7 +390,8 @@ public void testCaptureInitialOffsetsUsesVersionCommitTsoSnapshot() throws Excep
             Assertions.assertEquals(-1, offsets.get(0).getOffsetTso());
             Assertions.assertEquals(Cloud.TableStreamOffsetStatePB.TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING,
                     offsets.get(1).getState());
-            Assertions.assertEquals(101, offsets.get(1).getOffsetTso());
+            // Water marks use next-point semantics: source commit_tso 101 -> next tso to read 102.
+            Assertions.assertEquals(102, offsets.get(1).getOffsetTso());
         }
     }
 

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java

@@ -359,11 +359,13 @@ public void testIncrementalScanRangeUsesPartitionOffsetAsStartTSO() throws Excep
                 TPaloScanRange range = loc.getScanRange().getPaloScanRange();
                 long tabletId = range.getTabletId();
                 long pid = tabletIdToPartitionId.get(tabletId);
+                // Water marks use next-point semantics: getStreamUpdate already returns the next
+                // tso to read, and OlapScanNode passes it through as the half-open range start.
                 long expectedStart = 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 (next tso to read)");
                 assertedAtLeastOne = true;
             }
         }
@@ -404,7 +406,7 @@ public void testIncrementalScanStartTsoAdvancesAfterOffsetCommit() throws Except
             if (off.first != null) {
                 prevOffsets.put(pid, off.first);
             }
-            nextOffsets.put(pid, off.second != null ? off.second : baseTable.getPartition(pid).getTso());
+            nextOffsets.put(pid, off.second != null ? off.second : baseTable.getPartition(pid).getTso() + 1);
         }
         OlapTableStreamUpdate update = new OlapTableStreamUpdate(prevOffsets, nextOffsets);
         Assertions.assertFalse(nextOffsets.isEmpty());
@@ -440,6 +442,8 @@ public void testIncrementalScanStartTsoAdvancesAfterOffsetCommit() throws Except
             for (TScanRangeLocations loc : locations) {
                 TPaloScanRange range = loc.getScanRange().getPaloScanRange();
                 long pid = tabletIdToPartitionId.get(range.getTabletId());
+                // Water marks use next-point semantics: the committed next TSO is passed through
+                // as the half-open range start without any further shift.
                 Assertions.assertEquals(nextOffsets.get(pid), range.getStartTso(),
                         "after offset commit, new startTSO must equal the previously committed next TSO");
                 assertedAtLeastOne = true;
@@ -586,6 +590,8 @@ public void testIncrTimestampRangePropagatesToScanRangeTso() throws Exception {
         // 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 bounds come straight from the user-provided timestamps and are passed through as a
+        // half-open [start, end) scan range without any shift.
         long expectedStartTso = TSOTimestamp.composeFullTimestamp(OlapScanNode.parseChangeTimestamp(startTs));
         long expectedEndTso = TSOTimestamp.composeFullTimestamp(OlapScanNode.parseChangeTimestamp(endTs));