greptimedb/src/flow/AGENTS.md

flow — Agent & Contributor Guide

Navigation aid for src/flow. Keep it short and point to code. Paths are relative to the repo root.

Repo-wide rules that apply here: .agents/architecture-invariants.md.

What this crate does

Flownode is the stream-processing engine behind continuous aggregation / materialized views. It has two execution paths:

• 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.

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
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:

• 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).

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).
• 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

cargo nextest run -p flow

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

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.