feat: support existing changelog incremental reads

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/audit_log_table.rs

@@ -0,0 +1,87 @@
+// 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::incremental_scan::{IncrementalPlan, IncrementalScan, IncrementalScanMode};
+use super::{ArrowRecordBatchStream, Table};
+use crate::spec::{
+    BigIntType, DataField, DataType, VarCharType, SEQUENCE_NUMBER_FIELD_ID,
+    SEQUENCE_NUMBER_FIELD_NAME,
+};
+
+/// Wrapper that exposes table rows with a leading `rowkind` audit column.
+///
+/// Incremental reads produce:
+/// - Delta: every row is `+I`
+/// - Changelog: kinds come from physical `_VALUE_KIND` (`+I`/`-U`/`+U`/`-D`)
+/// - Diff: not implemented in this release
+#[derive(Debug, Clone)]
+pub struct AuditLogTable {
+    wrapped: Table,
+}
+
+const TABLE_READ_SEQUENCE_NUMBER_ENABLED: &str = "table-read.sequence-number.enabled";
+
+impl AuditLogTable {
+    pub fn new(wrapped: Table) -> Self {
+        Self { wrapped }
+    }
+
+    pub fn wrapped(&self) -> &Table {
+        &self.wrapped
+    }
+
+    /// Logical fields: `rowkind` (+ optional `_SEQUENCE_NUMBER`) then table fields.
+    pub fn fields(&self) -> crate::Result<Vec<DataField>> {
+        let mut fields = Vec::with_capacity(self.wrapped.schema().fields().len() + 2);
+        fields.push(DataField::new(
+            -1,
+            "rowkind".to_string(),
+            DataType::VarChar(VarCharType::new(8)?),
+        ));
+        if self.sequence_number_enabled() {
+            fields.push(DataField::new(
+                SEQUENCE_NUMBER_FIELD_ID,
+                SEQUENCE_NUMBER_FIELD_NAME.to_string(),
+                DataType::BigInt(BigIntType::new()),
+            ));
+        }
+        fields.extend(self.wrapped.schema().fields().iter().cloned());
+        Ok(fields)
+    }
+
+    fn sequence_number_enabled(&self) -> bool {
+        self.wrapped
+            .schema()
+            .options()
+            .get(TABLE_READ_SEQUENCE_NUMBER_ENABLED)
+            .is_some_and(|v| v.eq_ignore_ascii_case("true"))
+    }
+
+    pub fn new_incremental_scan(
+        &self,
+        mode: IncrementalScanMode,
+        start_exclusive: i64,
+        end_inclusive: i64,
+    ) -> IncrementalScan<'_> {
+        IncrementalScan::for_table(&self.wrapped, mode, start_exclusive, end_inclusive)
+    }
+
+    pub fn to_arrow(&self, plan: &IncrementalPlan) -> crate::Result<ArrowRecordBatchStream> {
+        let read = self.wrapped.new_read_builder().new_read()?;
+        read.to_audit_log_arrow(plan)
+    }
+}

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,231 @@
+// 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 existing changelog manifest files in the range.
+    ///
+    /// Skips [`OVERWRITE`](crate::spec::CommitKind::OVERWRITE) snapshots and
+    /// snapshots without a `changelog_manifest_list`. Does not generate
+    /// changelogs (no compact/lookup producer path).
+    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 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?;
+            // OVERWRITE rewrites table contents and does not contribute changelog
+            // files for batch incremental reads (Java IncrementalChangelogStartingScanner).
+            if snapshot.commit_kind() == &CommitKind::OVERWRITE {
+                continue;
+            }
+            if snapshot.changelog_manifest_list().is_none() {
+                continue;
+            }
+            let plan = self.scan.plan_snapshot_changelog(&snapshot).await?;
+            splits.extend(plan.splits().iter().cloned().map(IncrementalSplit::Data));
+        }
+        Ok(IncrementalPlan::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(),
+        })
+    }
+}

crates/paimon/src/table/mod.rs

@@ -18,6 +18,7 @@
 //! Table API for Apache Paimon
 
 pub(crate) mod aggregator;
+mod audit_log_table;
 pub(crate) mod bin_pack;
 mod bitmap_global_index_reader;
 mod blob_resolver;
@@ -48,6 +49,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;
@@ -79,6 +81,7 @@ mod write_builder;
 
 use crate::Result;
 use arrow_array::RecordBatch;
+pub use audit_log_table::AuditLogTable;
 pub use branch_manager::BranchManager;
 pub use btree_global_index_build_builder::BTreeGlobalIndexBuildBuilder;
 pub use commit_message::CommitMessage;
@@ -94,6 +97,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;