[Refactor] Deprecate lake tablet_row_nums; derive first-load row counts from tablet_stats

xiangguangyxg · xiangguangyxg · commit 5899a2aca59e · 2026-06-18T21:13:55.000+08:00
Mark PublishVersionResponse.tablet_row_nums (ordinal 4) as deprecated and stop
populating it. BE now reports first-import per-tablet stats through the newer
tablet_stats field: the publish gate broadens to emit stats for any tablet on
first import (base_version == 1) in addition to every range-distribution tablet,
computed via compute_tablet_stats without delvec I/O on the publish hot path. FE
derives first-load row counts from tablet_stats[*].num_rows via
collectFirstLoadRowCounts, replacing reads of the deprecated map. The
aggregate-publish merge of tablet_row_nums is removed.

For primary-key tablets the first-load row count is now the live-row count
(num_rows - num_dels) rather than the raw rowset sum the deprecated field
reported; this is more accurate and is self-healed by the periodic TabletStatMgr
scan. The field and ordinal are retained ([deprecated = true]) for wire
compatibility during a BE-before-FE rolling upgrade.

Signed-off-by: xiangguangyxg &lt;xiangguangyxg@gmail.com&gt;
diff --git a/be/src/service/service_be/lake_service.cpp b/be/src/service/service_be/lake_service.cpp
@@ -421,13 +421,18 @@ void LakeServiceImpl::publish_version(::google::protobuf::RpcController* control
                     if (res.ok()) {
                         auto metadata = std::move(res).value();
                         auto score = compaction_score(_tablet_mgr, metadata);
-                        // Per-tablet stats for range-distribution tablets, fed back to FE for real-time
-                        // split/merge triggering. Summed without any delvec I/O (publish hot path).
-                        const bool has_range_stats = metadata->has_range();
-                        int64_t range_num_rows = 0;
-                        int64_t range_data_size = 0;
-                        if (has_range_stats) {
-                            compute_tablet_stats(*metadata, &range_num_rows, &range_data_size);
+                        // Per-tablet stats returned to FE: for range-distribution tablets (every
+                        // publish) to drive real-time split/merge, and on first import (base_version==1)
+                        // for any tablet so FE can collect first-load statistics. Supersedes the
+                        // deprecated tablet_row_nums field; note num_rows here is the live-row count
+                        // (PK tablets subtract deletes) rather than that field's raw rowset row sum,
+                        // so a first load with intra-batch duplicate keys or deletes reports fewer
+                        // rows. Summed without delvec I/O (publish hot path).
+                        const bool emit_stats = metadata->has_range() || request->base_version() == 1;
+                        int64_t stats_num_rows = 0;
+                        int64_t stats_data_size = 0;
+                        if (emit_stats) {
+                            compute_tablet_stats(*metadata, &stats_num_rows, &stats_data_size);
                         }
                         // Copy metadata out of the lock(response_mtx), to let it execute in parallel.
                         TabletMetadataPB local_metadata;
@@ -437,17 +442,10 @@ void LakeServiceImpl::publish_version(::google::protobuf::RpcController* control
                         {
                             std::lock_guard l(response_mtx);
                             response->mutable_compaction_scores()->insert({metadata->id(), score});
-                            if (request->base_version() == 1) {
-                                int64_t row_nums = std::accumulate(
-                                        metadata->rowsets().begin(), metadata->rowsets().end(), 0,
-                                        [](int64_t sum, const auto& rowset) { return sum + rowset.num_rows(); });
-                                // Used to collect statistics when the partition is first imported
-                                response->mutable_tablet_row_nums()->insert({metadata->id(), row_nums});
-                            }
-                            if (has_range_stats) {
+                            if (emit_stats) {
                                 auto* stat = &(*response->mutable_tablet_stats())[metadata->id()];
-                                stat->set_num_rows(range_num_rows);
-                                stat->set_data_size(range_data_size);
+                                stat->set_num_rows(stats_num_rows);
+                                stat->set_data_size(stats_data_size);
                             }
                             if (skip_write_tablet_metadata) {
                                 (*response->mutable_tablet_metas())[metadata->id()].Swap(&local_metadata);
@@ -664,9 +662,6 @@ struct AggregatePublishContext {
         for (const auto& [tid, score] : resp->compaction_scores()) {
             (*response->mutable_compaction_scores())[tid] = score;
         }
-        for (const auto& [tid, row_num] : resp->tablet_row_nums()) {
-            (*response->mutable_tablet_row_nums())[tid] = row_num;
-        }
         for (const auto& [tid, stat] : resp->tablet_stats()) {
             (*response->mutable_tablet_stats())[tid] = stat;
         }
diff --git a/be/test/service/lake_service_test.cpp b/be/test/service/lake_service_test.cpp
@@ -6038,10 +6038,10 @@ TEST_F(LakeServiceTest, test_compute_tablet_stats) {
 // - A non-range tablet produces no tablet_stats entry.
 // - A range-distribution tablet (metadata has_range() == true) produces a
 //   tablet_stats entry with data_size > 0.
-TEST_F(LakeServiceTest, test_publish_returns_tablet_stats_for_range_tablet) {
-    // --- Negative: non-range tablet must NOT appear in tablet_stats ---
+TEST_F(LakeServiceTest, test_publish_returns_tablet_stats) {
+    // --- First load (base_version==1) of a non-range tablet: stats are now reported via
+    //     tablet_stats; the deprecated tablet_row_nums field is no longer populated. ---
     {
-        // Publish a txn on the default tablet (no range set).
         TxnLog log = generate_write_txn_log(2, 101, 4096);
         ASSERT_OK(_tablet_mgr->put_txn_log(log));
 
@@ -6055,7 +6055,30 @@ TEST_F(LakeServiceTest, test_publish_returns_tablet_stats_for_range_tablet) {
         _lake_service.publish_version(nullptr, &request, &response, nullptr);
 
         ASSERT_EQ(0, response.failed_tablets_size()) << response.status().error_msgs(0);
-        EXPECT_EQ(0, response.tablet_stats().count(_tablet_id)) << "non-range tablet must not appear in tablet_stats";
+        auto it = response.tablet_stats().find(_tablet_id);
+        ASSERT_NE(it, response.tablet_stats().end()) << "first-load tablet must report stats via tablet_stats";
+        EXPECT_EQ(101, it->second.num_rows()) << "first-load row count must flow through tablet_stats";
+        EXPECT_EQ(0, response.tablet_row_nums().count(_tablet_id))
+                << "deprecated tablet_row_nums must no longer be populated";
+    }
+
+    // --- Non-first-load (base_version>1) of a non-range tablet: must NOT appear in tablet_stats. ---
+    {
+        TxnLog log = generate_write_txn_log(1, 50, 2048);
+        ASSERT_OK(_tablet_mgr->put_txn_log(log));
+
+        PublishVersionRequest request;
+        request.set_base_version(2);
+        request.set_new_version(3);
+        request.add_tablet_ids(_tablet_id);
+        request.add_txn_ids(log.txn_id());
+
+        PublishVersionResponse response;
+        _lake_service.publish_version(nullptr, &request, &response, nullptr);
+
+        ASSERT_EQ(0, response.failed_tablets_size()) << response.status().error_msgs(0);
+        EXPECT_EQ(0, response.tablet_stats().count(_tablet_id))
+                << "non-range, non-first-load tablet must not appear in tablet_stats";
     }
 
     // --- Positive: range-distribution tablet MUST appear in tablet_stats ---
diff --git a/fe/fe-core/src/main/java/com/starrocks/lake/Utils.java b/fe/fe-core/src/main/java/com/starrocks/lake/Utils.java
@@ -14,6 +14,7 @@
 
 package com.starrocks.lake;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.staros.proto.ShardInfo;
 import com.starrocks.alter.reshard.PublishTabletsInfo;
@@ -179,6 +180,21 @@ public static boolean noOpPublishForForceSkip(long jobId, String reason, long wa
         }
     }
 
+    // Derive first-import (base_version==1) per-tablet row counts from the publish response's
+    // tablet_stats, now that the tablet_row_nums field is deprecated. BE populates tablet_stats
+    // for every first-load tablet (and every range-distribution tablet).
+    @VisibleForTesting
+    static void collectFirstLoadRowCounts(PublishVersionResponse response, Map<Long, Long> tabletRowNum) {
+        if (tabletRowNum == null || response == null || response.tabletStats == null) {
+            return;
+        }
+        response.tabletStats.forEach((tabletId, stat) -> {
+            if (stat != null && stat.numRows != null) {
+                tabletRowNum.put(tabletId, stat.numRows);
+            }
+        });
+    }
+
     public static void publishVersionBatch(@NotNull List<Tablet> tablets, List<TxnInfoPB> txnInfos,
                                            long baseVersion, long newVersion,
                                            Map<Long, Double> compactionScores,
@@ -256,8 +272,8 @@ public static void publishVersionBatch(@NotNull List<Tablet> tablets, List<TxnIn
                         tabletRanges.put(entry.getKey(), TabletRange.fromProto(entry.getValue()));
                     }
                 }
-                if (baseVersion == 1 && tabletRowNum != null && response != null && response.tabletRowNums != null) {
-                    tabletRowNum.putAll(response.tabletRowNums);
+                if (baseVersion == 1) {
+                    collectFirstLoadRowCounts(response, tabletRowNum);
                 }
                 if (tabletStats != null && response != null && response.tabletStats != null) {
                     tabletStats.putAll(response.tabletStats);
@@ -487,8 +503,8 @@ public static void sendAggregatePublishVersionRequest(AggregatePublishVersionReq
                     tabletRanges.put(entry.getKey(), TabletRange.fromProto(entry.getValue()));
                 }
             }
-            if (baseVersion == 1 && tabletRowNum != null && response != null && response.tabletRowNums != null) {
-                tabletRowNum.putAll(response.tabletRowNums);
+            if (baseVersion == 1) {
+                collectFirstLoadRowCounts(response, tabletRowNum);
             }
             if (tabletStats != null && response != null && response.tabletStats != null) {
                 tabletStats.putAll(response.tabletStats);
diff --git a/fe/fe-core/src/test/java/com/starrocks/lake/UtilsTest.java b/fe/fe-core/src/test/java/com/starrocks/lake/UtilsTest.java
@@ -16,6 +16,8 @@
 package com.starrocks.lake;
 
 import com.starrocks.common.StarRocksException;
+import com.starrocks.proto.PublishVersionResponse;
+import com.starrocks.proto.TabletStatPB;
 import com.starrocks.server.GlobalStateMgr;
 import com.starrocks.server.NodeMgr;
 import com.starrocks.server.WarehouseManager;
@@ -92,4 +94,33 @@ public void testGetWarehouseIdByNodeId() {
         Assertions.assertEquals(10001L, Utils.getWarehouseIdByNodeId(systemInfo, 10001).get().longValue());
         Assertions.assertEquals(10002L, Utils.getWarehouseIdByNodeId(systemInfo, 10002).get().longValue());
     }
+
+    @Test
+    public void testCollectFirstLoadRowCounts() {
+        java.util.Map<Long, Long> out = new java.util.HashMap<>();
+
+        // null response / null tablet_stats / null out-param -> no-op, no NPE
+        Utils.collectFirstLoadRowCounts(null, out);
+        PublishVersionResponse response = new PublishVersionResponse();
+        Utils.collectFirstLoadRowCounts(response, out);
+        Assertions.assertTrue(out.isEmpty());
+
+        // populated -> derive numRows; skip entries whose numRows is null
+        response.tabletStats = new java.util.HashMap<>();
+        TabletStatPB withRows = new TabletStatPB();
+        withRows.numRows = 7L;
+        withRows.dataSize = 70L;
+        TabletStatPB noRows = new TabletStatPB();
+        noRows.dataSize = 80L; // numRows left null -> must be skipped
+        response.tabletStats.put(11L, withRows);
+        response.tabletStats.put(22L, noRows);
+
+        Utils.collectFirstLoadRowCounts(response, out);
+        Assertions.assertEquals(1, out.size());
+        Assertions.assertEquals(Long.valueOf(7L), out.get(11L));
+        Assertions.assertNull(out.get(22L));
+
+        // null out-param -> no-op, no NPE
+        Utils.collectFirstLoadRowCounts(response, null);
+    }
 }
diff --git a/gensrc/proto/lake_service.proto b/gensrc/proto/lake_service.proto
@@ -60,16 +60,20 @@ message PublishVersionResponse {
     // Mapping from tablet id to compaction score.
     map<int64, double> compaction_scores = 2;
     optional StatusPB status = 3;
-    // Mapping from tablet id to row_nums when the partition is first imported
-    map<int64, int64> tablet_row_nums = 4;
+    // Deprecated: first-import row counts are now derived from |tablet_stats| (field 8),
+    // which BE also populates on first load. BE no longer writes this field; kept for
+    // wire compatibility during a BE-before-FE rolling upgrade. Do not reuse ordinal 4.
+    map<int64, int64> tablet_row_nums = 4 [deprecated = true];
     map<int64, TabletMetadataPB> tablet_metas = 5;
     // New tablet ranges for tablet splitting
     map<int64, TabletRangePB> tablet_ranges = 6;
     // Tablets that have async vector indexes and were published in this batch.
     // FE uses this to register new pending build tasks.
     repeated VectorIndexBuildInfoPB vector_index_build_infos = 7;
-    // Per-tablet aggregate stats for range-distribution tablets, returned on every
-    // publish so FE can trigger split/merge in real time. Empty for non-range tablets.
+    // Per-tablet aggregate stats. Returned for range-distribution tablets on every
+    // publish so FE can trigger split/merge in real time, and for any tablet on first
+    // import (base_version == 1) so FE can collect first-load statistics (supersedes the
+    // deprecated tablet_row_nums field). Empty for non-range tablets after first import.
     map<int64, TabletStatPB> tablet_stats = 8;
 }
 
