Streaming Execution

Overview

When ARO processes a large file, the naive approach loads everything into memory at once. A 10 GB CSV file would consume 20 GB or more of RAM, crashing most systems. With streaming execution, the same code runs in roughly 1 MB of memory by processing data one row at a time.

Syntax Transparency

There are no new keywords. The code you already write is the streaming code:

(* This is automatically streamed - no syntax change needed *)
Read the <data> from the <file: "huge.csv">.
Filter the <filtered> from <data> where <status> = "active".
Log <filtered> to the <console>.

Under the hood, ARO sees this as a pipeline of three stages. Each row flows from Read through Filter to Log without the entire dataset ever being held in memory.

  ┌──────────┐    ┌──────────┐    ┌──────────┐
  │  Read    │───>│  Filter  │───>│   Log    │
  │ (Stream) │    │ (Stream) │    │ (Drain)  │
  └──────────┘    └──────────┘    └──────────┘
       │               │               │
       │    Row 1      │    Pass?      │    Print
       │──────────────>│──────────────>│────────> console
       │    Row 2      │    Reject     │
       │──────────────>│───────X       │
       │    Row 3      │    Pass?      │    Print
       │──────────────>│──────────────>│────────> console

  Memory: O(1) per row, not O(n) for entire dataset

Action Classification

Actions fall into three categories that determine how they participate in a streaming pipeline:

Explicit Control

Streaming is the default. If you need the legacy eager behavior for small files that will be accessed multiple times, use the eager qualifier:

(* Default: streaming - constant memory *)
Read the <data> from the <file: "huge.csv">.

(* Explicit: load everything into memory *)
<Read: eager> the <data> from the <file: "small.csv">.

Stream Tee

A stream can only be consumed once. When the same variable feeds into multiple downstream operations of different types, ARO automatically creates a Stream Tee that splits the stream for parallel consumption:

Filter the <active> from <orders> where <status> = "active".
Reduce the <total> from <active> with sum(<amount>).  (* consumer 1 *)
Log <active> to the <console>.                       (* consumer 2 *)

The Stream Tee uses a bounded ring buffer so both consumers see every element. The fastest consumer drives the stream; slower consumers read from the buffer. If a consumer falls behind, the buffer spills to disk automatically.

                          ┌─────────────────┐
  active-stream ────────> │   Stream Tee    │
                          └─────────────────┘
                               │         │
                               v         v
                          Consumer 1  Consumer 2
                          (Reduce)    (Log)

Pipeline Optimization

Aggregation Fusion

When multiple Reduce operations consume the same stream, ARO's semantic analyzer fuses them into a single pass. Instead of iterating the data three times, the runtime maintains all accumulators simultaneously:

(* These three reduces are fused into one pass *)
Reduce the <total> from <active-orders> with sum(<amount>).
Reduce the <count> from <active-orders> with count().
Reduce the <avg> from <active-orders> with avg(<amount>).

The parser detects that all three operations share the same source (active-orders) and rewrites them into a single fused reduce. Memory usage is O(1) regardless of dataset size -- just the accumulators.

Unified Data Sources

Every data source in ARO produces the same stream abstraction. The same filter, reduce, and transform operations work identically whether the data comes from a file, an HTTP response, a socket, or a repository:

Example: Multi-Stage Pipeline

This example processes a list of transactions through multiple filter stages, then runs aggregation on the result. In streaming mode, each row flows through all stages immediately -- non-matching rows are discarded without allocating memory:

(Application-Start: Streaming Pipeline Demo) {
    Create the <transactions> with [
        {id: 1, year: "2024", amount: 1200, status: "completed", category: "electronics"},
        {id: 2, year: "2024", amount: 450, status: "pending", category: "electronics"},
        {id: 3, year: "2024", amount: 800, status: "completed", category: "furniture"},
        {id: 4, year: "2024", amount: 950, status: "completed", category: "electronics"},
        {id: 5, year: "2024", amount: 2000, status: "completed", category: "electronics"}
    ].

    (* Multi-stage filter pipeline *)
    Filter the <high-value> from <transactions>
        where <amount> > 500.

    Filter the <completed> from <high-value>
        where <status> = "completed".

    Filter the <electronics> from <completed>
        where <category> = "electronics".

    (* Aggregation fusion: all reduces run in a single pass *)
    Reduce the <total-amount> from <electronics> with sum(<amount>).
    Reduce the <count> from <electronics> with count().
    Reduce the <avg-amount> from <electronics> with avg(<amount>).
    Reduce the <max-amount> from <electronics> with max(<amount>).

    Log <total-amount> to the <console>.
    Log <count> to the <console>.
    Log <avg-amount> to the <console>.
    Log <max-amount> to the <console>.

    Return an <OK: status> for the <streaming-demo>.
}

Example: JSONL Streaming

JSONL (JSON Lines) is the ideal format for streaming because each line is an independently parseable JSON object. ARO reads JSONL files line by line without loading the entire file:

(Application-Start: Streaming Demo) {
    (* Read JSONL file - streamed line by line in production *)
    Read the <log-data> from "./events.jsonl".

    (* Filter errors - streaming: pass/reject each element *)
    Filter the <errors> from <log-data> where <level> = "ERROR".

    (* Multi-stage filter pipeline *)
    Filter the <api-events> from <log-data> where <service> = "api".
    Filter the <api-errors> from <api-events> where <level> = "ERROR".

    (* Reduce with O(1) memory accumulators *)
    Reduce the <error-count> from <errors> with count().
    Reduce the <total-time> from <api-events> with sum(<time>).
    Reduce the <avg-time> from <api-events> with avg(<time>).

    Log <error-count> to the <console>.
    Log <avg-time> to the <console>.

    Return an <OK: status> for the <demo>.
}