优化后的语法规则（DataFlow v2 规范）
目标：面向分析师与数据工程师的稳健 DSL，统一数据流式语义、空值处理、时间与时区、可审计性与可维护性。
核心设计原则：
- 显式链式：强制使用管道操作符 |>，所有步骤显式连接。
- 类型与模式先行：数据源接受 schema 明确声明，可选择推断但需记录为元数据。
- 一致的空值语义：唯一的空值常量 null，提供 null-safe 比较与空值合并。
- 时间一致性：全局 timezone 与可控解析，时间函数与时区一致。
- 可审计性：可生成等效 SQL/计划，错误提示可指向具体子句。
- 无二义性：去除二义性操作符，统一关键字与表达式优先级。
——
分层规则如下。
- 词法与字面量
- 标识符：由字母/数字/下划线组成，不能以数字开头。保留字需使用反引号包裹，如
group。
- 字符串常量：使用单引号，例如 'Asia/Shanghai'、'北区'、'2025-01-01'。
- 数字常量：int、decimal(支持 123.45)。
- 布尔常量：true/false。
- 空值常量：null（唯一空值记号）。
- 注释：-- 行注释；/* ... */ 块注释。
- 类型系统与模式声明
- 基本类型：bool, int, bigint, decimal(p,s), double, string, date, timestamp, interval。
- 模式声明：schema { col: type, ... }，示例：
schema { date: date, item: string, qty: int, price: decimal(10,2) }
- 显式与推断：source 支持 with infer_schema true/false；显式 schema 优先于推断。
- 强制转换：expr :: type，例如 '2025-01-01'::date。
- 全局设置与会话
- 设置语句：
- set timezone 'Asia/Shanghai'
- set sql_dialect 'ansi' // 影响 explain sql 方言
- 时区：用于 timestamp 解析与时间函数；数据源可覆盖但必须记录在元数据中。
- Null 读入策略：在 source 层配置 null_as 列表，将命中值读取为 null。
- 运算符与表达式
- 比较运算：=, !=, >, >=, <, <=
- Null-safe 比较：=?
- a =? b 等价于 (a = b) 或 (a is null and b is null)
- 逻辑：and, or, not；短路求值。
- 其他空值语法：is null, is not null, coalesce(a, b, ...)
- 算术：+, -, *, /；对空值参与运算结果为 null（除 coalesce 后）。
- 函数（选摘）：date_parse(s, fmt [, tz]), to_date(s [, fmt]), to_timestamp(s [, fmt, tz]),
now(), date_add(d, interval '7' day), lag(x, k, default), lead(...), nvl 别名到 coalesce。
- 管道与求值模型
- 管道：dataset |> step1 |> step2 ...
- 左结合，左侧数据集作为右侧变换的隐式第一个参数。
- 每个 step 返回数据集（或元数据），直到 sink（write/save/plot）或 explain。
- 数据源与接收器
- 读取：source [with ] [schema {...}]
- CSV：source csv 'sales.csv' with delim ',', header true, encoding 'utf-8', null_as ['NA',''], tz 'Asia/Shanghai'
- 其他：json/parquet/sql(table='orders') 等（保留扩展点）
- 写出：write [with ]
- write csv 'out.csv' with header true, encoding 'utf-8'
- 逻辑保存：save as // 保存为可复用的逻辑视图（缓存可选）
- 缓存：cache [as ] [storage memory|disk] // 命中时优先使用
- 选择、派生、过滤
- select：投影与重命名
- select col, expr as alias, *
- 规则：同一 select 子句中，后项不能引用前面刚起的别名（避免顺序依赖）；需使用 derive。
- derive：新增/覆盖列
- derive new_col = expr [, ...]
- 覆盖时需显式关键字 replace：derive replace qty = coalesce(qty, 0)
- filter：过滤
- filter expr
- 支持 null-safe：filter region =? '北区'
- 排序、限制
- order by col [asc|desc] [nulls first|last]
- limit n [offset m]
- 分组与聚合
- 统一语法：
group by key1, key2 aggregate
sum(qty) as total_qty,
avg(price) as avg_price
- 规则：
- 聚合别名必须唯一，且不得与 group by 键冲突（如冲突报错，建议使用 rename 步骤）。
- 聚合函数支持：sum, avg, min, max, count, count_distinct, any_value 等。
- 窗口分析
- 在 derive 或 select 中使用 over 子句：
- derive qty_7d = sum(qty) over by item order date rows 7
- 语法：(expr) over [by ] [order <ts_col>] [rows N|range interval '7' day]
- 常用窗口函数：sum, avg, min, max, count, row_number(), lag(x, k, default), lead(...), dense_rank()
- 连接
- 语法：
- join [inner|left|right|full] <dataset_or_name> on [with suffixes ['_l','_r']]
- 默认 inner；支持 using(key) 简写：join left orders using(item_id)
- 别名与冲突：
- 冲突列名按 suffixes 规则处理；未指定时默认 ['_left','_right']。
- 可使用限定名 left.col/right.col 或 .col
- 采样与重现性
- sample bernoulli [seed ] [repeatable]
- sample bernoulli 0.1 seed 42 repeatable
- sample rows [seed ] // 保证随机且可复现
- 可视化
- plot x= y= [color=] [facet=] [title='...'] [width=... height=...]
- 类型：line, bar, scatter, area
- 输出：plot ... |> write png 'chart.png'（或在交互环境中内联渲染）
- 审计与优化
- explain sql // 输出等效 SQL（遵循 set sql_dialect）
- explain plan // 输出逻辑/物理执行计划
- validate // 执行静态检查：空值策略、时区、别名冲突、未解析列等
- 错误与诊断（示例）
- 使用 == 时：E1001: Operator '==' is not supported. Use '=' or '=?' for null-safe equality.
- 聚合别名冲突：E2103: Alias 'item' collides with group key. Rename or change alias.
- 未指明时区进行 timestamp 解析：W1302: Parsing timestamp without timezone. Consider set timezone 'Asia/Shanghai'.
- 保留字（选摘）
- select, derive, filter, group, aggregate, window, over, by, order, rows, range, join, on, using, left, right, full, inner, save, cache, write, source, schema, with, set, explain, sample, plot, limit, offset, null, true, false
——
语法使用示例（基础到高级）
示例 1：基础读取、筛选、聚合并保存
set timezone 'Asia/Shanghai'
source csv 'sales.csv'
with delim ',', header true, encoding 'utf-8', null_as ['NA', '']
schema { date: date, item: string, region: string, qty: int, price: decimal(10,2) }
|> filter date >= '2025-01-01' and region = '北区'
|> group by item aggregate
sum(qty) as total_qty,
avg(price) as avg_price
|> order by total_qty desc
|> save as daily_sales
示例 2：空值与 null-safe 比较、一致的别名规则
source csv 'customers.csv'
with header true, encoding 'utf-8', null_as ['NA']
schema { id: string, region: string, tier: string }
|> filter tier =? 'gold' -- tier 与 'gold' 同为 null 时视为相等
|> select id, coalesce(region, '未填') as region, tier
示例 3：7 日滚动窗口销量
source csv 'sales.csv'
with header true
schema { date: date, item: string, qty: int }
|> derive
qty_7d = sum(qty) over by item order date rows 7,
qty_lag1 = lag(qty, 1, 0) over by item order date
|> select date, item, qty, qty_lag1, qty_7d
示例 4：连接订单明细，处理列名冲突
let sales =
source csv 'sales.csv'
with header true
schema { item_id: string, date: date, qty: int }
let orders =
source sql table='orders'
schema { item_id: string, region: string, price: decimal(10,2) }
sales
|> join left orders using(item_id) with suffixes ['_s','_o']
|> derive revenue = qty * price
|> select date, item_id, qty, price, revenue, region
示例 5：采样与缓存（可复现实验）
source csv 'sales.csv' with header true
|> sample bernoulli 0.1 seed 42 repeatable
|> cache as sampled_sales
|> group by date aggregate sum(qty) as qty_sum
示例 6：时间解析与时区
set timezone 'Asia/Shanghai'
source csv 'events.csv'
with header true
schema { ts_str: string, user_id: string }
|> derive ts = to_timestamp(ts_str, 'yyyy-MM-dd HH:mm:ss') -- 采用会话时区
|> group by date(ts) as d aggregate count(*) as uv
示例 7：可视化与导出
save as daily_sales -- 假设前文已构建
|> plot line x=item y=total_qty title='Item Sales'
|> write png 'sales.png'
示例 8：SQL 审计
source csv 'sales.csv' with header true schema { date: date, item: string, qty: int, price: decimal(10,2) }
|> filter date >= '2025-01-01'
|> group by item aggregate sum(qty) as total_qty
|> explain sql
示例 9：多指标聚合与严格别名
source csv 'sales.csv' with header true schema { item: string, qty: int, price: decimal(10,2) }
|> group by item aggregate
sum(qty) as total_qty,
count(*) as orders,
sum(qty * price) as gm_value
|> derive avg_price = gm_value / nullif(total_qty, 0)
|> select item, total_qty, orders, avg_price
——
语法特性对比（优化前后对比分析）
-
管道写法
- 前：换行隐式或 '->' 混用，链路不统一。
- 后：统一使用 |>，左结合，语义清晰、可组合。
-
空值语义
- 前：null 与 'NA' 混用，无定义。
- 后：唯一空值常量 null；source 通过 null_as 将特定文本映射为 null；提供 =? 与 coalesce。
-
比较运算
- 前：'==' 与 '=' 混杂。
- 后：统一 '='；提供 '=?' 支持 null-safe 比较；'==' 明确报错并给出修复建议。
-
时间与时区
- 前：未声明，解析不确定。
- 后：set timezone + date_parse/to_timestamp；source 可覆盖 tz；时区传导至窗口与时间函数。
-
模式与类型
- 前：隐式列/类型，聚合命名冲突未防护。
- 后：schema 明确声明或推断记录；别名冲突静态检查，聚合别名必须唯一且不得与分组键冲突。
-
聚合与窗口
- 前：聚合语法与命名不一致，窗口缺失。
- 后：group by ... aggregate 一致体；窗口 over by/order/rows|range 标准化。
-
连接
- 前：join 语法冗余（kind left），列冲突策略不统一。
- 后：join [inner|left|...] on/using；with suffixes 控制冲突处理；限定名与默认后缀。
-
编码与 CSV
- 前：编码不明。
- 后：encoding 显式；delim/header/quote/escape/null_as 参数化。
-
审计与优化
- 前：无等效 SQL 输出。
- 后：explain sql/plan；validate 静态检查与提示。
——
使用注意事项和最佳实践建议
-
一致性
- 始终使用 |> 连接步骤；不要在同一管道内混合 select 与 derive 的别名依赖。
- 使用 derive 做派生与覆盖；select 只做投影与重命名。
-
空值处理
- 数据源读取时使用 null_as 把 'NA'、空串等统一映射为 null。
- 使用 coalesce 替换业务默认值；需要判等时优先使用 =? 代替复杂的 (a=b or (a is null and b is null))。
-
时间与时区
- 设定会话时区 set timezone；解析字符串时间时使用 to_date/to_timestamp，并指定格式。
- 跨时区数据需在 source 中指定 tz，或在 derive 中显式转换。
-
别名与冲突
- 聚合别名避免与 group key 相同；join 后列冲突请使用 with suffixes 或限定名。
- 在 select 中的别名不可被同一 select 后续表达式引用，使用 derive 拆分步骤。
-
性能与复现
- 重复使用的数据集使用 cache；实验使用 sample bernoulli ... seed ... repeatable 确保复现性。
- 聚合前尽量 filter 以减少数据量；窗口前确保 order 列类型正确且基数合理。
-
可审计性
- 对关键产出在上线前执行 explain sql/plan；保留生成的 SQL 以便审计与回溯。
- validate 检查通过作为流水线质量门槛。
——
扩展性设计和未来优化方向
-
连接器扩展
- 扩展 source/write 至 parquet, iceberg, delta, jdbc，多方言 SQL 直连；支持 credentials 管理与密钥引用。
-
类型与函数
- 增加 geography/geometry、json 类型；扩展时间窗口函数（session window）与时序函数（ASOF join）。
-
配置与模块化
- 引入 pipeline 级 options {...} 区块与可复用宏 macro，支持参数化与环境切换（dev/staging/prod）。
-
编译与优化
- 规则推送（predicate/aggregation pushdown）、列裁剪、跨管道 CSE（公共子表达式消除）。
- 代价估计与自适应 join 策略选择；统计信息捕获与缓存。
-
质量与监控
- 数据质量 DSL 子集（expect/validate），产出指标（缺失率、分布漂移）；内建数据血缘 lineage 与影响分析。
-
安全与多租户
- 行列级权限、受限 UDF、审计日志与可追踪 explain 结果签名。
-
可视化生态
- plot 统一语义导出到 vega-lite/ggplot 后端；主题与模板系统；可视化快照的元数据管理。
——
以上规范在保证语法一致性与逻辑严谨的前提下，统一了 DataFlow 的核心语义，强化了空值、时间、管道、聚合/窗口、连接与可审计性，适配分析师与数据工程师复杂场景的可读性与可维护性需求。
