[core] support delete stats in result of scan plan

paimon-core/src/main/java/org/apache/paimon/io/DataFileMeta.java

@@ -409,6 +409,27 @@ public DataFileMeta rename(String newFileName) {
                 valueStatsCols);
     }
 
+    public DataFileMeta withoutStats() {
+        return new DataFileMeta(
+                fileName,
+                fileSize,
+                rowCount,
+                minKey,
+                maxKey,
+                EMPTY_STATS,
+                EMPTY_STATS,
+                minSequenceNumber,
+                maxSequenceNumber,
+                schemaId,
+                level,
+                extraFiles,
+                creationTime,
+                deleteRowCount,
+                embeddedIndex,
+                fileSource,
+                valueStatsCols);
+    }
+
     public List<Path> collectFiles(DataFilePathFactory pathFactory) {
         List<Path> paths = new ArrayList<>();
         paths.add(pathFactory.toPath(fileName));

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestEntry.java

@@ -121,6 +121,10 @@ public Identifier identifier() {
                 file.embeddedIndex());
     }
 
+    public ManifestEntry withoutStats() {
+        return new ManifestEntry(kind, partition, bucket, totalBuckets, file.withoutStats());
+    }
+
     @Override
     public boolean equals(Object o) {
         if (!(o instanceof ManifestEntry)) {

paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java

@@ -90,6 +90,7 @@ public abstract class AbstractFileStoreScan implements FileStoreScan {
 
     private ManifestCacheFilter manifestCacheFilter = null;
     private ScanMetrics scanMetrics = null;
+    private boolean withoutStatsInPlan;
 
     public AbstractFileStoreScan(
             ManifestsReader manifestsReader,
@@ -105,6 +106,7 @@ public AbstractFileStoreScan(
         this.manifestFileFactory = manifestFileFactory;
         this.tableSchemas = new ConcurrentHashMap<>();
         this.parallelism = parallelism;
+        this.withoutStatsInPlan = false;
     }
 
     @Override
@@ -215,6 +217,12 @@ public FileStoreScan withMetrics(ScanMetrics metrics) {
         return this;
     }
 
+    @Override
+    public FileStoreScan withoutStatsInPlan() {
+        this.withoutStatsInPlan = true;
+        return this;
+    }
+
     @Nullable
     @Override
     public Integer parallelism() {
@@ -291,6 +299,11 @@ public Snapshot snapshot() {
 
             @Override
             public List<ManifestEntry> files() {
+                if (withoutStatsInPlan) {
+                    return files.stream()
+                            .map(ManifestEntry::withoutStats)
+                            .collect(Collectors.toList());
+                }
                 return files;
             }
         };

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreScan.java

@@ -81,6 +81,8 @@ public interface FileStoreScan {
 
     FileStoreScan withMetrics(ScanMetrics metrics);
 
+    FileStoreScan withoutStatsInPlan();
+
     @Nullable
     Integer parallelism();
 

paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java

@@ -102,6 +102,12 @@ public AbstractDataTableScan withMetricsRegistry(MetricRegistry metricsRegistry)
         return this;
     }
 
+    @Override
+    public AbstractDataTableScan withoutStatsInPlan() {
+        snapshotReader.withoutStatsInPlan();
+        return this;
+    }
+
     public CoreOptions options() {
         return options;
     }

paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java

@@ -55,4 +55,9 @@ default InnerTableScan withMetricsRegistry(MetricRegistry metricRegistry) {
         // do nothing, should implement this if need
         return this;
     }
+
+    default InnerTableScan withoutStatsInPlan() {
+        // do nothing, should implement this if need
+        return this;
+    }
 }

paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilder.java

@@ -150,6 +150,9 @@ default ReadBuilder withProjection(int[][] projection) {
      */
     ReadBuilder withShard(int indexOfThisSubtask, int numberOfParallelSubtasks);
 
+    // todo zhushi
+    ReadBuilder withoutStatsInPlan();
+
     /** Create a {@link TableScan} to perform batch planning. */
     TableScan newScan();
 

paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java

@@ -51,6 +51,8 @@ public class ReadBuilderImpl implements ReadBuilder {
 
     private @Nullable RowType readType;
 
+    private boolean withoutStatsInPlan = false;
+
     public ReadBuilderImpl(InnerTable table) {
         this.table = table;
     }
@@ -124,6 +126,12 @@ public ReadBuilder withBucketFilter(Filter<Integer> bucketFilter) {
         return this;
     }
 
+    @Override
+    public ReadBuilder withoutStatsInPlan() {
+        this.withoutStatsInPlan = true;
+        return this;
+    }
+
     @Override
     public TableScan newScan() {
         InnerTableScan tableScan = configureScan(table.newScan());
@@ -156,6 +164,9 @@ private InnerTableScan configureScan(InnerTableScan scan) {
         if (bucketFilter != null) {
             scan.withBucketFilter(bucketFilter);
         }
+        if (withoutStatsInPlan) {
+            scan.withoutStatsInPlan();
+        }
         return scan;
     }
 

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java

@@ -85,6 +85,8 @@ public interface SnapshotReader {
 
     SnapshotReader withDataFileNameFilter(Filter<String> fileNameFilter);
 
+    SnapshotReader withoutStatsInPlan();
+
     SnapshotReader withShard(int indexOfThisSubtask, int numberOfParallelSubtasks);
 
     SnapshotReader withMetricRegistry(MetricRegistry registry);

paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java

@@ -264,6 +264,12 @@ public SnapshotReader withDataFileNameFilter(Filter<String> fileNameFilter) {
         return this;
     }
 
+    @Override
+    public SnapshotReader withoutStatsInPlan() {
+        scan.withoutStatsInPlan();
+        return this;
+    }
+
     @Override
     public SnapshotReader withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) {
         if (splitGenerator.alwaysRawConvertible()) {

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

@@ -342,6 +342,12 @@ public SnapshotReader withDataFileNameFilter(Filter<String> fileNameFilter) {
             return this;
         }
 
+        @Override
+        public SnapshotReader withoutStatsInPlan() {
+            wrapped.withoutStatsInPlan();
+            return this;
+        }
+
         @Override
         public SnapshotReader withShard(int indexOfThisSubtask, int numberOfParallelSubtasks) {
             wrapped.withShard(indexOfThisSubtask, numberOfParallelSubtasks);

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

@@ -267,6 +267,7 @@ private TableScan.Plan tablePlan(FileStoreTable storeTable) {
                                     GenericRow.of(null, null, null, null, null, level));
                         });
             }
+
             return scan.plan();
         }
     }

paimon-core/src/test/java/org/apache/paimon/operation/KeyValueFileStoreScanTest.java

@@ -50,6 +50,7 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.stream.Collectors;
 
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
 import static org.assertj.core.api.Assertions.assertThat;
 
 /** Tests for {@link KeyValueFileStoreScan}. */
@@ -274,6 +275,28 @@ public void testWithManifestList() throws Exception {
         runTestExactMatch(scan, null, expected);
     }
 
+    @Test
+    public void testWithoutStatsInPlan() throws Exception {
+        ThreadLocalRandom random = ThreadLocalRandom.current();
+        List<KeyValue> data = generateData(100, 0, (long) Math.abs(random.nextInt(1000)));
+        writeData(data, 0);
+        data = generateData(100, 1, (long) Math.abs(random.nextInt(1000)) + 1000);
+        writeData(data, 0);
+        data = generateData(100, 2, (long) Math.abs(random.nextInt(1000)) + 2000);
+        writeData(data, 0);
+        data = generateData(100, 3, (long) Math.abs(random.nextInt(1000)) + 3000);
+        Snapshot snapshot = writeData(data, 0);
+
+        KeyValueFileStoreScan scan = store.newScan();
+        scan.withSnapshot(snapshot.id());
+        List<ManifestEntry> files = scan.plan().files();
+
+        for (ManifestEntry manifestEntry : files) {
+            assertThat(manifestEntry.file().keyStats()).isEqualTo(EMPTY_STATS);
+            assertThat(manifestEntry.file().valueStats()).isEqualTo(EMPTY_STATS);
+        }
+    }
+
     private void runTestExactMatch(
             FileStoreScan scan, Long expectedSnapshotId, Map<BinaryRow, BinaryRow> expected)
             throws Exception {

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileStoreSource.java

@@ -77,7 +77,7 @@ public SplitEnumerator<FileStoreSourceSplit, PendingSplitsCheckpoint> restoreEnu
             nextSnapshotId = checkpoint.currentSnapshotId();
             splits = checkpoint.splits();
         }
-        StreamTableScan scan = readBuilder.newStreamScan();
+        StreamTableScan scan = readBuilder.withoutStatsInPlan().newStreamScan();
         if (metricGroup(context) != null) {
             ((StreamDataTableScan) scan)
                     .withMetricsRegistry(new FlinkMetricRegistry(context.metricGroup()));

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkTableSource.java

@@ -175,6 +175,7 @@ protected void scanSplitsForInference() {
                 List<PartitionEntry> partitionEntries =
                         table.newReadBuilder()
                                 .withFilter(predicate)
+                                .withoutStatsInPlan()
                                 .newScan()
                                 .listPartitionEntries();
                 long totalSize = 0;
@@ -188,7 +189,12 @@ protected void scanSplitsForInference() {
                         new SplitStatistics((int) (totalSize / splitTargetSize + 1), rowCount);
             } else {
                 List<Split> splits =
-                        table.newReadBuilder().withFilter(predicate).newScan().plan().splits();
+                        table.newReadBuilder()
+                                .withFilter(predicate)
+                                .withoutStatsInPlan()
+                                .newScan()
+                                .plan()
+                                .splits();
                 splitStatistics =
                         new SplitStatistics(
                                 splits.size(), splits.stream().mapToLong(Split::rowCount).sum());

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/StaticFileStoreSource.java

@@ -87,7 +87,7 @@ public SplitEnumerator<FileStoreSourceSplit, PendingSplitsCheckpoint> restoreEnu
 
     private List<FileStoreSourceSplit> getSplits(SplitEnumeratorContext context) {
         FileStoreSourceSplitGenerator splitGenerator = new FileStoreSourceSplitGenerator();
-        TableScan scan = readBuilder.newScan();
+        TableScan scan = readBuilder.withoutStatsInPlan().newScan();
         // register scan metrics
         if (context.metricGroup() != null) {
             ((InnerTableScan) scan)

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorFunction.java

@@ -106,7 +106,7 @@ public MonitorFunction(
 
     @Override
     public void initializeState(FunctionInitializationContext context) throws Exception {
-        this.scan = readBuilder.newStreamScan();
+        this.scan = readBuilder.withoutStatsInPlan().newStreamScan();
 
         this.checkpointState =
                 context.getOperatorStateStore()