Timeplus 流式 SQL 指南

你是一位 Timeplus 专家 —— 一个基于流式 SQL 引擎（Proton）构建的高性能实时流式分析平台。你能够编写正确且高效的 Timeplus SQL，并通过兼容 ClickHouse 的 HTTP API 执行。

快速参考

执行 SQL

环境配置

始终使用以下环境变量 —— 切勿硬编码凭证：

- TIMEPLUS_HOST       # 主机名或 IP 地址
- TIMEPLUS_USER       # 用户名
- TIMEPLUS_PASSWORD   # 密码（可为空）

通过 curl 执行 SQL（端口 8123）

端口 8123 是兼容 ClickHouse 的 HTTP 接口，用于执行 所有 DDL 和历史查询（如 CREATE、DROP、INSERT、SELECT from table(...)）。始终使用 -u 选项携带用户名和密码。

注意：如果 curl 返回空内容，不代表出错，仅表示查询无返回记录。可通过检查 HTTP 状态码确认结果（200 表示成功，4xx/5xx 表示失败）。

# 标准模式 —— 将 SQL 内容通过管道传给 curl
echo "YOUR SQL HERE" | curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

健康检查：

curl "http://${TIMEPLUS_HOST}:8123/"
# 返回: Ok.

DDL 示例 —— 创建数据流：

echo "CREATE STREAM IF NOT EXISTS sensor_data (
  device_id string,
  temperature float32,
  ts datetime64(3, 'UTC') DEFAULT now64(3, 'UTC')
) SETTINGS logstore_retention_ms=86400000" | \
curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

历史查询（JSON 输出格式）：

echo "SELECT * FROM table(sensor_data) LIMIT 10" | \
curl "http://${TIMEPLUS_HOST}:8123/?default_format=JSONEachRow" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

插入数据：

echo "INSERT INTO sensor_data (device_id, temperature) VALUES ('dev-1', 23.5), ('dev-2', 18.2)" | \
curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

通过 REST API 实现流式数据导入（端口 3218）

用于向数据流批量推送事件：

curl -s -X POST "http://${TIMEPLUS_HOST}:3218/proton/v1/ingest/streams/sensor_data" \
  -H "Content-Type: application/json" \
  -d '{
    "columns": ["device_id", "temperature"],
    "data": [
      ["dev-1", 23.5],
      ["dev-2", 18.2],
      ["dev-3", 31.0]
    ]
  }'

输出格式

在 URL 后追加 ?default_format=<format> 来指定输出格式：

核心概念

流式查询 vs 历史查询

-- 流式查询：持续运行，永不结束。默认行为。
SELECT device_id, temperature FROM sensor_data;

-- 历史查询：有界，立即返回结果。使用 table() 包裹。
SELECT device_id, temperature FROM table(sensor_data) LIMIT 100;

-- 历史 + 未来：包含所有过去事件和未来事件
SELECT * FROM sensor_data WHERE _tp_time >= earliest_timestamp();

流式查询中的 ORDER BY 与 LIMIT

流式查询的结果是无界的——新数据会持续到达，因此传统批处理 SQL 中的 ORDER BY / LIMIT 语义不直接适用：

• **非窗口化的流式查询不允许使用 ORDER BY**。排序需要有限输入。只能在上游为有界数据时使用（如 table(...) 查询，或带窗口聚合的 tumble/hop/session），每个窗口关闭后生成有限批次并可排序。
• **LIMIT n 在流式查询中表示在收到 n 条记录后终止查询**。这不是“按某顺序取前 n 条”，而是强制停止。建议搭配窗口使用（如 LIMIT n BY window_start），或使用流式原生的 top-k 聚合函数（如 top_k、min_k、max_k）实现每窗口排名。
• 对于历史查询（SELECT ... FROM table(stream)），ORDER BY 和 LIMIT 按常规方式工作。

详见 references/TRANSFORMATIONS.md → 每窗口 Top-N 获取推荐模式。

_tp_time 字段

每个数据流都内置一个 datetime64(3, 'UTC') 类型的 _tp_time 事件时间字段，其默认值为数据摄入时间。可通过在创建流时设置 SETTINGS event_time_column='your_column' 来指定自定义事件时间列。

数据流模式

常用模式

模式 1：创建流 → 插入数据 → 查询

# 1. 创建数据流
echo "CREATE STREAM IF NOT EXISTS orders (
  order_id string,
  product string,
  amount float32,
  region string
)" | curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

# 2. 插入数据
echo "INSERT INTO orders VALUES ('o-1','Widget',19.99,'US'), ('o-2','Gadget',49.99,'EU')" | \
  curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

# 3. 查询历史数据
echo "SELECT region, sum(amount) FROM table(orders) GROUP BY region" | \
  curl "http://${TIMEPLUS_HOST}:8123/?default_format=JSONEachRow" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

模式 2：窗口聚合（流式）

echo "SELECT window_start, region, sum(amount) AS revenue
FROM tumble(orders, 1m)
GROUP BY window_start, region
EMIT AFTER WATERMARK AND DELAY 5s" | \
  curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

模式 3：物化视图流水线

echo "CREATE MATERIALIZED VIEW IF NOT EXISTS mv_revenue_by_region
INTO revenue_by_region AS
SELECT window_start, region, sum(amount) AS total
FROM tumble(orders, 5m)
GROUP BY window_start, region" | \
  curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

模式 4：用于测试的随机数据流

echo "CREATE RANDOM STREAM IF NOT EXISTS mock_sensors (
  device_id string DEFAULT 'device-' || to_string(rand() % 10),
  temperature float32 DEFAULT 20 + (rand() % 30),
  status string DEFAULT ['ok','warn','error'][rand() % 3 + 1]
) SETTINGS eps=5" | \
  curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

错误处理

常见错误及解决方案：

查看流结构：

echo "DESCRIBE sensor_data" | curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

列出所有数据流：

echo "SHOW STREAMS" | curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

解释查询计划：

echo "EXPLAIN SELECT * FROM tumble(sensor_data, 1m) GROUP BY window_start" | \
  curl "http://${TIMEPLUS_HOST}:8123/" \
  -u "${TIMEPLUS_USER}:${TIMEPLUS_PASSWORD}" \
  --data-binary @-

何时查阅参考文件

当用户需求涉及更深入知识时，请加载对应的参考文件：

• 创建或修改 数据流、外部流、数据源 → references/INGESTION.md
• 窗口函数、JOIN、CTE、物化视图、聚合操作 → references/TRANSFORMATIONS.md
• 目标输出、外部表、Kafka 输出、Webhook → references/SINKS.md
• 数据类型、完整函数目录、查询设置、所有 DDL → references/SQL_REFERENCE.md
• 模拟数据、随机流、测试数据生成 → references/RANDOM_STREAMS.md
• 编写 Python UDF、JavaScript UDF、远程 UDF、SQL Lambda → references/UDFS.md
• Python 表函数 → references/Python_TABLE_FUNCTION.md
• **EMIT 子句术语、EMIT AFTER SESSION CLOSE / KEY EXPIRE、事件终止的会话窗口、UDAF has_customized_emit、MAXSPAN/TIMEOUT 约束** → references/EMIT_AND_UDAF.md
• 定时任务 → references/TASK.md
• 告警规则 → references/ALERT.md