feat: support batch incremental diff reads

Author: Pandas886

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,

crates/paimon/src/table/incremental_scan.rs

@@ -35,10 +35,11 @@ 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,
 }
 
@@ -223,9 +224,51 @@ impl<'a> IncrementalScan<'a> {
     }
 
     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(),
-        })
+        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 });
+        }
+
+        Ok(IncrementalPlan::new(mode, splits))
     }
 }