Struct OxiSqlStreamProvider 

pub struct OxiSqlStreamProvider { /* private fields */ }

A DataFusion TableProvider that executes a live SQL query at scan time, pushing filters and projections down to the OxiSQL backend.

Filter pushdown

Simple comparison predicates (=, <>, <, <=, >, >=, AND, OR, IS NULL, IS NOT NULL, NOT) are converted to SQL WHERE clauses. Complex expressions (functions, subqueries, arithmetic) are left to DataFusion’s post-scan filtering.

Projection pushdown

When DataFusion requests specific columns, the provider generates SELECT col1, col2, ... instead of SELECT *.

Limit pushdown

When DataFusion specifies a limit, the provider appends LIMIT N to the query, reducing the number of rows transferred.

Sort pushdown

An optional ORDER BY clause can be injected into the generated SQL via Self::with_sort. Each entry is a (column_name, SortOrder) pair.

Automatic partitioning

Self::with_auto_partition splits the scan into multiple parallel partitions using LIMIT / OFFSET pagination, each fetching at most target_batch_size rows. At most n_parallel partitions are created.

Implementations

impl OxiSqlStreamProvider

pub fn new( conn: Arc<dyn Connection>, table_name: impl Into<String>, schema: SchemaRef, ) -> Self

Construct from a connection, table name, and Arrow schema.

The schema must match the column layout returned by the backend for SELECT * FROM table_name. Column names in the schema are used verbatim as SQL column references.

pub fn with_sort(self, order: Vec<(String, SortOrder)>) -> Self

Attach an ORDER BY clause to the SQL generated at scan time.

Each entry is a (column_name, SortOrder) pair. The resulting SQL fragment is appended between the WHERE clause and any LIMIT clause.

This is a “sort-into-SQL” approach: the backend database performs the ordering, reducing the work DataFusion must do on the result set.

Example

use oxisql_datafusion::stream::{OxiSqlStreamProvider, SortOrder};

let provider = OxiSqlStreamProvider::new(conn, "users", schema)
    .with_sort(vec![
        ("score".into(), SortOrder::Desc),
        ("id".into(), SortOrder::Asc),
    ]);

pub fn sort_order(&self) -> Option<&[(String, SortOrder)]>

Return the configured sort order, if any.

Returns None if no sort has been attached via Self::with_sort.

pub fn with_auto_partition( self, n_parallel: usize, target_batch_size: usize, ) -> Self

Configure automatic partitioning for parallel scan execution.

When set, scan() creates up to n_parallel partition slots, each issuing LIMIT target_batch_size OFFSET i * target_batch_size to the backend. This allows DataFusion to execute multiple partitions in parallel without knowing the total row count ahead of time.

Partitions beyond the actual data silently return empty batches, so setting n_parallel generously is safe.

• n_parallel: maximum number of parallel partitions (e.g. CPU thread count).
• target_batch_size: rows fetched per partition page (e.g. 8192).

Example

let provider = OxiSqlStreamProvider::new(conn, "users", schema)
    .with_auto_partition(4, 8192);

pub fn auto_partition_config(&self) -> Option<(usize, usize)>

Return the auto-partition configuration (n_parallel, target_batch_size), if any.

Trait Implementations

impl Debug for OxiSqlStreamProvider

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl TableProvider for OxiSqlStreamProvider

fn schema(&self) -> SchemaRef

Get a reference to the schema for this table

fn table_type(&self) -> TableType

Get the type of this table for metadata/catalog purposes.

fn scan<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _state: &'life1 dyn Session, projection: Option<&'life2 Vec<usize>>, filters: &'life3 [Expr], limit: Option<usize>, ) -> Pin<Box<dyn Future<Output = DFResult<Arc<dyn ExecutionPlan>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Create an ExecutionPlan for scanning the table with optional projection, filter, and limit, described below.

fn supports_filters_pushdown( &self, filters: &[&Expr], ) -> DFResult<Vec<TableProviderFilterPushDown>>

Specify if DataFusion should provide filter expressions to the TableProvider to apply during the scan.

fn constraints(&self) -> Option<&Constraints>

Get a reference to the constraints of the table. Returns:

fn get_table_definition(&self) -> Option<&str>

Get the create statement used to create this table, if available.

fn get_logical_plan(&self) -> Option<Cow<'_, LogicalPlan>>

Get the LogicalPlan of this table, if available.

fn get_column_default(&self, _column: &str) -> Option<&Expr>

Get the default value for a column, if available.

fn scan_with_args<'a, 'life0, 'life1, 'async_trait>( &'life0 self, state: &'life1 dyn Session, args: ScanArgs<'a>, ) -> Pin<Box<dyn Future<Output = Result<ScanResult, DataFusionError>> + Send + 'async_trait>>

where 'a: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Create an ExecutionPlan for scanning the table using structured arguments.

fn statistics(&self) -> Option<Statistics>

Get statistics for this table, if available Although not presently used in mainline DataFusion, this allows implementation specific behavior for downstream repositories, in conjunction with specialized optimizer rules to perform operations such as re-ordering of joins.

fn insert_into<'life0, 'life1, 'async_trait>( &'life0 self, _state: &'life1 dyn Session, _input: Arc<dyn ExecutionPlan>, _insert_op: InsertOp, ) -> Pin<Box<dyn Future<Output = Result<Arc<dyn ExecutionPlan>, DataFusionError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Return an ExecutionPlan to insert data into this table, if supported.

fn delete_from<'life0, 'life1, 'async_trait>( &'life0 self, _state: &'life1 dyn Session, _filters: Vec<Expr>, ) -> Pin<Box<dyn Future<Output = Result<Arc<dyn ExecutionPlan>, DataFusionError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Delete rows matching the filter predicates.

fn update<'life0, 'life1, 'async_trait>( &'life0 self, _state: &'life1 dyn Session, _assignments: Vec<(String, Expr)>, _filters: Vec<Expr>, ) -> Pin<Box<dyn Future<Output = Result<Arc<dyn ExecutionPlan>, DataFusionError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Update rows matching the filter predicates.

fn truncate<'life0, 'life1, 'async_trait>( &'life0 self, _state: &'life1 dyn Session, ) -> Pin<Box<dyn Future<Output = Result<Arc<dyn ExecutionPlan>, DataFusionError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Remove all rows from the table.