[refactor](nereids) Refactor stats collection framework

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -2087,7 +2087,7 @@ public class Config extends ConfigBase {
     public static int force_olap_table_replication_num = 0;
 
     @ConfField
-    public static int full_auto_analyze_simultaneously_running_task_num = 5;
+    public static int full_auto_analyze_simultaneously_running_task_num = 1;
 
     @ConfField
     public static int cpu_resource_limit_per_analyze_task = 1;
@@ -2114,4 +2114,7 @@ public class Config extends ConfigBase {
 
     @ConfField
     public static boolean forbid_running_alter_job = false;
+
+    @ConfField
+    public static int table_stats_health_threshold = 80;
 }

fe/fe-core/src/main/cup/sql_parser.cup

@@ -5998,12 +5998,6 @@ with_analysis_properties ::=
             put("incremental", "true");
         }};
     :}
-    | KW_AUTO
-    {:
-        RESULT = new HashMap<String, String>() {{
-            put("automatic", "true");
-        }};
-    :}
     | KW_SAMPLE KW_PERCENT INTEGER_LITERAL:samplePercent
     {:
         RESULT = new HashMap<String, String>() {{

fe/fe-core/src/main/java/org/apache/doris/analysis/AnalyzeProperties.java

@@ -22,6 +22,7 @@
 import org.apache.doris.statistics.AnalysisInfo.AnalysisType;
 
 import com.google.common.collect.ImmutableSet;
+import com.google.gson.annotations.SerializedName;
 import org.apache.commons.lang3.StringUtils;
 import org.quartz.CronExpression;
 
@@ -32,8 +33,6 @@
 
 public class AnalyzeProperties {
 
-    private final Map<String, String> properties;
-
     public static final String PROPERTY_SYNC = "sync";
     public static final String PROPERTY_INCREMENTAL = "incremental";
     public static final String PROPERTY_AUTOMATIC = "automatic";
@@ -55,6 +54,9 @@ public class AnalyzeProperties {
 
     private CronExpression cronExpression;
 
+    @SerializedName("analyzeProperties")
+    private final Map<String, String> properties;
+
     private static final ImmutableSet<String> PROPERTIES_SET = new ImmutableSet.Builder<String>()
             .add(PROPERTY_SYNC)
             .add(PROPERTY_INCREMENTAL)

fe/fe-core/src/main/java/org/apache/doris/analysis/ShowTableStatsStmt.java

@@ -32,22 +32,28 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ShowResultSet;
 import org.apache.doris.qe.ShowResultSetMetaData;
-import org.apache.doris.statistics.TableStatistic;
-import org.apache.doris.statistics.util.StatisticsUtil;
+import org.apache.doris.statistics.TableStats;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
 
+import java.sql.Date;
+import java.util.ArrayList;
 import java.util.List;
 
 public class ShowTableStatsStmt extends ShowStmt {
 
     // TODO add more columns
     private static final ImmutableList<String> TITLE_NAMES =
             new ImmutableList.Builder<String>()
-                    .add("row_count")
-                    .add("update_time")
-                    .add("last_analyze_time")
+                    .add("updated_rows")
+                    .add("query_times")
+                    .add("row_count(for external_table only)")
+                    .add("method")
+                    .add("type")
+                    .add("updated_time")
+                    .add("columns")
+                    .add("trigger")
                     .build();
 
     private final TableName tableName;
@@ -126,12 +132,20 @@ public long getPartitionId() {
         return table.getPartition(partitionName).getId();
     }
 
-    public ShowResultSet constructResultSet(TableStatistic tableStatistic) {
+    public ShowResultSet constructResultSet(TableStats tableStatistic) {
+        if (tableStatistic == null) {
+            return new ShowResultSet(getMetaData(), new ArrayList<>());
+        }
         List<List<String>> result = Lists.newArrayList();
         List<String> row = Lists.newArrayList();
+        row.add(String.valueOf(tableStatistic.updatedRows));
+        row.add(String.valueOf(tableStatistic.queriedTimes.get()));
         row.add(String.valueOf(tableStatistic.rowCount));
-        row.add(String.valueOf(tableStatistic.updateTime));
-        row.add(StatisticsUtil.getReadableTime(tableStatistic.lastAnalyzeTimeInMs));
+        row.add(tableStatistic.analysisMethod.toString());
+        row.add(tableStatistic.analysisType.toString());
+        row.add(new Date(tableStatistic.updatedTime).toString());
+        row.add(tableStatistic.columns);
+        row.add(tableStatistic.jobType.toString());
         result.add(row);
         return new ShowResultSet(getMetaData(), result);
     }

fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchemaInitializer.java

@@ -84,7 +84,6 @@ public void run() {
             return;
         }
         Database database = op.get();
-        modifyTblReplicaCount(database, StatisticConstants.ANALYSIS_TBL_NAME);
         modifyTblReplicaCount(database, StatisticConstants.STATISTIC_TBL_NAME);
         modifyTblReplicaCount(database, StatisticConstants.HISTOGRAM_TBL_NAME);
     }
@@ -127,7 +126,6 @@ public void modifyTblReplicaCount(Database database, String tblName) {
     }
 
     private void createTbl() throws UserException {
-        Env.getCurrentEnv().getInternalCatalog().createTable(buildAnalysisTblStmt());
         Env.getCurrentEnv().getInternalCatalog().createTable(buildStatisticsTblStmt());
         Env.getCurrentEnv().getInternalCatalog().createTable(buildHistogramTblStmt());
     }
@@ -146,41 +144,6 @@ public static void createDB() {
         }
     }
 
-    @VisibleForTesting
-    public CreateTableStmt buildAnalysisTblStmt() throws UserException {
-        TableName tableName = new TableName("",
-                FeConstants.INTERNAL_DB_NAME, StatisticConstants.ANALYSIS_TBL_NAME);
-        List<ColumnDef> columnDefs = new ArrayList<>();
-        columnDefs.add(new ColumnDef("id", TypeDef.createVarchar(StatisticConstants.ID_LEN)));
-        columnDefs.add(new ColumnDef("catalog_id", TypeDef.createVarchar(StatisticConstants.MAX_NAME_LEN)));
-        columnDefs.add(new ColumnDef("db_id", TypeDef.createVarchar(StatisticConstants.MAX_NAME_LEN)));
-        columnDefs.add(new ColumnDef("tbl_id", TypeDef.createVarchar(StatisticConstants.MAX_NAME_LEN)));
-        columnDefs.add(new ColumnDef("idx_id", TypeDef.createVarchar(StatisticConstants.MAX_NAME_LEN)));
-        ColumnDef partId = new ColumnDef("part_id", TypeDef.createVarchar(StatisticConstants.MAX_NAME_LEN));
-        partId.setAllowNull(true);
-        columnDefs.add(partId);
-        columnDefs.add(new ColumnDef("count", TypeDef.create(PrimitiveType.BIGINT)));
-        columnDefs.add(new ColumnDef("last_analyze_time_in_ms", TypeDef.create(PrimitiveType.BIGINT)));
-        columnDefs.add(new ColumnDef("update_time", TypeDef.create(PrimitiveType.DATETIME)));
-        String engineName = "olap";
-        ArrayList<String> uniqueKeys = Lists.newArrayList("id", "catalog_id",
-                "db_id", "tbl_id", "idx_id", "part_id");
-        KeysDesc keysDesc = new KeysDesc(KeysType.UNIQUE_KEYS, uniqueKeys);
-        DistributionDesc distributionDesc = new HashDistributionDesc(
-                StatisticConstants.STATISTIC_TABLE_BUCKET_COUNT, uniqueKeys);
-        Map<String, String> properties = new HashMap<String, String>() {
-            {
-                put("replication_num", String.valueOf(
-                        Math.max(1, Config.min_replication_num_per_tablet)));
-            }
-        };
-        CreateTableStmt createTableStmt = new CreateTableStmt(true, false,
-                tableName, columnDefs, engineName, keysDesc, null, distributionDesc,
-                properties, null, "Doris internal statistics table, DO NOT MODIFY IT", null);
-        StatisticsUtil.analyze(createTableStmt);
-        return createTableStmt;
-    }
-
     @VisibleForTesting
     public CreateTableStmt buildStatisticsTblStmt() throws UserException {
         TableName tableName = new TableName("",
@@ -264,8 +227,7 @@ private boolean created() {
             return false;
         }
         Database db = optionalDatabase.get();
-        return db.getTable(StatisticConstants.ANALYSIS_TBL_NAME).isPresent()
-                && db.getTable(StatisticConstants.STATISTIC_TBL_NAME).isPresent()
+        return db.getTable(StatisticConstants.STATISTIC_TBL_NAME).isPresent()
                 && db.getTable(StatisticConstants.HISTOGRAM_TBL_NAME).isPresent();
     }
 

fe/fe-core/src/main/java/org/apache/doris/catalog/external/HMSExternalTable.java

@@ -32,7 +32,7 @@
 import org.apache.doris.statistics.ColumnStatistic;
 import org.apache.doris.statistics.ColumnStatisticBuilder;
 import org.apache.doris.statistics.HMSAnalysisTask;
-import org.apache.doris.statistics.TableStatistic;
+import org.apache.doris.statistics.TableStats;
 import org.apache.doris.statistics.util.StatisticsUtil;
 import org.apache.doris.thrift.THiveTable;
 import org.apache.doris.thrift.TTableDescriptor;
@@ -107,6 +107,9 @@ public class HMSExternalTable extends ExternalTable {
 
     private DLAType dlaType = DLAType.UNKNOWN;
 
+    // No as precise as row count in TableStats, but better than none.
+    private long estimatedRowCount = -1;
+
     public enum DLAType {
         UNKNOWN, HIVE, HUDI, ICEBERG
     }
@@ -422,13 +425,20 @@ public List<Column> getHudiSchema(List<FieldSchema> hmsSchema) {
     @Override
     public long estimatedRowCount() {
         try {
-            Optional<TableStatistic> tableStatistics = Env.getCurrentEnv().getStatisticsCache().getTableStatistics(
-                    catalog.getId(), catalog.getDbOrAnalysisException(dbName).getId(), id);
-            if (tableStatistics.isPresent()) {
-                long rowCount = tableStatistics.get().rowCount;
+            TableStats tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(id);
+            if (tableStats != null) {
+                long rowCount = tableStats.rowCount;
                 LOG.debug("Estimated row count for db {} table {} is {}.", dbName, name, rowCount);
                 return rowCount;
             }
+
+            if (estimatedRowCount != -1) {
+                return estimatedRowCount;
+            }
+            // Cache the estimated row count in this structure
+            // though the table never get analyzed, since the row estimation might be expensive caused by RPC.
+            estimatedRowCount = getRowCount();
+            return estimatedRowCount;
         } catch (Exception e) {
             LOG.warn("Fail to get row count for table {}", name, e);
         }

fe/fe-core/src/main/java/org/apache/doris/journal/JournalEntity.java

@@ -122,6 +122,7 @@
 import org.apache.doris.scheduler.job.Job;
 import org.apache.doris.scheduler.job.JobTask;
 import org.apache.doris.statistics.AnalysisInfo;
+import org.apache.doris.statistics.TableStats;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.Frontend;
 import org.apache.doris.transaction.TransactionState;
@@ -858,6 +859,11 @@ public void readFields(DataInput in) throws IOException {
                 isRead = true;
                 break;
             }
+            case OperationType.OP_UPDATE_TABLE_STATS: {
+                data = TableStats.read(in);
+                isRead = true;
+                break;
+            }
             default: {
                 IOException e = new IOException();
                 LOG.error("UNKNOWN Operation Type {}", opCode, e);

fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java

@@ -86,6 +86,7 @@
 import org.apache.doris.scheduler.job.Job;
 import org.apache.doris.scheduler.job.JobTask;
 import org.apache.doris.statistics.AnalysisInfo;
+import org.apache.doris.statistics.TableStats;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.Frontend;
 import org.apache.doris.transaction.TransactionState;
@@ -1088,6 +1089,10 @@ public static void loadJournal(Env env, Long logId, JournalEntity journal) {
                     env.replayAutoIncrementIdUpdateLog((AutoIncrementIdUpdateLog) journal.getData());
                     break;
                 }
+                case OperationType.OP_UPDATE_TABLE_STATS: {
+                    env.getAnalysisManager().replayUpdateTableStatsStatus((TableStats) journal.getData());
+                    break;
+                }
                 default: {
                     IOException e = new IOException();
                     LOG.error("UNKNOWN Operation Type {}", opCode, e);
@@ -1906,4 +1911,8 @@ public long logBarrier(BarrierLog log) {
     public void logUpdateAutoIncrementId(AutoIncrementIdUpdateLog log) {
         logEdit(OperationType.OP_UPDATE_AUTO_INCREMENT_ID, log);
     }
+
+    public void logCreateTableStats(TableStats tableStats) {
+        logEdit(OperationType.OP_UPDATE_TABLE_STATS, tableStats);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/persist/OperationType.java

@@ -323,6 +323,7 @@ public class OperationType {
     public static final short OP_CREATE_SCHEDULER_TASK = 453;
     public static final short OP_DELETE_SCHEDULER_TASK = 454;
 
+    public static final short OP_UPDATE_TABLE_STATS = 455;
 
     /**
      * Get opcode name by op code.

fe/fe-core/src/main/java/org/apache/doris/qe/DdlExecutor.java

@@ -38,7 +38,6 @@
 import org.apache.doris.analysis.AlterRoutineLoadStmt;
 import org.apache.doris.analysis.AlterSqlBlockRuleStmt;
 import org.apache.doris.analysis.AlterSystemStmt;
-import org.apache.doris.analysis.AlterTableStatsStmt;
 import org.apache.doris.analysis.AlterTableStmt;
 import org.apache.doris.analysis.AlterUserStmt;
 import org.apache.doris.analysis.AlterViewStmt;
@@ -165,8 +164,6 @@ public static void execute(Env env, DdlStmt ddlStmt) throws Exception {
             env.createMaterializedView((CreateMaterializedViewStmt) ddlStmt);
         } else if (ddlStmt instanceof AlterTableStmt) {
             env.alterTable((AlterTableStmt) ddlStmt);
-        } else if (ddlStmt instanceof AlterTableStatsStmt) {
-            StatisticsRepository.alterTableStatistics((AlterTableStatsStmt) ddlStmt);
         } else if (ddlStmt instanceof AlterColumnStatsStmt) {
             StatisticsRepository.alterColumnStatistics((AlterColumnStatsStmt) ddlStmt);
         } else if (ddlStmt instanceof AlterViewStmt) {

fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java

@@ -138,7 +138,6 @@
 import org.apache.doris.catalog.TabletInvertedIndex;
 import org.apache.doris.catalog.TabletMeta;
 import org.apache.doris.catalog.View;
-import org.apache.doris.catalog.external.ExternalTable;
 import org.apache.doris.catalog.external.HMSExternalTable;
 import org.apache.doris.clone.DynamicPartitionScheduler;
 import org.apache.doris.cluster.ClusterNamespace;
@@ -198,7 +197,7 @@
 import org.apache.doris.statistics.ColumnStatistic;
 import org.apache.doris.statistics.Histogram;
 import org.apache.doris.statistics.StatisticsRepository;
-import org.apache.doris.statistics.TableStatistic;
+import org.apache.doris.statistics.TableStats;
 import org.apache.doris.statistics.query.QueryStatsUtil;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.Diagnoser;
@@ -241,7 +240,6 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
-import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
@@ -2413,30 +2411,8 @@ private void handleShowDataSkew() throws AnalysisException {
     private void handleShowTableStats() {
         ShowTableStatsStmt showTableStatsStmt = (ShowTableStatsStmt) stmt;
         TableIf tableIf = showTableStatsStmt.getTable();
-        long partitionId = showTableStatsStmt.getPartitionId();
-        boolean showCache = showTableStatsStmt.isCached();
-        try {
-            if (tableIf instanceof ExternalTable && showCache) {
-                Optional<TableStatistic> tableStatistics = Env.getCurrentEnv().getStatisticsCache().getTableStatistics(
-                        tableIf.getDatabase().getCatalog().getId(),
-                        tableIf.getDatabase().getId(),
-                        tableIf.getId());
-                if (tableStatistics.isPresent()) {
-                    resultSet = showTableStatsStmt.constructResultSet(tableStatistics.get());
-                } else {
-                    resultSet = showTableStatsStmt.constructResultSet(TableStatistic.UNKNOWN);
-                }
-            } else if (partitionId > 0) {
-                TableStatistic partStats = StatisticsRepository.fetchTableLevelOfPartStats(partitionId);
-                resultSet = showTableStatsStmt.constructResultSet(partStats);
-            } else {
-                TableStatistic tableStats = StatisticsRepository.fetchTableLevelStats(tableIf.getId());
-                resultSet = showTableStatsStmt.constructResultSet(tableStats);
-            }
-        } catch (DdlException e) {
-            LOG.warn("Table statistics do not exist: {}", tableIf.getName());
-            resultSet = showTableStatsStmt.constructResultSet(TableStatistic.UNKNOWN);
-        }
+        TableStats tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(tableIf.getId());
+        resultSet = showTableStatsStmt.constructResultSet(tableStats);
     }
 
     private void handleShowColumnStats() throws AnalysisException {