How Flow Computation Enables Real-Time Stream Processing in GreptimeDB

Flow computation in GreptimeDB enables real-time stream processing through incremental state maintenance, sequence-based delta reads, and an embedded frontend that eliminates network hops, delivering sub-second latency for continuous aggregation queries.

GreptimeDB implements continuous aggregation through Flow computation, a built-in real-time stream processing engine that processes data incrementally as it arrives. Unlike traditional batch-oriented systems that require periodic full-table scans, Flow computation maintains aggregation state in memory and updates results continuously, enabling real-time analytics on streaming data with minimal latency.

Core Architecture of Flow Computation

The Flow computation architecture separates state management from computation logic, distributing coordination across lightweight nodes while centralizing aggregation state in memory.

Flownode as the Lightweight Coordinator

The Flownode serves as the primary coordinator for Flow computation. Defined in src/plugins/src/flownode.rs, this component stores flow state and forwards queries to an embedded frontend. It receives DDL commands (create, drop, flush) and data-ingest requests from the meta service, acting as the entry point for all Flow computation operations.

FlowDualEngine and Dual-Mode Execution

At the heart of the system lies the FlowDualEngine, implemented in src/flow/src/adapter/flownode_impl.rs. This engine manages two distinct execution modes:

• StreamingEngine – Handles low-latency incremental aggregation for real-time stream processing
• BatchingEngine – Processes larger, periodic jobs when data volume exceeds streaming thresholds

The FlowDualEngine routes inserts and DDL to the appropriate engine and synchronizes state with the meta service, ensuring consistent behavior across both execution paths.

StreamingEngine for Low-Latency Processing

The StreamingEngine, located at lines 1090-1130 in src/flow/src/adapter/flownode_impl.rs, executes incremental queries using the internal __aggr_state function. It maintains an in-memory FlowStat map with the structure Map<Timestamp, (Map<Key, Value>, Sequence)>.

When new data arrives, the StreamingEngine updates this state map and immediately emits results if the evaluation interval has expired. This design eliminates the need to scan historical data, achieving sub-second latency for real-time stream processing workloads.

BatchingEngine for High-Throughput Workloads

For scenarios where data volume exceeds the efficient capacity of incremental processing, the BatchingEngine (lines 1015-1050 in src/flow/src/adapter/flownode_impl.rs) executes the same aggregation logic on scheduled intervals (e.g., every second).

While it utilizes the same state map structure, the BatchingEngine performs heavy scans on the datanode side rather than maintaining purely incremental state. This dual approach allows Flow computation to adapt dynamically to varying workload characteristics without changing the underlying flow definition.

State Management and Incremental Computation

Real-time stream processing in GreptimeDB relies on sophisticated state management that minimizes I/O while maintaining correctness.

The FlowState Map Structure

The core state primitive is the FlowState map, defined as Map<Timestamp, (Map<Key, Value>, Sequence)>. This structure stores:

• Timestamp – The time window for the aggregation
• Key – Grouping keys for the aggregation
• Value – Computed aggregate values
• Sequence – A monotonic marker indicating the last processed write sequence

This map resides in memory within the Flownode, enabling microsecond-level state updates essential for real-time stream processing.

Sequence-Based Delta Reads

To avoid scanning entire tables, Flow computation implements sequence-based incremental reads. When requesting data from datanodes, the Flownode sends memtable_last_seq and sst_last_seq values in the gRPC headers.

The datanode returns only rows whose internal write sequence lies within the requested range, drastically reducing scan volume from millions of rows to mere deltas. This mechanism is crucial for maintaining low latency in high-throughput real-time stream processing scenarios.

Handling Refills After Compaction

When memtables flush to SST files, the exact sequence granularity may be lost. In these cases, the Flownode triggers a refill process, recomputing affected windows from source data.

While this temporarily increases processing overhead, it ensures correctness without requiring the entire historical state to remain in memory. The refill mechanism balances the trade-off between memory efficiency and computational accuracy in distributed real-time stream processing.

Real-Time Data Flow Path

Understanding the end-to-end data path clarifies how Flow computation achieves its performance characteristics.

When an Insert arrives, the client sends InsertRequests to the meta service, which routes to FlowDualEngine::handle_flow_inserts in src/flow/src/adapter/flownode_impl.rs.

The engine performs a lookup on the source table mapping. If the flow is registered in src_table2flow.stream, it routes to the streaming path.

StreamingEngine::handle_inserts_inner reorders columns, stamps a logical tick, and invokes handle_write_request. This updates the in-memory state map and, if the evaluation interval has expired, triggers an incremental query using the internal __aggr_state function on the embedded frontend.

Results materialize into the sink table on the datanode without a network hop, leveraging the laminar flow architecture to deliver sub-second latency for real-time stream processing.

Consistency and Recovery Mechanisms

Distributed real-time stream processing requires robust failure handling.

On startup, the ConsistentCheckTask scans meta-service flow metadata to recreate missing flows and waits for state report heartbeats to confirm recovery. This ensures that Flow computation resumes from the correct state after node restarts.

The FlowDualEngine::try_sync_with_check_task method guarantees that flush or drop operations only return after the underlying engine has observed the change. This synchronization prevents lost updates and ensures that real-time stream processing maintains exactly-once semantics for state modifications.

Practical Implementation Examples

The following examples demonstrate how to interact with Flow computation using GreptimeDB's client libraries.

Creating a Flow

The FlowRequester in src/client/src/flow.rs provides the interface for flow management.

use greptimedb::client::Client;
use greptimedb::client::flow::FlowRequester;

#[tokio::main]
async fn main() -> greptimedb::client::Result<()> {
    // Initialise a GreptimeDB client (address omitted for brevity)
    let client = Client::new("127.0.0.1:4001")?;
    let flow_req = FlowRequester::new(client);

    // Example CREATE FLOW DDL (same syntax as regular SQL)
    let create_sql = r#"
        CREATE FLOW my_flow
        INTO my_sink_table
        FROM my_source_table
        AGGREGATE avg(value) AS avg_val
        INTERVAL 5 SECOND
    "#;

    // Build a FlowRequest manually (the client library does this internally)
    let request = greptime_proto::v1::flow::FlowRequest {
        header: Some(greptime_proto::v1::flow::FlowRequestHeader {
            tracing_context: greptime_proto::v1::trace::TraceContext::default(),
            query_context: None,
        }),
        body: Some(greptime_proto::v1::flow::flow_request::Body::Create(
            greptime_proto::v1::flow::CreateRequest {
                flow_id: Some(greptime_proto::v1::FlowId { id: 0 }), // let server allocate
                source_table_ids: vec![greptime_proto::v1::TableId { id: 1 }],
                sink_table_name: Some(greptime_proto::v1::TableName {
                    catalog_name: "public".into(),
                    schema_name: "public".into(),
                    table_name: "my_sink_table".into(),
                }),
                create_if_not_exists: true,
                or_replace: false,
                expire_after: None,
                eval_interval: Some(greptime_proto::v1::flow::EvalInterval { seconds: 5 }),
                comment: "".into(),
                sql: create_sql.into(),
                flow_options: Default::default(),
            },
        )),
    };

    let resp = flow_req.handle_inner(request).await?;
    println!("Created flow IDs: {:?}", resp.affected_flows);
    Ok(())
}

Key source: FlowRequester::handle_inner in [src/client/src/flow.rs](https://github.com/greptimeteam/greptimedb/blob/main/src/client/src/flow.rs#L64-L79) sends the request to the flownode via the embedded gRPC client.

Inserting Data into Source Tables

Data ingestion triggers the real-time stream processing pipeline.

use greptimedb::client::{Client, Result};
use greptime_proto::v1::region::InsertRequests;

#[tokio::main]
async fn main() -> Result<()> {
    let client = Client::new("127.0.0.1:4001")?;
    let flow_req = greptimedb::client::flow::FlowRequester::new(client);

    // Build an InsertRequests payload (normally generated by the SDK)
    let insert = InsertRequests {
        requests: vec![greptime_proto::v1::region::InsertRequest {
            region_id: greptime_proto::v1::RegionId { id: 42 },
            rows: Some(greptime_proto::v1::Rows {
                schema: vec![], // column schema omitted for brevity
                rows: vec![],   // actual row data
            }),
            partition_expr_version: None,
        }],
    };

    // Forward to the flownode – the engine will route it to the streaming engine
    let _ = flow_req.handle_inserts_inner(insert).await?;
    Ok(())
}

Key path: FlowDualEngine::handle_flow_inserts (src/flow/src/adapter/flownode_impl.rs lines ~1085‑1130) decides whether the request goes to StreamingEngine or BatchingEngine.

Flushing a Flow

Force materialization of pending aggregation windows.

use greptimedb::client::flow::FlowRequester;
use greptime_proto::v1::flow::{FlowRequest, FlowRequestHeader, FlowId, flow_request};

#[tokio::main]
async fn main() -> greptimedb::client::Result<()> {
    let client = greptimedb::client::Client::new("127.0.0.1:4001")?;
    let flow_req = FlowRequester::new(client);

    let flush = FlowRequest {
        header: Some(FlowRequestHeader::default()),
        body: Some(flow_request::Body::Flush(
            greptime_proto::v1::flow::FlushFlow {
                flow_id: Some(FlowId { id: 1 }), // the ID returned by CREATE
            },
        )),
    };

    let resp = flow_req.handle_inner(flush).await?;
    println!("Flushed rows: {}", resp.affected_rows);
    Ok(())
}

Implementation: FlowServiceOperator::flush (src/operator/src/flow.rs lines 55‑61) forwards the request to the appropriate engine; the engine then executes __aggr_merge on all pending windows.

Summary

• Flow computation in GreptimeDB provides built-in real-time stream processing through incremental aggregation and in-memory state management.
• The FlowDualEngine automatically routes workloads between StreamingEngine (low-latency) and BatchingEngine (high-throughput) based on data characteristics.
• Sequence-based delta reads minimize I/O by requesting only new data since the last computation, while the FlowState map (Map<Timestamp, (Map<Key, Value>, Sequence)>) maintains aggregation windows in memory.
• The embedded frontend architecture eliminates network hops between compute and storage, delivering sub-second latency for materializing results into sink tables.
• Automatic refill mechanisms ensure correctness when memtable compaction invalidates sequence precision, balancing memory efficiency with computational accuracy.

Frequently Asked Questions

What is Flow computation in GreptimeDB?

Flow computation is GreptimeDB's built-in engine for real-time stream processing and continuous aggregation. It allows users to define persistent queries that automatically process incoming data streams, maintain aggregation state in memory, and emit results to sink tables without requiring external stream processing systems like Apache Flink or Kafka Streams.

How does Flow computation achieve sub-second latency?

Flow computation achieves sub-second latency through three key mechanisms: incremental state maintenance using the in-memory FlowStat map, sequence-based delta reads that fetch only new data rather than full table scans, and the embedded frontend architecture that processes queries within the flownode without network hops to separate query engines. The StreamingEngine in src/flow/src/adapter/flownode_impl.rs executes the __aggr_state function immediately upon data arrival when evaluation intervals expire.

What is the difference between StreamingEngine and BatchingEngine?

The StreamingEngine (lines 1090-1130 in src/flow/src/adapter/flownode_impl.rs) processes data incrementally as it arrives, updating the in-memory state map and emitting results immediately when evaluation intervals trigger. It is optimized for low-latency real-time stream processing. The BatchingEngine (lines 1015-1050) processes data on scheduled intervals (e.g., every second) and performs heavier scans on the datanode side rather than maintaining purely incremental state. This dual approach allows Flow computation to adapt dynamically to varying workload characteristics without changing the underlying flow definition.

How does GreptimeDB ensure consistency during node failures?

GreptimeDB ensures consistency through the ConsistentCheckTask, which scans meta-service flow metadata on startup to recreate missing flows and waits for state report heartbeats to confirm recovery. This ensures that Flow computation resumes from the correct state after node restarts. The FlowDualEngine::try_sync_with_check_task method guarantees that flush or drop operations only return after the underlying engine has observed the change. This synchronization prevents lost updates and ensures that real-time stream processing maintains exactly-once semantics for state modifications.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}