App::DuckETL

SQL-first ETL on DuckDB. Raku only plans SQL and renders results; rows stay inside DuckDB.

当前进度：M0 引擎 + M1 只读 TUI + M2 编辑闭环 + M3 生产化（ducketl run/tui/worker、增量水位、Parquet 分区写、--isolated worker、DuckDB 扩展装载、Pipeline 表单 / $EDITOR / 保存 JSON、history、失败回滚、原子落盘、运行取消）。设计依据见 TUI-ETL-设计分析.md。

快速开始

raku -Ilib bin/ducketl validate examples/pipelines/nightly_orders.json
raku -Ilib bin/ducketl run examples/pipelines/nightly_orders.json
# => examples/out/orders.parquet

raku -Ilib bin/ducketl run examples/pipelines/orders_table.json
# => examples/out/ducketl.duckdb, mart.orders（upsert）

raku -Ilib bin/ducketl tui examples/pipelines/nightly_orders.json
# => 全屏 TUI：Dashboard / Sources / Pipeline / Run

raku -Ilib bin/ducketl run examples/pipelines/orders_incremental.json
raku -Ilib bin/ducketl run examples/pipelines/orders_incremental.json
# => 第二次按 created_at 水位追加 0 行

raku -Ilib bin/ducketl run examples/pipelines/partitioned_orders.json
# => examples/out/orders_by_country/country=*/data_0.parquet

ducketl extensions                       # 列出 DuckDB 扩展
ducketl tui --isolated pipeline.json     # worker 进程独占数据库

用 zef 安装：

zef install .
ducketl run pipeline.json
ducketl tui pipeline.json

依赖：Rakudo 2026.09、DB::DuckDB 0.8.3、Alien::Libduckdb 1.1.0（DuckDB 2.x）、JSON::Fast、Terminal::UI 0.1.14（仅 TUI）。

命令

ducketl run <pipeline.json> [--db PATH] [--history PATH]
                            [--threads N] [--memory-limit SIZE]
                            [--keep-temp] [--json] [--quiet]
ducketl tui [pipeline.json] [--db PATH] [--history PATH]
                            [--threads N] [--memory-limit SIZE] [--isolated]
ducketl worker <pipeline.json> [--db PATH] [--history PATH]
                            [--threads N] [--memory-limit SIZE] [--keep-temp]
ducketl extensions [--json]
ducketl validate <pipeline.json>
ducketl list [--history PATH] [--limit N] [--json]
ducketl show <run-id> [--history PATH] [--json]
ducketl version

worker 是 tui --isolated 使用的子进程入口：stdout 每行一个 JSON 事件，stdin 收到 cancel 时取消当前运行。

--db 缺省时使用 pipeline 的 database，再缺省为 :memory:。
history 默认 ~/.local/share/ducketl/history.duckdb，与数据库分开文件。
run 的进度写 stderr，结果写 stdout；--json 输出完整运行结果。

Pipeline 格式

canonical 格式是 JSON。路径相对 pipeline 文件所在目录解析。

{
  "name": "nightly_orders",
  "database": "../out/ducketl.duckdb",
  "defaults": { "threads": 4, "memory_limit": "1GB" },
  "steps": [
    {
      "id": "extract",
      "type": "source",
      "format": "csv",
      "path": "../data/orders.csv",
      "options": { "header": true, "delim": ",", "sample_size": 100000 }
    },
    {
      "id": "clean",
      "type": "sql",
      "sql": "select id, cast(amount as decimal(18,2)) as amount from extract where amount is not null"
    },
    {
      "id": "check",
      "type": "assert",
      "sql": "select count(*) from clean where amount < 0",
      "expect": 0
    },
    {
      "id": "load",
      "type": "sink",
      "target": "mart.orders",
      "mode": "upsert",
      "key": ["id"]
    }
  ]
}

步骤类型

类型	作用	关键字段
`source`	CSV / Parquet / JSON / 表 / SQL	`format`、`path`、`table`、`sql`、`options`
`sql`	任意 SELECT	`sql`、`materialize`
`assert`	数据质量断言	`sql`、`expect`、`op`（`eq/ne/lt/lte/gt/gte`）
`sink`	表 / Parquet / CSV / 分区 Parquet 目录	`target`、`mode`（`replace/append/upsert`）、`key`、`watermark`、`partition_by`

source 支持 csv、parquet、json、table、sql。
- 文件路径作为绑定参数传给 read_csv_auto(?) / read_parquet(?) / read_json_auto(?)。
- path 也可以是路径列表。
sql / assert 用步骤 id 引用前一步的数据：from extract、join customers。只有 source 与 sql 步骤产出数据；引用尚未产出或非数据步骤会校验失败。
中间结果默认物化为临时表。"materialize": false 改为临时视图（DuckDB 会把视图内联进下游查询），适合把整条链压成一个查询。
sink 的 target 以 .parquet / .csv 结尾时写文件，否则写表（table 或 schema.table）。
- 文件先写同目录 staged 文件，成功提交后再 rename；失败不会碰目标文件。
- upsert 需要 key；首次运行会建表并创建唯一索引，后续按 key 更新。
- append 在目标不存在时自动建表。
表 sink 在事务内执行；任一步失败都会回滚，目标表回到运行前状态。

增量水位、分区写与扩展

{
  "name": "incremental",
  "extensions": ["parquet"],
  "steps": [
    { "id": "extract", "type": "source", "format": "csv", "path": "raw/events.csv" },
    { "id": "load", "type": "sink", "target": "mart.events",
      "mode": "append", "watermark": "created_at" }
  ]
}

watermark 只用于表 sink 的 append / upsert：首次运行全量，之后自动加 where <column> > <last value>；值存在 history 的 _etl.watermarks （pipeline + step_id + column_name）。水位只在数据 commit（以及文件安装）成功后更新，失败重跑不会跳过数据。重复运行同一份输入插入 0 行。
partition_by 用于 Parquet 分区目录：target 是目录路径（含 /、以 ~ 或 . 开头），mode 只能是 replace。支持 options.compression（默认 zstd）。写入先落到 <target>.tmp-…，commit 后整体替换目录；读取用 read_parquet('<target>/**/*.parquet')。重复运行是整目录替换，不会累加重复行。
extensions 是顶层字符串数组（如 ["parquet"]、["json"]），运行前在 job 连接上执行 load <name>；名称做字符白名单校验。ducketl extensions 列出 DuckDB 已安装/已加载的扩展。远程源（httpfs/sqlite/postgres/excel）仍需先自行 INSTALL 对应 DuckDB 扩展，项目不内置连接器。

TUI（M1 + M2 + M3）

ducketl tui 打开工作台。最小 50×10；宽度 ≥110 时显示左侧导航，≥80 时显示右侧运行面板，窗口缩放会重建布局。M1 为只读浏览与运行，M2 增加 Pipeline 屏的编辑闭环，M3 增加 --isolated worker 模式。

屏	内容
Dashboard	最近 20 次运行（来自 history）；`Enter` 查看步骤、行数与错误
Sources	当前数据库 catalog，加上当前 pipeline 的 `source` 步骤；`Enter`/`p` 预览（`LIMIT 100`）
Pipeline	可编辑草稿：步骤表 + 字段表单；`n` 新建、`Enter` 打开文件、`a` 加步骤、`s` 保存
Run	步骤状态、行数、耗时、日志；`Enter` 查看单步 SQL/错误，`Esc` 返回

键	作用
`1` `2` `3` `4`	Dashboard / Sources / Run / Pipeline
`Tab` / `Shift-Tab`	在导航、内容、详情面板间切换
`j` `k` / `↑` `↓`	移动选择；表单中移动字段
`Enter`	激活选中项；Pipeline 中编辑当前字段（枚举/布尔值直接切换）
`a` / `d`	追加步骤 / 删除选中步骤（二次确认）
`e`	用 `$VISUAL` / `$EDITOR` 编辑选中步骤的 SQL，返回后恢复界面
`s`	保存草稿；首次保存会提示路径（默认 `<cwd>/pipelines/<name>.json`）
`o` / `n`	打开选中 pipeline 文件 / 新建草稿（有未保存改动时确认）
`r` / `F2`	运行当前 pipeline；在 Pipeline 屏运行草稿（无需先保存）
`c` / `F3`	取消当前运行
`p`	预览 Sources 中选中的表或 source
`[` `]`	切换 pipeline（同时刷新 catalog）
`R`	重新加载 pipeline、catalog、history
`q`	退出；有运行中的任务时先取消并等待回滚，有未保存草稿时确认

编辑草稿：

表单字段随步骤类型/格式变化：source 的 path/table/sql 互斥展示，sink 的 key 只在 upsert 下出现。
文本字段用底部单行编辑器（插入、删除、←/→、Home/End、Enter 提交、Esc 取消）；options 以 JSON 对象文本编辑；expect 只接受整数。
保存前用临时文件走一遍 Pipeline.load 校验（id、引用、类型、必填项），校验通过才原子 rename 到目标路径；失败不会覆盖已有文件。
草稿标题带 * 表示未保存。未保存草稿运行或保存时，相对路径基准为当前工作目录；保存后基准变为 JSON 文件所在目录。
$EDITOR 为空或未设置时回退到 vi；外部编辑器退出码非 0 时 SQL 保持不变。

线程与连接：

UI 线程只渲染快照和读按键；run 在后台线程执行，事件经队列交给 UI。
数据 Duck 同时持有 job connection 与 session connection：catalog/预览走 session，不与运行事务抢临时表。
运行取消通过驱动已绑定的 duckdb_interrupt 打断 job connection；事务回滚、staged 文件删除、history 记为 canceled。
--db 优先于 pipeline 的 database，再回退到配置或 :memory:；threads / memory_limit 取 CLI、pipeline defaults、配置中的第一个可用值。
catalog 不会为了浏览而创建尚不存在的数据库文件。

隔离模式（ducketl tui --isolated）：

每次运行启动一个 ducketl worker 子进程；数据库与 history 由 worker 独占，TUI 只消费 JSONL 事件。
运行期间 TUI 关闭自己的 history 连接，结束后自动重开、刷新 Dashboard；Catalog 与预览禁用。
草稿必须先保存（worker 只接收 pipeline 文件）；history 必须是文件，不能用 :memory:。
c 通过 worker stdin 发送 cancel；worker 被 kill -9 或崩溃时 TUI 继续工作，运行标为 failed，未完成步骤标为失败。
worker 的 stdout/stderr 是独立管道；异常不会写入 TUI 的终端。

运行语义

单进程：一个数据 DB::DuckDB 实例；run 使用一条 job connection，TUI 另有一条 session connection；history 用独立 .duckdb 文件，避免跨库锁。
数据不过 Raku：只执行 SQL 和 COUNT(*)；没有逐行读取。
事务：source/sql/表 sink 在 run 级事务中；临时表、目标表、唯一索引一起提交或回滚。
文件原子性：staged 文件 + rename。
history：_etl.runs、_etl.steps、_etl.watermarks。水位只在对应 step 成功且 commit/文件安装完成后更新。
扩展：pipeline 的 extensions 在事务开始前执行 load；失败即整次运行失败，不写目标。
分区目录：COPY ... PARTITION_BY 先写 <target>.tmp-*，成功后整目录替换；失败清理 staging，旧目录保留。
隔离 worker：worker 在独立进程中打开数据库与 history，TUI 不打开数据库；worker 退出后才重新打开 history。
失败可见：失败步骤的 SQL 与错误写入 history；ducketl show <run-id> 查看。

当前边界

TUI 编辑是单草稿、线性步骤：没有拖拽排序、多选、撤销栈；步骤顺序用删除/追加调整（M2 范围）。
options 在表单里以 JSON 文本编辑，没有逐项控件；长 SQL 用 $EDITOR，内联输入是单行编辑器。
运行中不显示精确百分比；只显示已完成的步骤、行数与耗时（设计文档明确反对伪造进度）。
不做调度、多租户、远程编排、通用 DAG。
增量水位只支持表 sink 的 append/upsert；replace 与文件 sink 不支持水位。
分区写只支持 Parquet 目录 + replace；目录 SQL 需用 read_parquet('.../**/*.parquet')。
远程 HTTP / SQLite / Postgres / Excel 源需要自行安装对应 DuckDB 扩展；项目只提供 extensions 装载入口与 ducketl extensions 探测，不内置连接器。
隔离模式一次一个 worker，运行期间不能浏览 catalog/预览；草稿需先保存。
表名限制为 [A-Za-z_][A-Za-z0-9_]* 的 table 或 schema.table；其他标识符暂不支持。
步骤 id 在 SQL 的表位置具有保留含义；引用当前或未来步骤会报错。

bin/ducketl                  CLI 入口
lib/App/DuckETL/Pipeline     模型、校验、步骤引用检查
lib/App/DuckETL/SQL          quote/literal、步骤引用扫描与重写
lib/App/DuckETL/Planner      步骤 -> SQL（纯函数）
lib/App/DuckETL/Duck         连接、catalog、schema、预览
lib/App/DuckETL/Executor     事务、事件、取消点、水位、分区安装、失败清理
lib/App/DuckETL/Run          后台运行、事件队列、取消
lib/App/DuckETL/IsolatedRun  子进程 worker、JSONL 事件、取消/崩溃处理
lib/App/DuckETL/Store        history / watermarks（线程安全）
lib/App/DuckETL/CLI          run / tui / validate / list / show
lib/App/DuckETL/TUI          布局、四屏状态机、键位、表单与草稿
lib/App/DuckETL/TUI/Draft    草稿字段/枚举/序列化/校验辅助
lib/App/DuckETL/TUI/Render   纯渲染：宽度感知表格、状态行
examples/                    可直接运行的 pipeline 与样例数据
t/                           单元 + 集成测试

测试

prove6 -l t

覆盖：SQL 引用扫描（函数/字符串/注释/限定名）、pipeline 校验、Planner golden SQL、CSV→Parquet、upsert 幂等、失败回滚、文件原子性、CLI、TUI 纯渲染（含 CJK 列宽）、后台运行事件与取消、草稿字段/枚举/序列化与保存后往返、增量水位幂等、分区目录替换、扩展装载、隔离 worker 事件/取消/崩溃。

本机验收：

M0：1,000,000 行 CSV → 过滤/转型 → Parquet，约 346 ms（Rakudo 2026.09、DuckDB 2.0.0-dev1、WSL Ubuntu；含启动与 history 写入）。
M1：同一 1M 行 pipeline 运行期间切换 Dashboard/Sources/Run 不阻塞，约 277 ms 完成；慢查询运行中按 c 取消，UI 在 0.75 s 内显示 canceled，history 记为 canceled，目标文件无残留。
M2：在 TUI 从零新建草稿，用表单建 source → sql → sink 三步，并用 $EDITOR 改写 SQL；保存 JSON 后运行，源 3 行 → 过滤 2 行 → Parquet 2 行（sum=50）。再加 assert 步骤后保存并重跑，4 步全部 ok。
M3：带水位的 append 连续运行插入 3 → 0 → 1 行（新增一行后），目标表 4 行、水位推进到 2026-01-04；分区写生成 country=UK / country=US 两个目录，重跑替换后仍为 3 行且无 staging 残留；extensions: ["parquet"] 成功，未知扩展使运行失败且不写目标。--isolated 下 worker 正常运行、stdin cancel 约 0.16 s 生效；kill -9 worker 后 TUI 保持可操作、运行标为 failed，退出码 0。