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 OK 表示成功；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= 参数：

核心概念

流式查询 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 列

每个流都内置一个 _tp_time datetime64(3, 'UTC') 事件时间列，默认值为数据摄入时间。创建流时可通过 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
• 输出目标（Sinks）、外部表、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