[core] Fix total- metrics

paimon-core/src/main/java/org/apache/paimon/AbstractFileStore.java

@@ -167,7 +167,7 @@ public FileStoreCommitImpl newCommit(String commitUser) {
                 options.manifestMergeMinCount(),
                 partitionType.getFieldCount() > 0 && options.dynamicPartitionOverwrite(),
                 newKeyComparator(),
-                new CommitMetrics(pathFactory()));
+                new CommitMetrics(pathFactory(), fileIO));
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/metrics/commit/CommitMetrics.java

@@ -19,29 +19,82 @@
 package org.apache.paimon.metrics.commit;
 
 import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.FileStatus;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.io.DataFilePathFactory;
 import org.apache.paimon.metrics.Counter;
 import org.apache.paimon.metrics.DescriptiveStatisticsHistogram;
 import org.apache.paimon.metrics.Histogram;
 import org.apache.paimon.metrics.MetricGroup;
 import org.apache.paimon.metrics.groups.GenericMetricGroup;
 import org.apache.paimon.utils.FileStorePathFactory;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
 /** Metrics to measure a commit. */
 public class CommitMetrics {
+    private static Logger log = LoggerFactory.getLogger(CommitMetrics.class);
     private static final int HISTOGRAM_WINDOW_SIZE = 10_000;
     protected static final String GROUP_NAME = "commit";
 
     private final MetricGroup genericMetricGroup;
     private final FileStorePathFactory pathFactory;
 
-    public CommitMetrics(FileStorePathFactory pathFactory) {
+    private long initTableFilesCount = 0;
+    private long initChangelogFilesCount = 0;
+    public CommitMetrics(FileStorePathFactory pathFactory, FileIO fileIO) {
         this.pathFactory = pathFactory;
         this.genericMetricGroup =
                 GenericMetricGroup.createGenericMetricGroup(
                         pathFactory.root().getName(), GROUP_NAME);
+        initDataFilesCount(pathFactory, fileIO);
         registerGenericCommitMetrics();
     }
 
+    private void initDataFilesCount(FileStorePathFactory pathFactory, FileIO fileIO) {
+        try {
+            List<Path> dirs = Arrays.stream(fileIO.listStatus(pathFactory.root())).map(f -> f.getPath()).collect(Collectors.toList());
+            boolean hasPartition = true;
+            for (Path dir : dirs) {
+                if (dir.getName().startsWith("bucket-")) {
+                    hasPartition = false;
+                    break;
+                }
+            }
+            if (hasPartition) {
+                List<Path> buckets = new ArrayList<>();
+                for (Path dir : dirs) {
+                    FileStatus[] fileStatuses = fileIO.listStatus(dir);
+                    buckets.addAll(Arrays.stream(fileStatuses).filter(f -> f.isDir() && f.getPath().getName().startsWith("bucket-")).map(f -> f.getPath()).collect(Collectors.toList()));
+                }
+                for (Path bucket : buckets) {
+                    FileStatus[] fileStatuses = fileIO.listStatus(bucket);
+                    accFilesCount(fileStatuses);
+                }
+            } else {
+                for (Path dir : dirs) {
+                    FileStatus[] fileStatuses = fileIO.listStatus(dir);
+                    accFilesCount(fileStatuses);
+                }
+            }
+        } catch (IOException ie) {
+            log.warn("List table files failed, the 'total' prefixed commit metrics will calculate the number of total files since the job started. ");
+        }
+    }
+
+    private void accFilesCount(FileStatus[] fileStatuses) {
+        initTableFilesCount += Arrays.stream(fileStatuses).filter(f -> f.getPath().getName().startsWith(DataFilePathFactory.DATA_FILE_PREFIX)).count();
+        initChangelogFilesCount += Arrays.stream(fileStatuses).filter(f -> f.getPath().getName().startsWith(DataFilePathFactory.CHANGELOG_FILE_PREFIX)).count();
+    }
+
     public MetricGroup getGenericMetricGroup() {
         return genericMetricGroup;
     }
@@ -166,7 +219,9 @@ private void registerGenericCommitMetrics() {
                     return latestCommit == null ? 0L : latestCommit.getChangelogRecordsCompacted();
                 });
         totalTableFilesCounter = genericMetricGroup.counter(TOTAL_TABLE_FILES);
+        totalTableFilesCounter.inc(initTableFilesCount);
         totalChangelogFilesCounter = genericMetricGroup.counter(TOTAL_CHANGELOG_FILES);
+        totalChangelogFilesCounter.inc(initChangelogFilesCount);
     }
 
     public void reportCommit(CommitStats commitStats) {

paimon-core/src/test/java/org/apache/paimon/metrics/commit/CommitMetricsTest.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.local.LocalFileIO;
 import org.apache.paimon.manifest.FileKind;
 import org.apache.paimon.manifest.ManifestEntry;
 import org.apache.paimon.metrics.Counter;
@@ -300,6 +301,6 @@ private CommitMetrics getCommitMetrics() {
                         RowType.of(new IntType()),
                         "default",
                         CoreOptions.FILE_FORMAT.defaultValue().toString());
-        return new CommitMetrics(pathFactory);
+        return new CommitMetrics(pathFactory, LocalFileIO.create());
     }
 }