[fix](stream) Estimate stream rows from consumption baselines

cloud/src/meta-service/meta_service_table_stream.cpp

@@ -112,6 +112,9 @@ TableStreamReadResult fill_partition_read_state(const TableStreamIdentityPB& ide
 
     state->set_offset_state(offset->state());
     state->set_offset_tso(offset->offset_tso());
+    state->set_base_row_count(offset->has_base_row_count() ? offset->base_row_count() : -1);
+    state->set_changelog_row_count(offset->has_changelog_row_count() ? offset->changelog_row_count()
+                                                                     : -1);
     if (offset->has_last_consumption_time_ms()) {
         state->set_last_consumption_time_ms(offset->last_consumption_time_ms());
     }

cloud/src/meta-service/meta_service_txn.cpp

@@ -235,7 +235,10 @@ class TableStreamUpdateTxnContext {
             }
         }
         for (const auto& update : stream_update.partition_updates()) {
-            if (!write_next_offset(identity, update, offset_gc)) {
+            auto previous = stream_offsets.find(update.partition_id());
+            if (!write_next_offset(identity, update,
+                                   previous == stream_offsets.end() ? nullptr : &previous->second,
+                                   offset_gc)) {
                 return false;
             }
         }
@@ -315,12 +318,26 @@ class TableStreamUpdateTxnContext {
 
     bool write_next_offset(const TableStreamIdentityPB& identity,
                            const TableStreamPartitionUpdatePB& update,
+                           const TableStreamOffsetPB* previous,
                            TableStreamPartitionSetPB* offset_gc) {
         TableStreamOffsetPB next_offset;
         next_offset.set_partition_id(update.partition_id());
         next_offset.set_state(TableStreamOffsetStatePB::TABLE_STREAM_OFFSET_CONSUMED);
         next_offset.set_offset_tso(update.next_offset_tso());
         next_offset.set_last_consumption_time_ms(consumption_time_ms_);
+        if (previous != nullptr && previous->state() == TABLE_STREAM_OFFSET_CONSUMED &&
+            previous->offset_tso() == update.next_offset_tso()) {
+            // Reports arriving later cannot establish a new baseline for an unchanged offset.
+            next_offset.set_base_row_count(
+                    previous->has_base_row_count() ? previous->base_row_count() : -1);
+            next_offset.set_changelog_row_count(
+                    previous->has_changelog_row_count() ? previous->changelog_row_count() : -1);
+        } else {
+            next_offset.set_base_row_count(update.has_base_row_count() ? update.base_row_count()
+                                                                       : -1);
+            next_offset.set_changelog_row_count(
+                    update.has_changelog_row_count() ? update.changelog_row_count() : -1);
+        }
         std::string value;
         if (!next_offset.SerializeToString(&value)) {
             code_ = MetaServiceCode::PROTOBUF_SERIALIZE_ERR;

cloud/test/meta_service_table_stream_test.cpp

@@ -1041,17 +1041,37 @@ TEST_F(MetaServiceTableStreamTest, CommitUpdatesLatestOffsetAndIsIdempotent) {
     ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
 
     int64_t txn_id = begin_target_transaction("consume-existing-offset");
-    CommitTxnResponse response =
-            consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
+    CommitTxnRequest request =
+            make_consume_request(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
+    auto* update = request.mutable_table_stream_updates(0)->mutable_partition_updates(0);
+    update->set_base_row_count(50);
+    update->set_changelog_row_count(100);
+    CommitTxnResponse response = commit_transaction(request);
     ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
     TableStreamOffsetPB offset = get_latest_offset(2001);
     EXPECT_EQ(offset.state(), TABLE_STREAM_OFFSET_CONSUMED);
     EXPECT_EQ(offset.offset_tso(), 120);
     EXPECT_GT(offset.last_consumption_time_ms(), 0);
-
-    response = consume_partition(txn_id, 2001, TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
+    EXPECT_EQ(offset.base_row_count(), 50);
+    EXPECT_EQ(offset.changelog_row_count(), 100);
+    auto read = get_read_state({2001});
+    ASSERT_EQ(read.status().code(), MetaServiceCode::OK);
+    EXPECT_EQ(read.bindings(0).partition_states(0).base_row_count(), 50);
+    EXPECT_EQ(read.bindings(0).partition_states(0).changelog_row_count(), 100);
+
+    update->set_base_row_count(999);
+    update->set_changelog_row_count(999);
+    response = commit_transaction(request);
     ASSERT_EQ(response.status().code(), MetaServiceCode::OK) << response.status().msg();
     EXPECT_EQ(get_latest_offset(2001).offset_tso(), 120);
+    EXPECT_EQ(get_latest_offset(2001).base_row_count(), 50);
+    EXPECT_EQ(get_latest_offset(2001).changelog_row_count(), 100);
+    request = make_consume_request(begin_target_transaction("consume-unchanged-offset"), 2001,
+                                   TABLE_STREAM_OFFSET_CONSUMED, 120, 120);
+    request.mutable_table_stream_updates(0)->mutable_partition_updates(0)->set_base_row_count(999);
+    ASSERT_EQ(commit_transaction(request).status().code(), MetaServiceCode::OK);
+    EXPECT_EQ(get_latest_offset(2001).base_row_count(), 50);
+    EXPECT_EQ(get_latest_offset(2001).changelog_row_count(), 100);
 
     txn_id = begin_target_transaction("consume-unknown-offset");
     response = consume_partition(txn_id, 2002, TABLE_STREAM_OFFSET_UNKNOWN, 0, 140);
@@ -1326,6 +1346,9 @@ TEST_F(MetaServiceTableStreamTest, CommitRetryAfterMetaServiceRestartIsExactlyOn
     CommitTxnRequest request = make_consume_request(txn_id, kSourcePartitionId,
                                                     TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
     request.set_commit_tso(999);
+    request.mutable_table_stream_updates(0)->mutable_partition_updates(0)->set_base_row_count(50);
+    request.mutable_table_stream_updates(0)->mutable_partition_updates(0)->set_changelog_row_count(
+            100);
 
     // The commit succeeds, but its response is lost before the caller observes it.
     const CommitTxnResponse lost_response = commit_transaction(request);
@@ -1338,6 +1361,8 @@ TEST_F(MetaServiceTableStreamTest, CommitRetryAfterMetaServiceRestartIsExactlyOn
     ASSERT_EQ(retry_response.status().code(), MetaServiceCode::OK) << retry_response.status().msg();
 
     EXPECT_EQ(get_latest_offset(kSourcePartitionId).offset_tso(), 120);
+    EXPECT_EQ(get_latest_offset(kSourcePartitionId).base_row_count(), 50);
+    EXPECT_EQ(get_latest_offset(kSourcePartitionId).changelog_row_count(), 100);
     ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
     std::string value;
     EXPECT_EQ(txn->get(meta_rowset_tmp_key({instance_id_, txn_id, kTargetTabletId}), &value),
@@ -1515,13 +1540,24 @@ TEST_F(MetaServiceTableStreamTest, ConcurrentOffsetCasAllowsOnlyOneCommit) {
     put_latest_partition_state(txn.get(), kPartitionId, 8, 130, 100);
     ASSERT_EQ(txn->commit(), TxnErrorCode::TXN_OK);
 
-    const CommitTxnRequest first_request =
+    CommitTxnRequest first_request =
             make_consume_request(begin_target_transaction("concurrent-cas-first"), kPartitionId,
                                  TABLE_STREAM_OFFSET_CONSUMED, 100, 120);
-    const CommitTxnRequest second_request =
+    CommitTxnRequest second_request =
             make_consume_request(begin_target_transaction("concurrent-cas-second"), kPartitionId,
                                  TABLE_STREAM_OFFSET_CONSUMED, 100, 125);
 
+    first_request.mutable_table_stream_updates(0)->mutable_partition_updates(0)->set_base_row_count(
+            50);
+    first_request.mutable_table_stream_updates(0)
+            ->mutable_partition_updates(0)
+            ->set_changelog_row_count(100);
+    second_request.mutable_table_stream_updates(0)
+            ->mutable_partition_updates(0)
+            ->set_base_row_count(60);
+    second_request.mutable_table_stream_updates(0)
+            ->mutable_partition_updates(0)
+            ->set_changelog_row_count(120);
     auto [first_response, second_response] = commit_concurrently(first_request, second_request);
 
     const int successful_commits = (first_response.status().code() == MetaServiceCode::OK ? 1 : 0) +
@@ -1533,6 +1569,9 @@ TEST_F(MetaServiceTableStreamTest, ConcurrentOffsetCasAllowsOnlyOneCommit) {
     EXPECT_EQ(conflicted_commits, 1);
     const int64_t final_offset = get_latest_offset(kPartitionId).offset_tso();
     EXPECT_TRUE(final_offset == 120 || final_offset == 125);
+    EXPECT_EQ(get_latest_offset(kPartitionId).base_row_count(), final_offset == 120 ? 50 : 60);
+    EXPECT_EQ(get_latest_offset(kPartitionId).changelog_row_count(),
+              final_offset == 120 ? 100 : 120);
 }
 
 TEST_F(MetaServiceTableStreamTest, ConcurrentUnknownOffsetCasAllowsOnlyOneFirstConsumer) {
@@ -2119,13 +2158,27 @@ TEST_F(MetaServiceTableStreamTest, CommitWritesOffsetsForEachMultiVersionMode) {
                 mode_case.mode == MULTI_VERSION_READ_WRITE
                         ? TABLE_STREAM_OFFSET_INITIAL_SNAPSHOT_PENDING
                         : TABLE_STREAM_OFFSET_CONSUMED;
-        const CommitTxnResponse response = consume_partition(
+        // The first full consumption can keep its numeric TSO while changing offset state.
+        const int64_t next_tso = mode_case.mode == MULTI_VERSION_READ_WRITE ? 100 : 120;
+        CommitTxnRequest request = make_consume_request(
                 begin_target_transaction(fmt::format("consume-mode-{}", mode_case.label)),
-                partition_id, expected_state, 100, 120);
+                partition_id, expected_state, 100, next_tso);
+        auto* update = request.mutable_table_stream_updates(0)->mutable_partition_updates(0);
+        update->set_base_row_count(50);
+        update->set_changelog_row_count(100);
+        const CommitTxnResponse response = commit_transaction(request);
         EXPECT_EQ(response.status().code(),
                   mode_case.succeeds ? MetaServiceCode::OK : MetaServiceCode::INVALID_ARGUMENT)
                 << response.status().msg();
-        EXPECT_EQ(get_latest_offset(partition_id).offset_tso(), mode_case.succeeds ? 120 : 100);
+        EXPECT_EQ(get_latest_offset(partition_id).offset_tso(),
+                  mode_case.succeeds ? next_tso : 100);
+        if (mode_case.succeeds) {
+            EXPECT_EQ(get_latest_offset(partition_id).base_row_count(), 50);
+            EXPECT_EQ(get_latest_offset(partition_id).changelog_row_count(), 100);
+        } else {
+            EXPECT_FALSE(get_latest_offset(partition_id).has_base_row_count());
+            EXPECT_FALSE(get_latest_offset(partition_id).has_changelog_row_count());
+        }
 
         std::unique_ptr<Transaction> txn;
         ASSERT_EQ(service_->txn_kv()->create_txn(&txn), TxnErrorCode::TXN_OK);
@@ -2142,7 +2195,9 @@ TEST_F(MetaServiceTableStreamTest, CommitWritesOffsetsForEachMultiVersionMode) {
             TableStreamOffsetPB versioned_offset;
             ASSERT_TRUE(versioned_offset.ParseFromString(value));
             EXPECT_EQ(versioned_offset.state(), TABLE_STREAM_OFFSET_CONSUMED);
-            EXPECT_EQ(versioned_offset.offset_tso(), 120);
+            EXPECT_EQ(versioned_offset.offset_tso(), next_tso);
+            EXPECT_EQ(versioned_offset.base_row_count(), 50);
+            EXPECT_EQ(versioned_offset.changelog_row_count(), 100);
         } else {
             EXPECT_EQ(versioned_err, TxnErrorCode::TXN_KEY_NOT_FOUND);
         }

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

@@ -115,6 +115,14 @@ public void setIsChangelog(boolean isChangelog) {
         this.isChangelog = isChangelog;
     }
 
+    /**
+     * Local data identity, replaced on every tablet-set mutation (including same-ID restore).
+     * Writes, compaction and TTL do not replace this immutable topology snapshot.
+     */
+    public Object getDataEpoch() {
+        return tablets;
+    }
+
     public List<Tablet> getTablets() {
         // Volatile read: returns the current immutable snapshot; callers iterate without locking.
         return Collections.unmodifiableList(tablets);

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

@@ -20,10 +20,8 @@
 import org.apache.doris.catalog.stream.StreamReadMode;
 import org.apache.doris.cloud.catalog.CloudPartition;
 import org.apache.doris.common.Config;
-import org.apache.doris.common.Pair;
 
 import com.google.common.base.Preconditions;
-import com.google.common.collect.Maps;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -34,24 +32,21 @@
 import java.util.stream.Collectors;
 
 /**
- * A lightweight wrapper base for read changelog of table
+ * A changelog scan wrapper. Stream reads provide partition estimates for their resolved offset
+ * ranges; other changelog reads delegate to the corresponding changelog index statistics.
+ * These are approximate scales and can lag writes or retention changes.
  */
 public class OlapChangelogTableWrapper extends OlapTableWrapper {
 
     private final MaterializedIndexMeta changelogIndexMeta;
 
     public OlapChangelogTableWrapper(OlapTable originTable) {
-        this(originTable, Maps.newHashMap());
+        this(originTable, Collections.emptyMap());
     }
 
-    public OlapChangelogTableWrapper(OlapTable originTable, Map<Long, Pair<Long, Long>> partitionOffsetMap) {
-        this(originTable, partitionOffsetMap, Collections.emptyMap());
-    }
-
-    public OlapChangelogTableWrapper(OlapTable originTable, Map<Long, Pair<Long, Long>> partitionOffsetMap,
-            Map<Long, Long> partitionVisibleVersionMap) {
+    public OlapChangelogTableWrapper(OlapTable originTable, Map<Long, PartitionScanState> partitionScanStates) {
         super(originTable, originTable.getName(), originTable.getChangelogIndexMeta().getSchema(), KeysType.DUP_KEYS,
-                partitionOffsetMap, partitionVisibleVersionMap);
+                partitionScanStates);
         this.changelogIndexMeta = originTable.getChangelogIndexMeta();
         Preconditions.checkNotNull(changelogIndexMeta, "changelog meta is null, table=%s", originTable.getName());
         this.setBaseIndexId(changelogIndexMeta.getIndexId());