AGENTS.md@src/flow · diff

git:20260810.5d4699d to git:20260917.7c7132e

31 added, 52 removed. Audit A to A.

# flow — Agent & Contributor Guide

Navigation aid for `src/flow`. Keep it short and point to code. Paths are

- relative to the repo root.

+ relative to the repository root.

- Repo-wide rules that apply here: [`.agents/architecture-invariants.md`](../../.agents/architecture-invariants.md).

+ Repo-wide rules that apply to here: [`.agents/architecture-invariants.md`](../../.agents/architecture-invariants.md).

## What this crate does

- Flownode is the stream-processing engine behind continuous aggregation /

- materialized views. It has two execution paths:

+ Flownode provides two flow profiles:

- - **Batching mode** (the actively developed path): splits data

- into time windows and periodically runs aggregation SQL through the frontend,

- writing results back to a sink table.

- - **Streaming mode** (the legacy dataflow path): an incremental DFIR/dataflow

- compute graph that processes row-level diffs.

+ - **Streaming mode**: a single stateless DataFusion query profile. Each mirror

+ insert is materialized as a transient input table, executed against the

+ retained logical plan, and written directly to the sink through

+ `batching_mode::frontend_client::FrontendClient`.

+ - **Batching mode**: splits data into time windows and periodically runs

+ aggregation SQL through the frontend, writing results back to a sink table.

Users cannot select a mode directly: `flow_type` is a reserved internal option.

`StatementExecutor::determine_flow_type` in `src/operator/src/statement/ddl.rs`

owns current mode selection. `FlowDualEngine` defaults missing internal

`flow_type` metadata to batching for compatibility. Read those paths before

changing routing rules.

## Module map

| Module | Path | Purpose |

| --- | --- | --- |

- | `engine` | `src/flow/src/engine.rs` | `FlowEngine` trait: create/remove/flush/insert lifecycle |

- | `adapter` | `src/flow/src/adapter.rs`, `src/flow/src/adapter/` | `StreamingEngine`, worker pool, dual-engine dispatch, table sources/sinks |

- | `batching_mode` | `src/flow/src/batching_mode.rs`, `src/flow/src/batching_mode/` | `BatchingEngine`, task scheduling, time windows, frontend client, checkpoints |

- | `compute` | `src/flow/src/compute/` | Streaming dataflow render/state |

- | `expr` | `src/flow/src/expr.rs`, `src/flow/src/expr/` | Scalar/aggregate expressions and Map-Filter-Project |

- | `plan` | `src/flow/src/plan.rs` | `TypedPlan` (reduce/join/MFP) |

- | `transform` | `src/flow/src/transform.rs` | Substrait → flow plan |

- | `df_optimizer` | `src/flow/src/df_optimizer.rs` | SQL → DataFusion logical plan → optimized plan |

- | `repr` | `src/flow/src/repr.rs` | `Row`, `DiffRow`, `Batch`, `RelationDesc` |

- | `server` | `src/flow/src/server.rs` | gRPC `Flow` service, `FlownodeBuilder`/`FlownodeInstance` |

+ | `engine` | `src/flow/src/engine.rs` | `FlowEngine` lifecycle contract |

+ | `adapter` | `src/flow/src/adapter.rs`, `src/flow/src/adapter/` | Stateless streaming execution, dual-engine dispatch, shared source schema/default normalization, and sink creation |

+ | `batching_mode` | `src/flow/src/batching_mode.rs`, `src/flow/src/batching_mode/` | `BatchingEngine`, scheduling, time windows, frontend client, and checkpoints |

+ | `repr` | `src/flow/src/repr.rs` | Shared row and relation schema representations |

+ | `server` | `src/flow/src/server.rs` | gRPC `Flow` service and flownode builders |

| `heartbeat` | `src/flow/src/heartbeat.rs` | Reports flownode state/stats to metasrv |

Flow metadata lives in `common-meta`, not here:

`src/common/meta/src/key/flow/` and `src/common/meta/src/ddl/create_flow.rs`.

## Data flow

- `Frontend → Flownode (gRPC)` → `FlowService` (`server.rs`) →

- `FlowDualEngine` (`adapter/flownode_impl.rs`) routes by `FlowType`:

+ `Frontend → Flownode (gRPC)` → `FlowService` (`server.rs`) → `FlowDualEngine`

+ (`adapter/flownode_impl.rs`) routes by `FlowType`:

+ - Streaming: mirror rows are normalized against the source schema, evaluated by

+ the retained stateless DataFusion plan, and sent to the sink via

+ `FrontendClient`.

- Batching: marks dirty windows; a task later runs aggregation SQL via the

- frontend client and writes the sink table (`batching_mode/task.rs`).

- - Streaming: worker threads apply incremental diffs and push to the sink

- (`adapter/worker.rs`, `compute/render.rs`).

+ frontend client and writes the sink table.

+ `FlowType::Streaming` and mirror routing are shared contracts; preserve both

+ when changing dispatch. Preserve batching behavior and the shared schema,

+ default, and sink-creation helpers.

## Public surface

- - gRPC `Flow` service in `src/flow/src/server.rs`

- (`handle_create_remove`, `handle_mirror_request`, `handle_mark_dirty_time_window`).

- - Sink writes for the streaming engine go through `FrontendInvoker`

- (`row_inserts`, `row_deletes`).

+ - gRPC `Flow` service in `src/flow/src/server.rs`.

- `FlowEngine` trait in `src/flow/src/engine.rs`.

- Started from the `cmd` crate via `FlownodeBuilder` / `FlownodeInstance`.

- ## When you change X, also touch Y

- - **Flow definition / options**: validation in `common-meta`'s `ddl/create_flow.rs`

- and the serialized `FlowInfoValue` in `common-meta`'s `key/flow/`.

- - **New scalar/aggregate function** (`expr/`): also wire up evaluation in

- `compute/render.rs` (streaming) and ensure batching SQL handles it.

- - **Persisted flow metadata**: keep `FlowInfoValue` backward compatible

- (`serde(default)` / `serde(alias)`).

- - **Shared routing, metadata, or sink contracts**: check both engines. An

- engine-specific implementation change does not automatically need a mirror

- change in the other path.

## Testing

```bash

cargo nextest run -p flow

```

- Helpers in `src/flow/src/test_utils.rs` (test context, test query engine).

+ Stateless streaming tests are under `src/flow/src/adapter/stateless.rs`.

## Gotchas

- - Batching vs streaming differ in latency, state, and execution. Confirm the

- selected mode before reasoning about a flow.

- - Streaming workers are `!Send`; cross-thread interaction goes through

- `WorkerHandle`, not the worker directly.

- - Internal flow timestamps (`repr::Timestamp`, ms) are not necessarily the

- table's time column; window functions key off the diff timestamp.

- ## Maintenance contract

- Update this file when the dual-engine routing, the gRPC surface, or the flow

- metadata contract (shared with `common-meta`) changes.

+ - Batching and streaming differ in latency and state. Confirm the selected mode

+ before reasoning about a flow.

+ - Streaming is finite and stateless per mirror request; it has no worker graph,

+ background runtime, or replay state.

+ - Internal flow timestamps and table time columns are separate contracts.