diff --git a/crates/integrations/datafusion/src/incremental_query.rs b/crates/integrations/datafusion/src/incremental_query.rs new file mode 100644 index 00000000..d21f4605 --- /dev/null +++ b/crates/integrations/datafusion/src/incremental_query.rs @@ -0,0 +1,217 @@ +// 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. + +//! `paimon_incremental_query` table-valued function for DataFusion. +//! +//! Usage: +//! ```sql +//! SELECT * FROM paimon_incremental_query('table_name', start_snapshot_exclusive, end_snapshot_inclusive); +//! SELECT * FROM paimon_incremental_query('table_name$audit_log', 0, 2, 'diff'); +//! ``` +//! +//! This module only parses SQL arguments and adapts the resulting +//! [`IncrementalPlan`](paimon::table::IncrementalPlan) into a DataFusion +//! execution plan. Snapshot planning lives in paimon-core +//! (`IncrementalScan` / `AuditLogTable`). + +use std::fmt::Debug; +use std::sync::Arc; + +use async_trait::async_trait; +use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef; +use datafusion::catalog::Session; +use datafusion::catalog::TableFunctionImpl; +use datafusion::common::project_schema; +use datafusion::datasource::{TableProvider, TableType}; +use datafusion::error::Result as DFResult; +use datafusion::logical_expr::{Expr, TableProviderFilterPushDown}; +use datafusion::physical_plan::ExecutionPlan; +use datafusion::prelude::SessionContext; +use paimon::catalog::Catalog; +use paimon::table::{AuditLogTable, IncrementalScanMode, Table}; + +use crate::error::to_datafusion_error; +use crate::physical_plan::PaimonIncrementalScan; +use crate::runtime::{await_with_runtime, block_on_with_runtime}; +use crate::table::PaimonTableProvider; +use crate::table_function_args::{ + extract_snapshot_bound_literal, extract_string_literal, parse_incremental_scan_mode, + parse_incremental_table_ref, +}; + +const FUNCTION_NAME: &str = "paimon_incremental_query"; + +/// Register the `paimon_incremental_query` table-valued function on a [`SessionContext`]. +pub fn register_incremental_query( + ctx: &SessionContext, + catalog: Arc, + default_database: &str, +) { + ctx.register_udtf( + FUNCTION_NAME, + Arc::new(IncrementalQueryFunction::new(catalog, default_database)), + ); +} + +/// Table function that performs batch incremental reads between snapshot IDs. +/// +/// Arguments: +/// `(table_name STRING, start_snapshot_exclusive BIGINT, end_snapshot_inclusive BIGINT [, scan_mode STRING])` +pub struct IncrementalQueryFunction { + catalog: Arc, + default_database: String, +} + +impl Debug for IncrementalQueryFunction { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("IncrementalQueryFunction") + .field("default_database", &self.default_database) + .finish() + } +} + +impl IncrementalQueryFunction { + pub fn new(catalog: Arc, default_database: &str) -> Self { + Self { + catalog, + default_database: default_database.to_string(), + } + } +} + +impl TableFunctionImpl for IncrementalQueryFunction { + fn call(&self, args: &[Expr]) -> DFResult> { + if args.len() != 3 && args.len() != 4 { + return Err(datafusion::error::DataFusionError::Plan(format!( + "{FUNCTION_NAME} requires 3 or 4 arguments: \ + (table_name, start_snapshot_exclusive, end_snapshot_inclusive [, scan_mode])" + ))); + } + + let table_name = extract_string_literal(FUNCTION_NAME, &args[0], "table_name")?; + let start_exclusive = + extract_snapshot_bound_literal(FUNCTION_NAME, &args[1], "start_snapshot_exclusive")?; + let end_inclusive = + extract_snapshot_bound_literal(FUNCTION_NAME, &args[2], "end_snapshot_inclusive")?; + let scan_mode = if args.len() == 4 { + parse_incremental_scan_mode(FUNCTION_NAME, &args[3])? + } else { + IncrementalScanMode::Auto + }; + + let table_ref = + parse_incremental_table_ref(FUNCTION_NAME, &table_name, &self.default_database)?; + + let catalog = Arc::clone(&self.catalog); + let identifier = table_ref.identifier.clone(); + let table = block_on_with_runtime( + async move { catalog.get_table(&identifier).await }, + "paimon_incremental_query: catalog access thread panicked", + ) + .map_err(to_datafusion_error)?; + + let schema = if table_ref.audit_log { + let fields = AuditLogTable::new(table.clone()) + .fields() + .map_err(to_datafusion_error)?; + paimon::arrow::build_target_arrow_schema(&fields).map_err(to_datafusion_error)? + } else { + PaimonTableProvider::try_new(table.clone())?.schema() + }; + + Ok(Arc::new(IncrementalQueryTableProvider { + table, + schema, + start_exclusive, + end_inclusive, + scan_mode, + audit_log: table_ref.audit_log, + })) + } +} + +#[derive(Debug)] +struct IncrementalQueryTableProvider { + table: Table, + schema: ArrowSchemaRef, + start_exclusive: i64, + end_inclusive: i64, + scan_mode: IncrementalScanMode, + audit_log: bool, +} + +#[async_trait] +impl TableProvider for IncrementalQueryTableProvider { + fn schema(&self) -> ArrowSchemaRef { + self.schema.clone() + } + + fn table_type(&self) -> TableType { + TableType::Base + } + + async fn scan( + &self, + _state: &dyn Session, + projection: Option<&Vec>, + _filters: &[Expr], + _limit: Option, + ) -> DFResult> { + let table = self.table.clone(); + let start_exclusive = self.start_exclusive; + let end_inclusive = self.end_inclusive; + let scan_mode = self.scan_mode; + + // Planning is delegated entirely to paimon-core IncrementalScan. + let incremental_plan = await_with_runtime(async move { + table + .new_read_builder() + .new_incremental_scan(scan_mode, start_exclusive, end_inclusive) + .plan() + .await + .map_err(to_datafusion_error) + }) + .await?; + + let projected_schema = project_schema(&self.schema, projection)?; + let projected_columns = projection.map(|indices| { + indices + .iter() + .map(|&i| self.schema.field(i).name().clone()) + .collect::>() + }); + + Ok(Arc::new(PaimonIncrementalScan::new( + projected_schema, + self.table.clone(), + incremental_plan, + self.audit_log, + projected_columns, + ))) + } + + fn supports_filters_pushdown( + &self, + filters: &[&Expr], + ) -> DFResult> { + // Residual filter evaluation is performed by DataFusion above this plan. + Ok(vec![ + TableProviderFilterPushDown::Unsupported; + filters.len() + ]) + } +} diff --git a/crates/integrations/datafusion/src/lib.rs b/crates/integrations/datafusion/src/lib.rs index a60d52b2..dac060ea 100644 --- a/crates/integrations/datafusion/src/lib.rs +++ b/crates/integrations/datafusion/src/lib.rs @@ -45,6 +45,7 @@ mod filter_pushdown; #[cfg(feature = "fulltext")] mod full_text_search; mod hybrid_search; +mod incremental_query; mod lateral_vector_search; mod merge_into; mod physical_plan; @@ -78,6 +79,7 @@ pub use error::to_datafusion_error; #[cfg(feature = "fulltext")] pub use full_text_search::{register_full_text_search, FullTextSearchFunction}; pub use hybrid_search::{register_hybrid_search, HybridSearchFunction}; +pub use incremental_query::{register_incremental_query, IncrementalQueryFunction}; pub use physical_plan::PaimonTableScan; pub use relation_planner::PaimonRelationPlanner; pub use sql_context::SQLContext; diff --git a/crates/integrations/datafusion/src/physical_plan/incremental_scan.rs b/crates/integrations/datafusion/src/physical_plan/incremental_scan.rs new file mode 100644 index 00000000..b3103876 --- /dev/null +++ b/crates/integrations/datafusion/src/physical_plan/incremental_scan.rs @@ -0,0 +1,187 @@ +// 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. + +//! Physical plan for batch incremental reads via paimon-core +//! [`IncrementalScan`](paimon::table::IncrementalScan) / +//! [`AuditLogTable`](paimon::table::AuditLogTable). + +use std::sync::Arc; + +use datafusion::arrow::datatypes::SchemaRef as ArrowSchemaRef; +use datafusion::error::Result as DFResult; +use datafusion::execution::{SendableRecordBatchStream, TaskContext}; +use datafusion::physical_expr::EquivalenceProperties; +use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType}; +use datafusion::physical_plan::stream::RecordBatchStreamAdapter; +use datafusion::physical_plan::{DisplayAs, ExecutionPlan, Partitioning, PlanProperties}; +use futures::{StreamExt, TryStreamExt}; +use paimon::table::{AuditLogTable, IncrementalPlan, Table}; + +use crate::error::to_datafusion_error; + +pub(crate) const PLAN_NAME: &str = "PaimonIncrementalScan"; + +/// Execution plan that streams incremental snapshot-range rows from a Paimon table. +/// +/// Planning of snapshot ranges is performed eagerly in +/// [`crate::incremental_query::IncrementalQueryFunction`]; this plan only executes +/// the precomputed [`IncrementalPlan`]. +#[derive(Debug)] +pub(crate) struct PaimonIncrementalScan { + table: Table, + incremental_plan: IncrementalPlan, + audit_log: bool, + projected_columns: Option>, + schema: ArrowSchemaRef, + plan_properties: Arc, +} + +impl PaimonIncrementalScan { + pub(crate) fn new( + schema: ArrowSchemaRef, + table: Table, + incremental_plan: IncrementalPlan, + audit_log: bool, + projected_columns: Option>, + ) -> Self { + let plan_properties = Arc::new(PlanProperties::new( + EquivalenceProperties::new(schema.clone()), + Partitioning::UnknownPartitioning(1), + EmissionType::Incremental, + Boundedness::Bounded, + )); + Self { + table, + incremental_plan, + audit_log, + projected_columns, + schema, + plan_properties, + } + } +} + +impl ExecutionPlan for PaimonIncrementalScan { + fn name(&self) -> &str { + PLAN_NAME + } + + fn properties(&self) -> &Arc { + &self.plan_properties + } + + fn children(&self) -> Vec<&Arc> { + vec![] + } + + fn with_new_children( + self: Arc, + _children: Vec>, + ) -> DFResult> { + Ok(self) + } + + fn execute( + &self, + _partition: usize, + _context: Arc, + ) -> DFResult { + let table = self.table.clone(); + let plan = self.incremental_plan.clone(); + let audit_log = self.audit_log; + let projected_columns = self.projected_columns.clone(); + let schema = self.schema.clone(); + + let fut = async move { + let stream = if audit_log { + // AuditLogTable always materialises the full audit schema + // (rowkind + table fields). Column projection is applied below. + AuditLogTable::new(table) + .to_arrow(&plan) + .map_err(to_datafusion_error)? + } else { + let mut read_builder = table.new_read_builder(); + if let Some(ref columns) = projected_columns { + let col_refs: Vec<&str> = columns.iter().map(|s| s.as_str()).collect(); + read_builder + .with_projection(&col_refs) + .map_err(to_datafusion_error)?; + } + let read = read_builder.new_read().map_err(to_datafusion_error)?; + read.to_incremental_arrow(&plan) + .map_err(to_datafusion_error)? + }; + + let stream = stream.map(move |result| { + let batch = result.map_err(to_datafusion_error)?; + if audit_log { + if let Some(ref columns) = projected_columns { + let indices = columns + .iter() + .map(|name| { + batch.schema().index_of(name).map_err(|e| { + datafusion::error::DataFusionError::Internal(format!( + "PaimonIncrementalScan: projection column '{name}': {e}" + )) + }) + }) + .collect::>>()?; + return batch.project(&indices).map_err(|e| { + datafusion::error::DataFusionError::ArrowError(Box::new(e), None) + }); + } + } + Ok(batch) + }); + + Ok::<_, datafusion::error::DataFusionError>(RecordBatchStreamAdapter::new( + schema, + Box::pin(stream), + )) + }; + + Ok(Box::pin(RecordBatchStreamAdapter::new( + self.schema.clone(), + futures::stream::once(fut).try_flatten(), + ))) + } +} + +impl DisplayAs for PaimonIncrementalScan { + fn fmt_as( + &self, + _format: datafusion::physical_plan::DisplayFormatType, + f: &mut std::fmt::Formatter, + ) -> std::fmt::Result { + write!( + f, + "PaimonIncrementalScan(audit_log={}, splits={})", + self.audit_log, + self.incremental_plan.splits().len() + ) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn incremental_scan_plan_name_constant() { + assert_eq!(PLAN_NAME, "PaimonIncrementalScan"); + } +} diff --git a/crates/integrations/datafusion/src/physical_plan/mod.rs b/crates/integrations/datafusion/src/physical_plan/mod.rs index 2fa35bf1..aadee94d 100644 --- a/crates/integrations/datafusion/src/physical_plan/mod.rs +++ b/crates/integrations/datafusion/src/physical_plan/mod.rs @@ -15,8 +15,10 @@ // specific language governing permissions and limitations // under the License. +pub(crate) mod incremental_scan; pub(crate) mod scan; pub(crate) mod sink; +pub(crate) use incremental_scan::PaimonIncrementalScan; pub use scan::PaimonTableScan; pub use sink::PaimonDataSink; diff --git a/crates/integrations/datafusion/src/sql_context.rs b/crates/integrations/datafusion/src/sql_context.rs index ec720134..08c65617 100644 --- a/crates/integrations/datafusion/src/sql_context.rs +++ b/crates/integrations/datafusion/src/sql_context.rs @@ -153,11 +153,11 @@ impl SQLContext { /// /// `default_db = Some("")` is rejected — pass `None` to opt out instead. /// - /// **Note on built-in TVFs (`vector_search`, `full_text_search`):** when - /// `default_db = None`, bare table names inside these functions still resolve - /// against the literal namespace `"default"` (the fallback in - /// [`register_table_functions`]). Callers using `None` must qualify table names - /// (`'db.table'` or `'catalog.db.table'`) in those calls. + /// **Note on built-in TVFs (`paimon_incremental_query`, `vector_search`, + /// `full_text_search`):** when `default_db = None`, bare table names inside + /// these functions still resolve against the literal namespace `"default"` + /// (the fallback in [`register_table_functions`]). Callers using `None` must + /// qualify table names (`'db.table'` or `'catalog.db.table'`) in those calls. pub async fn register_catalog_with_default_db( &mut self, catalog_name: impl Into, @@ -3210,6 +3210,11 @@ fn register_table_functions( default_database: &str, ) { crate::blob_view::register_blob_view(ctx, Arc::clone(catalog), default_database); + crate::incremental_query::register_incremental_query( + ctx, + Arc::clone(catalog), + default_database, + ); crate::vector_search::register_vector_search(ctx, Arc::clone(catalog), default_database); #[cfg(feature = "fulltext")] crate::full_text_search::register_full_text_search(ctx, Arc::clone(catalog), default_database); @@ -5202,9 +5207,10 @@ mod tests { async fn register_catalog_with_none_table_function_resolves_bare_name_to_literal_default() { // Documents the fallback in register_table_functions: `default_db.unwrap_or("default")`. // When the caller opts out of default-db init, bare table names inside built-in TVFs - // (vector_search / full_text_search) still resolve against the literal namespace - // `"default"` — so a caller using `None` MUST use fully-qualified names with these - // functions or they'll hit a `default.` lookup that may not exist / be readable. + // (paimon_incremental_query / vector_search / full_text_search) still resolve + // against the literal namespace `"default"` — so a caller using `None` MUST use + // fully-qualified names with these functions or they'll hit a `default.` + // lookup that may not exist / be readable. let catalog = Arc::new(ProbeTrackingCatalog::new()); let mut ctx = SQLContext::new(); ctx.register_catalog_with_default_db("paimon", catalog, None) diff --git a/crates/integrations/datafusion/src/table_function_args.rs b/crates/integrations/datafusion/src/table_function_args.rs index c0a6c833..d637dfab 100644 --- a/crates/integrations/datafusion/src/table_function_args.rs +++ b/crates/integrations/datafusion/src/table_function_args.rs @@ -18,7 +18,109 @@ use datafusion::common::ScalarValue; use datafusion::error::{DataFusionError, Result as DFResult}; use datafusion::logical_expr::Expr; -use paimon::catalog::Identifier; +use paimon::catalog::{Identifier, SYSTEM_TABLE_SPLITTER}; +use paimon::table::IncrementalScanMode; + +/// Parsed target of `paimon_incremental_query`: base table plus optional `$audit_log`. +#[derive(Debug, Clone)] +pub(crate) struct IncrementalTableRef { + pub identifier: Identifier, + pub audit_log: bool, +} + +/// Parse table name for incremental query TVF. +/// +/// Accepts `table`, `database.table`, or `catalog.database.table`, with optional +/// `$audit_log` suffix on the object name. +pub(crate) fn parse_incremental_table_ref( + function_name: &str, + name: &str, + default_database: &str, +) -> DFResult { + let parts: Vec<&str> = name.split('.').collect(); + let (database, object_name) = match parts.len() { + 1 => (default_database, parts[0]), + 2 => (parts[0], parts[1]), + 3 => (parts[1], parts[2]), + _ => { + return Err(DataFusionError::Plan(format!( + "{function_name}: invalid table name '{name}', expected 'table', 'database.table', or 'catalog.database.table'" + ))); + } + }; + + let mut object_parts = object_name.splitn(2, SYSTEM_TABLE_SPLITTER); + let base = object_parts.next().unwrap_or(object_name); + let suffix = object_parts.next(); + let audit_log = match suffix { + None => false, + Some("audit_log") => true, + Some(other) => { + return Err(DataFusionError::Plan(format!( + "{function_name}: unsupported system table suffix '${other}' in '{name}'" + ))); + } + }; + + Ok(IncrementalTableRef { + identifier: Identifier::new(database.to_string(), base.to_string()), + audit_log, + }) +} + +/// Accept integer literals or decimal strings (Spark `'0'`, `'1'` compatibility). +pub(crate) fn extract_snapshot_bound_literal( + function_name: &str, + expr: &Expr, + name: &str, +) -> DFResult { + match expr { + Expr::Literal(scalar, _) => match scalar { + ScalarValue::Int8(Some(v)) => Ok(*v as i64), + ScalarValue::Int16(Some(v)) => Ok(*v as i64), + ScalarValue::Int32(Some(v)) => Ok(*v as i64), + ScalarValue::Int64(Some(v)) => Ok(*v), + ScalarValue::UInt8(Some(v)) => Ok(*v as i64), + ScalarValue::UInt16(Some(v)) => Ok(*v as i64), + ScalarValue::UInt32(Some(v)) => Ok(*v as i64), + ScalarValue::UInt64(Some(v)) => i64::try_from(*v).map_err(|_| { + DataFusionError::Plan(format!( + "{function_name}: {name} value {v} exceeds i64 range" + )) + }), + ScalarValue::Utf8(Some(s)) | ScalarValue::LargeUtf8(Some(s)) => { + s.parse::().map_err(|_| { + DataFusionError::Plan(format!( + "{function_name}: {name} must be an integer snapshot id, got string '{s}'" + )) + }) + } + _ => Err(DataFusionError::Plan(format!( + "{function_name}: {name} must be an integer or numeric string literal, got: {expr}" + ))), + }, + _ => Err(DataFusionError::Plan(format!( + "{function_name}: {name} must be a literal, got: {expr}" + ))), + } +} + +/// Parse optional TVF 4th argument; aligns with Java `incremental-between-scan-mode`. +pub(crate) fn parse_incremental_scan_mode( + function_name: &str, + expr: &Expr, +) -> DFResult { + let mode = extract_string_literal(function_name, expr, "scan_mode")?; + match mode.to_ascii_lowercase().as_str() { + "auto" => Ok(IncrementalScanMode::Auto), + "delta" => Ok(IncrementalScanMode::Delta), + "changelog" => Ok(IncrementalScanMode::Changelog), + "diff" => Ok(IncrementalScanMode::Diff), + other => Err(DataFusionError::Plan(format!( + "{function_name}: unsupported scan_mode '{other}', expected auto, delta, changelog, or diff" + ))), + } +} pub(crate) fn extract_string_literal( function_name: &str, @@ -80,3 +182,79 @@ pub(crate) fn parse_table_identifier( ))), } } + +#[cfg(test)] +mod incremental_query_args_tests { + use super::*; + + #[test] + fn parse_incremental_table_ref_plain_name() { + let parsed = + parse_incremental_table_ref("paimon_incremental_query", "orders", "default").unwrap(); + assert_eq!(parsed.identifier.database(), "default"); + assert_eq!(parsed.identifier.object(), "orders"); + assert!(!parsed.audit_log); + } + + #[test] + fn parse_incremental_table_ref_with_audit_log_suffix() { + let parsed = parse_incremental_table_ref( + "paimon_incremental_query", + "paimon.test_db.orders$audit_log", + "default", + ) + .unwrap(); + assert_eq!(parsed.identifier.database(), "test_db"); + assert_eq!(parsed.identifier.object(), "orders"); + assert!(parsed.audit_log); + } + + #[test] + fn extract_snapshot_bound_accepts_int_and_numeric_string() { + let int_expr = Expr::Literal(ScalarValue::Int64(Some(2)), None); + let str_expr = Expr::Literal(ScalarValue::Utf8(Some("2".to_string())), None); + assert_eq!( + extract_snapshot_bound_literal("paimon_incremental_query", &int_expr, "end").unwrap(), + 2 + ); + assert_eq!( + extract_snapshot_bound_literal("paimon_incremental_query", &str_expr, "end").unwrap(), + 2 + ); + } + + #[test] + fn parse_incremental_table_ref_rejects_unknown_system_suffix() { + let err = + parse_incremental_table_ref("paimon_incremental_query", "orders$snapshots", "default") + .unwrap_err() + .to_string(); + assert!(err.contains("snapshots"), "unexpected error: {err}"); + } + + #[test] + fn parse_incremental_scan_mode_accepts_aliases() { + let lit = |s: &str| Expr::Literal(ScalarValue::Utf8(Some(s.to_string())), None); + assert_eq!( + parse_incremental_scan_mode("paimon_incremental_query", &lit("auto")).unwrap(), + IncrementalScanMode::Auto + ); + assert_eq!( + parse_incremental_scan_mode("paimon_incremental_query", &lit("DIFF")).unwrap(), + IncrementalScanMode::Diff + ); + assert_eq!( + parse_incremental_scan_mode("paimon_incremental_query", &lit("changelog")).unwrap(), + IncrementalScanMode::Changelog + ); + } + + #[test] + fn parse_incremental_scan_mode_rejects_unknown() { + let lit = Expr::Literal(ScalarValue::Utf8(Some("nope".to_string())), None); + let err = parse_incremental_scan_mode("paimon_incremental_query", &lit) + .unwrap_err() + .to_string(); + assert!(err.contains("nope"), "unexpected error: {err}"); + } +} diff --git a/crates/integrations/datafusion/tests/incremental_query.rs b/crates/integrations/datafusion/tests/incremental_query.rs new file mode 100644 index 00000000..f7d22d06 --- /dev/null +++ b/crates/integrations/datafusion/tests/incremental_query.rs @@ -0,0 +1,387 @@ +// 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. + +//! Integration tests for `paimon_incremental_query` TVF. +//! +//! Uses pure Rust `changelog-producer=input` tables (no compact / lookup). + +mod common; + +use std::sync::Arc; + +use arrow_array::{Array, Int32Array, Int8Array, RecordBatch}; +use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; +use datafusion::arrow::array::StringArray; +use paimon::catalog::Identifier; +use paimon::spec::VALUE_KIND_FIELD_NAME; +use paimon::{Catalog, FileSystemCatalog}; + +async fn setup_table_with_three_snapshots(sql: &paimon_datafusion::SQLContext) { + common::exec( + sql, + "CREATE TABLE paimon.test_db.inc_t ( + id INT NOT NULL, + value INT, + PRIMARY KEY (id) + ) WITH ('bucket' = '1', 'changelog-producer' = 'input')", + ) + .await; + common::exec(sql, "INSERT INTO paimon.test_db.inc_t VALUES (1, 10)").await; + common::exec(sql, "INSERT INTO paimon.test_db.inc_t VALUES (2, 20)").await; + common::exec(sql, "INSERT INTO paimon.test_db.inc_t VALUES (3, 30)").await; +} + +fn collect_id_value(batches: &[datafusion::arrow::record_batch::RecordBatch]) -> Vec<(i32, i32)> { + let mut rows = Vec::new(); + for batch in batches { + let ids = batch + .column_by_name("id") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("id column"); + let values = batch + .column_by_name("value") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("value column"); + for row in 0..batch.num_rows() { + rows.push((ids.value(row), values.value(row))); + } + } + rows.sort_unstable(); + rows +} + +fn collect_audit_rows( + batches: &[datafusion::arrow::record_batch::RecordBatch], +) -> Vec<(String, i32, i32)> { + let mut rows = Vec::new(); + for batch in batches { + let kinds = batch + .column_by_name("rowkind") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("rowkind"); + let ids = batch + .column_by_name("id") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("id"); + let values = batch + .column_by_name("value") + .and_then(|c| c.as_any().downcast_ref::()) + .expect("value"); + for row in 0..batch.num_rows() { + rows.push(( + kinds.value(row).to_string(), + ids.value(row), + values.value(row), + )); + } + } + rows.sort_unstable(); + rows +} + +/// Commit a single-row input-changelog delete (`RowKind::Delete` = 3). +async fn commit_input_changelog_delete( + catalog: &Arc, + table_name: &str, + id: i32, +) { + let table = catalog + .get_table(&Identifier::new("test_db", table_name)) + .await + .expect("table should exist"); + let schema = Arc::new(ArrowSchema::new(vec![ + ArrowField::new("id", ArrowDataType::Int32, false), + ArrowField::new(VALUE_KIND_FIELD_NAME, ArrowDataType::Int8, false), + ])); + let batch = RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![id])), + Arc::new(Int8Array::from(vec![3])), // RowKind::Delete + ], + ) + .unwrap(); + let wb = table.new_write_builder(); + let mut writer = wb.new_write().expect("writer"); + writer + .write_arrow_batch(&batch) + .await + .expect("write changelog delete"); + let messages = writer.prepare_commit().await.expect("prepare"); + wb.new_commit() + .commit(messages) + .await + .expect("commit changelog delete"); +} + +/// 3-arg form defaults to Auto (input producer → changelog semantics). +#[tokio::test] +async fn incremental_query_without_audit_log_returns_delta_rows() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let batches = ctx + .sql("SELECT id, value FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 2)") + .await + .expect("plan") + .collect() + .await + .expect("execute"); + + assert_eq!(collect_id_value(&batches), vec![(1, 10), (2, 20)]); +} + +#[tokio::test] +async fn incremental_query_with_audit_log_suffix_exposes_rowkind() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let batches = ctx + .sql( + "SELECT rowkind, id, value \ + FROM paimon_incremental_query('paimon.test_db.inc_t$audit_log', 0, 1)", + ) + .await + .expect("plan") + .collect() + .await + .expect("execute"); + + let batch = &batches[0]; + assert_eq!(batch.schema().field(0).name(), "rowkind"); + let kinds = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + assert!(kinds.iter().all(|k| k == Some("+I"))); +} + +#[tokio::test] +async fn incremental_query_explicit_auto_and_changelog_match_default() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let default_rows = ctx + .sql("SELECT id, value FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 1)") + .await + .unwrap() + .collect() + .await + .unwrap(); + let auto_rows = ctx + .sql("SELECT id, value FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 1, 'auto')") + .await + .unwrap() + .collect() + .await + .unwrap(); + let changelog_rows = ctx + .sql( + "SELECT id, value FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 1, 'changelog')", + ) + .await + .unwrap() + .collect() + .await + .unwrap(); + + assert_eq!( + collect_id_value(&default_rows), + collect_id_value(&auto_rows) + ); + assert_eq!( + collect_id_value(&default_rows), + collect_id_value(&changelog_rows) + ); +} + +#[tokio::test] +async fn incremental_query_explicit_delta_mode() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let batches = ctx + .sql( + "SELECT id, value FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 2, 'delta')", + ) + .await + .expect("plan") + .collect() + .await + .expect("execute"); + + // Delta reads data files from APPEND snapshots in (0, 2] → rows from snapshots 1 and 2. + assert_eq!(collect_id_value(&batches), vec![(1, 10), (2, 20)]); +} + +#[tokio::test] +async fn incremental_query_diff_mode_on_pk_table_with_audit_log() { + let (_tmp, ctx) = common::setup_sql_context().await; + common::exec( + &ctx, + "CREATE TABLE paimon.test_db.inc_pk ( + id INT NOT NULL, + value INT, + PRIMARY KEY (id) + ) WITH ('bucket' = '1', 'merge-engine' = 'deduplicate')", + ) + .await; + common::exec(&ctx, "INSERT INTO paimon.test_db.inc_pk VALUES (1, 10)").await; + common::exec(&ctx, "INSERT INTO paimon.test_db.inc_pk VALUES (2, 20)").await; + // PK deduplicate merge treats a later insert with the same key as an update. + common::exec(&ctx, "INSERT INTO paimon.test_db.inc_pk VALUES (1, 11)").await; + + let batches = ctx + .sql( + "SELECT rowkind, id, value \ + FROM paimon_incremental_query('paimon.test_db.inc_pk$audit_log', 1, 3, 'diff')", + ) + .await + .expect("plan") + .collect() + .await + .expect("execute"); + + let rows = collect_audit_rows(&batches); + assert!( + rows.iter() + .any(|(k, id, v)| k == "-U" && *id == 1 && *v == 10), + "diff audit_log should include -U before image, got {rows:?}" + ); + assert!( + rows.iter() + .any(|(k, id, v)| k == "+U" && *id == 1 && *v == 11), + "diff audit_log should include +U after image, got {rows:?}" + ); +} + +#[tokio::test] +async fn incremental_query_rejects_invalid_scan_mode() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let err = ctx + .sql("SELECT * FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 1, 'nope')") + .await + .expect_err("invalid scan_mode should fail at planning"); + let msg = err.to_string(); + assert!( + msg.to_lowercase().contains("scan_mode") || msg.contains("nope"), + "unexpected error: {msg}" + ); +} + +#[tokio::test] +async fn incremental_query_rejects_illegal_snapshot_range() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let err = ctx + .sql("SELECT * FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 99)") + .await + .expect("plan may succeed; execution/planning of scan fails") + .collect() + .await + .expect_err("out-of-range end snapshot should fail"); + let msg = err.to_string().to_lowercase(); + assert!( + msg.contains("snapshot") || msg.contains("99") || msg.contains("range"), + "unexpected error: {err}" + ); +} + +#[tokio::test] +async fn incremental_query_supports_projection_and_filter() { + let (_tmp, ctx) = common::setup_sql_context().await; + setup_table_with_three_snapshots(&ctx).await; + + let batches = ctx + .sql("SELECT id FROM paimon_incremental_query('paimon.test_db.inc_t', 0, 2) WHERE id = 2") + .await + .expect("plan") + .collect() + .await + .expect("execute"); + + let mut ids = Vec::new(); + for batch in &batches { + assert_eq!(batch.num_columns(), 1); + assert_eq!(batch.schema().field(0).name(), "id"); + let col = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + ids.push(col.value(row)); + } + } + assert_eq!(ids, vec![2]); +} + +#[tokio::test] +async fn incremental_query_audit_log_sees_delete_rowkind() { + let (tmp, catalog) = common::create_test_env(); + let ctx = common::create_sql_context(Arc::clone(&catalog)).await; + common::exec(&ctx, "CREATE SCHEMA paimon.test_db").await; + + common::exec( + &ctx, + "CREATE TABLE paimon.test_db.inc_del ( + id INT NOT NULL, + PRIMARY KEY (id) + ) WITH ('bucket' = '1', 'changelog-producer' = 'input')", + ) + .await; + common::exec(&ctx, "INSERT INTO paimon.test_db.inc_del VALUES (1)").await; + common::exec(&ctx, "INSERT INTO paimon.test_db.inc_del VALUES (2)").await; + commit_input_changelog_delete(&catalog, "inc_del", 1).await; + let _ = tmp; + + let batches = ctx + .sql( + "SELECT rowkind, id FROM paimon_incremental_query('paimon.test_db.inc_del$audit_log', 1, 3)", + ) + .await + .expect("plan") + .collect() + .await + .expect("execute"); + + let mut kinds = Vec::new(); + for batch in &batches { + let kind_col = batch + .column(0) + .as_any() + .downcast_ref::() + .unwrap(); + let id_col = batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + kinds.push((kind_col.value(row).to_string(), id_col.value(row))); + } + } + kinds.sort_unstable(); + assert!( + kinds.iter().any(|(k, id)| k == "-D" && *id == 1), + "audit_log incremental query should expose -D, got {kinds:?}" + ); +} 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/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index b77ab95d..8408121d 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -71,6 +71,8 @@ pub(crate) const DISABLE_ALTER_COLUMN_NULL_TO_NOT_NULL_OPTION: &str = const MERGE_ENGINE_OPTION: &str = "merge-engine"; const CHANGELOG_PRODUCER_OPTION: &str = "changelog-producer"; const ROWKIND_FIELD_OPTION: &str = "rowkind.field"; +const DIFF_PARALLELISM_OPTION: &str = "diff.parallelism"; +const DEFAULT_DIFF_PARALLELISM: usize = 4; const DEFAULT_COMMIT_MAX_RETRIES: u32 = 10; const DEFAULT_COMMIT_TIMEOUT_MS: u64 = 120_000; const DEFAULT_COMMIT_MIN_RETRY_WAIT_MS: u64 = 1_000; @@ -384,6 +386,17 @@ impl<'a> CoreOptions<'a> { self.options.get(ROWKIND_FIELD_OPTION).map(String::as_str) } + /// Parallelism for batch incremental Diff pair reads (`diff.parallelism`). + /// + /// Default is 4; values below 1 are clamped to 1. + pub fn diff_parallelism(&self) -> usize { + self.options + .get(DIFF_PARALLELISM_OPTION) + .and_then(|s| s.parse().ok()) + .unwrap_or(DEFAULT_DIFF_PARALLELISM) + .max(1) + } + pub fn data_evolution_enabled(&self) -> bool { self.options .get(DATA_EVOLUTION_ENABLED_OPTION) @@ -1257,6 +1270,21 @@ mod tests { ); } + #[test] + fn test_diff_parallelism_defaults() { + let options = HashMap::new(); + let core = CoreOptions::new(&options); + assert_eq!(core.diff_parallelism(), 4); + + let options = HashMap::from([(DIFF_PARALLELISM_OPTION.to_string(), "0".into())]); + let core = CoreOptions::new(&options); + assert_eq!(core.diff_parallelism(), 1); + + let options = HashMap::from([(DIFF_PARALLELISM_OPTION.to_string(), "8".into())]); + let core = CoreOptions::new(&options); + assert_eq!(core.diff_parallelism(), 8); + } + #[test] fn test_changelog_producer_accepts_known_values() { for (value, expected) in [ 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 00000000..718df681 --- /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, SEQUENCE_NUMBER_FIELD_ID, + SEQUENCE_NUMBER_FIELD_NAME, +}; + +/// Wrapper that exposes table rows with a leading `rowkind` audit column. +/// +/// Incremental reads produce: +/// - Delta: every row is `+I` +/// - Changelog: kinds come from physical `_VALUE_KIND` (`+I`/`-U`/`+U`/`-D`) +/// - Diff: before/after image comparison (`+I`/`-U`/`+U`/`-D`, equal keys skipped) +#[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> { + let mut fields = Vec::with_capacity(self.wrapped.schema().fields().len() + 2); + fields.push(DataField::new( + -1, + "rowkind".to_string(), + DataType::VarChar(VarCharType::new(8)?), + )); + if self.sequence_number_enabled() { + fields.push(DataField::new( + SEQUENCE_NUMBER_FIELD_ID, + SEQUENCE_NUMBER_FIELD_NAME.to_string(), + DataType::BigInt(BigIntType::new()), + )); + } + fields.extend(self.wrapped.schema().fields().iter().cloned()); + Ok(fields) + } + + fn sequence_number_enabled(&self) -> bool { + self.wrapped + .schema() + .options() + .get(TABLE_READ_SEQUENCE_NUMBER_ENABLED) + .is_some_and(|v| v.eq_ignore_ascii_case("true")) + } + + pub fn new_incremental_scan( + &self, + mode: IncrementalScanMode, + start_exclusive: i64, + end_inclusive: i64, + ) -> IncrementalScan<'_> { + IncrementalScan::for_table(&self.wrapped, mode, start_exclusive, end_inclusive) + } + + pub fn to_arrow(&self, plan: &IncrementalPlan) -> crate::Result { + let read = self.wrapped.new_read_builder().new_read()?; + read.to_audit_log_arrow(plan) + } +} 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..ea4b1fbe --- /dev/null +++ b/crates/paimon/src/table/incremental_scan.rs @@ -0,0 +1,274 @@ +// 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 existing changelog manifest files in the range. + /// + /// 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). + Auto, + /// Diff before/after snapshot states for PK tables. + /// + /// Phase 1 supports only `merge-engine=deduplicate`. Planning compares the + /// full table state at `start_exclusive` vs `end_inclusive` and yields + /// per-(partition, bucket) [`IncrementalSplit::DiffPair`] units. + 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 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 { + let core_options = CoreOptions::new(self.table.schema().options()); + if core_options.merge_engine()? != crate::spec::MergeEngine::Deduplicate { + return Err(crate::Error::Unsupported { + message: "Batch incremental Diff only supports merge-engine=deduplicate in Phase 1" + .to_string(), + }); + } + let before = self + .snapshot_manager + .get_snapshot(self.start_exclusive) + .await?; + let after = self + .snapshot_manager + .get_snapshot(self.end_inclusive) + .await?; + let (before_plan, after_plan) = self.scan.plan_snapshot_diff(&before, &after).await?; + + use std::collections::BTreeMap; + type PBKey = (Vec, i32); + + let mut before_map: BTreeMap> = BTreeMap::new(); + for split in before_plan.splits() { + let key = (split.partition().to_serialized_bytes(), split.bucket()); + before_map.entry(key).or_default().push(split.clone()); + } + + let mut after_map: BTreeMap> = BTreeMap::new(); + for split in after_plan.splits() { + let key = (split.partition().to_serialized_bytes(), split.bucket()); + after_map.entry(key).or_default().push(split.clone()); + } + + let mut keys: std::collections::BTreeSet = before_map.keys().cloned().collect(); + keys.extend(after_map.keys().cloned()); + + let mut splits = Vec::new(); + for key in keys { + let before = before_map.remove(&key).unwrap_or_default(); + let after = after_map.remove(&key).unwrap_or_default(); + if before.is_empty() && after.is_empty() { + continue; + } + splits.push(IncrementalSplit::DiffPair { before, after }); + } + + Ok(IncrementalPlan::new(mode, splits)) + } +} diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 96e9e434..41aa1eda 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; @@ -48,6 +49,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; @@ -79,6 +81,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; @@ -94,6 +97,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..834dd292 100644 --- a/crates/paimon/src/table/table_read.rs +++ b/crates/paimon/src/table/table_read.rs @@ -18,11 +18,24 @@ 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}; -use crate::spec::{CoreOptions, DataField, MergeEngine, Predicate}; +use crate::arrow::build_target_arrow_schema; +use crate::spec::{ + BigIntType, CoreOptions, DataField, DataType, MergeEngine, Predicate, TinyIntType, + SEQUENCE_NUMBER_FIELD_ID, SEQUENCE_NUMBER_FIELD_NAME, VALUE_KIND_FIELD_ID, + VALUE_KIND_FIELD_NAME, +}; use crate::DataSplit; +use arrow_array::{builder::StringBuilder, Array, ArrayRef, RecordBatch, StringArray, UInt32Array}; +use arrow_schema::Schema as ArrowSchema; +use arrow_select::concat::concat as arrow_concat; +use arrow_select::take::take; +use futures::{stream, StreamExt}; +use std::cmp::Ordering; +use std::sync::Arc; /// Table read: reads data from splits (e.g. produced by [TableScan::plan]). /// @@ -107,6 +120,40 @@ impl<'a> TableRead<'a> { TableReadKind::Format(read) => read.to_arrow(data_splits), } } + + /// Returns an [`ArrowRecordBatchStream`] for an incremental scan plan. + /// + /// Delta/Changelog use [`IncrementalSplit::Data`]. Diff uses + /// [`IncrementalSplit::DiffPair`] and emits after-image rows only. + 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(), + }), + } + } + + /// Returns an audit-log [`ArrowRecordBatchStream`] for an incremental plan. + /// + /// Output schema is `rowkind` (+ optional `_SEQUENCE_NUMBER`) followed by + /// the projected user columns. Delta rows are all `+I`; Changelog rows take + /// kinds from `_VALUE_KIND`; Diff emits `+I`/`-U`/`+U`/`-D` from before/after + /// image comparison. + pub fn to_audit_log_arrow( + &self, + plan: &IncrementalPlan, + ) -> crate::Result { + 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)] @@ -160,6 +207,399 @@ 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 self.to_incremental_diff_arrow(plan); + } + + 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) + } + + fn to_incremental_diff_arrow( + &self, + plan: &IncrementalPlan, + ) -> crate::Result { + let pairs: Vec<(Vec, Vec)> = plan + .splits() + .iter() + .filter_map(|s| match s { + IncrementalSplit::DiffPair { before, after } => { + Some((before.clone(), after.clone())) + } + _ => None, + }) + .collect(); + let parallel = CoreOptions::new(self.table.schema().options()).diff_parallelism(); + let table = self.table.clone(); + let read_type = self.read_type.clone(); + let data_predicates = self.data_predicates.clone(); + + Ok(Box::pin(async_stream::try_stream! { + let mut workers = stream::iter(pairs.into_iter().map(|(before, after)| { + let table = table.clone(); + let read_type = read_type.clone(); + let data_predicates = data_predicates.clone(); + async move { + let pair_read = + PaimonTableRead::new(&table, read_type, data_predicates); + pair_read.to_diff_after_image_stream(&before, &after) + } + })) + .buffer_unordered(parallel); + while let Some(stream_result) = workers.next().await { + let mut pair_stream = stream_result?; + while let Some(batch) = pair_stream.next().await { + yield batch?; + } + } + })) + } + + /// Returns an audit-log stream for a planned incremental scan. + pub fn to_audit_log_arrow( + &self, + plan: &IncrementalPlan, + ) -> crate::Result { + match plan.mode() { + IncrementalScanMode::Diff => self.audit_diff_stream(plan), + IncrementalScanMode::Delta => self.audit_raw_stream(plan, false), + IncrementalScanMode::Changelog => self.audit_raw_stream(plan, true), + IncrementalScanMode::Auto => unreachable!("Auto resolved during plan()"), + } + } + + fn audit_raw_stream( + &self, + plan: &IncrementalPlan, + has_value_kind: bool, + ) -> crate::Result { + 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 = 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)), + } + })?; + } + })) + } + + fn audit_diff_stream(&self, plan: &IncrementalPlan) -> crate::Result { + let pairs: Vec<(Vec, Vec)> = plan + .splits() + .iter() + .filter_map(|s| match s { + IncrementalSplit::DiffPair { before, after } => { + Some((before.clone(), after.clone())) + } + _ => None, + }) + .collect(); + let parallel = CoreOptions::new(self.table.schema().options()).diff_parallelism(); + let table = self.table.clone(); + let read_type = self.read_type.clone(); + let data_predicates = self.data_predicates.clone(); + + Ok(Box::pin(async_stream::try_stream! { + let mut workers = stream::iter(pairs.into_iter().map(|(before, after)| { + let table = table.clone(); + let read_type = read_type.clone(); + let data_predicates = data_predicates.clone(); + async move { + let pair_read = PaimonTableRead::new(&table, read_type, data_predicates); + pair_read.to_audit_log_arrow_for_diff(&before, &after) + } + })) + .buffer_unordered(parallel); + while let Some(stream_result) = workers.next().await { + let mut pair_stream = stream_result?; + while let Some(batch) = pair_stream.next().await { + yield batch?; + } + } + })) + } + + fn to_audit_log_arrow_for_diff( + &self, + before: &[DataSplit], + after: &[DataSplit], + ) -> crate::Result { + ensure_diff_supported_read_type(&self.read_type)?; + let include_sequence = audit_sequence_number_enabled(self.table); + let audit_schema = audit_schema_for_read_type(&self.read_type, include_sequence)?; + + let mut diff_read_type = self.read_type.clone(); + if include_sequence { + diff_read_type.insert( + 0, + DataField::new( + SEQUENCE_NUMBER_FIELD_ID, + SEQUENCE_NUMBER_FIELD_NAME.to_string(), + DataType::BigInt(BigIntType::new()), + ), + ); + } + + let key_indices = primary_key_indices(self.table, &diff_read_type)?; + let value_indices = value_indices_for_diff(self.table, &diff_read_type); + + let before = before.to_vec(); + let after = after.to_vec(); + let table = self.table.clone(); + let read_type_for_output = self.read_type.clone(); + let data_predicates = self.data_predicates.clone(); + + Ok(Box::pin(async_stream::try_stream! { + let core_options = CoreOptions::new(table.schema().options()); + let pair_read = PaimonTableRead::new(&table, diff_read_type.clone(), data_predicates); + let before_stream = + pair_read.read_pk_sorted_for_diff_with_type(&before, &core_options, &diff_read_type)?; + let after_stream = + pair_read.read_pk_sorted_for_diff_with_type(&after, &core_options, &diff_read_type)?; + let mut bc = ArrowCursor::new(before_stream).await?; + let mut ac = ArrowCursor::new(after_stream).await?; + let mut data_col_indices: Option> = None; + let mut builder = AuditBatchBuilder::new(audit_schema.clone()); + + while bc.alive() || ac.alive() { + let indices = data_col_indices.get_or_insert_with(|| { + let sample = if bc.alive() { + bc.batch() + } else { + ac.batch() + }; + diff_output_col_indices(sample, &read_type_for_output, include_sequence) + .expect("diff output column indices") + }); + if !builder.has_data_columns() { + builder.set_data_col_indices(indices.clone()); + } + match cursor_cmp(&bc, &ac, &key_indices, &value_indices)? { + CursorOrd::BeforeOnly => { + builder.push("-D", bc.batch(), bc.row()); + bc.advance().await?; + } + CursorOrd::AfterOnly => { + builder.push("+I", ac.batch(), ac.row()); + ac.advance().await?; + } + CursorOrd::EqualSame => { + bc.advance().await?; + ac.advance().await?; + } + CursorOrd::EqualDiff => { + builder.push("-U", bc.batch(), bc.row()); + builder.push("+U", ac.batch(), ac.row()); + bc.advance().await?; + ac.advance().await?; + } + } + if builder.len() >= DIFF_BATCH_SIZE { + yield builder.flush()?; + } + } + if builder.len() > 0 { + yield builder.flush()?; + } + })) + } + + fn to_diff_after_image_stream( + &self, + before: &[DataSplit], + after: &[DataSplit], + ) -> crate::Result { + ensure_diff_supported_read_type(&self.read_type)?; + let key_indices = primary_key_indices(self.table, &self.read_type)?; + let value_indices = value_indices_for_diff(self.table, &self.read_type); + let output_schema = build_target_arrow_schema(&self.read_type)?; + let output_col_indices: Vec = (0..self.read_type.len()).collect(); + + let table = self.table.clone(); + let read_type = self.read_type.clone(); + let data_predicates = self.data_predicates.clone(); + let before = before.to_vec(); + let after = after.to_vec(); + + Ok(Box::pin(async_stream::try_stream! { + let core_options = CoreOptions::new(table.schema().options()); + let pair_read = PaimonTableRead::new(&table, read_type, data_predicates); + let before_stream = pair_read.read_pk_sorted_for_diff(&before, &core_options)?; + let after_stream = pair_read.read_pk_sorted_for_diff(&after, &core_options)?; + let mut bc = ArrowCursor::new(before_stream).await?; + let mut ac = ArrowCursor::new(after_stream).await?; + let mut builder = + DiffAfterImageBatchBuilder::new(output_schema.clone(), output_col_indices.clone()); + + while bc.alive() || ac.alive() { + match cursor_cmp(&bc, &ac, &key_indices, &value_indices)? { + CursorOrd::BeforeOnly => { + bc.advance().await?; + } + CursorOrd::AfterOnly => { + builder.push(ac.batch(), ac.row()); + ac.advance().await?; + } + CursorOrd::EqualSame => { + bc.advance().await?; + ac.advance().await?; + } + CursorOrd::EqualDiff => { + builder.push(ac.batch(), ac.row()); + bc.advance().await?; + ac.advance().await?; + } + } + if builder.len() >= DIFF_BATCH_SIZE { + yield builder.flush()?; + } + } + if builder.len() > 0 { + yield builder.flush()?; + } + })) + } + + fn read_pk_sorted_for_diff( + &self, + splits: &[DataSplit], + core_options: &CoreOptions, + ) -> crate::Result { + self.read_pk_sorted_for_diff_with_type(splits, core_options, &self.read_type) + } + + fn read_pk_sorted_for_diff_with_type( + &self, + splits: &[DataSplit], + core_options: &CoreOptions, + read_type: &[DataField], + ) -> crate::Result { + if splits.is_empty() { + return Ok(Box::pin(futures::stream::empty())); + } + for split in splits { + if split + .data_deletion_files() + .is_some_and(|files| files.iter().any(|file| file.is_some())) + { + return Err(crate::Error::Unsupported { + message: "Batch incremental Diff does not support deletion vectors".to_string(), + }); + } + } + let reader = KeyValueFileReader::new( + self.table.file_io.clone(), + KeyValueReadConfig { + table_name: self.table.identifier().full_name(), + table_options: self.table.schema().options().clone(), + schema_manager: self.table.schema_manager().clone(), + table_schema_id: self.table.schema().id(), + table_fields: self.table.schema.fields().to_vec(), + read_type: read_type.to_vec(), + predicates: self.data_predicates.clone(), + primary_keys: self.table.schema.trimmed_primary_keys(), + merge_engine: core_options.merge_engine()?, + sequence_fields: core_options + .sequence_fields() + .iter() + .map(|s| s.to_string()) + .collect(), + }, + ); + reader.read(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(); @@ -307,6 +747,497 @@ impl<'a> PaimonTableRead<'a> { } } +fn audit_schema_for_read_type( + read_type: &[DataField], + include_sequence: bool, +) -> crate::Result> { + let mut fields = Vec::with_capacity(read_type.len() + 2); + fields.push(DataField::new( + -1, + "rowkind".to_string(), + DataType::VarChar(crate::spec::VarCharType::new(8)?), + )); + if include_sequence { + fields.push(DataField::new( + SEQUENCE_NUMBER_FIELD_ID, + SEQUENCE_NUMBER_FIELD_NAME.to_string(), + DataType::BigInt(BigIntType::new()), + )); + } + fields.extend(read_type.iter().cloned()); + build_target_arrow_schema(&fields) +} + +fn audit_sequence_number_enabled(table: &Table) -> bool { + table + .schema() + .options() + .get("table-read.sequence-number.enabled") + .is_some_and(|v| v.eq_ignore_ascii_case("true")) +} + +fn rowkind_array_from_column(column: &dyn arrow_array::Array) -> crate::Result { + let values = column + .as_any() + .downcast_ref::() + .ok_or_else(|| crate::Error::DataInvalid { + message: "AuditLogTable _VALUE_KIND column must be Int8".to_string(), + source: None, + })?; + let strings: Vec<&'static str> = (0..values.len()) + .map(|idx| match values.value(idx) { + 0 => "+I", + 1 => "-U", + 2 => "+U", + 3 => "-D", + _ => "?", + }) + .collect(); + Ok(StringArray::from(strings)) +} + +const DIFF_BATCH_SIZE: usize = 8192; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CursorOrd { + BeforeOnly, + AfterOnly, + EqualSame, + EqualDiff, +} + +struct ArrowCursor { + stream: ArrowRecordBatchStream, + batch: Option, + row: usize, +} + +impl ArrowCursor { + async fn new(stream: ArrowRecordBatchStream) -> crate::Result { + let mut cursor = Self { + stream, + batch: None, + row: 0, + }; + cursor.advance().await?; + Ok(cursor) + } + + fn alive(&self) -> bool { + self.batch.is_some() + } + + fn batch(&self) -> &RecordBatch { + self.batch.as_ref().expect("cursor must be alive") + } + + fn row(&self) -> usize { + self.row + } + + async fn advance(&mut self) -> crate::Result<()> { + loop { + if let Some(ref batch) = self.batch { + if self.row + 1 < batch.num_rows() { + self.row += 1; + return Ok(()); + } + } + match self.stream.next().await { + Some(Ok(batch)) if batch.num_rows() > 0 => { + self.batch = Some(batch); + self.row = 0; + return Ok(()); + } + Some(Ok(_)) => continue, + Some(Err(e)) => return Err(e), + None => { + self.batch = None; + return Ok(()); + } + } + } + } +} + +struct AuditBatchBuilder { + schema: Arc, + rowkind: StringBuilder, + row_indices: Vec<(usize, usize)>, + pinned_batches: Vec, + data_col_indices: Vec, + len: usize, +} + +impl AuditBatchBuilder { + fn new(schema: Arc) -> Self { + Self { + schema, + rowkind: StringBuilder::new(), + row_indices: Vec::new(), + pinned_batches: Vec::new(), + data_col_indices: Vec::new(), + len: 0, + } + } + + fn has_data_columns(&self) -> bool { + !self.data_col_indices.is_empty() + } + + fn set_data_col_indices(&mut self, indices: Vec) { + self.data_col_indices = indices; + } + + fn len(&self) -> usize { + self.len + } + + fn push(&mut self, kind: &str, batch: &RecordBatch, row: usize) { + self.rowkind.append_value(kind); + let batch_id = self.pin_batch(batch); + self.row_indices.push((batch_id, row)); + self.len += 1; + } + + fn pin_batch(&mut self, batch: &RecordBatch) -> usize { + if let Some(last) = self.pinned_batches.last() { + if std::ptr::eq(batch, last) { + return self.pinned_batches.len() - 1; + } + } + let batch_id = self.pinned_batches.len(); + self.pinned_batches.push(batch.clone()); + batch_id + } + + fn flush(&mut self) -> crate::Result { + let mut columns: Vec = vec![Arc::new(self.rowkind.finish())]; + self.rowkind = StringBuilder::new(); + for &col_idx in &self.data_col_indices { + let taken: Vec = self + .row_indices + .iter() + .map(|(batch_id, row)| { + take( + self.pinned_batches[*batch_id].column(col_idx).as_ref(), + &UInt32Array::from(vec![*row as u32]), + None, + ) + .map_err(|e| crate::Error::UnexpectedError { + message: format!("Failed to take audit diff column: {e}"), + source: Some(Box::new(e)), + }) + }) + .collect::>>()?; + let refs: Vec<&dyn Array> = taken.iter().map(|array| array.as_ref()).collect(); + columns.push( + arrow_concat(&refs).map_err(|e| crate::Error::UnexpectedError { + message: format!("Failed to concat audit diff column: {e}"), + source: Some(Box::new(e)), + })?, + ); + } + self.row_indices.clear(); + self.pinned_batches.clear(); + self.len = 0; + RecordBatch::try_new(self.schema.clone(), columns).map_err(|e| { + crate::Error::UnexpectedError { + message: format!("Failed to build audit diff batch: {e}"), + source: Some(Box::new(e)), + } + }) + } +} + +struct DiffAfterImageBatchBuilder { + schema: Arc, + row_indices: Vec<(usize, usize)>, + pinned_batches: Vec, + col_indices: Vec, + len: usize, +} + +impl DiffAfterImageBatchBuilder { + fn new(schema: Arc, col_indices: Vec) -> Self { + Self { + schema, + row_indices: Vec::new(), + pinned_batches: Vec::new(), + col_indices, + len: 0, + } + } + + fn len(&self) -> usize { + self.len + } + + fn push(&mut self, batch: &RecordBatch, row: usize) { + let batch_id = self.pin_batch(batch); + self.row_indices.push((batch_id, row)); + self.len += 1; + } + + fn pin_batch(&mut self, batch: &RecordBatch) -> usize { + if let Some(last) = self.pinned_batches.last() { + if std::ptr::eq(batch, last) { + return self.pinned_batches.len() - 1; + } + } + let batch_id = self.pinned_batches.len(); + self.pinned_batches.push(batch.clone()); + batch_id + } + + fn flush(&mut self) -> crate::Result { + let mut columns = Vec::with_capacity(self.col_indices.len()); + for &col_idx in &self.col_indices { + let taken: Vec = self + .row_indices + .iter() + .map(|(batch_id, row)| { + take( + self.pinned_batches[*batch_id].column(col_idx).as_ref(), + &UInt32Array::from(vec![*row as u32]), + None, + ) + .map_err(|e| crate::Error::UnexpectedError { + message: format!("Failed to take diff after-image column: {e}"), + source: Some(Box::new(e)), + }) + }) + .collect::>>()?; + let refs: Vec<&dyn Array> = taken.iter().map(|array| array.as_ref()).collect(); + columns.push( + arrow_concat(&refs).map_err(|e| crate::Error::UnexpectedError { + message: format!("Failed to concat diff after-image column: {e}"), + source: Some(Box::new(e)), + })?, + ); + } + self.row_indices.clear(); + self.pinned_batches.clear(); + self.len = 0; + RecordBatch::try_new(self.schema.clone(), columns).map_err(|e| { + crate::Error::UnexpectedError { + message: format!("Failed to build diff after-image batch: {e}"), + source: Some(Box::new(e)), + } + }) + } +} + +fn diff_output_col_indices( + batch: &RecordBatch, + read_type: &[DataField], + include_sequence: bool, +) -> crate::Result> { + let mut indices = Vec::with_capacity(read_type.len() + usize::from(include_sequence)); + if include_sequence { + indices.push( + batch + .schema() + .index_of(SEQUENCE_NUMBER_FIELD_NAME) + .map_err(|e| crate::Error::DataInvalid { + message: format!("Diff read missing _SEQUENCE_NUMBER: {e}"), + source: None, + })?, + ); + } + for field in read_type { + indices.push(batch.schema().index_of(field.name()).map_err(|e| { + crate::Error::DataInvalid { + message: format!("Diff read missing column '{}': {e}", field.name()), + source: None, + } + })?); + } + Ok(indices) +} + +fn value_indices_for_diff(table: &Table, fields: &[DataField]) -> Vec { + let primary_key_names = table.schema().trimmed_primary_keys(); + let primary_keys: std::collections::HashSet<&str> = + primary_key_names.iter().map(|key| key.as_str()).collect(); + fields + .iter() + .enumerate() + .filter(|(_, field)| { + field.name() != SEQUENCE_NUMBER_FIELD_NAME && !primary_keys.contains(field.name()) + }) + .map(|(index, _)| index) + .collect() +} + +fn primary_key_indices(table: &Table, read_type: &[DataField]) -> crate::Result> { + let mut indices = Vec::new(); + for pk in table.schema().trimmed_primary_keys() { + let idx = read_type + .iter() + .position(|field| field.name() == pk) + .ok_or_else(|| crate::Error::DataInvalid { + message: format!("Primary key column '{pk}' missing from read projection"), + source: None, + })?; + indices.push(idx); + } + Ok(indices) +} + +fn ensure_diff_supported_read_type(read_type: &[DataField]) -> crate::Result<()> { + for field in read_type { + if !is_diff_supported_type(field.data_type()) { + return Err(crate::Error::Unsupported { + message: format!( + "Batch incremental Diff does not support nested or decimal column '{}'", + field.name() + ), + }); + } + } + Ok(()) +} + +fn is_diff_supported_type(data_type: &DataType) -> bool { + !matches!( + data_type, + DataType::Decimal(_) | DataType::Array(_) | DataType::Map(_) | DataType::Row(_) + ) +} + +fn cursor_cmp( + bc: &ArrowCursor, + ac: &ArrowCursor, + key_indices: &[usize], + value_indices: &[usize], +) -> crate::Result { + match (bc.alive(), ac.alive()) { + (false, false) => unreachable!("cursor_cmp called with both streams exhausted"), + (false, true) => return Ok(CursorOrd::AfterOnly), + (true, false) => return Ok(CursorOrd::BeforeOnly), + (true, true) => {} + } + match compare_pk(bc, ac, key_indices)? { + Ordering::Less => Ok(CursorOrd::BeforeOnly), + Ordering::Greater => Ok(CursorOrd::AfterOnly), + Ordering::Equal => { + if rows_equal_at(bc.batch(), bc.row(), ac.batch(), ac.row(), value_indices)? { + Ok(CursorOrd::EqualSame) + } else { + Ok(CursorOrd::EqualDiff) + } + } + } +} + +fn compare_pk( + bc: &ArrowCursor, + ac: &ArrowCursor, + key_indices: &[usize], +) -> crate::Result { + for &idx in key_indices { + let ord = scalar_compare( + bc.batch().column(idx), + bc.row(), + ac.batch().column(idx), + ac.row(), + )?; + if ord != Ordering::Equal { + return Ok(ord); + } + } + Ok(Ordering::Equal) +} + +fn rows_equal_at( + left_batch: &RecordBatch, + left_row: usize, + right_batch: &RecordBatch, + right_row: usize, + indices: &[usize], +) -> crate::Result { + for &idx in indices { + let ord = scalar_compare( + left_batch.column(idx), + left_row, + right_batch.column(idx), + right_row, + )?; + if ord != Ordering::Equal { + return Ok(false); + } + } + Ok(true) +} + +fn scalar_compare( + left: &dyn Array, + left_row: usize, + right: &dyn Array, + right_row: usize, +) -> crate::Result { + use arrow_array::{ + BooleanArray, Date32Array, Float32Array, Float64Array, Int16Array, Int32Array, Int64Array, + Int8Array, StringArray, UInt16Array, UInt32Array, UInt64Array, UInt8Array, + }; + + macro_rules! compare { + ($ty:ty, $getter:expr) => { + if let (Some(a), Some(b)) = ( + left.as_any().downcast_ref::<$ty>(), + right.as_any().downcast_ref::<$ty>(), + ) { + return Ok($getter(a, left_row).cmp(&$getter(b, right_row))); + } + }; + } + + compare!(Int8Array, |a: &Int8Array, r| a.value(r)); + compare!(Int16Array, |a: &Int16Array, r| a.value(r)); + compare!(Int32Array, |a: &Int32Array, r| a.value(r)); + compare!(Int64Array, |a: &Int64Array, r| a.value(r)); + compare!(UInt8Array, |a: &UInt8Array, r| a.value(r)); + compare!(UInt16Array, |a: &UInt16Array, r| a.value(r)); + compare!(UInt32Array, |a: &UInt32Array, r| a.value(r)); + compare!(UInt64Array, |a: &UInt64Array, r| a.value(r)); + compare!(BooleanArray, |a: &BooleanArray, r| a.value(r)); + compare!(Date32Array, |a: &Date32Array, r| a.value(r)); + + if let (Some(a), Some(b)) = ( + left.as_any().downcast_ref::(), + right.as_any().downcast_ref::(), + ) { + return Ok(a.value(left_row).cmp(b.value(right_row))); + } + + if let (Some(a), Some(b)) = ( + left.as_any().downcast_ref::(), + right.as_any().downcast_ref::(), + ) { + return Ok(a + .value(left_row) + .partial_cmp(&b.value(right_row)) + .unwrap_or(Ordering::Equal)); + } + if let (Some(a), Some(b)) = ( + left.as_any().downcast_ref::(), + right.as_any().downcast_ref::(), + ) { + return Ok(a + .value(left_row) + .partial_cmp(&b.value(right_row)) + .unwrap_or(Ordering::Equal)); + } + + Err(crate::Error::Unsupported { + message: format!( + "Batch incremental Diff does not support comparing column type {:?}", + left.data_type() + ), + }) +} + /// Whether a primary-key split must go through the sort-merge reader. /// /// Mirrors Java `PrimaryKeyTableRawFileSplitReadProvider#match`: a raw read @@ -410,4 +1341,28 @@ mod tests { "directly-constructed read of a query-auth.enabled table must fail closed" ); } + + #[test] + fn test_diff_rejects_nested_and_decimal_types() { + use crate::spec::{ArrayType, DecimalType, IntType}; + + let decimal = DataField::new( + 1, + "amount".to_string(), + DataType::Decimal(DecimalType::new(10, 2).unwrap()), + ); + let nested = DataField::new( + 2, + "tags".to_string(), + DataType::Array(ArrayType::new(DataType::Int(IntType::new()))), + ); + assert!(matches!( + ensure_diff_supported_read_type(&[decimal]), + Err(crate::Error::Unsupported { message }) if message.contains("nested or decimal") + )); + assert!(matches!( + ensure_diff_supported_read_type(&[nested]), + Err(crate::Error::Unsupported { message }) if message.contains("nested or decimal") + )); + } } diff --git a/crates/paimon/src/table/table_scan.rs b/crates/paimon/src/table/table_scan.rs index 58338abc..5af25eb5 100644 --- a/crates/paimon/src/table/table_scan.rs +++ b/crates/paimon/src/table/table_scan.rs @@ -706,6 +706,40 @@ 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(), + }), + } + } + + /// Plan data splits from a snapshot's changelog manifest list only. + pub(crate) async fn plan_snapshot_changelog(&self, snapshot: &Snapshot) -> crate::Result { + 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(), + }), + } + } + + /// Plan before/after full-snapshot splits for batch incremental Diff. + pub(crate) async fn plan_snapshot_diff( + &self, + before: &Snapshot, + after: &Snapshot, + ) -> crate::Result<(Plan, Plan)> { + match &self.0 { + TableScanKind::Paimon(scan) => scan.plan_snapshot_diff(before, after).await, + TableScanKind::Format(_) => Err(crate::Error::Unsupported { + message: "Format tables do not support incremental Diff scan".to_string(), + }), + } + } + #[cfg(test)] fn apply_limit_pushdown(&self, splits: Vec) -> Vec { match &self.0 { @@ -1048,11 +1082,287 @@ 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 + } + + /// 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 { + 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 + } + + /// Plan before/after full-snapshot states for Diff incremental scan. + /// + /// Loads full manifest entries for both snapshots, rejects bucket rescale, + /// prunes files that are identical and unaffected by exclusive changes, then + /// builds splits via the shared snapshot planning path. + pub(crate) async fn plan_snapshot_diff( + &self, + before: &Snapshot, + after: &Snapshot, + ) -> crate::Result<(Plan, Plan)> { + self.ensure_query_auth_allowed()?; + let mut before_entries = self.plan_manifest_entries(before).await?; + let mut after_entries = self.plan_manifest_entries(after).await?; + Self::validate_diff_bucket_layout(&before_entries, &after_entries)?; + Self::prune_unchanged_diff_files(&mut before_entries, &mut after_entries); + let data_evolution_read_field_ids = self.projected_read_field_ids()?; + let before_plan = self + .plan_snapshot_from_entries( + before.clone(), + before_entries, + data_evolution_read_field_ids.as_ref(), + None, + ) + .await?; + let after_plan = self + .plan_snapshot_from_entries( + after.clone(), + after_entries, + data_evolution_read_field_ids.as_ref(), + None, + ) + .await?; + Ok((before_plan, after_plan)) + } + + fn validate_diff_bucket_layout( + before_entries: &[ManifestEntry], + after_entries: &[ManifestEntry], + ) -> crate::Result<()> { + use std::collections::{BTreeMap, BTreeSet}; + + fn totals(entries: &[ManifestEntry]) -> BTreeMap, BTreeSet> { + let mut result: BTreeMap, BTreeSet> = BTreeMap::new(); + for entry in entries { + result + .entry(entry.partition().to_vec()) + .or_default() + .insert(entry.total_buckets()); + } + result + } + + let before = totals(before_entries); + let after = totals(after_entries); + for (partition, before_totals) in &before { + let Some(after_totals) = after.get(partition) else { + continue; + }; + if before_totals != after_totals { + return Err(crate::Error::Unsupported { + message: + "Batch incremental Diff does not support bucket rescale between snapshots" + .to_string(), + }); + } + } + Ok(()) + } + + fn prune_unchanged_diff_files(before: &mut Vec, after: &mut Vec) { + // Common files (byte-equal manifest entries) that do not overlap any + // exclusive before/after file key-range can be dropped from both sides. + let mut common: Vec = before + .iter() + .filter(|entry| after.iter().any(|other| other == *entry)) + .cloned() + .collect(); + if common.is_empty() { + return; + } + + let exclusive_after: Vec<&ManifestEntry> = after + .iter() + .filter(|entry| !common.iter().any(|c| c == *entry)) + .collect(); + let exclusive_before: Vec<&ManifestEntry> = before + .iter() + .filter(|entry| !common.iter().any(|c| c == *entry)) + .collect(); + common.retain(|entry| { + let overlaps_exclusive = |other: &&ManifestEntry| { + Self::key_ranges_overlap_bytes( + &entry.file().min_key, + &entry.file().max_key, + &other.file().min_key, + &other.file().max_key, + ) + }; + !exclusive_after.iter().any(overlaps_exclusive) + && !exclusive_before.iter().any(overlaps_exclusive) + }); + if common.is_empty() { + return; + } + before.retain(|entry| !common.iter().any(|c| c == entry)); + after.retain(|entry| !common.iter().any(|c| c == entry)); + } + + fn key_ranges_overlap_bytes(min_a: &[u8], max_a: &[u8], min_b: &[u8], max_b: &[u8]) -> bool { + min_a <= max_b && min_b <= max_a + } + + /// 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 +1376,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/audit_log_table_test.rs b/crates/paimon/tests/audit_log_table_test.rs new file mode 100644 index 00000000..ced4ba2b --- /dev/null +++ b/crates/paimon/tests/audit_log_table_test.rs @@ -0,0 +1,538 @@ +// 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::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::() + .unwrap(); + let ids = batch + .column(id_idx) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(value_idx) + .as_any() + .downcast_ref::() + .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::() + .unwrap(); + let seqs = batch + .column(seq_idx) + .as_any() + .downcast_ref::() + .unwrap(); + let ids = batch + .column(id_idx) + .as_any() + .downcast_ref::() + .unwrap(); + let values = batch + .column(value_idx) + .as_any() + .downcast_ref::() + .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 = 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 (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, 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 = audit.to_arrow(&plan).unwrap().try_collect().await.unwrap(); + + assert_eq!( + collect_audit_rows(&batches), + vec![("+I".to_string(), 1, 10), ("+I".to_string(), 2, 20),] + ); +} + +#[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 = 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 = audit.to_arrow(&plan).unwrap().try_collect().await.unwrap(); + let batch_schema: Vec = 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)); +} + +async fn audit_diff_rows( + table: &paimon::table::Table, + start: i64, + end: i64, +) -> Vec<(String, i32, i32)> { + let audit = AuditLogTable::new(table.clone()); + let plan = audit + .new_incremental_scan(IncrementalScanMode::Diff, start, end) + .plan() + .await + .unwrap(); + let batches: Vec = audit.to_arrow(&plan).unwrap().try_collect().await.unwrap(); + collect_audit_rows(&batches) +} + +fn assert_rows_contain(rows: &[(String, i32, i32)], expected: &[(&str, i32, i32)]) { + for (kind, id, value) in expected { + assert!( + rows.iter() + .any(|(k, i, v)| k == *kind && i == id && v == value), + "missing rowkind={kind} id={id} value={value} in {rows:?}" + ); + } +} + +fn assert_rows_exclude(rows: &[(String, i32, i32)], excluded: &[(&str, i32, i32)]) { + for (kind, id, value) in excluded { + assert!( + !rows + .iter() + .any(|(k, i, v)| k == *kind && i == id && v == value), + "unexpected rowkind={kind} id={id} value={value} in {rows:?}" + ); + } +} + +#[tokio::test] +async fn audit_log_diff_scan_emits_row_level_delete_insert_and_updates() { + let table_path = "memory:/audit_log/diff_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, 2], vec![10, 20])).await; + write_batch(&table, &make_batch(vec![2, 3], vec![25, 30])).await; + + let rows = audit_diff_rows(&table, 1, 2).await; + assert_eq!( + rows, + vec![ + ("+I".to_string(), 3, 30), + ("+U".to_string(), 2, 25), + ("-U".to_string(), 2, 20), + ] + ); +} + +#[tokio::test] +async fn audit_log_diff_same_snapshot_range_returns_no_rows() { + let table_path = "memory:/audit_log/diff_same_snapshot"; + 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 rows = audit_diff_rows(&table, 1, 1).await; + assert!( + rows.is_empty(), + "same start/end snapshot should yield empty diff" + ); +} + +#[tokio::test] +async fn audit_log_diff_insert_only_emits_plus_i() { + let table_path = "memory:/audit_log/diff_insert_only"; + 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![1, 2], vec![10, 20])).await; + + let rows = audit_diff_rows(&table, 1, 2).await; + assert_eq!(rows, vec![("+I".to_string(), 2, 20)]); +} + +#[tokio::test] +async fn audit_log_diff_update_only_emits_minus_u_and_plus_u_from_before_after() { + let table_path = "memory:/audit_log/diff_update_only"; + 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![1], vec![20])).await; + + let rows = audit_diff_rows(&table, 1, 2).await; + assert_eq!( + rows, + vec![("+U".to_string(), 1, 20), ("-U".to_string(), 1, 10),] + ); +} + +#[tokio::test] +async fn audit_log_diff_delete_via_input_delete_row() { + // Diff compares materialized PK state; input -D removes a key without compact. + let table_path = "memory:/audit_log/diff_delete_input"; + 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, 2], vec![10, 20], vec![0, 0]), + ) + .await; + write_batch(&table, &make_batch_with_kinds(vec![1], vec![10], vec![3])).await; + + let rows = audit_diff_rows(&table, 1, 2).await; + assert_rows_contain(&rows, &[("-D", 1, 10)]); + assert_rows_exclude(&rows, &[("+I", 1, 10), ("-U", 1, 10), ("+U", 1, 10)]); +} + +#[tokio::test] +async fn audit_log_diff_mixed_delete_insert_update_without_compact() { + let table_path = "memory:/audit_log/diff_mixed"; + 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, 2, 3], vec![10, 20, 30], vec![0, 0, 0]), + ) + .await; + write_batch( + &table, + &make_batch_with_kinds(vec![1, 2, 4], vec![10, 25, 40], vec![3, 2, 0]), + ) + .await; + + let rows = audit_diff_rows(&table, 1, 2).await; + assert_rows_contain( + &rows, + &[("-D", 1, 10), ("-U", 2, 20), ("+U", 2, 25), ("+I", 4, 40)], + ); + assert_rows_exclude(&rows, &[("+I", 3, 30), ("-D", 3, 30)]); +} + +#[tokio::test] +async fn audit_log_diff_processes_multiple_bucket_pairs() { + let table_path = "memory:/audit_log/diff_multi_bucket"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("changelog-producer", "none"), + ("merge-engine", "deduplicate"), + ("bucket", "4"), + ]), + ); + setup_dirs(&file_io, table_path).await; + persist_table_schema(&file_io, table_path, table.schema()).await; + + write_batch(&table, &make_batch(vec![1, 8], vec![10, 80])).await; + write_batch(&table, &make_batch(vec![1, 8], vec![11, 81])).await; + + let plan = table + .new_read_builder() + .new_incremental_scan(IncrementalScanMode::Diff, 1, 2) + .plan() + .await + .unwrap(); + let diff_pairs = plan + .splits() + .iter() + .filter(|split| matches!(split, paimon::table::IncrementalSplit::DiffPair { .. })) + .count(); + assert!( + diff_pairs >= 2, + "expected multiple (partition,bucket) diff pairs, got {diff_pairs}" + ); + + let rows = audit_diff_rows(&table, 1, 2).await; + assert_rows_contain( + &rows, + &[("-U", 1, 10), ("+U", 1, 11), ("-U", 8, 80), ("+U", 8, 81)], + ); +} + +#[tokio::test] +async fn audit_log_diff_with_sequence_number_enabled_exposes_ordered_columns() { + use std::collections::HashMap; + + let table_path = "memory:/audit_log/diff_sequence"; + let (file_io, table) = memory_table( + table_path, + pk_schema(&[ + ("changelog-producer", "none"), + ("merge-engine", "deduplicate"), + ("bucket", "1"), + ]) + .copy_with_options(HashMap::from([( + "table-read.sequence-number.enabled".to_string(), + "true".to_string(), + )])), + ); + 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![1], vec![20])).await; + + let audit = AuditLogTable::new(table.clone()); + let field_names: Vec = 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::Diff, 1, 2) + .plan() + .await + .unwrap(); + let batches: Vec = audit.to_arrow(&plan).unwrap().try_collect().await.unwrap(); + let rows = collect_audit_rows_with_sequence(&batches); + assert_eq!(rows.len(), 2); + assert_rows_contain( + &rows + .iter() + .map(|(k, _s, i, v)| (k.clone(), *i, *v)) + .collect::>(), + &[("-U", 1, 10), ("+U", 1, 20)], + ); + assert!(rows.iter().all(|(_, seq, _, _)| *seq >= 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..9ef5f62a --- /dev/null +++ b/crates/paimon/tests/common/incremental_helpers.rs @@ -0,0 +1,162 @@ +// 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, Int8Array, 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() +} + +/// 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, values: Vec, kinds: Vec) -> 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, 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..b0ff96d0 --- /dev/null +++ b/crates/paimon/tests/incremental_batch_scan_test.rs @@ -0,0 +1,515 @@ +// 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_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)> { + 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 reads existing changelog_manifest_list data files. +#[tokio::test] +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 = read + .to_incremental_arrow(&plan) + .unwrap() + .try_collect() + .await + .unwrap(); + + assert_eq!(collect_pairs(&batches), vec![(1, 10)]); +} + +#[tokio::test] +async fn diff_between_snapshots_returns_after_image_rows() { + let table_path = "memory:/incremental_batch/diff_after_image"; + 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, 2], vec![10, 20])).await; + write_batch(&table, &make_batch(vec![2, 3], vec![25, 30])).await; + + let rows = read_incremental_pairs(&table, IncrementalScanMode::Diff, 1, 2).await; + assert_eq!(rows, vec![(2, 25), (3, 30)]); +} + +#[tokio::test] +async fn diff_identical_rows_are_skipped_from_after_image() { + let table_path = "memory:/incremental_batch/diff_identical"; + 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![1, 2], vec![10, 20])).await; + + let rows = read_incremental_pairs(&table, IncrementalScanMode::Diff, 1, 2).await; + assert_eq!(rows, vec![(2, 20)]); +} + +#[tokio::test] +async fn diff_rejects_start_before_earliest_snapshot() { + let table_path = "memory:/incremental_batch/diff_earliest"; + 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::Diff, 0, 1) + .await + .unwrap_err(); + assert!( + matches!(err, paimon::Error::DataInvalid { .. }), + "expected DataInvalid, got {err:?}" + ); +} + +#[tokio::test] +async fn diff_rejects_non_deduplicate_merge_engine() { + for merge_engine in ["partial-update", "aggregation", "first-row"] { + let table_path = format!("memory:/incremental_batch/diff_engine_{merge_engine}"); + let (file_io, table) = memory_table( + &table_path, + pk_schema(&[ + ("changelog-producer", "none"), + ("merge-engine", merge_engine), + ("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 err = plan_incremental(&table, IncrementalScanMode::Diff, 1, 2) + .await + .unwrap_err(); + assert!( + matches!(err, paimon::Error::Unsupported { .. }), + "merge-engine={merge_engine} expected Unsupported, got {err:?}" + ); + } +} + +#[tokio::test] +async fn diff_rejects_bucket_rescale_between_snapshots() { + use std::collections::HashMap; + + let table_path = "memory:/incremental_batch/diff_bucket_rescale"; + 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 table = table.copy_with_options(HashMap::from([("bucket".to_string(), "2".to_string())])); + persist_table_schema(&file_io, table_path, table.schema()).await; + write_batch(&table, &make_batch(vec![2], vec![20])).await; + + let err = plan_incremental(&table, IncrementalScanMode::Diff, 1, 2) + .await + .unwrap_err(); + assert!( + matches!( + err, + paimon::Error::Unsupported { ref message } if message.contains("bucket rescale") + ), + "expected Unsupported for bucket rescale, got {err:?}" + ); +} diff --git a/docs/src/sql.md b/docs/src/sql.md index 50623e4d..f4d2c1e2 100644 --- a/docs/src/sql.md +++ b/docs/src/sql.md @@ -72,9 +72,9 @@ async fn example() -> Result<(), Box> { pre-registered. Use `register_catalog(...).await` to add one or more Paimon catalogs; registering a catalog also registers the built-in scalar function `blob_view` (alias `sys.blob_view`) and the built-in table-valued functions -(`vector_search`, `hybrid_search`, and `full_text_search` when the `fulltext` -feature is enabled) against it. It also manages session-scoped dynamic options -internally for `SET`/`RESET` support. +(`paimon_incremental_query`, `vector_search`, `hybrid_search`, and +`full_text_search` when the `fulltext` feature is enabled) against it. It also +manages session-scoped dynamic options internally for `SET`/`RESET` support. ### REST Catalog Views and SQL Functions @@ -1122,6 +1122,72 @@ CROSS JOIN LATERAL vector_search( ) AS r; ``` +## Incremental Query + +Paimon supports batch incremental reads between snapshot IDs via the +`paimon_incremental_query` table-valued function. This aligns with Spark's +`paimon_incremental_query` TVF: results are returned as `RecordBatch` streams +within a single SQL query (not a long-running CDC stream). + +### Registration + +When you use a `SQLContext`, `paimon_incremental_query` is registered +automatically for every catalog you register — no extra setup is needed. + +With a raw DataFusion `SessionContext`, register it explicitly: + +```rust +use paimon_datafusion::register_incremental_query; + +register_incremental_query(&ctx, catalog.clone(), "default"); +``` + +### Usage + +```sql +SELECT * FROM paimon_incremental_query('table_name', start_snapshot_id, end_snapshot_id); +SELECT * FROM paimon_incremental_query('table_name$audit_log', start_snapshot_id, end_snapshot_id); +SELECT * FROM paimon_incremental_query('table_name$audit_log', start_snapshot_id, end_snapshot_id, 'diff'); +``` + +| Argument | Type | Description | +|---|---|---| +| `table_name` | STRING | Table name, fully qualified (`catalog.db.table`) or short form. Append `$audit_log` to read the audit-log view with a `rowkind` column (`+I`, `-U`, `+U`, `-D`). | +| `start_snapshot_id` | BIGINT | Exclusive lower bound snapshot ID | +| `end_snapshot_id` | BIGINT | Inclusive upper bound snapshot ID | +| `scan_mode` | STRING | Optional. One of `auto` (default), `delta`, `changelog`, or `diff`. | + +Examples: + +```sql +-- Delta / auto rows between snapshots 0 and 2 (exclusive start, inclusive end) +SELECT * FROM paimon_incremental_query('paimon.my_db.orders', 0, 2); + +-- Audit log with row kinds +SELECT rowkind, * FROM paimon_incremental_query('paimon.my_db.orders$audit_log', 0, 2); + +-- Diff mode on a primary-key table (before/after images) +SELECT rowkind, * FROM paimon_incremental_query('paimon.my_db.orders$audit_log', 0, 2, 'diff'); +``` + +The snapshot interval is **(start, end]** — rows committed after +`start_snapshot_id` through `end_snapshot_id` inclusive. Omitting `scan_mode` +uses `auto`, which picks an appropriate mode based on table options (for +example `changelog-producer = input` tables often behave like changelog reads). + +Modes: + +| Mode | Behaviour | +|---|---| +| `auto` | Resolve to `delta` when `changelog-producer=none`, otherwise `changelog`. | +| `delta` | Read data files from APPEND snapshots in the range. | +| `changelog` | Read existing changelog manifest files in the range (skips OVERWRITE and snapshots without changelog manifests). Does not generate changelogs. | +| `diff` | Compare before/after snapshot states for primary-key tables (`merge-engine=deduplicate`). | + +Illegal `scan_mode` values and out-of-range snapshot bounds fail with clear +errors. Projection and residual filters are supported via the normal DataFusion +plan (column projection is applied by the incremental scan; filters are residual). + ## Vector Search Paimon supports approximate nearest neighbor (ANN) vector search through global