high level scan api draft

Author: myrrc

Cargo.lock

@@ -31,6 +31,7 @@ tracing = { workspace = true, features = ["std", "log"] }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 url = { workspace = true, features = [] }
 vortex = { workspace = true, features = ["object_store"] }
+arrow-array = { workspace = true, features = ["ffi"] }
 
 [dev-dependencies]
 tempfile = { workspace = true }

vortex-ffi/Cargo.toml

@@ -18,6 +18,12 @@
  */
 #define BinaryView_MAX_INLINED_SIZE 12
 
+typedef enum {
+    VX_CARD_UNKNOWN = 0,
+    VX_CARD_ESTIMATE = 1,
+    VX_CARD_MAXIMUM = 2,
+} vx_cardinality;
+
 /**
  * Variant enum for Vortex primitive types.
  */
@@ -204,6 +210,18 @@ typedef enum {
     LOG_LEVEL_TRACE = 5,
 } vx_log_level;
 
+typedef enum {
+    VX_S_INCLUDE_ALL = 0,
+    VX_S_INCLUDE_RANGE = 1,
+    VX_S_EXCLUDE_RANGE = 2,
+} vx_scan_selection_include;
+
+typedef enum {
+    VX_ESTIMATE_UNKNOWN = 0,
+    VX_ESTIMATE_EXACT = 1,
+    VX_ESTIMATE_INEXACT = 2,
+} vx_estimate_boundary;
+
 /**
  * Physical type enum, represents the in-memory physical layout but might represent a different logical type.
  */
@@ -295,6 +313,8 @@ typedef struct Nullability Nullability;
 
 typedef struct Primitive Primitive;
 
+typedef struct VxFileHandle VxFileHandle;
+
 /**
  * Base type for all Vortex arrays.
  *
@@ -344,6 +364,14 @@ typedef struct vx_array_sink vx_array_sink;
  */
 typedef struct vx_binary vx_binary;
 
+/**
+ * A data source is a reference to multiple possibly remote files. When
+ * created, it opens first file to determine the schema from DType, all
+ * other operations are deferred till a scan is requested. You can request
+ * multiple file scans from a data source
+ */
+typedef struct vx_data_source vx_data_source;
+
 /**
  * A Vortex data type.
  *
@@ -378,6 +406,14 @@ typedef struct vx_expression vx_expression;
  */
 typedef struct vx_file vx_file;
 
+/**
+ * A Partition is a unit of work for a worker thread from which you can
+ * get vx_arrays.
+ */
+typedef struct vx_partition vx_partition;
+
+typedef struct vx_scan vx_scan;
+
 /**
  * A handle to a Vortex session.
  */
@@ -398,6 +434,67 @@ typedef struct vx_struct_fields vx_struct_fields;
  */
 typedef struct vx_struct_fields_builder vx_struct_fields_builder;
 
+typedef int (*vx_fs_use_vortex)(const char *schema, const char *path);
+
+typedef void (*vx_fs_set_userdata)(void *userdata);
+
+typedef void (*vx_fs_open)(void *userdata, const char *path, vx_error **err);
+
+typedef void (*vx_fs_create)(void *userdata, const char *path, vx_error **err);
+
+typedef void (*vx_list_callback)(void *userdata, const char *name, int is_dir);
+
+typedef void (*vx_fs_list)(const void *userdata,
+                           const char *path,
+                           vx_list_callback callback,
+                           vx_error **error);
+
+typedef const VxFileHandle *vx_file_handle;
+
+typedef void (*vx_fs_close)(vx_file_handle handle);
+
+typedef uint64_t (*vx_fs_size)(vx_file_handle handle, vx_error **err);
+
+typedef uint64_t (
+    *vx_fs_read)(vx_file_handle handle, uint64_t offset, size_t len, uint8_t *buffer, vx_error **err);
+
+typedef uint64_t (
+    *vx_fs_write)(vx_file_handle handle, uint64_t offset, size_t len, uint8_t *buffer, vx_error **err);
+
+typedef void (*vx_fs_sync)(vx_file_handle handle, vx_error **err);
+
+typedef void (*vx_glob_callback)(void *userdata, const char *file);
+
+typedef void (*vx_glob)(const char *glob, vx_glob_callback callback, vx_error **err);
+
+/**
+ * Host must either implement all or none of fs_* callbacks.
+ */
+typedef struct {
+    const char *files;
+    /**
+     * Whether to use Vortex filesystem or host's filesystem.
+     * Return 1 to use Vortex for a given schema ("file", "s3") and path.
+     * Return 0 to use host's filesystem.
+     */
+    vx_fs_use_vortex fs_use_vortex;
+    vx_fs_set_userdata fs_set_userdata;
+    vx_fs_open fs_open;
+    vx_fs_create fs_create;
+    vx_fs_list fs_list;
+    vx_fs_close fs_close;
+    vx_fs_size fs_size;
+    vx_fs_read fs_read;
+    vx_fs_write fs_write;
+    vx_fs_sync fs_sync;
+    vx_glob glob;
+} vx_data_source_options;
+
+typedef struct {
+    vx_cardinality cardinality;
+    uint64_t rows;
+} vx_data_source_row_count;
+
 /**
  * Options supplied for opening a file.
  */
@@ -460,6 +557,28 @@ typedef struct {
     unsigned long row_offset;
 } vx_file_scan_options;
 
+typedef struct {
+    uint64_t *idx;
+    size_t idx_len;
+    vx_scan_selection_include include;
+} vx_scan_selection;
+
+typedef struct {
+    const vx_expression *projection;
+    const vx_expression *filter;
+    uint64_t row_range_begin;
+    uint64_t row_range_end;
+    vx_scan_selection selection;
+    uint64_t limit;
+    uint64_t max_threads;
+    int ordered;
+} vx_scan_options;
+
+typedef struct {
+    uint64_t estimate;
+    vx_estimate_boundary type;
+} vx_estimate;
+
 #ifdef __cplusplus
 extern "C" {
 #endif // __cplusplus
@@ -604,6 +723,34 @@ size_t vx_binary_len(const vx_binary *ptr);
  */
 const char *vx_binary_ptr(const vx_binary *ptr);
 
+/**
+ * Clone a borrowed [`vx_data_source`], returning an owned [`vx_data_source`].
+ *
+ *
+ * Must be released with [`vx_data_source_free`].
+ */
+const vx_data_source *vx_data_source_clone(const vx_data_source *ptr);
+
+/**
+ * Free an owned [`vx_data_source`] object.
+ */
+void vx_data_source_free(const vx_data_source *ptr);
+
+/**
+ * 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 **err);
+
+/**
+ * Create a non-owned dtype referencing dataframe.
+ * This dtype's lifetime is bound to underlying data source.
+ * Caller should not free this dtype.
+ */
+const vx_dtype *vx_data_source_dtype(const vx_data_source *ds);
+
+void vx_data_source_get_row_count(const vx_data_source *ds, vx_data_source_row_count *rc);
+
 /**
  * Clone a borrowed [`vx_dtype`], returning an owned [`vx_dtype`].
  *
@@ -747,6 +894,8 @@ uint8_t vx_dtype_time_unit(const DType *dtype);
  */
 const vx_string *vx_dtype_time_zone(const DType *dtype);
 
+void vx_type_to_arrow_schema(const vx_dtype *_dtype, FFI_ArrowSchema *_schema, vx_error **_err);
+
 /**
  * Free an owned [`vx_error`] object.
  */
@@ -927,6 +1076,47 @@ vx_array_iterator *vx_file_scan(const vx_session *session,
  */
 void vx_set_log_level(vx_log_level level);
 
+/**
+ * Free an owned [`vx_scan`] object.
+ */
+void vx_scan_free(vx_scan *ptr);
+
+/**
+ * Free an owned [`vx_partition`] object.
+ */
+void vx_partition_free(vx_partition *ptr);
+
+vx_scan *vx_data_source_scan(const vx_data_source *data_source,
+                             const vx_scan_options *options,
+                             vx_estimate *estimate,
+                             vx_error **err);
+
+/**
+ * Get next owned partition out of a scan request.
+ * Caller must free this partition using vx_partition_free.
+ * This method is thread-safe.
+ * If using in a sync multi-thread runtime, users 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.
+ */
+vx_partition *vx_scan_next(vx_scan *scan, vx_error **err);
+
+void vx_partition_row_count(const vx_partition *partition, vx_estimate *count, vx_error **err);
+
+void vx_partition_scan_arrow(const vx_partition *_partition, FFI_ArrowArrayStream *_stream, vx_error **err);
+
+/**
+ * Get next vx_array out of this partition.
+ * Thread-unsafe.
+ */
+const vx_array *vx_partition_next(vx_partition *partition, vx_error **err);
+
+/**
+ * Scan progress between 0.0 and 1.0
+ */
+double vx_scan_progress(const vx_scan *_scan);
+
 /**
  * Free an owned [`vx_session`] object.
  */