Skip to content

DataFusion SQL Layer

src/engine/sql.rs embeds Apache DataFusion. Both commands accept a single SQL string where the FROM clause references a file path or directory (single-quoted):

import basaltic_red as br

table  = br.sql.execute_sql("SELECT COUNT(*) FROM 'data/analytics'")       # collected Table
stream = br.sql.execute_sql_stream("SELECT * FROM 'data/analytics'")       # PyBatchIterator

Target Resolution

Before planning, the engine registers the FROM target as a table named br_target. Two paths exist:

  1. Native ListingTable, for extensions DataFusion reads itself:

    Extension DataFusion format
    parquet, pq ParquetFormat
    csv, tsv, psv CsvFormat (delimiter-aware)
    json, jsonl, ndjson JsonFormat (newline-delimited objects)
    arrow, ipc, feather ArrowFormat

    A directory of homogeneous extension registers as one ListingTable, allowing DataFusion to apply supported predicate and projection pruning across files. The resulting IO savings depend on the file format and query plan.

  2. Fallback MemTable, for formats without a native DataFusion reader (xlsx, avro, orc, msgpack, txt, mixed directories) and for JSON files whose top level is an array ([...]). The file is read through the format registry, collected into a MemTable, then queried in memory.

Native ListingTable readers use DataFusion's FileFormat directly and do not consult the dynamic FormatHandler registry. Registering a handler for a native extension such as csv therefore does not change how br.sql.* reads that extension.

Top-level JSON arrays

DataFusion's JsonFormat expects newline-delimited objects. Files starting with [ are detected by inspecting the first non-whitespace byte and routed to the streaming handler instead.

Streaming vs Collecting

Command Return Memory profile
execute_sql(query) PyArrow Table Fully collected
execute_sql_stream(query) PyBatchIterator Lazy, batches produced on demand

PyBatchIterator (src/pyapi/iterator.rs)

Two internal sources behind one class:

  • Lazy, a live SendableRecordBatchStream; each next() pulls one batch from the tokio runtime (memory::global_runtime()).
  • Eager, a pre-collected Vec<RecordBatch> (used when a MemTable fallback was required).

Python-facing API:

  • iteration: for batch in stream: yields PyArrow RecordBatches
  • stream.to_pyarrow() → complete PyArrow Table
  • repr(stream) → PyBatchIterator(batches=N, rows=M) (counts known eagerly; filled during consumption for lazy streams)

The Arrow C Data Interface can share compatible buffers when passing results to Polars or DuckDB. DataFusion still decodes input and a consumer may materialize or copy results; see Integrate with Polars & DuckDB.

What Gets Pushed Down

For Parquet ListingTables, DataFusion can prune row groups and avoid reading unselected column chunks when the query plan and file metadata allow it. Other native formats can use the projection or filter support provided by their reader, but do not necessarily skip physical input in the same way.

# Reads only passenger_count + fare_amount column chunks, prunes row groups by predicate
br.sql.execute_sql_stream(
    "SELECT passenger_count, AVG(fare_amount) "
    "FROM 'data/output/clean_trips.parquet' GROUP BY passenger_count"
)