[core][flink] Preserve file stats during primary-key compaction

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

@@ -760,7 +760,8 @@ public FileIO fileIO() {
     }
 
     private ManifestEntryChanges collectChanges(List<CommitMessage> commitMessages) {
-        ManifestEntryChanges changes = new ManifestEntryChanges(options.bucket());
+        ManifestEntryChanges changes =
+                new ManifestEntryChanges(options.bucket(), options.manifestDeleteFileDropStats());
         commitMessages.forEach(changes::collect);
         LOG.info("Finished collecting changes, including: {}", changes);
         return changes;

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

@@ -69,11 +69,6 @@ private FileSystemWriteRestore(
         this.scan = scan;
         this.indexFileHandler = indexFileHandler;
         this.snapshotId = snapshotId;
-        if (options.manifestDeleteFileDropStats()) {
-            if (this.scan != null) {
-                this.scan.dropStats();
-            }
-        }
     }
 
     @Override

paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java

@@ -61,9 +61,6 @@ public CommitScanner(
         this.indexManifestFile = indexManifestFile;
         // Stats in DELETE Manifest Entries is useless
         this.dropStats = options.manifestDeleteFileDropStats();
-        if (dropStats) {
-            this.scan.dropStats();
-        }
     }
 
     public List<SimpleFileEntry> readIncrementalChanges(
@@ -107,9 +104,6 @@ public Map<BinaryRow, Integer> readTotalBuckets(
             Set<BinaryRow> remainingPartitions = new HashSet<>(changedPartitions);
             Map<BinaryRow, Integer> totalBuckets = new HashMap<>();
             FileStoreScan freshScan = scanSupplier.get();
-            if (dropStats) {
-                freshScan.dropStats();
-            }
             Iterator<ManifestEntry> iterator =
                     freshScan
                             .withSnapshot(snapshot)

paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java

@@ -38,6 +38,7 @@
 public class ManifestEntryChanges {
 
     private final int defaultNumBucket;
+    private final boolean dropDeleteStats;
 
     public List<ManifestEntry> appendTableFiles;
     public List<ManifestEntry> appendChangelog;
@@ -47,7 +48,12 @@ public class ManifestEntryChanges {
     public List<IndexManifestEntry> compactIndexFiles;
 
     public ManifestEntryChanges(int defaultNumBucket) {
+        this(defaultNumBucket, false);
+    }
+
+    public ManifestEntryChanges(int defaultNumBucket, boolean dropDeleteStats) {
         this.defaultNumBucket = defaultNumBucket;
+        this.dropDeleteStats = dropDeleteStats;
         this.appendTableFiles = new ArrayList<>();
         this.appendChangelog = new ArrayList<>();
         this.appendIndexFiles = new ArrayList<>();
@@ -135,6 +141,10 @@ private ManifestEntry makeEntry(FileKind kind, CommitMessage commitMessage, Data
             totalBuckets = defaultNumBucket;
         }
 
+        if (kind == FileKind.DELETE && dropDeleteStats) {
+            file = file.copyWithoutStats();
+        }
+
         return ManifestEntry.create(
                 kind, commitMessage.partition(), commitMessage.bucket(), totalBuckets, file);
     }

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

@@ -2060,7 +2060,8 @@ private ManifestCommittable deleteIndexCommittable(
 
     private static List<ManifestEntry> tableFilesFrom(
             ManifestCommittable committable, CoreOptions options) {
-        ManifestEntryChanges changes = new ManifestEntryChanges(options.bucket());
+        ManifestEntryChanges changes =
+                new ManifestEntryChanges(options.bucket(), options.manifestDeleteFileDropStats());
         committable.fileCommittables().forEach(changes::collect);
         return new ArrayList<>(changes.appendTableFiles);
     }

paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java

@@ -19,13 +19,17 @@
 package org.apache.paimon.table.sink;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.BinaryRowWriter;
 import org.apache.paimon.data.GenericRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.disk.IOManagerImpl;
 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.manifest.ManifestFileMeta;
 import org.apache.paimon.operation.AbstractFileStoreWrite;
 import org.apache.paimon.options.MemorySize;
 import org.apache.paimon.options.Options;
@@ -61,6 +65,7 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Predicate;
 
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
@@ -264,6 +269,55 @@ public void testUpgradeToMaxLevel() throws Exception {
         assertThat(streamingRead(scan, read, latestSnapshotId)).hasSize(2);
     }
 
+    @Test
+    public void testDropStatsOnlyForDeleteManifestEntries() throws Exception {
+        Options conf = new Options();
+        conf.set(CoreOptions.BUCKET, 1);
+        conf.set(CoreOptions.MANIFEST_DELETE_FILE_DROP_STATS, true);
+
+        FileStoreTable table = createFileStoreTable(conf);
+        TableWriteImpl<?> write =
+                table.newWrite(commitUser).withIOManager(new IOManagerImpl(tempDir.toString()));
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(GenericRow.of(1, 1, 10L));
+        write.write(GenericRow.of(1, 2, 20L));
+        commit.commit(0, write.prepareCommit(false, 0));
+        write.close();
+        commit.close();
+
+        write = table.newWrite(commitUser).withIOManager(new IOManagerImpl(tempDir.toString()));
+        commit = table.newCommit(commitUser);
+        write.compact(partition(1), 0, true);
+        commit.commit(1, write.prepareCommit(true, 1));
+        write.close();
+        commit.close();
+
+        Snapshot snapshot = table.snapshotManager().latestSnapshot();
+        List<ManifestFileMeta> manifests =
+                table.manifestListReader().read(snapshot.deltaManifestList());
+        assertThat(manifests).hasSize(1);
+        List<ManifestEntry> entries = table.manifestFileReader().read(manifests.get(0).fileName());
+        assertThat(entries).hasSize(2);
+
+        ManifestEntry addEntry = null;
+        ManifestEntry deleteEntry = null;
+        for (ManifestEntry entry : entries) {
+            if (entry.kind() == FileKind.ADD) {
+                addEntry = entry;
+            } else if (entry.kind() == FileKind.DELETE) {
+                deleteEntry = entry;
+            }
+        }
+
+        assertThat(addEntry).isNotNull();
+        assertThat(deleteEntry).isNotNull();
+        assertThat(addEntry.file().fileName()).isEqualTo(deleteEntry.file().fileName());
+        assertThat(addEntry.file().level()).isGreaterThan(deleteEntry.file().level());
+        assertThat(addEntry.file().valueStats()).isNotEqualTo(EMPTY_STATS);
+        assertThat(deleteEntry.file().valueStats()).isEqualTo(EMPTY_STATS);
+    }
+
     @Test
     public void testWaitAllSnapshotsOfSpecificIdentifier() throws Exception {
         Options conf = new Options();

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java

@@ -69,9 +69,6 @@ public TableWriteCoordinator(FileStoreTable table) {
         checkNotNull(table.getManifestCache());
         this.latestCommittedIdentifiers = new ConcurrentHashMap<>();
         this.scan = table.store().newScan();
-        if (table.coreOptions().manifestDeleteFileDropStats()) {
-            scan.dropStats();
-        }
         this.indexFileHandler = table.store().newIndexFileHandler();
         this.pageSize =
                 (int)
@@ -107,9 +104,6 @@ private synchronized void refresh() {
             // used so the shared request `scan`'s bucket/partition state never narrows the
             // warm-up.
             FileStoreScan prefetchScan = table.store().newScan().withSnapshot(snapshot);
-            if (table.coreOptions().manifestDeleteFileDropStats()) {
-                prefetchScan.dropStats();
-            }
             prefetchScan.plan();
         }
     }

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

@@ -100,9 +100,6 @@ public CompactorSourceBuilder withPartitionIdleTime(@Nullable Duration partition
         if (partitionPredicate != null) {
             readBuilder.withPartitionFilter(partitionPredicate);
         }
-        if (CoreOptions.fromMap(table.options()).manifestDeleteFileDropStats()) {
-            readBuilder = readBuilder.dropStats();
-        }
         if (isContinuous) {
             return new ContinuousFileStoreSource(readBuilder, compactBucketsTable.options(), null);
         } else {

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java

@@ -55,6 +55,7 @@
 import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_COORDINATOR_CACHE_EXPIRE_AFTER_ACCESS;
 import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_COORDINATOR_CACHE_MEMORY;
 import static org.apache.paimon.flink.FlinkConnectorOptions.SINK_WRITER_COORDINATOR_CACHE_SOFT_VALUES;
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
 import static org.apache.paimon.utils.SerializationUtils.serializeBinaryRow;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -130,6 +131,31 @@ public void testScanVectorIndexPayloads() throws Exception {
         assertThat(scan.extractVectorIndexPayloads()).containsExactly(ann);
     }
 
+    @Test
+    public void testScanPreservesStatsWhenDeleteManifestStatsAreDropped() throws Exception {
+        Identifier identifier = new Identifier("db", "table");
+        Schema schema =
+                Schema.newBuilder()
+                        .column("k", DataTypes.INT())
+                        .column("v", DataTypes.INT())
+                        .primaryKey("k")
+                        .option(CoreOptions.BUCKET.key(), "1")
+                        .option(CoreOptions.MANIFEST_DELETE_FILE_DROP_STATS.key(), "true")
+                        .build();
+        catalog.createDatabase("db", false);
+        catalog.createTable(identifier, schema, false);
+        FileStoreTable table = getTable(identifier);
+        write(table, GenericRow.of(1, 10));
+
+        TableWriteCoordinator coordinator = new TableWriteCoordinator(table);
+        ScanCoordinationRequest request =
+                new ScanCoordinationRequest(serializeBinaryRow(EMPTY_ROW), 0, false, false, false);
+        ScanCoordinationResponse scan = coordinator.scan(request);
+
+        assertThat(scan.extractDataFiles()).hasSize(1);
+        assertThat(scan.extractDataFiles().get(0).valueStats()).isNotEqualTo(EMPTY_STATS);
+    }
+
     @Test
     public void testPrefetchManifestsWarmsCache() throws Exception {
         Identifier identifier = new Identifier("db", "table");

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java

@@ -27,6 +27,7 @@
 import org.apache.paimon.flink.util.AbstractTestBase;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.io.DataFileMeta;
 import org.apache.paimon.io.DataFileMetaSerializer;
 import org.apache.paimon.partition.PartitionPredicate;
 import org.apache.paimon.schema.Schema;
@@ -65,6 +66,7 @@
 import java.util.Map;
 import java.util.UUID;
 
+import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
 import static org.apache.paimon.utils.SerializationUtils.deserializeBinaryRow;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -210,6 +212,39 @@ public void testStreamingRead(boolean defaultOptions) throws Exception {
         it.close();
     }
 
+    @Test
+    public void testStreamingReadPreservesStatsWhenDeleteManifestStatsAreDropped()
+            throws Exception {
+        FileStoreTable table =
+                createFileStoreTable()
+                        .copy(
+                                Collections.singletonMap(
+                                        CoreOptions.MANIFEST_DELETE_FILE_DROP_STATS.key(), "true"));
+        StreamWriteBuilder streamWriteBuilder =
+                table.newStreamWriteBuilder().withCommitUser(commitUser);
+        StreamTableWrite write = streamWriteBuilder.newWrite();
+        StreamTableCommit commit = streamWriteBuilder.newCommit();
+        write.write(rowData(1, 1510, BinaryString.fromString("20221208"), 15));
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        StreamExecutionEnvironment env =
+                streamExecutionEnvironmentBuilder().streamingMode().build();
+        DataStreamSource<RowData> compactorSource =
+                new CompactorSourceBuilder("test", table)
+                        .withContinuousMode(true)
+                        .withEnv(env)
+                        .build();
+        CloseableIterator<RowData> it = compactorSource.executeAndCollect();
+
+        List<DataFileMeta> files = dataFileMetaSerializer.deserializeList(it.next().getBinary(3));
+        assertThat(files).hasSize(1);
+        assertThat(files.get(0).valueStats()).isNotEqualTo(EMPTY_STATS);
+
+        write.close();
+        commit.close();
+        it.close();
+    }
+
     @Test
     public void testStreamingPartitionSpec() throws Exception {
         testPartitionSpec(