NATS テーブルエンジン

• メッセージ subject をパブリッシュまたはサブスクライブする。
• 新しいメッセージが利用可能になり次第処理する。

​

テーブルの作成

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
(
    name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
    name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
    ...
) ENGINE = NATS SETTINGS
    nats_url = 'host:port',
    nats_subjects = 'subject1,subject2,...',
    nats_format = 'data_format'[,]
    [nats_schema = '',]
    [nats_num_consumers = N,]
    [nats_queue_group = 'group_name',]
    [nats_secure = false,]
    [nats_max_reconnect = N,]
    [nats_reconnect_wait = N,]
    [nats_server_list = 'host1:port1,host2:port2,...',]
    [nats_skip_broken_messages = N,]
    [nats_max_block_size = N,]
    [nats_flush_interval_ms = N,]
    [nats_username = 'user',]
    [nats_password = 'password',]
    [nats_token = 'clickhouse',]
    [nats_credentials = '-----BEGIN NATS USER JWT----- ...',]
    [nats_startup_connect_tries = 5,]
    [nats_max_rows_per_message = 1,]
    [nats_commit_on_select = false,]
    [nats_handle_error_mode = 'default']

• nats_url – host:port (例: localhost:4222) ..
• nats_subjects – NATS table がサブスクライブ/パブリッシュする subject の一覧。foo.*.bar や baz.> のようなワイルドカード subject をサポートします
• nats_format – メッセージのフォーマット。JSONEachRow など、SQL の FORMAT 関数と同じ記法を使用します。詳細は フォーマット セクションを参照してください。

• nats_schema – フォーマットでスキーマ定義が必要な場合に使用する必要があるパラメータです。たとえば、Cap’n Proto では、スキーマファイルへのパスとルート schema.capnp:Message オブジェクト名が必要です。
• nats_stream – NATS JetStream 内の既存の stream 名。
• nats_consumer_name – NATS JetStream 内の既存の durable pull コンシューマー名。
• nats_num_consumers – テーブルごとのコンシューマー数。デフォルト: 1。NATS core のみを使用していて、1 つのコンシューマーのスループットが不十分な場合は、より多くのコンシューマーを指定します。
• nats_queue_group – NATS subscriber の queue group 名。デフォルトはテーブル名です。
• nats_max_reconnect – 非推奨であり、効果はありません。再接続は nats_reconnect_wait タイムアウトで恒久的に実行されます。
• nats_reconnect_wait – 再接続試行のたびに待機する時間 (ミリ秒単位) 。デフォルト: 2000。
• nats_server_list - 接続先の server 一覧。NATS クラスターに接続するために指定できます。
• nats_skip_broken_messages - ブロックごとに許容する、スキーマ非互換の NATS メッセージ の数。デフォルト: 0。nats_skip_broken_messages = N の場合、このエンジンは解析できない N 件の NATS メッセージ をスキップします (1 メッセージ は 1 行のデータに相当します) 。
• nats_max_block_size - NATS からデータを flush するために poll で収集する行数。デフォルト: max_insert_block_size。
• nats_flush_interval_ms - NATS から読み取ったデータを flush するまでのタイムアウト。デフォルト: stream_flush_interval_ms。
• nats_wait_for_flush_interval - true の場合、background streaming cycle は、コンシューマー queue が空になるとすぐに終了するのではなく、flush interval 全体 (nats_flush_interval_ms、指定されていない場合は stream_flush_interval_ms) にわたって開いたままとなります。これにより、最大 1 flush interval 分の追加インジェストレイテンシーと引き換えに、より多くの メッセージ を単一のブロックに蓄積できます。デフォルト: false (低レイテンシーの drain-and-go 動作) 。
• nats_username - NATS username。サーバー設定ファイルで定義された named collection に保存されている場合、クエリでその collection の nats_url または nats_server_list を override することはできません。
• nats_password - NATS password。サーバー設定ファイルで定義された named collection に保存されている場合、クエリでその collection の nats_url または nats_server_list を override することはできません。
• nats_token - NATS auth token。サーバー設定ファイルで定義された named collection に保存されている場合、クエリでその collection の nats_url または nats_server_list を override することはできません。
• nats_credential_file - NATS credentials file への path。server が自身の権限で path を開くため、クエリによって nats_url と nats_server_list が override されない、サーバー設定ファイルで定義された named collection からのみ受け付けられます。クエリでは、代わりにファイルの内容を nats_credentials に渡します。
• nats_credentials - NATS credentials の内容 (user JWT と seed を含む .creds file と同じペイロード)。クエリで使用できる唯一の指定方法であるため、<nats_credential_file overridable="false"> により operator がその path をロックしている場合を除き、競合するのではなく named collection から継承した nats_credential_file を置き換えます。named collection が保持する credentials を削除するために空文字列を割り当てることはできません。
• nats_ca_file - NATS server certificate の検証に使用する、信頼された CA certificates を含む file への path。nats_secure が必要です。nats_credential_file と同様に、server が自身の権限で path を開くため、クエリによって nats_url と nats_server_list が override されない、サーバー設定ファイルで定義された named collection からのみ受け付けられます。
• nats_client_cert_file - NATS server に提示する client certificate への path。nats_secure および nats_client_key_file が必要です。nats_ca_file と同じソースから受け付けられます。
• nats_client_key_file - nats_client_cert_file の秘密鍵への path。nats_ca_file と同じソースから受け付けられます。
• nats_startup_connect_tries - 起動時の接続試行回数。デフォルト: 5。
• nats_max_rows_per_message — 行ベースのフォーマットで、1 つの NATS メッセージ に書き込まれる最大行数。 (デフォルト: 1) 。
• nats_commit_on_select - クエリ実行時に メッセージ を commit します。JetStream にのみ適用されます。core NATS には acknowledgement はありません。デフォルト: 0。
• nats_handle_error_mode — NATS エンジンでの error の処理方法。設定可能な値: default (メッセージ の解析に失敗すると exception が throw されます) 、stream (exception メッセージ と raw メッセージ が仮想カラム _error および _raw_message に保存されます) 。

CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1,subject2',
             nats_format = 'JSONEachRow';

INSERT INTO queue
SETTINGS stream_like_engine_insert_queue = 'subject2'
VALUES (1, 1);

CREATE TABLE queue (
    key UInt64,
    value UInt64,
    date DateTime
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';

<nats>
    <user>click</user>
    <password>house</password>
    <token>clickhouse</token>
</nats>

​

説明

1. engine を使用して NATS コンシューマーを作成し、それをデータストリームとして扱います。
2. 必要な structure を持つ table を作成します。
3. engine からのデータを変換し、あらかじめ作成した table に格納する materialized view を作成します。

CREATE TABLE queue (
    key UInt64,
    value UInt64
  ) ENGINE = NATS
    SETTINGS nats_url = 'localhost:4444',
             nats_subjects = 'subject1',
             nats_format = 'JSONEachRow',
             date_time_input_format = 'best_effort';

CREATE TABLE daily (key UInt64, value UInt64)
    ENGINE = MergeTree() ORDER BY key;

CREATE MATERIALIZED VIEW consumer TO daily
    AS SELECT key, value FROM queue;

SELECT key, value FROM daily ORDER BY key;

DETACH TABLE consumer;
ATTACH TABLE consumer;

​

仮想カラム

• _subject - NATS メッセージの subject。データ型: String。

• _raw_message - 正常にパースできなかった生のメッセージ。データ型: Nullable(String)。
• _error - パース失敗時に発生した例外メッセージ。データ型: Nullable(String)。

​

データフォーマットのサポート

• 行ベースのフォーマットでは、1 つの NATS メッセージに含める行数を nats_max_rows_per_message の設定で制御できます。
• ブロックベースのフォーマットでは、ブロックをより小さなパーツに分割することはできませんが、1 つのブロックに含まれる行数は一般設定の max_block_size で制御できます。

​

JetStream の使用

$ nats stream add
? Stream Name stream_name
? Subjects stream_subject
? Storage file
? Replication 1
? Retention Policy Limits
? Discard Policy Old
? Stream Messages Limit -1
? Per Subject Messages Limit -1
? Total Stream Size -1
? Message TTL -1
? Max Message Size -1
? Duplicate tracking time window 2m0s
? Allow message Roll-ups No
? Allow message deletion Yes
? Allow purging subjects or the entire stream Yes
Stream stream_name was created

Information for Stream stream_name created 2025-10-03 14:12:51

                Subjects: stream_subject
                Replicas: 1
                 Storage: File

Options:

               Retention: Limits
         Acknowledgments: true
          Discard Policy: Old
        Duplicate Window: 2m0s
              Direct Get: true
       Allows Msg Delete: true
            Allows Purge: true
Allows Per-Message TTL: false
          Allows Rollups: false

Limits:

        Maximum Messages: unlimited
     Maximum Per Subject: unlimited
           Maximum Bytes: unlimited
             Maximum Age: unlimited
    Maximum Message Size: unlimited
       Maximum Consumers: unlimited

State:

                Messages: 0
                   Bytes: 0 B
          First Sequence: 0
           Last Sequence: 0
        Active Consumers: 0

$ nats consumer add
? Select a Stream stream_name
? Consumer name consumer_name
? Delivery target (Pull Consumers の場合は空)
? Start policy (all, new, last, subject, 1h, msg sequence) all
? Acknowledgment policy explicit
? Replay policy instant
? Filter Stream by subjects (すべての場合は空)
? Maximum Allowed Deliveries -1
? Maximum Acknowledgments Pending 0
? Deliver headers only without bodies No
? Add a Retry Backoff Policy No
Information for Consumer stream_name > consumer_name created 2025-10-03T14:13:51+03:00

Configuration:

                    Name: consumer_name
               Pull Mode: true
          Deliver Policy: All
              Ack Policy: Explicit
                Ack Wait: 30.00s
           Replay Policy: Instant
         Max Ack Pending: 1,000
       Max Waiting Pulls: 512

State:

Last Delivered Message: Consumer sequence: 0 Stream sequence: 0
    Acknowledgment Floor: Consumer sequence: 0 Stream sequence: 0
        Outstanding Acks: 0 out of maximum 1,000
    Redelivered Messages: 0
    Unprocessed Messages: 0
           Waiting Pulls: 0 of maximum 512

CREATE TABLE nats_jet_stream (
    key UInt64,
    value UInt64
  ) ENGINE NATS
    SETTINGS  nats_url = 'localhost:4222',
              nats_stream = 'stream_name',
              nats_consumer_name = 'consumer_name',
              nats_subjects = 'stream_subject',
              nats_format = 'JSONEachRow';

​

データ耐久性