From b560efbfc75c0931deda9fd0b240a57060a9a116 Mon Sep 17 00:00:00 2001 From: pandas886 <123344357+pandas886@users.noreply.github.com> Date: Sat, 11 Jul 2026 22:28:19 +0800 Subject: [PATCH] feat: support batch incremental delta scans --- crates/paimon/src/lib.rs | 7 +- .../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 +++++++++- .../tests/common/incremental_helpers.rs | 142 ++++++++++ crates/paimon/tests/common/mod.rs | 20 ++ .../tests/incremental_batch_scan_test.rs | 267 ++++++++++++++++++ 10 files changed, 885 insertions(+), 6 deletions(-) create mode 100644 crates/paimon/src/table/incremental_scan.rs create mode 100644 crates/paimon/tests/common/incremental_helpers.rs create mode 100644 crates/paimon/tests/common/mod.rs create mode 100644 crates/paimon/tests/incremental_batch_scan_test.rs diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs index d9cd2900..ddf207d3 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 06acf51d..f44d367b 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::>(); 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 00000000..8088d827 --- /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, + after: Vec, + }, +} + +/// Planned incremental scan: resolved mode plus splits. +#[derive(Debug, Clone)] +pub struct IncrementalPlan { + mode: IncrementalScanMode, + splits: Vec, +} + +impl IncrementalPlan { + pub fn new(mode: IncrementalScanMode, splits: Vec) -> 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 { + 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 { + 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 { + 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 { + 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 { + 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 96e9e434..e454e985 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 8bca5a73..a0fc6d77 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> { match &self.0 { diff --git a/crates/paimon/src/table/table_read.rs b/crates/paimon/src/table/table_read.rs index 0b8e5505..f9e31ca2 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 { + 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 { + 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 { 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 58338abc..6653423a 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 { + 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) -> Vec { 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 { + 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> { + 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 = 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::>() + }; + 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>> = 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>, mut trace: Option<&mut ScanTrace>, + ) -> crate::Result { + 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, + data_evolution_read_field_ids: Option<&HashSet>, + mut trace: Option<&mut ScanTrace>, ) -> crate::Result { 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 00000000..e4c1abf9 --- /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, values: Vec) -> 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, values: Vec) -> 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 00000000..a26f7984 --- /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 00000000..26e9fccb --- /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::() + .unwrap(); + let values = batch + .column(1) + .as_any() + .downcast_ref::() + .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 = 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 { + 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 = 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:?}" + ); +}