[core] Fix FileMonitorTable with Deletion Vectors Enabled

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/mergetree/SortedRun.java

@@ -19,9 +19,9 @@
 package org.apache.paimon.mergetree;
 
 import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.io.DataFileMeta;
-import org.apache.paimon.utils.Preconditions;
 
 import java.util.Collections;
 import java.util.Comparator;
@@ -87,9 +87,13 @@ public long totalSize() {
     @VisibleForTesting
     public void validate(Comparator<InternalRow> comparator) {
         for (int i = 1; i < files.size(); i++) {
-            Preconditions.checkState(
-                    comparator.compare(files.get(i).minKey(), files.get(i - 1).maxKey()) > 0,
-                    "SortedRun is not sorted and may contain overlapping key intervals. This is a bug.");
+            BinaryRow minKey = files.get(i).minKey();
+            BinaryRow maxKey = files.get(i - 1).maxKey();
+            if (comparator.compare(minKey, maxKey) <= 0) {
+                throw new IllegalArgumentException(
+                        "SortedRun is not sorted and may contain overlapping key intervals. "
+                                + "This is a bug.");
+            }
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/table/source/DataTableStreamScan.java

@@ -19,6 +19,7 @@
 package org.apache.paimon.table.source;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.CoreOptions.StreamScanMode;
 import org.apache.paimon.Snapshot;
 import org.apache.paimon.consumer.Consumer;
 import org.apache.paimon.lookup.LookupStrategy;
@@ -48,13 +49,15 @@
 import java.util.List;
 
 import static org.apache.paimon.CoreOptions.ChangelogProducer.FULL_COMPACTION;
+import static org.apache.paimon.CoreOptions.StreamScanMode.FILE_MONITOR;
 
 /** {@link StreamTableScan} implementation for streaming planning. */
 public class DataTableStreamScan extends AbstractDataTableScan implements StreamDataTableScan {
 
     private static final Logger LOG = LoggerFactory.getLogger(DataTableStreamScan.class);
 
     private final CoreOptions options;
+    private final StreamScanMode scanMode;
     private final SnapshotManager snapshotManager;
     private final boolean supportStreamingReadOverwrite;
     private final DefaultValueAssigner defaultValueAssigner;
@@ -80,6 +83,7 @@ public DataTableStreamScan(
             DefaultValueAssigner defaultValueAssigner) {
         super(options, snapshotReader);
         this.options = options;
+        this.scanMode = options.toConfiguration().get(CoreOptions.STREAM_SCAN_MODE);
         this.snapshotManager = snapshotManager;
         this.supportStreamingReadOverwrite = supportStreamingReadOverwrite;
         this.defaultValueAssigner = defaultValueAssigner;
@@ -139,7 +143,9 @@ private void initScanner() {
 
     private Plan tryFirstPlan() {
         StartingScanner.Result result;
-        if (options.needLookup()) {
+        if (scanMode == FILE_MONITOR) {
+            result = startingScanner.scan(snapshotReader);
+        } else if (options.needLookup()) {
             result = startingScanner.scan(snapshotReader.withLevelFilter(level -> level > 0));
             snapshotReader.withLevelFilter(Filter.alwaysTrue());
         } else if (options.changelogProducer().equals(FULL_COMPACTION)) {
@@ -157,7 +163,9 @@ private Plan tryFirstPlan() {
             currentWatermark = scannedResult.currentWatermark();
             long currentSnapshotId = scannedResult.currentSnapshotId();
             LookupStrategy lookupStrategy = options.lookupStrategy();
-            if (!lookupStrategy.produceChangelog && lookupStrategy.deletionVector) {
+            if (scanMode == FILE_MONITOR) {
+                nextSnapshotId = currentSnapshotId + 1;
+            } else if (!lookupStrategy.produceChangelog && lookupStrategy.deletionVector) {
                 // For DELETION_VECTOR_ONLY mode, we need to return the remaining data from level 0
                 // in the subsequent plan.
                 nextSnapshotId = currentSnapshotId;
@@ -250,9 +258,7 @@ protected SnapshotReader.Plan handleOverwriteSnapshot(Snapshot snapshot) {
     }
 
     protected FollowUpScanner createFollowUpScanner() {
-        CoreOptions.StreamScanMode type =
-                options.toConfiguration().get(CoreOptions.STREAM_SCAN_MODE);
-        switch (type) {
+        switch (scanMode) {
             case COMPACT_BUCKET_TABLE:
                 return new DeltaFollowUpScanner();
             case FILE_MONITOR:

paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeySimpleTableTest.java

@@ -1674,7 +1674,7 @@ public void testFullCompactedRead() throws Exception {
     }
 
     @Test
-    public void testInnerStreamScanMode() throws Exception {
+    public void testFileMonitorTableScan() throws Exception {
         FileStoreTable table = createFileStoreTable();
 
         FileMonitorTable monitorTable = new FileMonitorTable(table);
@@ -1751,6 +1751,38 @@ public void testInnerStreamScanMode() throws Exception {
         commit.close();
     }
 
+    @Test
+    public void testFileMonitorTableScanWithDv() throws Exception {
+        FileStoreTable table =
+                createFileStoreTable(options -> options.set(DELETION_VECTORS_ENABLED, true));
+
+        FileMonitorTable monitorTable = new FileMonitorTable(table);
+        ReadBuilder readBuilder = monitorTable.newReadBuilder();
+        StreamTableScan scan = readBuilder.newStreamScan();
+        TableRead read = readBuilder.newRead();
+
+        BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+        BatchTableWrite write = writeBuilder.newWrite();
+        write.withIOManager(IOManager.create(tempDir.toString()));
+        BatchTableCommit commit = writeBuilder.newCommit();
+
+        write.write(rowData(1, 10, 100L));
+        write.write(rowData(1, 11, 101L));
+        commit.commit(write.prepareCommit());
+
+        List<InternalRow> results = new ArrayList<>();
+        read.createReader(scan.plan()).forEachRemaining(results::add);
+        read.createReader(scan.plan()).forEachRemaining(results::add);
+        assertThat(results).hasSize(1);
+        FileMonitorTable.FileChange change = FileMonitorTable.toFileChange(results.get(0));
+        assertThat(change.beforeFiles()).hasSize(0);
+        assertThat(change.dataFiles()).hasSize(1);
+        results.clear();
+
+        write.close();
+        commit.close();
+    }
+
     @Test
     public void testReadOptimizedTable() throws Exception {
         // let max level has many files