diff --git a/LICENSE b/LICENSE index fc8d26538..a672c4a10 100644 --- a/LICENSE +++ b/LICENSE @@ -400,7 +400,8 @@ This product includes code from Apache ORC. * Upstream source patches: * cmake_modules/orc.diff * Adapted read-ahead cache: - * include/paimon/utils/read_ahead_cache.h (adapted from Cache.hh) + * include/paimon/utils/prefetch_cache_config.h (adapted from Cache.hh) + * src/paimon/common/utils/read_ahead_cache.h (adapted from Cache.hh) * src/paimon/common/utils/read_ahead_cache.cpp (adapted from Cache.cc) * Adapted byte range combiner: * src/paimon/common/utils/byte_range_combiner.h (adapted from Cache.hh) @@ -413,6 +414,18 @@ License: https://www.apache.org/licenses/LICENSE-2.0 -------------------------------------------------------------------------------- +This product includes code from Vortex 0.75.0. + +* Callback I/O sink utilities in crates/vortex_callback_io/callback_io.rs + (adapted from vortex-ffi/src/sink.rs) +* Upstream source patch in cmake_modules/vortex.diff + +Copyright the Vortex contributors +Home page: https://github.com/vortex-data/vortex +License: https://www.apache.org/licenses/LICENSE-2.0 + +-------------------------------------------------------------------------------- + This product includes code from xxHash. * MMH_rotl32 utility in src/paimon/common/utils/murmurhash_utils.h diff --git a/crates/vortex_callback_io/callback_io.rs b/crates/vortex_callback_io/callback_io.rs index f4d5fb857..d3d2faadc 100644 --- a/crates/vortex_callback_io/callback_io.rs +++ b/crates/vortex_callback_io/callback_io.rs @@ -14,37 +14,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Callback-based I/O for the Vortex C API. +//! Callback I/O for Vortex using Paimon's `InputStream` and `OutputStream`. //! -//! The stock C API can only read a whole file already materialized in memory -//! (`vx_data_source_new_buffer`) or a path that Vortex itself resolves -//! (`vx_data_source_new`), and it can only write to a local filesystem path -//! (`vx_array_sink_open_file`, which uses `async_fs::File::create`). paimon owns its own pluggable -//! filesystem, so none of these fit: reading forces the entire file into memory, and writing forces -//! a local temporary file that then has to be copied back. +//! Copied into `vortex-ffi/src/` at build time; see `cmake_modules/vortex.diff`. +//! Keep FFI types and signatures in sync with `src/paimon/format/vortex/vortex_ffi.h`. //! -//! This module adds entry points that take host callbacks instead, so Vortex issues positional -//! reads and sequential writes straight against paimon's `InputStream` / `OutputStream`. Vortex's -//! core I/O traits are the plug points; the shape of the bridge mirrors `vortex-jni`'s -//! `JavaReadable` and `JavaWrite`. -//! -//! The write side needs its own handle type rather than reusing `vx_array_sink`, whose fields are -//! private to `crate::sink`. -//! -//! The sink entry points below (`vx_callback_sink_push` / `_close` / `_abort`) are adapted from -//! vortex-ffi's `src/sink.rs` (SPDX-License-Identifier: Apache-2.0, Copyright the Vortex -//! contributors), reworked to drive the write pipeline through host callbacks instead of a -//! Vortex-resolved filesystem path. -//! -//! This file is maintained in the paimon-cpp tree (`crates/vortex_callback_io/`) and copied into -//! `vortex-ffi/src/` at build time; see `cmake_modules/vortex.diff`. -//! -//! The C declarations that mirror the types and entry points below live in paimon-cpp's -//! `src/paimon/format/vortex/vortex_ffi.h` (cbindgen does not see this module, so they are written -//! by hand). The two sides are only checked by human review, and a mismatch in field order or a -//! function signature is silent memory corruption, not a compile error. Any change to a -//! `#[repr(C)]` struct, a callback `type`, or a `#[no_mangle]` entry point here MUST be mirrored in -//! that header in the same change. +//! Sink entry points are adapted from Vortex's `vortex-ffi/src/sink.rs` +//! (SPDX-License-Identifier: Apache-2.0, Copyright the Vortex contributors). use std::ffi::c_void; use std::io; @@ -137,11 +113,7 @@ pub struct vx_output_callbacks { pub release_fn: Option, } -/// The host context, owned by the reader. -/// -/// Held behind an `Arc` so in-flight reads keep the context alive, and so `release_fn` runs exactly -/// once when the last reference goes away. It carries only the context, not the read callback, so -/// that ownership can be taken over before anything else is validated. +/// Shared host context, released when the last reader or writer reference is dropped. struct CallbackTarget { ctx: *mut c_void, release_fn: Option, @@ -164,11 +136,7 @@ impl Drop for CallbackTarget { } } -/// A [`VortexReadAt`] that forwards positional reads to host callbacks. -/// -/// Reads run on the runtime's blocking pool, since the host callback is synchronous and must not -/// occupy an async executor thread. The file size is supplied at construction time, so `size()` -/// never crosses the FFI boundary. +/// A [`VortexReadAt`] that runs host callbacks on the blocking pool. struct CallbackReadAt { target: Arc, read_at_fn: vx_read_at_fn, @@ -177,16 +145,12 @@ struct CallbackReadAt { concurrency: usize, } -/// Number of concurrent host read callbacks Vortex may have in flight for one file. -/// -/// The host filesystem may be remote, where concurrency hides latency; it is also the bound on -/// blocking threads occupied by one reader. +/// Maximum concurrent host reads per file, also bounding blocking-pool usage. const DEFAULT_CONCURRENCY: usize = 16; impl VortexReadAt for CallbackReadAt { fn coalesce_config(&self) -> Option { - // Each read costs an FFI hop plus a blocking-pool hand-off, and the host filesystem may be - // remote, so favor fewer and larger reads. + // Coalesce reads to reduce remote I/O and blocking-pool overhead. Some(CoalesceConfig::object_storage()) } @@ -255,8 +219,7 @@ unsafe fn data_source_new_callback( callbacks: vx_input_callbacks, size: u64, ) -> VortexResult<*const vx_data_source> { - // Take ownership of the host context before anything else, so an early error below still - // releases it exactly once when this `Arc` is dropped. + // Take ownership before validation so errors also release the context. let target = Arc::new(CallbackTarget { ctx: callbacks.ctx, release_fn: callbacks.release_fn, @@ -282,11 +245,8 @@ unsafe fn data_source_new_callback( /// Create a data source that reads through host callbacks. /// -/// `size` is the total length of the file in bytes; the host knows it up front, so it is passed -/// here instead of being fetched through another callback. -/// -/// The callbacks (and the context they carry) are owned by the returned data source and released -/// when it is freed, including when this call fails. +/// `size` is the total file length in bytes. Takes ownership of the host context, +/// which is released when the data source is freed or this call fails. /// /// On error, returns NULL and sets "err". /// @@ -306,11 +266,7 @@ pub unsafe extern "C-unwind" fn vx_data_source_new_callback( }) } -/// A [`VortexWrite`] that forwards sequential writes to host callbacks. -/// -/// Writes run inline on the runtime thread driving the write task, the same way `vortex-jni` -/// forwards to Java sinks: the host callback is synchronous and the layout writer hands over one -/// buffer at a time, so there is nothing to overlap. +/// A [`VortexWrite`] that calls host callbacks inline on the writer's runtime thread. struct CallbackWrite { target: Arc, write_fn: vx_write_fn, @@ -364,11 +320,7 @@ impl VortexWrite for CallbackWrite { } } -/// A sink that writes a Vortex file through host callbacks. -/// -/// Mirrors `vx_array_sink`: pushed arrays go through a channel feeding a write task on the session's -/// runtime, and errors surface when the sink is closed. This is a separate type because -/// `vx_array_sink`'s fields are private to its own module. +/// A Vortex file sink backed by host callbacks and a background write task. pub struct vx_callback_sink { sink: Sender>, writer: Task>, @@ -380,8 +332,7 @@ unsafe fn callback_sink_open( callbacks: vx_output_callbacks, dtype: *const vx_dtype, ) -> VortexResult<*mut vx_callback_sink> { - // Take ownership of the host context before anything else, so an early error below still - // releases it exactly once when this `Arc` is dropped. + // Take ownership before validation so errors also release the context. let target = Arc::new(CallbackTarget { ctx: callbacks.ctx, release_fn: callbacks.release_fn, @@ -401,7 +352,7 @@ unsafe fn callback_sink_open( flush_fn: callbacks.flush_fn, }; - // Small on purpose so the producing thread feels back-pressure, as in vortex-jni's writer. + // Limit queued arrays to apply backpressure to the producer. let (sink, rx) = mpsc::channel(4); let array_stream = ArrayStreamAdapter::new(dtype.clone(), rx.into_stream()); @@ -557,7 +508,6 @@ mod tests { use crate::sink::vx_array_sink_open_file; use crate::sink::vx_array_sink_push; - /// Host context standing in for paimon's `InputStream`: positional reads against a file. struct FileContext { file: File, } @@ -584,8 +534,7 @@ mod tests { drop(unsafe { Box::from_raw(ctx.cast::()) }); } - /// Writing a file with the stock sink and reading it back through the callback data source - /// must yield the same dtype and row count: the callback plug point is wired up correctly. + // Read a file written by the standard sink through host callbacks. #[test] #[cfg_attr(miri, ignore)] fn test_callback_data_source_round_trip() { @@ -638,8 +587,6 @@ mod tests { } } - /// A rejected request must still release the host context: the entry point takes ownership - /// before it validates anything, so the host never has to guess whether to clean up. #[test] #[cfg_attr(miri, ignore)] fn test_rejected_callbacks_release_context() { @@ -652,7 +599,6 @@ mod tests { unsafe { let session = vx_session_new(); let mut error = ptr::null_mut(); - // No read callback, so the call must fail. let callbacks = vx_input_callbacks { ctx: ptr::null_mut(), read_at_fn: None, @@ -672,12 +618,10 @@ mod tests { } } - /// Writing through the callback sink and reading the bytes back through the callback data source - /// must round-trip, with no filesystem involved in either direction. + // Round-trip through an in-memory buffer using only callbacks. #[test] #[cfg_attr(miri, ignore)] fn test_callback_sink_round_trip() { - /// Host context standing in for a paimon stream pair: an in-memory byte buffer. struct BufferContext { data: Mutex>, }