feat: support batch incremental delta scans

crates/paimon/src/lib.rs

@@ -47,9 +47,10 @@ pub use catalog::FileSystemCatalog;
 
 pub use table::{
     CommitMessage, DataEvolutionDeleteWriter, DataEvolutionWriter, DataSplit, DataSplitBuilder,
-    DeletionFile, PartitionBucket, Plan, RESTEnv, RESTSnapshotCommit, ReadBuilder,
-    RenamingSnapshotCommit, RowRange, ScanTrace, SnapshotCommit, SnapshotManager, Table,
-    TableCommit, TableRead, TableScan, TableUpdate, TableWrite, TagManager, WriteBuilder,
+    DeletionFile, IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit,
+    PartitionBucket, Plan, RESTEnv, RESTSnapshotCommit, ReadBuilder, RenamingSnapshotCommit,
+    RowRange, ScanTrace, SnapshotCommit, SnapshotManager, Table, TableCommit, TableRead, TableScan,
+    TableUpdate, TableWrite, TagManager, WriteBuilder,
 };
 
 pub use table::{

crates/paimon/src/table/format_read_builder.rs

@@ -45,6 +45,10 @@ impl<'a> FormatReadBuilder<'a> {
         }
     }
 
+    pub(crate) fn table(&self) -> &'a Table {
+        self.table
+    }
+
     pub(crate) fn with_projection(&mut self, columns: &[&str]) -> Result<&mut Self> {
         let projection_names = columns.iter().map(|c| (*c).to_string()).collect::<Vec<_>>();
         self.read_type = Some(resolve_projected_fields(

crates/paimon/src/table/incremental_scan.rs

@@ -0,0 +1,219 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::{DataSplit, SnapshotManager, Table, TableScan};
+use crate::spec::{CommitKind, CoreOptions};
+
+/// Batch incremental scan mode.
+///
+/// Range semantics: `(start_exclusive, end_inclusive]` — start is exclusive and
+/// end is inclusive. An empty range (`start == end`) yields an empty plan.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum IncrementalScanMode {
+    /// Read data files from APPEND snapshots in the range (delta manifests).
+    Delta,
+    /// Read changelog manifest files in the range.
+    ///
+    /// Not fully implemented in this release; planning returns
+    /// [`Error::Unsupported`](crate::Error::Unsupported).
+    Changelog,
+    /// Resolve to [`Delta`](Self::Delta) when `changelog-producer=none`,
+    /// otherwise to [`Changelog`](Self::Changelog).
+    Auto,
+    /// Diff before/after snapshots.
+    ///
+    /// Not fully implemented in this release; planning returns
+    /// [`Error::Unsupported`](crate::Error::Unsupported).
+    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.
+    DiffPair {
+        before: Vec<DataSplit>,
+        after: Vec<DataSplit>,
+    },
+}
+
+/// Planned incremental scan: resolved mode plus splits.
+#[derive(Debug, Clone)]
+pub struct IncrementalPlan {
+    mode: IncrementalScanMode,
+    splits: Vec<IncrementalSplit>,
+}
+
+impl IncrementalPlan {
+    pub fn new(mode: IncrementalScanMode, splits: Vec<IncrementalSplit>) -> Self {
+        Self { mode, splits }
+    }
+
+    /// Resolved mode (`Auto` already collapsed to `Delta` / `Changelog`).
+    pub fn mode(&self) -> IncrementalScanMode {
+        self.mode
+    }
+
+    pub fn splits(&self) -> &[IncrementalSplit] {
+        &self.splits
+    }
+
+    pub fn data_splits(&self) -> Vec<DataSplit> {
+        self.splits
+            .iter()
+            .filter_map(|split| match split {
+                IncrementalSplit::Data(data) => Some(data.clone()),
+                IncrementalSplit::DiffPair { .. } => None,
+            })
+            .collect()
+    }
+}
+
+/// Batch incremental scan over a snapshot id range.
+pub struct IncrementalScan<'a> {
+    table: &'a Table,
+    scan: TableScan<'a>,
+    snapshot_manager: SnapshotManager,
+    mode: IncrementalScanMode,
+    start_exclusive: i64,
+    end_inclusive: i64,
+}
+
+impl<'a> IncrementalScan<'a> {
+    pub(crate) fn for_table(
+        table: &'a Table,
+        mode: IncrementalScanMode,
+        start_exclusive: i64,
+        end_inclusive: i64,
+    ) -> Self {
+        let scan = TableScan::new(table, None, Vec::new(), None, None, None);
+        Self::new(table, scan, mode, start_exclusive, end_inclusive)
+    }
+
+    pub(crate) fn new(
+        table: &'a Table,
+        scan: TableScan<'a>,
+        mode: IncrementalScanMode,
+        start_exclusive: i64,
+        end_inclusive: i64,
+    ) -> Self {
+        let snapshot_manager =
+            SnapshotManager::new(table.file_io().clone(), table.location().to_string());
+        Self {
+            table,
+            scan,
+            snapshot_manager,
+            mode,
+            start_exclusive,
+            end_inclusive,
+        }
+    }
+
+    pub async fn plan(&self) -> crate::Result<IncrementalPlan> {
+        let mode = self.resolve_mode();
+        self.validate_snapshot_range(mode).await?;
+        if self.start_exclusive == self.end_inclusive {
+            return Ok(IncrementalPlan::new(mode, Vec::new()));
+        }
+        match mode {
+            IncrementalScanMode::Delta => self.plan_delta(mode).await,
+            IncrementalScanMode::Changelog => self.plan_changelog(mode).await,
+            IncrementalScanMode::Auto => unreachable!("Auto must resolve before planning"),
+            IncrementalScanMode::Diff => self.plan_diff(mode).await,
+        }
+    }
+
+    fn resolve_mode(&self) -> IncrementalScanMode {
+        match self.mode {
+            IncrementalScanMode::Auto => {
+                let core_options = CoreOptions::new(self.table.schema().options());
+                let producer = core_options.changelog_producer();
+                if producer.eq_ignore_ascii_case("none") {
+                    IncrementalScanMode::Delta
+                } else {
+                    IncrementalScanMode::Changelog
+                }
+            }
+            mode => mode,
+        }
+    }
+
+    async fn validate_snapshot_range(&self, mode: IncrementalScanMode) -> crate::Result<()> {
+        let earliest = self
+            .snapshot_manager
+            .earliest_snapshot_id()
+            .await?
+            .ok_or_else(|| crate::Error::DataInvalid {
+                message: "No snapshots available for incremental scan".to_string(),
+                source: None,
+            })?;
+        let latest = self
+            .snapshot_manager
+            .get_latest_snapshot_id()
+            .await?
+            .ok_or_else(|| crate::Error::DataInvalid {
+                message: "No snapshots available for incremental scan".to_string(),
+                source: None,
+            })?;
+        let min_start = match mode {
+            IncrementalScanMode::Diff => earliest,
+            IncrementalScanMode::Delta | IncrementalScanMode::Changelog => earliest - 1,
+            IncrementalScanMode::Auto => unreachable!("Auto must resolve before validation"),
+        };
+        if self.start_exclusive < min_start
+            || self.end_inclusive > latest
+            || self.start_exclusive > self.end_inclusive
+        {
+            return Err(crate::Error::DataInvalid {
+                message: format!(
+                    "Incremental snapshot range [{}, {}] is out of available range [{}, {}] for {:?}",
+                    self.start_exclusive, self.end_inclusive, min_start, latest, mode
+                ),
+                source: None,
+            });
+        }
+        Ok(())
+    }
+
+    async fn plan_delta(&self, mode: IncrementalScanMode) -> crate::Result<IncrementalPlan> {
+        let mut splits = Vec::new();
+        for snapshot_id in (self.start_exclusive + 1)..=self.end_inclusive {
+            let snapshot = self.snapshot_manager.get_snapshot(snapshot_id).await?;
+            if snapshot.commit_kind() != &CommitKind::APPEND {
+                continue;
+            }
+            let plan = self.scan.plan_snapshot_delta(&snapshot).await?;
+            splits.extend(plan.splits().iter().cloned().map(IncrementalSplit::Data));
+        }
+        Ok(IncrementalPlan::new(mode, splits))
+    }
+
+    async fn plan_changelog(&self, mode: IncrementalScanMode) -> crate::Result<IncrementalPlan> {
+        let _ = mode;
+        Err(crate::Error::Unsupported {
+            message: "Batch incremental Changelog scan is not implemented yet".to_string(),
+        })
+    }
+
+    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(),
+        })
+    }
+}

crates/paimon/src/table/mod.rs

@@ -48,6 +48,7 @@ mod global_index_drop_builder;
 pub(crate) mod global_index_scanner;
 mod global_index_types;
 mod hybrid_search_builder;
+mod incremental_scan;
 mod kv_file_reader;
 mod kv_file_writer;
 mod lumina_index_build_builder;
@@ -94,6 +95,9 @@ pub use global_index_types::{
 pub use hybrid_search_builder::{
     HybridSearchBuilder, HybridSearchRanker, HybridSearchRoute, HybridSearchRouteKind,
 };
+pub use incremental_scan::{
+    IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit,
+};
 pub use lumina_index_build_builder::LuminaIndexBuildBuilder;
 pub use read_builder::ReadBuilder;
 pub use rest_env::RESTEnv;

crates/paimon/src/table/read_builder.rs

@@ -22,6 +22,7 @@
 
 use super::bucket_filter::{extract_predicate_for_keys, split_partition_and_data_predicates};
 use super::format_read_builder::FormatReadBuilder;
+use super::incremental_scan::{IncrementalScan, IncrementalScanMode};
 use super::partition_filter::PartitionFilter;
 use super::table_read::TableRead;
 use super::{Table, TableScan};
@@ -210,6 +211,32 @@ impl<'a> ReadBuilder<'a> {
         }
     }
 
+    /// Create a batch incremental scan over snapshot id range
+    /// `(start_exclusive, end_inclusive]`.
+    ///
+    /// Filters and projection configured on this builder are pushed into the
+    /// incremental plan (partition / bucket pruning on the delta path).
+    pub fn new_incremental_scan(
+        &self,
+        mode: IncrementalScanMode,
+        start_exclusive: i64,
+        end_inclusive: i64,
+    ) -> IncrementalScan<'a> {
+        match &self.0 {
+            ReadBuilderKind::Paimon(builder) => IncrementalScan::new(
+                builder.table,
+                builder.new_scan(),
+                mode,
+                start_exclusive,
+                end_inclusive,
+            ),
+            // Format tables share the API surface; planning fails with Unsupported.
+            ReadBuilderKind::Format(builder) => {
+                IncrementalScan::for_table(builder.table(), mode, start_exclusive, end_inclusive)
+            }
+        }
+    }
+
     /// Create a table read for consuming splits (e.g. from a scan plan).
     pub fn new_read(&self) -> Result<TableRead<'a>> {
         match &self.0 {

crates/paimon/src/table/table_read.rs

@@ -18,6 +18,7 @@
 use super::data_evolution_reader::DataEvolutionReader;
 use super::data_file_reader::DataFileReader;
 use super::format_table_read::FormatTableRead;
+use super::incremental_scan::{IncrementalPlan, IncrementalScanMode, IncrementalSplit};
 use super::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig};
 use super::read_builder::split_scan_predicates;
 use super::{ArrowRecordBatchStream, Table};
@@ -107,6 +108,22 @@ impl<'a> TableRead<'a> {
             TableReadKind::Format(read) => read.to_arrow(data_splits),
         }
     }
+
+    /// Returns an [`ArrowRecordBatchStream`] for an incremental scan plan.
+    ///
+    /// Only [`IncrementalSplit::Data`] is supported in this release. Diff
+    /// planning/read remains unimplemented.
+    pub fn to_incremental_arrow(
+        &self,
+        plan: &IncrementalPlan,
+    ) -> crate::Result<ArrowRecordBatchStream> {
+        match &self.0 {
+            TableReadKind::Paimon(read) => read.to_incremental_arrow(plan),
+            TableReadKind::Format(_) => Err(crate::Error::Unsupported {
+                message: "Format tables do not support incremental batch read".to_string(),
+            }),
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -160,6 +177,34 @@ impl<'a> PaimonTableRead<'a> {
         self
     }
 
+    /// Returns an [`ArrowRecordBatchStream`] for an incremental scan plan.
+    pub fn to_incremental_arrow(
+        &self,
+        plan: &IncrementalPlan,
+    ) -> crate::Result<ArrowRecordBatchStream> {
+        if plan.mode() == IncrementalScanMode::Diff {
+            return Err(crate::Error::Unsupported {
+                message: "Batch incremental Diff read not yet implemented".to_string(),
+            });
+        }
+
+        let mut data_splits = Vec::new();
+        for split in plan.splits() {
+            match split {
+                IncrementalSplit::Data(data) => data_splits.push(data.clone()),
+                IncrementalSplit::DiffPair { .. } => {
+                    return Err(crate::Error::UnexpectedError {
+                        message: "DiffPair appeared in non-Diff incremental plan".to_string(),
+                        source: None,
+                    });
+                }
+            }
+        }
+        // Delta / Changelog rows are read as-is from planned files (no full-table
+        // merge against historical base versions).
+        self.new_data_file_reader().read(&data_splits)
+    }
+
     /// Returns an [`ArrowRecordBatchStream`].
     pub fn to_arrow(&self, data_splits: &[DataSplit]) -> crate::Result<ArrowRecordBatchStream> {
         let has_primary_keys = !self.table.schema.primary_keys().is_empty();