[fix](balance) do not pick tablets whose data size is zero for BE balance (#66499)

fe/fe-core/src/main/java/org/apache/doris/clone/BeLoadRebalancer.java

@@ -123,6 +123,10 @@ protected List<TabletSchedCtx> selectAlternativeTabletsForCluster(
                 .collect(Collectors.toList());
 
         boolean hasCandidateTablet = false;
+        // Only for logging. This counts scanned tablets, not balance attempts, so it is summarized
+        // once per round here instead of going into TabletSchedulerStat: when every replica size is
+        // still unreported, every tablet of every high load backend hits it on every round.
+        int zeroSizeTabletNum = 0;
 
         // choose tablets from high load backends.
         // BackendLoadStatistic is sorted by load score in ascend order,
@@ -225,6 +229,18 @@ protected List<TabletSchedCtx> selectAlternativeTabletsForCluster(
                         continue;
                     }
 
+                    // A tablet with no data relocates nothing, so moving it can not improve the disk usage
+                    // difference that triggered this balance. It only shifts the replica count term of the
+                    // load score, which destroys the round-robin distribution a newly created table got
+                    // from createTablets(). Zero also means the size has not been reported yet: replica
+                    // data size is not persisted in the image (see LocalTablet), so it remains zero after
+                    // an FE restart until the next tablet stat update, and balancing on an unknown size is
+                    // guesswork. This is the same rule as DiskRebalancer.completeSchedCtx().
+                    if (replicaDataSize <= 0) {
+                        zeroSizeTabletNum++;
+                        continue;
+                    }
+
                     hasCandidateTablet = true;
 
                     // for urgent disk, pick tablets order by size,
@@ -273,11 +289,16 @@ protected List<TabletSchedCtx> selectAlternativeTabletsForCluster(
         } // end for high backends
 
         if (!alternativeTablets.isEmpty()) {
-            LOG.info("select alternative tablets, medium: {}, is urgent: {}, num: {}, detail: {}",
-                    medium, isUrgent, alternativeTablets.size(), alternativeTabletInfos);
+            LOG.info("select alternative tablets, medium: {}, is urgent: {}, num: {},"
+                            + " skip {} tablets whose size is zero, detail: {}",
+                    medium, isUrgent, alternativeTablets.size(), zeroSizeTabletNum, alternativeTabletInfos);
         } else if (isUrgent && !hasCandidateTablet) {
-            LOG.info("urgent balance cann't found candidate tablets. medium: {}, tag: {}",
-                    medium, clusterStat.getTag());
+            LOG.info("urgent balance cann't found candidate tablets. medium: {}, tag: {},"
+                            + " skip {} tablets whose size is zero",
+                    medium, clusterStat.getTag(), zeroSizeTabletNum);
+        } else if (zeroSizeTabletNum > 0) {
+            LOG.info("no alternative tablet, {} tablets are skipped because their size is zero."
+                            + " medium: {}, tag: {}", zeroSizeTabletNum, medium, clusterStat.getTag());
         }
         return alternativeTablets;
     }
@@ -334,6 +355,16 @@ public void completeSchedCtx(TabletSchedCtx tabletCtx) throws SchedException {
                     "no replica on high load backend" + isUrgentInfo);
         }
 
+        // Recheck because selection and scheduling can be far apart. Zero also means the size has not been
+        // reported yet: replica data size is not persisted in the image (see LocalTablet), so it remains zero
+        // after an FE restart until the next tablet stat update. Balancing on an unknown size is guesswork.
+        // This is the same rule as DiskRebalancer.completeSchedCtx().
+        if (tabletCtx.getTabletSize() <= 0) {
+            schedulerStat.counterBalanceRejectByZeroDataSize.incrementAndGet();
+            throw new SchedException(Status.UNRECOVERABLE, SubCode.DIAGNOSE_IGNORE,
+                    "size of src replica is zero");
+        }
+
         // select a replica as source
         boolean setSource = false;
         for (Replica replica : replicas) {

fe/fe-core/src/main/java/org/apache/doris/clone/Rebalancer.java

@@ -64,6 +64,9 @@ public abstract class Rebalancer {
     protected Map<Long, PathSlot> backendsWorkingSlots;
     protected TabletInvertedIndex invertedIndex;
     protected SystemInfoService infoService;
+    // Owned by TabletScheduler, which injects itself via setSchedulerStat() after construction.
+    // Defaults to a standalone instance so that unit tests can build a Rebalancer on its own.
+    protected TabletSchedulerStat schedulerStat = new TabletSchedulerStat();
     // be id -> end time of prio
     protected Map<Long, Long> prioBackends = Maps.newConcurrentMap();
 
@@ -163,6 +166,10 @@ public void updateLoadStatistic(Map<Tag, LoadStatisticForTag> statisticMap) {
         this.statisticMap = statisticMap;
     }
 
+    public void setSchedulerStat(TabletSchedulerStat schedulerStat) {
+        this.schedulerStat = schedulerStat;
+    }
+
     public void updateAlterTableIds(Set<Long> alterTableIds) {
         this.alterTableIds = alterTableIds;
     }

fe/fe-core/src/main/java/org/apache/doris/clone/TabletScheduler.java

@@ -165,6 +165,8 @@ public TabletScheduler(Env env, SystemInfoService infoService, TabletInvertedInd
         }
         // if rebalancer can not get new task, then use diskRebalancer to get task
         this.diskRebalancer = new DiskRebalancer(infoService, invertedIndex, backendsWorkingSlots);
+        this.rebalancer.setSchedulerStat(stat);
+        this.diskRebalancer.setSchedulerStat(stat);
     }
 
     // for fe ut

fe/fe-core/src/main/java/org/apache/doris/clone/TabletSchedulerStat.java

@@ -111,6 +111,8 @@ public class TabletSchedulerStat {
     public AtomicLong counterReplicaMissingForTagErr = new AtomicLong(0L);
     @StatField("num of balance scheduled")
     public AtomicLong counterBalanceSchedule = new AtomicLong(0L);
+    @StatField("num of BE balance rejected for zero data size")
+    public AtomicLong counterBalanceRejectByZeroDataSize = new AtomicLong(0L);
     @StatField("num of colocate replica mismatch")
     public AtomicLong counterReplicaColocateMismatch = new AtomicLong(0L);
     @StatField("num of colocate replica redundant")