feat: add host read provider for dataset I/O

Cargo.toml

@@ -18,6 +18,8 @@ rust-version = "1.91.0"
 crate-type = ["cdylib", "staticlib", "rlib"]
 
 [dependencies]
+async-trait = "0.1"
+bytes = "1"
 lance = { git = "https://github.com/lance-format/lance.git", rev = "e934cc2c", features = ["substrait"] }
 lance-core = { git = "https://github.com/lance-format/lance.git", rev = "e934cc2c" }
 lance-file = { git = "https://github.com/lance-format/lance.git", rev = "e934cc2c" }
@@ -39,6 +41,7 @@ tokio = { version = "1", features = ["rt-multi-thread", "sync"] }
 futures = "0.3"
 log = "0.4"
 libc = "0.2"
+object_store = "0.13.2"
 pin-project = "1.0"
 prost = "0.14"
 snafu = "0.9"

README.md

@@ -68,6 +68,7 @@ Based on the [liblance RFC](https://github.com/lance-format/lance/discussions/60
 | [x] | Async scan | Callback-based `lance_scanner_scan_async()` for non-blocking scans |
 | [x] | Dataset metadata | `lance_dataset_version()`, `lance_dataset_count_rows()`, `lance_dataset_latest_version()` |
 | [x] | Filter pushdown | `lance_scanner_set_substrait_filter()` accepts a serialized Substrait `ExtendedExpression`; `lance_scanner_additional_sql_filter()` adds SQL predicates with AND before scanning starts |
+| [x] | Host read provider | Query engines can route range reads through their native file reader and data cache while independently sharing Lance Session caches |
 
 ## Building
 
@@ -197,6 +198,43 @@ auto ds = lance::Dataset::open_with_session(session, "data.lance");
 auto stats = session.cache_stats();
 ```
 
+### Combine a shared Session with a host read provider
+
+`LanceSession` and `LanceReadProvider` are intentionally independent. A
+Session is long-lived and caches portable metadata/index state. A read provider
+is bound to a Dataset and owns the current query engine's file readers,
+credentials, cancellation state, data-cache policy, and I/O statistics.
+
+```c
+LanceReadProviderOps provider_ops = {
+    .open = host_open,
+    .read_at = host_read_at,
+    .close_reader = host_close_reader,
+    .destroy_context = host_destroy_context,
+    .last_error_message = host_last_error_message,
+};
+LanceReadProvider* provider =
+    lance_read_provider_new(&provider_ops, host_context, 16);
+
+LanceDatasetOpenOptions options = {
+    .uri = "s3://bucket/data.lance",
+    .storage_options = storage_options,
+    .version = 0,
+    .session = session,
+    .read_provider = provider,
+};
+LanceDataset* ds = lance_dataset_open_with_options(&options);
+
+/* The Dataset retains both shared objects. */
+lance_read_provider_close(provider);
+lance_session_close(session);
+```
+
+The provider receives object metadata from Lance's native object store and is
+used only for object contents. Listing, metadata lookup, and writes continue to
+use the native store. Returning `LANCE_READ_NOT_SUPPORTED` from `open` falls
+back to the native read path for that object.
+
 ### Open at a specific version
 
 `lance_dataset_open` takes a `version` argument — `0` means the latest, any

include/lance/lance.h

@@ -182,6 +182,7 @@ typedef struct LanceDataset  LanceDataset;
 typedef struct LanceScanner  LanceScanner;
 typedef struct LanceBatch    LanceBatch;
 typedef struct LanceSession  LanceSession;
+typedef struct LanceReadProvider LanceReadProvider;
 typedef struct LanceVersions LanceVersions;
 typedef struct LanceDataStatistics LanceDataStatistics;
 typedef struct LanceIndexSegmentBuilder LanceIndexSegmentBuilder;
@@ -233,8 +234,104 @@ int32_t lance_session_get_cache_stats(
     LanceSessionCacheStats* out_stats
 );
 
+/* ─── Host read provider ─── */
+
+/** Status returned by host read-provider callbacks. */
+typedef enum LanceReadStatus {
+    LANCE_READ_OK = 0,
+    LANCE_READ_NOT_SUPPORTED = 1,
+    LANCE_READ_NOT_FOUND = 2,
+    LANCE_READ_CANCELLED = 3,
+    LANCE_READ_IO_ERROR = 4
+} LanceReadStatus;
+
+/**
+ * Stable identity of an object opened through a host read provider.
+ *
+ * All strings are borrowed and remain valid only for the duration of the
+ * `open` callback. `path` is relative to `store_prefix`. `e_tag` and
+ * `version` may be NULL.
+ */
+typedef struct LanceFileIdentity {
+    const char* store_prefix;
+    const char* path;
+    uint64_t size;
+    int64_t last_modified_millis;
+    const char* e_tag;
+    const char* version;
+} LanceFileIdentity;
+
+/**
+ * Host callbacks for random-access reads.
+ *
+ * `open` and `read_at` may run concurrently on blocking worker threads and
+ * must be thread-safe. A successful `open` must set `out_reader` to a non-NULL
+ * value. `close_reader` is called exactly once for every successfully opened
+ * reader. Callbacks must not throw or unwind across the C ABI.
+ */
+typedef struct LanceReadProviderOps {
+    int32_t (*open)(
+        void* context,
+        const LanceFileIdentity* identity,
+        void** out_reader
+    );
+    int32_t (*read_at)(
+        void* reader,
+        uint64_t offset,
+        uint8_t* buffer,
+        uint64_t length,
+        uint64_t* bytes_read
+    );
+    void (*close_reader)(void* reader);
+    void (*destroy_context)(void* context);
+    /**
+     * Return the current thread's last host error. The string is borrowed and
+     * copied by lance-c immediately. May be NULL.
+     */
+    const char* (*last_error_message)(void* context);
+} LanceReadProviderOps;
+
+/**
+ * Create a reference-counted host read provider.
+ *
+ * `max_concurrency` bounds simultaneous blocking `open` and `read_at`
+ * callbacks and must be greater than zero. On success the provider owns
+ * `context` and, when supplied, eventually calls `destroy_context` exactly
+ * once.
+ */
+LanceReadProvider* lance_read_provider_new(
+    const LanceReadProviderOps* ops,
+    void* context,
+    uint32_t max_concurrency
+);
+
+/**
+ * Close a provider handle. Datasets already opened with it remain valid and
+ * retain the provider until their outstanding reads are complete.
+ */
+void lance_read_provider_close(LanceReadProvider* provider);
+
 /* ─── Dataset lifecycle ─── */
 
+/**
+ * Complete options for opening a dataset.
+ *
+ * `session` and `read_provider` are optional and borrowed for the duration of
+ * this call. The returned dataset retains shared ownership of both.
+ */
+typedef struct LanceDatasetOpenOptions {
+    const char* uri;
+    const char* const* storage_options;
+    uint64_t version;
+    const LanceSession* session;
+    const LanceReadProvider* read_provider;
+} LanceDatasetOpenOptions;
+
+/** Open a dataset with an optional shared session and host read provider. */
+LanceDataset* lance_dataset_open_with_options(
+    const LanceDatasetOpenOptions* options
+);
+
 /**
  * Open a Lance dataset.
  *
@@ -1016,7 +1113,7 @@ typedef struct {
  * best-effort and may be omitted if they cannot be materialized. `metrics` is
  * NULL when `metrics_len` is zero.
  */
-typedef struct {
+typedef struct LanceScanStatistics {
     uint64_t iops;
     uint64_t requests;
     uint64_t bytes_read;

include/lance/lance.hpp

@@ -190,6 +190,40 @@ class Session {
     const LanceSession* c_handle() const { return handle_.get(); }
 };
 
+// ─── Host Read Provider ─────────────────────────────────────────────────────
+
+/// Reference-counted host random-access reader. The provider takes ownership
+/// of `context` after successful construction and, when supplied, invokes
+/// `destroy_context` when the last Dataset/provider handle releases it.
+class ReadProvider {
+    Handle<LanceReadProvider, lance_read_provider_close> handle_;
+
+public:
+    ReadProvider(
+        const LanceReadProviderOps& ops,
+        void* context,
+        uint32_t max_concurrency)
+        : handle_(lance_read_provider_new(&ops, context, max_concurrency)) {
+        if (!handle_) check_error();
+    }
+
+    ReadProvider(ReadProvider&&) noexcept = default;
+    ReadProvider& operator=(ReadProvider&&) noexcept = default;
+    ReadProvider(const ReadProvider&) = delete;
+    ReadProvider& operator=(const ReadProvider&) = delete;
+
+    const LanceReadProvider* c_handle() const { return handle_.get(); }
+};
+
+/// Composable options for Dataset::open. Session caches and host reads have
+/// independent lifetimes and may be enabled separately or together.
+struct DatasetOpenOptions {
+    std::vector<std::pair<std::string, std::string>> storage_options;
+    uint64_t version = 0;
+    const Session* session = nullptr;
+    const ReadProvider* read_provider = nullptr;
+};
+
 // ─── Process-local FTS query context ────────────────────────────────────────
 
 /// Immutable, query-specific global BM25 scorer plus pinned FTS segment list.
@@ -225,6 +259,21 @@ class Dataset {
     }
 
 public:
+    /// Open a dataset with composable Session and host-read-provider options.
+    static Dataset open(const std::string& uri, const DatasetOpenOptions& options) {
+        auto kv = to_c_storage_options(options.storage_options);
+        LanceDatasetOpenOptions c_options{
+            uri.c_str(),
+            options.storage_options.empty() ? nullptr : kv.data(),
+            options.version,
+            options.session ? options.session->c_handle() : nullptr,
+            options.read_provider ? options.read_provider->c_handle() : nullptr,
+        };
+        auto* ds = lance_dataset_open_with_options(&c_options);
+        if (!ds) check_error();
+        return Dataset(ds);
+    }
+
     /// Open a dataset at the given URI. Pass `version` = 0 (the default) for
     /// the latest, or a specific version id from `versions()` to check out
     /// that version, e.g. `lance::Dataset::open("data.lance", {}, /*version=*/42)`.

src/dataset.rs

@@ -13,9 +13,11 @@ use arrow_schema::Schema as ArrowSchema;
 use lance::Dataset;
 use lance::dataset::builder::DatasetBuilder;
 use lance_core::Result;
+use lance_io::object_store::{ObjectStoreParams, StorageOptionsAccessor};
 
 use crate::error::{ffi_try, swallow_unwind};
 use crate::helpers;
+use crate::read_provider::{LanceReadProvider, ProviderWrapper};
 use crate::runtime::block_on;
 use crate::session::LanceSession;
 use crate::stream_guard::guarded_ffi_stream_from_reader;
@@ -25,6 +27,20 @@ pub struct LanceDataset {
     pub(crate) inner: RwLock<Arc<Dataset>>,
 }
 
+/// Complete set of options for opening a dataset.
+///
+/// `session` and `read_provider` are borrowed for this call. The returned
+/// dataset retains shared ownership of their underlying state.
+#[repr(C)]
+#[derive(Clone, Copy, Debug)]
+pub struct LanceDatasetOpenOptions {
+    pub uri: *const c_char,
+    pub storage_options: *const *const c_char,
+    pub version: u64,
+    pub session: *const LanceSession,
+    pub read_provider: *const LanceReadProvider,
+}
+
 impl LanceDataset {
     /// Take a consistent snapshot of the inner dataset.
     /// Returns a cloned Arc so the caller can hold it without keeping the lock.
@@ -122,7 +138,7 @@ pub unsafe extern "C" fn lance_dataset_open(
     version: u64,
 ) -> *mut LanceDataset {
     ffi_try!(
-        unsafe { open_dataset_inner(uri, storage_options, version, None) },
+        unsafe { open_dataset_inner(uri, storage_options, version, None, None) },
         null
     )
 }
@@ -156,23 +172,71 @@ unsafe fn open_dataset_with_session_inner(
         ));
     }
     let session = unsafe { &*session };
-    unsafe { open_dataset_inner(uri, storage_options, version, Some(session)) }
+    unsafe { open_dataset_inner(uri, storage_options, version, Some(session), None) }
+}
+
+/// Open a dataset with an optional shared session and host read provider.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn lance_dataset_open_with_options(
+    options: *const LanceDatasetOpenOptions,
+) -> *mut LanceDataset {
+    ffi_try!(unsafe { open_dataset_with_options_inner(options) }, null)
+}
+
+unsafe fn open_dataset_with_options_inner(
+    options: *const LanceDatasetOpenOptions,
+) -> Result<*mut LanceDataset> {
+    if options.is_null() {
+        return Err(lance_core::Error::invalid_input_source(
+            "options must not be NULL".into(),
+        ));
+    }
+    let options = unsafe { &*options };
+    let session = if options.session.is_null() {
+        None
+    } else {
+        Some(unsafe { &*options.session })
+    };
+    let read_provider = if options.read_provider.is_null() {
+        None
+    } else {
+        Some(unsafe { &*options.read_provider })
+    };
+    unsafe {
+        open_dataset_inner(
+            options.uri,
+            options.storage_options,
+            options.version,
+            session,
+            read_provider,
+        )
+    }
 }
 
 unsafe fn open_dataset_inner(
     uri: *const c_char,
     storage_options: *const *const c_char,
     version: u64,
     session: Option<&LanceSession>,
+    read_provider: Option<&LanceReadProvider>,
 ) -> Result<*mut LanceDataset> {
     let uri_str = unsafe { helpers::parse_c_string(uri)? }
         .ok_or_else(|| lance_core::Error::invalid_input_source("uri must not be NULL".into()))?;
 
     let opts = unsafe { helpers::parse_storage_options(storage_options)? };
 
     let mut builder = DatasetBuilder::from_uri(uri_str);
-    if !opts.is_empty() {
-        builder = builder.with_storage_options(opts);
+    if !opts.is_empty() || read_provider.is_some() {
+        let mut store_params = ObjectStoreParams::default();
+        if !opts.is_empty() {
+            store_params.storage_options_accessor =
+                Some(Arc::new(StorageOptionsAccessor::with_static_options(opts)));
+        }
+        if let Some(read_provider) = read_provider {
+            store_params.object_store_wrapper =
+                Some(Arc::new(ProviderWrapper::new(&read_provider.inner)));
+        }
+        builder = builder.with_store_params(store_params);
     }
     if version != 0 {
         builder = builder.with_version(version);