Merge branch 'branch-2.0' into fix_hudi_scan_lock

Author: lide-reed

be/src/common/daemon.cpp

@@ -409,6 +409,7 @@ static void init_doris_metrics(const std::vector<StorePath>& store_paths) {
 void signal_handler(int signal) {
     if (signal == SIGINT || signal == SIGTERM) {
         k_doris_exit = true;
+        k_doris_start = false;
         LOG(INFO) << "doris start to exit";
     }
 }

be/src/common/daemon.h

@@ -27,6 +27,7 @@ namespace doris {
 
 struct StorePath;
 inline bool k_doris_exit = false;
+inline bool k_doris_start = false;
 
 class Daemon {
 public:

be/src/runtime/thread_context.h

@@ -100,6 +100,7 @@ class MemTracker;
 class RuntimeState;
 
 extern bool k_doris_exit;
+extern bool k_doris_start;
 extern bthread_key_t btls_key;
 
 // Using gcc11 compiles thread_local variable on lower versions of GLIBC will report an error,
@@ -388,17 +389,17 @@ class AddThreadMemTrackerConsumer {
 // which is different from the previous behavior.
 #define CONSUME_MEM_TRACKER(size)                                                                  \
     do {                                                                                           \
-        if (doris::thread_context_ptr.init) {                                                      \
+        if (doris::k_doris_start && doris::thread_context_ptr.init) {                              \
             doris::thread_context()->consume_memory(size);                                         \
-        } else if (doris::ExecEnv::GetInstance()->initialized()) {                                 \
+        } else if (doris::k_doris_start && doris::ExecEnv::GetInstance()->initialized()) {         \
             doris::ExecEnv::GetInstance()->orphan_mem_tracker_raw()->consume_no_update_peak(size); \
         }                                                                                          \
     } while (0)
 #define RELEASE_MEM_TRACKER(size)                                                            \
     do {                                                                                     \
-        if (doris::thread_context_ptr.init) {                                                \
+        if (doris::k_doris_start && doris::thread_context_ptr.init) {                        \
             doris::thread_context()->consume_memory(-size);                                  \
-        } else if (doris::ExecEnv::GetInstance()->initialized()) {                           \
+        } else if (doris::k_doris_start && doris::ExecEnv::GetInstance()->initialized()) {   \
             doris::ExecEnv::GetInstance()->orphan_mem_tracker_raw()->consume_no_update_peak( \
                     -size);                                                                  \
         }                                                                                    \

be/src/service/doris_main.cpp

@@ -82,6 +82,7 @@ int __llvm_profile_write_file();
 
 namespace doris {
 extern bool k_doris_exit;
+extern bool k_doris_start;
 
 static void thrift_output(const char* x) {
     LOG(WARNING) << "thrift internal message: " << x;
@@ -463,6 +464,7 @@ int main(int argc, char** argv) {
 
     // init exec env
     auto exec_env = doris::ExecEnv::GetInstance();
+    doris::k_doris_start = true;
     doris::ExecEnv::init(exec_env, paths);
     doris::TabletSchemaCache::create_global_schema_cache();
 

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

@@ -219,6 +219,9 @@ public void createAnalysisJob(AnalyzeTblStmt stmt, boolean proxy) throws DdlExce
             }
             List<AnalysisInfo> jobs = new ArrayList<>();
             autoCollector.createAnalyzeJobForTbl(stmt.getDb(), jobs, stmt.getTable());
+            if (jobs.isEmpty()) {
+                return;
+            }
             AnalysisInfo job = autoCollector.getReAnalyzeRequiredPart(jobs.get(0));
             if (job != null) {
                 Env.getCurrentEnv().getStatisticsAutoCollector().createSystemAnalysisJob(job);

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

@@ -144,7 +144,6 @@ protected List<AnalysisInfo> constructAnalysisInfo(DatabaseIf<? extends TableIf>
             } catch (Throwable t) {
                 LOG.warn("Failed to analyze table {}.{}.{}",
                         db.getCatalog().getName(), db.getFullName(), table.getName(), t);
-                continue;
             }
         }
         return analysisInfos;
@@ -186,7 +185,19 @@ protected void createAnalyzeJobForTbl(DatabaseIf<? extends TableIf> db,
                 return;
             }
         }
-        long rowCount = StatisticsUtil.isEmptyTable(table, analysisMethod) ? 0 : 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.
+        AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+        TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
+        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);
+            }
+            return;
+        }
         AnalysisInfo jobInfo = new AnalysisInfoBuilder()
                 .setJobId(Env.getCurrentEnv().getNextId())
                 .setCatalogId(db.getCatalog().getId())

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

@@ -242,4 +242,8 @@ protected void clearStaleIndexRowCount(OlapTable table) {
     protected void addIndexRowForTest(long indexId, long rowCount) {
         indexesRowCount.put(indexId, rowCount);
     }
+
+    public boolean isColumnsStatsEmpty() {
+        return colNameToColStatsMeta == null || colNameToColStatsMeta.isEmpty();
+    }
 }

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

@@ -142,6 +142,11 @@ public List<Column> getSchemaAllIndexes(boolean full) {
                 columns.add(new Column("c2", PrimitiveType.HLL));
                 return columns;
             }
+
+            @Mock
+            public long getRowCount() {
+                return 1;
+            }
         };
         StatisticsAutoCollector saa = new StatisticsAutoCollector();
         List<AnalysisInfo> analysisInfoList = saa.constructAnalysisInfo(new Database(1, "anydb"));
@@ -397,6 +402,11 @@ public List<Long> getMvColumnIndexIds(String columnName) {
                 objects.add(-1L);
                 return objects;
             }
+
+            @Mock
+            public long getRowCount() {
+                return 1;
+            }
         };
 
         new MockUp<StatisticsUtil>() {
@@ -469,6 +479,11 @@ public List<Long> getMvColumnIndexIds(String columnName) {
                 objects.add(-1L);
                 return objects;
             }
+
+            @Mock
+            public long getRowCount() {
+                return 1;
+            }
         };
 
         new MockUp<StatisticsUtil>() {

regression-test/suites/statistics/test_analyze_mv.groovy

@@ -689,6 +689,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)
@@ -703,6 +715,27 @@ 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 """analyze table mvTestDup with sync"""
+        empty_test = sql """show column stats mvTestDup"""
+        assertEquals(12, empty_test.size())
+
         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())