[core] Drop stats for deleted data files to reduce memory

paimon-core/src/main/java/org/apache/paimon/append/UnawareAppendTableCompactionCoordinator.java

@@ -380,6 +380,8 @@ public FilesIterator(
             if (filter != null) {
                 snapshotReader.withFilter(filter);
             }
+            // drop stats to reduce memory
+            snapshotReader.dropStats();
             this.streamingMode = isStreaming;
         }
 

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

@@ -101,7 +101,8 @@ protected AbstractFileStoreWrite(
             int writerNumberMax,
             boolean legacyPartitionName) {
         this.snapshotManager = snapshotManager;
-        this.scan = scan;
+        // Statistic is useless in writer
+        this.scan = scan == null ? null : scan.dropStats();
         this.indexFactory = indexFactory;
         this.dvMaintainerFactory = dvMaintainerFactory;
         this.totalBuckets = totalBuckets;

paimon-core/src/main/java/org/apache/paimon/stats/SimpleStatsEvolution.java

@@ -22,6 +22,8 @@
 import org.apache.paimon.casting.CastedRow;
 import org.apache.paimon.data.BinaryString;
 import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalArray;
 import org.apache.paimon.data.InternalMap;
 import org.apache.paimon.data.InternalRow;
@@ -33,9 +35,9 @@
 
 import javax.annotation.Nullable;
 
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 /** Converter for array of {@link SimpleColStats}. */
 public class SimpleStatsEvolution {
@@ -46,14 +48,19 @@ public class SimpleStatsEvolution {
 
     private final Map<List<String>, int[]> indexMappings;
 
+    private final GenericRow emptyValues;
+    private final GenericArray emptyNullCounts;
+
     public SimpleStatsEvolution(
             RowType rowType,
             @Nullable int[] indexMapping,
             @Nullable CastFieldGetter[] castFieldGetters) {
         this.fieldNames = rowType.getFieldNames();
         this.indexMapping = indexMapping;
         this.castFieldGetters = castFieldGetters;
-        this.indexMappings = new HashMap<>();
+        this.indexMappings = new ConcurrentHashMap<>();
+        this.emptyValues = new GenericRow(fieldNames.size());
+        this.emptyNullCounts = new GenericArray(new Object[fieldNames.size()]);
     }
 
     public Result evolution(
@@ -62,7 +69,12 @@ public Result evolution(
         InternalRow maxValues = stats.maxValues();
         InternalArray nullCounts = stats.nullCounts();
 
-        if (denseFields != null) {
+        if (denseFields != null && denseFields.isEmpty()) {
+            // optimize for empty dense fields
+            minValues = emptyValues;
+            maxValues = emptyValues;
+            nullCounts = emptyNullCounts;
+        } else if (denseFields != null) {
             int[] denseIndexMapping =
                     indexMappings.computeIfAbsent(
                             denseFields,

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/LookupDataTableScan.java

@@ -59,6 +59,7 @@ public LookupDataTableScan(
                 defaultValueAssigner);
         this.startupMode = options.startupMode();
         this.lookupScanMode = lookupScanMode;
+        dropStats();
     }
 
     @Override

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyPartialLookupTable.java

@@ -207,6 +207,7 @@ private LocalQueryExecutor(
 
             this.scan =
                     table.newReadBuilder()
+                            .dropStats()
                             .withFilter(filter)
                             .withBucketFilter(
                                     requireCachedBucketIds == null

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/service/QueryFileMonitor.java

@@ -83,7 +83,7 @@ public void open(OpenContext openContext) throws Exception {
      */
     public void open(Configuration parameters) throws Exception {
         FileMonitorTable monitorTable = new FileMonitorTable((FileStoreTable) table);
-        ReadBuilder readBuilder = monitorTable.newReadBuilder();
+        ReadBuilder readBuilder = monitorTable.newReadBuilder().dropStats();
         this.scan = readBuilder.newStreamScan();
         this.read = readBuilder.newRead();
     }

paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonCommand.scala

@@ -94,7 +94,8 @@ trait PaimonCommand extends WithFileStoreTable with ExpressionHelper with SQLCon
       condition: Expression,
       output: Seq[Attribute]): Seq[DataSplit] = {
     // low level snapshot reader, it can not be affected by 'scan.mode'
-    val snapshotReader = table.newSnapshotReader()
+    // dropStats after filter push down
+    val snapshotReader = table.newSnapshotReader().dropStats()
     if (condition != TrueLiteral) {
       val filter =
         convertConditionToPaimonPredicate(condition, output, rowType, ignoreFailure = true)