feat: Streaming execution mode — MVP

[malon64]

Context

Floe's current execution model is entirely batch-based: every stage (read, validate, split, write) materializes a full DataFrame in memory before passing it to the next stage. For large files this means peak memory usage equals the full file size, which becomes a bottleneck at scale.

This issue defines the scope and implementation plan for a streaming MVP: processing data chunk-by-chunk from source to sink, never holding the full file in memory at once.

What "streaming" means in Floe

Floe's validation layer is imperative Rust code — it cannot be integrated into the Polars streaming execution engine (which only runs Polars expressions end-to-end). The correct approach is manual chunk iteration: the source emits fixed-size batches of rows, each batch goes through validation independently, and accepted/rejected chunks are written immediately without accumulation.

source → [chunk N rows] → validate → split → write accepted chunk
                                            → write rejected chunk
         [chunk N rows] → validate → split → write accepted chunk
                                            → write rejected chunk
         ...
         → commit (Delta snapshot / Iceberg snapshot)

Constraints and out-of-scope for MVP

Feature	Streaming MVP	Reason
cast_mismatch check	✅ Supported	Chunk-stateless — operates on column masks
not_null check	✅ Supported	Chunk-stateless
unique check	❌ Disabled	Requires global HashSet across all chunks + existing table (seed_from_df) — fundamentally incompatible with stateless chunk processing
merge_scd1 / merge_scd2	❌ Disabled	Requires full table scan for MERGE INTO
Append / Overwrite write modes	✅ Supported	Delta and Iceberg both support incremental batch writes
Schema mismatch (file-level)	✅ Supported	Detected before chunked read begins

Entities declaring unique: true on any column or using write_mode: merge_scd1/merge_scd2 must be rejected at config validation time with an explicit error if streaming mode is requested.

Implementation plan

Gap 1 — ReadInput struct: add Stream variant

File: crates/floe-core/src/io/format.rs

pub enum ReadInput {
    // Existing batch variant — unchanged
    Data {
        input_file: InputFile,
        raw_df: Option<DataFrame>,
        typed_df: DataFrame,
    },
    // New streaming variant
    Stream {
        input_file: InputFile,
        chunks: Box<dyn Iterator<Item = FloeResult<(DataFrame, DataFrame)>>>,
        // Each item: (raw_chunk, typed_chunk) in lockstep
    },
    FileError { ... },
}

The (raw_chunk, typed_chunk) lockstep is required because cast_mismatch compares raw string values against typed values column-by-column.

Readers to update: csv.rs and parquet.rs — remove the immediate .collect() calls and emit chunk iterators. LazyCsvReader and scan_parquet are already lazy; the .collect() is the only thing to remove. Other formats (Avro, JSON, ORC, XML) remain batch-only for now.

---

Gap 2 — Checks: no changes needed

cast_mismatch_errors_sparse and not_null_errors_sparse already operate on a &DataFrame of arbitrary size. A chunk of N rows is processed identically to a full DataFrame. Zero changes required.

Note: if issue #252 (checks → Polars expressions) lands first, this gap remains zero effort.

---

Gap 3 — AcceptedSinkAdapter trait: add streaming method

File: crates/floe-core/src/io/format.rs

Add an opt-in method with a default fallback so existing sink implementations are unaffected:

pub trait AcceptedSinkAdapter: Send + Sync {
    // Existing batch API — unchanged
    fn write_accepted(&self, df: &mut DataFrame, ...) -> FloeResult<AcceptedWriteOutput>;

    // New streaming API — default collects chunks and calls write_accepted (safe fallback)
    fn write_accepted_streaming(
        &self,
        chunks: &mut dyn Iterator<Item = DataFrame>,
        ...
    ) -> FloeResult<AcceptedWriteOutput> {
        let df = collect_chunks(chunks)?;
        self.write_accepted(&mut df, ...)
    }
}

Delta and Iceberg override write_accepted_streaming with true incremental writes (multiple batch commits → single snapshot at the end). Parquet inherits the fallback.

---

Gap 4 — Run loop: new run_streaming() entry point

File: crates/floe-core/src/run/entity/mod.rs

The current run_entity() accumulates all accepted DataFrames into a Vec<DataFrame> and vstacks them before writing. In streaming mode, each accepted chunk is written immediately:

// Current batch loop (simplified)
let mut accepted_accum = vec![];
for input in inputs {
    let (accepted_df, rejected_df) = validate_split(input)?;
    accepted_accum.push(accepted_df);
    write_rejected(rejected_df)?;
}
write_accepted(concat(accepted_accum))?;  // ← one big write at the end

// New streaming loop
let mut streaming_sink = sink_adapter.begin_streaming(...)?;
for chunk in source.chunks() {
    let (accepted_chunk, rejected_chunk) = validate_split_chunk(chunk)?;
    streaming_sink.write_chunk(accepted_chunk)?;
    write_rejected(rejected_chunk)?;
}
streaming_sink.commit()?;  // ← single Delta/Iceberg snapshot at the end

This is a new code path (run_entity_streaming()), not a modification of the existing batch path. Both coexist.

---

Gap 5 — Config: opt-in execution mode

Add execution_mode to EntityConfig (or SinkTarget):

entity:
  name: orders
  execution_mode: streaming   # "batch" (default) or "streaming"
  source:
    format: csv
    path: s3://bucket/orders/*.csv
  sink:
    accepted:
      format: delta
      path: s3://bucket/warehouse/orders
      write_mode: append

Static validation at startup rejects streaming mode if:

• Any column declares unique: true
• write_mode is merge_scd1 or merge_scd2
• Source format is not csv or parquet (Avro/JSON/ORC/XML not supported in MVP)
• Sink format is not delta or iceberg (Parquet falls back to batch via default impl)

---

Dependency on #252

#252 (checks → Polars expressions) is not a hard prerequisite for this issue. The existing imperative checks already work on arbitrary-sized DataFrames. However, if #252 lands first, the streaming pipeline becomes a candidate for the Polars streaming engine on entities without unique constraints — a significant additional benefit.

Recommended order: #252 → this issue.

Estimated effort

Gap	Effort
ReadInput Stream variant + CSV/Parquet readers	3-4 days
AcceptedSinkAdapter streaming method + Delta/Iceberg impl	3-4 days
run_entity_streaming() loop	4-5 days
Config + static validation	2 days
Tests (unit + integration with large file fixture)	3-4 days
Total	~3 weeks