StreamFusion

Run Apache Flink SQL faster by executing supported operators natively (Rust + Apache Arrow/DataFusion over JNI) while Flink continues to own planning, coordination, and everything not yet supported. Substitution is transparent and conservative: a query is planned by Flink, the operators we can reproduce exactly are swapped for native ones, and anything else falls back to Flink with identical results.

It is DataFusion Comet's idea — a native, columnar accelerator behind an unchanged SQL front end — applied to streaming instead of batch: stateful windowing, joins, aggregations, changelog processing, and columnar sources/sinks, not just stateless projection and filter.

What it accelerates

A query accelerates only when it forms one fully-columnar island: every operator except a rowwise source/sink runs natively, exchanging Arrow batches (the row↔Arrow transpose is paid once at the host edges, never between native operators). A single unsupported interior operator drags the whole query back to Flink.

Native coverage is broad — most of the streaming SQL surface:

• Stateless: projection/Calc, filter, UNION ALL, GROUPING SETS/CUBE/ROLLUP, UNNEST.
• Windowed aggregates: TUMBLE/HOP/SESSION/CUMULATE (event-time and proctime, one- and two-phase), and OVER window functions.
• Joins: regular (updating) equi-joins, event-time/proctime interval and window joins, event-time temporal-table joins, and processing-time lookup joins (sync and async).
• Changelog: non-windowed GROUP BY, streaming Top-N / LIMIT, deduplication, changelog normalization — all consuming and emitting a retract changelog.
• Connectors: Parquet/ORC file sources and a Parquet sink (native Arrow scan/write); Kafka source decode for JSON/CSV/raw/Avro/protobuf and Debezium/OGG CDC — JSON/Avro/protobuf via a fully native rdkafka source (the default path; it also regenerates the table's watermark per partition, exactly as Flink's source does).
• UDFs: a Flink ScalarFunction the expression engine can't implement itself is invoked over Arrow columns by a native→JVM upcall (Comet's JvmScalarUdfExpr pattern), one JNI crossing per batch, so the pipeline stays native through the UDF and the result is byte-identical.

Determinism. Results are byte-identical to stock Flink for everything admitted. The one caveat is late-data dropping on out-of-order event-time streams, where Flink is itself non-deterministic (periodic watermarks); we match Flink's deterministic path, which governs in-order data and every benchmark.

Inspiration

StreamFusion is built by porting established engines rather than reinventing operators:

• DataFusion Comet — the model for the whole project (native columnar accelerator behind an unchanged SQL planner) and the reference for the JNI / Arrow C Data Interface bridge, off-heap memory accounting, the config surface, and fallback-reason reporting.
• Arroyo — the streaming-operator implementations we port (it already runs on DataFusion); the reference for join/window/changelog logic.
• Apache DataFusion — the native execution and expression engine underneath (hash joins, aggregates, Arrow kernels).
• RisingWave — the reference for changelog semantics and memcomparable arrow-row state encoding.
• Apache Flink — the parity target: every operator is a faithful port of Flink's own, verified for identical output by a parity harness.

Running and configuration

Install acceleration by hooking the planner once (NativePlanner.install(env)), then run Flink SQL as normal. Two things to set in a real deployment:

Deployment JVM flags — run the TaskManager JVM with Arrow's safety checks off (as Comet/Spark do); profiling showed ~1/3 of the transpose CPU was per-accessor bounds/refcount checks:

-Darrow.enable_unsafe_memory_access=true -Darrow.enable_null_check_for_get=false

Configuration (JVM system properties, mirroring Comet's config surface):

• -Dstreamfusion.native.enabled=false — master switch; run entirely on Flink.
• -Dstreamfusion.operator.<name>.enabled=false — keep one operator on the host (e.g. leave a lone cheap filter on a row source, which can't earn back the transpose round-trip).
• -Dstreamfusion.expression.<NAME>.allowIncompatible=true — opt into the faster pure-Rust path for expressions that otherwise use a byte-exact JVM upcall or fall back (UPPER/LOWER/ REGEXP_EXTRACT, ROUND on float, transcendental math). Off by default (parity-first).
• -Dstreamfusion.memory.accounting.enabled (default on) — native stateful operators reserve an operator-scope share of the slot's managed memory from Flink's MemoryManager and bound their state by it, failing with a NativeMemoryLimitException naming the remedy rather than an unattributed OOM.

Related work

Three native Flink accelerators exist, all closed source:

• Flash (Alibaba Cloud) — a C++ native + SIMD vectorized engine with a custom state backend (ForStDB). Stateful, production-deployed at scale; claims 5–10× on streaming Nexmark, 3×+ on batch TPC-DS, and ~50% cost reduction across 100k+ compute units.
• Vera X (Ververica, the original Flink creators) — a proprietary native vectorized engine with a drop-in compatibility layer and a new state store. Stateful; claims 5–10× on Nexmark SQL and ~52% lower resource usage.
• Iron Vector (Irontools) — the same stack as us (Rust + Arrow + DataFusion over zero-copy JNI, Substrait plan serialization, transparent fallback), but stateless only today (projections, filters, expressions); windows, joins, and exactly-once are described as planned.

Where StreamFusion differs: it is open source, and every substitution is gated and verified for identical results against stock Flink by a parity harness rather than asserted. It is already native on stateful windowing, joins, and changelog processing — the hard, closed part of the field — where Iron Vector is stateless-only.

License

Licensed under the Apache License, Version 2.0.