PolarDB:列ストアインデックスによる時系列データ分析の高速化

ワークフロー

1. データ書き込み：ビジネスアプリケーションが時系列データ (トランザクションレコードなど) を PolarDB for PostgreSQL または クラスターに書き込みます。

2. 列ストアインデックス：ベーステーブルに列ストアインデックスを作成します。 PolarDB for PostgreSQL または は、ローストアと並行して列形式のデータを自動的に維持します。 行ストレージと比較して、列ストレージはデータを列単位で編成するため、より高い圧縮率を実現し、集計クエリの実行時には関連する列のみを読み取ることで I/O を削減します。

3. クエリの高速化：分析クエリ (ローソク足の集計など) は、オプティマイザまたは明示的な ヒント を通じて、優先的に列ストアインデックスにルーティングされます。 クエリエンジンは列ストレージと並列処理を使用してデータをスキャンおよび集計し、結果を返します。

ソリューションのメリット

• 使いやすさ：ビジネスのリファクタリングや複雑な ETL は不要です。 ベーステーブルに列ストアインデックスを作成するだけで、分析クエリが透過的に高速化されます。

• 豊富な機能：パーティションテーブルをネイティブにサポートし、 time_bucket、 first、 last などの豊富な時系列分析関数を搭載しているため、SQL 開発を簡素化できます。

ステップ 1: 環境の準備

1. クラスターのバージョンと設定が次の要件を満たしていることを確認します。

   • クラスターのバージョン：

     • PostgreSQL 14 (マイナーエンジンバージョン 2.0.14.10.20.0 以降)

     • PostgreSQL 15 (マイナーエンジンバージョン 2.0.15.15.7.0 以降)

     • PostgreSQL 16 (マイナーエンジンバージョン 2.0.16.8.3.0 以降)

     • PostgreSQL 17 (マイナーエンジンバージョン 2.0.17.7.5.0 以降)

     説明

     コンソールで、または SHOW polardb_version; ステートメントを実行してマイナーエンジンバージョンを確認できます。マイナーエンジンバージョンが要件を満たさない場合は、マイナーエンジンバージョンをアップグレードしてください。

   • wal_level パラメーターを logical に設定する必要があります。この設定は、ロジカルデコーディングに必要な情報を先行書き込みログ (WAL) に追加します。

     説明

     コンソールでwal_level パラメーターを設定することができます。このパラメーターを変更するとクラスターが再起動されるため、ビジネスへの影響を考慮し、慎重に操作を行ってください。

   • ソーステーブルにはプライマリキーが必要で、列ストアインデックスの作成時にはそのプライマリキー列を含める必要があります。プライマリキーに SERIAL または BIGSERIAL データ型を使用すると、データ同期の効率が大幅に向上するため、推奨します。

   • 各テーブルに作成できる列ストアインデックスは 1 つだけです。

2. 列ストアインデックス機能を有効にします。

   IMCI を有効化する方法は、PolarDB for PostgreSQL または クラスターのマイナーエンジンバージョンによって異なります。

   PostgreSQL 16 (2.0.16.9.8.0 以降) または PostgreSQL 14 (2.0.14.17.35.0 以降)

   これらのバージョンの PolarDB for PostgreSQL または クラスターでは、2 つの方法が利用できます。以下の表に違いを示します。

   比較項目	[推奨] IMCI 読み取り専用ノードの追加	事前インストール済みの列ストアインデックス拡張機能を直接使用
   方法	コンソールで IMCI 読み取り専用ノードを手動で追加できます。	操作は不要です。拡張機能を直接使用できます。
   リソース割り当て	列ストアエンジンは、利用可能なすべてのメモリを含むノードのリソースを専有します。	列ストアエンジンはノードメモリの 25% に制限されます。残りのメモリはローストアエンジンに割り当てられます。
   業務への影響	トランザクション処理 (TP) と分析処理 (AP) のワークロードは異なるノードに分離され、互いに影響しません。	TP と AP のワークロードは同じノード上で実行され、互いに影響する可能性があります。
   コスト	IMCI 読み取り専用ノードには追加料金がかかり、通常のコンピューティングノードと同じ料金で課金されます。	追加コストはかかりません。

   IMCI 読み取り専用ノードの追加

   IMCI 読み取り専用ノードを追加するには、2 つの方法があります。

   説明

   クラスターには少なくとも 1 つの読み取り専用ノードが必要です。単一ノードクラスターに IMCI 読み取り専用ノードを追加することはできません。

   コンソール

   1. PolarDB コンソールにログインし、クラスターのリージョンを選択します。 次のいずれかの方法でノードの追加/削除 ウィザードを開くことができます。

      • クラスター ページで、操作 列の ノードの追加/削除 をクリックします。

      • 対象のクラスターの 概要 ページで、データベースノード セクションの ノードの追加/削除 をクリックします。

   2. 列ストアインデックス読み取り専用ノードの追加 を選択し、OK をクリックします。

   3. クラスターのアップグレード/ダウングレードページで、IMCI 読み取り専用ノードを追加し、支払いを完了します。

      1. 1 つの列ストアインデックス読み取り専用ノードの追加 をクリックし、ノード仕様を選択します。

      2. 切り替え時間を選択します。

      3. (オプション) 製品利用規約とサービスレベルアグリーメント (SLA) を確認します。

      4. 今すぐ購入 をクリックします。

   4. 支払いが完了したら、クラスター詳細ページに戻り、IMCI 読み取り専用ノードが追加されるのを待ちます。ノードは、そのステータスが実行中に変わると使用可能になります。

   購入時

   PolarDB 購入ページで、ノード数 セクションで、[IMCI 読み取り専用ノード] の数を指定します。

   PostgreSQL 16 (2.0.16.8.3.0 から 2.0.16.9.8.0 未満) または PostgreSQL 14 (2.0.14.10.20.0 から 2.0.14.17.35.0 未満)

   これらのバージョンの PolarDB for PostgreSQL または クラスターでは、IMCI 機能は polar_csi 拡張機能として提供されています。IMCI を使用するには、まず対象のデータベースに拡張機能を作成する必要があります。

   説明

   • polar_csi 拡張機能はデータベースレベルでスコープされます。クラスター内の複数のデータベースで IMCI を使用するには、各データベースに polar_csi 拡張機能を作成する必要があります。

   • 拡張機能のインストールに使用するデータベースアカウントは、特権アカウントである必要があります。

   polar_csi 拡張機能をインストールするには、2 つの方法があります。

   コンソール

   1. PolarDB コンソールにログインし、左側のナビゲーションペインで クラスター をクリックします。次に、クラスターが配置されている[リージョン]を選択し、クラスター ID をクリックしてクラスター詳細ページを開きます。

   2. 左側のナビゲーションペインで、設定と管理 > プラグインの管理 を選択します。プラグインの管理 タブで、プラグインがインストールされていません を選択します。

   3. ページの右上隅で、ターゲットデータベースを選択します。 polar_csi 拡張機能の行で、操作列の インストール をクリックします。 プラグインのインストール ダイアログボックスで、ターゲットの データベースアカウント を選択し、確認 をクリックしてターゲットデータベースに拡張機能をインストールします。

   CLI

   データベースクラスターに接続し、polar_csi 拡張機能を作成するための十分な権限があるデータベースで、次のステートメントを実行します。

   CREATE EXTENSION polar_csi;

ステップ 2: データの準備

1. このソリューションでは、トランザクションレコードテーブルを使用し、約 2 日間にわたる 1 億件のトランザクションレコードの生成をシミュレートします。取引時間は毎日 08:00～16:00 で、1 日あたり約 4,000 万件のレコードが生成されます。

   -- トランザクションレコードテーブル
   CREATE TABLE market_trades (
       trade_id   BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,  -- 自動インクリメントプライマリキー
       trade_ts   TIMESTAMP,                                        -- トランザクションタイムスタンプ
       market_id  VARCHAR,                                          -- マーケット ID
       price      DECIMAL,                                          -- トランザクション価格
       amount     DECIMAL,                                          -- トランザクション量
       insert_ts  TIMESTAMP                                         -- システム書き込みタイムスタンプ
   );
   
   INSERT INTO market_trades(trade_ts, market_id, price, amount, insert_ts)
   SELECT
       trade_ts,
       market_id,
       price,
       amount,
       trade_ts + (random() * 500)::INT * INTERVAL '1 millisecond' AS insert_ts
   FROM (
       -- ========================
       -- 1. 1 日目のピーク：2025-06-01 8:00 - 16:00、4,000 万行
       -- ========================
       SELECT
           '2025-06-01 08:00:00'::TIMESTAMP +
           (random() * 28800)::INT * INTERVAL '1 second' +  -- 28800 秒 = 8 時間
           (random() * 1000)::INT * INTERVAL '1 millisecond' AS trade_ts,
           CASE WHEN random() < 0.6 THEN 'BTC-USDT' ELSE 'ETH-USDT' END AS market_id,
           CASE WHEN random() < 0.6 THEN 30000 + (random() * 1000) ELSE 2000 + (random() * 100) END AS price,
           random() * 10 + 0.1 AS amount
       FROM generate_series(1, 40000000)
   
       UNION ALL
   
       -- ========================
       -- 2. 1 日目のオフピーク：2025-06-01 16:00 - 2025-06-02 08:00、1,000 万行
       -- ========================
       SELECT
           CASE
               WHEN random() < 0.5 THEN
                   -- 16:00 - 24:00
                   '2025-06-01 16:00:00'::TIMESTAMP + (random() * 28800)::INT * INTERVAL '1 second'
               ELSE
                   -- 00:00 - 08:00 (2 日目の早朝)
                   '2025-06-02 00:00:00'::TIMESTAMP + (random() * 28800)::INT * INTERVAL '1 second'
           END +
           (random() * 1000)::INT * INTERVAL '1 millisecond' AS trade_ts,
           CASE WHEN random() < 0.6 THEN 'BTC-USDT' ELSE 'ETH-USDT' END AS market_id,
           CASE WHEN random() < 0.6 THEN 30000 + (random() * 1000) ELSE 2000 + (random() * 100) END AS price,
           random() * 10 + 0.1 AS amount
       FROM generate_series(1, 10000000)
   
       UNION ALL
   
       -- ========================
       -- 3. 2 日目のピーク：2025-06-02 8:00 - 16:00、4,000 万行
       -- ========================
       SELECT
           '2025-06-02 08:00:00'::TIMESTAMP +
           (random() * 28800)::INT * INTERVAL '1 second' +
           (random() * 1000)::INT * INTERVAL '1 millisecond' AS trade_ts,
           CASE WHEN random() < 0.6 THEN 'BTC-USDT' ELSE 'ETH-USDT' END AS market_id,
           CASE WHEN random() < 0.6 THEN 30000 + (random() * 1000) ELSE 2000 + (random() * 100) END AS price,
           random() * 10 + 0.1 AS amount
       FROM generate_series(1, 40000000)
   
       UNION ALL
   
       -- ========================
       -- 4. 2 日目のオフピーク：2025-06-02 16:00 - 2025-06-03 08:00、1,000 万行
       -- ========================
       SELECT
           CASE
               WHEN random() < 0.5 THEN
                   -- 16:00 - 24:00
                   '2025-06-02 16:00:00'::TIMESTAMP + (random() * 28800)::INT * INTERVAL '1 second'
               ELSE
                   -- 00:00 - 08:00 (3 日目の早朝)
                   '2025-06-03 00:00:00'::TIMESTAMP + (random() * 28800)::INT * INTERVAL '1 second'
           END +
           (random() * 1000)::INT * INTERVAL '1 millisecond' AS trade_ts,
           CASE WHEN random() < 0.6 THEN 'BTC-USDT' ELSE 'ETH-USDT' END AS market_id,
           CASE WHEN random() < 0.6 THEN 30000 + (random() * 1000) ELSE 2000 + (random() * 100) END AS price,
           random() * 10 + 0.1 AS amount
       FROM generate_series(1, 10000000)
   ) AS data;

2. トランザクションレコードテーブルに列ストアインデックスを作成します。

   CREATE INDEX idx_csi_market_trades ON market_trades USING CSI;

ステップ 3: ローソク足集計クエリの実行

ユースケース：固定の時間枠でローソク足を計算します。

アプリケーション例：秒ごとの高値、安値、始値、終値、および総取引量を計算します。

以下の例では、それぞれ秒単位、分単位、時間単位、日単位のローソク足データを計算します。

-- 秒レベルのローソク足集計
/*+ SET (polar_csi.enable_query on) */
SELECT
    time_bucket('1 second', trade_ts) AS candle_ts,   -- 1 秒以内のデータ
    market_id,
    MIN(price) AS low,                                -- 1 秒以内の安値
    MAX(price) AS high,                               -- 1 秒以内の高値
    FIRST(price ORDER BY trade_ts) AS open,           -- 1 秒以内の始値
    LAST(price ORDER BY trade_ts) AS close,           -- 1 秒以内の終値
    SUM(amount) AS vol                                -- 1 秒以内の総取引量
FROM market_trades
WHERE trade_ts >= '2025-06-01 00:00:00' AND trade_ts <= '2025-06-02 00:00:00'
GROUP BY candle_ts, market_id
ORDER BY candle_ts, market_id;

-- 分レベルのローソク足集計
/*+ SET (polar_csi.enable_query on) */
SELECT
    time_bucket('1 minute', trade_ts) AS candle_ts,   -- 1 分以内のデータ
    market_id,
    MIN(price) AS low,                                -- 1 分以内の安値
    MAX(price) AS high,                               -- 1 分以内の高値
    FIRST(price ORDER BY trade_ts) AS open,           -- 1 分以内の始値
    LAST(price ORDER BY trade_ts) AS close,           -- 1 分以内の終値
    SUM(amount) AS vol                                -- 1 分以内の総取引量
FROM market_trades
WHERE trade_ts >= '2025-06-01 00:00:00' AND trade_ts <= '2025-06-02 00:00:00'
GROUP BY candle_ts, market_id
ORDER BY candle_ts, market_id;

-- 時間レベルのローソク足集計
/*+ SET (polar_csi.enable_query on) */
SELECT
    time_bucket('1 hour', trade_ts) AS candle_ts,     -- 1 時間以内のデータ
    market_id,
    MIN(price) AS low,                                -- 1 時間以内の安値
    MAX(price) AS high,                               -- 1 時間以内の高値
    FIRST(price ORDER BY trade_ts) AS open,           -- 1 時間以内の始値
    LAST(price ORDER BY trade_ts) AS close,           -- 1 時間以内の終値
    SUM(amount) AS vol                                -- 1 時間以内の総取引量
FROM market_trades
WHERE trade_ts >= '2025-06-01 00:00:00' AND trade_ts <= '2025-06-02 00:00:00'
GROUP BY candle_ts, market_id
ORDER BY candle_ts, market_id;

-- 日レベルのローソク足集計
/*+ SET (polar_csi.enable_query on) */
SELECT
    time_bucket('1 day', trade_ts) AS candle_ts,      -- 1 日以内のデータ
    market_id,
    MIN(price) AS low,                                -- 1 日以内の安値
    MAX(price) AS high,                               -- 1 日以内の高値
    FIRST(price ORDER BY trade_ts) AS open,           -- 1 日以内の始値
    LAST(price ORDER BY trade_ts) AS close,           -- 1 日以内の終値
    SUM(amount) AS vol                                -- 1 日以内の総取引量
FROM market_trades
WHERE trade_ts >= '2025-06-01 00:00:00' AND trade_ts <= '2025-06-02 00:00:00'
GROUP BY candle_ts, market_id
ORDER BY candle_ts, market_id;

SQL に関する注記

• /*+ SET (polar_csi.enable_query on) */：クエリに列ストアインデックスの実行計画を強制的に使用させます。一部のシナリオでは、オプティマイザが行ストアの方が効率的であると誤って推定することがあります。このヒントは、クエリが確実に列ストアパスを使用するようにします。

• time_bucket(bucket_width, ts) は、時系列データベース () が提供する関数です。この関数は、タイムスタンプ ts を、指定された間隔 bucket_width でグループ化します。