feat(c): expose read builder scan options and time-travel through the C binding (#524)

Author: JunRuiLee

bindings/c/src/table.rs

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::collections::HashMap;
 use std::ffi::c_void;
 
 use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
@@ -66,30 +67,141 @@ pub unsafe extern "C" fn paimon_table_free(table: *mut paimon_table) {
     free_table_wrapper(table, |t| t.inner);
 }
 
+/// Time-travel selector option names, in the core's resolution priority order.
+const TIME_TRAVEL_SELECTORS: [&str; 4] = [
+    "scan.timestamp-millis",
+    "scan.version",
+    "scan.snapshot-id",
+    "scan.tag-name",
+];
+
+/// Build a `ReadBuilderState` from a table and a scan-option map, resolving any
+/// time-travel selector at construction time.
+///
+/// Rejects more than one time-travel selector up front (the core silently falls
+/// back on a conflict, which would misattribute the failure). Otherwise runs the
+/// core's `copy_with_time_travel`, which validates unsupported scan options and
+/// resolves the selector; a set selector that does not resolve to a snapshot is
+/// an error, so a mistyped or missing selector can never silently read latest.
+unsafe fn new_read_builder_state(
+    table: &Table,
+    options: HashMap<String, String>,
+) -> Result<ReadBuilderState, *mut paimon_error> {
+    let present: Vec<&str> = TIME_TRAVEL_SELECTORS
+        .iter()
+        .copied()
+        .filter(|name| options.contains_key(*name))
+        .collect();
+    if present.len() > 1 {
+        return Err(paimon_error::new(
+            PaimonErrorCode::InvalidInput,
+            format!(
+                "Only one time-travel selector may be set, found: {}",
+                present.join(", ")
+            ),
+        ));
+    }
+    let selector = TIME_TRAVEL_SELECTORS
+        .iter()
+        .find_map(|&name| options.get(name).map(|v| (name.to_string(), v.clone())));
+
+    let resolved = match runtime().block_on(table.copy_with_time_travel(options)) {
+        Ok(t) => t,
+        Err(e) => return Err(paimon_error::from_paimon(e)),
+    };
+
+    if let Some((name, value)) = selector {
+        if !resolved.has_resolved_travel_snapshot() {
+            return Err(paimon_error::new(
+                PaimonErrorCode::InvalidInput,
+                format!("time-travel selector {name}={value} did not resolve to any snapshot"),
+            ));
+        }
+    }
+
+    Ok(ReadBuilderState {
+        table: resolved,
+        projected_columns: None,
+        filter: None,
+        case_sensitive: true,
+    })
+}
+
 /// Create a new ReadBuilder from a Table.
 ///
 /// # Safety
 /// `table` must be a valid pointer from `paimon_catalog_get_table`, or null (returns error).
 #[no_mangle]
 pub unsafe extern "C" fn paimon_table_new_read_builder(
     table: *const paimon_table,
+) -> paimon_result_read_builder {
+    paimon_table_new_read_builder_with_options(table, std::ptr::null(), 0)
+}
+
+/// Create a ReadBuilder from a Table with scan options (e.g. time-travel
+/// selectors `scan.snapshot-id` / `scan.tag-name` / `scan.timestamp-millis` /
+/// `scan.version`). At most one time-travel selector may be set. A selector that
+/// does not resolve to a snapshot is an error (never a silent read-of-latest).
+///
+/// # Safety
+/// `table` must be a valid pointer. `options` must be a valid pointer to
+/// `options_len` `paimon_option` values, or null when `options_len` is 0.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_new_read_builder_with_options(
+    table: *const paimon_table,
+    options: *const paimon_option,
+    options_len: usize,
 ) -> paimon_result_read_builder {
     if let Err(e) = check_non_null(table, "table") {
         return paimon_result_read_builder {
             read_builder: std::ptr::null_mut(),
             error: e,
         };
     }
+    if options.is_null() && options_len > 0 {
+        return paimon_result_read_builder {
+            read_builder: std::ptr::null_mut(),
+            error: paimon_error::new(
+                PaimonErrorCode::InvalidInput,
+                "null options pointer with non-zero length".to_string(),
+            ),
+        };
+    }
+    let mut map = HashMap::with_capacity(options_len);
+    if options_len > 0 {
+        let slice = std::slice::from_raw_parts(options, options_len);
+        for opt in slice {
+            let key = match validate_cstr(opt.key, "option key") {
+                Ok(s) => s,
+                Err(e) => {
+                    return paimon_result_read_builder {
+                        read_builder: std::ptr::null_mut(),
+                        error: e,
+                    }
+                }
+            };
+            let value = match validate_cstr(opt.value, "option value") {
+                Ok(s) => s,
+                Err(e) => {
+                    return paimon_result_read_builder {
+                        read_builder: std::ptr::null_mut(),
+                        error: e,
+                    }
+                }
+            };
+            map.insert(key, value);
+        }
+    }
     let table_ref = &*((*table).inner as *const Table);
-    let state = ReadBuilderState {
-        table: table_ref.clone(),
-        projected_columns: None,
-        filter: None,
-        case_sensitive: true,
-    };
-    paimon_result_read_builder {
-        read_builder: box_read_builder_state(state),
-        error: std::ptr::null_mut(),
+    match new_read_builder_state(table_ref, map) {
+        Ok(state) => paimon_result_read_builder {
+            read_builder: box_read_builder_state(state),
+            error: std::ptr::null_mut(),
+        },
+        Err(e) => paimon_result_read_builder {
+            read_builder: std::ptr::null_mut(),
+            error: e,
+        },
     }
 }
 
@@ -1611,6 +1723,18 @@ const _: unsafe extern "C" fn(
     paimon_datum,
 ) -> paimon_result_predicate = paimon_predicate_not_between;
 
+// Read builder ABI signature guards. These pin the C-linked read-builder
+// constructors so an accidental signature change fails to compile rather than
+// silently breaking header consumers. To add behavior, introduce a new
+// `paimon_table_new_read_builder_*` symbol instead of changing one of these.
+const _: unsafe extern "C" fn(*const paimon_table) -> paimon_result_read_builder =
+    paimon_table_new_read_builder;
+const _: unsafe extern "C" fn(
+    *const paimon_table,
+    *const paimon_option,
+    usize,
+) -> paimon_result_read_builder = paimon_table_new_read_builder_with_options;
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1848,4 +1972,163 @@ mod tests {
             paimon_table_free(table);
         }
     }
+
+    /// Assert the result is a built read builder (read_builder non-null, error
+    /// null) and free it.
+    unsafe fn assert_rb_ok_and_free(r: paimon_result_read_builder) {
+        assert!(!r.read_builder.is_null(), "expected a read builder");
+        assert!(r.error.is_null(), "expected no error");
+        paimon_read_builder_free(r.read_builder);
+    }
+
+    /// Assert the result is an error (read_builder null, error non-null) and free
+    /// it.
+    unsafe fn assert_rb_err_and_free(r: paimon_result_read_builder) {
+        assert!(r.read_builder.is_null(), "expected no read builder");
+        assert!(!r.error.is_null(), "expected an error");
+        paimon_error_free(r.error);
+    }
+
+    /// Assert the result is an error, extract its code and message, then free the
+    /// error exactly once. Lets a test pin the error identity (which failure
+    /// fired) instead of only its shape.
+    unsafe fn assert_rb_err_code_message(r: paimon_result_read_builder) -> (i32, String) {
+        assert!(r.read_builder.is_null(), "expected no read builder");
+        assert!(!r.error.is_null(), "expected an error");
+        let err = &*r.error;
+        let code = err.code;
+        let message = if err.message.data.is_null() {
+            String::new()
+        } else {
+            let bytes = std::slice::from_raw_parts(err.message.data, err.message.len);
+            String::from_utf8_lossy(bytes).into_owned()
+        };
+        paimon_error_free(r.error);
+        (code, message)
+    }
+
+    /// A `paimon_option` borrowing `key`/`value` (kept alive by the caller).
+    fn opt(key: &std::ffi::CStr, value: &std::ffi::CStr) -> paimon_option {
+        paimon_option {
+            key: key.as_ptr(),
+            value: value.as_ptr(),
+        }
+    }
+
+    #[test]
+    fn empty_options_builds_like_plain_new_read_builder() {
+        unsafe {
+            let table = boxed_test_table();
+            // Null pointer + zero length.
+            assert_rb_ok_and_free(paimon_table_new_read_builder_with_options(
+                table,
+                std::ptr::null(),
+                0,
+            ));
+            // Non-null pointer to a zero-length array + zero length.
+            let empty: [paimon_option; 0] = [];
+            assert_rb_ok_and_free(paimon_table_new_read_builder_with_options(
+                table,
+                empty.as_ptr(),
+                0,
+            ));
+            // The plain entry point (delegates with an empty map).
+            assert_rb_ok_and_free(paimon_table_new_read_builder(table));
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn non_selector_option_builds() {
+        unsafe {
+            let table = boxed_test_table();
+            let k = CString::new("some.unrelated.option").unwrap();
+            let v = CString::new("value").unwrap();
+            let opts = [opt(&k, &v)];
+            assert_rb_ok_and_free(paimon_table_new_read_builder_with_options(
+                table,
+                opts.as_ptr(),
+                1,
+            ));
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn more_than_one_selector_is_rejected() {
+        unsafe {
+            let table = boxed_test_table();
+            let k1 = CString::new("scan.snapshot-id").unwrap();
+            let v1 = CString::new("1").unwrap();
+            let k2 = CString::new("scan.tag-name").unwrap();
+            let v2 = CString::new("t").unwrap();
+            let opts = [opt(&k1, &v1), opt(&k2, &v2)];
+            let (code, message) = assert_rb_err_code_message(
+                paimon_table_new_read_builder_with_options(table, opts.as_ptr(), 2),
+            );
+            // Binding-constructed rejection: InvalidInput naming both selectors.
+            assert_eq!(code, PaimonErrorCode::InvalidInput as i32);
+            assert!(
+                message.contains("scan.snapshot-id") && message.contains("scan.tag-name"),
+                "message should name both selectors, got: {message}"
+            );
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn unsupported_scan_option_is_rejected() {
+        unsafe {
+            let table = boxed_test_table();
+            let k = CString::new("scan.watermark").unwrap();
+            let v = CString::new("0").unwrap();
+            let opts = [opt(&k, &v)];
+            // Core's validate_scan_options rejects this before resolution; the
+            // binding surfaces core's Unsupported code.
+            let (code, message) = assert_rb_err_code_message(
+                paimon_table_new_read_builder_with_options(table, opts.as_ptr(), 1),
+            );
+            assert_eq!(code, PaimonErrorCode::Unsupported as i32);
+            assert!(
+                message.contains("not supported"),
+                "message should say the option is not supported, got: {message}"
+            );
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn malformed_selector_value_does_not_silently_read_latest() {
+        unsafe {
+            let table = boxed_test_table();
+            let k = CString::new("scan.snapshot-id").unwrap();
+            let v = CString::new("abc").unwrap();
+            let opts = [opt(&k, &v)];
+            // Core swallows the parse error and falls back; the binding reports
+            // the unified "did not resolve" error rather than building a
+            // latest-reading builder.
+            let (code, message) = assert_rb_err_code_message(
+                paimon_table_new_read_builder_with_options(table, opts.as_ptr(), 1),
+            );
+            assert_eq!(code, PaimonErrorCode::InvalidInput as i32);
+            assert!(
+                message.contains("did not resolve"),
+                "message should report the selector did not resolve, got: {message}"
+            );
+            paimon_table_free(table);
+        }
+    }
+
+    #[test]
+    fn null_options_with_nonzero_length_is_rejected() {
+        unsafe {
+            let table = boxed_test_table();
+            assert_rb_err_and_free(paimon_table_new_read_builder_with_options(
+                table,
+                std::ptr::null(),
+                2,
+            ));
+            paimon_table_free(table);
+        }
+    }
 }