This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 94c698c656 [core] Derive partition layout from active files (#10052)
94c698c656 is described below
commit 94c698c656599802a4fa36926de61195ac4e27e1
Author: dwangatt <[email protected]>
AuthorDate: Thu Sep 24 19:37:29 2026 +1000
[core] Derive partition layout from active files (#10052)
---
.../paimon/operation/AbstractFileStoreScan.java | 38 ++++++-
.../FileStoreScanPartitionBucketEntryTest.java | 32 ++++++
.../paimon/table/PartitionLayoutScanTest.java | 124 +++++++++++++++++++++
3 files changed, 189 insertions(+), 5 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index e40d27e789..d5aaa35766 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -56,6 +56,7 @@ import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -368,13 +369,40 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
@Override
public List<PartitionEntry> readPartitionEntries() {
List<ManifestFileMeta> manifests = readManifests().filteredManifests;
+ Set<Identifier> deletedEntries =
+ FileEntry.readDeletedEntries(
+ manifest ->
+ readManifest(
+ manifest,
+ SimpleFileEntry::from,
+ FileEntry.deletedFilter(),
+ null),
+ manifests,
+ parallelism);
+
Map<BinaryRow, PartitionEntry> partitions = new ConcurrentHashMap<>();
Consumer<ManifestFileMeta> processor =
- m ->
- PartitionEntry.merge(
- readManifest(m,
PartitionEntry::fromManifestEntry, null, null),
- partitions);
- randomlyOnlyExecute(getExecutorService(parallelism), processor,
manifests);
+ manifest -> {
+ Map<BinaryRow, PartitionEntry> entries = new HashMap<>();
+ for (ManifestEntry manifestEntry :
+ readManifest(
+ manifest,
+ Function.identity(),
+ FileEntry.addFilter(),
+ entry ->
!deletedEntries.contains(entry.identifier()))) {
+ PartitionEntry entry =
PartitionEntry.fromManifestEntry(manifestEntry);
+ entries.compute(
+ entry.partition(),
+ (partition, old) -> old == null ? entry :
old.merge(entry));
+ }
+ PartitionEntry.merge(entries.values(), partitions);
+ };
+ randomlyOnlyExecute(
+ getExecutorService(parallelism),
+ processor,
+ manifests.stream()
+ .filter(manifest -> manifest.numAddedFiles() > 0)
+ .collect(Collectors.toList()));
return partitions.values().stream()
.filter(p -> p.fileCount() > 0)
.collect(Collectors.toList());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java
index 8ff5023b16..b91464f9e6 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreScanPartitionBucketEntryTest.java
@@ -18,12 +18,18 @@
package org.apache.paimon.operation;
+import org.apache.paimon.CoreOptions;
import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.fs.Path;
import org.apache.paimon.manifest.BucketEntry;
import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
import org.apache.paimon.table.sink.CommitMessage;
+import org.apache.paimon.table.sink.PostponeFixedBucketWriteBuilder;
import org.apache.paimon.table.sink.TableCommitImpl;
+import org.apache.paimon.table.sink.TableWriteImpl;
import org.apache.paimon.table.source.snapshot.ScannerTestBase;
import org.apache.paimon.utils.Pair;
@@ -157,6 +163,32 @@ public class FileStoreScanPartitionBucketEntryTest extends
ScannerTestBase {
assertThat(entries.get(0).recordCount()).isEqualTo(5);
}
+ @Test
+ public void testReadPartitionEntriesIgnoresDeletedBucketLayout() throws
Exception {
+ Options options = new Options();
+ options.set(CoreOptions.BUCKET, -2);
+ table = createFileStoreTable(true, options, new Path(tablePath,
"postpone-bucket-table"));
+ snapshotReader = table.newSnapshotReader();
+
+ PostponeFixedBucketWriteBuilder builder =
table.newPostponeFixedBucketWriteBuilder();
+ try (TableWriteImpl<?> write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.writeAndReturn(rowData(1, 1, 1L), 0, 4);
+ commit.commit(write.prepareCommit());
+ }
+
+ builder =
table.newPostponeFixedBucketWriteBuilder().withOverwrite(Collections.emptyMap());
+ try (TableWriteImpl<?> write = builder.newWrite();
+ BatchTableCommit commit = builder.newCommit()) {
+ write.writeAndReturn(rowData(1, 2, 2L), 0, 2);
+ commit.commit(write.prepareCommit());
+ }
+
+ assertThat(snapshotReader.partitionEntries())
+ .extracting(PartitionEntry::totalBuckets)
+ .containsExactly(2);
+ }
+
@Test
public void testReadPartitionEntriesAppendOnlyTable() throws Exception {
// Create an append-only table (no primary keys)
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/PartitionLayoutScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/PartitionLayoutScanTest.java
new file mode 100644
index 0000000000..5cfa170f2b
--- /dev/null
+++
b/paimon-core/src/test/java/org/apache/paimon/table/PartitionLayoutScanTest.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.table;
+
+import org.apache.paimon.FileStore;
+import org.apache.paimon.Snapshot;
+import org.apache.paimon.manifest.FileKind;
+import org.apache.paimon.manifest.ManifestEntry;
+import org.apache.paimon.manifest.ManifestFileMeta;
+import org.apache.paimon.manifest.PartitionEntry;
+import org.apache.paimon.manifest.PojoManifestEntry;
+import org.apache.paimon.utils.Pair;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests partition-layout metadata derived while scanning manifests. */
+public class PartitionLayoutScanTest extends TableTestBase {
+
+ @Test
+ public void testPartitionLayoutIgnoresDeletedFileInSameManifest() throws
Exception {
+ createTableDefault();
+ FileStoreTable table = getTableDefault();
+ writeDataDefault(Collections.singletonList(dataDefault(1, 1)));
+ writeDataDefault(Collections.singletonList(dataDefault(2, 2)));
+
+ FileStore<?> store = table.store();
+ List<ManifestEntry> existingEntries =
+ store.manifestListFactory().create()
+
.readDataManifests(store.snapshotManager().latestSnapshot()).stream()
+ .flatMap(
+ manifest ->
+ store.manifestFileFactory().create()
+ .read(manifest.fileName(),
manifest.fileSize())
+ .stream())
+ .filter(entry -> entry.kind() == FileKind.ADD)
+ .collect(Collectors.toList());
+
+ ManifestEntry oldLayoutFile = existingEntries.get(0);
+ ManifestEntry liveLayoutFile = existingEntries.get(1);
+ List<ManifestFileMeta> manifests =
+ store.manifestFileFactory()
+ .create()
+ .write(
+ Arrays.asList(
+ new PojoManifestEntry(
+ FileKind.ADD,
+ oldLayoutFile.partition(),
+ oldLayoutFile.bucket(),
+ 4,
+ oldLayoutFile.file()),
+ new PojoManifestEntry(
+ FileKind.ADD,
+ liveLayoutFile.partition(),
+ liveLayoutFile.bucket(),
+ 2,
+ liveLayoutFile.file()),
+ new PojoManifestEntry(
+ FileKind.DELETE,
+ oldLayoutFile.partition(),
+ oldLayoutFile.bucket(),
+ 4,
+ oldLayoutFile.file())));
+
+ Pair<String, Long> baseManifestList =
store.manifestListFactory().create().write(manifests);
+ Pair<String, Long> emptyManifestList =
+
store.manifestListFactory().create().write(Collections.emptyList());
+ long snapshotId = store.snapshotManager().latestSnapshotId() + 1;
+ Snapshot snapshot =
+ new Snapshot(
+ snapshotId,
+ table.schema().id(),
+ baseManifestList.getKey(),
+ baseManifestList.getValue(),
+ emptyManifestList.getKey(),
+ emptyManifestList.getValue(),
+ null,
+ null,
+ null,
+ commitUser,
+ null,
+ snapshotId,
+ Snapshot.CommitKind.OVERWRITE,
+ System.currentTimeMillis(),
+ liveLayoutFile.file().rowCount(),
+ 0,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
+ store.snapshotManager()
+ .fileIO()
+ .tryToWriteAtomic(
+ store.snapshotManager().snapshotPath(snapshotId),
snapshot.toJson());
+
+
assertThat(table.newSnapshotReader().withSnapshot(snapshotId).partitionEntries())
+ .extracting(PartitionEntry::totalBuckets)
+ .containsExactly(2);
+ }
+}
