[core] Align incremental scans and version selection with Java

.github/workflows/ci.yml

@@ -262,15 +262,15 @@ jobs:
       - name: Core Lumina Native Build Test
         if: matrix.suite == 'lumina'
         run: >
-          cargo test --locked -p paimon
+          cargo test --locked -p paimon --lib
           table::lumina_index_build_builder::tests::test_execute_writes_lumina_index_manifest
           --features fulltext,vortex
           -- --ignored --exact
 
       - name: DataFusion Lumina Build Query E2E Test
         if: matrix.suite == 'lumina'
         run: >
-          cargo test --locked -p paimon-datafusion
+          cargo test --locked -p paimon-datafusion --test read_tables
           --features vortex
           vector_search_tests::test_lumina_build_then_vector_search_query
           -- --ignored --exact

bindings/python/python/pypaimon_rust/datafusion.pyi

@@ -41,6 +41,7 @@ class Split:
     def __init__(self, state: bytes) -> None: ...
     def row_count(self) -> int: ...
     # Java SplitSerializer v1 binary: a DataSplit (v8), or an IndexedSplit when the split has row ranges.
+    def is_streaming(self) -> bool: ...
     def serialize(self) -> bytes: ...
 
 class Plan:
@@ -81,7 +82,7 @@ class ReadBuilder:
         ...
     def new_scan(self) -> TableScan: ...
     def new_incremental_scan(self, start_snapshot_id: int, end_snapshot_id: int) -> TableScan:
-        """Plan APPEND deltas in (start, end] together, merging primary-key versions across snapshots.
+        """Plan APPEND deltas in (start, end] together, preserving physical change events.
 
         Snapshot IDs are used, not timestamps. The end snapshot must exist.
         Row-position slicing and sharding use the combined delta batch as their position space.

bindings/python/src/read.rs

@@ -135,12 +135,14 @@ impl PyReadBuilder {
         // conflict error via its Java-parity silent fallback, so the strict
         // gate below would otherwise misattribute the failure to a single
         // selector. Surface the real conflict, listing the keys the user set.
+        // scan.version must first be adapted by the core: Java allows it to
+        // overwrite a selector of the same kind after resolving tag precedence.
         let present: Vec<&str> = TIME_TRAVEL_SELECTORS
             .iter()
             .copied()
             .filter(|name| opts.contains_key(*name))
             .collect();
-        if present.len() > 1 {
+        if present.len() > 1 && !opts.contains_key("scan.version") {
             return Err(PyValueError::new_err(format!(
                 "Only one time-travel selector may be set, found: {}",
                 present.join(", ")
@@ -472,9 +474,13 @@ impl PySplit {
         self.inner.row_count()
     }
 
-    /// Serialize this planned split to the Java `SplitSerializer` (v1) binary, so pypaimon (or
-    /// any Paimon reader) can rebuild it without re-planning. A split carrying row ranges is
-    /// serialized as an `IndexedSplit`.
+    /// Whether the split must be read as physical change events.
+    fn is_streaming(&self) -> bool {
+        self.inner.is_streaming()
+    }
+
+    /// Serialize to Java SplitSerializer v1, using IndexedSplit for row ranges.
+    /// Preserves the streaming flag for physical change-event reads.
     fn serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>> {
         let bytes = self.inner.serialize_split_v1().map_err(to_py_err)?;
         Ok(PyBytes::new(py, &bytes))

bindings/python/tests/test_read.py

@@ -694,6 +694,24 @@ def test_time_travel_by_tag_name():
         assert _rows(builder.new_read().read(splits)) == 1
 
 
+def test_scan_version_adapts_before_binding_selector_validation():
+    with tempfile.TemporaryDirectory() as warehouse:
+        ctx = _make_two_snapshot_table(warehouse)
+        ctx.sql("CALL sys.create_tag(table => 'tdb.t', tag => 'v1', snapshot_id => 1)")
+        table = PaimonCatalog({"warehouse": warehouse}).get_table("tdb.t")
+        for version, key in [("1", "scan.snapshot-id"), ("v1", "scan.tag-name")]:
+            builder = table.new_read_builder({"scan.version": version, key: "invalid"})
+            plan = builder.new_scan().plan()
+            assert plan.snapshot_id() == 1
+            assert pa.Table.from_batches(builder.new_read().read(plan.splits())).to_pydict() == {
+                "id": [1], "name": ["a"]}
+        for opts in [
+                {"scan.version": "1", "scan.tag-name": "v1"},
+                {"scan.version": "v1", "scan.snapshot-id": "1"}]:
+            with pytest.raises(ValueError, match="did not resolve"):
+                table.new_read_builder(opts)
+
+
 def test_time_travel_unresolved_snapshot_raises():
     with tempfile.TemporaryDirectory() as warehouse:
         _make_two_snapshot_table(warehouse)
@@ -869,7 +887,7 @@ def test_split_serialize_encodes_deletions_and_external_path():
         assert b"s3://ext/data-0.parquet" in data  # external path
 
 
-def test_combined_incremental_plan_merges_pk_versions_and_preserves_range():
+def test_combined_incremental_plan_preserves_pk_events_and_range():
     with tempfile.TemporaryDirectory() as warehouse:
         ctx = SQLContext()
         ctx.register_catalog("paimon", {"warehouse": warehouse})
@@ -885,9 +903,16 @@ def test_combined_incremental_plan_merges_pk_versions_and_preserves_range():
         assert plan.snapshot_id() == 2
         assert len(plan.splits()) == 1
         assert pa.Table.from_batches(builder.new_read().read(plan.splits())).to_pydict() == {
-            "id": [1], "value": [20]}
+            "id": [1, 1], "value": [10, 20]}
         assert [s.serialize() for s in scan.plan().splits()] == [
             s.serialize() for s in plan.splits()]
+        assert all(s.is_streaming() for s in plan.splits())
+        assert plan.splits()[0].serialize()[-2] == 1  # Java isStreaming flag
+        import pickle
+        restored = pickle.loads(pickle.dumps(plan.splits()[0]))
+        assert restored.is_streaming()
+        assert restored.serialize() == (
+            plan.splits()[0].serialize())
         selected = builder.new_incremental_scan(1, 2).plan()
         assert selected.snapshot_id() == 2
         assert pa.Table.from_batches(builder.new_read().read(selected.splits())).to_pydict() == {
@@ -981,7 +1006,8 @@ def test_incremental_row_positions_use_combined_delta_batch():
             assert plan.snapshot_id() == 2
             restored = [pickle.loads(pickle.dumps(split)) for split in plan.splits()]
             assert pa.Table.from_batches(builder.new_read().read(restored)).column("id").to_pylist() == expected
-            assert [s.serialize() for s in scan.plan().splits()] == [s.serialize() for s in plan.splits()]
+            assert [s.serialize() for s in scan.plan().splits()] == [
+                s.serialize() for s in plan.splits()]
         builder.with_row_ranges([(1, 3)]).with_limit(2)
         plan = builder.new_incremental_scan(0, 2).with_row_position_slice(2, 5).plan()
         assert pa.Table.from_batches(builder.new_read().read(plan.splits())).column("id").to_pylist() == [2, 3]

crates/integration_tests/tests/read_tables.rs

@@ -2957,34 +2957,39 @@ async fn test_time_travel_conflicting_selectors_fail() {
     let catalog = create_file_system_catalog();
     let table = get_table_from_catalog(&catalog, "time_travel_table").await;
 
-    let conflicted = table.copy_with_options(HashMap::from([
-        ("scan.version".to_string(), "snapshot1".to_string()),
-        ("scan.timestamp-millis".to_string(), "1234".to_string()),
-    ]));
-
-    let plan_err = conflicted
-        .new_read_builder()
-        .new_scan()
-        .plan()
-        .await
-        .expect_err("conflicting time-travel selectors should fail");
+    // Java resolves scan.version before validating conflicts with other selectors.
+    for (version, selector) in [
+        ("snapshot1", "scan.tag-name"),
+        ("1", "scan.snapshot-id"),
+        ("watermark-1", "scan.watermark"),
+    ] {
+        let conflicted = table.copy_with_options(HashMap::from([
+            ("scan.version".to_string(), version.to_string()),
+            ("scan.timestamp-millis".to_string(), "1234".to_string()),
+        ]));
 
-    match plan_err {
-        Error::DataInvalid { message, .. } => {
-            assert!(
-                message.contains("Only one time-travel selector may be set"),
-                "unexpected conflict error: {message}"
-            );
-            assert!(
-                message.contains("scan.version"),
-                "conflict error should mention scan.version: {message}"
-            );
-            assert!(
-                message.contains("scan.timestamp-millis"),
-                "conflict error should mention scan.timestamp-millis: {message}"
-            );
+        let plan_err = conflicted
+            .new_read_builder()
+            .new_scan()
+            .plan()
+            .await
+            .expect_err("conflicting time-travel selectors should fail");
+
+        match plan_err {
+            Error::DataInvalid { message, .. } => {
+                assert!(
+                    message.contains("Only one time-travel selector may be set"),
+                    "unexpected conflict error for version {version}: {message}"
+                );
+                for key in [selector, "scan.timestamp-millis"] {
+                    assert!(
+                        message.contains(key),
+                        "conflict error should mention {key}: {message}"
+                    );
+                }
+            }
+            other => panic!("unexpected error for version {version}: {other:?}"),
         }
-        other => panic!("unexpected error: {other:?}"),
     }
 }
 

crates/integrations/datafusion/tests/read_tables.rs

@@ -1008,38 +1008,45 @@ async fn time_travel_schema_evolution() {
 
 #[tokio::test]
 async fn test_time_travel_conflicting_selectors_fail() {
-    // When both scan.version and scan.timestamp-millis are set on the same
-    // provider, Paimon rejects the combination at scan time.
-    let provider = create_provider_with_options(
-        "time_travel_table",
-        HashMap::from([
-            ("scan.version".to_string(), "1".to_string()),
-            ("scan.timestamp-millis".to_string(), "1234".to_string()),
-        ]),
-    )
-    .await;
+    // Java resolves scan.version before validating conflicts with other selectors.
+    for (version, selector) in [
+        ("snapshot1", "scan.tag-name"),
+        ("1", "scan.snapshot-id"),
+        ("watermark-1", "scan.watermark"),
+    ] {
+        let provider = create_provider_with_options(
+            "time_travel_table",
+            HashMap::from([
+                ("scan.version".to_string(), version.to_string()),
+                ("scan.timestamp-millis".to_string(), "1234".to_string()),
+            ]),
+        )
+        .await;
 
-    let ctx = create_context().await;
-    ctx.register_temp_table("paimon.default.time_travel_table", Arc::new(provider))
-        .expect("Failed to register temp table");
+        let ctx = create_context().await;
+        ctx.register_temp_table("paimon.default.time_travel_table", Arc::new(provider))
+            .expect("Failed to register temp table");
 
-    let err = ctx
-        .sql("SELECT id, name FROM paimon.default.time_travel_table")
-        .await
-        .expect("query should parse")
-        .collect()
-        .await
-        .expect_err("conflicting time-travel selectors should fail");
+        let err = ctx
+            .sql("SELECT id, name FROM paimon.default.time_travel_table")
+            .await
+            .expect("query should parse")
+            .collect()
+            .await
+            .expect_err("conflicting time-travel selectors should fail");
 
-    let message = err.to_string();
-    assert!(
-        message.contains("Only one time-travel selector may be set"),
-        "unexpected conflict error: {message}"
-    );
-    assert!(
-        message.contains("scan.version"),
-        "conflict error should mention scan.version: {message}"
-    );
+        let message = err.to_string();
+        assert!(
+            message.contains("Only one time-travel selector may be set"),
+            "unexpected conflict error for version {version}: {message}"
+        );
+        for key in [selector, "scan.timestamp-millis"] {
+            assert!(
+                message.contains(key),
+                "conflict error should mention {key}: {message}"
+            );
+        }
+    }
 }
 
 #[tokio::test]

crates/paimon/src/table/data_file_reader.rs

@@ -2754,13 +2754,27 @@ mod tests {
             Vec::new(),
         );
         let batches = reader
-            .read(&[split])
+            .clone()
+            .read(std::slice::from_ref(&split))
             .unwrap()
             .try_collect::<Vec<_>>()
             .await
             .unwrap();
 
         assert_eq!(collect_ids(&batches), vec![1, 3]);
+        // Event planners do not attach endpoint DVs, but readers still honor
+        // explicitly supplied DVs in a Java streaming frame.
+        let mut bytes = split.serialize().unwrap();
+        let flag = bytes.len() - 2;
+        bytes[flag] = 1;
+        let streaming = DataSplit::deserialize(&bytes).unwrap();
+        let events = reader
+            .read(&[streaming])
+            .unwrap()
+            .try_collect::<Vec<_>>()
+            .await
+            .unwrap();
+        assert_eq!(collect_ids(&events), vec![1, 3]);
     }
 
     /// A Mosaic file and a Parquet file in the same split must both be read and concatenated.

crates/paimon/src/table/hybrid_search_builder.rs

@@ -539,7 +539,7 @@ impl<'a> HybridSearchBuilder<'a> {
         let core = CoreOptions::new(self.table.schema().options());
         // Already targeting a fixed snapshot (resolved travel copy or a selector
         // that resolves deterministically): every route agrees without pinning.
-        if self.table.has_resolved_travel_snapshot() || core.try_time_travel_selector()?.is_some() {
+        if self.table.has_resolved_travel_snapshot() || core.has_time_travel_selector() {
             return Ok(None);
         }
         // Read-latest: pin the current latest snapshot once so a concurrent commit

crates/paimon/src/table/incremental_scan.rs

@@ -277,13 +277,14 @@ impl<'a> IncrementalScan<'a> {
         }
     }
 
-    /// Plan APPEND deltas as one batch, merging their manifest entries and
-    /// overlapping primary-key files across the whole snapshot range.
+    /// Plan APPEND deltas with batch split packing and streaming read semantics.
+    /// Each physical change is retained, including repeated keys and retracts.
     ///
     /// Unlike [`Self::plan`], this returns an ordinary [`Plan`] for a normal
     /// table reader. It does not preserve a separate result for each commit.
     /// Only Delta (or Auto resolving to Delta) is supported. The end snapshot
-    /// must exist, and supplies the plan's snapshot metadata and deletion vectors.
+    /// must exist and supplies snapshot metadata. Snapshot deletion vectors and
+    /// automatic global-index pruning do not apply to these historical events.
     pub async fn plan_combined_delta(&self) -> crate::Result<Plan> {
         CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
         let mode = self.resolve_mode();

crates/paimon/src/table/sorted_global_index_build_builder/tests.rs

@@ -1133,17 +1133,22 @@ async fn test_empty_global_index_ranges_skip_legacy_manifests() {
         assert!(plan.splits().is_empty());
 
         let (traced_plan, trace) = read_builder.new_scan().plan_with_trace().await.unwrap();
-        let delta_plan = read_builder
+        let delta_error = read_builder
             .new_scan()
             .plan_snapshot_delta(&snapshot)
             .await
-            .unwrap();
+            .unwrap_err();
 
         assert!(traced_plan.splits().is_empty());
         assert_eq!(trace.manifest_entries_read, 0);
         assert_eq!(trace.final_splits, 0);
         assert_eq!(trace.final_files, 0);
-        assert!(delta_plan.splits().is_empty());
+        // A snapshot index may short-circuit a batch read, but cannot hide
+        // an invalid historical event file from incremental planning.
+        assert!(
+            matches!(delta_error, crate::Error::DataInvalid { ref message, .. }
+            if message.contains("First row id"))
+        );
     }
 }