[refactor](stats) Refactor TableStatsMeta

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

@@ -32,6 +32,7 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ShowResultSet;
 import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.statistics.ColStatsMeta;
 import org.apache.doris.statistics.ColumnStatistic;
 
 import com.google.common.collect.ImmutableList;
@@ -54,6 +55,10 @@ public class ShowColumnStatsStmt extends ShowStmt {
                     .add("avg_size_byte")
                     .add("min")
                     .add("max")
+                    .add("method")
+                    .add("type")
+                    .add("trigger")
+                    .add("query_times")
                     .add("updated_time")
                     .build();
 
@@ -136,6 +141,7 @@ public ShowResultSet constructResultSet(List<Pair<String, ColumnStatistic>> colu
             if (p.second.isUnKnown) {
                 return;
             }
+
             List<String> row = Lists.newArrayList();
             row.add(p.first);
             row.add(String.valueOf(p.second.count));
@@ -145,6 +151,12 @@ public ShowResultSet constructResultSet(List<Pair<String, ColumnStatistic>> colu
             row.add(String.valueOf(p.second.avgSizeByte));
             row.add(String.valueOf(p.second.minExpr == null ? "N/A" : p.second.minExpr.toSql()));
             row.add(String.valueOf(p.second.maxExpr == null ? "N/A" : p.second.maxExpr.toSql()));
+            ColStatsMeta colStatsMeta = Env.getCurrentEnv().getAnalysisManager().findColStatsMeta(table.getId(),
+                    p.first);
+            row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.analysisMethod));
+            row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.analysisType));
+            row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.jobType));
+            row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.queriedTimes));
             row.add(String.valueOf(p.second.updatedTime));
             result.add(row);
         });

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

@@ -32,7 +32,7 @@
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.qe.ShowResultSet;
 import org.apache.doris.qe.ShowResultSetMetaData;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
@@ -49,8 +49,6 @@ public class ShowTableStatsStmt extends ShowStmt {
                     .add("updated_rows")
                     .add("query_times")
                     .add("row_count")
-                    .add("method")
-                    .add("type")
                     .add("updated_time")
                     .add("columns")
                     .add("trigger")
@@ -132,7 +130,7 @@ public long getPartitionId() {
         return table.getPartition(partitionName).getId();
     }
 
-    public ShowResultSet constructResultSet(TableStats tableStatistic) {
+    public ShowResultSet constructResultSet(TableStatsMeta tableStatistic) {
         if (tableStatistic == null) {
             return new ShowResultSet(getMetaData(), new ArrayList<>());
         }
@@ -141,8 +139,6 @@ public ShowResultSet constructResultSet(TableStats tableStatistic) {
         row.add(String.valueOf(tableStatistic.updatedRows));
         row.add(String.valueOf(tableStatistic.queriedTimes.get()));
         row.add(String.valueOf(tableStatistic.rowCount));
-        row.add(tableStatistic.analysisMethod.toString());
-        row.add(tableStatistic.analysisType.toString());
         row.add(new Date(tableStatistic.updatedTime).toString());
         row.add(tableStatistic.analyzeColumns().toString());
         row.add(tableStatistic.jobType.toString());

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

@@ -56,7 +56,7 @@
 import org.apache.doris.statistics.HistogramTask;
 import org.apache.doris.statistics.MVAnalysisTask;
 import org.apache.doris.statistics.OlapAnalysisTask;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.statistics.util.StatisticsUtil;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.SystemInfoService;
@@ -1125,12 +1125,15 @@ public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) {
         return new MVAnalysisTask(info);
     }
 
-    @Override
-    public boolean needReAnalyzeTable(TableStats tblStats) {
+    public boolean needReAnalyzeTable(TableStatsMeta tblStats) {
         if (tblStats == null) {
             return true;
         }
         long rowCount = getRowCount();
+        // TODO: Do we need to analyze an empty table?
+        if (rowCount == 0) {
+            return false;
+        }
         if (!tblStats.analyzeColumns().containsAll(getBaseSchema()
                 .stream()
                 .map(Column::getName)
@@ -1145,7 +1148,7 @@ public boolean needReAnalyzeTable(TableStats tblStats) {
     @Override
     public Map<String, Set<String>> findReAnalyzeNeededPartitions() {
         TableIf table = this;
-        TableStats tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(table.getId());
+        TableStatsMeta tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(table.getId());
         Set<String> allPartitions = table.getPartitionNames().stream().map(table::getPartition)
                 .filter(Partition::hasData).map(Partition::getName).collect(Collectors.toSet());
         if (tableStats == null) {

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

@@ -30,7 +30,7 @@
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.BaseAnalysisTask;
 import org.apache.doris.statistics.ColumnStatistic;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.thrift.TTableDescriptor;
 
 import com.google.common.base.Preconditions;
@@ -575,7 +575,7 @@ public Optional<ColumnStatistic> getColumnStatistic(String colName) {
     public void analyze(String dbName) {}
 
     @Override
-    public boolean needReAnalyzeTable(TableStats tblStats) {
+    public boolean needReAnalyzeTable(TableStatsMeta tblStats) {
         return true;
     }
 

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

@@ -23,7 +23,7 @@
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.BaseAnalysisTask;
 import org.apache.doris.statistics.ColumnStatistic;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.thrift.TTableDescriptor;
 
 import com.google.common.collect.Lists;
@@ -138,7 +138,7 @@ default int getBaseColumnIdxByName(String colName) {
 
     Optional<ColumnStatistic> getColumnStatistic(String colName);
 
-    boolean needReAnalyzeTable(TableStats tblStats);
+    boolean needReAnalyzeTable(TableStatsMeta tblStats);
 
     Map<String, Set<String>> findReAnalyzeNeededPartitions();
 

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

@@ -35,7 +35,7 @@
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.BaseAnalysisTask;
 import org.apache.doris.statistics.ColumnStatistic;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.thrift.TTableDescriptor;
 
 import com.google.common.collect.Sets;
@@ -383,7 +383,7 @@ public void gsonPostProcess() throws IOException {
     }
 
     @Override
-    public boolean needReAnalyzeTable(TableStats tblStats) {
+    public boolean needReAnalyzeTable(TableStatsMeta tblStats) {
         // TODO: Find a way to decide if this external table need to be reanalyzed.
         // For now, simply return true for all external tables.
         return true;

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.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.statistics.util.StatisticsUtil;
 import org.apache.doris.thrift.THiveTable;
 import org.apache.doris.thrift.TTableDescriptor;
@@ -436,7 +436,7 @@ public List<Column> getHudiSchema(List<FieldSchema> hmsSchema) {
     @Override
     public long estimatedRowCount() {
         try {
-            TableStats tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(id);
+            TableStatsMeta 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);

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

@@ -24,7 +24,7 @@
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.BaseAnalysisTask;
 import org.apache.doris.statistics.JdbcAnalysisTask;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.thrift.TTableDescriptor;
 
 import org.apache.logging.log4j.LogManager;
@@ -111,7 +111,7 @@ public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) {
     @Override
     public long getRowCount() {
         makeSureInitialized();
-        TableStats tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(id);
+        TableStatsMeta 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);

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

@@ -126,7 +126,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.statistics.TableStatsMeta;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.Frontend;
 import org.apache.doris.transaction.TransactionState;
@@ -889,7 +889,7 @@ public void readFields(DataInput in) throws IOException {
                 break;
             }
             case OperationType.OP_UPDATE_TABLE_STATS: {
-                data = TableStats.read(in);
+                data = TableStatsMeta.read(in);
                 isRead = true;
                 break;
             }

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

@@ -87,7 +87,7 @@
 import org.apache.doris.scheduler.job.JobTask;
 import org.apache.doris.statistics.AnalysisInfo;
 import org.apache.doris.statistics.AnalysisManager;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.Frontend;
 import org.apache.doris.transaction.TransactionState;
@@ -1124,7 +1124,7 @@ public static void loadJournal(Env env, Long logId, JournalEntity journal) {
                     break;
                 }
                 case OperationType.OP_UPDATE_TABLE_STATS: {
-                    env.getAnalysisManager().replayUpdateTableStatsStatus((TableStats) journal.getData());
+                    env.getAnalysisManager().replayUpdateTableStatsStatus((TableStatsMeta) journal.getData());
                     break;
                 }
                 case OperationType.OP_PERSIST_AUTO_JOB: {
@@ -1981,7 +1981,7 @@ public void logUpdateAutoIncrementId(AutoIncrementIdUpdateLog log) {
         logEdit(OperationType.OP_UPDATE_AUTO_INCREMENT_ID, log);
     }
 
-    public void logCreateTableStats(TableStats tableStats) {
+    public void logCreateTableStats(TableStatsMeta tableStats) {
         logEdit(OperationType.OP_UPDATE_TABLE_STATS, tableStats);
     }
 

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

@@ -199,7 +199,7 @@
 import org.apache.doris.statistics.ColumnStatistic;
 import org.apache.doris.statistics.Histogram;
 import org.apache.doris.statistics.StatisticsRepository;
-import org.apache.doris.statistics.TableStats;
+import org.apache.doris.statistics.TableStatsMeta;
 import org.apache.doris.statistics.query.QueryStatsUtil;
 import org.apache.doris.system.Backend;
 import org.apache.doris.system.Diagnoser;
@@ -2454,7 +2454,7 @@ private void handleShowDataSkew() throws AnalysisException {
     private void handleShowTableStats() {
         ShowTableStatsStmt showTableStatsStmt = (ShowTableStatsStmt) stmt;
         TableIf tableIf = showTableStatsStmt.getTable();
-        TableStats tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(tableIf.getId());
+        TableStatsMeta tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(tableIf.getId());
         /*
            HMSExternalTable table will fetch row count from HMS
            or estimate with file size and schema if it's not analyzed.

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

@@ -120,7 +120,7 @@ public class AnalysisManager extends Daemon implements Writable {
     // Tracking and control sync analyze tasks, keep in mem only
     private final ConcurrentMap<ConnectContext, SyncTaskCollection> ctxToSyncTask = new ConcurrentHashMap<>();
 
-    private final Map<Long, TableStats> idToTblStats = new ConcurrentHashMap<>();
+    private final Map<Long, TableStatsMeta> idToTblStats = new ConcurrentHashMap<>();
 
     protected SimpleQueue<AnalysisInfo> autoJobs = createSimpleQueue(null, this);
 
@@ -511,6 +511,7 @@ public AnalysisInfo buildAnalysisJobInfo(AnalyzeTblStmt stmt) throws DdlExceptio
         infoBuilder.setCatalogName(catalogName);
         infoBuilder.setDbName(db);
         infoBuilder.setTblName(tblName);
+        // TODO: Refactor later, DON'T MODIFY IT RIGHT NOW
         StringJoiner stringJoiner = new StringJoiner(",", "[", "]");
         for (String colName : columnNames) {
             stringJoiner.add(colName);
@@ -643,8 +644,14 @@ public void updateTableStats(AnalysisInfo jobInfo) {
         if (tbl instanceof ExternalTable) {
             return;
         }
-        // TODO: set updatedRows to 0, when loadedRows of transaction info is ready.
-        updateTableStatsStatus(new TableStats(tbl.getId(), tbl.estimatedRowCount(), jobInfo));
+        TableStatsMeta tableStats = findTableStatsStatus(tbl.getId());
+        if (tableStats == null) {
+            updateTableStatsStatus(new TableStatsMeta(tbl.getId(), tbl.estimatedRowCount(), jobInfo));
+        } else {
+            tableStats.updateByJob(jobInfo);
+            logCreateTableStats(tableStats);
+        }
+
     }
 
     public List<AnalysisInfo> showAnalysisJob(ShowAnalyzeStmt stmt) {
@@ -726,7 +733,7 @@ public void dropStats(DropStatsStmt dropStatsStmt) throws DdlException {
 
         Set<String> cols = dropStatsStmt.getColumnNames();
         long tblId = dropStatsStmt.getTblId();
-        TableStats tableStats = findTableStatsStatus(dropStatsStmt.getTblId());
+        TableStatsMeta tableStats = findTableStatsStatus(dropStatsStmt.getTblId());
         if (tableStats == null) {
             return;
         }
@@ -949,10 +956,10 @@ public static boolean needAbandon(AnalysisInfo analysisInfo) {
         return false;
     }
 
-    private static void readIdToTblStats(DataInput in, Map<Long, TableStats> map) throws IOException {
+    private static void readIdToTblStats(DataInput in, Map<Long, TableStatsMeta> map) throws IOException {
         int size = in.readInt();
         for (int i = 0; i < size; i++) {
-            TableStats tableStats = TableStats.read(in);
+            TableStatsMeta tableStats = TableStatsMeta.read(in);
             map.put(tableStats.tblId, tableStats);
         }
     }
@@ -980,7 +987,7 @@ private void writeJobInfo(DataOutput out, Map<Long, AnalysisInfo> infoMap) throw
 
     private void writeTableStats(DataOutput out) throws IOException {
         out.writeInt(idToTblStats.size());
-        for (Entry<Long, TableStats> entry : idToTblStats.entrySet()) {
+        for (Entry<Long, TableStatsMeta> entry : idToTblStats.entrySet()) {
             entry.getValue().write(out);
         }
     }
@@ -996,29 +1003,29 @@ public void addToJobIdTasksMap(long jobId, Map<Long, BaseAnalysisTask> tasks) {
         analysisJobIdToTaskMap.put(jobId, tasks);
     }
 
-    public TableStats findTableStatsStatus(long tblId) {
+    public TableStatsMeta findTableStatsStatus(long tblId) {
         return idToTblStats.get(tblId);
     }
 
     // Invoke this when load transaction finished.
     public void updateUpdatedRows(long tblId, long rows) {
-        TableStats statsStatus = idToTblStats.get(tblId);
+        TableStatsMeta statsStatus = idToTblStats.get(tblId);
         if (statsStatus != null) {
             statsStatus.updatedRows.addAndGet(rows);
             logCreateTableStats(statsStatus);
         }
     }
 
-    public void updateTableStatsStatus(TableStats tableStats) {
+    public void updateTableStatsStatus(TableStatsMeta tableStats) {
         replayUpdateTableStatsStatus(tableStats);
         logCreateTableStats(tableStats);
     }
 
-    public void replayUpdateTableStatsStatus(TableStats tableStats) {
+    public void replayUpdateTableStatsStatus(TableStatsMeta tableStats) {
         idToTblStats.put(tableStats.tblId, tableStats);
     }
 
-    public void logCreateTableStats(TableStats tableStats) {
+    public void logCreateTableStats(TableStatsMeta tableStats) {
         Env.getCurrentEnv().getEditLog().logCreateTableStats(tableStats);
     }
 
@@ -1030,7 +1037,7 @@ public void registerSysJob(AnalysisInfo jobInfo, Map<Long, BaseAnalysisTask> tas
 
     @VisibleForTesting
     protected Set<String> findReAnalyzeNeededPartitions(TableIf table) {
-        TableStats tableStats = findTableStatsStatus(table.getId());
+        TableStatsMeta tableStats = findTableStatsStatus(table.getId());
         if (tableStats == null) {
             return table.getPartitionNames().stream().map(table::getPartition)
                     .filter(Partition::hasData).map(Partition::getName).collect(Collectors.toSet());
@@ -1071,7 +1078,7 @@ protected SimpleQueue<AnalysisInfo> createSimpleQueue(Collection<AnalysisInfo> c
     // Remove col stats status from TableStats if failed load some col stats after analyze corresponding column so that
     // we could make sure it would be analyzed again soon if user or system submit job for that column again.
     public void removeColStatsStatus(long tblId, String colName) {
-        TableStats tableStats = findTableStatsStatus(tblId);
+        TableStatsMeta tableStats = findTableStatsStatus(tblId);
         if (tableStats != null) {
             tableStats.removeColumn(colName);
         }
@@ -1089,4 +1096,12 @@ public void removeTableStats(long tblId) {
     public void replayTableStatsDeletion(TableStatsDeletionLog log) {
         idToTblStats.remove(log.id);
     }
+
+    public ColStatsMeta findColStatsMeta(long tblId, String colName) {
+        TableStatsMeta tableStats = findTableStatsStatus(tblId);
+        if (tableStats == null) {
+            return null;
+        }
+        return tableStats.findColumnStatsMeta(colName);
+    }
 }