diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java index 88d6083dec0d..fd322687ca1f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java @@ -760,7 +760,8 @@ public FileIO fileIO() { } private ManifestEntryChanges collectChanges(List 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; diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java index 740e7399f087..a0224bf6e325 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/FileSystemWriteRestore.java +++ b/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 diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java index e63e76590d78..2045b538e8c3 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/CommitScanner.java +++ b/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 readIncrementalChanges( @@ -107,9 +104,6 @@ public Map readTotalBuckets( Set remainingPartitions = new HashSet<>(changedPartitions); Map totalBuckets = new HashMap<>(); FileStoreScan freshScan = scanSupplier.get(); - if (dropStats) { - freshScan.dropStats(); - } Iterator iterator = freshScan .withSnapshot(snapshot) diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java b/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java index faa65bba4750..28887dafd4c8 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/commit/ManifestEntryChanges.java +++ b/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 appendTableFiles; public List appendChangelog; @@ -47,7 +48,12 @@ public class ManifestEntryChanges { public List 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); } diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java index f326be18dcab..d82a1c652507 100644 --- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java @@ -2060,7 +2060,8 @@ private ManifestCommittable deleteIndexCommittable( private static List 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); } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java index 659c010d529a..474bb42886b1 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/sink/TableWriteTest.java @@ -19,6 +19,7 @@ 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; @@ -26,6 +27,9 @@ 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 manifests = + table.manifestListReader().read(snapshot.deltaManifestList()); + assertThat(manifests).hasSize(1); + List 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(); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java index ac7d5475ee80..d0f37d0b8453 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinator.java +++ b/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(); } } diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java index 96961272f12a..cf8176069144 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java +++ b/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 { diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java index 87b6bbca6cf7..ead539442f9d 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/sink/coordinator/TableWriteCoordinatorTest.java +++ b/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"); diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java index 8f7686dbff9f..de74b1248bd9 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/CompactorSourceITCase.java +++ b/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 compactorSource = + new CompactorSourceBuilder("test", table) + .withContinuousMode(true) + .withEnv(env) + .build(); + CloseableIterator it = compactorSource.executeAndCollect(); + + List 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(