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 2760b14 feat: support batch incremental delta scans (#508)
2760b14 is described below
commit 2760b149ea644951ea5a05342e6269a5d0f5bbf8
Author: Pandas <[email protected]>
AuthorDate: Mon Jul 13 13:08:54 2026 +0800
feat: support batch incremental delta scans (#508)
---
crates/paimon/src/lib.rs | 7 +-
crates/paimon/src/table/format_read_builder.rs | 4 +
crates/paimon/src/table/incremental_scan.rs | 219 +++++++++++++++++
crates/paimon/src/table/mod.rs | 4 +
crates/paimon/src/table/read_builder.rs | 27 +++
crates/paimon/src/table/table_read.rs | 45 ++++
crates/paimon/src/table/table_scan.rs | 156 +++++++++++-
crates/paimon/tests/common/incremental_helpers.rs | 142 +++++++++++
crates/paimon/tests/common/mod.rs | 20 ++
crates/paimon/tests/incremental_batch_scan_test.rs | 267 +++++++++++++++++++++
10 files changed, 885 insertions(+), 6 deletions(-)
diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs
index d9cd290..ddf207d 100644
--- a/crates/paimon/src/lib.rs
+++ b/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::{
diff --git a/crates/paimon/src/table/format_read_builder.rs
b/crates/paimon/src/table/format_read_builder.rs
index 06acf51..f44d367 100644
--- a/crates/paimon/src/table/format_read_builder.rs
+++ b/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(
diff --git a/crates/paimon/src/table/incremental_scan.rs
b/crates/paimon/src/table/incremental_scan.rs
new file mode 100644
index 0000000..8088d82
--- /dev/null
+++ b/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(),
+ })
+ }
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 96e9e43..e454e98 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/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;
diff --git a/crates/paimon/src/table/read_builder.rs
b/crates/paimon/src/table/read_builder.rs
index 8bca5a7..a0fc6d7 100644
--- a/crates/paimon/src/table/read_builder.rs
+++ b/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 {
diff --git a/crates/paimon/src/table/table_read.rs
b/crates/paimon/src/table/table_read.rs
index 0b8e550..f9e31ca 100644
--- a/crates/paimon/src/table/table_read.rs
+++ b/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();
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 58338ab..6653423 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -706,6 +706,16 @@ impl<'a> TableScan<'a> {
}
}
+ /// Plan data splits from a snapshot's delta manifest list only.
+ pub(crate) async fn plan_snapshot_delta(&self, snapshot: &Snapshot) ->
crate::Result<Plan> {
+ match &self.0 {
+ TableScanKind::Paimon(scan) =>
scan.plan_snapshot_delta(snapshot).await,
+ TableScanKind::Format(_) => Err(crate::Error::Unsupported {
+ message: "Format tables do not support incremental delta
scan".to_string(),
+ }),
+ }
+ }
+
#[cfg(test)]
fn apply_limit_pushdown(&self, splits: Vec<DataSplit>) -> Vec<DataSplit> {
match &self.0 {
@@ -1048,11 +1058,154 @@ impl<'a> PaimonTableScan<'a> {
}
}
+ /// Plan data splits from a snapshot's delta manifest list (APPEND deltas).
+ ///
+ /// Reuses the same split-building path as a full snapshot plan, but only
+ /// reads the delta manifest list and keeps ADD entries.
+ pub(crate) async fn plan_snapshot_delta(&self, snapshot: &Snapshot) ->
crate::Result<Plan> {
+ self.ensure_query_auth_allowed()?;
+ let entries = self
+ .plan_manifest_list_entries(snapshot.delta_manifest_list())
+ .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(
+ &self,
+ manifest_list_name: &str,
+ ) -> crate::Result<Vec<ManifestEntry>> {
+ let file_io = self.table.file_io();
+ let table_path = self.table.location();
+ let core_options = CoreOptions::new(self.table.schema().options());
+ let has_primary_keys = !self.table.schema().primary_keys().is_empty();
+ let partition_fields = self.table.schema().partition_fields();
+
+ let mut manifest_metas =
+ read_manifest_list(file_io, table_path, manifest_list_name).await?;
+
+ if let Some(pf) = self.partition_filter.as_ref() {
+ if !partition_fields.is_empty() {
+ manifest_metas.retain(|meta| {
+ let stats = meta.partition_stats();
+ let min_values =
BinaryRow::from_serialized_bytes(stats.min_values()).ok();
+ let max_values =
BinaryRow::from_serialized_bytes(stats.max_values()).ok();
+ let null_counts = stats.null_counts().clone();
+ let file_stats = FileStatsRows::for_manifest_partition(
+ meta.num_added_files() + meta.num_deleted_files(),
+ min_values,
+ max_values,
+ null_counts,
+ );
+ pf.matches_manifest(&file_stats, &partition_fields)
+ });
+ }
+ }
+
+ let bucket_key_fields: Vec<DataField> = if
self.bucket_predicate.is_none() {
+ Vec::new()
+ } else {
+ let bucket_keys = core_options.bucket_key().unwrap_or_else(|| {
+ if has_primary_keys {
+ self.table.schema().trimmed_primary_keys()
+ } else {
+ Vec::new()
+ }
+ });
+ bucket_keys
+ .iter()
+ .filter_map(|key| {
+ self.table
+ .schema()
+ .fields()
+ .iter()
+ .find(|f| f.name() == key)
+ .cloned()
+ })
+ .collect::<Vec<_>>()
+ };
+ let bucket_function_type = core_options.bucket_function_type()?;
+
+ let base_path = format!("{}/{}", table_path.trim_end_matches('/'),
MANIFEST_DIR);
+ let shared_cache = SharedSchemaCache::new();
+ let partition_filter = self.partition_filter.as_ref();
+ let bucket_predicate = self.bucket_predicate.as_ref();
+ let scan_all_files = self.scan_all_files;
+
+ let mut entries = Vec::new();
+ for meta in manifest_metas {
+ let path = format!("{base_path}/{}", meta.file_name());
+ let input = file_io.new_input(&path)?;
+ let bytes = input.read().await?;
+ let mut bucket_cache: HashMap<i32, Option<HashSet<i32>>> =
HashMap::new();
+ let manifest_entries =
crate::spec::avro::from_manifest_bytes_filtered_shared(
+ &bytes,
+ &shared_cache,
+ &mut |_kind, partition_bytes, bucket, total_buckets| {
+ if has_primary_keys && !scan_all_files && bucket < 0 {
+ return false;
+ }
+ if let Some(pred) = bucket_predicate {
+ let targets =
bucket_cache.entry(total_buckets).or_insert_with(|| {
+ compute_target_buckets(
+ pred,
+ &bucket_key_fields,
+ bucket_function_type,
+ total_buckets,
+ )
+ });
+ if let Some(targets) = targets {
+ if !targets.contains(&bucket) {
+ return false;
+ }
+ }
+ }
+ if let Some(pf) = partition_filter {
+ match pf.matches_entry(partition_bytes) {
+ Ok(false) => return false,
+ Ok(true) => {}
+ Err(_) => {}
+ }
+ }
+ true
+ },
+ )?;
+ entries.extend(manifest_entries);
+ }
+ let entries = entries
+ .into_iter()
+ .filter(|entry| *entry.kind() == FileKind::Add)
+ .collect();
+ Ok(entries)
+ }
+
async fn plan_snapshot(
&self,
snapshot: Snapshot,
data_evolution_read_field_ids: Option<&HashSet<i32>>,
mut trace: Option<&mut ScanTrace>,
+ ) -> crate::Result<Plan> {
+ let entries = self
+ .plan_manifest_entries_with_trace(&snapshot, trace.as_deref_mut())
+ .await?;
+ self.plan_snapshot_from_entries(snapshot, entries,
data_evolution_read_field_ids, trace)
+ .await
+ }
+
+ async fn plan_snapshot_from_entries(
+ &self,
+ snapshot: Snapshot,
+ entries: Vec<ManifestEntry>,
+ data_evolution_read_field_ids: Option<&HashSet<i32>>,
+ mut trace: Option<&mut ScanTrace>,
) -> crate::Result<Plan> {
let file_io = self.table.file_io();
let table_path = self.table.location();
@@ -1066,9 +1219,6 @@ impl<'a> PaimonTableScan<'a> {
let open_file_cost = core_options.source_split_open_file_cost();
let partition_keys = self.table.schema().partition_keys();
- let entries = self
- .plan_manifest_entries_with_trace(&snapshot, trace.as_deref_mut())
- .await?;
if entries.is_empty() {
if let Some(trace) = trace {
trace.record_final_plan(0, 0, 0);
diff --git a/crates/paimon/tests/common/incremental_helpers.rs
b/crates/paimon/tests/common/incremental_helpers.rs
new file mode 100644
index 0000000..e4c1abf
--- /dev/null
+++ b/crates/paimon/tests/common/incremental_helpers.rs
@@ -0,0 +1,142 @@
+// 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.
+
+//! Minimal helpers for batch incremental scan tests (no compact/lookup APIs).
+
+use arrow_array::{Int32Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
+use paimon::catalog::Identifier;
+use paimon::io::FileIOBuilder;
+use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+use paimon::table::Table;
+use std::sync::Arc;
+
+pub async fn setup_dirs(file_io: &paimon::io::FileIO, table_path: &str) {
+ file_io
+ .mkdirs(&format!("{table_path}/schema"))
+ .await
+ .unwrap();
+ file_io
+ .mkdirs(&format!("{table_path}/snapshot"))
+ .await
+ .unwrap();
+}
+
+pub async fn persist_table_schema(
+ file_io: &paimon::io::FileIO,
+ table_path: &str,
+ schema: &TableSchema,
+) {
+ use bytes::Bytes;
+
+ let path = format!("{table_path}/schema/schema-{}", schema.id());
+ let json = serde_json::to_vec(schema).unwrap();
+ file_io
+ .new_output(&path)
+ .unwrap()
+ .write(Bytes::from(json))
+ .await
+ .unwrap();
+}
+
+/// Primary-key schema `(id, value)` with caller-supplied options.
+pub fn pk_schema(options: &[(&str, &str)]) -> TableSchema {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1");
+ for (k, v) in options {
+ builder = builder.option(*k, *v);
+ }
+ TableSchema::new(0, &builder.build().unwrap())
+}
+
+pub fn partitioned_pk_schema(bucket: &str) -> TableSchema {
+ let schema = Schema::builder()
+ .column("pt", DataType::VarChar(VarCharType::string_type()))
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .partition_keys(["pt"])
+ .option("bucket", bucket)
+ .option("target-file-size", "1b")
+ .option("num-sorted-run.compaction-trigger", "2")
+ .build()
+ .unwrap();
+ TableSchema::new(0, &schema)
+}
+
+pub fn memory_table(path: &str, schema: TableSchema) -> (paimon::io::FileIO,
Table) {
+ let file_io = FileIOBuilder::new("memory").build().unwrap();
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "incremental_test"),
+ path.to_string(),
+ schema,
+ None,
+ );
+ (file_io, table)
+}
+
+pub fn make_batch(ids: Vec<i32>, values: Vec<i32>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .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),
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("value", ArrowDataType::Int32, false),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(pts)),
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(Int32Array::from(values)),
+ ],
+ )
+ .unwrap()
+}
+
+pub async fn write_batch(table: &Table, batch: &RecordBatch) {
+ let builder = table.new_write_builder();
+ let mut w = builder.new_write().unwrap();
+ w.write_arrow_batch(batch).await.unwrap();
+ let msgs = w.prepare_commit().await.unwrap();
+ builder.new_commit().commit(msgs).await.unwrap();
+}
+
+pub async fn write_partitioned(table: &Table, batch: RecordBatch) {
+ let builder = table.new_write_builder();
+ let mut w = builder.new_write().unwrap();
+ w.write_arrow_batch(&batch).await.unwrap();
+ let msgs = w.prepare_commit().await.unwrap();
+ builder.new_commit().commit(msgs).await.unwrap();
+}
diff --git a/crates/paimon/tests/common/mod.rs
b/crates/paimon/tests/common/mod.rs
new file mode 100644
index 0000000..a26f798
--- /dev/null
+++ b/crates/paimon/tests/common/mod.rs
@@ -0,0 +1,20 @@
+// 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.
+
+#![allow(dead_code)]
+
+pub mod incremental_helpers;
diff --git a/crates/paimon/tests/incremental_batch_scan_test.rs
b/crates/paimon/tests/incremental_batch_scan_test.rs
new file mode 100644
index 0000000..26e9fcc
--- /dev/null
+++ b/crates/paimon/tests/incremental_batch_scan_test.rs
@@ -0,0 +1,267 @@
+// 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, RecordBatch};
+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,
+};
+
+fn collect_pairs(batches: &[RecordBatch]) -> Vec<(i32, i32)> {
+ let mut rows = Vec::new();
+ for batch in batches {
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let values = batch
+ .column(1)
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ for row in 0..batch.num_rows() {
+ rows.push((ids.value(row), values.value(row)));
+ }
+ }
+ rows.sort_unstable();
+ rows
+}
+
+async fn read_incremental_pairs(
+ table: &paimon::table::Table,
+ mode: IncrementalScanMode,
+ start_exclusive: i64,
+ end_inclusive: i64,
+) -> Vec<(i32, i32)> {
+ let builder = table.new_read_builder();
+ let plan = builder
+ .new_incremental_scan(mode, start_exclusive, end_inclusive)
+ .plan()
+ .await
+ .unwrap();
+ let read = table.new_read_builder().new_read().unwrap();
+ let batches: Vec<RecordBatch> = read
+ .to_incremental_arrow(&plan)
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ collect_pairs(&batches)
+}
+
+async fn plan_incremental(
+ table: &paimon::table::Table,
+ mode: IncrementalScanMode,
+ start_exclusive: i64,
+ end_inclusive: i64,
+) -> Result<paimon::table::IncrementalPlan, paimon::Error> {
+ table
+ .new_read_builder()
+ .new_incremental_scan(mode, start_exclusive, end_inclusive)
+ .plan()
+ .await
+}
+
+/// Start exclusive / end inclusive: (0, 2] includes both appends; (1, 2] only
the second.
+#[tokio::test]
+async fn
delta_between_snapshots_reads_only_append_snapshots_in_left_open_range() {
+ let table_path = "memory:/incremental_batch/delta_range";
+ 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 rows = read_incremental_pairs(&table, IncrementalScanMode::Delta, 0,
2).await;
+ assert_eq!(rows, vec![(1, 10), (2, 20)]);
+
+ let rows = read_incremental_pairs(&table, IncrementalScanMode::Delta, 1,
2).await;
+ assert_eq!(rows, vec![(2, 20)]);
+}
+
+#[tokio::test]
+async fn auto_uses_delta_when_changelog_producer_is_none() {
+ let table_path = "memory:/incremental_batch/auto_delta";
+ 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![3], vec![30])).await;
+ write_batch(&table, &make_batch(vec![4], vec![40])).await;
+
+ let delta = read_incremental_pairs(&table, IncrementalScanMode::Delta, 0,
2).await;
+ let auto = read_incremental_pairs(&table, IncrementalScanMode::Auto, 0,
2).await;
+ assert_eq!(auto, delta);
+}
+
+/// Empty range (start == end) yields no splits / no rows.
+#[tokio::test]
+async fn delta_empty_range_returns_no_rows() {
+ let table_path = "memory:/incremental_batch/delta_empty";
+ 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;
+
+ let plan = plan_incremental(&table, IncrementalScanMode::Delta, 1, 1)
+ .await
+ .unwrap();
+ assert!(plan.splits().is_empty());
+ assert_eq!(plan.mode(), IncrementalScanMode::Delta);
+
+ let rows = read_incremental_pairs(&table, IncrementalScanMode::Delta, 1,
1).await;
+ assert!(rows.is_empty());
+}
+
+/// Out-of-bounds ranges fail loudly with DataInvalid.
+#[tokio::test]
+async fn delta_rejects_out_of_range_snapshot_ids() {
+ let table_path = "memory:/incremental_batch/delta_oob";
+ 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;
+
+ // end past latest
+ let err = plan_incremental(&table, IncrementalScanMode::Delta, 0, 99)
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::DataInvalid { .. }),
+ "expected DataInvalid for end > latest, got {err:?}"
+ );
+
+ // start below earliest - 1 (earliest=1, min_start=0)
+ let err = plan_incremental(&table, IncrementalScanMode::Delta, -2, 1)
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::DataInvalid { .. }),
+ "expected DataInvalid for start < earliest-1, got {err:?}"
+ );
+
+ // start > end
+ let err = plan_incremental(&table, IncrementalScanMode::Delta, 2, 1)
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::DataInvalid { .. }),
+ "expected DataInvalid for start > end, got {err:?}"
+ );
+}
+
+/// Partition filter from ReadBuilder is pushed into the delta plan path.
+#[tokio::test]
+async fn incremental_delta_scan_applies_partition_filter_from_read_builder() {
+ use paimon::spec::{Datum, PredicateBuilder};
+ use std::collections::HashMap;
+
+ let table_path = "memory:/incremental_batch/delta_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(),
+ "none".to_string(),
+ )]));
+ setup_dirs(&file_io, table_path).await;
+ persist_table_schema(&file_io, table_path, table.schema()).await;
+
+ write_partitioned(&table, make_partitioned_batch(vec!["a"], vec![1],
vec![10])).await;
+ write_partitioned(&table, make_partitioned_batch(vec!["b"], vec![2],
vec![20])).await;
+
+ 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::Delta, 0, 2)
+ .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)]);
+}
+
+/// Changelog mode is reserved but not implemented in this PR.
+#[tokio::test]
+async fn changelog_mode_is_unsupported() {
+ let table_path = "memory:/incremental_batch/changelog_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;
+
+ let err = plan_incremental(&table, IncrementalScanMode::Changelog, 0, 1)
+ .await
+ .unwrap_err();
+ assert!(
+ matches!(err, paimon::Error::Unsupported { .. }),
+ "expected Unsupported for Changelog, got {err:?}"
+ );
+}
