Data: Parquet and SQL with DataFusion

DataFusion is a query engine you embed in your own program: you give it Parquet or CSV files, and it answers SQL. The crate docs call it “an extensible query engine written in Rust that uses Apache Arrow as its in-memory format”. Its FAQ says how its authors see the neighbours: DuckDB is mostly used directly as a database, while DataFusion is “designed to be embedded”; Polars is also written in Rust and uses the Arrow memory model, but has fewer extension points. That is the project describing itself, not a comparison I ran. Polars’ own docs say it uses the Arrow columnar format too, and the DataFusion introduction page lists InfluxDB among its known users.

RememberRegister a file as a table, then ask questions in SQL or with the DataFrame API.

Project layout

sales-report/
├── examples/                   two more programs, further down
│   ├── explain.rs
│   └── results.rs
├── src/
│   └── main.rs                 the program on this page
├── Cargo.lock                  created by Cargo
└── Cargo.toml                  dependencies go here

RustCargo.toml

[package]
name = "sales-report"
version = "0.1.0"
edition = "2024"

[dependencies]
datafusion = "55.2.0"
tokio = { version = "1.54.1", features = ["full"] }

Rustsrc/main.rs

use std::sync::Arc;

use datafusion::arrow::array::{Float64Array, Int64Array, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::parquet::arrow::ArrowWriter;
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    // 1. Build a small table in memory as Arrow columns, then save it as Parquet.
    let schema = Arc::new(Schema::new(vec![
        Field::new("city", DataType::Utf8, false),
        Field::new("month", DataType::Int64, false),
        Field::new("sales", DataType::Float64, false),
    ]));
    let batch = RecordBatch::try_new(
        schema.clone(),
        vec![
            Arc::new(StringArray::from(vec![
                "Pune", "Pune", "Delhi", "Delhi", "Chennai",
            ])),
            Arc::new(Int64Array::from(vec![1, 2, 1, 2, 1])),
            Arc::new(Float64Array::from(vec![120.0, 135.5, 210.0, 190.25, 99.0])),
        ],
    )?;
    let path = std::env::temp_dir().join("sales.parquet");
    let mut writer = ArrowWriter::try_new(std::fs::File::create(&path)?, schema, None)?;
    writer.write(&batch)?;
    writer.close()?;

    // 2. Register the file as a table and ask a question in SQL.
    let ctx = SessionContext::new();
    ctx.register_parquet(
        "sales",
        path.to_str().unwrap(),
        ParquetReadOptions::default(),
    )
    .await?;
    ctx.sql("SELECT city, SUM(sales) AS total FROM sales GROUP BY city ORDER BY total DESC")
        .await?
        .show()
        .await?;

    // 3. A similar question with the DataFrame API. Both queries are sorted, so the
    //    rows come out in the same order every time.
    ctx.table("sales")
        .await?
        .filter(col("month").eq(lit(2)))?
        .select(vec![col("city"), col("sales")])?
        .sort(vec![col("sales").sort(false, true)])?
        .show()
        .await?;
    Ok(())
}

$ cargo run

+---------+--------+
| city    | total  |
+---------+--------+
| Delhi   | 400.25 |
| Pune    | 255.5  |
| Chennai | 99.0   |
+---------+--------+
+-------+--------+
| city  | sales  |
+-------+--------+
| Delhi | 190.25 |
| Pune  | 135.5  |
+-------+--------+

Both queries end in a sort, so the rows come out in the same order on every run. This page is about DataFusion 55.2.0, as pinned in Cargo.lock.

Arrow is the in-memory format. The arrow-rs docs describe a RecordBatch as “a two-dimensional batch of column-oriented data with a defined schema”. The program builds one from three arrays.
Parquet is the file format. Its project page calls it “an open source, column-oriented data file format designed for efficient data storage and retrieval”. ArrowWriter wrote the file and register_parquet made it a table named sales.
Two ways to ask: a SQL string, or DataFrame calls (filter, select, sort). The DataFrame docs say most DataFrame methods only build up a plan (the steps DataFusion will take; What the plan says shows one), and that collect runs it “using the same DataFusion planning and execution process used to execute SQL”.
It is async, and runs on Tokio. That is why main is #[tokio::main] and the calls that do the work (register_parquet, sql, show) end in .await; filter, select and sort only build the plan, so they do not. The crate docs say DataFusion runs each plan on multiple cores “using a Tokio Runtime as a thread pool”, with the number of partitions (pieces of the data that can be processed side by side) set by target_partitions, which “defaults to the number of CPU cores”. For Tokio itself see the Tokio guide; for the idea of a future see Async Basics.
Use DataFusion’s copies of Arrow and Parquet (datafusion::arrow, datafusion::parquet), as the program does. The crate docs list both as re-exports. The section on adding your own arrow crate shows what goes wrong otherwise.
The first build is big. The lock file lists 286 packages, including this program:

terminal

$ cargo tree --depth 1
sales-report v0.1.0 (/home/you/sales-report)
├── datafusion v55.2.0
└── tokio v1.54.1
$ cargo tree -i arrow --depth 0
arrow v59.3.1
$ grep -c '^name = ' Cargo.lock
286

I did not time the build. After the examples on this page were built, the debug build folder for this project measured 6.8 GB with du on this VM (once, so read it as “several gigabytes”).

What the plan says

A plan is the list of steps the engine will run; the logical plan says what to compute, the physical plan says how, and the optimizer rewrites the first before the second is made (Reading explain plans describes the lines). Whether DataFusion skips the columns and row groups a query does not need is something you can check. examples/explain.rs writes the same five rows with two rows per row group, asks the same question in SQL and with DataFrame calls, and prints the plans. A row group is, in the words of the Parquet concepts page, “a logical horizontal partitioning of the data into rows”.

Rustexamples/explain.rs

use std::sync::Arc;

use datafusion::arrow::array::{Float64Array, Int64Array, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::parquet::arrow::ArrowWriter;
use datafusion::parquet::file::properties::WriterProperties;
use datafusion::physical_plan::{ExecutionPlan, collect, displayable};
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    // The same five rows as the main program, but written two rows per Parquet row group.
    let schema = Arc::new(Schema::new(vec![
        Field::new("city", DataType::Utf8, false),
        Field::new("month", DataType::Int64, false),
        Field::new("sales", DataType::Float64, false),
    ]));
    let batch = RecordBatch::try_new(
        schema.clone(),
        vec![
            Arc::new(StringArray::from(vec![
                "Pune", "Pune", "Delhi", "Delhi", "Chennai",
            ])),
            Arc::new(Int64Array::from(vec![1, 2, 1, 2, 1])),
            Arc::new(Float64Array::from(vec![120.0, 135.5, 210.0, 190.25, 99.0])),
        ],
    )?;
    let path = std::env::temp_dir().join("sales_groups.parquet");
    let props = WriterProperties::builder()
        .set_max_row_group_row_count(Some(2))
        .build();
    let mut writer = ArrowWriter::try_new(std::fs::File::create(&path)?, schema, Some(props))?;
    writer.write(&batch)?;
    writer.close()?;

    // One partition, so the plan does not depend on how many CPU cores this machine has
    // (the default is one partition per core).
    let ctx = SessionContext::new_with_config(SessionConfig::new().with_target_partitions(1));
    ctx.register_parquet(
        "sales",
        path.to_str().unwrap(),
        ParquetReadOptions::default(),
    )
    .await?;

    // The same question asked two ways.
    // `cargo run --example explain -- 100` changes the number (200 if you give none).
    let limit: f64 = std::env::args()
        .nth(1)
        .map_or(200.0, |a| a.parse().expect("a number"));
    let from_sql = ctx
        .sql(&format!(
            "SELECT city, sales FROM sales WHERE sales > {limit}"
        ))
        .await?;
    let from_api = ctx
        .table("sales")
        .await?
        .filter(col("sales").gt(lit(limit)))?
        .select(vec![col("city"), col("sales")])?;
    println!(
        "-- as written (SQL)\n{}",
        from_sql.logical_plan().display_indent()
    );
    println!(
        "-- as written (DataFrame)\n{}",
        from_api.logical_plan().display_indent()
    );

    // The optimizer rewrites both. The unoptimized plans differ (Int64(200) against
    // Float64(200)); the optimized ones are equal.
    let sql_plan = from_sql.clone().into_optimized_plan()?;
    let api_plan = from_api.into_optimized_plan()?;
    println!("-- optimized\n{}", sql_plan.display_indent());
    println!("optimized plans equal: {}", sql_plan == api_plan);

    // Run the physical plan, then ask it what the Parquet reader did.
    let physical = from_sql.create_physical_plan().await?;
    // The plan prints the temp folder of the machine that runs it; show a placeholder
    // instead, so the page does not depend on that path (the plan drops the leading `/`).
    let shown = displayable(physical.as_ref()).indent(false).to_string().replace(
        path.to_str().unwrap().trim_start_matches('/'),
        "<temp dir>/sales_groups.parquet",
    );
    println!("-- physical\n{shown}");
    let rows: usize = collect(physical.clone(), ctx.task_ctx())
        .await?
        .iter()
        .map(|b| b.num_rows())
        .sum();
    println!("rows returned: {rows}");
    print_metrics(physical.as_ref());
    Ok(())
}

fn print_metrics(node: &dyn ExecutionPlan) {
    if let Some(v) = node
        .metrics()
        .and_then(|set| set.sum_by_name("row_groups_pruned_statistics"))
    {
        println!("{}: row_groups_pruned_statistics = {v}", node.name());
    }
    for child in node.children() {
        print_metrics(child.as_ref());
    }
}

$ cargo run --example explain

-- as written (SQL)
Projection: sales.city, sales.sales
  Filter: sales.sales > Int64(200)
    TableScan: sales
-- as written (DataFrame)
Projection: sales.city, sales.sales
  Filter: sales.sales > Float64(200)
    TableScan: sales
-- optimized
Filter: sales.sales > Float64(200)
  TableScan: sales projection=[city, sales], partial_filters=[sales.sales > Float64(200)]
optimized plans equal: true
-- physical
FilterExec: sales@1 > 200
  DataSourceExec: file_groups={1 group: [[<temp dir>/sales_groups.parquet]]}, projection=[city, sales], file_type=parquet, predicate=sales@2 > 200, pruning_predicate=sales_null_count@1 != row_count@2 AND sales_max@0 > 200, required_guarantees=[]

rows returned: 1
DataSourceExec: row_groups_pruned_statistics = 3 total → 1 matched

$ cargo run --example explain -- 100 | tail -2

rows returned: 4
DataSourceExec: row_groups_pruned_statistics = 3 total → 2 matched -> 2 fully matched

What the run showed:

The two queries start as different plans (Int64(200) against Float64(200)), and after the optimizer rewrites them the plans are equal. That is for this query only; I did not test others.
TableScan says projection=[city, sales]: the month column is not in the scan’s projection.
The scan carries a pruning_predicate that compares sales_max, a statistic of the column, with 200, and the metric printed row_groups_pruned_statistics = 3 total → 1 matched: two of the three row groups were ruled out by their statistics.
The example sets with_target_partitions(1) so the plan does not depend on how many cores the machine has. When I ran it by hand without that line on this 8-core VM, the physical plan had an extra RepartitionExec step with 8 partitions (that run is not one of the sessions on this page).
The physical plan names the file by its temp folder, which differs per machine, so the example replaces that part with <temp dir> before printing.
The file is five rows, so this shows the mechanism and says nothing about speed. I measured no times.

Results as data, and mistakes

examples/results.rs reads a CSV file instead, with register_csv. A CSV file has no column types, so DataFusion inferred them from the data; the first lines of the output show what it chose. collect returns the answer as Arrow arrays you can read from code, and mistakes in SQL come back as Err values:

Rustexamples/results.rs

use datafusion::arrow::array::AsArray;
use datafusion::arrow::datatypes::Float64Type;
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    // A CSV file this time: DataFusion reads the header and guesses the column types.
    let path = std::env::temp_dir().join("sales.csv");
    std::fs::write(
        &path,
        "city,month,sales\nPune,1,120.0\nPune,2,135.5\nDelhi,1,210.0\nDelhi,2,190.25\nChennai,1,99.0\n",
    )?;

    let ctx = SessionContext::new();
    ctx.register_csv("sales", path.to_str().unwrap(), CsvReadOptions::new())
        .await?;

    for field in ctx.table("sales").await?.schema().fields() {
        println!("column {} has type {}", field.name(), field.data_type());
    }

    // collect() gives the result back as Arrow record batches, so you can use it in code.
    let batches = ctx
        .sql("SELECT city, SUM(sales) AS total FROM sales GROUP BY city ORDER BY total DESC")
        .await?
        .collect()
        .await?;
    for batch in &batches {
        let city = batch.column_by_name("city").unwrap().as_string::<i32>();
        let total = batch
            .column_by_name("total")
            .unwrap()
            .as_primitive::<Float64Type>();
        for i in 0..batch.num_rows() {
            println!("{} -> {:.2}", city.value(i), total.value(i));
        }
    }

    // Mistakes come back as errors, not panics.
    for sql in [
        "SELECT * FROM sale",
        "SELECT cty FROM sales",
        "SELEC city FROM sales",
    ] {
        match ctx.sql(sql).await {
            Ok(_) => println!("{sql}: ok"),
            Err(e) => println!("{sql}\n  Error: {e}"),
        }
    }
    Ok(())
}

$ cargo run --example results

column city has type Utf8
column month has type Int64
column sales has type Float64
Delhi -> 400.25
Pune -> 255.50
Chennai -> 99.00
SELECT * FROM sale
  Error: Error during planning: table 'datafusion.public.sale' not found
SELECT cty FROM sales
  Error: Schema error: No field named cty. Did you mean 'sales.city'?
Valid fields are sales.city, sales.month, sales.sales.
SELEC city FROM sales
  Error: SQL error: ParserError("Expected: an SQL statement, found: SELEC at Line: 1, Column: 1")

Adding your own `arrow` crate

cargo add arrow is a natural thing to type. On 25 October 2026 the newest release on crates.io was 60.0.0, while DataFusion 55.2.0 depends on arrow ^59.2.0 (docs.rs dependency list) and the lock file has 59.3.1 (the tree output above prints arrow v59.3.1). Those are Arrow’s version numbers, not DataFusion’s: DataFusion is 55.2.0. Two versions of a crate are two different crates to the compiler, so a RecordBatch from one is not a RecordBatch from the other. The session below builds a shorter program than the one above, not the main program: it has arrow = "=60.0.0" added to Cargo.toml, takes its imports from arrow instead of datafusion::arrow, and hands one column to register_batch:

Rust

[package]
name = "sales-report"
version = "0.1.0"
edition = "2024"

[dependencies]
datafusion = "55.2.0"
tokio = { version = "1.54.1", features = ["full"] }
arrow = "=60.0.0"

use std::sync::Arc;

use arrow::array::Int64Array;
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use datafusion::prelude::*;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    let schema = Arc::new(Schema::new(vec![Field::new("month", DataType::Int64, false)]));
    let batch = RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(vec![1, 2]))]).unwrap();

    let ctx = SessionContext::new();
    ctx.register_batch("months", batch)?;
    ctx.sql("SELECT month FROM months").await?.show().await?;
    Ok(())
}

$ cargo build

error[E0308]: mismatched types
   --> src/main.rs:14:34
    |
 14 |     ctx.register_batch("months", batch)?;
    |         --------------           ^^^^^ expected `datafusion::arrow::array::RecordBatch`, found `arrow::array::RecordBatch`
    |         |
    |         arguments to this method are incorrect
  ... (notes pointing into library source code trimmed)

When I ran cargo build by hand, the full message also had a note saying “there are multiple different versions of crate arrow_array in the dependency graph”. The page trims notes that point into library sources. The fix is the one the program already uses: import from datafusion::arrow, or choose the arrow version DataFusion uses (cargo tree -i arrow --depth 0 above shows it).

The same job in your language

Only Python ran for this page. The other rows only name tools, and I did not run them. A name here is not a recommendation or a popularity claim, and I did not check that each tool does both Parquet and SQL: the notes say only what the linked page says.

Language	Tools named, and what the linked page says	Ran here
Go	DuckDB’s Go client (uses cgo and needs a C compiler, per DuckDB’s docs); arrow-go, the Go implementation of Arrow, which also has Parquet	no
Python	pandas, Polars, DuckDB, PyArrow, DataFusion’s own Python bindings	yes, below
TypeScript	DuckDB-Wasm and DuckDB’s Node.js client, both in DuckDB’s client list; Apache Arrow JS, for Arrow data (its page does not mention Parquet or SQL)	no
Java	DuckDB’s JDBC driver, in DuckDB’s client list; Apache Spark (named only)	no
Kotlin	Kotlin DataFrame (its page lists SQL databases as a data source and does not mention Parquet)	no
C#	DuckDB.NET, a .NET client listed in DuckDB’s client overview; Microsoft.Data.Analysis, a DataFrame type (its page mentions neither Parquet nor SQL)	no
C	the DuckDB C API	no
C++	Arrow C++ (with its Acero engine); DuckDB’s C++ API	no

The runner image has Python but no pip, so setup.sh downloads pip and installs the packages from requirements.txt into a scratch folder. This session needs internet access and cannot be reproduced offline: with the network switched off, setup.sh failed at its first step (curl could not resolve bootstrap.pypa.io). Only the five packages in requirements.txt are pinned; get-pip.py is fetched unpinned and run as downloaded, and the packages’ own dependencies are not pinned either, so a later run may install different versions (the script prints the five it got). On your own machine, a virtual environment and pip install -r requirements.txt do the same job. compare.py writes the same five rows to Parquet and asks each library for the total per city:

Python

import duckdb
import datafusion
import pandas as pd
import polars as pl
import pyarrow as pa
import pyarrow.parquet as pq

# The same five rows as the Rust program, written to a Parquet file by PyArrow.
pq.write_table(
    pa.table({
        "city": ["Pune", "Pune", "Delhi", "Delhi", "Chennai"],
        "month": [1, 2, 1, 2, 1],
        "sales": [120.0, 135.5, 210.0, 190.25, 99.0],
    }),
    "sales.parquet",
)

# Total sales per city, biggest first, with each library.
results = {}

pdf = pd.read_parquet("sales.parquet").groupby("city")["sales"].sum().sort_values(ascending=False)
results["pandas"] = list(pdf.items())

plf = pl.read_parquet("sales.parquet").group_by("city").agg(pl.col("sales").sum().alias("total")).sort("total", descending=True)
results["polars"] = plf.rows()

results["duckdb"] = duckdb.sql(
    "SELECT city, SUM(sales) AS total FROM 'sales.parquet' GROUP BY city ORDER BY total DESC"
).fetchall()

ctx = datafusion.SessionContext()
ctx.register_parquet("sales", "sales.parquet")
batches = ctx.sql("SELECT city, SUM(sales) AS total FROM sales GROUP BY city ORDER BY total DESC").collect()
results["datafusion"] = [row for b in batches for row in zip(b.column("city").to_pylist(), b.column("total").to_pylist())]

for name, rows in results.items():
    print(f"{name:<10}", [(c, float(t)) for c, t in rows])
print("all agree:", len({str([(c, float(t)) for c, t in r]) for r in results.values()}) == 1)
print("versions:", ", ".join(f"{m.__name__} {m.__version__}" for m in (pd, pl, duckdb, pa, datafusion)))

#!/bin/bash
# The runner image has Python but no pip, so fetch pip, then install the pinned
# packages into a scratch folder (nothing is installed system-wide).
set -e
cd "$(dirname "$0")"
{
  curl -sS https://bootstrap.pypa.io/get-pip.py -o /tmp/get-pip.py
  python3 /tmp/get-pip.py -q --break-system-packages --target /tmp/pylibs
  PYTHONPATH=/tmp/pylibs python3 -m pip install -q --target /tmp/pylibs -r requirements.txt
} > /tmp/pip.log 2>&1 || { cat /tmp/pip.log; exit 1; }
echo installed

$ bash setup.sh # needs network: fetches pip, then installs requirements.txt (the image has no pip)

installed

$ python3 compare.py

pandas     [('Delhi', 400.25), ('Pune', 255.5), ('Chennai', 99.0)]
polars     [('Delhi', 400.25), ('Pune', 255.5), ('Chennai', 99.0)]
duckdb     [('Delhi', 400.25), ('Pune', 255.5), ('Chennai', 99.0)]
datafusion [('Delhi', 400.25), ('Pune', 255.5), ('Chennai', 99.0)]
all agree: True
versions: pandas 3.1.0, polars 2.0.0, duckdb 2.0.0, pyarrow 26.0.0, datafusion 55.0.0

All four gave the same three rows. The Python datafusion package is DataFusion’s own engine behind a Python API; its version number (55.0.0) matches the Rust crate’s major version here, but I did not check that the two builds contain the same code. Nothing in this run measures speed.

Try it yourself

Change the filter. cargo run --example explain -- 100 uses sales > 100. The second command in the plan panel above ran it (with tail -2 to keep only the end): 4 rows come back, and the Parquet metric line now reads 3 total → 2 matched -> 2 fully matched (the line also reports how many row groups matched fully), because the third row group’s largest value is 99.
Make three mistakes in SQL. The results session sent a misspelled table, a misspelled column and a misspelled keyword. The column error suggested sales.city; the keyword error is a ParserError. All three were Err values and the program kept going.

Add arrow to the main program. Add arrow = "=60.0.0" to Cargo.toml and change the three use datafusion::arrow::... lines of src/main.rs to use arrow::.... This is the whole program, not the shorter one above, so it does not give the single error shown there: Cargo reported three errors, two mismatched types (E0308, where the schema goes to ArrowWriter::try_new and where &batch goes to writer.write) and one E0277, because ? cannot turn the new arrow::error::ArrowError into a DataFusionError: