# 数据流

> 介绍 Flownode 如何选择并运行 batching 和旧 streaming 两条执行路径。

# 数据流

Flownode 内部有两条执行路径：

- **Batching mode** 是聚合和 TQL workload 的主要执行路径。它查询已经持久化的 source 数据，并将物化结果写入 sink table。
- **Streaming mode** 是为兼容已有 workload 而保留的旧执行路径，不推荐新 workload 使用。Frontend 会把新到达的行同步给它进行增量处理。

用户不能直接选择执行模式。创建 Flow 时，GreptimeDB 先根据 source table、再根据查询判定执行路径，顺序如下：

```mermaid
flowchart LR
    MISS{"存在缺失的源表？"}
    MISS -->|"是，已声明延迟创建"| PEND["pendingbatching Flow"]
    MISS -->|"是，其余情况"| ERR["拒绝创建"]
    MISS -->|"否"| TTL{"存在 ttl = 'instant'的源表？"}
    TTL -->|"是"| STREAM["Streaming mode"]
    TTL -->|"否"| TQL{"TQL 查询？"}
    TQL -->|"是"| BATCH["Batching mode"]
    TQL -->|"否"| AGG{"计划含 Aggregate或 Distinct？"}
    AGG -->|"是"| BATCH
    AGG -->|"否"| STREAM
```

source table 的判定在前，因此聚合 `ttl = 'instant'` source table 的 Flow 走的是 streaming mode，而不是 batching mode。source table 尚不存在时，Flow 会被拒绝创建，除非建表时声明 `WITH (defer_on_missing_source = true)`，此时会得到一个 pending batching Flow。

## Batching mode

Batching mode 复用 GreptimeDB 的查询引擎，不需要为每一行输入维护一张算子图。对于基于时间窗口的 Flow，主循环如下：

1. Source table 收到写入后，把受影响的时间窗口标记为 dirty。
2. `BatchingTask` 按求值调度或自适应轮询节奏运行，并在该次求值时收集待处理的 dirty window。标记 dirty window 不会唤醒任务。
3. 任务把这些窗口转换成时间谓词，加入 Flow 查询，再请求 Frontend 查询 source table。
4. 查询结果写入 sink table，更新已重新计算窗口对应的物化结果。
5. 成功处理的窗口从 dirty set 中移除；执行失败的工作仍可在后续调度中处理。

TQL Flow，以及计划无法按 dirty window 安全裁剪的 evaluation interval Flow，执行的是完整查询，而不是按时间过滤的查询；这条路径上 dirty set 只作为调度信号。这条路径还可以使用 streaming renderer 尚未实现的查询引擎能力。任务和 dirty window 组件的进一步说明见[批处理模式](./batching_mode.md)。

## Streaming mode

Dataflow 模块（参见 `flow::compute` 模块）是 `flow` 的核心计算模块。
它接收 SQL 查询并将其转换为 `flow` 的内部执行计划。
然后，该执行计划被转化为实际的数据流，而数据流本质上是一个由带有输入和输出端口的函数组成的有向无环图（DAG）。
新到达的行变更会增量驱动这张图执行。

Renderer 支持 map/filter/project 和 reduce 操作。执行计划中已经有 join 和 union 节点，但 streaming renderer 尚未实现它们。

在内部，数据流使用 `tuple(row, time, diff)` 以行格式处理数据。
这里 `row` 表示实际传递的数据，可能包含多个 `value` 对象。
`time` 是系统时间，用于跟踪数据流的进度，`diff` 通常表示行的插入或删除（+1 或 -1）。
因此，`tuple` 表示给定系统时间的 `row` 的插入/删除操作。有状态算子通过 [Arrangement](./arrangement.md) 保存这些变更的索引 trace。