risingwave

RisingWave

RisingWave 是一种流式 SQL 数据库，实现了 PostgreSQL wire 协议。 核心流水线为：Source（摄取）→ Materialized View（持续计算）→ Sink（输出）。

核心原则

1. 端口是 4566，而不是 5432 — RisingWave 在 4566 端口监听 SQL 连接。仪表板位于 5691。
2. SOURCE ≠ TABLE — CREATE SOURCE 用于连接数据流，但不会持久化数据。CREATE TABLE 会持久化数据。对于 CDC（Debezium、Maxwell、Canal），必须使用 CREATE TABLE ... FROM source。
3. Watermark 可启用窗口关闭 — 如果没有 WATERMARK FOR col AS col - INTERVAL '...'，EMIT ON WINDOW CLOSE 将无法工作；请在 source 或 table 定义中使用它。在 RisingWave 2.8 及更高版本中，TABLE 上的 watermark 要求使用 APPEND ONLY — 对于需要 watermark 的非仅追加流，请使用 CREATE SOURCE。
4. EMIT ON WINDOW CLOSE 与默认行为 — 默认的更新时发送会在每个检查点后发送部分结果。使用 EMIT ON WINDOW CLOSE 获取最终且不可变的窗口结果。
5. 大型回填需要 BACKGROUND_DDL — 如果不设置 SET BACKGROUND_DDL = true，在大型表上创建 MV 会造成阻塞。使用 SELECT * FROM rw_catalog.rw_ddl_progress 进行监控。
6. sink 上的 snapshot = false — 向现有 MV 添加 sink 时，如果没有此标志，所有历史数据都会被重放到 sink 中。
7. 核实文档 — RisingWave 发展迅速。如有疑问，请查看 docs.risingwave.com，或对现有对象执行 SHOW CREATE。

连接

# Default local connection
psql -h localhost -p 4566 -d dev -U root

# Docker (single-node)
docker run -it --pull=always -p 4566:4566 -p 5691:5691 \
  risingwavelabs/risingwave:latest single_node

参数	默认值
主机	localhost
端口	4566
数据库	dev
用户	root
密码	（无）

连接字符串：postgresql://root:@localhost:4566/dev

任何兼容 PostgreSQL 的客户端都可使用：psql、JDBC、psycopg2、pgx、SQLAlchemy、dbt、Grafana。

MCP 服务配置

RisingWave 提供官方 MCP 服务，包含 100 多种工具（查询执行、架构探索、流式作业监控、CDC 进度、Kafka 延迟、Hummock 存储分析）。

git clone https://github.com/risingwavelabs/risingwave-mcp.git
cd risingwave-mcp && pip install -r requirements.txt

通过环境变量进行配置：

RISINGWAVE_CONNECTION_STR=postgresql://root:@localhost:4566/dev

将其添加到 Agent 的 MCP 配置中（Claude Code：~/.claude/claude_desktop_config.json，VS Code：.vscode/mcp.json）：

{
  "mcpServers": {
    "risingwave": {
      "type": "stdio",
      "command": "python",
      "args": ["/path/to/risingwave-mcp/src/main.py"],
      "env": {
        "RISINGWAVE_CONNECTION_STR": "postgresql://root:@localhost:4566/dev"
      }
    }
  }
}

流水线模式

Stream data → SOURCE → MATERIALIZED VIEW(s) → SINK(s)
Static/CDC data → TABLE ─────────────────────────────╯

步骤 1：Source（流式，无持久化）

CREATE SOURCE user_events (
    user_id     INT,
    action      VARCHAR,
    event_time  TIMESTAMP,
    WATERMARK FOR event_time AS event_time - INTERVAL '5 SECOND'
)
WITH (
    connector = 'kafka',
    topic = 'user-events',
    properties.bootstrap.server = 'kafka:9092',
    scan.startup.mode = 'latest'
)
FORMAT PLAIN ENCODE JSON;

步骤 2：Materialized View（持续计算）

-- Windowed aggregation with final results on window close
CREATE MATERIALIZED VIEW active_users_per_minute AS
SELECT
    action,
    COUNT(DISTINCT user_id) AS unique_users,
    window_start,
    window_end
FROM TUMBLE(user_events, event_time, INTERVAL '1 MINUTE')
GROUP BY action, window_start, window_end
EMIT ON WINDOW CLOSE;

-- Plain aggregation (emit on every update)
CREATE MATERIALIZED VIEW user_action_counts AS
SELECT user_id, action, COUNT(*) AS cnt
FROM user_events
GROUP BY user_id, action;

步骤 3：Sink（输出）

CREATE SINK alerts_to_kafka FROM active_users_per_minute
WITH (
    connector = 'kafka',
    topic = 'user-alerts',
    properties.bootstrap.server = 'kafka:9092',
    snapshot = false   -- skip historical backfill
)
FORMAT PLAIN ENCODE JSON;

CDC 模式（数据库复制）

CDC 需要两步设置：共享连接 source + 每张表对应一个 TABLE。

-- Step 1: shared CDC source connection
CREATE SOURCE pg_cdc WITH (
    connector = 'postgres-cdc',
    hostname = 'postgres-host',
    port = '5432',
    username = 'replicator',
    password = '<your-password>',
    database.name = 'mydb',
    slot.name = 'rw_slot'
);

-- Step 2: per-table ingestion (TABLE, not SOURCE)
CREATE TABLE orders (
    id          INT PRIMARY KEY,
    customer_id INT,
    total       DECIMAL,
    created_at  TIMESTAMP
)
FROM pg_cdc TABLE 'public.orders';

时间窗口

所有三种窗口类型都会添加 window_start 和 window_end 列。

-- TUMBLE: non-overlapping fixed windows
FROM TUMBLE(table, time_col, INTERVAL '5 MINUTES')

-- HOP (sliding): overlapping windows
-- hop_size = slide interval, window_size = total duration
FROM HOP(table, time_col, INTERVAL '1 MINUTE', INTERVAL '5 MINUTES')

**注意：**在 RisingWave 2.x 中，SESSION 窗口仅支持批处理模式。对于流式处理，请使用 TUMBLE 或 HOP。

**模式：**始终按 window_start, window_end 分组，并在使用 watermarks 时添加 EMIT ON WINDOW CLOSE。

实用系统目录查询

-- All materialized views with definitions
SELECT name, definition FROM rw_catalog.rw_materialized_views;

-- DDL progress during MV creation / backfill
SELECT ddl_id, ddl_statement, progress FROM rw_catalog.rw_ddl_progress;

-- Active sources and their connectors
SELECT name, connector FROM rw_catalog.rw_sources;

-- Sink info
SELECT name, sink_type, connector FROM rw_catalog.rw_sinks;

-- CDC backfill progress
SELECT job_id, split_total_count, split_backfilled_count, split_completed_count
FROM rw_catalog.rw_cdc_progress;

-- Cluster nodes
SELECT id, host, type, state FROM rw_catalog.rw_worker_nodes;

-- Inspect object definition
SHOW CREATE MATERIALIZED VIEW my_mv;
SHOW CREATE SOURCE my_source;
SHOW CREATE SINK my_sink;

实用会话设置

-- Large backfills: non-blocking DDL
SET BACKGROUND_DDL = true;

-- Share Kafka source across multiple MVs (v2.1+)
SET streaming_use_shared_source = true;

故障排除

MCP 服务无法连接

• 确认已设置 RISINGWAVE_CONNECTION_STR，且 RisingWave 正在端口 4566 上运行
• 先测试连接：psql -h localhost -p 4566 -d dev -U root
• 检查 Python 版本：python --version（需要 Python 3.8+）

EMIT ON WINDOW CLOSE 未产生任何输出

• 必须为源或表定义 WATERMARK FOR col AS col - INTERVAL '...'
• 确认水位线正在推进：插入带有近期时间戳的行，而不是历史数据
• 检查 MV 定义：SHOW CREATE MATERIALIZED VIEW my_mv

MV 创建卡住 / 会话超时

• 在 CREATE MATERIALIZED VIEW 之前使用 SET BACKGROUND_DDL = true
• 监控进度：SELECT ddl_id, ddl_statement, progress FROM rw_catalog.rw_ddl_progress

CDC 表未接收更新

• 确认 PostgreSQL 设置了 wal_level = logical，且复制用户具有 REPLICATION 角色
• 检查 CDC 延迟：SELECT * FROM rw_catalog.rw_cdc_progress
• 确保 CREATE SOURCE 中的 slot.name 唯一，且上游数据库中不存在同名项

Sink 发送重复的历史数据

• 在 Sink 的 WITH 子句中添加 snapshot = false，跳过对现有 MV 数据的回填

参考资料

• 连接器参考 — 所有受支持的源和 Sink
• 系统目录参考 — 完整的 rw_catalog 表列表
• RisingWave 文档
• RisingWave MCP 服务