diff --git a/Cargo.lock b/Cargo.lock index dbf1a2ad0..87abd75d6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4644,7 +4644,9 @@ dependencies = [ "paimon", "paimon-vindex-core", "serde_json", + "tempfile", "tokio", + "url", ] [[package]] diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml index 51e3731c6..6e3604c5b 100644 --- a/bindings/c/Cargo.toml +++ b/bindings/c/Cargo.toml @@ -31,7 +31,13 @@ crate-type = ["cdylib", "staticlib", "rlib"] doc = false [dependencies] -paimon = { path = "../../crates/paimon" } +paimon = { path = "../../crates/paimon", features = [ + "storage-azdls", + "storage-cos", + "storage-gcs", + "storage-obs", + "storage-s3", +] } tokio = { workspace = true, features = ["rt-multi-thread"] } futures = "0.3" arrow = { workspace = true } @@ -44,3 +50,5 @@ serde_json = "1.0.120" # IVF-flat ANN segment fixture in-process. Versions match crates/paimon. bytes = "1.7.1" paimon-vindex-core = "0.4.0" +tempfile = "3" +url = "2.5.2" diff --git a/bindings/c/DEPENDENCIES.rust.tsv b/bindings/c/DEPENDENCIES.rust.tsv index ebc06fa48..ba4f75585 100644 --- a/bindings/c/DEPENDENCIES.rust.tsv +++ b/bindings/c/DEPENDENCIES.rust.tsv @@ -1,5 +1,6 @@ crate 0BSD Apache-2.0 Apache-2.0 WITH LLVM-exception BSD-2-Clause BSD-3-Clause BSL-1.0 CC0-1.0 CDLA-Permissive-2.0 ISC LGPL-2.1-or-later MIT MIT-0 Unicode-3.0 Unlicense Zlib adler2@2.0.1 X X X +aes@0.8.4 X X ahash@0.8.12 X X aho-corasick@1.1.4 X X alloc-no-stdlib@2.0.4 X @@ -33,10 +34,12 @@ aws-lc-rs@1.17.3 X X aws-lc-sys@0.43.0 X X X X X backon@1.6.0 X base64@0.22.1 X X +base64ct@1.8.3 X X bigdecimal@0.4.10 X X bitflags@2.13.1 X X block-buffer@0.10.4 X X block-buffer@0.12.1 X X +block-padding@0.3.3 X X bon@3.9.3 X X bon-macros@3.9.3 X X brotli@8.0.4 X X @@ -45,15 +48,18 @@ bumpalo@3.20.3 X X bytemuck@1.25.2 X X X byteorder@1.5.0 X X bytes@1.12.1 X +cbc@0.1.2 X X cc@1.3.0 X X cfg-if@1.0.4 X X chrono@0.4.45 X X chrono-tz@0.10.4 X X +cipher@0.4.4 X X cmake@0.1.58 X X cmov@0.5.4 X X combine@4.6.7 X comfy-table@7.2.2 X const-oid@0.10.2 X X +const-oid@0.9.6 X X const-random@0.1.18 X X const-random-macro@0.1.16 X X core-foundation@0.10.1 X X @@ -61,6 +67,7 @@ core-foundation@0.9.4 X X core-foundation-sys@0.8.7 X X cpufeatures@0.2.17 X X cpufeatures@0.3.0 X X +crc-fast@1.10.0 X X crc32fast@1.5.0 X X crossbeam-channel@0.5.16 X X crossbeam-deque@0.8.7 X X @@ -75,6 +82,7 @@ ctutils@0.4.2 X X darling@0.23.0 X darling_core@0.23.0 X darling_macro@0.23.0 X +der@0.7.10 X X diff@0.1.13 X X digest@0.10.7 X X digest@0.11.3 X X @@ -141,6 +149,7 @@ ident_case@1.0.1 X X idna@1.1.0 X X idna_adapter@1.2.2 X X indexmap@2.14.0 X X +inout@0.1.4 X X integer-encoding@3.0.4 X ipnet@2.12.0 X X itertools@0.14.0 X X @@ -155,6 +164,7 @@ jni-sys@0.4.1 X X jni-sys-macros@0.4.1 X X jobserver@0.1.35 X X js-sys@0.3.103 X X +lazy_static@1.5.0 X X lexical-core@1.0.6 X X lexical-parse-float@1.0.6 X X lexical-parse-integer@1.0.6 X X @@ -184,6 +194,7 @@ nalgebra-macros@0.2.2 X native-tls@0.2.18 X X num@0.4.3 X X num-bigint@0.4.8 X X +num-bigint-dig@0.8.6 X X num-complex@0.4.6 X X num-integer@0.1.46 X X num-iter@0.1.46 X X @@ -193,8 +204,14 @@ once_cell@1.21.4 X X opendal-core@0.58.0 X opendal-http-transport-reqwest@0.58.0 X opendal-layer-retry@0.58.0 X +opendal-service-azdls@0.58.0 X +opendal-service-azure-common@0.58.0 X +opendal-service-cos@0.58.0 X opendal-service-fs@0.58.0 X +opendal-service-gcs@0.58.0 X +opendal-service-obs@0.58.0 X opendal-service-oss@0.58.0 X +opendal-service-s3@0.58.0 X openssl@0.10.81 X openssl-macros@0.1.1 X X openssl-probe@0.2.1 X X @@ -208,10 +225,16 @@ paimon-mosaic-core@0.2.0 X paimon-vindex-core@0.4.0 X parquet@58.3.0 X paste@1.0.15 X X +pbkdf2@0.12.2 X X +pem@3.0.6 X +pem-rfc7468@0.7.0 X X percent-encoding@2.3.2 X X phf@0.12.1 X phf_shared@0.12.1 X pin-project-lite@0.2.17 X X +pkcs1@0.7.5 X X +pkcs5@0.7.1 X X +pkcs8@0.10.2 X X pkg-config@0.3.33 X X portable-atomic@1.14.0 X X portable-atomic-util@0.2.7 X X @@ -241,11 +264,17 @@ regex-automata@0.4.16 X X regex-lite@0.1.9 X X regex-syntax@0.8.11 X X reqsign-aliyun-oss@3.1.1 X +reqsign-aws-v4@3.0.2 X +reqsign-azure-storage@3.1.0 X reqsign-core@3.1.0 X reqsign-file-read-tokio@3.0.2 X +reqsign-google@3.0.2 X +reqsign-huaweicloud-obs@3.0.2 X +reqsign-tencent-cos@3.0.2 X reqwest@0.12.28 X X reqwest@0.13.4 X X roaring@0.11.4 X X +rsa@0.9.10 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X @@ -258,8 +287,10 @@ rustls-webpki@0.103.13 X rustversion@1.0.23 X X ryu@1.0.23 X X safe_arch@0.7.4 X X X +salsa20@0.10.2 X X same-file@1.0.6 X X schannel@0.1.29 X +scrypt@0.11.0 X X security-framework@3.7.0 X X security-framework-sys@2.17.0 X X semver@1.0.28 X X @@ -278,6 +309,7 @@ sha1@0.11.0 X X sha2@0.10.9 X X sha2@0.11.0 X X shlex@2.0.1 X X +signature@2.2.0 X X simba@0.9.1 X simd-adler32@0.3.10 X simd_cesu8@1.2.0 X X @@ -291,6 +323,9 @@ snafu-derive@0.8.9 X X snafu-derive@0.9.1 X X snap@1.1.2 X socket2@0.6.5 X X +spin@0.10.1 X +spin@0.9.9 X +spki@0.7.3 X X stable_deref_trait@1.2.1 X X strsim@0.11.1 X strum@0.27.2 X diff --git a/bindings/c/src/blob_reader.rs b/bindings/c/src/blob_reader.rs new file mode 100644 index 000000000..75ce5433d --- /dev/null +++ b/bindings/c/src/blob_reader.rs @@ -0,0 +1,326 @@ +// 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 std::collections::HashMap; +use std::ffi::c_void; +use std::io::SeekFrom; + +use paimon::{BlobReader, BlobStream}; + +use crate::error::{check_non_null, paimon_error, validate_cstr, PaimonErrorCode}; +use crate::result::{ + paimon_result_blob_reader, paimon_result_blob_stream, paimon_result_blob_stream_read, + paimon_result_blob_stream_seek, paimon_result_read_blobs, +}; +use crate::runtime; +use crate::types::{ + paimon_blob_reader, paimon_blob_stream, paimon_byte_slice, paimon_bytes_array, paimon_option, + paimon_table, +}; + +fn new_reader(reader: BlobReader) -> paimon_result_blob_reader { + let reader = Box::new(reader); + let wrapper = Box::new(paimon_blob_reader { + inner: Box::into_raw(reader) as *mut c_void, + }); + paimon_result_blob_reader { + reader: Box::into_raw(wrapper), + error: std::ptr::null_mut(), + } +} + +fn read_error(error: *mut paimon_error) -> paimon_result_read_blobs { + paimon_result_read_blobs { + blobs: paimon_bytes_array::empty(), + error, + } +} + +fn stream_error(error: *mut paimon_error) -> paimon_result_blob_stream { + paimon_result_blob_stream { + stream: std::ptr::null_mut(), + error, + } +} + +/// # Safety +/// `options` is null for zero length or points to valid UTF-8 C-string pairs. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_reader_new( + options: *const paimon_option, + options_len: usize, +) -> paimon_result_blob_reader { + if options_len > 0 && options.is_null() { + return paimon_result_blob_reader { + reader: std::ptr::null_mut(), + error: paimon_error::new( + PaimonErrorCode::InvalidInput, + "null pointer passed for `options`".to_string(), + ), + }; + } + + let mut storage_options = HashMap::with_capacity(options_len); + if options_len > 0 { + for option in std::slice::from_raw_parts(options, options_len) { + let key = match validate_cstr(option.key, "option key") { + Ok(value) => value, + Err(error) => { + return paimon_result_blob_reader { + reader: std::ptr::null_mut(), + error, + }; + } + }; + let value = match validate_cstr(option.value, "option value") { + Ok(value) => value, + Err(error) => { + return paimon_result_blob_reader { + reader: std::ptr::null_mut(), + error, + }; + } + }; + storage_options.insert(key, value); + } + } + + new_reader(BlobReader::new(storage_options)) +} + +/// Create a reader using a table's FileIO. +/// +/// # Safety +/// `table` is a valid handle returned by the Paimon C API. +#[no_mangle] +pub unsafe extern "C" fn paimon_table_new_blob_reader( + table: *const paimon_table, +) -> paimon_result_blob_reader { + if let Err(error) = check_non_null(table, "table") { + return paimon_result_blob_reader { + reader: std::ptr::null_mut(), + error, + }; + } + + let table = &*((*table).inner as *const paimon::Table); + new_reader(BlobReader::from_file_io(table.file_io().clone())) +} + +/// # Safety +/// The handle and input slices are valid for this call. Free the output with +/// `paimon_bytes_array_free`. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_reader_read_blobs( + reader: *const paimon_blob_reader, + descriptors: *const paimon_byte_slice, + descriptors_len: usize, +) -> paimon_result_read_blobs { + if let Err(error) = check_non_null(reader, "blob reader") { + return read_error(error); + } + if descriptors_len > 0 && descriptors.is_null() { + return read_error(paimon_error::new( + PaimonErrorCode::InvalidInput, + "null pointer passed for `descriptors`".to_string(), + )); + } + + let mut owned = Vec::with_capacity(descriptors_len); + if descriptors_len > 0 { + for (index, descriptor) in std::slice::from_raw_parts(descriptors, descriptors_len) + .iter() + .enumerate() + { + if descriptor.len > 0 && descriptor.data.is_null() { + return read_error(paimon_error::new( + PaimonErrorCode::InvalidInput, + format!( + "null data pointer for BlobDescriptor input index {index}, URI unavailable" + ), + )); + } + let bytes = if descriptor.len == 0 { + &[] + } else { + std::slice::from_raw_parts(descriptor.data, descriptor.len) + }; + owned.push(bytes.to_vec()); + } + } + + let reader = &*((*reader).inner as *const BlobReader); + match runtime().block_on(reader.read_blobs(&owned)) { + Ok(values) => paimon_result_read_blobs { + blobs: paimon_bytes_array::new(values), + error: std::ptr::null_mut(), + }, + Err(error) => read_error(paimon_error::from_paimon(error)), + } +} + +/// Open one descriptor for incremental reads. +/// +/// # Safety +/// `reader` is valid and `descriptor` points to `descriptor_len` bytes. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_reader_open_blob( + reader: *const paimon_blob_reader, + descriptor: *const u8, + descriptor_len: usize, +) -> paimon_result_blob_stream { + if let Err(error) = check_non_null(reader, "blob reader") { + return stream_error(error); + } + if descriptor_len > 0 && descriptor.is_null() { + return stream_error(paimon_error::new( + PaimonErrorCode::InvalidInput, + "null pointer passed for `descriptor`".to_string(), + )); + } + let bytes = if descriptor_len == 0 { + &[] + } else { + std::slice::from_raw_parts(descriptor, descriptor_len) + }; + let reader = &*((*reader).inner as *const BlobReader); + match reader.open_blob(bytes) { + Ok(stream) => { + let stream = Box::new(stream); + let wrapper = Box::new(paimon_blob_stream { + inner: Box::into_raw(stream) as *mut c_void, + }); + paimon_result_blob_stream { + stream: Box::into_raw(wrapper), + error: std::ptr::null_mut(), + } + } + Err(error) => stream_error(paimon_error::from_paimon(error)), + } +} + +/// Read at most `buffer_len` bytes into caller-owned memory. +/// +/// A zero `bytes_read` result means end of stream when `buffer_len` is nonzero. +/// +/// # Safety +/// `stream` is valid and `buffer` points to `buffer_len` writable bytes. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_stream_read( + stream: *mut paimon_blob_stream, + buffer: *mut u8, + buffer_len: usize, +) -> paimon_result_blob_stream_read { + if let Err(error) = check_non_null(stream, "blob stream") { + return paimon_result_blob_stream_read { + bytes_read: 0, + error, + }; + } + if buffer_len > 0 && buffer.is_null() { + return paimon_result_blob_stream_read { + bytes_read: 0, + error: paimon_error::new( + PaimonErrorCode::InvalidInput, + "null pointer passed for `buffer`".to_string(), + ), + }; + } + + let stream = &mut *((*stream).inner as *mut BlobStream); + match runtime().block_on(stream.read(buffer_len)) { + Ok(bytes) => { + if !bytes.is_empty() { + std::ptr::copy_nonoverlapping(bytes.as_ptr(), buffer, bytes.len()); + } + paimon_result_blob_stream_read { + bytes_read: bytes.len(), + error: std::ptr::null_mut(), + } + } + Err(error) => paimon_result_blob_stream_read { + bytes_read: 0, + error: paimon_error::from_paimon(error), + }, + } +} + +/// Seek within the descriptor's range. `whence` uses the standard 0, 1, 2 values. +/// +/// # Safety +/// `stream` is valid. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_stream_seek( + stream: *mut paimon_blob_stream, + offset: i64, + whence: i32, +) -> paimon_result_blob_stream_seek { + if let Err(error) = check_non_null(stream, "blob stream") { + return paimon_result_blob_stream_seek { position: 0, error }; + } + let from = match whence { + 0 if offset >= 0 => SeekFrom::Start(offset as u64), + 1 => SeekFrom::Current(offset), + 2 => SeekFrom::End(offset), + _ => { + return paimon_result_blob_stream_seek { + position: 0, + error: paimon_error::new( + PaimonErrorCode::InvalidInput, + "invalid blob stream seek".to_string(), + ), + }; + } + }; + let stream = &mut *((*stream).inner as *mut BlobStream); + match runtime().block_on(stream.seek(from)) { + Ok(position) => paimon_result_blob_stream_seek { + position, + error: std::ptr::null_mut(), + }, + Err(error) => paimon_result_blob_stream_seek { + position: 0, + error: paimon_error::from_paimon(error), + }, + } +} + +/// # Safety +/// `stream` is null or was returned by `paimon_blob_reader_open_blob`. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_stream_free(stream: *mut paimon_blob_stream) { + if stream.is_null() { + return; + } + let stream = Box::from_raw(stream); + if !stream.inner.is_null() { + drop(Box::from_raw(stream.inner as *mut BlobStream)); + } +} + +/// # Safety +/// `reader` is null or was returned by `paimon_blob_reader_new`. +#[no_mangle] +pub unsafe extern "C" fn paimon_blob_reader_free(reader: *mut paimon_blob_reader) { + if reader.is_null() { + return; + } + let reader = Box::from_raw(reader); + if !reader.inner.is_null() { + drop(Box::from_raw(reader.inner as *mut BlobReader)); + } +} diff --git a/bindings/c/src/lib.rs b/bindings/c/src/lib.rs index 27972b5f1..2c1af96fa 100644 --- a/bindings/c/src/lib.rs +++ b/bindings/c/src/lib.rs @@ -19,6 +19,7 @@ // So it's type node can't meet camel case. #![allow(non_camel_case_types)] +mod blob_reader; mod catalog; mod error; mod identifier; diff --git a/bindings/c/src/result.rs b/bindings/c/src/result.rs index 94f87c2de..808b64913 100644 --- a/bindings/c/src/result.rs +++ b/bindings/c/src/result.rs @@ -18,6 +18,36 @@ use crate::error::paimon_error; use crate::types::*; +#[repr(C)] +pub struct paimon_result_blob_reader { + pub reader: *mut paimon_blob_reader, + pub error: *mut paimon_error, +} + +#[repr(C)] +pub struct paimon_result_blob_stream { + pub stream: *mut paimon_blob_stream, + pub error: *mut paimon_error, +} + +#[repr(C)] +pub struct paimon_result_blob_stream_read { + pub bytes_read: usize, + pub error: *mut paimon_error, +} + +#[repr(C)] +pub struct paimon_result_blob_stream_seek { + pub position: u64, + pub error: *mut paimon_error, +} + +#[repr(C)] +pub struct paimon_result_read_blobs { + pub blobs: paimon_bytes_array, + pub error: *mut paimon_error, +} + #[repr(C)] pub struct paimon_result_catalog_new { pub catalog: *mut paimon_catalog, diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index 37fa0c919..af181e398 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -38,9 +38,12 @@ use arrow_array::{Array, Int32Array, RecordBatch, StringArray, StructArray}; use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as ArrowSchema}; use paimon::catalog::Identifier; use paimon::io::FileIOBuilder; -use paimon::spec::{CommitKind, DataType, IntType, Schema, TableSchema, VarCharType}; +use paimon::spec::{ + BlobDescriptor, CommitKind, DataType, IntType, Schema, TableSchema, VarCharType, +}; use paimon::table::{SnapshotManager, Table}; +use crate::blob_reader::*; use crate::error::*; use crate::table::*; use crate::types::*; @@ -3311,3 +3314,218 @@ fn vector_search_projection_unknown_column_errors_at_execute_read() { unwrap_table(handle); } } + +#[test] +fn blob_reader_reads_batch_and_owns_output_buffers() { + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), b"abcdefghij").unwrap(); + let uri = url::Url::from_file_path(file.path()).unwrap().to_string(); + let mut descriptors = vec![ + BlobDescriptor::new(uri.clone(), 3, -1).serialize(), + BlobDescriptor::new(uri.clone(), 1, 3).serialize(), + BlobDescriptor::new(uri, 5, 0).serialize(), + ]; + let slices = descriptors + .iter() + .map(|value| paimon_byte_slice { + data: value.as_ptr(), + len: value.len(), + }) + .collect::>(); + + unsafe { + let created = paimon_blob_reader_new(ptr::null(), 0); + assert!(created.error.is_null()); + assert!(!created.reader.is_null()); + + let result = paimon_blob_reader_read_blobs(created.reader, slices.as_ptr(), slices.len()); + assert!(result.error.is_null()); + assert_eq!(result.blobs.len, 3); + + descriptors.clear(); + paimon_blob_reader_free(created.reader); + let values = std::slice::from_raw_parts(result.blobs.data, result.blobs.len) + .iter() + .map(|value| std::slice::from_raw_parts(value.data, value.len).to_vec()) + .collect::>(); + assert_eq!( + values, + vec![b"defghij".to_vec(), b"bcd".to_vec(), Vec::new()] + ); + paimon_bytes_array_free(result.blobs); + } +} + +#[test] +fn blob_reader_from_table_keeps_file_io_alive() { + let file_io = memory_file_io(); + let uri = "memory:/blob_reader_from_table"; + crate::runtime().block_on(async { + file_io + .new_output(uri) + .unwrap() + .write(bytes::Bytes::from_static(b"abcdefghij")) + .await + .unwrap(); + }); + let table = Table::new( + file_io, + Identifier::new("default", "blob_table"), + "memory:/blob_table".to_string(), + simple_table_schema(), + None, + ); + let table = unsafe { wrap_table(table) }; + let descriptor = BlobDescriptor::new(uri.to_string(), 2, 4).serialize(); + let descriptor_slice = paimon_byte_slice { + data: descriptor.as_ptr(), + len: descriptor.len(), + }; + + unsafe { + let created = paimon_table_new_blob_reader(table); + assert!(created.error.is_null()); + assert!(!created.reader.is_null()); + unwrap_table(table); + + let result = paimon_blob_reader_read_blobs(created.reader, &descriptor_slice, 1); + assert!(result.error.is_null()); + let values = std::slice::from_raw_parts(result.blobs.data, result.blobs.len); + assert_eq!( + std::slice::from_raw_parts(values[0].data, values[0].len), + b"cdef" + ); + + paimon_bytes_array_free(result.blobs); + paimon_blob_reader_free(created.reader); + } +} + +#[test] +fn blob_reader_handles_empty_and_error_batches() { + unsafe { + let null_table = paimon_table_new_blob_reader(ptr::null()); + assert!(null_table.reader.is_null()); + assert!(!null_table.error.is_null()); + paimon_error_free(null_table.error); + + let created = paimon_blob_reader_new(ptr::null(), 0); + assert!(created.error.is_null()); + + let empty = paimon_blob_reader_read_blobs(created.reader, ptr::null(), 0); + assert!(empty.error.is_null()); + assert!(empty.blobs.data.is_null()); + assert_eq!(empty.blobs.len, 0); + paimon_bytes_array_free(empty.blobs); + + let invalid_bytes = [0_u8; 1]; + let invalid_slice = paimon_byte_slice { + data: invalid_bytes.as_ptr(), + len: invalid_bytes.len(), + }; + let invalid = paimon_blob_reader_read_blobs(created.reader, &invalid_slice, 1); + assert!(!invalid.error.is_null()); + assert!(invalid.blobs.data.is_null()); + paimon_error_free(invalid.error); + + let null_slice = paimon_byte_slice { + data: ptr::null(), + len: 1, + }; + let null_data = paimon_blob_reader_read_blobs(created.reader, &null_slice, 1); + assert!(!null_data.error.is_null()); + assert!(null_data.blobs.data.is_null()); + paimon_error_free(null_data.error); + + paimon_blob_reader_free(created.reader); + paimon_blob_reader_free(ptr::null_mut()); + } +} + +#[test] +fn blob_stream_reads_chunks_and_outlives_reader() { + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), b"abcdefghij").unwrap(); + let uri = url::Url::from_file_path(file.path()).unwrap().to_string(); + let descriptor = BlobDescriptor::new(uri, 2, 5).serialize(); + + unsafe { + let created = paimon_blob_reader_new(ptr::null(), 0); + assert!(created.error.is_null()); + let opened = + paimon_blob_reader_open_blob(created.reader, descriptor.as_ptr(), descriptor.len()); + assert!(opened.error.is_null()); + assert!(!opened.stream.is_null()); + paimon_blob_reader_free(created.reader); + + let mut buffer = [0xFF_u8; 3]; + let first = paimon_blob_stream_read(opened.stream, buffer.as_mut_ptr(), buffer.len()); + assert!(first.error.is_null()); + assert_eq!(first.bytes_read, 3); + assert_eq!(&buffer, b"cde"); + + let seek = paimon_blob_stream_seek(opened.stream, -2, 2); + assert!(seek.error.is_null()); + assert_eq!(seek.position, 3); + + buffer.fill(0xFF); + let second = paimon_blob_stream_read(opened.stream, buffer.as_mut_ptr(), buffer.len()); + assert!(second.error.is_null()); + assert_eq!(second.bytes_read, 2); + assert_eq!(&buffer[..2], b"fg"); + assert_eq!(buffer[2], 0xFF); + + let end = paimon_blob_stream_read(opened.stream, buffer.as_mut_ptr(), buffer.len()); + assert!(end.error.is_null()); + assert_eq!(end.bytes_read, 0); + + paimon_blob_stream_free(opened.stream); + paimon_blob_stream_free(ptr::null_mut()); + } +} + +#[test] +fn blob_stream_validates_handles_and_buffers() { + unsafe { + let null_reader = paimon_blob_reader_open_blob(ptr::null(), ptr::null(), 0); + assert!(null_reader.stream.is_null()); + assert!(!null_reader.error.is_null()); + paimon_error_free(null_reader.error); + + let created = paimon_blob_reader_new(ptr::null(), 0); + let invalid = paimon_blob_reader_open_blob(created.reader, ptr::null(), 0); + assert!(invalid.stream.is_null()); + assert!(!invalid.error.is_null()); + paimon_error_free(invalid.error); + + let file = tempfile::NamedTempFile::new().unwrap(); + let uri = url::Url::from_file_path(file.path()).unwrap().to_string(); + let descriptor = BlobDescriptor::new(uri, 0, 0).serialize(); + let opened = + paimon_blob_reader_open_blob(created.reader, descriptor.as_ptr(), descriptor.len()); + assert!(opened.error.is_null()); + + let null_buffer = paimon_blob_stream_read(opened.stream, ptr::null_mut(), 1); + assert!(!null_buffer.error.is_null()); + paimon_error_free(null_buffer.error); + + let zero = paimon_blob_stream_read(opened.stream, ptr::null_mut(), 0); + assert!(zero.error.is_null()); + assert_eq!(zero.bytes_read, 0); + + let null_stream = paimon_blob_stream_read(ptr::null_mut(), ptr::null_mut(), 0); + assert!(!null_stream.error.is_null()); + paimon_error_free(null_stream.error); + + let invalid_seek = paimon_blob_stream_seek(opened.stream, -1, 0); + assert!(!invalid_seek.error.is_null()); + paimon_error_free(invalid_seek.error); + + let null_seek = paimon_blob_stream_seek(ptr::null_mut(), 0, 0); + assert!(!null_seek.error.is_null()); + paimon_error_free(null_seek.error); + + paimon_blob_stream_free(opened.stream); + paimon_blob_reader_free(created.reader); + } +} diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs index 98ada0359..dd8eb593c 100644 --- a/bindings/c/src/types.rs +++ b/bindings/c/src/types.rs @@ -62,12 +62,71 @@ pub unsafe extern "C" fn paimon_bytes_free(bytes: paimon_bytes) { } } +#[repr(C)] +#[derive(Clone, Copy)] +pub struct paimon_byte_slice { + pub data: *const u8, + pub len: usize, +} + +#[repr(C)] +pub struct paimon_bytes_array { + pub data: *mut paimon_bytes, + pub len: usize, +} + +impl paimon_bytes_array { + pub fn empty() -> Self { + Self { + data: std::ptr::null_mut(), + len: 0, + } + } + + pub fn new(values: Vec>) -> Self { + if values.is_empty() { + return Self::empty(); + } + let boxed = values + .into_iter() + .map(paimon_bytes::new) + .collect::>() + .into_boxed_slice(); + let len = boxed.len(); + let data = Box::into_raw(boxed) as *mut paimon_bytes; + Self { data, len } + } +} + +/// # Safety +/// `array` was returned by `paimon_blob_reader_read_blobs`. +#[no_mangle] +pub unsafe extern "C" fn paimon_bytes_array_free(array: paimon_bytes_array) { + if array.data.is_null() { + return; + } + let values = Box::from_raw(std::ptr::slice_from_raw_parts_mut(array.data, array.len)); + for value in values.iter().copied() { + paimon_bytes_free(value); + } +} + /// Opaque wrapper around a heap-allocated Rust object. #[repr(C)] pub struct paimon_catalog { pub inner: *mut c_void, } +#[repr(C)] +pub struct paimon_blob_reader { + pub inner: *mut c_void, +} + +#[repr(C)] +pub struct paimon_blob_stream { + pub inner: *mut c_void, +} + #[repr(C)] pub struct paimon_identifier { pub inner: *mut c_void, diff --git a/bindings/go/DEPENDENCIES.rust.tsv b/bindings/go/DEPENDENCIES.rust.tsv index ebc06fa48..ba4f75585 100644 --- a/bindings/go/DEPENDENCIES.rust.tsv +++ b/bindings/go/DEPENDENCIES.rust.tsv @@ -1,5 +1,6 @@ crate 0BSD Apache-2.0 Apache-2.0 WITH LLVM-exception BSD-2-Clause BSD-3-Clause BSL-1.0 CC0-1.0 CDLA-Permissive-2.0 ISC LGPL-2.1-or-later MIT MIT-0 Unicode-3.0 Unlicense Zlib adler2@2.0.1 X X X +aes@0.8.4 X X ahash@0.8.12 X X aho-corasick@1.1.4 X X alloc-no-stdlib@2.0.4 X @@ -33,10 +34,12 @@ aws-lc-rs@1.17.3 X X aws-lc-sys@0.43.0 X X X X X backon@1.6.0 X base64@0.22.1 X X +base64ct@1.8.3 X X bigdecimal@0.4.10 X X bitflags@2.13.1 X X block-buffer@0.10.4 X X block-buffer@0.12.1 X X +block-padding@0.3.3 X X bon@3.9.3 X X bon-macros@3.9.3 X X brotli@8.0.4 X X @@ -45,15 +48,18 @@ bumpalo@3.20.3 X X bytemuck@1.25.2 X X X byteorder@1.5.0 X X bytes@1.12.1 X +cbc@0.1.2 X X cc@1.3.0 X X cfg-if@1.0.4 X X chrono@0.4.45 X X chrono-tz@0.10.4 X X +cipher@0.4.4 X X cmake@0.1.58 X X cmov@0.5.4 X X combine@4.6.7 X comfy-table@7.2.2 X const-oid@0.10.2 X X +const-oid@0.9.6 X X const-random@0.1.18 X X const-random-macro@0.1.16 X X core-foundation@0.10.1 X X @@ -61,6 +67,7 @@ core-foundation@0.9.4 X X core-foundation-sys@0.8.7 X X cpufeatures@0.2.17 X X cpufeatures@0.3.0 X X +crc-fast@1.10.0 X X crc32fast@1.5.0 X X crossbeam-channel@0.5.16 X X crossbeam-deque@0.8.7 X X @@ -75,6 +82,7 @@ ctutils@0.4.2 X X darling@0.23.0 X darling_core@0.23.0 X darling_macro@0.23.0 X +der@0.7.10 X X diff@0.1.13 X X digest@0.10.7 X X digest@0.11.3 X X @@ -141,6 +149,7 @@ ident_case@1.0.1 X X idna@1.1.0 X X idna_adapter@1.2.2 X X indexmap@2.14.0 X X +inout@0.1.4 X X integer-encoding@3.0.4 X ipnet@2.12.0 X X itertools@0.14.0 X X @@ -155,6 +164,7 @@ jni-sys@0.4.1 X X jni-sys-macros@0.4.1 X X jobserver@0.1.35 X X js-sys@0.3.103 X X +lazy_static@1.5.0 X X lexical-core@1.0.6 X X lexical-parse-float@1.0.6 X X lexical-parse-integer@1.0.6 X X @@ -184,6 +194,7 @@ nalgebra-macros@0.2.2 X native-tls@0.2.18 X X num@0.4.3 X X num-bigint@0.4.8 X X +num-bigint-dig@0.8.6 X X num-complex@0.4.6 X X num-integer@0.1.46 X X num-iter@0.1.46 X X @@ -193,8 +204,14 @@ once_cell@1.21.4 X X opendal-core@0.58.0 X opendal-http-transport-reqwest@0.58.0 X opendal-layer-retry@0.58.0 X +opendal-service-azdls@0.58.0 X +opendal-service-azure-common@0.58.0 X +opendal-service-cos@0.58.0 X opendal-service-fs@0.58.0 X +opendal-service-gcs@0.58.0 X +opendal-service-obs@0.58.0 X opendal-service-oss@0.58.0 X +opendal-service-s3@0.58.0 X openssl@0.10.81 X openssl-macros@0.1.1 X X openssl-probe@0.2.1 X X @@ -208,10 +225,16 @@ paimon-mosaic-core@0.2.0 X paimon-vindex-core@0.4.0 X parquet@58.3.0 X paste@1.0.15 X X +pbkdf2@0.12.2 X X +pem@3.0.6 X +pem-rfc7468@0.7.0 X X percent-encoding@2.3.2 X X phf@0.12.1 X phf_shared@0.12.1 X pin-project-lite@0.2.17 X X +pkcs1@0.7.5 X X +pkcs5@0.7.1 X X +pkcs8@0.10.2 X X pkg-config@0.3.33 X X portable-atomic@1.14.0 X X portable-atomic-util@0.2.7 X X @@ -241,11 +264,17 @@ regex-automata@0.4.16 X X regex-lite@0.1.9 X X regex-syntax@0.8.11 X X reqsign-aliyun-oss@3.1.1 X +reqsign-aws-v4@3.0.2 X +reqsign-azure-storage@3.1.0 X reqsign-core@3.1.0 X reqsign-file-read-tokio@3.0.2 X +reqsign-google@3.0.2 X +reqsign-huaweicloud-obs@3.0.2 X +reqsign-tencent-cos@3.0.2 X reqwest@0.12.28 X X reqwest@0.13.4 X X roaring@0.11.4 X X +rsa@0.9.10 X X rust-ini@0.21.3 X rustc_version@0.4.1 X X rustix@1.1.4 X X X @@ -258,8 +287,10 @@ rustls-webpki@0.103.13 X rustversion@1.0.23 X X ryu@1.0.23 X X safe_arch@0.7.4 X X X +salsa20@0.10.2 X X same-file@1.0.6 X X schannel@0.1.29 X +scrypt@0.11.0 X X security-framework@3.7.0 X X security-framework-sys@2.17.0 X X semver@1.0.28 X X @@ -278,6 +309,7 @@ sha1@0.11.0 X X sha2@0.10.9 X X sha2@0.11.0 X X shlex@2.0.1 X X +signature@2.2.0 X X simba@0.9.1 X simd-adler32@0.3.10 X simd_cesu8@1.2.0 X X @@ -291,6 +323,9 @@ snafu-derive@0.8.9 X X snafu-derive@0.9.1 X X snap@1.1.2 X socket2@0.6.5 X X +spin@0.10.1 X +spin@0.9.9 X +spki@0.7.3 X X stable_deref_trait@1.2.1 X X strsim@0.11.1 X strum@0.27.2 X diff --git a/bindings/go/blob_reader.go b/bindings/go/blob_reader.go new file mode 100644 index 000000000..7f0071e51 --- /dev/null +++ b/bindings/go/blob_reader.go @@ -0,0 +1,227 @@ +/* + * 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. + */ + +package paimon + +import ( + "context" + "fmt" + "runtime" + "sync" + "unsafe" + + "github.com/jupiterrider/ffi" +) + +// BlobReader resolves serialized BlobDescriptors. +type BlobReader struct { + ctx context.Context + lib *libRef + inner *paimonBlobReader + mu sync.RWMutex +} + +// NewBlobReader creates a descriptor reader with FileIO options. +func NewBlobReader(storageOptions map[string]string) (*BlobReader, error) { + ctx, lib, err := ensureLoaded() + if err != nil { + return nil, err + } + inner, err := ffiBlobReaderNew.symbol(ctx)(storageOptions) + if err != nil { + return nil, err + } + lib.acquire() + return &BlobReader{ctx: ctx, lib: lib, inner: inner}, nil +} + +// NewBlobReader creates a descriptor reader using this table's FileIO. +func (t *Table) NewBlobReader() (*BlobReader, error) { + if t.inner == nil { + return nil, ErrClosed + } + inner, err := ffiTableNewBlobReader.symbol(t.ctx)(t.inner) + if err != nil { + return nil, err + } + t.lib.acquire() + return &BlobReader{ctx: t.ctx, lib: t.lib, inner: inner}, nil +} + +// ReadBlob resolves one descriptor. +func (r *BlobReader) ReadBlob(descriptor []byte) ([]byte, error) { + values, err := r.ReadBlobs([][]byte{descriptor}) + if err != nil { + return nil, err + } + return values[0], nil +} + +// ReadBlobs resolves a batch in input order. +func (r *BlobReader) ReadBlobs(descriptors [][]byte) ([][]byte, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if r.inner == nil { + return nil, ErrClosed + } + return ffiBlobReaderReadBlobs.symbol(r.ctx)(r.inner, descriptors) +} + +// Close releases the reader and is idempotent. +func (r *BlobReader) Close() { + r.mu.Lock() + defer r.mu.Unlock() + if r.inner == nil { + return + } + ffiBlobReaderFree.symbol(r.ctx)(r.inner) + r.inner = nil + r.lib.release() +} + +var ffiBlobReaderNew = newFFI(ffiOpts{ + sym: "paimon_blob_reader_new", + rType: &typeResultBlobReader, + aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer}, +}, func(ctx context.Context, ffiCall ffiCall) func(map[string]string) (*paimonBlobReader, error) { + return func(options map[string]string) (*paimonBlobReader, error) { + type paimonOption struct { + key *byte + value *byte + } + opts := make([]paimonOption, 0, len(options)) + for key, value := range options { + keyPtr, err := bytePtrFromString(key) + if err != nil { + return nil, err + } + valuePtr, err := bytePtrFromString(value) + if err != nil { + return nil, err + } + opts = append(opts, paimonOption{key: keyPtr, value: valuePtr}) + } + + var optsPtr unsafe.Pointer + if len(opts) > 0 { + optsPtr = unsafe.Pointer(&opts[0]) + } + optsLen := uintptr(len(opts)) + var result resultBlobReader + ffiCall( + unsafe.Pointer(&result), + unsafe.Pointer(&optsPtr), + unsafe.Pointer(&optsLen), + ) + runtime.KeepAlive(opts) + if result.error != nil { + return nil, parseError(ctx, result.error) + } + return result.reader, nil + } +}) + +var ffiTableNewBlobReader = newFFI(ffiOpts{ + sym: "paimon_table_new_blob_reader", + rType: &typeResultBlobReader, + aTypes: []*ffi.Type{&ffi.TypePointer}, +}, func(ctx context.Context, ffiCall ffiCall) func(*paimonTable) (*paimonBlobReader, error) { + return func(table *paimonTable) (*paimonBlobReader, error) { + var result resultBlobReader + ffiCall( + unsafe.Pointer(&result), + unsafe.Pointer(&table), + ) + if result.error != nil { + return nil, parseError(ctx, result.error) + } + return result.reader, nil + } +}) + +var ffiBlobReaderReadBlobs = newFFI(ffiOpts{ + sym: "paimon_blob_reader_read_blobs", + rType: &typeResultReadBlobs, + aTypes: []*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + &ffi.TypePointer, + }, +}, func(ctx context.Context, ffiCall ffiCall) func(*paimonBlobReader, [][]byte) ([][]byte, error) { + return func(reader *paimonBlobReader, descriptors [][]byte) ([][]byte, error) { + slices := make([]paimonByteSlice, len(descriptors)) + for index, descriptor := range descriptors { + if len(descriptor) > 0 { + slices[index].data = &descriptor[0] + } + slices[index].len = uintptr(len(descriptor)) + } + var slicesPtr unsafe.Pointer + if len(slices) > 0 { + slicesPtr = unsafe.Pointer(&slices[0]) + } + slicesLen := uintptr(len(slices)) + var result resultReadBlobs + ffiCall( + unsafe.Pointer(&result), + unsafe.Pointer(&reader), + unsafe.Pointer(&slicesPtr), + unsafe.Pointer(&slicesLen), + ) + runtime.KeepAlive(descriptors) + runtime.KeepAlive(slices) + if result.error != nil { + return nil, parseError(ctx, result.error) + } + defer ffiBytesArrayFree.symbol(ctx)(result.blobs) + if result.blobs.len > 0 && result.blobs.data == nil { + return nil, fmt.Errorf("paimon: native BlobReader returned a null result array") + } + + values := make([][]byte, result.blobs.len) + for index, value := range unsafe.Slice(result.blobs.data, result.blobs.len) { + if value.len == 0 { + values[index] = []byte{} + } else { + values[index] = parseBytes(value) + } + } + return values, nil + } +}) + +var ffiBlobReaderFree = newFFI(ffiOpts{ + sym: "paimon_blob_reader_free", + rType: &ffi.TypeVoid, + aTypes: []*ffi.Type{&ffi.TypePointer}, +}, func(_ context.Context, ffiCall ffiCall) func(*paimonBlobReader) { + return func(reader *paimonBlobReader) { + ffiCall(nil, unsafe.Pointer(&reader)) + } +}) + +var ffiBytesArrayFree = newFFI(ffiOpts{ + sym: "paimon_bytes_array_free", + rType: &ffi.TypeVoid, + aTypes: []*ffi.Type{&typePaimonBytesArray}, +}, func(_ context.Context, ffiCall ffiCall) func(paimonBytesArray) { + return func(values paimonBytesArray) { + ffiCall(nil, unsafe.Pointer(&values)) + } +}) diff --git a/bindings/go/blob_stream.go b/bindings/go/blob_stream.go new file mode 100644 index 000000000..e6a6b7b96 --- /dev/null +++ b/bindings/go/blob_stream.go @@ -0,0 +1,204 @@ +/* + * 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. + */ + +package paimon + +import ( + "context" + "fmt" + "io" + "runtime" + "sync" + "unsafe" + + "github.com/jupiterrider/ffi" +) + +// BlobStream incrementally reads one BlobDescriptor. +type BlobStream struct { + ctx context.Context + lib *libRef + inner *paimonBlobStream + mu sync.Mutex +} + +var _ io.ReadSeekCloser = (*BlobStream)(nil) + +// OpenBlob opens one descriptor without reading its contents. +func (r *BlobReader) OpenBlob(descriptor []byte) (*BlobStream, error) { + r.mu.RLock() + defer r.mu.RUnlock() + if r.inner == nil { + return nil, ErrClosed + } + inner, err := ffiBlobReaderOpenBlob.symbol(r.ctx)(r.inner, descriptor) + if err != nil { + return nil, err + } + r.lib.acquire() + return &BlobStream{ctx: r.ctx, lib: r.lib, inner: inner}, nil +} + +// Read implements io.Reader. +func (s *BlobStream) Read(buffer []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.inner == nil { + return 0, ErrClosed + } + if len(buffer) == 0 { + return 0, nil + } + + read, err := ffiBlobStreamRead.symbol(s.ctx)(s.inner, buffer) + if err != nil { + return 0, err + } + if read > len(buffer) { + return 0, fmt.Errorf("paimon: native BlobStream returned %d bytes for a %d-byte buffer", read, len(buffer)) + } + if read == 0 { + return 0, io.EOF + } + return read, nil +} + +// Seek implements io.Seeker within the descriptor's range. +func (s *BlobStream) Seek(offset int64, whence int) (int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + if s.inner == nil { + return 0, ErrClosed + } + if whence != io.SeekStart && whence != io.SeekCurrent && whence != io.SeekEnd { + return 0, fmt.Errorf("paimon: invalid BlobStream whence %d", whence) + } + position, err := ffiBlobStreamSeek.symbol(s.ctx)(s.inner, offset, int32(whence)) + if err != nil { + return 0, err + } + if position > uint64(^uint64(0)>>1) { + return 0, fmt.Errorf("paimon: BlobStream position exceeds int64") + } + return int64(position), nil +} + +// Close releases the stream and is idempotent. +func (s *BlobStream) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + if s.inner == nil { + return nil + } + ffiBlobStreamFree.symbol(s.ctx)(s.inner) + s.inner = nil + s.lib.release() + return nil +} + +var ffiBlobReaderOpenBlob = newFFI(ffiOpts{ + sym: "paimon_blob_reader_open_blob", + rType: &typeResultBlobStream, + aTypes: []*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + &ffi.TypePointer, + }, +}, func(ctx context.Context, ffiCall ffiCall) func(*paimonBlobReader, []byte) (*paimonBlobStream, error) { + return func(reader *paimonBlobReader, descriptor []byte) (*paimonBlobStream, error) { + var descriptorPtr unsafe.Pointer + if len(descriptor) > 0 { + descriptorPtr = unsafe.Pointer(&descriptor[0]) + } + descriptorLen := uintptr(len(descriptor)) + var result resultBlobStream + ffiCall( + unsafe.Pointer(&result), + unsafe.Pointer(&reader), + unsafe.Pointer(&descriptorPtr), + unsafe.Pointer(&descriptorLen), + ) + runtime.KeepAlive(descriptor) + if result.error != nil { + return nil, parseError(ctx, result.error) + } + return result.stream, nil + } +}) + +var ffiBlobStreamRead = newFFI(ffiOpts{ + sym: "paimon_blob_stream_read", + rType: &typeResultBlobStreamRead, + aTypes: []*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + &ffi.TypePointer, + }, +}, func(ctx context.Context, ffiCall ffiCall) func(*paimonBlobStream, []byte) (int, error) { + return func(stream *paimonBlobStream, buffer []byte) (int, error) { + bufferPtr := unsafe.Pointer(&buffer[0]) + bufferLen := uintptr(len(buffer)) + var result resultBlobStreamRead + ffiCall( + unsafe.Pointer(&result), + unsafe.Pointer(&stream), + unsafe.Pointer(&bufferPtr), + unsafe.Pointer(&bufferLen), + ) + runtime.KeepAlive(buffer) + if result.error != nil { + return 0, parseError(ctx, result.error) + } + return int(result.bytesRead), nil + } +}) + +var ffiBlobStreamSeek = newFFI(ffiOpts{ + sym: "paimon_blob_stream_seek", + rType: &typeResultBlobStreamSeek, + aTypes: []*ffi.Type{ + &ffi.TypePointer, + &ffi.TypeSint64, + &ffi.TypeSint32, + }, +}, func(ctx context.Context, ffiCall ffiCall) func(*paimonBlobStream, int64, int32) (uint64, error) { + return func(stream *paimonBlobStream, offset int64, whence int32) (uint64, error) { + var result resultBlobStreamSeek + ffiCall( + unsafe.Pointer(&result), + unsafe.Pointer(&stream), + unsafe.Pointer(&offset), + unsafe.Pointer(&whence), + ) + if result.error != nil { + return 0, parseError(ctx, result.error) + } + return result.position, nil + } +}) + +var ffiBlobStreamFree = newFFI(ffiOpts{ + sym: "paimon_blob_stream_free", + rType: &ffi.TypeVoid, + aTypes: []*ffi.Type{&ffi.TypePointer}, +}, func(_ context.Context, ffiCall ffiCall) func(*paimonBlobStream) { + return func(stream *paimonBlobStream) { + ffiCall(nil, unsafe.Pointer(&stream)) + } +}) diff --git a/bindings/go/tests/blob_reader_test.go b/bindings/go/tests/blob_reader_test.go new file mode 100644 index 000000000..94a5eb714 --- /dev/null +++ b/bindings/go/tests/blob_reader_test.go @@ -0,0 +1,296 @@ +/* + * 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. + */ + +package paimon_test + +import ( + "bytes" + "encoding/binary" + "errors" + "io" + "net/url" + "os" + "strings" + "testing" + + paimon "github.com/apache/paimon-rust/bindings/go" +) + +func blobDescriptorV2(uri string, offset, length int64) []byte { + result := make([]byte, 0, 29+len(uri)) + result = append(result, 2) + result = binary.LittleEndian.AppendUint64(result, 0x424C4F4244455343) + result = binary.LittleEndian.AppendUint32(result, uint32(len(uri))) + result = append(result, uri...) + result = binary.LittleEndian.AppendUint64(result, uint64(offset)) + result = binary.LittleEndian.AppendUint64(result, uint64(length)) + return result +} + +func localFileURI(path string) string { + return (&url.URL{Scheme: "file", Path: path}).String() +} + +func TestBlobReaderReadBlobAndBatch(t *testing.T) { + first, err := os.CreateTemp(t.TempDir(), "blob-first-*") + if err != nil { + t.Fatal(err) + } + if _, err := first.WriteString("abcdefghij"); err != nil { + t.Fatal(err) + } + if err := first.Close(); err != nil { + t.Fatal(err) + } + second, err := os.CreateTemp(t.TempDir(), "blob-second-*") + if err != nil { + t.Fatal(err) + } + if _, err := second.WriteString("UVWXYZ"); err != nil { + t.Fatal(err) + } + if err := second.Close(); err != nil { + t.Fatal(err) + } + + reader, err := paimon.NewBlobReader(nil) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + + value, err := reader.ReadBlob(blobDescriptorV2(localFileURI(first.Name()), 1, 3)) + if err != nil { + t.Fatal(err) + } + if string(value) != "bcd" { + t.Fatalf("ReadBlob returned %q, want %q", value, "bcd") + } + + values, err := reader.ReadBlobs([][]byte{ + blobDescriptorV2(localFileURI(second.Name()), 1, 3), + blobDescriptorV2(localFileURI(first.Name()), 3, -1), + blobDescriptorV2(localFileURI(first.Name()), 5, 0), + blobDescriptorV2(localFileURI(first.Name()), 2, 4), + blobDescriptorV2(localFileURI(first.Name()), 2, 4), + }) + if err != nil { + t.Fatal(err) + } + want := []string{"VWX", "defghij", "", "cdef", "cdef"} + for index, value := range values { + if string(value) != want[index] { + t.Fatalf("ReadBlobs result %d = %q, want %q", index, value, want[index]) + } + } + + empty, err := reader.ReadBlobs(nil) + if err != nil { + t.Fatal(err) + } + if empty == nil || len(empty) != 0 { + t.Fatalf("empty batch returned %#v", empty) + } +} + +func TestBlobReaderFromTableOutlivesTable(t *testing.T) { + file, err := os.CreateTemp(t.TempDir(), "blob-table-*") + if err != nil { + t.Fatal(err) + } + if _, err := file.WriteString("abcdefghij"); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + + table := openCopiedTestTable(t) + reader, err := table.NewBlobReader() + if err != nil { + t.Fatal(err) + } + table.Close() + defer reader.Close() + + value, err := reader.ReadBlob(blobDescriptorV2(localFileURI(file.Name()), 2, 4)) + if err != nil { + t.Fatal(err) + } + if string(value) != "cdef" { + t.Fatalf("ReadBlob returned %q, want %q", value, "cdef") + } +} + +func TestBlobReaderErrorsAndClose(t *testing.T) { + reader, err := paimon.NewBlobReader(map[string]string{}) + if err != nil { + t.Fatal(err) + } + + if _, err := reader.ReadBlob(nil); err == nil { + t.Fatal("expected invalid descriptor error") + } + + missingURI := localFileURI(t.TempDir() + "/missing.blob") + _, err = reader.ReadBlobs([][]byte{ + blobDescriptorV2(missingURI, 0, 1), + }) + if err == nil { + t.Fatal("expected missing object error") + } + if !strings.Contains(err.Error(), "input indices [0]") || !strings.Contains(err.Error(), missingURI) { + t.Fatalf("error lacks descriptor context: %v", err) + } + + reader.Close() + reader.Close() + if _, err := reader.ReadBlob(blobDescriptorV2(missingURI, 0, 0)); !errors.Is(err, paimon.ErrClosed) { + t.Fatalf("ReadBlob after Close returned %v, want ErrClosed", err) + } + if _, err := reader.ReadBlobs(nil); !errors.Is(err, paimon.ErrClosed) { + t.Fatalf("ReadBlobs after Close returned %v, want ErrClosed", err) + } +} + +func TestBlobStreamReadsIncrementally(t *testing.T) { + file, err := os.CreateTemp(t.TempDir(), "blob-stream-*") + if err != nil { + t.Fatal(err) + } + if _, err := file.WriteString("abcdefghij"); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + + reader, err := paimon.NewBlobReader(nil) + if err != nil { + t.Fatal(err) + } + stream, err := reader.OpenBlob(blobDescriptorV2(localFileURI(file.Name()), 2, 5)) + if err != nil { + t.Fatal(err) + } + reader.Close() + if size, err := stream.Seek(0, io.SeekEnd); err != nil || size != 5 { + t.Fatalf("SeekEnd returned (%d, %v), want (5, nil)", size, err) + } + if position, err := stream.Seek(1, io.SeekStart); err != nil || position != 1 { + t.Fatalf("SeekStart returned (%d, %v), want (1, nil)", position, err) + } + var ranged bytes.Buffer + if _, err := io.CopyN(&ranged, stream, 3); err != nil { + t.Fatal(err) + } + if ranged.String() != "def" { + t.Fatalf("range returned %q, want %q", ranged.String(), "def") + } + if _, err := stream.Seek(0, io.SeekStart); err != nil { + t.Fatal(err) + } + + buffer := make([]byte, 2) + var value []byte + for { + read, err := stream.Read(buffer) + value = append(value, buffer[:read]...) + if errors.Is(err, io.EOF) { + break + } + if err != nil { + t.Fatal(err) + } + } + if string(value) != "cdefg" { + t.Fatalf("stream returned %q, want %q", value, "cdefg") + } + if err := stream.Close(); err != nil { + t.Fatal(err) + } + if err := stream.Close(); err != nil { + t.Fatal(err) + } + if _, err := stream.Read(buffer); !errors.Is(err, paimon.ErrClosed) { + t.Fatalf("Read after Close returned %v, want ErrClosed", err) + } + if _, err := stream.Seek(0, io.SeekStart); !errors.Is(err, paimon.ErrClosed) { + t.Fatalf("Seek after Close returned %v, want ErrClosed", err) + } +} + +func TestBlobStreamToEndEmptyAndLazyErrors(t *testing.T) { + file, err := os.CreateTemp(t.TempDir(), "blob-stream-tail-*") + if err != nil { + t.Fatal(err) + } + if _, err := file.WriteString("abcdefghij"); err != nil { + t.Fatal(err) + } + if err := file.Close(); err != nil { + t.Fatal(err) + } + + reader, err := paimon.NewBlobReader(nil) + if err != nil { + t.Fatal(err) + } + defer reader.Close() + + stream, err := reader.OpenBlob(blobDescriptorV2(localFileURI(file.Name()), 4, -1)) + if err != nil { + t.Fatal(err) + } + value, err := io.ReadAll(stream) + if err != nil { + t.Fatal(err) + } + stream.Close() + if string(value) != "efghij" { + t.Fatalf("stream returned %q, want %q", value, "efghij") + } + + empty, err := reader.OpenBlob(blobDescriptorV2(localFileURI(file.Name()), 3, 0)) + if err != nil { + t.Fatal(err) + } + value, err = io.ReadAll(empty) + if err != nil { + t.Fatal(err) + } + empty.Close() + if len(value) != 0 { + t.Fatalf("empty stream returned %q", value) + } + + missing := localFileURI(t.TempDir() + "/missing.blob") + lazy, err := reader.OpenBlob(blobDescriptorV2(missing, 0, -1)) + if err != nil { + t.Fatalf("OpenBlob performed eager I/O: %v", err) + } + defer lazy.Close() + if _, err := lazy.Read(make([]byte, 1)); err == nil { + t.Fatal("expected missing object error on first Read") + } + + if _, err := reader.OpenBlob(nil); err == nil { + t.Fatal("expected invalid descriptor error") + } +} diff --git a/bindings/go/types.go b/bindings/go/types.go index 04fc57ae0..6adbac941 100644 --- a/bindings/go/types.go +++ b/bindings/go/types.go @@ -29,6 +29,61 @@ import ( // FFI type definitions mirroring C repr structs from paimon-c. var ( + typeResultBlobReader = ffi.Type{ + Type: ffi.Struct, + Elements: &[]*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + nil, + }[0], + } + + typeResultBlobStream = ffi.Type{ + Type: ffi.Struct, + Elements: &[]*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + nil, + }[0], + } + + typeResultBlobStreamRead = ffi.Type{ + Type: ffi.Struct, + Elements: &[]*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + nil, + }[0], + } + + typeResultBlobStreamSeek = ffi.Type{ + Type: ffi.Struct, + Elements: &[]*ffi.Type{ + &ffi.TypeUint64, + &ffi.TypePointer, + nil, + }[0], + } + + typePaimonBytesArray = ffi.Type{ + Type: ffi.Struct, + Elements: &[]*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + nil, + }[0], + } + + typeResultReadBlobs = ffi.Type{ + Type: ffi.Struct, + Elements: &[]*ffi.Type{ + &ffi.TypePointer, + &ffi.TypePointer, + &ffi.TypePointer, + nil, + }[0], + } + // Result types: { value, *error } // paimon_result_catalog_new { catalog: paimon_catalog, error: *paimon_error } typeResultCatalogNew = ffi.Type{ @@ -243,6 +298,16 @@ type paimonBytes struct { len uintptr } +type paimonByteSlice struct { + data *byte + len uintptr +} + +type paimonBytesArray struct { + data *paimonBytes + len uintptr +} + type paimonError struct { code int32 message paimonBytes @@ -250,6 +315,8 @@ type paimonError struct { // Opaque pointer wrappers type paimonCatalog struct{} +type paimonBlobReader struct{} +type paimonBlobStream struct{} type paimonIdentifier struct{} type paimonTable struct{} type paimonReadBuilder struct{} @@ -273,6 +340,31 @@ type resultCatalogNew struct { error *paimonError } +type resultBlobReader struct { + reader *paimonBlobReader + error *paimonError +} + +type resultBlobStream struct { + stream *paimonBlobStream + error *paimonError +} + +type resultBlobStreamRead struct { + bytesRead uintptr + error *paimonError +} + +type resultBlobStreamSeek struct { + position uint64 + error *paimonError +} + +type resultReadBlobs struct { + blobs paimonBytesArray + error *paimonError +} + type resultGetTable struct { table *paimonTable error *paimonError diff --git a/crates/paimon/src/catalog/rest/rest_token_file_io.rs b/crates/paimon/src/catalog/rest/rest_token_file_io.rs index c34967a93..d7d17f787 100644 --- a/crates/paimon/src/catalog/rest/rest_token_file_io.rs +++ b/crates/paimon/src/catalog/rest/rest_token_file_io.rs @@ -185,6 +185,8 @@ mod tests { use super::*; use crate::api::GetTableTokenResponse; use crate::io::cache::create_local_cache; + use crate::spec::BlobDescriptor; + use crate::BlobReader; async fn token(State(requests): State>) -> Json { let request = requests.fetch_add(1, Ordering::SeqCst); @@ -292,4 +294,60 @@ mod tests { assert_eq!(requests.load(Ordering::SeqCst), 2); server.abort(); } + + #[tokio::test] + async fn test_blob_reader_reuses_refreshing_file_io() { + let table_directory = tempfile::tempdir().unwrap(); + let file_path = table_directory.path().join("blob"); + std::fs::write(&file_path, b"abcdefghij").unwrap(); + let (options, api, requests, server) = token_api().await; + let token_file_io = Arc::new(RESTTokenFileIO::new( + Identifier::new("database", "table"), + table_directory.path().to_string_lossy().into_owned(), + options, + api, + None, + )); + + let file_io = token_file_io.build_file_io().await.unwrap(); + assert_eq!(requests.load(Ordering::SeqCst), 1); + let uri = url::Url::from_file_path(file_path).unwrap().to_string(); + let descriptor = BlobDescriptor::new(uri, 2, 4).serialize(); + let values = BlobReader::from_file_io(file_io) + .read_blobs(&[descriptor]) + .await + .unwrap(); + + assert_eq!(values, vec![b"cdef".to_vec()]); + assert_eq!(requests.load(Ordering::SeqCst), 2); + server.abort(); + } + + #[tokio::test] + async fn test_blob_stream_reuses_refreshing_file_io() { + let table_directory = tempfile::tempdir().unwrap(); + let file_path = table_directory.path().join("blob"); + std::fs::write(&file_path, b"abcdefghij").unwrap(); + let (options, api, requests, server) = token_api().await; + let token_file_io = Arc::new(RESTTokenFileIO::new( + Identifier::new("database", "table"), + table_directory.path().to_string_lossy().into_owned(), + options, + api, + None, + )); + + let file_io = token_file_io.build_file_io().await.unwrap(); + assert_eq!(requests.load(Ordering::SeqCst), 1); + let uri = url::Url::from_file_path(file_path).unwrap().to_string(); + let descriptor = BlobDescriptor::new(uri, 2, 4).serialize(); + let mut stream = BlobReader::from_file_io(file_io) + .open_blob(&descriptor) + .unwrap(); + + assert_eq!(stream.read(2).await.unwrap(), b"cd"); + assert_eq!(stream.read(2).await.unwrap(), b"ef"); + assert_eq!(requests.load(Ordering::SeqCst), 2); + server.abort(); + } } diff --git a/crates/paimon/src/lib.rs b/crates/paimon/src/lib.rs index a86a95e92..5a33f65c4 100644 --- a/crates/paimon/src/lib.rs +++ b/crates/paimon/src/lib.rs @@ -48,12 +48,12 @@ pub use catalog::CatalogFactory; pub use catalog::FileSystemCatalog; pub use table::{ - CommitMessage, DataEvolutionDeleteWriter, DataEvolutionWriter, DataSplit, DataSplitBuilder, - DeletionFile, IncrementalPlan, IncrementalScan, IncrementalScanMode, IncrementalSplit, - PartitionBucket, Plan, PostponeBucketPlan, PostponeFixedBucketTableCommit, - PostponeFixedBucketTableWrite, RESTEnv, RESTSnapshotCommit, ReadBuilder, - RenamingSnapshotCommit, RowRange, ScanTrace, SnapshotCommit, SnapshotManager, Table, - TableCommit, TableRead, TableScan, TableUpdate, TableWrite, TagManager, WriteBuilder, + BlobReader, BlobStream, CommitMessage, DataEvolutionDeleteWriter, DataEvolutionWriter, + DataSplit, DataSplitBuilder, DeletionFile, IncrementalPlan, IncrementalScan, + IncrementalScanMode, IncrementalSplit, PartitionBucket, Plan, PostponeBucketPlan, + PostponeFixedBucketTableCommit, PostponeFixedBucketTableWrite, RESTEnv, RESTSnapshotCommit, + ReadBuilder, RenamingSnapshotCommit, RowRange, ScanTrace, SnapshotCommit, SnapshotManager, + Table, TableCommit, TableRead, TableScan, TableUpdate, TableWrite, TagManager, WriteBuilder, }; pub use table::{ diff --git a/crates/paimon/src/table/blob_resolver.rs b/crates/paimon/src/table/blob_resolver.rs index 673efb9fd..e5448254d 100644 --- a/crates/paimon/src/table/blob_resolver.rs +++ b/crates/paimon/src/table/blob_resolver.rs @@ -23,6 +23,7 @@ use arrow_array::{Array, BinaryArray}; use bytes::Bytes; use futures::{stream, StreamExt, TryStreamExt}; use std::collections::HashMap; +use std::io::SeekFrom; use std::sync::Arc; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; @@ -32,12 +33,284 @@ pub(crate) const BLOB_DESCRIPTOR_READ_CONCURRENCY: usize = 8; const BLOB_DESCRIPTOR_READ_BYTE_UNIT: u64 = 1024 * 1024; const BLOB_DESCRIPTOR_READ_MAX_IN_FLIGHT_BYTES: u64 = 64 * 1024 * 1024; +/// Reads serialized [`BlobDescriptor`] values without requiring a table. +#[derive(Clone, Debug)] +pub struct BlobReader { + storage_options: HashMap, + file_io: Option, + limiter: BlobReadLimiter, +} + +impl Default for BlobReader { + fn default() -> Self { + Self::new(HashMap::new()) + } +} + +impl BlobReader { + pub fn new(storage_options: HashMap) -> Self { + Self { + storage_options, + file_io: None, + limiter: BlobReadLimiter::new(), + } + } + + /// Create a reader that reuses an existing FileIO. + pub fn from_file_io(file_io: FileIO) -> Self { + Self { + storage_options: HashMap::new(), + file_io: Some(file_io), + limiter: BlobReadLimiter::new(), + } + } + + /// Open one descriptor for incremental reads. + pub fn open_blob(&self, bytes: &[u8]) -> Result { + let descriptor = BlobDescriptor::deserialize(bytes) + .map_err(|error| blob_error_with_context(error, &[0], None))?; + let range = descriptor + .range_spec() + .map_err(|error| blob_error_with_context(error, &[0], Some(descriptor.uri())))?; + let file_io = match &self.file_io { + Some(file_io) => file_io.clone(), + None => FileIO::from_path(descriptor.uri()) + .and_then(|builder| builder.with_props(self.storage_options.iter()).build()) + .map_err(|error| blob_error_with_context(error, &[0], Some(descriptor.uri())))?, + }; + + Ok(BlobStream { + file_io, + uri: descriptor.uri().to_string(), + offset: range.offset(), + resolved_length: range.length(), + position: 0, + limiter: self.limiter.clone(), + }) + } + + /// Read a descriptor batch in input order. + pub async fn read_blobs(&self, descriptors: &[Vec]) -> Result>> { + let mut by_uri = HashMap::>::new(); + for (index, bytes) in descriptors.iter().enumerate() { + let descriptor = BlobDescriptor::deserialize(bytes) + .map_err(|error| blob_error_with_context(error, &[index], None))?; + descriptor.range_spec().map_err(|error| { + blob_error_with_context(error, &[index], Some(descriptor.uri())) + })?; + by_uri + .entry(descriptor.uri().to_string()) + .or_default() + .push((index, bytes)); + } + + let limiter = self.limiter.clone(); + let groups: Vec)>> = stream::iter(by_uri) + .map(|(uri, entries)| { + let limiter = limiter.clone(); + async move { + let indices = entries.iter().map(|(index, _)| *index).collect::>(); + let file_io = match &self.file_io { + Some(file_io) => file_io.clone(), + None => FileIO::from_path(&uri) + .and_then(|builder| { + builder.with_props(self.storage_options.iter()).build() + }) + .map_err(|error| { + blob_error_with_context(error, &indices, Some(&uri)) + })?, + }; + let mut builder = BinaryBuilder::with_capacity(entries.len(), 0); + for (_, descriptor) in &entries { + builder.append_value(descriptor); + } + let resolved = resolve_blob_column(&builder.finish(), &file_io, limiter) + .await + .map_err(|error| blob_error_with_context(error, &indices, Some(&uri)))?; + Ok::<_, crate::Error>( + entries + .into_iter() + .enumerate() + .map(|(position, (index, _))| { + (index, resolved.value(position).to_vec()) + }) + .collect(), + ) + } + }) + .buffer_unordered(BLOB_DESCRIPTOR_READ_CONCURRENCY) + .try_collect() + .await?; + + let mut values = groups.into_iter().flatten().collect::>(); + values.sort_unstable_by_key(|(index, _)| *index); + Ok(values.into_iter().map(|(_, value)| value).collect()) + } +} + +/// Incremental reader for one serialized [`BlobDescriptor`]. +#[derive(Debug)] +pub struct BlobStream { + file_io: FileIO, + uri: String, + offset: u64, + resolved_length: Option, + position: u64, + limiter: BlobReadLimiter, +} + +impl BlobStream { + /// Read at most `max_bytes`, returning an empty buffer at end of stream. + pub async fn read(&mut self, max_bytes: usize) -> Result> { + self.read_inner(max_bytes) + .await + .map_err(|error| blob_error_with_context(error, &[0], Some(&self.uri))) + } + + /// Seek within the descriptor's range. + pub async fn seek(&mut self, from: SeekFrom) -> Result { + self.seek_inner(from) + .await + .map_err(|error| blob_error_with_context(error, &[0], Some(&self.uri))) + } + + async fn seek_inner(&mut self, from: SeekFrom) -> Result { + let position = match from { + SeekFrom::Start(position) => i128::from(position), + SeekFrom::Current(offset) => i128::from(self.position) + i128::from(offset), + SeekFrom::End(offset) => i128::from(self.length().await?) + i128::from(offset), + }; + self.position = u64::try_from(position).map_err(|_| crate::Error::DataInvalid { + message: "invalid BlobDescriptor stream seek".to_string(), + source: None, + })?; + Ok(self.position) + } + + async fn length(&mut self) -> Result { + if let Some(length) = self.resolved_length { + return Ok(length); + } + let input = self.file_io.new_input(&self.uri)?; + let _permit = self.limiter.acquire_request(&self.uri, "metadata").await?; + let length = input.metadata().await?.size.saturating_sub(self.offset); + self.resolved_length = Some(length); + Ok(length) + } + + async fn read_inner(&mut self, max_bytes: usize) -> Result> { + if max_bytes == 0 { + return Ok(Vec::new()); + } + + let length = self.length().await?; + let remaining = length.saturating_sub(self.position); + if remaining == 0 { + return Ok(Vec::new()); + } + + let to_read = remaining.min(u64::try_from(max_bytes).unwrap_or(u64::MAX)); + let start = + self.offset + .checked_add(self.position) + .ok_or_else(|| crate::Error::DataInvalid { + message: "BlobDescriptor stream position overflows u64".to_string(), + source: None, + })?; + let end = start + .checked_add(to_read) + .ok_or_else(|| crate::Error::DataInvalid { + message: "BlobDescriptor stream range overflows u64".to_string(), + source: None, + })?; + let input = self.file_io.new_input(&self.uri)?; + let reader = input.reader().await?; + let _permits = self.limiter.acquire_read(to_read, &self.uri).await?; + let bytes = reader.read(start..end).await?; + if bytes.len() as u64 != to_read { + return Err(crate::Error::DataInvalid { + message: format!( + "short read for range {start}..{end}, expected={to_read} bytes, actual={} bytes", + bytes.len() + ), + source: None, + }); + } + self.position += to_read; + Ok(bytes.to_vec()) + } +} + +fn blob_error_with_context( + error: crate::Error, + indices: &[usize], + uri: Option<&str>, +) -> crate::Error { + let location = match uri { + Some(uri) => format!( + "input indices {indices:?}, URI '{}'", + sanitize_blob_uri(uri) + ), + None => format!("input indices {indices:?}, URI unavailable"), + }; + let sanitize = |message: String| match uri { + Some(uri) => message.replace(uri, &sanitize_blob_uri(uri)), + None => message, + }; + match error { + crate::Error::Unsupported { message } => crate::Error::Unsupported { + message: format!("BlobDescriptor {location}: {}", sanitize(message)), + }, + crate::Error::IoUnsupported { message } => crate::Error::IoUnsupported { + message: format!("BlobDescriptor {location}: {}", sanitize(message)), + }, + crate::Error::ConfigInvalid { .. } => crate::Error::ConfigInvalid { + message: format!("BlobDescriptor {location}: invalid storage URI or options"), + }, + crate::Error::DataInvalid { message, .. } => crate::Error::DataInvalid { + message: format!("BlobDescriptor {location}: {}", sanitize(message)), + source: None, + }, + crate::Error::IoUnexpected { source, .. } + if source.kind() == opendal::ErrorKind::NotFound => + { + crate::Error::UnexpectedError { + message: format!("BlobDescriptor {location}: object not found"), + source: None, + } + } + crate::Error::IoUnexpected { .. } => crate::Error::UnexpectedError { + message: format!("BlobDescriptor {location}: storage I/O failed"), + source: None, + }, + crate::Error::UnexpectedError { message, .. } => crate::Error::UnexpectedError { + message: format!("BlobDescriptor {location}: {}", sanitize(message)), + source: None, + }, + _ => crate::Error::UnexpectedError { + message: format!("BlobDescriptor {location}: operation failed"), + source: None, + }, + } +} + +fn sanitize_blob_uri(uri: &str) -> String { + if let Ok(mut url) = url::Url::parse(uri) { + let _ = url.set_username(""); + let _ = url.set_password(None); + url.set_query(None); + url.set_fragment(None); + return url.to_string(); + } + uri.split(['?', '#']).next().unwrap_or(uri).to_string() +} + /// Shared admission control for external descriptor metadata and range reads. /// /// The byte semaphore budgets active range I/O only. A single range larger than /// the budget consumes every byte permit and runs alone, but can still allocate /// more than the configured budget because the complete value is required. -#[derive(Clone)] +#[derive(Clone, Debug)] pub(crate) struct BlobReadLimiter { requests: Arc, bytes: Arc, @@ -399,6 +672,7 @@ mod tests { bytes: Bytes, in_flight: std::sync::Arc, max_in_flight: std::sync::Arc, + ranges: std::sync::Arc>>>, } impl TrackingFileRead { @@ -407,6 +681,7 @@ mod tests { bytes, in_flight: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)), max_in_flight: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)), + ranges: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())), } } @@ -419,17 +694,23 @@ mod tests { bytes, in_flight, max_in_flight, + ranges: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())), } } fn max_in_flight(&self) -> usize { self.max_in_flight.load(std::sync::atomic::Ordering::SeqCst) } + + fn ranges(&self) -> Vec> { + self.ranges.lock().unwrap().clone() + } } #[async_trait::async_trait] impl FileRead for TrackingFileRead { async fn read(&self, range: std::ops::Range) -> crate::Result { + self.ranges.lock().unwrap().push(range.clone()); let in_flight = self .in_flight .fetch_add(1, std::sync::atomic::Ordering::SeqCst) @@ -443,6 +724,15 @@ mod tests { } } + struct ShortFileRead; + + #[async_trait::async_trait] + impl FileRead for ShortFileRead { + async fn read(&self, _range: std::ops::Range) -> crate::Result { + Ok(Bytes::from_static(b"x")) + } + } + #[tokio::test] async fn test_blob_range_reads_use_bounded_parallelism() { let reader = TrackingFileRead::new(Bytes::from_static(b"abcdefghijkl")); @@ -575,6 +865,63 @@ mod tests { .unwrap(); } + #[tokio::test] + async fn test_merged_descriptors_issue_one_underlying_read() { + let reader = TrackingFileRead::new(Bytes::from_static(b"abcdefghijkl")); + let reads = merge_blob_read_requests(vec![ + BlobReadRequest { + row: 0, + offset: 0, + length: 4, + }, + BlobReadRequest { + row: 1, + offset: 4, + length: 4, + }, + BlobReadRequest { + row: 2, + offset: 2, + length: 6, + }, + ]); + + let results = read_merged_blob_ranges( + "memory:/blob.bin", + Arc::new(reader.clone()), + reads, + BlobReadLimiter::new(), + ) + .await + .unwrap(); + + assert_eq!(results.len(), 1); + assert_eq!(reader.ranges(), vec![0..8]); + } + + #[tokio::test] + async fn test_blob_range_read_rejects_short_data() { + let error = read_merged_blob_ranges( + "memory:/blob.bin", + Arc::new(ShortFileRead), + vec![MergedBlobRead { + start: 4, + end: 8, + requests: vec![BlobReadRequest { + row: 0, + offset: 4, + length: 4, + }], + }], + BlobReadLimiter::new(), + ) + .await + .err() + .expect("short read must fail"); + + assert!(error.to_string().contains("short read")); + } + #[test] fn test_merge_blob_read_requests_merges_nearby_ranges() { let merged = merge_blob_read_requests(vec![ @@ -614,4 +961,137 @@ mod tests { assert_eq!(merged[1].start, BLOB_RANGE_MERGE_MAX_SPAN + 1); assert_eq!(merged[1].end, BLOB_RANGE_MERGE_MAX_SPAN + 5); } + + fn java_v2_descriptor(uri: &str, offset: i64, length: i64) -> Vec { + let mut bytes = Vec::new(); + bytes.push(2); + bytes.extend_from_slice(&0x424C4F4244455343_u64.to_le_bytes()); + bytes.extend_from_slice(&(uri.len() as i32).to_le_bytes()); + bytes.extend_from_slice(uri.as_bytes()); + bytes.extend_from_slice(&offset.to_le_bytes()); + bytes.extend_from_slice(&length.to_le_bytes()); + bytes + } + + fn file_uri(path: &std::path::Path) -> String { + url::Url::from_file_path(path).unwrap().to_string() + } + + #[tokio::test] + async fn test_standalone_blob_reader_reads_ranges_in_input_order() { + let first = tempfile::NamedTempFile::new().unwrap(); + let second = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(first.path(), b"abcdefghij").unwrap(); + std::fs::write(second.path(), b"UVWXYZ").unwrap(); + let first_uri = file_uri(first.path()); + let second_uri = file_uri(second.path()); + let descriptors = vec![ + java_v2_descriptor(&second_uri, 1, 3), + java_v2_descriptor(&first_uri, 3, -1), + java_v2_descriptor(&first_uri, 5, 0), + java_v2_descriptor(&first_uri, 2, 4), + java_v2_descriptor(&first_uri, 2, 4), + ]; + + let values = BlobReader::default() + .read_blobs(&descriptors) + .await + .unwrap(); + + assert_eq!( + values, + vec![ + b"VWX".to_vec(), + b"defghij".to_vec(), + Vec::new(), + b"cdef".to_vec(), + b"cdef".to_vec(), + ] + ); + } + + #[tokio::test] + async fn test_standalone_blob_reader_validates_input() { + let reader = BlobReader::default(); + + let error = reader.read_blobs(&[Vec::new()]).await.unwrap_err(); + assert!(error.to_string().contains("input indices [0]")); + + let error = reader + .read_blobs(&[BlobDescriptor::new("file:///tmp/a".to_string(), -1, 1).serialize()]) + .await + .unwrap_err(); + assert!(error.to_string().contains("offset must be non-negative")); + + let secret_uri = "ftp://access-key:secret@example.com/a?token=sensitive"; + let error = reader + .read_blobs(&[BlobDescriptor::new(secret_uri.to_string(), 0, 1).serialize()]) + .await + .unwrap_err(); + let message = error.to_string(); + assert!(message.contains("ftp://example.com/a")); + assert!(!message.contains("access-key")); + assert!(!message.contains("sensitive")); + } + + #[tokio::test] + async fn test_standalone_blob_reader_empty_batch() { + assert!(BlobReader::default() + .read_blobs(&[]) + .await + .unwrap() + .is_empty()); + } + + #[tokio::test] + async fn test_blob_stream_reads_incrementally() { + let file = tempfile::NamedTempFile::new().unwrap(); + std::fs::write(file.path(), b"abcdefghij").unwrap(); + let uri = file_uri(file.path()); + let reader = BlobReader::default(); + + let mut fixed = reader.open_blob(&java_v2_descriptor(&uri, 2, 5)).unwrap(); + assert_eq!(fixed.read(2).await.unwrap(), b"cd"); + assert_eq!(fixed.seek(SeekFrom::Start(1)).await.unwrap(), 1); + assert_eq!(fixed.read(2).await.unwrap(), b"de"); + assert_eq!(fixed.seek(SeekFrom::Current(-1)).await.unwrap(), 2); + assert_eq!(fixed.read(2).await.unwrap(), b"ef"); + assert_eq!(fixed.seek(SeekFrom::End(-2)).await.unwrap(), 3); + assert_eq!(fixed.read(8).await.unwrap(), b"fg"); + assert!(fixed.read(1).await.unwrap().is_empty()); + assert!(fixed.seek(SeekFrom::Current(-6)).await.is_err()); + + let mut to_end = reader.open_blob(&java_v2_descriptor(&uri, 4, -1)).unwrap(); + assert_eq!(to_end.seek(SeekFrom::End(-3)).await.unwrap(), 3); + assert_eq!(to_end.read(3).await.unwrap(), b"hij"); + assert_eq!(to_end.seek(SeekFrom::Start(0)).await.unwrap(), 0); + assert_eq!(to_end.read(8).await.unwrap(), b"efghij"); + assert!(to_end.read(1).await.unwrap().is_empty()); + + let mut empty = reader.open_blob(&java_v2_descriptor(&uri, 3, 0)).unwrap(); + assert!(empty.read(1).await.unwrap().is_empty()); + + let mut short = reader.open_blob(&java_v2_descriptor(&uri, 8, 4)).unwrap(); + assert!(short.read(4).await.is_err()); + } + + #[tokio::test] + async fn test_blob_stream_is_lazy_and_validates_input() { + let reader = BlobReader::default(); + let directory = tempfile::tempdir().unwrap(); + let missing = file_uri(&directory.path().join("missing")); + let mut stream = reader + .open_blob(&java_v2_descriptor(&missing, 0, -1)) + .unwrap(); + + assert!(stream.read(0).await.unwrap().is_empty()); + let error = stream.read(1).await.unwrap_err().to_string(); + assert!(error.contains("input indices [0]")); + assert!(error.contains("object not found") || error.contains("storage I/O failed")); + + assert!(reader.open_blob(&[]).is_err()); + assert!(reader + .open_blob(&BlobDescriptor::new("file:///tmp/a".to_string(), -1, 1).serialize()) + .is_err()); + } } diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs index 35f1e45a0..3b0f447d4 100644 --- a/crates/paimon/src/table/mod.rs +++ b/crates/paimon/src/table/mod.rs @@ -109,6 +109,7 @@ mod write_builder; use crate::Result; use arrow_array::RecordBatch; pub use audit_log_table::AuditLogTable; +pub use blob_resolver::{BlobReader, BlobStream}; pub use branch_manager::BranchManager; pub use commit_message::CommitMessage; pub use cow_writer::{CopyOnWriteMergeWriter, FileInfo}; diff --git a/docs/src/go-binding.md b/docs/src/go-binding.md index 26999d9bd..10bfdf0bd 100644 --- a/docs/src/go-binding.md +++ b/docs/src/go-binding.md @@ -39,6 +39,137 @@ go get github.com/apache/paimon-rust/bindings/go The native library is embedded and loaded automatically. Build with `CGO_ENABLED=1`. +## Reading BlobDescriptor Values + +`BlobReader` reads serialized `BlobDescriptor` values directly; it does not scan a Paimon table. +Use `ReadBlobs` to resolve a batch in one call. `ReadBlob` is a convenience +method for one descriptor; local `file://` reads can use nil storage options. + +```go +package main + +import ( + "database/sql" + "log" + "os" + + paimon "github.com/apache/paimon-rust/bindings/go" + _ "github.com/go-sql-driver/mysql" +) + +func main() { + db, err := sql.Open("mysql", os.Getenv("STARROCKS_DSN")) + if err != nil { + log.Fatal(err) + } + defer db.Close() + + rows, err := db.Query("SELECT blob_descriptor FROM catalog.db.my_table") + if err != nil { + log.Fatal(err) + } + defer rows.Close() + + var descriptors [][]byte + for rows.Next() { + var descriptor []byte + if err := rows.Scan(&descriptor); err != nil { + log.Fatal(err) + } + descriptors = append(descriptors, append([]byte(nil), descriptor...)) + } + if err := rows.Err(); err != nil { + log.Fatal(err) + } + + reader, err := paimon.NewBlobReader(map[string]string{ + "fs.oss.accessKeyId": os.Getenv("OSS_ACCESS_KEY_ID"), + "fs.oss.accessKeySecret": os.Getenv("OSS_ACCESS_KEY_SECRET"), + "fs.oss.endpoint": os.Getenv("OSS_ENDPOINT"), + }) + if err != nil { + log.Fatal(err) + } + defer reader.Close() + + blobs, err := reader.ReadBlobs(descriptors) + if err != nil { + log.Fatal(err) + } + for _, blob := range blobs { + log.Printf("read %d bytes", len(blob)) + } +} +``` + +The descriptor contains only URI, offset, and length. Pass OSS/S3 credentials +with the same FileIO option names used by catalogs. If StarRocks returns a hex +or base64 string, decode it to the original descriptor bytes before calling +`ReadBlobs`. + +Stream a large value without holding it all in memory: + +```go +stream, err := reader.OpenBlob(descriptor) +if err != nil { + log.Fatal(err) +} +defer stream.Close() +if _, err := io.Copy(destination, stream); err != nil { + log.Fatal(err) +} +``` + +For an HTTP byte range, seek relative to the descriptor and copy only that range: + +```go +size, err := stream.Seek(0, io.SeekEnd) +_, err = stream.Seek(start, io.SeekStart) +_, err = io.CopyN(w, stream, end-start+1) +``` + +`OpenBlob` is lazy and returns an `io.ReadSeekCloser`. `ReadBlobs` remains the batch +API that groups and merges ranges; separate streams are not merged together. + +For DLF temporary data tokens, reuse a table's refreshing FileIO: + +```go +catalog, err := paimon.NewCatalog(map[string]string{ + "metastore": "rest", + "uri": os.Getenv("DLF_ENDPOINT"), + "warehouse": os.Getenv("DLF_CATALOG"), + "token.provider": "dlf", + "dlf.region": os.Getenv("DLF_REGION"), + "dlf.oss-endpoint": os.Getenv("DLF_OSS_ENDPOINT"), + "dlf.token-loader": "ecs", + "dlf.token-ecs-role-name": os.Getenv("DLF_ECS_ROLE"), + "data-token.enabled": "true", +}) +if err != nil { + log.Fatal(err) +} +defer catalog.Close() +table, err := catalog.GetTable(paimon.NewIdentifier("db", "descriptor_table")) +if err != nil { + log.Fatal(err) +} +defer table.Close() +reader, err := table.NewBlobReader() +if err != nil { + log.Fatal(err) +} +defer reader.Close() +``` + +The reader and its streams keep the table FileIO and refresh DLF data tokens +before expiry. Set `dlf.oss-endpoint` when the server-provided endpoint is not +reachable from the application. Static options passed to +`paimon.NewBlobReader` are not refreshed. + +Reads are grouped by URI and nearby ranges are merged. The fixed limits are a +64 KiB merge gap, 8 MiB merged span, 8 concurrent requests, and 64 MiB of +in-flight range reads. Results retain descriptor input order. + ## Creating a Catalog Use `NewCatalog` with a map of options to create a catalog. The catalog type is determined by the `metastore` option (default: `filesystem`).