feat: support batch incremental diff reads

crates/paimon/src/spec/core_options.rs

@@ -86,6 +86,8 @@ const IGNORE_DELETE_FALLBACK_KEYS: &[&str] = &[
     "deduplicate.ignore-delete",
     "partial-update.ignore-delete",
 ];
+const DIFF_PARALLELISM_OPTION: &str = "diff.parallelism";
+const DEFAULT_DIFF_PARALLELISM: usize = 4;
 const DEFAULT_COMMIT_MAX_RETRIES: u32 = 10;
 const DEFAULT_COMMIT_TIMEOUT_MS: u64 = 120_000;
 const DEFAULT_COMMIT_MIN_RETRY_WAIT_MS: u64 = 1_000;
@@ -469,6 +471,17 @@ impl<'a> CoreOptions<'a> {
             .is_some_and(|v| v.eq_ignore_ascii_case("true"))
     }
 
+    /// Parallelism for batch incremental Diff pair reads (`diff.parallelism`).
+    ///
+    /// Default is 4; values below 1 are clamped to 1.
+    pub fn diff_parallelism(&self) -> usize {
+        self.options
+            .get(DIFF_PARALLELISM_OPTION)
+            .and_then(|s| s.parse().ok())
+            .unwrap_or(DEFAULT_DIFF_PARALLELISM)
+            .max(1)
+    }
+
     pub fn data_evolution_enabled(&self) -> bool {
         self.options
             .get(DATA_EVOLUTION_ENABLED_OPTION)
@@ -1481,6 +1494,21 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_diff_parallelism_defaults() {
+        let options = HashMap::new();
+        let core = CoreOptions::new(&options);
+        assert_eq!(core.diff_parallelism(), 4);
+
+        let options = HashMap::from([(DIFF_PARALLELISM_OPTION.to_string(), "0".into())]);
+        let core = CoreOptions::new(&options);
+        assert_eq!(core.diff_parallelism(), 1);
+
+        let options = HashMap::from([(DIFF_PARALLELISM_OPTION.to_string(), "8".into())]);
+        let core = CoreOptions::new(&options);
+        assert_eq!(core.diff_parallelism(), 8);
+    }
+
     #[test]
     fn test_changelog_producer_accepts_known_values() {
         for (value, expected) in [

crates/paimon/src/table/audit_log_table.rs

@@ -27,7 +27,7 @@ use crate::spec::{
 /// Incremental reads produce:
 /// - Delta: primary-key rows use physical `_VALUE_KIND`; append rows are `+I`
 /// - Changelog: kinds come from physical `_VALUE_KIND` (`+I`/`-U`/`+U`/`-D`)
-/// - Diff: not implemented in this release
+/// - Diff: before/after image comparison (`+I`/`-U`/`+U`/`-D`, equal keys skipped)
 #[derive(Debug, Clone)]
 pub struct AuditLogTable {
     wrapped: Table,
@@ -81,6 +81,7 @@ impl AuditLogTable {
     }
 
     pub fn to_arrow(&self, plan: &IncrementalPlan) -> crate::Result<ArrowRecordBatchStream> {
+        plan.validate()?;
         let read = self.wrapped.new_read_builder().new_read()?;
         read.to_audit_log_arrow(plan)
     }

crates/paimon/src/table/incremental_scan.rs

@@ -35,18 +35,19 @@ pub enum IncrementalScanMode {
     /// Resolve to [`Delta`](Self::Delta) when `changelog-producer=none`,
     /// otherwise to [`Changelog`](Self::Changelog).
     Auto,
-    /// Diff before/after snapshots.
+    /// Diff before/after snapshot states for PK tables.
     ///
-    /// Not fully implemented in this release; planning returns
-    /// [`Error::Unsupported`](crate::Error::Unsupported).
+    /// Phase 1 supports only `merge-engine=deduplicate`. Planning compares the
+    /// full table state at `start_exclusive` vs `end_inclusive` and yields
+    /// per-(partition, bucket) [`IncrementalSplit::DiffPair`] units.
     Diff,
 }
 
 /// A unit of work produced by an incremental plan.
 #[derive(Debug, Clone)]
 pub enum IncrementalSplit {
     Data(DataSplit),
-    /// Per-(partition, bucket) diff pair. Memory bounded by one bucket's data.
+    /// Per-(partition, bucket) diff pair.
     DiffPair {
         before: Vec<DataSplit>,
         after: Vec<DataSplit>,
@@ -65,6 +66,80 @@ impl IncrementalPlan {
         Self { mode, splits }
     }
 
+    pub fn try_new(
+        mode: IncrementalScanMode,
+        splits: Vec<IncrementalSplit>,
+    ) -> crate::Result<Self> {
+        let plan = Self::new(mode, splits);
+        plan.validate()?;
+        Ok(plan)
+    }
+
+    /// Validate the plan at every point it crosses into a reader.
+    ///
+    /// `new` is retained for source compatibility, so callers can still build
+    /// an invalid plan. Readers must call this method instead of assuming a
+    /// plan came from the scanner.
+    pub fn validate(&self) -> crate::Result<()> {
+        if self.mode == IncrementalScanMode::Auto {
+            return Err(crate::Error::DataInvalid {
+                message: "Incremental plan mode Auto must be resolved before consumption"
+                    .to_string(),
+                source: None,
+            });
+        }
+        if self.mode == IncrementalScanMode::Diff {
+            let mut before_snapshot_id = None;
+            let mut after_snapshot_id = None;
+            for split in &self.splits {
+                let IncrementalSplit::DiffPair { before, after } = split else {
+                    return Err(crate::Error::DataInvalid {
+                        message: "Diff incremental plan contains a Data split".to_string(),
+                        source: None,
+                    });
+                };
+                validate_diff_pair(before, after)?;
+                if let Some(snapshot_id) = before.first().map(DataSplit::snapshot_id) {
+                    if before_snapshot_id.is_some_and(|expected| expected != snapshot_id) {
+                        return Err(crate::Error::DataInvalid {
+                            message: "Diff plan contains different before snapshots".to_string(),
+                            source: None,
+                        });
+                    }
+                    before_snapshot_id = Some(snapshot_id);
+                }
+                if let Some(snapshot_id) = after.first().map(DataSplit::snapshot_id) {
+                    if after_snapshot_id.is_some_and(|expected| expected != snapshot_id) {
+                        return Err(crate::Error::DataInvalid {
+                            message: "Diff plan contains different after snapshots".to_string(),
+                            source: None,
+                        });
+                    }
+                    after_snapshot_id = Some(snapshot_id);
+                }
+            }
+            if let (Some(before), Some(after)) = (before_snapshot_id, after_snapshot_id) {
+                if before >= after {
+                    return Err(crate::Error::DataInvalid {
+                        message: "Diff plan before snapshot must be earlier than after snapshot"
+                            .to_string(),
+                        source: None,
+                    });
+                }
+            }
+        } else if self
+            .splits
+            .iter()
+            .any(|split| matches!(split, IncrementalSplit::DiffPair { .. }))
+        {
+            return Err(crate::Error::DataInvalid {
+                message: "Non-Diff incremental plan contains a DiffPair".to_string(),
+                source: None,
+            });
+        }
+        Ok(())
+    }
+
     /// Resolved mode (`Auto` already collapsed to `Delta` / `Changelog`).
     pub fn mode(&self) -> IncrementalScanMode {
         self.mode
@@ -85,6 +160,49 @@ impl IncrementalPlan {
     }
 }
 
+pub(crate) fn validate_diff_pair(before: &[DataSplit], after: &[DataSplit]) -> crate::Result<()> {
+    if before
+        .iter()
+        .chain(after)
+        .any(|split| split.row_ranges().is_some())
+    {
+        return Err(crate::Error::DataInvalid {
+            message: "Diff pair must not contain physical row ranges".to_string(),
+            source: None,
+        });
+    }
+    let first = before.first().or(after.first());
+    let Some(first) = first else {
+        return Ok(());
+    };
+    for side in [before, after] {
+        if let Some(first_in_side) = side.first() {
+            if side
+                .iter()
+                .any(|split| split.snapshot_id() != first_in_side.snapshot_id())
+            {
+                return Err(crate::Error::DataInvalid {
+                    message: "Diff pair side contains splits from different snapshots".to_string(),
+                    source: None,
+                });
+            }
+        }
+    }
+    for split in before.iter().chain(after) {
+        if split.partition() != first.partition()
+            || split.bucket() != first.bucket()
+            || split.bucket_path() != first.bucket_path()
+            || split.total_buckets() != first.total_buckets()
+        {
+            return Err(crate::Error::DataInvalid {
+                message: "Diff pair contains splits from different partition buckets".to_string(),
+                source: None,
+            });
+        }
+    }
+    Ok(())
+}
+
 /// Batch incremental scan over a snapshot id range.
 pub struct IncrementalScan<'a> {
     table: &'a Table,
@@ -201,7 +319,7 @@ impl<'a> IncrementalScan<'a> {
             let plan = self.scan.plan_snapshot_delta(&snapshot).await?;
             splits.extend(plan.splits().iter().cloned().map(IncrementalSplit::Data));
         }
-        Ok(IncrementalPlan::new(mode, splits))
+        IncrementalPlan::try_new(mode, splits)
     }
 
     async fn plan_changelog(&self, mode: IncrementalScanMode) -> crate::Result<IncrementalPlan> {
@@ -219,13 +337,60 @@ impl<'a> IncrementalScan<'a> {
             let plan = self.scan.plan_snapshot_changelog(&snapshot).await?;
             splits.extend(plan.splits().iter().cloned().map(IncrementalSplit::Data));
         }
-        Ok(IncrementalPlan::new(mode, splits))
+        IncrementalPlan::try_new(mode, splits)
     }
 
     async fn plan_diff(&self, mode: IncrementalScanMode) -> crate::Result<IncrementalPlan> {
-        let _ = mode;
-        Err(crate::Error::Unsupported {
-            message: "Batch incremental Diff scan is not implemented yet".to_string(),
-        })
+        if self.table.schema().primary_keys().is_empty() {
+            return Err(crate::Error::Unsupported {
+                message: "Batch incremental Diff requires a table with primary keys".to_string(),
+            });
+        }
+        let core_options = CoreOptions::new(self.table.schema().options());
+        if core_options.merge_engine()? != crate::spec::MergeEngine::Deduplicate {
+            return Err(crate::Error::Unsupported {
+                message: "Batch incremental Diff only supports merge-engine=deduplicate in Phase 1"
+                    .to_string(),
+            });
+        }
+        let before = self
+            .snapshot_manager
+            .get_snapshot(self.start_exclusive)
+            .await?;
+        let after = self
+            .snapshot_manager
+            .get_snapshot(self.end_inclusive)
+            .await?;
+        let (before_plan, after_plan) = self.scan.plan_snapshot_diff(&before, &after).await?;
+
+        use std::collections::BTreeMap;
+        type PBKey = (Vec<u8>, i32);
+
+        let mut before_map: BTreeMap<PBKey, Vec<DataSplit>> = BTreeMap::new();
+        for split in before_plan.splits() {
+            let key = (split.partition().to_serialized_bytes(), split.bucket());
+            before_map.entry(key).or_default().push(split.clone());
+        }
+
+        let mut after_map: BTreeMap<PBKey, Vec<DataSplit>> = BTreeMap::new();
+        for split in after_plan.splits() {
+            let key = (split.partition().to_serialized_bytes(), split.bucket());
+            after_map.entry(key).or_default().push(split.clone());
+        }
+
+        let mut keys: std::collections::BTreeSet<PBKey> = before_map.keys().cloned().collect();
+        keys.extend(after_map.keys().cloned());
+
+        let mut splits = Vec::new();
+        for key in keys {
+            let before = before_map.remove(&key).unwrap_or_default();
+            let after = after_map.remove(&key).unwrap_or_default();
+            if before.is_empty() && after.is_empty() {
+                continue;
+            }
+            splits.push(IncrementalSplit::DiffPair { before, after });
+        }
+
+        IncrementalPlan::try_new(mode, splits)
     }
 }

crates/paimon/src/table/kv_file_reader.rs

@@ -70,6 +70,8 @@ pub(crate) struct KeyValueReadConfig {
     pub primary_keys: Vec<String>,
     pub merge_engine: MergeEngine,
     pub sequence_fields: Vec<String>,
+    /// Merge files from all supplied splits into one globally key-sorted stream.
+    pub merge_splits: bool,
 }
 
 /// Keep only the conjuncts of `predicates` that reference primary-key columns,
@@ -301,7 +303,15 @@ impl KeyValueFileReader {
             }
         }
 
-        let splits: Vec<DataSplit> = data_splits.to_vec();
+        let split_groups: Vec<Vec<DataSplit>> = if self.config.merge_splits {
+            vec![data_splits.to_vec()]
+        } else {
+            data_splits
+                .iter()
+                .cloned()
+                .map(|split| vec![split])
+                .collect()
+        };
         let file_io = self.file_io;
         let merge_engine = self.config.merge_engine;
         let schema_manager = self.config.schema_manager;
@@ -321,21 +331,23 @@ impl KeyValueFileReader {
         let merge_output_schema = build_target_arrow_schema(&merge_output_fields)?;
 
         Ok(try_stream! {
-            for split in &splits {
+            for split_group in &split_groups {
                 // DV mode should not reach KeyValueFileReader.
-                if split
-                    .data_deletion_files()
-                    .is_some_and(|files| files.iter().any(Option::is_some))
-                {
-                    Err(Error::Unsupported {
-                        message: "KeyValueFileReader does not support deletion vectors".to_string(),
-                    })?;
+                for split in split_group {
+                    if split
+                        .data_deletion_files()
+                        .is_some_and(|files| files.iter().any(Option::is_some))
+                    {
+                        Err(Error::Unsupported {
+                            message: "KeyValueFileReader does not support deletion vectors".to_string(),
+                        })?;
+                    }
                 }
-
                 // Create one stream per data file.
                 let mut file_streams: Vec<ArrowRecordBatchStream> = Vec::new();
 
-                for file_meta in split.data_files().to_vec() {
+                for split in split_group {
+                    for file_meta in split.data_files().to_vec() {
                     let data_fields: Option<Vec<DataField>> = if file_meta.schema_id != table_schema_id {
                         let data_schema = schema_manager.schema(file_meta.schema_id).await?;
                         Some(data_schema.fields().to_vec())
@@ -357,9 +369,10 @@ impl KeyValueFileReader {
                         file_meta,
                         data_fields,
                         None,
-                        None,
+                        split.row_ranges().map(|ranges| ranges.to_vec()),
                     )?;
                     file_streams.push(stream);
+                    }
                 }
 
                 if file_streams.is_empty() {