From 261df44168defa9b906efce35b6e7987109c5f8d Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Wed, 26 Aug 2026 14:51:15 +0800 Subject: [PATCH] feat(c): expose batch incremental reads --- bindings/c/src/result.rs | 12 ++ bindings/c/src/table.rs | 298 ++++++++++++++++++++++++++++++-- bindings/c/src/tests.rs | 197 +++++++++++++++++++++ bindings/c/src/types.rs | 29 +++- docs/src/c-binding.md | 75 +++++++- docs/src/incremental-reading.md | 11 ++ 6 files changed, 600 insertions(+), 22 deletions(-) diff --git a/bindings/c/src/result.rs b/bindings/c/src/result.rs index 94f87c2de..48e332b6b 100644 --- a/bindings/c/src/result.rs +++ b/bindings/c/src/result.rs @@ -54,12 +54,24 @@ pub struct paimon_result_table_scan { pub error: *mut paimon_error, } +#[repr(C)] +pub struct paimon_result_incremental_scan { + pub scan: *mut paimon_incremental_scan, + pub error: *mut paimon_error, +} + #[repr(C)] pub struct paimon_result_plan { pub plan: *mut paimon_plan, pub error: *mut paimon_error, } +#[repr(C)] +pub struct paimon_result_incremental_plan { + pub plan: *mut paimon_incremental_plan, + pub error: *mut paimon_error, +} + #[repr(C)] pub struct paimon_result_record_batch_reader { pub reader: *mut paimon_record_batch_reader, diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs index 298a3c561..1a5e68114 100644 --- a/bindings/c/src/table.rs +++ b/bindings/c/src/table.rs @@ -24,14 +24,16 @@ use futures::StreamExt; use paimon::catalog::{Identifier, DEFAULT_MAIN_BRANCH}; use paimon::io::FileIO; use paimon::spec::{DataField, DataType, Datum, Predicate, PredicateBuilder, TableSchema}; -use paimon::table::{ArrowRecordBatchStream, DataSplit, Table}; +use paimon::table::{ + ArrowRecordBatchStream, DataSplit, IncrementalPlan, IncrementalScanMode, ReadBuilder, Table, +}; use paimon::Plan; use crate::error::{check_non_null, paimon_error, validate_cstr, PaimonErrorCode}; use crate::result::{ - paimon_result_get_table, paimon_result_new_read, paimon_result_next_batch, paimon_result_plan, - paimon_result_predicate, paimon_result_read_builder, paimon_result_record_batch_reader, - paimon_result_table_scan, + paimon_result_get_table, paimon_result_incremental_plan, paimon_result_incremental_scan, + paimon_result_new_read, paimon_result_next_batch, paimon_result_plan, paimon_result_predicate, + paimon_result_read_builder, paimon_result_record_batch_reader, paimon_result_table_scan, }; use crate::runtime; use crate::types::*; @@ -59,6 +61,37 @@ unsafe fn box_table_read_state(state: TableReadState) -> *mut paimon_table_read Box::into_raw(Box::new(paimon_table_read { inner })) } +fn build_core_read_builder<'a>( + table: &'a Table, + projected_columns: Option<&[String]>, + filter: Option<&Predicate>, + case_sensitive: bool, +) -> paimon::Result> { + let mut builder = table.new_read_builder(); + builder.with_case_sensitive(case_sensitive); + if let Some(columns) = projected_columns { + let columns = columns.iter().map(String::as_str).collect::>(); + builder.with_projection(&columns)?; + } + if let Some(filter) = filter { + builder.with_filter(filter.clone()); + } + Ok(builder) +} + +fn incremental_scan_mode_from_c(mode: i32) -> Result { + match mode { + PAIMON_INCREMENTAL_SCAN_MODE_DELTA => Ok(IncrementalScanMode::Delta), + PAIMON_INCREMENTAL_SCAN_MODE_CHANGELOG => Ok(IncrementalScanMode::Changelog), + PAIMON_INCREMENTAL_SCAN_MODE_AUTO => Ok(IncrementalScanMode::Auto), + PAIMON_INCREMENTAL_SCAN_MODE_DIFF => Ok(IncrementalScanMode::Diff), + _ => Err(paimon_error::new( + PaimonErrorCode::InvalidInput, + format!("unknown incremental scan mode {mode}"), + )), + } +} + // ======================= Table =============================== /// Create a table directly from a resolved Paimon table schema JSON. @@ -540,24 +573,20 @@ pub unsafe extern "C" fn paimon_read_builder_new_read( }; } let state = &*((*rb).inner as *const ReadBuilderState); - let mut rb_rust = state.table.new_read_builder(); - rb_rust.with_case_sensitive(state.case_sensitive); - - // Apply projection if set - if let Some(ref columns) = state.projected_columns { - let col_refs: Vec<&str> = columns.iter().map(|s| s.as_str()).collect(); - if let Err(e) = rb_rust.with_projection(&col_refs) { + let rb_rust = match build_core_read_builder( + &state.table, + state.projected_columns.as_deref(), + state.filter.as_ref(), + state.case_sensitive, + ) { + Ok(builder) => builder, + Err(e) => { return paimon_result_new_read { read: std::ptr::null_mut(), error: paimon_error::from_paimon(e), - }; + } } - } - - // Apply filter if set - if let Some(ref filter) = state.filter { - rb_rust.with_filter(filter.clone()); - } + }; match rb_rust.new_read() { Ok(table_read) => { @@ -578,6 +607,117 @@ pub unsafe extern "C" fn paimon_read_builder_new_read( } } +// ======================= IncrementalScan =============================== + +/// Create a fixed-range incremental scan from a ReadBuilder. +/// +/// The snapshot range is `(start_exclusive, end_inclusive]`. `mode` must be one +/// of `PAIMON_INCREMENTAL_SCAN_MODE_DELTA`, `_CHANGELOG`, `_AUTO`, or `_DIFF`. +/// Projection and filter settings already applied to the ReadBuilder are +/// preserved by this scan for incremental planning. Build the TableRead from +/// the same ReadBuilder to apply the same projection and filter while reading. +/// +/// # Safety +/// `rb` must be a valid pointer returned by `paimon_table_new_read_builder`, or +/// null (returns error). +#[no_mangle] +pub unsafe extern "C" fn paimon_read_builder_new_incremental_scan( + rb: *const paimon_read_builder, + mode: i32, + start_exclusive: i64, + end_inclusive: i64, +) -> paimon_result_incremental_scan { + if let Err(error) = check_non_null(rb, "rb") { + return paimon_result_incremental_scan { + scan: std::ptr::null_mut(), + error, + }; + } + let mode = match incremental_scan_mode_from_c(mode) { + Ok(mode) => mode, + Err(error) => { + return paimon_result_incremental_scan { + scan: std::ptr::null_mut(), + error, + } + } + }; + let state = &*((*rb).inner as *const ReadBuilderState); + let scan_state = IncrementalScanState { + table: state.table.clone(), + projected_columns: state.projected_columns.clone(), + filter: state.filter.clone(), + case_sensitive: state.case_sensitive, + mode, + start_exclusive, + end_inclusive, + }; + let inner = Box::into_raw(Box::new(scan_state)) as *mut c_void; + paimon_result_incremental_scan { + scan: Box::into_raw(Box::new(paimon_incremental_scan { inner })), + error: std::ptr::null_mut(), + } +} + +/// Free a fixed-range incremental scan. +/// +/// # Safety +/// Only call with a scan returned by `paimon_read_builder_new_incremental_scan`. +#[no_mangle] +pub unsafe extern "C" fn paimon_incremental_scan_free(scan: *mut paimon_incremental_scan) { + if !scan.is_null() { + let wrapper = Box::from_raw(scan); + if !wrapper.inner.is_null() { + drop(Box::from_raw(wrapper.inner as *mut IncrementalScanState)); + } + } +} + +/// Plan a fixed snapshot range for incremental reading. +/// +/// # Safety +/// `scan` must be a valid pointer returned by +/// `paimon_read_builder_new_incremental_scan`, or null (returns error). +#[no_mangle] +pub unsafe extern "C" fn paimon_incremental_scan_plan( + scan: *const paimon_incremental_scan, +) -> paimon_result_incremental_plan { + if let Err(error) = check_non_null(scan, "scan") { + return paimon_result_incremental_plan { + plan: std::ptr::null_mut(), + error, + }; + } + let state = &*((*scan).inner as *const IncrementalScanState); + let builder = match build_core_read_builder( + &state.table, + state.projected_columns.as_deref(), + state.filter.as_ref(), + state.case_sensitive, + ) { + Ok(builder) => builder, + Err(error) => { + return paimon_result_incremental_plan { + plan: std::ptr::null_mut(), + error: paimon_error::from_paimon(error), + } + } + }; + let scan = builder.new_incremental_scan(state.mode, state.start_exclusive, state.end_inclusive); + match runtime().block_on(scan.plan()) { + Ok(plan) => paimon_result_incremental_plan { + plan: Box::into_raw(Box::new(paimon_incremental_plan { + inner: Box::into_raw(Box::new(plan)) as *mut c_void, + })), + error: std::ptr::null_mut(), + }, + Err(error) => paimon_result_incremental_plan { + plan: std::ptr::null_mut(), + error: paimon_error::from_paimon(error), + }, + } +} + // ======================= TableScan =============================== /// Free a paimon_table_scan. @@ -706,6 +846,39 @@ pub unsafe extern "C" fn paimon_plan_num_splits(plan: *const paimon_plan) -> usi plan_ref.splits().len() } +/// Free an incremental plan. +/// +/// # Safety +/// Only call with a plan returned by `paimon_incremental_scan_plan`. +#[no_mangle] +pub unsafe extern "C" fn paimon_incremental_plan_free(plan: *mut paimon_incremental_plan) { + if !plan.is_null() { + let wrapper = Box::from_raw(plan); + if !wrapper.inner.is_null() { + drop(Box::from_raw(wrapper.inner as *mut IncrementalPlan)); + } + } +} + +/// Return the number of incremental work units in a plan. +/// +/// Delta, Changelog, and Auto plans contain data splits. Diff plans contain +/// before/after split pairs, each counted as one work unit. +/// +/// # Safety +/// `plan` must be a valid pointer returned by `paimon_incremental_scan_plan`, +/// or null (returns 0). +#[no_mangle] +pub unsafe extern "C" fn paimon_incremental_plan_num_splits( + plan: *const paimon_incremental_plan, +) -> usize { + if plan.is_null() { + return 0; + } + let plan = &*((*plan).inner as *const IncrementalPlan); + plan.splits().len() +} + // ======================= TableRead =============================== /// Free a paimon_table_read. @@ -785,6 +958,71 @@ pub unsafe extern "C" fn paimon_table_read_to_arrow( } } +/// Read a sub-range of an incremental plan as Arrow record batches. +/// +/// `offset` and `length` select a contiguous range of incremental work units. +/// The range is clamped to the available units. The returned reader uses the +/// same `paimon_record_batch_reader_next` and free functions as a batch read. +/// +/// # Safety +/// `read` and `plan` must be valid pointers returned by previous Paimon C API +/// calls, or null (returns error). +#[no_mangle] +pub unsafe extern "C" fn paimon_table_read_to_incremental_arrow( + read: *const paimon_table_read, + plan: *const paimon_incremental_plan, + offset: usize, + length: usize, +) -> paimon_result_record_batch_reader { + if let Err(error) = check_non_null(read, "read") { + return paimon_result_record_batch_reader { + reader: std::ptr::null_mut(), + error, + }; + } + if let Err(error) = check_non_null(plan, "plan") { + return paimon_result_record_batch_reader { + reader: std::ptr::null_mut(), + error, + }; + } + + let state = &*((*read).inner as *const TableReadState); + let plan = &*((*plan).inner as *const IncrementalPlan); + let start = offset.min(plan.splits().len()); + let end = offset.saturating_add(length).min(plan.splits().len()); + let selected = match IncrementalPlan::try_new(plan.mode(), plan.splits()[start..end].to_vec()) { + Ok(plan) => plan, + Err(error) => { + return paimon_result_record_batch_reader { + reader: std::ptr::null_mut(), + error: paimon_error::from_paimon(error), + } + } + }; + let table_read = paimon::table::TableRead::new( + &state.table, + state.read_type.clone(), + state.data_predicates.clone(), + ); + match table_read.to_incremental_arrow(&selected) { + Ok(stream) => { + let reader = Box::new(stream); + let wrapper = Box::new(paimon_record_batch_reader { + inner: Box::into_raw(reader) as *mut c_void, + }); + paimon_result_record_batch_reader { + reader: Box::into_raw(wrapper), + error: std::ptr::null_mut(), + } + } + Err(error) => paimon_result_record_batch_reader { + reader: std::ptr::null_mut(), + error: paimon_error::from_paimon(error), + }, + } +} + // ======================= RecordBatchReader =============================== /// Get the next Arrow record batch from the reader. @@ -865,7 +1103,8 @@ pub unsafe extern "C" fn paimon_record_batch_reader_next( /// Free a paimon_record_batch_reader. /// /// # Safety -/// Only call with a reader returned from `paimon_table_read_to_arrow` or +/// Only call with a reader returned from `paimon_table_read_to_arrow`, +/// `paimon_table_read_to_incremental_arrow`, or /// `paimon_vector_search_builder_execute_read`. #[no_mangle] pub unsafe extern "C" fn paimon_record_batch_reader_free(reader: *mut paimon_record_batch_reader) { @@ -1952,6 +2191,27 @@ const _: unsafe extern "C" fn( usize, ) -> paimon_result_read_builder = paimon_table_new_read_builder_with_options; +// Incremental-read ABI signature guards. Incremental plans intentionally use +// distinct opaque types from batch plans so callers cannot cross-cast them. +const _: unsafe extern "C" fn( + *const paimon_read_builder, + i32, + i64, + i64, +) -> paimon_result_incremental_scan = paimon_read_builder_new_incremental_scan; +const _: unsafe extern "C" fn(*mut paimon_incremental_scan) = paimon_incremental_scan_free; +const _: unsafe extern "C" fn(*const paimon_incremental_scan) -> paimon_result_incremental_plan = + paimon_incremental_scan_plan; +const _: unsafe extern "C" fn(*mut paimon_incremental_plan) = paimon_incremental_plan_free; +const _: unsafe extern "C" fn(*const paimon_incremental_plan) -> usize = + paimon_incremental_plan_num_splits; +const _: unsafe extern "C" fn( + *const paimon_table_read, + *const paimon_incremental_plan, + usize, + usize, +) -> paimon_result_record_batch_reader = paimon_table_read_to_incremental_arrow; + // Plan constructor ABI signature guard. Pins the symbol that builds a plan from // serialized split bytes so an accidental signature change fails to compile // rather than silently breaking header consumers. diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index e7da43068..ef6961cd6 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -342,6 +342,178 @@ unsafe fn read_rows_ffi(table: *const paimon_table) -> Vec<(i32, String)> { rows } +// ========================================================================= +// Incremental read tests +// ========================================================================= + +#[test] +fn test_incremental_delta_read_uses_left_open_snapshot_range() { + let path = "memory:/test_c_incremental_delta_range"; + let file_io = memory_file_io(); + setup_table_dirs(&file_io, path); + let table = Table::new( + file_io, + Identifier::new("default", "test"), + path.to_string(), + simple_table_schema(), + None, + ); + write_data_rust(&table, &[make_batch(vec![1], vec!["first"])]); + write_data_rust(&table, &[make_batch(vec![2], vec!["second"])]); + let handle = unsafe { wrap_table(table) }; + + unsafe { + let rb_result = paimon_table_new_read_builder(handle); + assert!(rb_result.error.is_null()); + let rb = rb_result.read_builder; + + let scan_result = + paimon_read_builder_new_incremental_scan(rb, PAIMON_INCREMENTAL_SCAN_MODE_DELTA, 1, 2); + assert!(scan_result.error.is_null()); + let scan = scan_result.scan; + + let plan_result = paimon_incremental_scan_plan(scan); + assert!(plan_result.error.is_null()); + let plan = plan_result.plan; + assert_eq!(paimon_incremental_plan_num_splits(plan), 1); + + let read_result = paimon_read_builder_new_read(rb); + assert!(read_result.error.is_null()); + let read = read_result.read; + + let reader_result = paimon_table_read_to_incremental_arrow(read, plan, 0, usize::MAX); + assert!(reader_result.error.is_null()); + let reader = reader_result.reader; + assert_eq!(collect_rows(reader), vec![(2, "second".to_string())]); + + paimon_record_batch_reader_free(reader); + paimon_table_read_free(read); + paimon_incremental_plan_free(plan); + paimon_incremental_scan_free(scan); + paimon_read_builder_free(rb); + unwrap_table(handle); + } +} + +#[test] +fn test_incremental_empty_snapshot_range_returns_empty_stream() { + let path = "memory:/test_c_incremental_empty_range"; + let file_io = memory_file_io(); + setup_table_dirs(&file_io, path); + let table = Table::new( + file_io, + Identifier::new("default", "test"), + path.to_string(), + simple_table_schema(), + None, + ); + write_data_rust(&table, &[make_batch(vec![1], vec!["first"])]); + let handle = unsafe { wrap_table(table) }; + + unsafe { + let rb = paimon_table_new_read_builder(handle).read_builder; + let scan_result = + paimon_read_builder_new_incremental_scan(rb, PAIMON_INCREMENTAL_SCAN_MODE_DELTA, 1, 1); + assert!(scan_result.error.is_null()); + let plan_result = paimon_incremental_scan_plan(scan_result.scan); + assert!(plan_result.error.is_null()); + assert_eq!(paimon_incremental_plan_num_splits(plan_result.plan), 0); + + let read_result = paimon_read_builder_new_read(rb); + assert!(read_result.error.is_null()); + let reader_result = paimon_table_read_to_incremental_arrow( + read_result.read, + plan_result.plan, + 0, + usize::MAX, + ); + assert!(reader_result.error.is_null()); + assert!(collect_rows(reader_result.reader).is_empty()); + + paimon_record_batch_reader_free(reader_result.reader); + paimon_table_read_free(read_result.read); + paimon_incremental_plan_free(plan_result.plan); + paimon_incremental_scan_free(scan_result.scan); + paimon_read_builder_free(rb); + unwrap_table(handle); + } +} + +#[test] +fn test_incremental_scan_rejects_unknown_mode() { + let table = Table::new( + memory_file_io(), + Identifier::new("default", "test"), + "memory:/test_c_incremental_unknown_mode".to_string(), + simple_table_schema(), + None, + ); + let handle = unsafe { wrap_table(table) }; + + unsafe { + let rb = paimon_table_new_read_builder(handle).read_builder; + let result = paimon_read_builder_new_incremental_scan(rb, 99, 0, 1); + assert!(result.scan.is_null()); + assert!(!result.error.is_null()); + assert_eq!((*result.error).code, PaimonErrorCode::InvalidInput as i32); + let message = std::str::from_utf8(std::slice::from_raw_parts( + (*result.error).message.data, + (*result.error).message.len, + )) + .unwrap(); + assert!(message.contains("incremental scan mode"), "got: {message}"); + + paimon_error_free(result.error); + paimon_read_builder_free(rb); + unwrap_table(handle); + } +} + +#[test] +fn test_incremental_read_preserves_builder_filter() { + let path = "memory:/test_c_incremental_filter"; + let file_io = memory_file_io(); + setup_table_dirs(&file_io, path); + let table = Table::new( + file_io, + Identifier::new("default", "test"), + path.to_string(), + simple_table_schema(), + None, + ); + write_data_rust(&table, &[make_batch(vec![1], vec!["first"])]); + write_data_rust(&table, &[make_batch(vec![2, 3], vec!["second", "third"])]); + let handle = unsafe { wrap_table(table) }; + + unsafe { + let rb = paimon_table_new_read_builder(handle).read_builder; + let predicate = build_predicate_equal(handle, "id", 3); + assert!(paimon_read_builder_with_filter(rb, predicate).is_null()); + + let scan_result = + paimon_read_builder_new_incremental_scan(rb, PAIMON_INCREMENTAL_SCAN_MODE_DELTA, 1, 2); + assert!(scan_result.error.is_null()); + let scan = scan_result.scan; + let plan_result = paimon_incremental_scan_plan(scan); + assert!(plan_result.error.is_null()); + let plan = plan_result.plan; + let read_result = paimon_read_builder_new_read(rb); + assert!(read_result.error.is_null()); + let read = read_result.read; + let reader_result = paimon_table_read_to_incremental_arrow(read, plan, 0, usize::MAX); + assert!(reader_result.error.is_null()); + let reader = reader_result.reader; + assert_eq!(collect_rows(reader), vec![(3, "third".to_string())]); + + paimon_record_batch_reader_free(reader); + paimon_table_read_free(read); + paimon_incremental_plan_free(plan); + paimon_incremental_scan_free(scan); + paimon_read_builder_free(rb); + unwrap_table(handle); + } +} + // ========================================================================= // Catalog-free table construction tests // ========================================================================= @@ -2058,6 +2230,31 @@ fn test_abort_commit() { #[test] fn test_null_pointer_handling() { unsafe { + let result = paimon_read_builder_new_incremental_scan( + ptr::null(), + PAIMON_INCREMENTAL_SCAN_MODE_DELTA, + 0, + 1, + ); + assert!(result.scan.is_null()); + assert!(!result.error.is_null()); + paimon_error_free(result.error); + + let result = paimon_incremental_scan_plan(ptr::null()); + assert!(result.plan.is_null()); + assert!(!result.error.is_null()); + paimon_error_free(result.error); + + let result = + paimon_table_read_to_incremental_arrow(ptr::null(), ptr::null(), 0, usize::MAX); + assert!(result.reader.is_null()); + assert!(!result.error.is_null()); + paimon_error_free(result.error); + + assert_eq!(paimon_incremental_plan_num_splits(ptr::null()), 0); + paimon_incremental_scan_free(ptr::null_mut()); + paimon_incremental_plan_free(ptr::null_mut()); + let result = paimon_table_new_write_builder(ptr::null()); assert!(!result.error.is_null()); assert!(result.write_builder.is_null()); diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs index 98ada0359..c6e2f1b55 100644 --- a/bindings/c/src/types.rs +++ b/bindings/c/src/types.rs @@ -21,10 +21,16 @@ use std::sync::Arc; use arrow_schema::Schema as ArrowSchema; use paimon::spec::{DataField, Predicate}; use paimon::table::{ - CommitMessage, PostponeBucketPlan, PostponeFixedBucketTableCommit, + CommitMessage, IncrementalScanMode, PostponeBucketPlan, PostponeFixedBucketTableCommit, PostponeFixedBucketTableWrite, Table, TableCommit, TableWrite, }; +/// Values accepted by `paimon_read_builder_new_incremental_scan`. +pub const PAIMON_INCREMENTAL_SCAN_MODE_DELTA: i32 = 0; +pub const PAIMON_INCREMENTAL_SCAN_MODE_CHANGELOG: i32 = 1; +pub const PAIMON_INCREMENTAL_SCAN_MODE_AUTO: i32 = 2; +pub const PAIMON_INCREMENTAL_SCAN_MODE_DIFF: i32 = 3; + /// C-compatible key-value pair for options. #[repr(C)] pub struct paimon_option { @@ -102,6 +108,22 @@ pub struct paimon_table_scan { pub inner: *mut c_void, } +/// Internal state for a fixed-range incremental scan. +pub(crate) struct IncrementalScanState { + pub table: Table, + pub projected_columns: Option>, + pub filter: Option, + pub case_sensitive: bool, + pub mode: IncrementalScanMode, + pub start_exclusive: i64, + pub end_inclusive: i64, +} + +#[repr(C)] +pub struct paimon_incremental_scan { + pub inner: *mut c_void, +} + #[repr(C)] pub struct paimon_table_read { pub inner: *mut c_void, @@ -119,6 +141,11 @@ pub struct paimon_plan { pub inner: *mut c_void, } +#[repr(C)] +pub struct paimon_incremental_plan { + pub inner: *mut c_void, +} + #[repr(C)] pub struct paimon_record_batch_reader { pub inner: *mut c_void, diff --git a/docs/src/c-binding.md b/docs/src/c-binding.md index 1b059505d..63c2170f4 100644 --- a/docs/src/c-binding.md +++ b/docs/src/c-binding.md @@ -20,8 +20,9 @@ under the License. # C Integration The C integration exposes Apache Paimon Rust through a C ABI. It provides -catalog and table access, scan planning, predicate push-down, streaming reads, -writes and commits, and vector search. Record batches cross the ABI through the +catalog and table access, batch and incremental scan planning, predicate +push-down, streaming reads, writes and commits, and vector search. Record +batches cross the ABI through the [Arrow C Data Interface](https://arrow.apache.org/docs/format/CDataInterface.html). The C binding is currently built from source. The repository does not check in @@ -221,6 +222,76 @@ range is clamped to the number of available splits. reversed: `paimon_table_write_write_arrow_batch` consumes the exported Arrow structures, so the caller must not release them again. +## Batch Incremental Reading + +The C API can plan and read a fixed snapshot range with +`(start_exclusive, end_inclusive]` semantics. For example, `(3, 5]` reads +changes from snapshots 4 and 5. The generated header defines these scan modes: + +| C mode | Behavior | +|--------|----------| +| `PAIMON_INCREMENTAL_SCAN_MODE_DELTA` | Reads data files added by `APPEND` snapshots. | +| `PAIMON_INCREMENTAL_SCAN_MODE_CHANGELOG` | Reads existing changelog files, skipping `OVERWRITE` snapshots and snapshots without changelog files. | +| `PAIMON_INCREMENTAL_SCAN_MODE_AUTO` | Selects Delta when `changelog-producer=none`; otherwise selects Changelog. | +| `PAIMON_INCREMENTAL_SCAN_MODE_DIFF` | Compares the complete table states at the start and end snapshots. | + +Create the incremental scan from the configured read builder, plan it, and +consume the plan with an incremental Arrow reader: + +```c +paimon_result_incremental_scan scan_result = + paimon_read_builder_new_incremental_scan( + read_builder, + PAIMON_INCREMENTAL_SCAN_MODE_DELTA, + 3, /* start_exclusive */ + 5 /* end_inclusive */); +CHECK_RESULT(scan_result); +paimon_incremental_scan *incremental_scan = scan_result.scan; + +paimon_result_incremental_plan plan_result = + paimon_incremental_scan_plan(incremental_scan); +CHECK_RESULT(plan_result); +paimon_incremental_plan *incremental_plan = plan_result.plan; + +paimon_result_new_read read_result = + paimon_read_builder_new_read(read_builder); +CHECK_RESULT(read_result); +paimon_table_read *read = read_result.read; + +size_t work_count = + paimon_incremental_plan_num_splits(incremental_plan); +paimon_result_record_batch_reader reader_result = + paimon_table_read_to_incremental_arrow( + read, incremental_plan, 0, work_count); +CHECK_RESULT(reader_result); +paimon_record_batch_reader *reader = reader_result.reader; + +/* Consume reader with paimon_record_batch_reader_next as shown above. */ + +paimon_record_batch_reader_free(reader); +paimon_table_read_free(read); +paimon_incremental_plan_free(incremental_plan); +paimon_incremental_scan_free(incremental_scan); +``` + +Projection and predicates configured before creating the incremental scan are +used during planning. Create the `paimon_table_read` from the same read builder +so the same projection and predicates are also applied while reading. + +`paimon_table_read_to_incremental_arrow` accepts an `offset` and `length` over +incremental work units. Delta, Changelog, and resolved Auto plans contain data +splits; Diff plans contain before/after split pairs, with each pair counted as +one unit. + +!!! note "Fixed-range API" + This API performs batch incremental reading over a fixed snapshot range. It + does not wait for future snapshots. A continuous consumer must discover new + snapshot IDs, create subsequent ranges, and persist its last successfully + consumed `end_inclusive` checkpoint. + +For mode semantics and current Diff restrictions, see +[Batch Incremental Reading](incremental-reading.md). + ## Projection and Predicates Projection uses a null-terminated array of column names: diff --git a/docs/src/incremental-reading.md b/docs/src/incremental-reading.md index 36e43db1a..26a296263 100644 --- a/docs/src/incremental-reading.md +++ b/docs/src/incremental-reading.md @@ -92,3 +92,14 @@ records by comparing the before and after images. If table option The start snapshot must still exist because `Diff` reads both endpoint states. An equal start and end snapshot produces an empty result. + +## C API + +The C binding exposes the same fixed-range incremental planner through +`paimon_read_builder_new_incremental_scan`, `paimon_incremental_scan_plan`, and +`paimon_table_read_to_incremental_arrow`. It supports all four modes listed +above and uses the same `(start_exclusive, end_inclusive]` range semantics. + +The returned reader is consumed with `paimon_record_batch_reader_next`, just +like a regular C batch read. See [C Integration](c-binding.md#batch-incremental-reading) +for a complete C example and resource-ownership rules.