You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "linliu-code (via GitHub)" <gi...@apache.org> on 2026/08/28 04:25:36 UTC
[PR] feat(core): read the metadata table as Arrow, and expose hudi-rs over a C ABI for JVM callers [hudi-rs]
linliu-code opened a new pull request, #700:
URL: https://github.com/apache/hudi-rs/pull/700
## Description
A JVM cannot call hudi-rs today. The `cpp` bridge is generated by `cxx`, so its symbols are C++-mangled and its arguments are `CxxString`/`CxxVector`; neither binds from Java. This adds `crates/jvm-ffi`, a plain C ABI over the reads that already exist, and one new read the metadata table needed.
That gap matters most for the metadata table, because Gluten offloads `SparkPlan` nodes and a metadata read is not one: `HoodieFileIndex` consults the metadata table while the plan is being built, and its output is what produces the scan Gluten later offloads. There is no node to hand over, so a JVM caller wanting hudi-rs's metadata reader currently has no route at all.
**The core change is a split, not new behaviour.** Reading the `files` partition built a `RecordBatch` and immediately decoded it into `FilesPartitionRecord`s. An Arrow consumer would have taken those structs and re-encoded them, paying for a decode it did not want. `read_files_partition` is now `read_files_partition_batch` plus the decode that was already in it, and `Table::read_metadata_table_files_partition_arrow` exposes the batch. Slice resolution is shared by both, so neither path can drift from the other.
**The shim holds no reading logic.** Null-terminated strings in, opaque handles and an `ArrowArrayStream` out. Every export routes through a guard that turns an error or a panic into a null plus a message retrievable with `hudi_ffi_last_error`, because a panic unwinding across `extern "C"` aborts the process that called us.
Two things a reviewer should know rather than infer. `hudi_ffi_read_slice` reads a slice through the generic file group reader, which refuses an HFile base file by design, so it serves parquet slices and not metadata ones; the metadata read is the separate entry point above. And a JVM consumer must run with `--add-opens=java.base/java.nio=ALL-UNNAMED`, which arrow-java requires and fails loudly without.
Stacked on #696. Review that first; this diff is the last commit.
## How are the changes test-covered
- [ ] N/A
- [x] Automated tests (unit and/or integration tests)
- [x] Manual tests
- [x] Details are described below
Six tests cover the boundary, and three are mutation-checked, since a guard that is never exercised is not a guard:
| Mutation | Result |
| --- | --- |
| remove `catch_unwind` from the guard | `a_panic_becomes_a_null_and_a_message` fails |
| remove the null check in `hudi_ffi_free_stream` | `freeing_null_is_a_no_op` aborts with SIGABRT |
| weaken the UTF-8 error message | `invalid_utf8_is_refused` fails |
The others cover an error becoming a null plus a message, null arguments being refused rather than dereferenced, and the error slot being per-thread so one thread's failure is not another's.
Manually verified end to end from Java 25, using FFM to call the shim and arrow-java 15 to import the stream, against a real metadata table:
```
reader opened
columns: 12
key=city=chennai type=2
key=city=san_francisco type=2
key=__all_partitions__ type=1
key=city=sao_paulo type=2
rows: 4
```
Those are the same four records `hudi_table_read_metadata_table_files_partition` asserts in Rust, with the `type` discriminator intact.
Locally: 1386 `hudi-core` lib tests pass, 6 in `hudi-jvm-ffi`, clippy clean on both with `-D warnings`, fmt clean.
**Not covered, and stated rather than implied.** The JVM half of the ownership contract is untested here: that the stream is freed exactly once including on an exception path, and that a panic does not abort a real JVM, are asserted on the Rust side only. Proving them from Java needs a test module that does not exist yet in this repo. The panicking export used for that is behind a default-off `ffi-test-hooks` feature so it is absent from a shipped library.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(core): read the metadata table as Arrow, and expose hudi-rs over a C ABI for JVM callers [hudi-rs]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #700:
URL: https://github.com/apache/hudi-rs/pull/700#issuecomment-5517222515
## [Codecov](https://app.codecov.io/gh/apache/hudi-rs/pull/700?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `43.13725%` with `58 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 79.34%. Comparing base ([`04a3853`](https://app.codecov.io/gh/apache/hudi-rs/commit/04a3853ada7adbf18612b7a839553c3f924b9ec0?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`a5a4a09`](https://app.codecov.io/gh/apache/hudi-rs/commit/a5a4a09d2baa4d2aceb0568d91445cbf00380522?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on main.
| [Files with missing lines](https://app.codecov.io/gh/apache/hudi-rs/pull/700?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [crates/jvm-ffi/src/lib.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/700?src=pr&el=tree&filepath=crates%2Fjvm-ffi%2Fsrc%2Flib.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2p2bS1mZmkvc3JjL2xpYi5ycw==) | 47.56% | [43 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/700?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/metadata/table/mod.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/700?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fmetadata%2Ftable%2Fmod.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL21ldGFkYXRhL3RhYmxlL21vZC5ycw==) | 18.75% | [13 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/700?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [crates/core/src/metadata/table/v2\_reader.rs](https://app.codecov.io/gh/apache/hudi-rs/pull/700?src=pr&el=tree&filepath=crates%2Fcore%2Fsrc%2Fmetadata%2Ftable%2Fv2_reader.rs&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y3JhdGVzL2NvcmUvc3JjL21ldGFkYXRhL3RhYmxlL3YyX3JlYWRlci5ycw==) | 50.00% | [2 Missing :warning: ](https://app.codecov.io/gh/apache/hudi-rs/pull/700?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #700 +/- ##
==========================================
- Coverage 79.62% 79.34% -0.28%
==========================================
Files 121 122 +1
Lines 11892 11988 +96
==========================================
+ Hits 9469 9512 +43
- Misses 2423 2476 +53
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/hudi-rs/pull/700?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(core): read the metadata table as Arrow, and expose hudi-rs over a C ABI for JVM callers [hudi-rs]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #700:
URL: https://github.com/apache/hudi-rs/pull/700#discussion_r3919083767
##########
crates/jvm-ffi/src/lib.rs:
##########
@@ -0,0 +1,444 @@
+/*
+ * 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.
+ */
+//! Crate `hudi-core`.
+//!
+//! # The [config] module is responsible for managing configurations.
+//!
+
+//! A plain C ABI over hudi-rs, so a JVM can read a file slice.
+//!
+//! The `cpp` crate already exposes the same reads, but `cxx` generates
+//! C++-mangled symbols and `CxxString`/`CxxVector` arguments, neither of which a
+//! JVM can bind to. This crate restates those calls in C: null-terminated
+//! strings in, opaque handles and an `ArrowArrayStream` out. It holds no reading
+//! logic of its own.
+//!
+//! # Ownership
+//!
+//! Every non-null pointer returned here is owned by the caller and must be
+//! released exactly once, with the matching `hudi_ffi_free_*`. Releasing twice
+//! is a double free; not releasing leaks the reader or the Arrow buffers behind
+//! the stream.
+//!
+//! # Errors and panics
+//!
+//! A failing call returns null and leaves a message retrievable with
+//! [`hudi_ffi_last_error`] until the next call on the same thread. Panics are
+//! caught at every boundary and turned into that same null-plus-message, because
+//! a panic unwinding into the JVM aborts the process.
+
+use std::cell::RefCell;
+use std::ffi::{CStr, CString, c_char};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+
+use arrow::array::RecordBatchIterator;
+use arrow::ffi_stream::FFI_ArrowArrayStream;
+use hudi::file_group::reader::FileGroupReader;
+use hudi::table::{ReadOptions, Table};
+
+thread_local! {
+ static LAST_ERROR: RefCell<Option<CString>> = const { RefCell::new(None) };
+}
+
+fn set_error(message: impl Into<String>) {
+ let text = message.into();
+ let encoded = CString::new(text).unwrap_or_else(|_| {
+ CString::new("hudi-rs error message contained an interior nul byte")
+ .expect("this literal has no nul")
+ });
+ LAST_ERROR.with(|slot| *slot.borrow_mut() = Some(encoded));
+}
+
+/// Run `body`, turning any error or panic into a null return plus a message.
+///
+/// Every exported function goes through this. A panic that unwinds across the C
+/// ABI aborts the process, so catching it here is what keeps a Rust bug from
+/// killing the JVM that called us.
+fn guard<T>(what: &str, body: impl FnOnce() -> Result<*mut T, String>) -> *mut T {
+ match catch_unwind(AssertUnwindSafe(body)) {
+ Ok(Ok(ptr)) => ptr,
+ Ok(Err(message)) => {
+ set_error(format!("{what}: {message}"));
+ std::ptr::null_mut()
+ }
+ Err(payload) => {
+ let detail = payload
+ .downcast_ref::<&str>()
+ .map(|s| (*s).to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .unwrap_or_else(|| "non-string panic payload".to_string());
+ set_error(format!("{what} panicked: {detail}"));
+ std::ptr::null_mut()
+ }
+ }
+}
+
+/// Borrow a C string, naming the argument in any error.
+unsafe fn as_str<'a>(ptr: *const c_char, name: &str) -> Result<&'a str, String> {
+ if ptr.is_null() {
+ return Err(format!("{name} is null"));
+ }
+ unsafe { CStr::from_ptr(ptr) }
+ .to_str()
+ .map_err(|e| format!("{name} is not valid UTF-8: {e}"))
+}
+
+/// Borrow `len` C strings from an array.
+unsafe fn as_strs<'a>(
+ ptr: *const *const c_char,
+ len: usize,
+ name: &str,
+) -> Result<Vec<&'a str>, String> {
+ if len == 0 {
+ return Ok(Vec::new());
+ }
+ if ptr.is_null() {
+ return Err(format!("{name} is null but len is {len}"));
+ }
+ let slice = unsafe { std::slice::from_raw_parts(ptr, len) };
+ slice
+ .iter()
+ .enumerate()
+ .map(|(i, p)| unsafe { as_str(*p, &format!("{name}[{i}]")) })
+ .collect()
+}
+
+/// A reader plus the runtime its async calls are driven on.
+///
+/// The runtime is owned here rather than created per read: a current-thread
+/// runtime is cheap to build but the reader holds object-store clients whose
+/// connection pools should outlive one call.
+pub struct HudiFfiReader {
+ inner: FileGroupReader,
+ runtime: tokio::runtime::Runtime,
+}
+
+/// A table handle, for the metadata read.
+///
+/// Separate from [`HudiFfiReader`] because the two wrap different things: a file
+/// group reader is handed a slice, while a metadata read resolves its own.
+pub struct HudiFfiTable {
+ inner: Table,
+ runtime: tokio::runtime::Runtime,
+}
+
+/// Open the table at `base_uri` for a metadata read.
+///
+/// # Safety
+/// Pointers must be null-terminated C strings valid for the call, and the two
+/// option arrays must each hold `option_count` entries.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_table_open(
+ base_uri: *const c_char,
+ option_keys: *const *const c_char,
+ option_values: *const *const c_char,
+ option_count: usize,
+) -> *mut HudiFfiTable {
+ guard("hudi_ffi_table_open", || {
+ let base_uri = unsafe { as_str(base_uri, "base_uri") }?;
+ let keys = unsafe { as_strs(option_keys, option_count, "option_keys") }?;
+ let values = unsafe { as_strs(option_values, option_count, "option_values") }?;
+ let options: Vec<(&str, &str)> = keys.into_iter().zip(values).collect();
+
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("failed to build a tokio runtime: {e}"))?;
+ let inner = runtime
+ .block_on(Table::new_with_options(base_uri, options))
+ .map_err(|e| format!("failed to open {base_uri}: {e}"))?;
+ Ok(Box::into_raw(Box::new(HudiFfiTable { inner, runtime })))
+ })
+}
+
+/// Read the metadata table's `files` partition as an Arrow stream.
+///
+/// `keys` may be null when `key_count` is zero, which reads every record.
+/// Keys are matched as stored, so a non-partitioned table's record is asked for
+/// as `"."`.
+///
+/// # Safety
+/// `table` must come from [`hudi_ffi_table_open`] and not yet be freed. The
+/// returned stream must be released with [`hudi_ffi_free_stream`] exactly once,
+/// including on paths where the caller stops reading partway.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_read_metadata_files_partition(
+ table: *mut HudiFfiTable,
+ keys: *const *const c_char,
+ key_count: usize,
+) -> *mut FFI_ArrowArrayStream {
+ guard("hudi_ffi_read_metadata_files_partition", || {
+ if table.is_null() {
+ return Err("table is null".to_string());
+ }
+ let table = unsafe { &*table };
+ let keys = unsafe { as_strs(keys, key_count, "keys") }?;
+
+ let batch = table
+ .runtime
+ .block_on(table.inner.read_metadata_table_files_partition_arrow(&keys))
+ .map_err(|e| format!("failed to read the metadata table: {e}"))?;
+
+ let schema = batch.schema();
+ let iterator = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
+ Ok(Box::into_raw(Box::new(FFI_ArrowArrayStream::new(
+ Box::new(iterator),
+ ))))
+ })
+}
+
+/// Release a table handle. Null is accepted and ignored.
+///
+/// # Safety
+/// `table` must come from [`hudi_ffi_table_open`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_table(table: *mut HudiFfiTable) {
+ if !table.is_null() {
+ drop(unsafe { Box::from_raw(table) });
+ }
+}
+
+/// Open a reader for the table at `base_uri`.
+///
+/// `option_keys` and `option_values` are parallel arrays of `option_count`
+/// entries. Returns null on failure; see [`hudi_ffi_last_error`].
+///
+/// # Safety
+/// All pointers must be null-terminated C strings valid for the call, and the
+/// two option arrays must each hold `option_count` entries.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_reader_open(
+ base_uri: *const c_char,
+ option_keys: *const *const c_char,
+ option_values: *const *const c_char,
+ option_count: usize,
+) -> *mut HudiFfiReader {
+ guard("hudi_ffi_reader_open", || {
+ let base_uri = unsafe { as_str(base_uri, "base_uri") }?;
+ let keys = unsafe { as_strs(option_keys, option_count, "option_keys") }?;
+ let values = unsafe { as_strs(option_values, option_count, "option_values") }?;
+ let options: Vec<(&str, &str)> = keys.into_iter().zip(values).collect();
+
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("failed to build a tokio runtime: {e}"))?;
+ let inner = runtime
+ .block_on(FileGroupReader::new_with_options(base_uri, options))
+ .map_err(|e| format!("failed to open {base_uri}: {e}"))?;
+ Ok(Box::into_raw(Box::new(HudiFfiReader { inner, runtime })))
+ })
+}
+
+/// Read one file slice, named by its base file and log files, into a stream.
+///
+/// `log_file_paths` may be null when `log_file_count` is zero, which reads a
+/// base file on its own. Returns null on failure.
+///
+/// # Safety
+/// `reader` must come from [`hudi_ffi_reader_open`] and not yet be freed. The
+/// returned stream must be released with [`hudi_ffi_free_stream`] exactly once,
+/// including on paths where the caller aborts partway through reading it.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_read_slice(
+ reader: *mut HudiFfiReader,
+ base_file_path: *const c_char,
+ log_file_paths: *const *const c_char,
+ log_file_count: usize,
+) -> *mut FFI_ArrowArrayStream {
+ guard("hudi_ffi_read_slice", || {
+ if reader.is_null() {
+ return Err("reader is null".to_string());
+ }
+ let reader = unsafe { &*reader };
+ let base_file_path = unsafe { as_str(base_file_path, "base_file_path") }?;
+ let logs = unsafe { as_strs(log_file_paths, log_file_count, "log_file_paths") }?;
+
+ let batch = reader
+ .runtime
+ .block_on(reader.inner.read_file_slice_from_paths(
+ base_file_path,
+ logs,
+ &ReadOptions::new(),
+ ))
+ .map_err(|e| format!("failed to read {base_file_path}: {e}"))?;
+
+ // Eager, matching what the cxx bridge does today: one batch wrapped as a
+ // stream. `read_file_slice_from_paths_stream` exists and would avoid
+ // materialising the slice, but driving an async stream from inside the
+ // stream's synchronous `get_next` callback needs care that belongs in its
+ // own change.
+ let schema = batch.schema();
+ let iterator = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
+ Ok(Box::into_raw(Box::new(FFI_ArrowArrayStream::new(
+ Box::new(iterator),
+ ))))
+ })
+}
+
+/// Release a reader. Null is accepted and ignored.
+///
+/// # Safety
+/// `reader` must come from [`hudi_ffi_reader_open`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_reader(reader: *mut HudiFfiReader) {
+ if !reader.is_null() {
+ drop(unsafe { Box::from_raw(reader) });
+ }
+}
+
+/// Release a stream. Null is accepted and ignored.
+///
+/// # Safety
+/// `stream` must come from [`hudi_ffi_read_slice`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_stream(stream: *mut FFI_ArrowArrayStream) {
+ if !stream.is_null() {
+ drop(unsafe { Box::from_raw(stream) });
+ }
+}
+
+/// The last error on this thread, or null if the last call succeeded.
+///
+/// The returned string is owned by this library and stays valid until the next
+/// call on the same thread.
+#[unsafe(no_mangle)]
+pub extern "C" fn hudi_ffi_last_error() -> *const c_char {
Review Comment:
Fixed in 2a1a582f — guard now clears the slot on entry, with a regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(core): read the metadata table as Arrow, and expose hudi-rs over a C ABI for JVM callers [hudi-rs]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua commented on code in PR #700:
URL: https://github.com/apache/hudi-rs/pull/700#discussion_r3919057286
##########
crates/jvm-ffi/src/lib.rs:
##########
@@ -0,0 +1,444 @@
+/*
+ * 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.
+ */
+//! Crate `hudi-core`.
Review Comment:
non-blocking: this first doc block ("Crate `hudi-core`. / # The [config] module is responsible for managing configurations.") is a copy-paste leftover from hudi-core sitting above the real crate doc — worth deleting.
##########
crates/core/src/metadata/table/mod.rs:
##########
@@ -234,9 +257,49 @@ impl Table {
&self,
keys: &[&str],
) -> Result<HashMap<String, FilesPartitionRecord>> {
- let Some(timestamp) = self.timeline.get_latest_commit_timestamp_as_option() else {
+ let Some((reader, file_slice)) = self.files_partition_reader().await? else {
return Ok(HashMap::new());
};
+ reader.read_files_partition(&file_slice, keys).await
+ }
+
+ /// The `files` partition's merged batch, for a caller that wants Arrow.
+ ///
+ /// An empty batch when the metadata table has no commits, matching the empty
+ /// map the decoded form returns in that case.
+ ///
+ /// # Note
+ /// Must be called on a METADATA table instance.
+ pub(crate) async fn read_files_partition_batch(
+ &self,
+ keys: &[&str],
+ ) -> Result<arrow_array::RecordBatch> {
+ let Some((reader, file_slice)) = self.files_partition_reader().await? else {
+ return Ok(arrow_array::RecordBatch::new_empty(std::sync::Arc::new(
Review Comment:
A metadata table with no commits returns an empty batch with an empty schema, and through the FFI that becomes a schema-less ArrowArrayStream. Is that the contract the JVM side wants, or should the empty case carry the files-partition schema so a consumer can bind columns without special-casing emptiness?
##########
crates/jvm-ffi/src/lib.rs:
##########
@@ -0,0 +1,444 @@
+/*
+ * 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.
+ */
+//! Crate `hudi-core`.
+//!
+//! # The [config] module is responsible for managing configurations.
+//!
+
+//! A plain C ABI over hudi-rs, so a JVM can read a file slice.
+//!
+//! The `cpp` crate already exposes the same reads, but `cxx` generates
+//! C++-mangled symbols and `CxxString`/`CxxVector` arguments, neither of which a
+//! JVM can bind to. This crate restates those calls in C: null-terminated
+//! strings in, opaque handles and an `ArrowArrayStream` out. It holds no reading
+//! logic of its own.
+//!
+//! # Ownership
+//!
+//! Every non-null pointer returned here is owned by the caller and must be
+//! released exactly once, with the matching `hudi_ffi_free_*`. Releasing twice
+//! is a double free; not releasing leaks the reader or the Arrow buffers behind
+//! the stream.
+//!
+//! # Errors and panics
+//!
+//! A failing call returns null and leaves a message retrievable with
+//! [`hudi_ffi_last_error`] until the next call on the same thread. Panics are
+//! caught at every boundary and turned into that same null-plus-message, because
+//! a panic unwinding into the JVM aborts the process.
+
+use std::cell::RefCell;
+use std::ffi::{CStr, CString, c_char};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+
+use arrow::array::RecordBatchIterator;
+use arrow::ffi_stream::FFI_ArrowArrayStream;
+use hudi::file_group::reader::FileGroupReader;
+use hudi::table::{ReadOptions, Table};
+
+thread_local! {
+ static LAST_ERROR: RefCell<Option<CString>> = const { RefCell::new(None) };
+}
+
+fn set_error(message: impl Into<String>) {
+ let text = message.into();
+ let encoded = CString::new(text).unwrap_or_else(|_| {
+ CString::new("hudi-rs error message contained an interior nul byte")
+ .expect("this literal has no nul")
+ });
+ LAST_ERROR.with(|slot| *slot.borrow_mut() = Some(encoded));
+}
+
+/// Run `body`, turning any error or panic into a null return plus a message.
+///
+/// Every exported function goes through this. A panic that unwinds across the C
+/// ABI aborts the process, so catching it here is what keeps a Rust bug from
+/// killing the JVM that called us.
+fn guard<T>(what: &str, body: impl FnOnce() -> Result<*mut T, String>) -> *mut T {
+ match catch_unwind(AssertUnwindSafe(body)) {
+ Ok(Ok(ptr)) => ptr,
+ Ok(Err(message)) => {
+ set_error(format!("{what}: {message}"));
+ std::ptr::null_mut()
+ }
+ Err(payload) => {
+ let detail = payload
+ .downcast_ref::<&str>()
+ .map(|s| (*s).to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .unwrap_or_else(|| "non-string panic payload".to_string());
+ set_error(format!("{what} panicked: {detail}"));
+ std::ptr::null_mut()
+ }
+ }
+}
+
+/// Borrow a C string, naming the argument in any error.
+unsafe fn as_str<'a>(ptr: *const c_char, name: &str) -> Result<&'a str, String> {
+ if ptr.is_null() {
+ return Err(format!("{name} is null"));
+ }
+ unsafe { CStr::from_ptr(ptr) }
+ .to_str()
+ .map_err(|e| format!("{name} is not valid UTF-8: {e}"))
+}
+
+/// Borrow `len` C strings from an array.
+unsafe fn as_strs<'a>(
+ ptr: *const *const c_char,
+ len: usize,
+ name: &str,
+) -> Result<Vec<&'a str>, String> {
+ if len == 0 {
+ return Ok(Vec::new());
+ }
+ if ptr.is_null() {
+ return Err(format!("{name} is null but len is {len}"));
+ }
+ let slice = unsafe { std::slice::from_raw_parts(ptr, len) };
+ slice
+ .iter()
+ .enumerate()
+ .map(|(i, p)| unsafe { as_str(*p, &format!("{name}[{i}]")) })
+ .collect()
+}
+
+/// A reader plus the runtime its async calls are driven on.
+///
+/// The runtime is owned here rather than created per read: a current-thread
+/// runtime is cheap to build but the reader holds object-store clients whose
+/// connection pools should outlive one call.
+pub struct HudiFfiReader {
Review Comment:
non-blocking: JVM callers will freely move and share these handles across threads. They are Send + Sync today (verified with a compile probe), but nothing pins that — a future !Sync field in Table would silently turn concurrent use into UB with no compile failure here. A const assertion (fn assert_send_sync<T: Send + Sync>() {} applied to both handle types) would make the C-ABI assumption checkable.
##########
crates/jvm-ffi/src/lib.rs:
##########
@@ -0,0 +1,444 @@
+/*
+ * 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.
+ */
+//! Crate `hudi-core`.
+//!
+//! # The [config] module is responsible for managing configurations.
+//!
+
+//! A plain C ABI over hudi-rs, so a JVM can read a file slice.
+//!
+//! The `cpp` crate already exposes the same reads, but `cxx` generates
+//! C++-mangled symbols and `CxxString`/`CxxVector` arguments, neither of which a
+//! JVM can bind to. This crate restates those calls in C: null-terminated
+//! strings in, opaque handles and an `ArrowArrayStream` out. It holds no reading
+//! logic of its own.
+//!
+//! # Ownership
+//!
+//! Every non-null pointer returned here is owned by the caller and must be
+//! released exactly once, with the matching `hudi_ffi_free_*`. Releasing twice
+//! is a double free; not releasing leaks the reader or the Arrow buffers behind
+//! the stream.
+//!
+//! # Errors and panics
+//!
+//! A failing call returns null and leaves a message retrievable with
+//! [`hudi_ffi_last_error`] until the next call on the same thread. Panics are
+//! caught at every boundary and turned into that same null-plus-message, because
+//! a panic unwinding into the JVM aborts the process.
+
+use std::cell::RefCell;
+use std::ffi::{CStr, CString, c_char};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+
+use arrow::array::RecordBatchIterator;
+use arrow::ffi_stream::FFI_ArrowArrayStream;
+use hudi::file_group::reader::FileGroupReader;
+use hudi::table::{ReadOptions, Table};
+
+thread_local! {
+ static LAST_ERROR: RefCell<Option<CString>> = const { RefCell::new(None) };
+}
+
+fn set_error(message: impl Into<String>) {
+ let text = message.into();
+ let encoded = CString::new(text).unwrap_or_else(|_| {
+ CString::new("hudi-rs error message contained an interior nul byte")
+ .expect("this literal has no nul")
+ });
+ LAST_ERROR.with(|slot| *slot.borrow_mut() = Some(encoded));
+}
+
+/// Run `body`, turning any error or panic into a null return plus a message.
+///
+/// Every exported function goes through this. A panic that unwinds across the C
+/// ABI aborts the process, so catching it here is what keeps a Rust bug from
+/// killing the JVM that called us.
+fn guard<T>(what: &str, body: impl FnOnce() -> Result<*mut T, String>) -> *mut T {
+ match catch_unwind(AssertUnwindSafe(body)) {
+ Ok(Ok(ptr)) => ptr,
+ Ok(Err(message)) => {
+ set_error(format!("{what}: {message}"));
+ std::ptr::null_mut()
+ }
+ Err(payload) => {
+ let detail = payload
+ .downcast_ref::<&str>()
+ .map(|s| (*s).to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .unwrap_or_else(|| "non-string panic payload".to_string());
+ set_error(format!("{what} panicked: {detail}"));
+ std::ptr::null_mut()
+ }
+ }
+}
+
+/// Borrow a C string, naming the argument in any error.
+unsafe fn as_str<'a>(ptr: *const c_char, name: &str) -> Result<&'a str, String> {
+ if ptr.is_null() {
+ return Err(format!("{name} is null"));
+ }
+ unsafe { CStr::from_ptr(ptr) }
+ .to_str()
+ .map_err(|e| format!("{name} is not valid UTF-8: {e}"))
+}
+
+/// Borrow `len` C strings from an array.
+unsafe fn as_strs<'a>(
+ ptr: *const *const c_char,
+ len: usize,
+ name: &str,
+) -> Result<Vec<&'a str>, String> {
+ if len == 0 {
+ return Ok(Vec::new());
+ }
+ if ptr.is_null() {
+ return Err(format!("{name} is null but len is {len}"));
+ }
+ let slice = unsafe { std::slice::from_raw_parts(ptr, len) };
+ slice
+ .iter()
+ .enumerate()
+ .map(|(i, p)| unsafe { as_str(*p, &format!("{name}[{i}]")) })
+ .collect()
+}
+
+/// A reader plus the runtime its async calls are driven on.
+///
+/// The runtime is owned here rather than created per read: a current-thread
+/// runtime is cheap to build but the reader holds object-store clients whose
+/// connection pools should outlive one call.
+pub struct HudiFfiReader {
+ inner: FileGroupReader,
+ runtime: tokio::runtime::Runtime,
+}
+
+/// A table handle, for the metadata read.
+///
+/// Separate from [`HudiFfiReader`] because the two wrap different things: a file
+/// group reader is handed a slice, while a metadata read resolves its own.
+pub struct HudiFfiTable {
+ inner: Table,
+ runtime: tokio::runtime::Runtime,
+}
+
+/// Open the table at `base_uri` for a metadata read.
+///
+/// # Safety
+/// Pointers must be null-terminated C strings valid for the call, and the two
+/// option arrays must each hold `option_count` entries.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_table_open(
+ base_uri: *const c_char,
+ option_keys: *const *const c_char,
+ option_values: *const *const c_char,
+ option_count: usize,
+) -> *mut HudiFfiTable {
+ guard("hudi_ffi_table_open", || {
+ let base_uri = unsafe { as_str(base_uri, "base_uri") }?;
+ let keys = unsafe { as_strs(option_keys, option_count, "option_keys") }?;
+ let values = unsafe { as_strs(option_values, option_count, "option_values") }?;
+ let options: Vec<(&str, &str)> = keys.into_iter().zip(values).collect();
+
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("failed to build a tokio runtime: {e}"))?;
+ let inner = runtime
+ .block_on(Table::new_with_options(base_uri, options))
+ .map_err(|e| format!("failed to open {base_uri}: {e}"))?;
+ Ok(Box::into_raw(Box::new(HudiFfiTable { inner, runtime })))
+ })
+}
+
+/// Read the metadata table's `files` partition as an Arrow stream.
+///
+/// `keys` may be null when `key_count` is zero, which reads every record.
+/// Keys are matched as stored, so a non-partitioned table's record is asked for
+/// as `"."`.
+///
+/// # Safety
+/// `table` must come from [`hudi_ffi_table_open`] and not yet be freed. The
+/// returned stream must be released with [`hudi_ffi_free_stream`] exactly once,
+/// including on paths where the caller stops reading partway.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_read_metadata_files_partition(
+ table: *mut HudiFfiTable,
+ keys: *const *const c_char,
+ key_count: usize,
+) -> *mut FFI_ArrowArrayStream {
+ guard("hudi_ffi_read_metadata_files_partition", || {
+ if table.is_null() {
+ return Err("table is null".to_string());
+ }
+ let table = unsafe { &*table };
+ let keys = unsafe { as_strs(keys, key_count, "keys") }?;
+
+ let batch = table
+ .runtime
+ .block_on(table.inner.read_metadata_table_files_partition_arrow(&keys))
+ .map_err(|e| format!("failed to read the metadata table: {e}"))?;
+
+ let schema = batch.schema();
+ let iterator = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
+ Ok(Box::into_raw(Box::new(FFI_ArrowArrayStream::new(
+ Box::new(iterator),
+ ))))
+ })
+}
+
+/// Release a table handle. Null is accepted and ignored.
+///
+/// # Safety
+/// `table` must come from [`hudi_ffi_table_open`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_table(table: *mut HudiFfiTable) {
+ if !table.is_null() {
+ drop(unsafe { Box::from_raw(table) });
+ }
+}
+
+/// Open a reader for the table at `base_uri`.
+///
+/// `option_keys` and `option_values` are parallel arrays of `option_count`
+/// entries. Returns null on failure; see [`hudi_ffi_last_error`].
+///
+/// # Safety
+/// All pointers must be null-terminated C strings valid for the call, and the
+/// two option arrays must each hold `option_count` entries.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_reader_open(
+ base_uri: *const c_char,
+ option_keys: *const *const c_char,
+ option_values: *const *const c_char,
+ option_count: usize,
+) -> *mut HudiFfiReader {
+ guard("hudi_ffi_reader_open", || {
+ let base_uri = unsafe { as_str(base_uri, "base_uri") }?;
+ let keys = unsafe { as_strs(option_keys, option_count, "option_keys") }?;
+ let values = unsafe { as_strs(option_values, option_count, "option_values") }?;
+ let options: Vec<(&str, &str)> = keys.into_iter().zip(values).collect();
+
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("failed to build a tokio runtime: {e}"))?;
+ let inner = runtime
+ .block_on(FileGroupReader::new_with_options(base_uri, options))
+ .map_err(|e| format!("failed to open {base_uri}: {e}"))?;
+ Ok(Box::into_raw(Box::new(HudiFfiReader { inner, runtime })))
+ })
+}
+
+/// Read one file slice, named by its base file and log files, into a stream.
+///
+/// `log_file_paths` may be null when `log_file_count` is zero, which reads a
+/// base file on its own. Returns null on failure.
+///
+/// # Safety
+/// `reader` must come from [`hudi_ffi_reader_open`] and not yet be freed. The
+/// returned stream must be released with [`hudi_ffi_free_stream`] exactly once,
+/// including on paths where the caller aborts partway through reading it.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_read_slice(
+ reader: *mut HudiFfiReader,
+ base_file_path: *const c_char,
+ log_file_paths: *const *const c_char,
+ log_file_count: usize,
+) -> *mut FFI_ArrowArrayStream {
+ guard("hudi_ffi_read_slice", || {
+ if reader.is_null() {
+ return Err("reader is null".to_string());
+ }
+ let reader = unsafe { &*reader };
+ let base_file_path = unsafe { as_str(base_file_path, "base_file_path") }?;
+ let logs = unsafe { as_strs(log_file_paths, log_file_count, "log_file_paths") }?;
+
+ let batch = reader
+ .runtime
+ .block_on(reader.inner.read_file_slice_from_paths(
+ base_file_path,
+ logs,
+ &ReadOptions::new(),
+ ))
+ .map_err(|e| format!("failed to read {base_file_path}: {e}"))?;
+
+ // Eager, matching what the cxx bridge does today: one batch wrapped as a
+ // stream. `read_file_slice_from_paths_stream` exists and would avoid
+ // materialising the slice, but driving an async stream from inside the
+ // stream's synchronous `get_next` callback needs care that belongs in its
+ // own change.
+ let schema = batch.schema();
+ let iterator = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
+ Ok(Box::into_raw(Box::new(FFI_ArrowArrayStream::new(
+ Box::new(iterator),
+ ))))
+ })
+}
+
+/// Release a reader. Null is accepted and ignored.
+///
+/// # Safety
+/// `reader` must come from [`hudi_ffi_reader_open`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_reader(reader: *mut HudiFfiReader) {
+ if !reader.is_null() {
+ drop(unsafe { Box::from_raw(reader) });
+ }
+}
+
+/// Release a stream. Null is accepted and ignored.
+///
+/// # Safety
+/// `stream` must come from [`hudi_ffi_read_slice`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_stream(stream: *mut FFI_ArrowArrayStream) {
+ if !stream.is_null() {
+ drop(unsafe { Box::from_raw(stream) });
+ }
+}
+
+/// The last error on this thread, or null if the last call succeeded.
+///
+/// The returned string is owned by this library and stays valid until the next
+/// call on the same thread.
+#[unsafe(no_mangle)]
+pub extern "C" fn hudi_ffi_last_error() -> *const c_char {
Review Comment:
The doc here promises null when the last call succeeded, but guard() only ever sets the slot on failure and nothing clears it — after one failed call, every later successful call still reports the old message, so a binding that checks last_error rather than the return value throws stale errors. Clearing the slot at the top of guard() (or on the success arm) would make the doc true.
##########
crates/jvm-ffi/src/lib.rs:
##########
@@ -0,0 +1,444 @@
+/*
+ * 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.
+ */
+//! Crate `hudi-core`.
+//!
+//! # The [config] module is responsible for managing configurations.
+//!
+
+//! A plain C ABI over hudi-rs, so a JVM can read a file slice.
+//!
+//! The `cpp` crate already exposes the same reads, but `cxx` generates
+//! C++-mangled symbols and `CxxString`/`CxxVector` arguments, neither of which a
+//! JVM can bind to. This crate restates those calls in C: null-terminated
+//! strings in, opaque handles and an `ArrowArrayStream` out. It holds no reading
+//! logic of its own.
+//!
+//! # Ownership
+//!
+//! Every non-null pointer returned here is owned by the caller and must be
+//! released exactly once, with the matching `hudi_ffi_free_*`. Releasing twice
+//! is a double free; not releasing leaks the reader or the Arrow buffers behind
+//! the stream.
+//!
+//! # Errors and panics
+//!
+//! A failing call returns null and leaves a message retrievable with
+//! [`hudi_ffi_last_error`] until the next call on the same thread. Panics are
+//! caught at every boundary and turned into that same null-plus-message, because
+//! a panic unwinding into the JVM aborts the process.
+
+use std::cell::RefCell;
+use std::ffi::{CStr, CString, c_char};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+
+use arrow::array::RecordBatchIterator;
+use arrow::ffi_stream::FFI_ArrowArrayStream;
+use hudi::file_group::reader::FileGroupReader;
+use hudi::table::{ReadOptions, Table};
+
+thread_local! {
+ static LAST_ERROR: RefCell<Option<CString>> = const { RefCell::new(None) };
+}
+
+fn set_error(message: impl Into<String>) {
+ let text = message.into();
+ let encoded = CString::new(text).unwrap_or_else(|_| {
+ CString::new("hudi-rs error message contained an interior nul byte")
+ .expect("this literal has no nul")
+ });
+ LAST_ERROR.with(|slot| *slot.borrow_mut() = Some(encoded));
+}
+
+/// Run `body`, turning any error or panic into a null return plus a message.
+///
+/// Every exported function goes through this. A panic that unwinds across the C
+/// ABI aborts the process, so catching it here is what keeps a Rust bug from
+/// killing the JVM that called us.
+fn guard<T>(what: &str, body: impl FnOnce() -> Result<*mut T, String>) -> *mut T {
+ match catch_unwind(AssertUnwindSafe(body)) {
+ Ok(Ok(ptr)) => ptr,
+ Ok(Err(message)) => {
+ set_error(format!("{what}: {message}"));
+ std::ptr::null_mut()
+ }
+ Err(payload) => {
+ let detail = payload
+ .downcast_ref::<&str>()
+ .map(|s| (*s).to_string())
+ .or_else(|| payload.downcast_ref::<String>().cloned())
+ .unwrap_or_else(|| "non-string panic payload".to_string());
+ set_error(format!("{what} panicked: {detail}"));
+ std::ptr::null_mut()
+ }
+ }
+}
+
+/// Borrow a C string, naming the argument in any error.
+unsafe fn as_str<'a>(ptr: *const c_char, name: &str) -> Result<&'a str, String> {
+ if ptr.is_null() {
+ return Err(format!("{name} is null"));
+ }
+ unsafe { CStr::from_ptr(ptr) }
+ .to_str()
+ .map_err(|e| format!("{name} is not valid UTF-8: {e}"))
+}
+
+/// Borrow `len` C strings from an array.
+unsafe fn as_strs<'a>(
+ ptr: *const *const c_char,
+ len: usize,
+ name: &str,
+) -> Result<Vec<&'a str>, String> {
+ if len == 0 {
+ return Ok(Vec::new());
+ }
+ if ptr.is_null() {
+ return Err(format!("{name} is null but len is {len}"));
+ }
+ let slice = unsafe { std::slice::from_raw_parts(ptr, len) };
+ slice
+ .iter()
+ .enumerate()
+ .map(|(i, p)| unsafe { as_str(*p, &format!("{name}[{i}]")) })
+ .collect()
+}
+
+/// A reader plus the runtime its async calls are driven on.
+///
+/// The runtime is owned here rather than created per read: a current-thread
+/// runtime is cheap to build but the reader holds object-store clients whose
+/// connection pools should outlive one call.
+pub struct HudiFfiReader {
+ inner: FileGroupReader,
+ runtime: tokio::runtime::Runtime,
+}
+
+/// A table handle, for the metadata read.
+///
+/// Separate from [`HudiFfiReader`] because the two wrap different things: a file
+/// group reader is handed a slice, while a metadata read resolves its own.
+pub struct HudiFfiTable {
+ inner: Table,
+ runtime: tokio::runtime::Runtime,
+}
+
+/// Open the table at `base_uri` for a metadata read.
+///
+/// # Safety
+/// Pointers must be null-terminated C strings valid for the call, and the two
+/// option arrays must each hold `option_count` entries.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_table_open(
+ base_uri: *const c_char,
+ option_keys: *const *const c_char,
+ option_values: *const *const c_char,
+ option_count: usize,
+) -> *mut HudiFfiTable {
+ guard("hudi_ffi_table_open", || {
+ let base_uri = unsafe { as_str(base_uri, "base_uri") }?;
+ let keys = unsafe { as_strs(option_keys, option_count, "option_keys") }?;
+ let values = unsafe { as_strs(option_values, option_count, "option_values") }?;
+ let options: Vec<(&str, &str)> = keys.into_iter().zip(values).collect();
+
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("failed to build a tokio runtime: {e}"))?;
+ let inner = runtime
+ .block_on(Table::new_with_options(base_uri, options))
+ .map_err(|e| format!("failed to open {base_uri}: {e}"))?;
+ Ok(Box::into_raw(Box::new(HudiFfiTable { inner, runtime })))
+ })
+}
+
+/// Read the metadata table's `files` partition as an Arrow stream.
+///
+/// `keys` may be null when `key_count` is zero, which reads every record.
+/// Keys are matched as stored, so a non-partitioned table's record is asked for
+/// as `"."`.
+///
+/// # Safety
+/// `table` must come from [`hudi_ffi_table_open`] and not yet be freed. The
+/// returned stream must be released with [`hudi_ffi_free_stream`] exactly once,
+/// including on paths where the caller stops reading partway.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_read_metadata_files_partition(
+ table: *mut HudiFfiTable,
+ keys: *const *const c_char,
+ key_count: usize,
+) -> *mut FFI_ArrowArrayStream {
+ guard("hudi_ffi_read_metadata_files_partition", || {
+ if table.is_null() {
+ return Err("table is null".to_string());
+ }
+ let table = unsafe { &*table };
+ let keys = unsafe { as_strs(keys, key_count, "keys") }?;
+
+ let batch = table
+ .runtime
+ .block_on(table.inner.read_metadata_table_files_partition_arrow(&keys))
+ .map_err(|e| format!("failed to read the metadata table: {e}"))?;
+
+ let schema = batch.schema();
+ let iterator = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
+ Ok(Box::into_raw(Box::new(FFI_ArrowArrayStream::new(
+ Box::new(iterator),
+ ))))
+ })
+}
+
+/// Release a table handle. Null is accepted and ignored.
+///
+/// # Safety
+/// `table` must come from [`hudi_ffi_table_open`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_table(table: *mut HudiFfiTable) {
+ if !table.is_null() {
+ drop(unsafe { Box::from_raw(table) });
+ }
+}
+
+/// Open a reader for the table at `base_uri`.
+///
+/// `option_keys` and `option_values` are parallel arrays of `option_count`
+/// entries. Returns null on failure; see [`hudi_ffi_last_error`].
+///
+/// # Safety
+/// All pointers must be null-terminated C strings valid for the call, and the
+/// two option arrays must each hold `option_count` entries.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_reader_open(
+ base_uri: *const c_char,
+ option_keys: *const *const c_char,
+ option_values: *const *const c_char,
+ option_count: usize,
+) -> *mut HudiFfiReader {
+ guard("hudi_ffi_reader_open", || {
+ let base_uri = unsafe { as_str(base_uri, "base_uri") }?;
+ let keys = unsafe { as_strs(option_keys, option_count, "option_keys") }?;
+ let values = unsafe { as_strs(option_values, option_count, "option_values") }?;
+ let options: Vec<(&str, &str)> = keys.into_iter().zip(values).collect();
+
+ let runtime = tokio::runtime::Builder::new_current_thread()
+ .enable_all()
+ .build()
+ .map_err(|e| format!("failed to build a tokio runtime: {e}"))?;
+ let inner = runtime
+ .block_on(FileGroupReader::new_with_options(base_uri, options))
+ .map_err(|e| format!("failed to open {base_uri}: {e}"))?;
+ Ok(Box::into_raw(Box::new(HudiFfiReader { inner, runtime })))
+ })
+}
+
+/// Read one file slice, named by its base file and log files, into a stream.
+///
+/// `log_file_paths` may be null when `log_file_count` is zero, which reads a
+/// base file on its own. Returns null on failure.
+///
+/// # Safety
+/// `reader` must come from [`hudi_ffi_reader_open`] and not yet be freed. The
+/// returned stream must be released with [`hudi_ffi_free_stream`] exactly once,
+/// including on paths where the caller aborts partway through reading it.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_read_slice(
+ reader: *mut HudiFfiReader,
+ base_file_path: *const c_char,
+ log_file_paths: *const *const c_char,
+ log_file_count: usize,
+) -> *mut FFI_ArrowArrayStream {
+ guard("hudi_ffi_read_slice", || {
+ if reader.is_null() {
+ return Err("reader is null".to_string());
+ }
+ let reader = unsafe { &*reader };
+ let base_file_path = unsafe { as_str(base_file_path, "base_file_path") }?;
+ let logs = unsafe { as_strs(log_file_paths, log_file_count, "log_file_paths") }?;
+
+ let batch = reader
+ .runtime
+ .block_on(reader.inner.read_file_slice_from_paths(
+ base_file_path,
+ logs,
+ &ReadOptions::new(),
+ ))
+ .map_err(|e| format!("failed to read {base_file_path}: {e}"))?;
+
+ // Eager, matching what the cxx bridge does today: one batch wrapped as a
+ // stream. `read_file_slice_from_paths_stream` exists and would avoid
+ // materialising the slice, but driving an async stream from inside the
+ // stream's synchronous `get_next` callback needs care that belongs in its
+ // own change.
+ let schema = batch.schema();
+ let iterator = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
+ Ok(Box::into_raw(Box::new(FFI_ArrowArrayStream::new(
+ Box::new(iterator),
+ ))))
+ })
+}
+
+/// Release a reader. Null is accepted and ignored.
+///
+/// # Safety
+/// `reader` must come from [`hudi_ffi_reader_open`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_reader(reader: *mut HudiFfiReader) {
+ if !reader.is_null() {
+ drop(unsafe { Box::from_raw(reader) });
+ }
+}
+
+/// Release a stream. Null is accepted and ignored.
+///
+/// # Safety
+/// `stream` must come from [`hudi_ffi_read_slice`] and be released once only.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn hudi_ffi_free_stream(stream: *mut FFI_ArrowArrayStream) {
Review Comment:
non-blocking: the three free functions and hudi_ffi_last_error are the only exports outside guard(), and drop(Box::from_raw(..)) can panic (e.g. dropping the tokio runtime from inside another runtime's context), which would unwind across extern "C" and abort the JVM — the exact failure this crate exists to prevent. Wrapping the drops in catch_unwind would keep the contract uniform.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(core): read the metadata table as Arrow, and expose hudi-rs over a C ABI for JVM callers [hudi-rs]
Posted by "yihua (via GitHub)" <gi...@apache.org>.
yihua merged PR #700:
URL: https://github.com/apache/hudi-rs/pull/700
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org