This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 677cc06 feat: support existing changelog incremental reads (#509)
677cc06 is described below
commit 677cc065497ca8f9e198caade5c7c3356d59420f
Author: Pandas <[email protected]>
AuthorDate: Thu Jul 16 15:27:14 2026 +0800
feat: support existing changelog incremental reads (#509)
---
crates/paimon/src/spec/schema.rs | 5 +
crates/paimon/src/table/audit_log_table.rs | 87 +++++
crates/paimon/src/table/incremental_scan.rs | 26 +-
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/table_read.rs | 217 ++++++++++++-
crates/paimon/src/table/table_scan.rs | 31 ++
crates/paimon/tests/audit_log_table_test.rs | 349 +++++++++++++++++++++
crates/paimon/tests/common/incremental_helpers.rs | 22 +-
crates/paimon/tests/incremental_batch_scan_test.rs | 163 +++++++++-
9 files changed, 886 insertions(+), 16 deletions(-)
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index c00d6de..f92e5ef 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/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>
diff --git a/crates/paimon/src/table/audit_log_table.rs
b/crates/paimon/src/table/audit_log_table.rs
new file mode 100644
index 0000000..e5ce757
--- /dev/null
+++ b/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)
+ }
+}
diff --git a/crates/paimon/src/table/incremental_scan.rs
b/crates/paimon/src/table/incremental_scan.rs
index 8088d82..3aa169e 100644
--- a/crates/paimon/src/table/incremental_scan.rs
+++ b/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> {
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 2a239e9..cf581aa 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/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;
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index f9e31ca..f3d1f64 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/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,
+ ROW_KIND_FIELD_ID, ROW_KIND_FIELD_NAME, SEQUENCE_NUMBER_FIELD_ID,
SEQUENCE_NUMBER_FIELD_NAME,
+ VALUE_KIND_FIELD_ID, VALUE_KIND_FIELD_NAME,
+};
use crate::DataSplit;
+use arrow_array::{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,24 @@ 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. Primary-key Delta and Changelog rows take
+ /// kinds from `_VALUE_KIND`; append-only Delta rows are `+I`. 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 +232,111 @@ 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,
!self.table.schema().primary_keys().is_empty())
+ }
+ 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 +484,70 @@ 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(
+ ROW_KIND_FIELD_ID,
+ ROW_KIND_FIELD_NAME.to_string(),
+ DataType::VarChar(crate::spec::VarCharType::string_type()),
+ ));
+ 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 mut strings = Vec::with_capacity(values.len());
+ for idx in 0..values.len() {
+ if values.is_null(idx) {
+ return Err(crate::Error::DataInvalid {
+ message: format!("AuditLogTable _VALUE_KIND is null at row
{idx}"),
+ source: None,
+ });
+ }
+ let rowkind = match values.value(idx) {
+ 0 => "+I",
+ 1 => "-U",
+ 2 => "+U",
+ 3 => "-D",
+ value => {
+ return Err(crate::Error::DataInvalid {
+ message: format!(
+ "AuditLogTable _VALUE_KIND has invalid value {value}
at row {idx}"
+ ),
+ source: None,
+ });
+ }
+ };
+ strings.push(rowkind);
+ }
+ Ok(StringArray::from(strings))
+}
+
/// Whether a primary-key split must go through the sort-merge reader.
///
/// Mirrors Java `PrimaryKeyTableRawFileSplitReadProvider#match`: a raw read
@@ -441,6 +637,25 @@ mod tests {
assert!(!pk_split_needs_merge(&dv_compacted, true));
}
+ #[test]
+ fn test_rowkind_rejects_null_value_kind() {
+ let values = arrow_array::Int8Array::from(vec![Some(0), None]);
+ assert!(matches!(
+ rowkind_array_from_column(&values),
+ Err(crate::Error::DataInvalid { ref message, .. }) if
message.contains("null at row 1")
+ ));
+ }
+
+ #[test]
+ fn test_rowkind_rejects_invalid_value_kind() {
+ let values = arrow_array::Int8Array::from(vec![4]);
+ assert!(matches!(
+ rowkind_array_from_column(&values),
+ Err(crate::Error::DataInvalid { ref message, .. })
+ if message.contains("invalid value 4 at row 0")
+ ));
+ }
+
#[test]
fn test_direct_table_read_fails_closed_when_query_auth_enabled() {
let table = query_auth_table();
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 6653423..8044590 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/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(
diff --git a/crates/paimon/tests/audit_log_table_test.rs
b/crates/paimon/tests/audit_log_table_test.rs
new file mode 100644
index 0000000..8839673
--- /dev/null
+++ b/crates/paimon/tests/audit_log_table_test.rs
@@ -0,0 +1,349 @@
+// 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.
+
+mod common;
+
+use arrow_array::{Array, Int32Array, Int64Array, RecordBatch, StringArray};
+use futures::TryStreamExt;
+use paimon::spec::{
+ DataType, IntType, Schema, TableSchema, VarCharType, ROW_KIND_FIELD_ID,
ROW_KIND_FIELD_NAME,
+ SEQUENCE_NUMBER_FIELD_NAME,
+};
+use paimon::table::{AuditLogTable, IncrementalScanMode};
+
+use common::incremental_helpers::{
+ make_batch, make_batch_with_kinds, memory_table, persist_table_schema,
pk_schema, setup_dirs,
+ write_batch,
+};
+
+fn collect_audit_rows(batches: &[RecordBatch]) -> Vec<(String, i32, i32)> {
+ let mut rows = Vec::new();
+ for batch in batches {
+ let schema = batch.schema();
+ let kind_idx = schema.index_of("rowkind").unwrap();
+ let id_idx = schema.index_of("id").unwrap();
+ let value_idx = schema.index_of("value").unwrap();
+ let kinds = batch
+ .column(kind_idx)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let ids = batch
+ .column(id_idx)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(value_idx)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((
+ kinds.value(row).to_string(),
+ ids.value(row),
+ values.value(row),
+ ));
+ }
+ }
+ rows.sort_unstable();
+ rows
+}
+
+fn collect_audit_rows_with_sequence(batches: &[RecordBatch]) -> Vec<(String,
i64, i32, i32)> {
+ let mut rows = Vec::new();
+ for batch in batches {
+ let schema = batch.schema();
+ let kind_idx = schema.index_of("rowkind").unwrap();
+ let seq_idx = schema.index_of(SEQUENCE_NUMBER_FIELD_NAME).unwrap();
+ let id_idx = schema.index_of("id").unwrap();
+ let value_idx = schema.index_of("value").unwrap();
+ let kinds = batch
+ .column(kind_idx)
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+ let seqs = batch
+ .column(seq_idx)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .unwrap();
+ let ids = batch
+ .column(id_idx)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(value_idx)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((
+ kinds.value(row).to_string(),
+ seqs.value(row),
+ ids.value(row),
+ values.value(row),
+ ));
+ }
+ }
+ rows.sort_unstable();
+ rows
+}
+
+#[tokio::test]
+async fn audit_log_changelog_scan_exposes_rowkind_as_first_column() {
+ let table_path = "memory:/audit_log/changelog_rowkind";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "input"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&make_batch_with_kinds(
+ vec![1, 1, 2, 2],
+ vec![10, 20, 25, 30],
+ vec![0, 1, 2, 3],
+ ))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let audit = AuditLogTable::new(table.clone());
+ let plan = audit
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 1)
+ .plan()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> =
audit.to_arrow(&plan).unwrap().try_collect().await.unwrap();
+
+ assert_eq!(batches[0].schema().field(0).name(), "rowkind");
+ assert_eq!(
+ collect_audit_rows(&batches),
+ vec![
+ ("+I".to_string(), 1, 10),
+ ("+U".to_string(), 2, 25),
+ ("-D".to_string(), 2, 30),
+ ("-U".to_string(), 1, 20),
+ ]
+ );
+}
+
+#[tokio::test]
+async fn audit_log_delta_scan_emits_plus_i_for_all_rows() {
+ let table_path = "memory:/audit_log/delta_plus_i";
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .option("bucket", "1")
+ .option("bucket-key", "id")
+ .build()
+ .unwrap();
+ let (file_io, table) = memory_table(table_path, TableSchema::new(0,
&schema));
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await;
+
+ let audit = AuditLogTable::new(table.clone());
+ let plan = audit
+ .new_incremental_scan(IncrementalScanMode::Delta, 0, 1)
+ .plan()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> =
audit.to_arrow(&plan).unwrap().try_collect().await.unwrap();
+
+ let rowkind_field = batches[0].schema().field(0).clone();
+ assert_eq!(rowkind_field.name(), ROW_KIND_FIELD_NAME);
+ assert_eq!(rowkind_field.data_type(), &arrow_schema::DataType::Utf8);
+ assert!(rowkind_field.is_nullable());
+ assert_eq!(
+ rowkind_field.metadata().get("PARQUET:field_id"),
+ Some(&ROW_KIND_FIELD_ID.to_string())
+ );
+
+ assert_eq!(
+ collect_audit_rows(&batches),
+ vec![("+I".to_string(), 1, 10), ("+I".to_string(), 2, 20),]
+ );
+}
+
+#[tokio::test]
+async fn audit_log_delta_scan_preserves_pk_row_kinds() {
+ let table_path = "memory:/audit_log/delta_rowkind";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "none"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&make_batch_with_kinds(
+ // Use distinct keys so the PK writer does not merge multiple
+ // changes for one key before the audit read sees the data file.
+ vec![1, 2, 3, 4],
+ vec![10, 20, 25, 30],
+ vec![0, 1, 2, 3],
+ ))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let audit = AuditLogTable::new(table.clone());
+ let plan = audit
+ .new_incremental_scan(IncrementalScanMode::Delta, 0, 1)
+ .plan()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> =
audit.to_arrow(&plan).unwrap().try_collect().await.unwrap();
+
+ assert_eq!(
+ collect_audit_rows(&batches),
+ vec![
+ ("+I".to_string(), 1, 10),
+ ("+U".to_string(), 3, 25),
+ ("-D".to_string(), 4, 30),
+ ("-U".to_string(), 2, 20),
+ ]
+ );
+}
+
+#[test]
+fn audit_log_rowkind_field_matches_java_special_field() {
+ let (_, table) = memory_table("memory:/audit_log/rowkind_field",
pk_schema(&[]));
+ let field = AuditLogTable::new(table).fields().unwrap().remove(0);
+
+ assert_eq!(field.id(), ROW_KIND_FIELD_ID);
+ assert_eq!(field.name(), ROW_KIND_FIELD_NAME);
+ assert!(matches!(
+ field.data_type(),
+ DataType::VarChar(varchar) if varchar.length() ==
VarCharType::MAX_LENGTH
+ ));
+}
+
+#[tokio::test]
+async fn audit_log_exposes_sequence_number_when_enabled() {
+ let table_path = "memory:/audit_log/sequence_number";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "input"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ("table-read.sequence-number.enabled", "true"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&make_batch_with_kinds(vec![1, 1], vec![10, 20],
vec![0, 1]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let audit = AuditLogTable::new(table.clone());
+ let field_names: Vec<String> = audit
+ .fields()
+ .unwrap()
+ .into_iter()
+ .map(|f| f.name().to_string())
+ .collect();
+ assert_eq!(
+ field_names,
+ vec![
+ "rowkind".to_string(),
+ SEQUENCE_NUMBER_FIELD_NAME.to_string(),
+ "id".to_string(),
+ "value".to_string(),
+ ]
+ );
+
+ let plan = audit
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 1)
+ .plan()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> =
audit.to_arrow(&plan).unwrap().try_collect().await.unwrap();
+ let batch_schema: Vec<String> = batches[0]
+ .schema()
+ .fields()
+ .iter()
+ .map(|f| f.name().clone())
+ .collect();
+ assert_eq!(
+ batch_schema,
+ vec![
+ "rowkind".to_string(),
+ SEQUENCE_NUMBER_FIELD_NAME.to_string(),
+ "id".to_string(),
+ "value".to_string(),
+ ]
+ );
+
+ let rows = collect_audit_rows_with_sequence(&batches);
+ assert_eq!(rows.len(), 2);
+ assert!(rows.iter().all(|(_, seq, _, _)| *seq >= 0));
+}
+
+#[tokio::test]
+async fn audit_log_diff_mode_is_unsupported() {
+ let table_path = "memory:/audit_log/diff_unsupported";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "none"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ write_batch(&table, &make_batch(vec![2], vec![20])).await;
+
+ let audit = AuditLogTable::new(table.clone());
+ let err = audit
+ .new_incremental_scan(IncrementalScanMode::Diff, 1, 2)
+ .plan()
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::Unsupported { .. }),
+ "expected Unsupported for Diff audit plan, got {err:?}"
+ );
+}
diff --git a/crates/paimon/tests/common/incremental_helpers.rs
b/crates/paimon/tests/common/incremental_helpers.rs
index e4c1abf..9ef5f62 100644
--- a/crates/paimon/tests/common/incremental_helpers.rs
+++ b/crates/paimon/tests/common/incremental_helpers.rs
@@ -17,7 +17,7 @@
//! Minimal helpers for batch incremental scan tests (no compact/lookup APIs).
-use arrow_array::{Int32Array, RecordBatch, StringArray};
+use arrow_array::{Int32Array, Int8Array, RecordBatch, StringArray};
use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
use paimon::catalog::Identifier;
use paimon::io::FileIOBuilder;
@@ -108,6 +108,26 @@ pub fn make_batch(ids: Vec<i32>, values: Vec<i32>) ->
RecordBatch {
.unwrap()
}
+/// Batch with explicit `_VALUE_KIND` for `changelog-producer=input`.
+///
+/// Kind codes: 0=+I, 1=-U, 2=+U, 3=-D.
+pub fn make_batch_with_kinds(ids: Vec<i32>, values: Vec<i32>, kinds: Vec<i8>)
-> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ArrowField::new("_VALUE_KIND", ArrowDataType::Int8, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ Arc::new(Int8Array::from(kinds)),
+ ],
+ )
+ .unwrap()
+}
+
pub fn make_partitioned_batch(pts: Vec<&str>, ids: Vec<i32>, values: Vec<i32>)
-> RecordBatch {
let schema = Arc::new(ArrowSchema::new(vec![
ArrowField::new("pt", ArrowDataType::Utf8, false),
diff --git a/crates/paimon/tests/incremental_batch_scan_test.rs
b/crates/paimon/tests/incremental_batch_scan_test.rs
index 26e9fcc..c1a0a8c 100644
--- a/crates/paimon/tests/incremental_batch_scan_test.rs
+++ b/crates/paimon/tests/incremental_batch_scan_test.rs
@@ -22,8 +22,8 @@ use futures::TryStreamExt;
use paimon::table::IncrementalScanMode;
use common::incremental_helpers::{
- make_batch, make_partitioned_batch, memory_table, partitioned_pk_schema,
persist_table_schema,
- pk_schema, setup_dirs, write_batch, write_partitioned,
+ make_batch, make_batch_with_kinds, make_partitioned_batch, memory_table,
partitioned_pk_schema,
+ persist_table_schema, pk_schema, setup_dirs, write_batch,
write_partitioned,
};
fn collect_pairs(batches: &[RecordBatch]) -> Vec<(i32, i32)> {
@@ -241,10 +241,157 @@ async fn
incremental_delta_scan_applies_partition_filter_from_read_builder() {
assert_eq!(collect_pairs(&batches), vec![(1, 10)]);
}
-/// Changelog mode is reserved but not implemented in this PR.
+/// Changelog mode reads existing changelog_manifest_list data files.
#[tokio::test]
-async fn changelog_mode_is_unsupported() {
- let table_path = "memory:/incremental_batch/changelog_unsupported";
+async fn changelog_between_snapshots_reads_changelog_manifest_files() {
+ let table_path = "memory:/incremental_batch/changelog_range";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "input"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ write
+ .write_arrow_batch(&make_batch_with_kinds(vec![1, 1], vec![10, 20],
vec![0, 2]))
+ .await
+ .unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let rows = read_incremental_pairs(&table, IncrementalScanMode::Changelog,
0, 1).await;
+ assert_eq!(rows, vec![(1, 10), (1, 20)]);
+}
+
+/// Multi-snapshot changelog range is left-open / right-closed and ordered by
snapshot id.
+#[tokio::test]
+async fn changelog_multi_snapshot_range_is_ordered_and_left_open() {
+ let table_path = "memory:/incremental_batch/changelog_multi";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "input"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(&table, &make_batch_with_kinds(vec![1], vec![10],
vec![0])).await;
+ write_batch(&table, &make_batch_with_kinds(vec![2], vec![20],
vec![0])).await;
+
+ let all = read_incremental_pairs(&table, IncrementalScanMode::Changelog,
0, 2).await;
+ assert_eq!(all, vec![(1, 10), (2, 20)]);
+
+ let second_only = read_incremental_pairs(&table,
IncrementalScanMode::Changelog, 1, 2).await;
+ assert_eq!(second_only, vec![(2, 20)]);
+}
+
+/// Auto resolves to Changelog when producer is not `none`.
+#[tokio::test]
+async fn auto_uses_changelog_when_producer_is_input() {
+ let table_path = "memory:/incremental_batch/auto_changelog";
+ let (file_io, table) = memory_table(
+ table_path,
+ pk_schema(&[
+ ("changelog-producer", "input"),
+ ("merge-engine", "deduplicate"),
+ ("bucket", "1"),
+ ]),
+ );
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_batch(
+ &table,
+ &make_batch_with_kinds(vec![1, 1], vec![10, 20], vec![0, 2]),
+ )
+ .await;
+
+ let plan = plan_incremental(&table, IncrementalScanMode::Auto, 0, 1)
+ .await
+ .unwrap();
+ assert_eq!(plan.mode(), IncrementalScanMode::Changelog);
+
+ let auto = read_incremental_pairs(&table, IncrementalScanMode::Auto, 0,
1).await;
+ let changelog = read_incremental_pairs(&table,
IncrementalScanMode::Changelog, 0, 1).await;
+ assert_eq!(auto, changelog);
+ assert_eq!(auto, vec![(1, 10), (1, 20)]);
+}
+
+/// Partition filter from ReadBuilder is pushed into the changelog plan path.
+#[tokio::test]
+async fn
incremental_changelog_scan_applies_partition_filter_from_read_builder() {
+ use paimon::spec::{Datum, PredicateBuilder};
+ use std::collections::HashMap;
+
+ let table_path = "memory:/incremental_batch/changelog_partition_filter";
+ let (file_io, mut table) = memory_table(table_path,
partitioned_pk_schema("1"));
+ table = table.copy_with_options(HashMap::from([(
+ "changelog-producer".to_string(),
+ "input".to_string(),
+ )]));
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ let builder = table.new_write_builder();
+ let mut write = builder.new_write().unwrap();
+ // Two partitions in one commit → one snapshot with both changelog files.
+ let schema = std::sync::Arc::new(arrow_schema::Schema::new(vec![
+ arrow_schema::Field::new("pt", arrow_schema::DataType::Utf8, false),
+ arrow_schema::Field::new("id", arrow_schema::DataType::Int32, false),
+ arrow_schema::Field::new("value", arrow_schema::DataType::Int32,
false),
+ arrow_schema::Field::new("_VALUE_KIND", arrow_schema::DataType::Int8,
false),
+ ]));
+ let batch = RecordBatch::try_new(
+ schema,
+ vec![
+ std::sync::Arc::new(arrow_array::StringArray::from(vec!["a",
"b"])),
+ std::sync::Arc::new(arrow_array::Int32Array::from(vec![1, 2])),
+ std::sync::Arc::new(arrow_array::Int32Array::from(vec![10, 20])),
+ std::sync::Arc::new(arrow_array::Int8Array::from(vec![0, 0])),
+ ],
+ )
+ .unwrap();
+ write.write_arrow_batch(&batch).await.unwrap();
+ let messages = write.prepare_commit().await.unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+
+ let filter = PredicateBuilder::new(table.schema().fields())
+ .equal("pt", Datum::String("a".to_string()))
+ .unwrap();
+ let mut builder = table.new_read_builder();
+ builder
+ .with_projection(&["id", "value"])
+ .unwrap()
+ .with_filter(filter);
+ let plan = builder
+ .new_incremental_scan(IncrementalScanMode::Changelog, 0, 1)
+ .plan()
+ .await
+ .unwrap();
+ let read = builder.new_read().unwrap();
+ let batches: Vec<RecordBatch> = read
+ .to_incremental_arrow(&plan)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+
+ assert_eq!(collect_pairs(&batches), vec![(1, 10)]);
+}
+
+/// Diff mode remains unsupported in this PR.
+#[tokio::test]
+async fn diff_mode_is_unsupported() {
+ let table_path = "memory:/incremental_batch/diff_unsupported";
let (file_io, table) = memory_table(
table_path,
pk_schema(&[
@@ -256,12 +403,14 @@ async fn changelog_mode_is_unsupported() {
setup_dirs(&file_io, table_path).await;
persist_table_schema(&file_io, table_path, table.schema()).await;
write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ write_batch(&table, &make_batch(vec![2], vec![20])).await;
- let err = plan_incremental(&table, IncrementalScanMode::Changelog, 0, 1)
+ // Non-empty range so planning reaches plan_diff (empty range
short-circuits).
+ let err = plan_incremental(&table, IncrementalScanMode::Diff, 1, 2)
.await
.unwrap_err();
assert!(
matches!(err, paimon::Error::Unsupported { .. }),
- "expected Unsupported for Changelog, got {err:?}"
+ "expected Unsupported for Diff, got {err:?}"
);
}
