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-design-analysis.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

依赖装在默认 site 仓库以外的仓库时（例如 RAKULIB=inst#/path/to/repo），zef 的安装 预编译看不到那个仓库，会报 Could not find Terminal::UI / Could not find Terminal::ANSI （单独装 zef install Terminal::UI 也一样）。把发行版装进同一个仓库即可：

zef install --install-to=inst#/path/to/repo .

Windows 上 zef 解压 DB::DuckDB 0.9.1 时报 Malformed UTF-8 near bytes ...：该归档带中文 文件名，bsdtar 列目录时输出非法 UTF-8。手动解包后从目录安装该依赖，再装本发行版：

zef install <解包后的 DB::DuckDB 目录>
zef install .

依赖：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 query "<sql>" [--db PATH] [--limit N] [--json]
ducketl version

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

• --db 缺省时使用 pipeline 的 database，再缺省为 :memory:。

• history 默认 ~/.local/share/ducketl/history.duckdb，与数据库分开文件。

• run 的进度写 stderr，结果写 stdout；--json 输出完整运行结果。

• query 对数据数据库执行一条 SQL：select / with / values 最多取 --limit 行（默认 1000），show / describe / explain 原样执行，其余语句执行并报告影响行数。结果表格写 stdout（数值右对齐、NULL 原样），N rows in Xms 写 stderr；--json 输出 columns / rows / limited / affected / elapsed_ms。

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、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；宽度 ≥100 时显示左侧导航，≥84 时显示右侧运行面板，窗口缩放会重建布局。M1 为只读浏览与运行，M2 增加 Pipeline 屏的编辑闭环，M3 增加 --isolated worker 模式。

界面由 TUI 自己绘制：圆角边框 + 内联标题、truecolor 配色（NO_COLOR=1 或 DUCKETL_MONO=1 关闭）、聚焦面板边框高亮、整行反色的选择条；表格列宽按内容自适应、超宽用 … 截断；运行/步骤状态用颜色区分（ok 绿、failed 红、canceled 黄、running 青）。Terminal::UI 只提供备用屏与按键解析。

终端图表是纯文本：Dashboard 表格下方是最近运行的 rows/run、time/run 迷你折线（sparkline + 范围），Run 屏步骤表下方是按耗时等比缩放的条形图（步骤失败时变红）。实现见 lib/App/DuckETL/TUI/Chart，不依赖任何绘图库。

编辑草稿：

• 表单字段随步骤类型/格式变化：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/Repo         当前 repository 链 -> 子进程 -I 参数
lib/App/DuckETL/Store        history / watermarks（线程安全）
lib/App/DuckETL/CLI          run / tui / query / validate / list / show
lib/App/DuckETL/TUI          布局、四屏状态机、键位、表单与草稿
lib/App/DuckETL/TUI/Draft    草稿字段/枚举/序列化/校验辅助
lib/App/DuckETL/TUI/Render   纯渲染：宽度感知表格、状态行
lib/App/DuckETL/TUI/Chart    纯文本图表：sparkline、条形图、区间标签
lib/App/DuckETL/TUI/Theme    truecolor 调色板与着色（NO_COLOR 关闭）
lib/App/DuckETL/TUI/Region   面板矩形与内边距（自绘边框用）
examples/                    可直接运行的 pipeline 与样例数据
t/                           单元 + 集成测试

测试

prove6 -l t

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