risingwave

RisingWave

RisingWave は PostgreSQL ワイヤープロトコルを実装したストリーミング SQL データベースです。 基本的なパイプラインは、ソース（取り込み）→ マテリアライズドビュー（継続的な計算）→ シンク（出力）です。

基本原則

1. ポートは 5432 ではなく 4566 — RisingWave は SQL 接続用に 4566 で待ち受けます。ダッシュボードは 5691 です。
2. SOURCE ≠ TABLE — CREATE SOURCE はストリームに接続しますが、データは永続化しません。CREATE TABLE はデータを永続化します。CDC（Debezium、Maxwell、Canal）では、必ず CREATE TABLE ... FROM source を使用してください。
3. ウォーターマークでウィンドウを閉じられるようになります — WATERMARK FOR col AS col - INTERVAL '...' がないと、EMIT ON WINDOW CLOSE は動作しません。ソースまたはテーブルの定義に指定してください。RisingWave 2.8 以降では、TABLE のウォーターマークには APPEND ONLY が必要です。ウォーターマークが必要な非 append-only ストリームには、CREATE SOURCE を使用してください。
4. EMIT ON WINDOW CLOSE とデフォルトの動作の違い — デフォルトの更新時出力では、各チェックポイントの後に部分的な結果を送信します。確定済みで変更されないウィンドウ結果には、EMIT ON WINDOW CLOSE を使用してください。
5. 大量のバックフィルには BACKGROUND_DDL が必要です — 大きなテーブル上に MV を作成すると、SET BACKGROUND_DDL = true がない場合は処理がブロックされます。SELECT * FROM rw_catalog.rw_ddl_progress で進捗を監視してください。
6. シンクでは snapshot = false を指定します — このフラグを指定せずに既存の MV にシンクを追加すると、過去のデータがすべてシンクに再送されます。
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 には、100 以上のツール（クエリ実行、スキーマの調査、ストリーミングジョブの監視、CDC の進捗、Kafka のラグ、Hummock ストレージ分析）を備えた公式 MCP サーバーがあります。

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

エージェントの 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：ソース（ストリーミング、永続化なし）

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：マテリアライズドビュー（継続的な計算）

-- 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：シンク（出力）

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 のセットアップには、共有接続ソースとテーブルごとの TABLE の2 つのステップが必要です。

-- 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';

時間ウィンドウ

3 種類すべてのウィンドウで、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')

注： SESSION ウィンドウは RisingWave 2.x のバッチモードでのみサポートされています。ストリーミングでは TUMBLE または HOP を使用してください。

パターン： 常に window_start, window_end でグループ化し、ウォーターマークを使用する場合は 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 が一意で、アップストリーム DB にまだ存在していないことを確認してください

シンクから過去のデータが重複して送信される

• 既存の MV データのバックフィルをスキップするには、シンクの WITH 句に snapshot = false を追加してください

参考資料

• コネクターのリファレンス — サポートされているすべてのソースとシンク
• システムカタログのリファレンス — rw_catalog テーブルの完全な一覧
• RisingWave のドキュメント
• RisingWave MCP サーバー