internal review

Author: myrrc

proposed/0030-c-scan-api.md

@@ -32,9 +32,9 @@ capabilities to the host:
 ## Motivation
 
 Vortex is a Rust project, and thus integration with non-Rust clients requires a
-decent amount of scaffolding[^3] which lowers down the probability of such
+decent amount of scaffolding[^3] which lowers the probability of such
 integration. As reading files is most important feature for a file format,
-exposing it to C-compatible uses lowers down the integration costs.
+exposing it to C-compatible users lowers integration costs.
 
 Additional motivation is our DuckDB integration which requires C++-to-Rust
 and Rust-to-C++ bridging. Replacing it with direct calls to C API would make
@@ -49,9 +49,9 @@ reasons:
 - Lack of support for exporting to `ArrowSchema` and `ArrowArrayStream`.
 - Lack of introspection over produced `vx_array` objects.
 
-    Vortex also has a C++ API in vortex-cxx which wraps Rust code. Its future is
-    outside of scope of this proposal but it's expected to wrap C scan API,
-    removing dependency on cxx.rs for vortex-cxx.
+> Vortex also has a C++ API in vortex-cxx which wraps Rust code. Its future is
+> outside of scope of this proposal but it's expected to wrap C scan API,
+> removing dependency on cxx.rs for vortex-cxx.
 
 ## Motivation for layered API
 
@@ -70,10 +70,9 @@ no way to suspend to host on IO operations. Exposing only the middle layer
 requires sync hosts to manage their own event loop which is tedious.
 
 Another benefit of splitting API into layers is that it can be discussed and
-implemented separately.
-
-As this is a big change, this PR will from now on focus just on the high level
-scan API. Other levels may be implemented later on demand.
+implemented separately. As this is a big change, this PR will from now on focus
+just on the high level scan API. Other levels may be implemented later on
+demand.
 
 ## Overview
 
@@ -94,6 +93,15 @@ scan API. Other levels may be implemented later on demand.
  └─►Array (thread unsafe)
 ```
 
+## Event loop
+
+Vortex manages the event loop by using a CurrentThreadRuntime runtime handle
+which in turn has a `smol::Executor<'static>` without a background thread pool.
+This means the event loop will run on host's thread pool scheduler (in other
+terms, use the calling thread) implicitly. In case of multiple threads, the
+underlying executor is shared between them, and drives from from all of these
+threads. The executor is initialized once on loading FFI library.
+
 ## DataSource
 
 A DataSource is a reference to multiple possibly remote files. When created, it
@@ -105,38 +113,41 @@ scans from a DataSource.
 // bindgen
 typedef struct vx_data_source vx_data_source;
 typedef struct vx_file_handle vx_file_handle;
-typedef struct vx_cache_handle *vx_cache_handle;
 
-typedef void (*vx_list_callback)(void* userdata, const char* name, int is_dir);
-typedef void (*vx_glob_callback)(void* userdata, const char* file);
+typedef void* vx_fs_userdata;
+typedef void (*vx_list_callback)(vx_fs_userdata userdata, const char* name,
+    int is_dir);
+typedef void (*vx_glob_callback)(vx_fs_userdata userdata, const char* file);
 
 typedef struct vx_data_source_options {
+    const char* files;
+
     // (1) Filesystem customization
-    bool (*fs_use_vortex)(const char* schema, const char* path);
-    void (*fs_set_userdata)(void* userdata);
+    vx_fs_userdata fs_userdata;
+    void (*fs_free_userdata)(vx_fs_userdata fs_userdata);
 
     // Called after glob expansion, single-file mode
-    vx_file_handle (*fs_open)(void* userdata, const char* path,
+    vx_file_handle (*fs_open)(vx_fs_userdata userdata, const char* path,
         vx_error** err);
-    vx_file_handle (*fs_create)(void* userdata, const char* path,
+    vx_file_handle (*fs_create)(vx_fs_userdata userdata, const char* path,
         vx_error** err);
 
     // non-recursive, callback is invoked with each path
-    void fs_list(void* userdata, const char* path, vx_list_callback cb,
-        vx_error** err);
+    void (*fs_list)(vx_fs_userdata userdata, const char* path,
+        vx_list_callback cb, vx_error** err);
 
-    void fs_close(vx_file_handle handle);
-    uint64_t fs_size(vx_file_handle handle, vx_error *err);
+    void (*fs_close)(vx_file_handle handle);
+    uint64_t (*fs_size)(vx_file_handle handle, vx_error** err);
 
     // positional read, doesn't change file open cursor
-    void fs_read(vx_file_handle handle, uint64_t offset, size_t len,
+    void (*fs_read)(vx_file_handle handle, uint64_t offset, uint64_t len,
         uint8_t *buffer, vx_error** err);
 
     // not needed for scanning but makes FS API complete
-    void fs_write(vx_file_handle handle, uint64_t offset, size_t len,
+    void (*fs_write)(vx_file_handle handle, uint64_t offset, uint64_t len,
         uint8_t *buffer, vx_error** err);
 
-    void fs_sync(vx_file_handle handle, vx_error** err);
+    void (*fs_sync)(vx_file_handle handle, vx_error** err);
 
     // (2) Globbing customization
     void (*glob)(const char* glob, vx_glob_callback cb, vx_error** err);
@@ -146,12 +157,12 @@ typedef struct vx_data_source_options {
 // be freed by the caller using release callback. If err is populated, out
 // is not set.
 void vx_dtype_to_arrow_schema(const vx_dtype* dtype, ArrowSchema* out,
-    vx_error* err);
+    vx_error** err);
 
 /// Create a new owned datasource which must be freed by the caller.
 const vx_data_source *
 vx_data_source_new(const vx_session *session,
-    const vx_data_source_options *opts, vx_error_out err);
+    const vx_data_source_options *opts, vx_error** err);
 
 // datasource is Arc'd inside so a clone creates another reference rather
 // than cloning the state fully.
@@ -176,19 +187,32 @@ void vx_data_source_get_row_count(const vx_data_source *ds,
 void vx_data_source_free(const vx_data_source *ds);
 ```
 
+> const vx_data_source* may be misleading as it's actually owned, but that's
+> limitation of arc_dyn_wrapper! as for now. This would likely be changed soon
+
 1. Open local or remote file. Allow using Vortex's filesystem or host's
-   filesystem e.g. DuckDB fs. Allow partial customization e.g. DuckDB fs for
-   local reads, but Vortex fs for remote reads. Host filesystem customization
+   filesystem e.g. DuckDB fs. Host filesystem customization
    point has the benefit of not duplicating credentials e.g. S3 access key
    between host and Vortex. Local implementation may be more performant.
+   User must either implement all `fs_*` callback and set `fs_userdata`, or set
+   all callbacks in (1) and `fs_userdata` to NULL.
+
 2. Open single file or multiple files. Hosts may have their own glob expansion
    [^4] which does HTTP IO.
 
 When both customization points are implemented, Vortex offloads all IO
 to a query engine.
 
-    Memory allocation customization is out of scope of this proposal, but it's
-    possible for Vortex to expose bringing allocator from outside for buffers.
+> Memory allocation customization is out of scope of this proposal, but it's
+> possible for Vortex to expose bringing allocator from outside for buffers.
+
+To allow stateful filesystem operations, a DataSource allows providing a custom
+filesystem userdata with associated deleter. It may be set per DataSource.
+Filesystem customizations do syncrohonous IO without batching. Clients who want
+control over IO should use the middle layer or lower.
+As filesystem callbacks are synchronous, middle layer would use different option
+structs, and there would be no way to re-use DataSource or other structures
+between layers.
 
 ## Scan
 
@@ -207,13 +231,14 @@ either and this is a low priority task.
 Scan options:
 
 ```c
+// roaring bitmap include and exclude won't be supported as for now because of
+// exposure complexity
 typedef enum {
     VX_S_INCLUDE_ALL = 0,
     VX_S_INCLUDE_RANGE = 1,
     VX_S_EXCLUDE_RANGE = 2,
 } vx_scan_selection_include;
 
-// Roaring bitmaps won't be supported.
 // If selection is VX_S_INCLUDE_ALL, idx is not read. idx is copied
 // by host on scan invocation and can be freed after first partition is
 // requested
@@ -226,12 +251,13 @@ typedef struct {
 typedef struct vx_scan_options {
     const vx_expression *projection; // NULL means "return all columns"
     const vx_expression *filter; // NULL means "no filter"
-    uint64_t row_range_begin; // Inclusive, [0; 0) means "all rows"
+    // special case: row_range_begin == 0 && row_range_end == 0 means "all rows"
+    uint64_t row_range_begin; // Inclusive
     uint64_t row_range_end; // Exclusive
     vx_scan_selection selection;
     uint64_t limit; // 0 means no limit
     uint64_t max_threads; // 0 means no limit.
-    bool ordered; // 0 means unordered scan
+    int ordered; // 0 means unordered scan
 } vx_scan_options;
 ```
 
@@ -249,7 +275,7 @@ typedef enum {
     VX_ESTIMATE_INEXACT = 2,
 } vx_estimate_boundary;
 
-// If type is VX_P_ESTIMATE_UNKNOWN, estimate is not populated.
+// If type is VX_ESTIMATE_UNKNOWN, estimate is not populated.
 typedef struct {
     uint64_t estimate;
     vx_estimate_boundary type;
@@ -259,10 +285,17 @@ typedef struct {
 // estimate->estimate to distribute work.
 // options and estimate fields may be NUL.
 // Calling vx_data_source_scan doesn't do IO unless vx_scan_next is called.
-// vx_scan can outlive vx_data_source.
+//
+// As an implementation detail, a scan has a reference to a data source thus
+// a scan can outlive a data source.
 vx_scan *
 vx_data_source_scan(const vx_data_source *ds, const vx_scan_options *options,
     vx_estimate* estimate, vx_error **err);
+
+// Return scan progress from 0.0 to 1.0.
+// Reports estimates of read vs. total rows. Thread-safe.
+// May be inaccurate, main use case is progress bars in clients.
+double vx_scan_progress(const vx_scan *scan);
 ```
 
 ## Partition
@@ -277,21 +310,36 @@ data.
 // Get next owned partition out of a scan request.
 // Caller must free this partition using vx_partition_free.
 // This method is thread-safe.
+// You can call vx_scan_free or vx_data_source_free while you are holding
+// partitions.
 // Hosts are encouraged to create a worker thread per partition.
-// Returns NULL and doesn't set err on exhaustion.
-// Returns NULL and sets err on error.
+// Returns NULL and sets *err to NULL on exhaustion.
+// Returns NULL and sets *err on error.
 vx_partition *vx_scan_next(vx_scan *scan, vx_error **err);
 
 // Request an array stream in Arrow format from a partition, consuming it
-// fully. Not thread-safe, should be called once.
-// stream is owned and must be freed using the release callback
+// fully. Calling vx_partition_next on partition after calling this function
+// is undefined behaviour.
+// User does not need to call vx_partition_free in partition after calling
+// this function.
+// Not thread-safe, should be called once.
+// stream is owned and must be freed using the release callback.
+// Does not modify stream and sets err on error.
 void vx_partition_scan_arrow(const vx_partition *partition,
-    ArrowArrayStream *stream, vx_error_out err);
+    ArrowArrayStream *stream, vx_error** err);
 
 // Thread-unsafe. Get an owned vx_array of an iterator.
-// Returns NULL and sets err to NULL if iterator is exhausted.
+// Calling from multiple threads is undefined behaviour.
 // Array must be freed by caller.
+// You can call vx_scan_free, vx_data_source_free, or vx_partition_free 
+// while you are holding arrays from it.
+// Returns NULL and sets *err to NULL on exhaustion.
+// Returns NULL and sets *err on error.
 const vx_array *vx_partition_next(vx_partition *partition, vx_error **err);
+
+// Return a (possibly exact) estimate of rows in a partition.
+void vx_partition_row_count(const vx_partition *partition, vx_estimate *count,
+    vx_error **err);
 ```
 
 ## Array introspection
@@ -300,7 +348,9 @@ The main question is how to transform outputs of iteration, `vx_array`, into
 something query engines can operate with. You need to execute the array
 iteratively till you recognize data and start exporting it. Thus API provides a
 way to scan partitions directly into ArrowArrayStream which should be good
-enough for most hosts.
+enough for most hosts. However, some hosts may want to work with vx_array
+directly as it provides types like ConstantArray which Arrow doesn't have, and
+thus convertion requires CPU-intensive work.
 
 ## Compatibility
 
@@ -311,19 +361,21 @@ eventually.
 
 - Dividing scan requests into Partitions for threads is taken from DataFusion's
   [partitioned streams](https://docs.rs/datafusion/latest/datafusion/physical_plan/enum.Partitioning.html).
-- Making Partitions produce Arrays without synchronization mimicks
+- Making Partitions produce Arrays without synchronization mimics
   [Arrow Stream](https://arrow.apache.org/docs/format/CStreamInterface.html)
   interface.
 
 ## Unresolved Questions
 
+- Should high level API support reading a file from a buffer and not from a
+  path? Is there a use case for that?
 - Should high level API have a way to use host's persistent caching options?
   In-memory caching may be implemented using host allocator only but for
   persistent caching we need additional filesystem customizations i.e. cache
   location path.
 - Should high level API expose batch reads from Partition? There are plans to
   deprecate Partitions on Rust side.
-- What introspecion should high level API have for hosts which aren't satisfied
+- What introspection should high level API have for hosts which aren't satisfied
   with `Vortex -> ArrowArrayStream` conversion? Should there be iterative
   execution API?
 - How should API expose definitions of `ArrowSchema` and `ArrowArrayStream`?
@@ -337,6 +389,9 @@ eventually.
   #include "vortex.h"
   ```
 
+  Current implementation copy-pastes ArrowSchema and ArrowArrayStream and gates
+  them behind a macro.
+
 ## High level API integration example: DuckDB
 
 ```
@@ -355,15 +410,15 @@ duckdb -> TableFunction vtable -> ffi_wrapping -> vx_scan()*
 * - vx_array -> DataChunk reuses existing Rust code
 ```
 
-https://github.com/vortex-data/vortex/tree/myrrc/scan-api-duckdb has an
-implementation of using C Scan API for Duckdb scan integration. Duckdb has a
-sync multi-threaded runtime, and table function is called from multiple threads
-simultaneously. Users can save a per-thread state.
+https://github.com/vortex-data/vortex/tree/myrrc/scan-api-duckdb has a PoC (not
+production-ready) implementation of using C Scan API for Duckdb scan
+integration. Duckdb has a sync multi-threaded runtime, and table function is
+called from multiple threads simultaneously. Users can save a per-thread state.
 
 The integration splits into following parts:
-- DType -> LogicalType integration, done sans Temporal extension.
+- DType -> LogicalType integration, done except for Temporal extension support.
 - Table function binding (creating a DataSource), done.
-- Global state initialization (creating a Scan), done sans filter pushdown.
+- Global state initialization (creating a Scan), done except filter pushdown.
 - Local state initialization (export batch id), done.
 - Utility functions like cardinality estimates, done.
 - vx_array -> DataChunk export, delegated to existing Rust code.
@@ -385,15 +440,16 @@ works on its own partition without synchronization.
 [^1]: Clients of this API would mostly be query engines like Velox or
     ClickHouse, but may as well be our own integrations like vortex-duckdb.
 [^2]: Like opening the first file in a glob expression to determine schema.
-[^3]: Exposed Rust ABI is not stable, so clients can't use cbindgen. C++ clients
-    can use cxx.rs but this still requires writing manual bindings and runtime
-    bridging.
+[^3]: Rust ABI is not stable if you don't use `repr(C)`, so clients can't use
+    cbindgen directly. C++ clients can use cxx.rs but this still requires
+    writing manual bindings and runtime bridging.
 [^4]: DuckDB MultiFileReader and MultiFileList.
 [^5]: The name may be misleading as it doesn't correspond to Rust side's
-    Partitions.
+    Partitions. Alternative considered was vx_split but it conflicts with
+    register_splits in LayoutReader.
 [^6]: DuckDB integration currently hides Partitions and Arrays behind a single
     [thread-safe iterator](https://github.com/vortex-data/vortex/blob/e8cd130c8ccac45082a0b458b1f843c4313555bf/vortex-duckdb/src/datasource.rs#L151)
-    which implies unnecessary intra-thread synchronization on pulling data.
-    On the other hand, the producer, an async crossbeam queue, allows smoothing
-    out uneven workloads from the Vortex side, and if that's removed, Vortex's
+    which implies unnecessary intra-thread synchronization on pulling data. On
+    the other hand, the producer, an async crossbeam queue, allows smoothing out
+    uneven workloads from the Vortex side, and if that's removed, Vortex's
     partition scheduling must be precise.