Skip to main content

Sources

the source

A pipeline has to read from somewhere, and fusion's first structural decision is that the engine is a streaming consumer defined against a source, not an array trick. The same stages and terminals compile over three source shapes, and which one you're on is written in the pipeline's type.

The three shapes​

  • An in-memory FArray: xs.fuse. The degenerate case: the whole collection is one chunk, and the emitted code is the tight indexed loop shown on the previous page. run hands you an FArray back, so fused pipelines slot invisibly into eager code.

  • A chunked Source[A]: Source.fromIterator(it).fuse, Source.unfold(z)(step), Source.iterate(x)(f). Pull-based and chunk-granular: one virtual pullChunk call hands the loop a whole unboxed leaf (~4096 elements), so the hot path stays the same element loop, with no per-element virtual call and no boxed sentinel. Constant memory: the loop holds one chunk, not the data, so a gigabyte input needs kilobytes of heap. And the driver wraps the loop in try/finally close(), so a file or socket behind the source is released on exhaustion, on short-circuit (take/find stopping early), and on exception alike.

  • A record-framed byte source (ByteRecordSource): Json.ndjson[Rec](bytes).stream: elements decoded straight out of raw bytes, fields the pipeline doesn't read skipped at the byte level. Same stages, same terminals, same optimizer; the details are on the JSON integration page.

The chunked contract is two methods. A source hands back leaves until it's done, and promises a chunk is only valid until the next pull, which is the license an InputStream-backed source needs to reuse one read buffer across pulls:

loading source…

Streaming, measured​

Constant-memory processing on the JVM usually costs throughput: Iterator chains and java.util.stream pipelines pay per-element virtual dispatch, LazyList (Scala's lazily evaluated linked list) allocates a cons cell per element, effect-system streams pay their runtime. The chunk-granular design keeps fusion's loop shape over lazy data. The benchmark folds a generated sequence, and a file's lines, without holding either in memory:

measuring…

The shape is in the type​

The entry point stamps the pipeline: xs.fuse gives Fuse[A, Chunks], and the JSON module's .stream gives Fuse[T, Ndjson]. That phantom parameter, the source shape, drives the rest of the machine: stages are shape-generic, but which terminals exist, and how they lower, is decided per shape by a typeclass (an interface implementation the compiler selects by type, like a ServiceLoader resolved at compile time). It has no runtime cost; it exists so that an unsupported operation is a compile error rather than a runtime surprise.