Fix auto analyze doesn't filter unsupported type bug. (#27547)

Fix auto analyze doesn't filter unsupported type bug. Catch throwable in auto analyze thread for each database, otherwise the thread will quit when one database failed to create jobs and all other databases will not get analyzed. change FE config item full_auto_analyze_simultaneously_running_task_num to auto_analyze_simultaneously_running_task_num backport #27559
apache · Nov 24, 2023 · bb48353 · bb48353
1 parent 114052b
commit bb48353
Show file tree

Hide file tree

Showing 6 changed files with 27 additions and 12 deletions.
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -2156,7 +2156,7 @@ public class Config extends ConfigBase {
     public static int autobucket_min_buckets = 1;
 
     @ConfField
-    public static int full_auto_analyze_simultaneously_running_task_num = 1;
+    public static int auto_analyze_simultaneously_running_task_num = 1;
 
     @ConfField
     public static final int period_analyze_simultaneously_running_task_num = 1;

diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -1134,6 +1134,7 @@ public boolean needReAnalyzeTable(TableStatsMeta tblStats) {
         }
         if (!tblStats.analyzeColumns().containsAll(getBaseSchema()
                 .stream()
+                .filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
                 .map(Column::getName)
                 .collect(Collectors.toSet()))) {
             return true;
@@ -2297,10 +2298,14 @@ public Map<String, Set<String>> findReAnalyzeNeededPartitions() {
         Set<String> allPartitions = table.getPartitionNames().stream().map(table::getPartition)
                 .filter(Partition::hasData).map(Partition::getName).collect(Collectors.toSet());
         if (tableStats == null) {
-            return table.getBaseSchema().stream().collect(Collectors.toMap(Column::getName, v -> allPartitions));
+            return table.getBaseSchema().stream().filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+                    .collect(Collectors.toMap(Column::getName, v -> allPartitions));
         }
         Map<String, Set<String>> colToPart = new HashMap<>();
         for (Column col : table.getBaseSchema()) {
+            if (StatisticsUtil.isUnsupportedType(col.getType())) {
+                continue;
+            }
             long lastUpdateTime = tableStats.findColumnLastUpdateTime(col.getName());
             Set<String> partitions = table.getPartitionNames().stream()
                     .map(table::getPartition)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/ExternalTable.java
@@ -36,6 +36,7 @@
 import org.apache.doris.statistics.BaseAnalysisTask;
 import org.apache.doris.statistics.ColumnStatistic;
 import org.apache.doris.statistics.TableStatsMeta;
+import org.apache.doris.statistics.util.StatisticsUtil;
 import org.apache.doris.thrift.TTableDescriptor;
 
 import com.google.common.collect.Sets;
@@ -397,7 +398,8 @@ public Map<String, Set<String>> findReAnalyzeNeededPartitions() {
         HashSet<String> partitions = Sets.newHashSet();
         // TODO: Find a way to collect external table partitions that need to be analyzed.
         partitions.add("Dummy Partition");
-        return getBaseSchema().stream().collect(Collectors.toMap(Column::getName, k -> partitions));
+        return getBaseSchema().stream().filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+                .collect(Collectors.toMap(Column::getName, k -> partitions));
     }
 
     @Override

diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
@@ -24,6 +24,7 @@
 import org.apache.doris.catalog.TableIf;
 import org.apache.doris.catalog.external.ExternalTable;
 import org.apache.doris.common.Config;
+import org.apache.doris.common.DdlException;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.CatalogIf;
 import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
@@ -51,7 +52,7 @@ public class StatisticsAutoCollector extends StatisticsCollector {
     public StatisticsAutoCollector() {
         super("Automatic Analyzer",
                 TimeUnit.MINUTES.toMillis(Config.auto_check_statistics_in_minutes),
-                new AnalysisTaskExecutor(Config.full_auto_analyze_simultaneously_running_task_num));
+                new AnalysisTaskExecutor(Config.auto_analyze_simultaneously_running_task_num));
     }
 
     @Override
@@ -77,12 +78,17 @@ private void analyzeAll() {
                 if (StatisticConstants.SYSTEM_DBS.contains(databaseIf.getFullName())) {
                     continue;
                 }
-                analyzeDb(databaseIf);
+                try {
+                    analyzeDb(databaseIf);
+                } catch (Throwable t) {
+                    LOG.warn("Failed to analyze database {}.{}", ctl.getName(), databaseIf.getFullName(), t);
+                    continue;
+                }
             }
         }
     }
 
-    public void analyzeDb(DatabaseIf<TableIf> databaseIf) {
+    public void analyzeDb(DatabaseIf<TableIf> databaseIf) throws DdlException {
         List<AnalysisInfo> analysisInfos = constructAnalysisInfo(databaseIf);
         for (AnalysisInfo analysisInfo : analysisInfos) {
             analysisInfo = getReAnalyzeRequiredPart(analysisInfo);
@@ -92,7 +98,8 @@ public void analyzeDb(DatabaseIf<TableIf> databaseIf) {
             try {
                 createSystemAnalysisJob(analysisInfo);
             } catch (Exception e) {
-                LOG.warn("Failed to create analysis job", e);
+                analysisInfo.message = e.getMessage();
+                throw e;
             }
         }
     }
@@ -136,8 +143,7 @@ protected void createAnalyzeJobForTbl(DatabaseIf<? extends TableIf> db,
                 .setTblId(table.getId())
                 .setColName(
                         table.getBaseSchema().stream().filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
-                                .map(
-                                        Column::getName).collect(Collectors.joining(","))
+                                .map(Column::getName).collect(Collectors.joining(","))
                 )
                 .setAnalysisType(AnalysisInfo.AnalysisType.FUNDAMENTALS)
                 .setAnalysisMode(AnalysisInfo.AnalysisMode.INCREMENTAL)

diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
@@ -23,6 +23,7 @@
 import org.apache.doris.common.io.Writable;
 import org.apache.doris.persist.gson.GsonUtils;
 import org.apache.doris.statistics.AnalysisInfo.JobType;
+import org.apache.doris.statistics.util.StatisticsUtil;
 
 import com.google.gson.annotations.SerializedName;
 
@@ -136,8 +137,9 @@ public void update(AnalysisInfo analyzedJob, TableIf tableIf) {
         }
         jobType = analyzedJob.jobType;
         if (tableIf != null && analyzedJob.colToPartitions.keySet()
-                .containsAll(tableIf.getBaseSchema().stream().map(Column::getName).collect(
-                        Collectors.toSet()))) {
+                .containsAll(tableIf.getBaseSchema().stream()
+                    .filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+                    .map(Column::getName).collect(Collectors.toSet()))) {
             updatedRows.set(0);
         }
     }

diff --git a/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java b/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java
@@ -247,7 +247,7 @@ public boolean enableAutoAnalyze() {
     @Test
     public void checkAvailableThread() {
         StatisticsAutoCollector autoCollector = new StatisticsAutoCollector();
-        Assertions.assertEquals(Config.full_auto_analyze_simultaneously_running_task_num,
+        Assertions.assertEquals(Config.auto_analyze_simultaneously_running_task_num,
                 autoCollector.analysisTaskExecutor.executors.getMaximumPoolSize());
     }