[improvement](statistics)Skip auto analyze empty table. (#43865)

### What problem does this PR solve?

When doing auto analyze, skip the table if the row count of this table
is empty. This is more safe than write all 0 stats for this empty table,
because the row count reported by BE is not realtime.

Issue Number: close #xxx

Related PR: #xxx

Problem Summary:

### Release note

None

Author: Jibing-Li

fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java

@@ -145,9 +145,6 @@ protected void processOneJob(TableIf table, Set<Pair<String, String>> columns,
         // appendMvColumn(table, columns);
         appendAllColumns(table, columns);
         columns = columns.stream().filter(c -> StatisticsUtil.needAnalyzeColumn(table, c)).collect(Collectors.toSet());
-        if (columns.isEmpty()) {
-            return;
-        }
         AnalysisInfo analyzeJob = createAnalyzeJobForTbl(table, columns, priority);
         if (analyzeJob == null) {
             return;
@@ -206,17 +203,30 @@ protected AnalysisInfo createAnalyzeJobForTbl(
         if (StatisticsUtil.enablePartitionAnalyze() && table.isPartitionedTable()) {
             analysisMethod = AnalysisMethod.FULL;
         }
+        AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+        TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
         if (table instanceof OlapTable && analysisMethod.equals(AnalysisMethod.SAMPLE)) {
             OlapTable ot = (OlapTable) table;
             if (ot.getRowCountForIndex(ot.getBaseIndexId(), true) == TableIf.UNKNOWN_ROW_COUNT) {
                 LOG.info("Table {} row count is not fully reported, skip auto analyzing this time.", ot.getName());
                 return null;
             }
         }
-        AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
-        TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
-        long rowCount = StatisticsUtil.isEmptyTable(table, analysisMethod) ? 0 :
-                (table.getRowCount() <= 0 ? table.fetchRowCount() : table.getRowCount());
+        // We don't auto analyze empty table to avoid all 0 stats.
+        // Because all 0 is more dangerous than unknown stats when row count report is delayed.
+        long rowCount = table.getRowCount();
+        if (rowCount <= 0) {
+            LOG.info("Table {} is empty, remove its old stats and skip auto analyze it.", table.getName());
+            // Remove the table's old stats if exists.
+            if (tableStatsStatus != null && !tableStatsStatus.isColumnsStatsEmpty()) {
+                manager.dropStats(table, null);
+            }
+            return null;
+        }
+        if (jobColumns == null || jobColumns.isEmpty()) {
+            return null;
+        }
+        LOG.info("Auto analyze table {} row count is {}", table.getName(), rowCount);
         StringJoiner stringJoiner = new StringJoiner(",", "[", "]");
         for (Pair<String, String> pair : jobColumns) {
             stringJoiner.add(pair.toString());

fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java

@@ -255,4 +255,8 @@ public long getBaseIndexDeltaRowCount(OlapTable table) {
         }
         return updatedRows.get() - maxUpdateRows;
     }
+
+    public boolean isColumnsStatsEmpty() {
+        return colToColStatsMeta == null || colToColStatsMeta.isEmpty();
+    }
 }

fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java

@@ -166,6 +166,6 @@ public long getRowCountForIndex(long indexId, boolean strict) {
                 return 100;
             }
         };
-        Assertions.assertThrows(NullPointerException.class, () -> collector.createAnalyzeJobForTbl(table, null, null));
+        Assertions.assertNull(collector.createAnalyzeJobForTbl(table, null, null));
     }
 }

regression-test/suites/statistics/test_analyze_mv.groovy

@@ -671,7 +671,7 @@ suite("test_analyze_mv") {
     verifyTaskStatus(result_sample, "mva_MIN__`value3`", "mv3")
     verifyTaskStatus(result_sample, "mva_SUM__CAST(`value1` AS bigint)", "mv3")
 
-    // Test row count report and report for nereids
+    // * Test row count report and report for nereids
     sql """truncate table mvTestDup"""
     result_row = sql """show index stats mvTestDup mv3"""
     assertEquals(1, result_row.size())
@@ -680,6 +680,18 @@ suite("test_analyze_mv") {
     assertEquals("0", result_row[0][3])
     assertEquals("-1", result_row[0][4])
 
+    // ** Embedded test for skip auto analyze when table is empty
+    sql """analyze table mvTestDup properties ("use.auto.analyzer" = "true")"""
+    def empty_test = sql """show auto analyze mvTestDup"""
+    assertEquals(0, empty_test.size())
+    empty_test = sql """show column stats mvTestDup"""
+    assertEquals(0, empty_test.size())
+    // ** End of embedded test
+
+    sql """analyze table mvTestDup with sync"""
+    empty_test = sql """show column stats mvTestDup"""
+    assertEquals(12, empty_test.size())
+
     for (int i = 0; i < 120; i++) {
         result_row = sql """show index stats mvTestDup mv3"""
         logger.info("mv3 stats: " + result_row)
@@ -694,6 +706,23 @@ suite("test_analyze_mv") {
     assertEquals("mv3", result_row[0][1])
     assertEquals("0", result_row[0][3])
     assertEquals("0", result_row[0][4])
+
+    // ** Embedded test for skip auto analyze when table is empty again
+    sql """analyze table mvTestDup properties ("use.auto.analyzer" = "true")"""
+    empty_test = sql """show auto analyze mvTestDup"""
+    assertEquals(0, empty_test.size())
+    empty_test = sql """show column stats mvTestDup"""
+    for (int i = 0; i < 100; i++) {
+        empty_test = sql """show column stats mvTestDup"""
+        if (empty_test.size() != 0) {
+            logger.info("async delete is not finished yet.")
+            Thread.sleep(1000)
+        }
+        break
+    }
+    assertEquals(0, empty_test.size())
+    // ** End of embedded test
+
     sql """insert into mvTestDup values (1, 2, 3, 4, 5), (1, 2, 3, 4, 5), (10, 20, 30, 40, 50), (10, 20, 30, 40, 50), (100, 200, 300, 400, 500), (1001, 2001, 3001, 4001, 5001);"""
     result_row = sql """show index stats mvTestDup mv3"""
     assertEquals(1, result_row.size())

regression-test/suites/statistics/test_auto_analyze_black_white_list.groovy

@@ -68,8 +68,9 @@ suite("test_auto_analyze_black_white_list") {
         )
     """
 
+    sql """insert into test_bw values (1, 1, 1, 1, 1)"""
     try {
-        wait_row_count_reported("test_auto_analyze_black_white_list", "test_bw", 0, 4, "0")
+        wait_row_count_reported("test_auto_analyze_black_white_list", "test_bw", 0, 4, "1")
     } catch (Exception e) {
         logger.info(e.getMessage());
         return;