feat: support existing changelog incremental reads

Author: Pandas886

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

@@ -26,10 +26,11 @@ use crate::spec::{CommitKind, CoreOptions};
 pub enum IncrementalScanMode {
     /// Read data files from APPEND snapshots in the range (delta manifests).
     Delta,
-    /// Read changelog manifest files in the range.
+    /// Read existing changelog manifest files in the range.
     ///
-    /// Not fully implemented in this release; planning returns
-    /// [`Error::Unsupported`](crate::Error::Unsupported).
+    /// 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).
@@ -204,10 +205,21 @@ impl<'a> IncrementalScan<'a> {
     }
 
     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(),
-        })
+        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> {

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;
@@ -82,6 +83,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;

crates/paimon/src/table/table_read.rs

@@ -22,8 +22,17 @@ use super::incremental_scan::{IncrementalPlan, IncrementalScanMode, IncrementalS
 use super::kv_file_reader::{KeyValueFileReader, KeyValueReadConfig};
 use super::read_builder::split_scan_predicates;
 use super::{ArrowRecordBatchStream, Table};
-use crate::spec::{CoreOptions, DataField, MergeEngine, Predicate};
+use crate::arrow::build_target_arrow_schema;
+use crate::spec::{
+    BigIntType, CoreOptions, DataField, DataType, MergeEngine, Predicate, TinyIntType,
+    SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID,
+    VALUE_KIND_FIELD_NAME,
+};
 use crate::DataSplit;
+use arrow_array::{ArrayRef, RecordBatch, StringArray};
+use arrow_schema::Schema as ArrowSchema;
+use futures::StreamExt;
+use std::sync::Arc;
 
 /// Table read: reads data from splits (e.g. produced by [TableScan::plan]).
 ///
@@ -124,6 +133,23 @@ impl<'a> TableRead<'a> {
             }),
         }
     }
+
+    /// Returns an audit-log [`ArrowRecordBatchStream`] for an incremental plan.
+    ///
+    /// Output schema is `rowkind` (+ optional `_SEQUENCE_NUMBER`) followed by
+    /// the projected user columns. Delta rows are all `+I`; Changelog rows take
+    /// kinds from `_VALUE_KIND`. Diff remains unsupported.
+    pub fn to_audit_log_arrow(
+        &self,
+        plan: &IncrementalPlan,
+    ) -> crate::Result<ArrowRecordBatchStream> {
+        match &self.0 {
+            TableReadKind::Paimon(read) => read.to_audit_log_arrow(plan),
+            TableReadKind::Format(_) => Err(crate::Error::Unsupported {
+                message: "Format tables do not support audit log batch read".to_string(),
+            }),
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -205,6 +231,109 @@ impl<'a> PaimonTableRead<'a> {
         self.new_data_file_reader().read(&data_splits)
     }
 
+    /// Returns an audit-log stream for a planned incremental scan.
+    pub fn to_audit_log_arrow(
+        &self,
+        plan: &IncrementalPlan,
+    ) -> crate::Result<ArrowRecordBatchStream> {
+        match plan.mode() {
+            IncrementalScanMode::Diff => Err(crate::Error::Unsupported {
+                message: "Batch incremental Diff audit read not yet implemented".to_string(),
+            }),
+            IncrementalScanMode::Delta => self.audit_raw_stream(plan, false),
+            IncrementalScanMode::Changelog => self.audit_raw_stream(plan, true),
+            IncrementalScanMode::Auto => unreachable!("Auto resolved during plan()"),
+        }
+    }
+
+    fn audit_raw_stream(
+        &self,
+        plan: &IncrementalPlan,
+        has_value_kind: bool,
+    ) -> crate::Result<ArrowRecordBatchStream> {
+        let data_splits = plan.data_splits();
+        let user_read_type = self.read_type.clone();
+        let include_sequence = audit_sequence_number_enabled(self.table);
+        let audit_schema = audit_schema_for_read_type(&user_read_type, include_sequence)?;
+
+        let mut read_type = user_read_type.clone();
+        if include_sequence {
+            read_type.insert(
+                0,
+                DataField::new(
+                    SEQUENCE_NUMBER_FIELD_ID,
+                    SEQUENCE_NUMBER_FIELD_NAME.to_string(),
+                    DataType::BigInt(BigIntType::new()),
+                ),
+            );
+        }
+        if has_value_kind {
+            read_type.push(DataField::new(
+                VALUE_KIND_FIELD_ID,
+                VALUE_KIND_FIELD_NAME.to_string(),
+                DataType::TinyInt(TinyIntType::new()),
+            ));
+        }
+
+        let reader = DataFileReader::new(
+            self.table.file_io.clone(),
+            self.table.schema_manager().clone(),
+            self.table.schema().id(),
+            self.table.schema.fields().to_vec(),
+            read_type,
+            self.data_predicates.clone(),
+        );
+        let raw_stream = reader.read(&data_splits)?;
+
+        Ok(Box::pin(async_stream::try_stream! {
+            futures::pin_mut!(raw_stream);
+            while let Some(batch) = raw_stream.next().await {
+                let batch = batch?;
+                let rowkind_col: ArrayRef = if has_value_kind {
+                    let col = batch
+                        .column_by_name(VALUE_KIND_FIELD_NAME)
+                        .ok_or_else(|| crate::Error::DataInvalid {
+                            message: "Changelog audit read missing _VALUE_KIND column".to_string(),
+                            source: None,
+                        })?;
+                    Arc::new(rowkind_array_from_column(col)?)
+                } else {
+                    let inserts: Vec<&'static str> = (0..batch.num_rows()).map(|_| "+I").collect();
+                    Arc::new(StringArray::from(inserts))
+                };
+
+                let mut columns: Vec<ArrayRef> = vec![rowkind_col];
+                if include_sequence {
+                    let seq_col = batch
+                        .column_by_name(SEQUENCE_NUMBER_FIELD_NAME)
+                        .ok_or_else(|| crate::Error::DataInvalid {
+                            message: "Audit read missing _SEQUENCE_NUMBER column".to_string(),
+                            source: None,
+                        })?;
+                    columns.push(seq_col.clone());
+                }
+                for field in &user_read_type {
+                    let col = batch
+                        .column_by_name(field.name())
+                        .ok_or_else(|| crate::Error::DataInvalid {
+                            message: format!(
+                                "Audit read missing column '{}'",
+                                field.name()
+                            ),
+                            source: None,
+                        })?;
+                    columns.push(col.clone());
+                }
+                yield RecordBatch::try_new(audit_schema.clone(), columns).map_err(|e| {
+                    crate::Error::UnexpectedError {
+                        message: format!("Failed to build audit log batch: {e}"),
+                        source: Some(Box::new(e)),
+                    }
+                })?;
+            }
+        }))
+    }
+
     /// 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();
@@ -352,6 +481,55 @@ impl<'a> PaimonTableRead<'a> {
     }
 }
 
+fn audit_schema_for_read_type(
+    read_type: &[DataField],
+    include_sequence: bool,
+) -> crate::Result<Arc<ArrowSchema>> {
+    let mut fields = Vec::with_capacity(read_type.len() + 2);
+    fields.push(DataField::new(
+        -1,
+        "rowkind".to_string(),
+        DataType::VarChar(crate::spec::VarCharType::new(8)?),
+    ));
+    if include_sequence {
+        fields.push(DataField::new(
+            SEQUENCE_NUMBER_FIELD_ID,
+            SEQUENCE_NUMBER_FIELD_NAME.to_string(),
+            DataType::BigInt(BigIntType::new()),
+        ));
+    }
+    fields.extend(read_type.iter().cloned());
+    build_target_arrow_schema(&fields)
+}
+
+fn audit_sequence_number_enabled(table: &Table) -> bool {
+    table
+        .schema()
+        .options()
+        .get("table-read.sequence-number.enabled")
+        .is_some_and(|v| v.eq_ignore_ascii_case("true"))
+}
+
+fn rowkind_array_from_column(column: &dyn arrow_array::Array) -> crate::Result<StringArray> {
+    let values = column
+        .as_any()
+        .downcast_ref::<arrow_array::Int8Array>()
+        .ok_or_else(|| crate::Error::DataInvalid {
+            message: "AuditLogTable _VALUE_KIND column must be Int8".to_string(),
+            source: None,
+        })?;
+    let strings: Vec<&'static str> = (0..values.len())
+        .map(|idx| match values.value(idx) {
+            0 => "+I",
+            1 => "-U",
+            2 => "+U",
+            3 => "-D",
+            _ => "?",
+        })
+        .collect();
+    Ok(StringArray::from(strings))
+}
+
 /// Whether a primary-key split must go through the sort-merge reader.
 ///
 /// Mirrors Java `PrimaryKeyTableRawFileSplitReadProvider#match`: a raw read

crates/paimon/src/table/table_scan.rs

@@ -716,6 +716,16 @@ impl<'a> TableScan<'a> {
         }
     }
 
+    /// Plan data splits from a snapshot's changelog manifest list only.
+    pub(crate) async fn plan_snapshot_changelog(&self, snapshot: &Snapshot) -> crate::Result<Plan> {
+        match &self.0 {
+            TableScanKind::Paimon(scan) => scan.plan_snapshot_changelog(snapshot).await,
+            TableScanKind::Format(_) => Err(crate::Error::Unsupported {
+                message: "Format tables do not support incremental changelog scan".to_string(),
+            }),
+        }
+    }
+
     #[cfg(test)]
     fn apply_limit_pushdown(&self, splits: Vec<DataSplit>) -> Vec<DataSplit> {
         match &self.0 {
@@ -1077,6 +1087,27 @@ impl<'a> PaimonTableScan<'a> {
         .await
     }
 
+    /// Plan data splits from a snapshot's changelog manifest list.
+    ///
+    /// Reuses the same split-building path as a full snapshot plan, but only
+    /// reads the changelog manifest list and keeps ADD entries. Snapshots
+    /// without a changelog list yield an empty plan.
+    pub(crate) async fn plan_snapshot_changelog(&self, snapshot: &Snapshot) -> crate::Result<Plan> {
+        self.ensure_query_auth_allowed()?;
+        let Some(list_name) = snapshot.changelog_manifest_list() else {
+            return Ok(Plan::new(Vec::new()));
+        };
+        let entries = self.plan_manifest_list_entries(list_name).await?;
+        let data_evolution_read_field_ids = self.projected_read_field_ids()?;
+        self.plan_snapshot_from_entries(
+            snapshot.clone(),
+            entries,
+            data_evolution_read_field_ids.as_ref(),
+            None,
+        )
+        .await
+    }
+
     /// Read entries from a single manifest list (delta or changelog) with
     /// partition / bucket filter pushdown matching the full scan path.
     async fn plan_manifest_list_entries(