[core] Support statistic with time travel

docs/content/maintenance/system-tables.md

@@ -369,3 +369,19 @@ SELECT * FROM sys.catalog_options;
 */
 ```
 
+### Statistic Table
+You can query the statistic information through statistic table.
+
+```sql
+SELECT * FROM T$statistics;
+
+/*
++--------------+------------+-----------------------+------------------+----------+
+|  snapshot_id |  schema_id |     mergedRecordCount | mergedRecordSize |  colstat |
++--------------+------------+-----------------------+------------------+----------+
+|            2 |          0 |              2        |         2        |    {}    |
++--------------+------------+-----------------------+------------------+----------+
+1 rows in set
+*/
+```
+

paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java

@@ -166,8 +166,18 @@ public Identifier identifier() {
 
     @Override
     public Optional<Statistics> statistics() {
-        // todo: support time travel
-        Snapshot latestSnapshot = snapshotManager().latestSnapshot();
+        Snapshot latestSnapshot;
+        Long snapshotId = coreOptions().scanSnapshotId();
+        if (snapshotId == null) {
+            snapshotId = snapshotManager().latestSnapshotId();
+        }
+
+        if (snapshotId != null && snapshotManager().snapshotExists(snapshotId)) {
+            latestSnapshot = snapshotManager().snapshot(snapshotId);
+        } else {
+            latestSnapshot = snapshotManager().latestSnapshot();
+        }
+
         if (latestSnapshot != null) {
             return store().newStatsFileHandler().readStats(latestSnapshot);
         }

paimon-core/src/main/java/org/apache/paimon/table/system/StatisticTable.java

@@ -65,7 +65,7 @@ public class StatisticTable implements ReadonlyTable {
 
     public static final String STATISTICS = "statistics";
 
-    public static final RowType TABLE_TYPE =
+    private static final RowType TABLE_TYPE =
             new RowType(
                     Arrays.asList(
                             new DataField(0, "snapshot_id", new BigIntType(false)),

paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala

@@ -73,6 +73,50 @@ abstract class AnalyzeTableTestBase extends PaimonSparkTestBase {
       Row(2, 0, 2, "{ }"))
   }
 
+  test("Paimon analyze: test statistic system table with snapshot") {
+    spark.sql(s"""
+                 |CREATE TABLE T (id STRING, name STRING, i INT, l LONG)
+                 |USING PAIMON
+                 |TBLPROPERTIES ('primary-key'='id')
+                 |""".stripMargin)
+
+    spark.sql(s"INSERT INTO T VALUES ('1', 'a', 1, 1)")
+    spark.sql(s"INSERT INTO T VALUES ('2', 'aaa', 1, 2)")
+    Assertions.assertEquals(0, spark.sql("select * from `T$statistics`").count())
+
+    spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
+
+    spark.sql(s"INSERT INTO T VALUES ('3', 'b', 2, 1)")
+    spark.sql(s"INSERT INTO T VALUES ('4', 'bbb', 3, 2)")
+
+    spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
+
+    withSQLConf("spark.paimon.scan.snapshot-id" -> "3") {
+      checkAnswer(
+        sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM `T$statistics`"),
+        Row(2, 0, 2, "{ }"))
+    }
+
+    withSQLConf("spark.paimon.scan.snapshot-id" -> "4") {
+      checkAnswer(
+        sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM `T$statistics`"),
+        Row(2, 0, 2, "{ }"))
+    }
+
+    withSQLConf("spark.paimon.scan.snapshot-id" -> "6") {
+      checkAnswer(
+        sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM `T$statistics`"),
+        Row(5, 0, 4, "{ }"))
+    }
+
+    withSQLConf("spark.paimon.scan.snapshot-id" -> "100") {
+      checkAnswer(
+        sql("SELECT snapshot_id, schema_id, mergedRecordCount, colstat FROM `T$statistics`"),
+        Row(5, 0, 4, "{ }"))
+    }
+
+  }
+
   test("Paimon analyze: analyze table without snapshot") {
     spark.sql(s"CREATE TABLE T (id STRING, name STRING)")
     spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")