feat: support existing changelog incremental reads (#509)

Author: Pandas886

crates/paimon/src/spec/schema.rs

@@ -811,6 +811,11 @@ pub const VALUE_KIND_FIELD_NAME: &str = "_VALUE_KIND";
 /// Must match Java Paimon's `SpecialFields.VALUE_KIND` (Integer.MAX_VALUE - 2).
 pub const VALUE_KIND_FIELD_ID: i32 = i32::MAX - 2;
 
+pub const ROW_KIND_FIELD_NAME: &str = "rowkind";
+
+/// Must match Java Paimon's `SpecialFields.ROW_KIND` (Integer.MAX_VALUE - 4).
+pub const ROW_KIND_FIELD_ID: i32 = i32::MAX - 4;
+
 /// Data field for paimon table.
 ///
 /// Impl Reference: <https://github.com/apache/paimon/blob/release-0.8.2/paimon-common/src/main/java/org/apache/paimon/types/DataField.java#L40>

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, ROW_KIND_FIELD_ID, ROW_KIND_FIELD_NAME,
+    SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME,
+};
+
+/// Wrapper that exposes table rows with a leading `rowkind` audit column.
+///
+/// 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
+#[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(
+            ROW_KIND_FIELD_ID,
+            ROW_KIND_FIELD_NAME.to_string(),
+            DataType::VarChar(VarCharType::string_type()),
+        ));
+        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;
@@ -85,6 +86,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;