diff --git a/be/src/exec/scan/olap_scanner.cpp b/be/src/exec/scan/olap_scanner.cpp index 4fafd8ffb30b20..d0d3a03bee9f99 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( + _tablet_reader_params.predicates.push_back(create_comparison_predicate( tso_ordinal, tso_column->name(), tso_data_type, Field::create_field(*_start_tso), false)); } if (_end_tso.has_value()) { - _tablet_reader_params.predicates.push_back(create_comparison_predicate( + _tablet_reader_params.predicates.push_back(create_comparison_predicate( tso_ordinal, tso_column->name(), tso_data_type, Field::create_field(*_end_tso), false)); } diff --git a/cloud/src/meta-service/meta_service_partition.cpp b/cloud/src/meta-service/meta_service_partition.cpp index c730037d3c7667..7ca12c1c5742a7 100644 --- a/cloud/src/meta-service/meta_service_partition.cpp +++ b/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); diff --git a/cloud/src/meta-service/meta_service_table_stream.cpp b/cloud/src/meta-service/meta_service_table_stream.cpp index 77823a2162c1f4..ece040731cd38a 100644 --- a/cloud/src/meta-service/meta_service_table_stream.cpp +++ b/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 {}; diff --git a/cloud/src/meta-service/meta_service_txn.cpp b/cloud/src/meta-service/meta_service_txn.cpp index 76963cb0627e8f..ad38d02dab8485 100644 --- a/cloud/src/meta-service/meta_service_txn.cpp +++ b/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; 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 20f6bf915e8cf2..0e6fabc153a003 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 @@ -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 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 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 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 // otherwise, return 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 diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java index 052cf16dcad995..a8d49c1feaf6e1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/stream/TableStreamManager.java @@ -516,7 +516,10 @@ private void fillRows(Map 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( diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java index 283a53b455ba05..a78d68b392d6ed 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/datasource/CloudInternalCatalog.java @@ -210,10 +210,14 @@ protected List 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; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/StreamConsumptionInfoExtractor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/StreamConsumptionInfoExtractor.java index bb9d7d276109d2..08e9f59aed96fc 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/StreamConsumptionInfoExtractor.java +++ b/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); 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 3418137d57c734..20e84418c9f883 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 @@ -581,13 +581,15 @@ private void addScanRangeLocations(Partition partition, parseBinlogScanType(scanParams, ((OlapTableWrapper) olapTable).getOriginTable()); Pair 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; diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/CloudTableStreamConsumptionTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/CloudTableStreamConsumptionTest.java index 60110061711f84..e726745336d325 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/stream/CloudTableStreamConsumptionTest.java +++ b/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()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java b/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java index b25ff6e7a6b178..2351b29667a3d0 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/cloud/datasource/CloudInternalCatalogTableStreamTest.java +++ b/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()); } } 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 7e9288721e433d..3fc305f8d7d849 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 @@ -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));