Rust ate Spark's lunch, Really?

Vitthal Mirji

Software Engineer · Data platforms · Stuttgart, Germany

vitthalmirji.com

vim89

Choose the engine and the work you will own

Both plan and execute queries. Choose by workload and operating needs.

Where does the time go between the disk I/O?

SELECT region, SUM(amount) AS revenue
FROM orders
WHERE order_date = DATE '2026-10-01'
GROUP BY region;

Orders partitioned by order_date · illustrative query

Parquet columns still need decoding

// This fixture stores amount at leaf index 1.
let projection = ProjectionMask::leaves(
    builder.parquet_schema(), [1]
);
let reader = builder
    .with_projection(projection)
    .build()?;
for decoded_batch in reader {
    let batch: RecordBatch = decoded_batch?;
}

Mechanism only; not a Parquet-vs-IPC speedup.

Different layouts add handoff work

// The next operator needs batches
if (!plan.supportsColumnar)
  RowToColumnarExec(
    ensureOutputsRowBased(plan)
  )

// The next operator needs rows
if (plan.supportsColumnar &&
    !plan.supportsRowBased)
  ColumnarToRowExec(
    ensureOutputsColumnar(plan)
  )

Same answer. Different work to reach it.

SELECT COUNT(*), SUM(amount)
FROM orders
WHERE amount > 50.00;

Schematic paths. Operators may fuse or mix row and batch work. With GROUP BY/HAVING, the batch path groups once per batch instead of updating one global total — mechanism on slide 180.

Python submits a plan. A UDF moves data.

df.select(
    (col("amount") + 1).alias("next")
)

@udf("long", useArrow=False)
def bump(x): return x + 1
df.select(bump("amount"))

@udf("long", useArrow=True)
def bump_arrow(x): return x + 1
df.select(bump_arrow("amount"))

A Python DataFrame API call is not a per-row Python round trip.

One row-only step can need two conversions

SELECT amount + 1 AS next_amount
FROM orders
WHERE amount > 50;

for (row <- rows) {
  val amount = row.getLong(0)
  if (amount > 50L)
    emit(amount + 1L)
}

Loop illustrates row processing, not generated Spark code. Bridges depend on the actual plan.

Keep the coordination your job depends on

Photon (Databricks, C++) and Comet (DataFusion, Rust) accelerate supported execution paths.

Keep Spark’s reach. Change the hot path.

Keep Spark. Accelerate supported work.

Spark plans; the executor hosts the native library

The CPU likes a boring loop

WHERE amount > 50.00
  AND note IS NOT NULL

let big  = gt(&amount, &fifty)?;
let kept = is_not_null(&note)?;
let mask = and(&big, &kept)?;

Arrow chooses the layout; Rust checks ownership. Python/JVM can use Arrow too. No promised 4×/8× speedup.

Four instructions. Eight additions. No miracle.

let incremented =
    Float64Array::from_iter_values(
        leaf.values().iter()
            .map(|value| value + 1.0)
    );

fadd.2d  v0, v0, v4
fadd.2d  v1, v1, v4
fadd.2d  v2, v2, v4
fadd.2d  v3, v3, v4

inspected binary Not a cross-platform speedup.

Packing took 414 ms. The Rust phase took 25 ms.

orders.zipWithIndex.foreach {
  case (order, i) =>
    valueColumn.set(i, order.value)
}
// later, in writeIpc:
writer.writeBatch()

measured 500k objects: 414 ms pack · 25 ms Rust subprocess.

50 columns: 15.30 ms versus 217.98 ms

measured M2 Max · synthetic, in-process · flat: mean of 5; nested: median of 3.

DataFusion builds rows too — on purpose

// Configured for the key fields.
let keys = converter.convert_columns(&columns)?;

let ordering = keys.row(left)
    .cmp(&keys.row(right));

Spark bridge: feed a row-based operator. Here: encode comparison keys.

Avoid the row detour: 3-5x here

5M input rows

Native path skips Spark rows. Parquet still needs decoding and encoding.

Our plain Parquet: Spark 3.5.5 / Comet 0.16 · local[2] · 6 measured pairs · experimental writer

Share batch owners. Know when bytes are reused.

let batches = self.batches.clone();
let schema = Arc::clone(&self.schema);

MemoryStream::try_new(
    batches, schema,
    self.projection.clone(),
)?

let (batch, reused) = unsafe {
    mmap_decode_single_i32_column(
        &path
    )
};

Shared ownership is not zero-copy I/O. Parquet still decodes and encodes.

Keep enough state to combine partial answers

for min in mins.iter().flatten() {
    self.min = Some(self.min.map_or(
        min, |current| current.min(min)));
}
for max in maxes.iter().flatten() {
    self.max = Some(self.max.map_or(
        max, |current| current.max(max)));
}

[4, 9] + [20, 21] → min 4, max 21 → range 17.

Repository UDAF example; the timed query uses built-in COUNT/SUM.

Out of memory budget? The query can fail.

Standalone DataFusion experiment · Spark users: think memory-related task failure

A memory budget alone does not guarantee completion. Execution strategy matters.

Spark: memory failure can trigger task retries or executor loss; repeated failures can abort the job.

The pool limits tracked allocations. It does not cap all process memory.

Steep start. Earlier feedback.

Familiar: SQL and schema reasoning.

My experience: slower at first; less debugging later.

Count learning, builds and debugging together.

Compiles ≠ correct business logic. Rust ≠ automatic speed.

Reject bad handoffs before they travel further

Compile-time using Macros : Rust → DataFusion · Scala → Catalyst

Bad declaration or selected SQL: build stops. Bad batch: stop, or quarantine by policy.

Change id to text; the build stops.

- name: id
  data_type: varchar
  tests: [not_null]

const _: () = SchemaConforms::<
    DbtSourceOfTruth,
    StewardInputModel,
    ExactOrdered,
>::CHECK;

Model mismatch: build fails. Bad batch: stop or quarantine.

Catch the stale accessor before query planning

#[derive(Contract, Connector)]
struct Order {
    id: i64,
    amount: i32,
}

Order::col_quantity()

Reject incompatible declared connections at build

ContractPipeline::source::<Bronze, _>(
    read
)
    .transform::<Silver, Gold>(enrich)
    .sink::<Gold>(write)
    .build()?
    .run()?;

SQL is valid. The handoff is not.

Rust macro · DataFusion planner

checked_sql! {
    output: Planned,
    input: Raw,
    required: Gold,
    source: "snapshot.json",
    sql: "SELECT * FROM raw",
}

#[derive(Contract)]
struct Gold {
    id: i64,
    total: i64,
}

SELECT id, amount AS total FROM raw

Compile time: captured schema + SQL output. Runtime: actual batches + values.

Same idea. Use the engine that runs it.

Two macros · different guarantees

Catalyst checks SQL. The Rust prototype also checks the sink contract.

Bad SQL? Let the build take the hit.

Scala 3 macro · Spark Catalyst · existing Parquet

val orders = table("""
  CREATE TABLE orders (id BIGINT,
    amount DECIMAL(10,2),
    placed_at TIMESTAMP_NTZ, note STRING)
""")
val domain = catalog(orders)
val checked = domain.sql("""
  SELECT COUNT(*) AS order_count,
         SUM(amount) AS total_amount
  FROM orders WHERE amount > 50.00
""")
spark.sql(checked)

'Project ['missing_column]
+- RelationV2[id] orders

Declared catalog only · Spark 3.5.5 / Comet 0.16 · correctness check, not a speed result.

Clean is a type. Not a pinky promise.

Tested prototype · separate from ContractPipeline

fn clean(raw: RawBatch)
    -> Result<CleanBatch, DemoError>
{
    Ok(raw)
}

fn clean(raw: RawBatch)
    -> Result<CleanBatch, DemoError>
{
    raw.clean()
}

Compile time: declared schemas and return types.

Runtime: actual batch schemas and data quality.

Wrong schema? Stop before Parquet decoding.

Tested local-file prototype · footer check runs at runtime

let source =
    ParquetSource::open(input, batch_size)?;

for raw in source {
    let clean = raw?.clean()?;
    sink(clean)?;
}

Schema mismatch: no row decoding. Data quality failure: stop before sink.

Starting code still needs a safe multi-batch sink

The generated example collects eagerly and overwrites the same sink per batch.

Three engines matched the independent answer

SELECT COUNT(*) AS order_count,
       SUM(amount) AS total_amount
FROM orders WHERE amount > 50.00;

correctness check 5M synthetic rows · one Parquet file · local, 21 Sep

Spark = Spark+Comet = DataFusion

2,500,000 orders · total_amount 188,750,000.00

Same Spark job: 45.25 s → 13.94 s

(n) = runs. Spark/Comet include startup; DataFusion is query-only; Ballista is client elapsed.

Spark/Comet: 7.748 GiB VM. Rust: 64 GiB host. Local M2 Max, not a cloud-bill measurement.

14 GB: too close. 91 GB: DataFusion ahead.

measured Same query · local M2 Max runs.

818 → 12,268 lookups/s, with specialized rebinding

// Reuse scan config + cached Parquet metadata.
let decoder_pushdown = false;
let plan = rebind(
    file_scan, decoder_pushdown, order_id
)?;
let mut output =
    execute_stream(plan, ctx.task_ctx())?;

5M rows · 1 warm local immutable file · 1 client · M2 Max · 5 rounds × 128 holdout keys/path

Complete lookup; setup excluded. 15× includes specialized physical rebinding, LZ4 layout and mimalloc.

Move one job. Keep a way back.

Query implementations, Process, patterns and Production migration steps.

My chart cannot decide your migration

Does it hold on your cloud and cluster?

How does it behave under sustained load?

What does the complete pipeline cost to run?

Count engineering and recovery in the decision

Files and other processes still need runtime checks.