[Refactor] Rename deprecated_tablet_row_nums, unify first-load stats on tablet_stats, precompute reshard signals

Address review feedback on the tablet_row_nums deprecation:

- Rename the deprecated proto field tablet_row_nums -> deprecated_tablet_row_nums
  ([deprecated = true], json_name = "tablet_row_nums" for wire compatibility), matching
  the lake_types.proto deprecation convention.

- Fully consolidate first-load per-tablet stats onto PartitionCommitInfo.tabletStats: the
  shared-nothing publish path (LeaderImpl) now builds TabletStatPB from TTabletInfo,
  removing tabletIdToRowCountForPartitionFirstLoad and the Map<Long,Long> tabletRowNum
  parameter threaded through the Utils publish overloads; StatisticsCollectionTrigger and
  InsertOverwriteJobRunner read tabletStats[id].numRows. tabletStats is cleared after the
  leader-side apply to bound FE heap; lake first-load statistics sample from
  LakeTablet.getFuzzyRowCount() (set during apply) and the shared-nothing path keeps its
  own tabletStats (OlapTableTxnLogApplier never clears it), so the clear is unconditional
  and batch-publish-safe (no per-version coupling).

- Eliminate the separate locked full-table reshard re-walk: TabletStatMgr piggybacks the
  split/merge signal computation on its existing row-count scan, and the publish apply
  computes its partition-local signal and enqueues a candidate carrying the precomputed
  signal only when it crosses a threshold. TabletReshardJobMgr.computeReshardSignals is
  removed; the drain triggers from the carried signal. Signals are monotone, so a
  per-partition crossing is decision-safe; the periodic scan remains the authoritative
  fallback.

Signed-off-by: xiangguangyxg <xiangguangyxg@gmail.com>

Author: xiangguangyxg

be/test/service/lake_service_test.cpp

@@ -6058,7 +6058,7 @@ TEST_F(LakeServiceTest, test_publish_returns_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))
+        EXPECT_EQ(0, response.deprecated_tablet_row_nums().count(_tablet_id))
                 << "deprecated tablet_row_nums must no longer be populated";
     }
 

fe/fe-core/src/main/java/com/starrocks/alter/LakeRollupJob.java

@@ -776,7 +776,7 @@ protected boolean lakePublishVersion() {
                 if (useAggregatePublish) {
                     List<VectorIndexBuildInfoPB> vectorIndexBuildInfos = new ArrayList<>();
                     Utils.sendAggregatePublishVersionRequest(request, 1, computeResource, null, null,
-                            null, vectorIndexBuildInfos);
+                            vectorIndexBuildInfos);
                     VectorIndexBuildScheduler.onPublishComplete(vectorIndexBuildInfos, /* fromCompaction= */ false);
                 }
             }

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAlterMetaJobBase.java

@@ -338,7 +338,7 @@ protected boolean lakePublishVersion() {
                 if (useAggregatePublish) {
                     List<VectorIndexBuildInfoPB> vectorIndexBuildInfos = new ArrayList<>();
                     Utils.aggregatePublishVersion(tablets, Lists.newArrayList(txnInfo), commitVersion - 1, commitVersion,
-                                null, null, computeResource, null, null, vectorIndexBuildInfos);
+                                null, null, computeResource, null, vectorIndexBuildInfos);
                     VectorIndexBuildScheduler.onPublishComplete(vectorIndexBuildInfos, /* fromCompaction= */ false);
                 }
             }

fe/fe-core/src/main/java/com/starrocks/alter/LakeTableSchemaChangeJob.java

@@ -889,7 +889,7 @@ protected boolean lakePublishVersion() {
                 if (isFileBundling) {
                     List<VectorIndexBuildInfoPB> vectorIndexBuildInfos = new ArrayList<>();
                     Utils.sendAggregatePublishVersionRequest(request, 1, computeResource, null, null,
-                            null, vectorIndexBuildInfos);
+                            vectorIndexBuildInfos);
                     VectorIndexBuildScheduler.onPublishComplete(vectorIndexBuildInfos, /* fromCompaction= */ false);
                 }
             }

fe/fe-core/src/main/java/com/starrocks/alter/reshard/MergeTabletJob.java

@@ -538,7 +538,7 @@ private Map<Long, TabletRange> publishVersion(List<Tablet> tablets, long commitV
             Map<Long, TabletRange> tabletRange = new HashMap<>();
             List<VectorIndexBuildInfoPB> vectorIndexBuildInfos = new ArrayList<>();
             Utils.publishVersion(tablets, txnInfo, commitVersion - 1, commitVersion, null, tabletRange,
-                    computeResource, null, null, useAggregatePublish, vectorIndexBuildInfos);
+                    computeResource, null, useAggregatePublish, vectorIndexBuildInfos);
             VectorIndexBuildScheduler.onPublishComplete(vectorIndexBuildInfos, /* fromCompaction= */ false);
 
             return tabletRange;

fe/fe-core/src/main/java/com/starrocks/alter/reshard/SplitTabletJob.java

@@ -578,7 +578,7 @@ private Map<Long, TabletRange> publishVersion(List<Tablet> tablets, long commitV
             Map<Long, TabletRange> tabletRange = new HashMap<>();
             List<VectorIndexBuildInfoPB> vectorIndexBuildInfos = new ArrayList<>();
             Utils.publishVersion(tablets, txnInfo, commitVersion - 1, commitVersion, null, tabletRange,
-                    computeResource, null, null, useAggregatePublish, vectorIndexBuildInfos);
+                    computeResource, null, useAggregatePublish, vectorIndexBuildInfos);
             VectorIndexBuildScheduler.onPublishComplete(vectorIndexBuildInfos, /* fromCompaction= */ false);
 
             return tabletRange;

fe/fe-core/src/main/java/com/starrocks/alter/reshard/TabletReshardJobMgr.java

@@ -18,18 +18,12 @@
 import com.google.common.collect.Maps;
 import com.google.gson.annotations.SerializedName;
 import com.starrocks.catalog.Database;
-import com.starrocks.catalog.MaterializedIndex;
-import com.starrocks.catalog.MaterializedIndex.IndexExtState;
 import com.starrocks.catalog.OlapTable;
-import com.starrocks.catalog.Partition;
-import com.starrocks.catalog.PhysicalPartition;
 import com.starrocks.catalog.Table;
 import com.starrocks.catalog.Tablet;
 import com.starrocks.common.Config;
 import com.starrocks.common.StarRocksException;
 import com.starrocks.common.util.FrontendDaemon;
-import com.starrocks.common.util.concurrent.lock.LockType;
-import com.starrocks.common.util.concurrent.lock.Locker;
 import com.starrocks.lake.LakeTablet;
 import com.starrocks.metric.MetricRepo;
 import com.starrocks.persist.ImageWriter;
@@ -52,7 +46,6 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
 public class TabletReshardJobMgr extends FrontendDaemon implements GsonPostProcessable {
@@ -75,13 +68,23 @@ public class TabletReshardJobMgr extends FrontendDaemon implements GsonPostProce
     public record ReshardCandidateKey(long dbId, long tableId) {
     }
 
-    private final Set<ReshardCandidateKey> reshardCandidates = ConcurrentHashMap.newKeySet();
+    private final Map<ReshardCandidateKey, ReshardSignals> reshardCandidates = new ConcurrentHashMap<>();
 
-    public void markReshardCandidate(long dbId, long tableId) {
+    // Enqueue a table whose just-published partition already crossed a split/merge threshold,
+    // carrying the precomputed signal so the drain triggers without re-walking the whole table.
+    // Concurrent marks for the same table before the next drain coalesce by max/min.
+    public void markReshardCandidate(long dbId, long tableId,
+                                     long maxTabletSize, long minAdjacentTabletPairSize) {
         if (!isLeaderAdmissionOpen()) {
             return;
         }
-        reshardCandidates.add(new ReshardCandidateKey(dbId, tableId));
+        ReshardSignals fresh = new ReshardSignals();
+        fresh.maxTabletSize = maxTabletSize;
+        fresh.minAdjacentTabletPairSize = minAdjacentTabletPairSize;
+        reshardCandidates.merge(new ReshardCandidateKey(dbId, tableId), fresh, (old, incoming) -> {
+            old.merge(incoming);
+            return old;
+        });
     }
 
     public TabletReshardJobMgr() {
@@ -125,88 +128,73 @@ public void createTabletReshardJob(Database db, OlapTable table, MergeTabletClau
         addTabletReshardJob(job);
     }
 
-    private static class ReshardSignals {
-        long maxTabletSize = 0L;
-        long minAdjacentTabletPairSize = Long.MAX_VALUE;
+    public static class ReshardSignals {
+        public long maxTabletSize = 0L;
+        public long minAdjacentTabletPairSize = Long.MAX_VALUE;
+
+        void merge(ReshardSignals other) {
+            maxTabletSize = Math.max(maxTabletSize, other.maxTabletSize);
+            minAdjacentTabletPairSize = Math.min(minAdjacentTabletPairSize, other.minAdjacentTabletPairSize);
+        }
+    }
+
+    // Accumulate split/merge signals over one VISIBLE index's tablets into {@code signals}.
+    // maxTabletSize drives split (never gated); the adjacent-pair min drives merge and only
+    // contributes when the index is above the parallelism floor and both neighbors are fresh
+    // (LakeTablet dataSizeUpdateTime >= visibleVersionTime), so auto-merge cannot shrink an
+    // index below its pre-split parallelism. Pure in-memory field reads, no RPC. Called by the
+    // periodic TabletStatMgr scan (whole table) and the publish apply (one partition), both of
+    // which already iterate these tablets, so the signal is precomputed instead of re-walked
+    // under a fresh lock.
+    public static void accumulateIndexSignals(List<Tablet> tablets, int parallelismFloor,
+                                              long visibleVersionTime, ReshardSignals signals) {
+        boolean eligibleForMerge = tablets.size() > parallelismFloor;
+        long prevFreshTabletSize = -1L;
+        for (Tablet tablet : tablets) {
+            long dataSize = tablet.getDataSize(true);
+            signals.maxTabletSize = Math.max(signals.maxTabletSize, dataSize);
+            if (!(tablet instanceof LakeTablet)
+                    || ((LakeTablet) tablet).getDataSizeUpdateTime() < visibleVersionTime) {
+                prevFreshTabletSize = -1L;
+                continue;
+            }
+            if (prevFreshTabletSize >= 0 && eligibleForMerge) {
+                signals.minAdjacentTabletPairSize = Math.min(signals.minAdjacentTabletPairSize,
+                        prevFreshTabletSize + dataSize);
+            }
+            prevFreshTabletSize = dataSize;
+        }
+    }
+
+    // Parallelism floor for merge eligibility; returns 0 (ungated) when warehouse state is unavailable.
+    public static int safeComputeParallelismFloor(long tableId) {
+        try {
+            return TabletReshardUtils.computeParallelismFloor(tableId);
+        } catch (RuntimeException e) {
+            LOG.warn("Parallelism floor unavailable for table {}; auto-merge will not be floor-gated.", tableId, e);
+            return 0;
+        }
     }
 
     /**
      * Shared reshard-trigger entrypoint used by both the publish-driven candidate drain and the
-     * periodic TabletStatMgr scan. Self-gates on leader/admission, cloud-native range distribution,
-     * and NORMAL table state; recomputes split/merge signals from in-memory LakeTablet stats (no RPC).
+     * periodic TabletStatMgr scan, with split/merge signals precomputed by the caller (which is
+     * already walking the relevant tablets — no second lock or full-table re-walk here). Self-gates
+     * on leader/admission, cloud-native range distribution, and NORMAL table state; the authoritative
+     * NORMAL re-check happens in the job factory under its own lock.
      */
-    public void triggerTabletReshardForTable(Database db, OlapTable table) {
+    public void triggerTabletReshardForTable(Database db, OlapTable table,
+                                             long maxTabletSize, long minAdjacentTabletPairSize) {
         if (!isLeaderAdmissionOpen()) {
             return;
         }
         if (!table.isCloudNativeTableOrMaterializedView() || !table.isRangeDistribution()) {
             return;
         }
-        // Cheap pre-check; the authoritative NORMAL check happens under the read lock in computeReshardSignals.
         if (table.getState() != OlapTable.OlapTableState.NORMAL) {
             return;
         }
-        ReshardSignals signals = computeReshardSignals(db, table);
-        if (signals == null) {
-            return;
-        }
-        triggerTabletReshard(db, table, signals.maxTabletSize, signals.minAdjacentTabletPairSize);
-    }
-
-    // Returns null if the table is no longer reshard-eligible (state changed under the lock).
-    private ReshardSignals computeReshardSignals(Database db, OlapTable table) {
-        int parallelismFloor = resolveMergeParallelismFloor(db, table); // resolves warehouse state before the lock
-        ReshardSignals signals = new ReshardSignals();
-        Locker locker = new Locker();
-        locker.lockTableWithIntensiveDbLock(db.getId(), table.getId(), LockType.READ);
-        try {
-            if (table.getState() != OlapTable.OlapTableState.NORMAL) {
-                return null;
-            }
-            for (Partition partition : table.getAllPartitions()) {
-                for (PhysicalPartition physicalPartition : partition.getSubPartitions()) {
-                    long visibleVersionTime = physicalPartition.getVisibleVersionTime();
-                    for (MaterializedIndex index :
-                            physicalPartition.getLatestMaterializedIndices(IndexExtState.VISIBLE)) {
-                        List<Tablet> tablets = index.getTablets();
-                        // Only an index above the parallelism floor contributes the merge signal;
-                        // otherwise auto-merge could shrink it below the pre-split parallelism count.
-                        // Split detection (maxTabletSize) is never gated.
-                        boolean eligibleForMerge = tablets.size() > parallelismFloor;
-                        long prevFreshTabletSize = -1L;
-                        for (Tablet tablet : tablets) {
-                            long dataSize = tablet.getDataSize(true);
-                            signals.maxTabletSize = Math.max(signals.maxTabletSize, dataSize);
-                            if (!(tablet instanceof LakeTablet)
-                                    || ((LakeTablet) tablet).getDataSizeUpdateTime() < visibleVersionTime) {
-                                prevFreshTabletSize = -1L;
-                                continue;
-                            }
-                            if (prevFreshTabletSize >= 0 && eligibleForMerge) {
-                                signals.minAdjacentTabletPairSize = Math.min(signals.minAdjacentTabletPairSize,
-                                        prevFreshTabletSize + dataSize);
-                            }
-                            prevFreshTabletSize = dataSize;
-                        }
-                    }
-                }
-            }
-        } finally {
-            locker.unLockTableWithIntensiveDbLock(db.getId(), table.getId(), LockType.READ);
-        }
-        return signals;
-    }
-
-    // Precondition: caller has verified the table is a cloud-native range-distribution table.
-    private int resolveMergeParallelismFloor(Database db, OlapTable table) {
-        try {
-            return TabletReshardUtils.computeParallelismFloor(table.getId());
-        } catch (RuntimeException e) {
-            LOG.warn("Parallelism floor unavailable for table {}.{}; "
-                            + "auto-merge will not be floor-gated for this table.",
-                    db.getFullName(), table.getName(), e);
-            return 0;
-        }
+        triggerTabletReshard(db, table, maxTabletSize, minAdjacentTabletPairSize);
     }
 
     // Precondition: caller has verified the table is a cloud-native range-distribution table.
@@ -340,9 +328,12 @@ private void drainReshardCandidates() {
         if (reshardCandidates.isEmpty()) {
             return;
         }
-        List<ReshardCandidateKey> batch = new ArrayList<>(reshardCandidates);
-        reshardCandidates.removeAll(batch);
-        for (ReshardCandidateKey key : batch) {
+        // Snapshot the keys; remove each atomically so a concurrent re-mark is re-evaluated next tick.
+        for (ReshardCandidateKey key : new ArrayList<>(reshardCandidates.keySet())) {
+            ReshardSignals signals = reshardCandidates.remove(key);
+            if (signals == null) {
+                continue;
+            }
             // db and table lookups are not atomic; the null guards are conservative — a dropped db/table
             // is simply skipped this cycle.
             Database db = GlobalStateMgr.getCurrentState().getLocalMetastore().getDb(key.dbId());
@@ -354,7 +345,8 @@ private void drainReshardCandidates() {
             if (!(table instanceof OlapTable)) {
                 continue;
             }
-            triggerTabletReshardForTable(db, (OlapTable) table);
+            triggerTabletReshardForTable(db, (OlapTable) table,
+                    signals.maxTabletSize, signals.minAdjacentTabletPairSize);
         }
     }
 

fe/fe-core/src/main/java/com/starrocks/catalog/TabletStatMgr.java

@@ -37,6 +37,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.starrocks.alter.reshard.TabletReshardJobMgr;
 import com.starrocks.catalog.MaterializedIndex.IndexExtState;
 import com.starrocks.common.Config;
 import com.starrocks.common.ErrorReportException;
@@ -136,6 +137,15 @@ protected void runAfterCatalogReady() {
                 Map<Pair<Long, Long>, Long> indexRowCountMap = Maps.newHashMap();
                 // NOTE: calculate the row first with read lock, then update the stats with write lock
                 OlapTable olapTable = (OlapTable) table;
+                // Periodic reshard fallback: piggyback the split/merge signal computation on this
+                // existing all-tablet read-locked scan instead of re-locking and re-walking the table
+                // in TabletReshardJobMgr. Only cloud-native range-distribution tables can reshard.
+                boolean reshardEligible = olapTable.isCloudNativeTableOrMaterializedView()
+                        && olapTable.isRangeDistribution();
+                TabletReshardJobMgr.ReshardSignals reshardSignals =
+                        reshardEligible ? new TabletReshardJobMgr.ReshardSignals() : null;
+                int reshardParallelismFloor =
+                        reshardEligible ? TabletReshardJobMgr.safeComputeParallelismFloor(table.getId()) : 0;
                 locker.lockTableWithIntensiveDbLock(db.getId(), table.getId(), LockType.READ);
                 try {
                     for (Partition partition : olapTable.getAllPartitions()) {
@@ -149,6 +159,10 @@ protected void runAfterCatalogReady() {
                                 for (Tablet tablet : tablets) {
                                     indexRowCount += tablet.getRowCount(version);
                                 } // end for tablets
+                                if (reshardEligible) {
+                                    TabletReshardJobMgr.accumulateIndexSignals(tablets, reshardParallelismFloor,
+                                            physicalPartition.getVisibleVersionTime(), reshardSignals);
+                                }
                                 indexRowCountMap.put(Pair.create(physicalPartition.getId(), index.getId()),
                                         indexRowCount);
                                 if (!olapTable.isTempPartition(partition.getId())) {
@@ -183,9 +197,12 @@ protected void runAfterCatalogReady() {
                     locker.unLockTableWithIntensiveDbLock(db.getId(), table.getId(), LockType.WRITE);
                 }
 
-                // Trigger tablet reshard (publish-driven path is primary; this is the periodic fallback)
-                GlobalStateMgr.getCurrentState().getTabletReshardJobMgr()
-                        .triggerTabletReshardForTable(db, olapTable);
+                // Trigger tablet reshard with the signals computed above (publish-driven path is
+                // primary; this periodic scan is the authoritative fallback).
+                if (reshardEligible) {
+                    GlobalStateMgr.getCurrentState().getTabletReshardJobMgr().triggerTabletReshardForTable(
+                            db, olapTable, reshardSignals.maxTabletSize, reshardSignals.minAdjacentTabletPairSize);
+                }
             }
         }
         LOG.info("finished to update index row num of all databases. cost: {} ms",