(paimon) branch master updated: [core] Restrict level 0 skip to full reads on deletion vector tables (#8712)

[lzljs3620320]

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 2414e915f2 [core] Restrict level 0 skip to full reads on deletion 
vector tables (#8712)
2414e915f2 is described below

commit 2414e915f22d054f87b2f0223144f4d3a41f5255
Author: Arnav Balyan <60175178+arnavbal...@users.noreply.github.com>
AuthorDate: Sat Jul 18 09:34:56 2026 +0530

    [core] Restrict level 0 skip to full reads on deletion vector tables (#8712)
---
 .../table/source/AbstractBatchTableScan.java       |  7 ++-
 .../apache/paimon/table/IncrementalTableTest.java  | 69 ++++++++++++++++++++++
 2 files changed, 74 insertions(+), 2 deletions(-)

diff --git 
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
 
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
index ef9d289b6b..9e4e77cf42 100644
--- 
a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
+++ 
b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractBatchTableScan.java
@@ -72,9 +72,12 @@ public abstract class AbstractBatchTableScan extends 
AbstractDataTableScan {
         this.hasNext = true;
         this.schemaManager = schemaManager;
         if (!schema.primaryKeys().isEmpty() && options.batchScanSkipLevel0()) {
+            // Incremental scans read the delta or changelog files of 
historical snapshots, which
+            // are always recorded at level 0. Skipping level 0 would drop all 
of their input.
             if (options.toConfiguration()
-                    .get(CoreOptions.BATCH_SCAN_MODE)
-                    .equals(CoreOptions.BatchScanMode.NONE)) {
+                            .get(CoreOptions.BATCH_SCAN_MODE)
+                            .equals(CoreOptions.BatchScanMode.NONE)
+                    && options.startupMode() != 
CoreOptions.StartupMode.INCREMENTAL) {
                 snapshotReader.withLevelFilter(level -> level > 
0).enableValueFilter();
             }
         }
diff --git 
a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java 
b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
index 4c218bd32a..3a85110bec 100644
--- 
a/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
+++ 
b/paimon-core/src/test/java/org/apache/paimon/table/IncrementalTableTest.java
@@ -444,6 +444,75 @@ public class IncrementalTableTest extends TableTestBase {
                 .containsExactly(GenericRow.of(3, 
BinaryString.fromString("c")));
     }
 
+    @Test
+    public void testDeletionVectorTable() throws Exception {
+        Identifier identifier = identifier("T");
+        Schema schema =
+                Schema.newBuilder()
+                        .column("a", DataTypes.INT())
+                        .column("b", DataTypes.INT())
+                        .primaryKey("a")
+                        .option("bucket", "1")
+                        .option("deletion-vectors.enabled", "true")
+                        .build();
+        catalog.createTable(identifier, schema, true);
+        FileStoreTable table = (FileStoreTable) catalog.getTable(identifier);
+
+        write(table, ioManager, GenericRow.of(1, 1), GenericRow.of(2, 1));
+        write(table, ioManager, GenericRow.of(1, 2), GenericRow.of(3, 1));
+
+        long latest = table.snapshotManager().latestSnapshotId();
+        assertThat(read(table, Pair.of(INCREMENTAL_BETWEEN, "0," + latest)))
+                .containsExactlyInAnyOrder(
+                        GenericRow.of(1, 1),
+                        GenericRow.of(2, 1),
+                        GenericRow.of(1, 2),
+                        GenericRow.of(3, 1));
+    }
+
+    @Test
+    public void testDeletionVectorTableWithInputChangelog() throws Exception {
+        Identifier identifier = identifier("T");
+        Schema schema =
+                Schema.newBuilder()
+                        .column("pt", DataTypes.INT())
+                        .column("pk", DataTypes.INT())
+                        .column("col1", DataTypes.INT())
+                        .partitionKeys("pt")
+                        .primaryKey("pk", "pt")
+                        .option("bucket", "1")
+                        .option("changelog-producer", "input")
+                        .option("deletion-vectors.enabled", "true")
+                        .build();
+        catalog.createTable(identifier, schema, true);
+        Table table = catalog.getTable(identifier);
+
+        write(
+                table,
+                ioManager,
+                GenericRow.of(1, 1, 1),
+                GenericRow.of(1, 2, 1),
+                GenericRow.of(1, 3, 1),
+                GenericRow.of(2, 1, 1));
+
+        compact(table, row(1), 0, ioManager, true);
+
+        write(
+                table,
+                ioManager,
+                GenericRow.ofKind(RowKind.DELETE, 1, 1, 1),
+                GenericRow.ofKind(RowKind.DELETE, 1, 2, 1),
+                GenericRow.of(1, 4, 1),
+                GenericRow.of(2, 1, 2));
+
+        assertThat(read(table, Pair.of(INCREMENTAL_BETWEEN, "1,3")))
+                .containsExactlyInAnyOrder(
+                        GenericRow.ofKind(RowKind.DELETE, 1, 1, 1),
+                        GenericRow.ofKind(RowKind.DELETE, 1, 2, 1),
+                        GenericRow.of(1, 4, 1),
+                        GenericRow.of(2, 1, 2));
+    }
+
     @Test
     public void testIncrementalEmptyResult() throws Exception {
         Identifier identifier = identifier("T");