set changelog version to 1

apache · Mar 30, 2024 · 69dae82 · 69dae82
1 parent 28f6cee
commit 69dae82
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 6 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/Changelog.java b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
@@ -39,7 +39,7 @@
  */
 public class Changelog extends Snapshot {
 
-    private static final int CURRENT_VERSION = 3;
+    private static final int CURRENT_VERSION = 1;
 
     public Changelog(
             long id,

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
@@ -18,6 +18,7 @@
 
 package org.apache.paimon.operation;
 
+import org.apache.paimon.Changelog;
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.data.BinaryRow;
@@ -386,6 +387,10 @@ private List<ManifestFileMeta> readManifests(Snapshot snapshot) {
             case DELTA:
                 return snapshot.deltaManifests(manifestList);
             case CHANGELOG:
+                if (snapshot instanceof Changelog) {
+                    return snapshot.changelogManifests(manifestList);
+                }
+
                 if (snapshot.version() > Snapshot.TABLE_STORE_02_VERSION) {
                     return snapshot.changelogManifests(manifestList);
                 }

diff --git a/...a/org/apache/paimon/table/source/snapshot/ContinuousFromTimestampStartingScannerTest.java b/...a/org/apache/paimon/table/source/snapshot/ContinuousFromTimestampStartingScannerTest.java
@@ -21,7 +21,6 @@
 import org.apache.paimon.CoreOptions;
 import org.apache.paimon.fs.Path;
 import org.apache.paimon.options.Options;
-import org.apache.paimon.table.ExpireSnapshots;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.sink.StreamTableCommit;
 import org.apache.paimon.table.sink.StreamTableWrite;
@@ -117,6 +116,7 @@ public void testScanFromChangelog() throws Exception {
         Options options = new Options();
         options.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX, 2);
         options.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN, 1);
+        options.set(CoreOptions.CHANGELOG_PRODUCER, CoreOptions.ChangelogProducer.INPUT);
         FileStoreTable table =
                 createFileStoreTable(
                         true,
@@ -128,7 +128,6 @@ public void testScanFromChangelog() throws Exception {
                                         + "/"
                                         + UUID.randomUUID()));
         SnapshotManager snapshotManager = table.snapshotManager();
-        ExpireSnapshots expireSnapshots = table.newExpireSnapshots();
         StreamTableWrite write = table.newWrite(commitUser);
         StreamTableCommit commit = table.newCommit(commitUser);
 
@@ -157,20 +156,20 @@ public void testScanFromChangelog() throws Exception {
 
         ContinuousFromTimestampStartingScanner scanner =
                 new ContinuousFromTimestampStartingScanner(
-                        snapshotManager, snapshotManager.snapshot(3).timeMillis(), false, false);
+                        snapshotManager, snapshotManager.snapshot(3).timeMillis(), true, true);
         StartingScanner.NextSnapshot result =
                 (StartingScanner.NextSnapshot) scanner.scan(snapshotReader);
         assertThat(result.nextSnapshotId()).isEqualTo(3);
         scanner =
                 new ContinuousFromTimestampStartingScanner(
-                        snapshotManager, snapshotManager.snapshot(2).timeMillis(), false, false);
+                        snapshotManager, snapshotManager.snapshot(2).timeMillis(), true, true);
 
         assertThat(((StartingScanner.NextSnapshot) scanner.scan(snapshotReader)).nextSnapshotId())
                 .isEqualTo(2);
 
         scanner =
                 new ContinuousFromTimestampStartingScanner(
-                        snapshotManager, snapshotManager.changelog(1).timeMillis(), false, false);
+                        snapshotManager, snapshotManager.changelog(1).timeMillis(), true, true);
         assertThat(((StartingScanner.NextSnapshot) scanner.scan(snapshotReader)).nextSnapshotId())
                 .isEqualTo(1);