Hologres:DLF カタログによる Paimon データへのアクセス

外部データベースの使用 (推奨)

外部データベースは、DLF カタログを Hologres にマッピングすることで、Paimon テーブルのクエリ、管理、ミラーリングを可能にします。2 つの認証モードをサポートしています。

外部サーバーの使用 (レガシー)

外部サーバーアプローチは、タイムトラベル、ブランチクエリ、データミラーリングをサポートしていません。 これらの機能には外部データベースを使用してください。

ステップ 1：外部サーバーの作成

DLF V2.5 以降の場合：

CREATE SERVER IF NOT EXISTS <server_name> FOREIGN DATA WRAPPER dlf_fdw OPTIONS (
  catalog_type 'paimon',
  metastore_type 'dlf-rest',
  dlf_catalog '<dlf_catalog_name>'
);

DLF V1.0 の場合：

CREATE SERVER IF NOT EXISTS <server_name> FOREIGN DATA WRAPPER dlf_fdw OPTIONS (
  dlf_region '<region_id>',
  dlf_endpoint 'dlf-share.<region>.aliyuncs.com',
  oss_endpoint 'oss-<region_id>-internal.aliyuncs.com',
  dlf_catalog '<dlf_catalog_name>'
);

外部サーバーのその他の操作については、「外部サーバーのその他の操作」をご参照ください。

ステップ 2：外部テーブルの作成

IMPORT FOREIGN SCHEMA を使用して、DLF データベース内のすべてのテーブルをインポートしてください。

-- すべてのテーブルをインポート
IMPORT FOREIGN SCHEMA <dlf_db_name>
FROM SERVER <server_name>
INTO <holo_schema_name>
OPTIONS (if_table_exist 'update');

-- 特定のテーブルをインポート
IMPORT FOREIGN SCHEMA <dlf_db_name>
LIMIT TO (<table_name>[, ...])
FROM SERVER <server_name>
INTO <holo_schema_name>
OPTIONS (if_table_exist 'update');

または、CREATE FOREIGN TABLE を使用して単一の外部テーブルを作成してください。

CREATE FOREIGN TABLE <foreign_table_name>
(
  <column_name> <data_type>
  [, ...]
)
SERVER <server_name>
OPTIONS
(
  schema_name '<dlf_db_name>',
  table_name '<dlf_table_name>'
);

構文の詳細については、「IMPORT FOREIGN SCHEMA」および「CREATE FOREIGN TABLE」をご参照ください。

クエリ最適化のための統計情報の収集

ANALYZE または AUTO ANALYZE を実行してテーブルの統計情報を収集します。クエリ オプティマイザーはこれらの統計情報を使用して、最適なクエリプランを構築します。

-- すべての列の統計情報を収集
ANALYZE <ext_database_name>.<ext_schema_name>.<ext_table_name>;

-- 特定の列の統計情報を収集 (より正確。フィルター列に使用)
ANALYZE <ext_database_name>.<ext_schema_name>.<ext_table_name>(<col_name>[, <col_name>]);

-- 外部データベースに対して AUTO ANALYZE を有効化
ALTER EXTERNAL DATABASE <ext_database_name> WITH enable_auto_analyze 'true';

詳細については、「ANALYZE と AUTO ANALYZE」をご参照ください。

タイムトラベルを使用したクエリ

V3.2 以降、Hologres は、タイムスタンプ、バージョン (スナップショット ID) 、またはタグを使用して Paimon テーブルの履歴スナップショットをクエリできます。

利用可能なスナップショットの一覧表示：

SELECT * FROM hologres.hg_list_snapshots('<ext_database_name>.<ext_schema_name>.<ext_table_name>');

snapshots テーブルは次のフィールドを返します：

hg_list_versions 関数を使用して Paimon の snapshots システムテーブルをクエリすることもできます。Paimon テーブルの場合、バージョンはスナップショット ID に対応します。

SELECT * FROM hologres.hg_list_versions('<ext_database_name>.<ext_schema_name>.<ext_table_name>');

利用可能なタグの一覧表示：

SELECT * FROM hologres.hg_list_tags('<ext_database_name>.<ext_schema_name>.<ext_table_name>');

タイムスタンプによるクエリ：

SELECT * FROM <ext_database_name>.<ext_schema_name>.<ext_table_name>
FOR TIMESTAMP AS OF '<timestamp>';

バージョン (スナップショット ID) によるクエリ：

SELECT * FROM '<ext_database_name>.<ext_schema_name>.<ext_table_name>'
FOR VERSION AS OF '<version>';

タグによるクエリ：

SELECT * FROM '<ext_database_name>.<ext_schema_name>.<ext_table_name>'
FOR TAG AS OF '<tag>';

Hologres V3.2.6 および V4.0.6 以降では、期限切れの Paimon スナップショットに対応するタグからデータを読み取ることができます。

Paimon システムテーブルの詳細については、「Paimon システムテーブル」をご参照ください。

Paimon ブランチのクエリ

V3.2 以降、Hologres は特定の Paimon ブランチからデータをクエリできます。これには、そのブランチ内の履歴スナップショットも含まれます。

利用可能なブランチの一覧表示：

SELECT * FROM hologres.hg_list_branches('<ext_database_name>.<ext_schema_name>.<ext_table_name>');

ブランチの最新データのクエリ：

SELECT * FROM '<ext_database_name>.<ext_schema_name>.<ext_table_name>'
FOR branch AS OF '<branch_name>';

Paimon テーブルで scan.fallback-branch が設定されている場合、Hologres は、見つからないパーティションをフォールバックブランチから自動的に読み取ります。詳細については、「ブランチの管理」をご参照ください。

ブランチ内の履歴スナップショットのクエリ：

-- タイムスタンプによる指定
SELECT * FROM '<ext_database_name>.<ext_schema_name>.<ext_table_name>'
FOR branch AS OF '<branch_name>' TIMESTAMP AS OF '<timestamp>';

-- タグによる指定
SELECT * FROM '<ext_database_name>.<ext_schema_name>.<ext_table_name>'
FOR branch AS OF '<branch_name>' TAG AS '<tag_name>';

レイクテーブルミラーリングによるクエリの高速化

V3.2 以降、Hologres はレイクテーブルミラーリングをサポートします。この機能は、Paimon テーブルのメタデータとデータをほぼリアルタイムまたはスケジュールされた間隔で Hologres に同期します。ミラーリングされたデータは Hologres ストレージから提供されるため、クエリレイテンシーが大幅に削減されます。テーブル全体のミラーリングと、パーティション化されたテーブルの部分的なミラーリングの両方がサポートされています。

設定の詳細については、「レイクテーブルミラーリング」をご参照ください。

ビットマップファイルインデックスによるクエリの高速化

V5.0 以降、Hologres は Paimon テーブル上のビットマップファイルインデックス (インデックス形式 V2) を直接読み取ります。インデックス付きの列を等価 (=) または IN 述語でフィルタリングすると、Hologres リーダーはビットマップを使用して一致する行を特定し、無関係な行をスキップします。これにより、スキャンされるデータ量が削減され、レイクテーブルに対するポイントクエリと複数値クエリが高速化されます。この機能はデフォルトで有効になっています。

仕組みと適用条件：

• インデックスは Paimon のライターによって生成されます。Flink や Spark などのエンジンがこのインデックスを書き込むことができます。また、Hologres も、テーブルに file-index.bitmap.columns が宣言されていれば、Paimon 外部テーブルに対して INSERT を実行する際に自動的にインデックスを生成します。

• インデックスはデータファイルごとに保存されます。各データファイルには、データファイル名をプレフィックスとする .index ファイルが付属します。

• Hologres リーダーはインデックス形式のバージョンを自動的に検出し、両方の形式が同じテーブル内に共存できます。インデックス形式 V2 は V2 パーサーを使用し、インデックス形式 V1 は V1 パーサーを引き続き使用します。既存のクエリに影響はありません。

• サポートされているインデックス列の型には、固定長型 (SMALLINT、INT、BIGINT、DATE、TIME、TIMESTAMP など) と可変長型 (STRING または TEXT) があります。FLOAT、DOUBLE、DECIMAL はサポートされていません。

• 等価 (=) 述語と IN 述語のみが高速化の対象となります。範囲述語、LIKE、その他の述語はビットマップインデックスを使用しません。

手順：

1. テーブルを作成するときに、ビットマップインデックス列を宣言します。

   -- Paimon 外部テーブルを作成し、ビットマップインデックス列を宣言します
   CREATE EXTERNAL TABLE EXT_DATABASE_NAME.EXT_SCHEMA_NAME.EXT_TABLE_NAME (
     id        INT,
     c_id      INT,
     c_name    TEXT
   )
   WITH (
     "table_format"              = 'paimon',
     "file_format"               = 'parquet',       -- Parquet と ORC の両方がビットマップファイルインデックスの読み取りをサポートします
     "bucket"                    = '4',
     "bucket-key"                = 'id',
     "file-index.bitmap.columns" = 'c_id,c_name'    -- インデックスを作成する列 (カンマ区切り)
   );

   説明

   特定の列をインデックス形式 V1 にダウングレードするには、その列に対して "file-index.bitmap.<COLUMN_NAME>.version" = '1' を設定します。デフォルトはインデックス形式 V2 です。

2. ビットマップファイルインデックスのフラグが有効になっていることを確認します。Hologres は V5.0 以降、次のフラグをデフォルトでオンにします。状態を確認したり、手動でオンにしたりするには、次を実行します：

   -- ビットマップファイルインデックスの読み取りを制御するグローバルフラグ
   SELECT hg_admin_command('set_global_flag','enable_paimon_bitmap_index=true');                    -- マスター スイッチ
   SELECT hg_admin_command('set_global_flag','enable_push_bitmap_to_fse_parquet_reader=true');      -- Parquet リーダーでのプッシュダウン
   SELECT hg_admin_command('set_global_flag','enable_push_bitmap_to_fse_orc_reader=true');          -- ORC リーダーでのプッシュダウン

3. 等価クエリまたは IN クエリを実行し、EXPLAIN (ANALYZE) でインデックスが有効になったことを確認します。

   -- クエリと検証
   EXPLAIN (ANALYZE)
   SELECT count(*) FROM EXT_DATABASE_NAME.EXT_SCHEMA_NAME.EXT_TABLE_NAME WHERE c_id = 1;

   クエリプランでは、最下層の Seq Scan ノードが bitmap_used=N を報告します。ここで N はビットマップインデックスによって処理された述語の数です。scan_rows は、テーブル全体の行数ではなく、一致する行数に近くなります。これらの指標はいずれも、インデックスが有効になったことを示します。出力例：

   ->  Seq Scan on EXT_TABLE_NAME  (cost=0.00..5.03 rows=1000 width=1)
         Filter: (c_id = 1)
         Engine: FSE
         Foreign Table Type: DLF
       [id=1 split_count=1 rows=4(4/4/4) scan_rows=4(4/4/4) bitmap_used=1]

動的テーブルを使用した継続的なリフレッシュ

動的テーブルは、1 つ以上の Paimon 外部テーブルからの集計結果を自動的に処理して保存します。また、組み込みのリフレッシュ戦略も備えています。MaxCompute 外部テーブルと DLF 外部テーブルは、V3.0 以降、ベーステーブルとしてサポートされています。

完全リフレッシュ — 各リフレッシュサイクルですべてのデータを置き換えます：

CREATE DYNAMIC TABLE paimon_dt_table
WITH (
  auto_refresh_mode = 'full',
  freshness = '3 minutes'
) AS SELECT * FROM <ext_database_name>.<ext_schema_name>.<ext_table_name>;

増分リフレッシュ — 新規または変更されたデータのみを処理します：

CREATE DYNAMIC TABLE paimon_dt_table_incremental
WITH (
  auto_refresh_mode = 'incremental',
  freshness = '3 minutes'
) AS SELECT * FROM <ext_database_name>.<ext_schema_name>.<ext_table_name>;

自動リフレッシュ — データの変更に基づいて完全リフレッシュまたは増分リフレッシュを自動的に選択します：

CREATE DYNAMIC TABLE paimon_dt_table_auto
WITH (
  auto_refresh_mode = 'auto',
  freshness = '3 minutes'
) AS SELECT * FROM <ext_database_name>.<ext_schema_name>.<ext_table_name>;

詳細については、「動的テーブルの概要」をご参照ください。

INSERT INTO

次の例では、Paimon プライマリキーテーブルを作成し、行を挿入します。

-- Paimon プライマリキーテーブルを作成
CREATE EXTERNAL TABLE <ext_database_name>.<ext_schema_name>.<ext_table_name> (
  id         BIGINT,
  customer_id BIGINT,
  item       TEXT,
  amount     FLOAT,
  dt         TEXT,
  PRIMARY KEY (id)
)
WITH (
  "table_format"  = 'paimon',
  "file_format"   = 'orc',
  "merge-engine"  = 'deduplicate',  -- 必要に応じて調整
  "bucket"        = '4'             -- 必要に応じて調整
);

-- 行を挿入
INSERT INTO <ext_database_name>.<ext_schema_name>.<ext_table_name>
  (id, customer_id, item, amount, dt)
VALUES
  (1, 1001, 'apple',  3.50, '2025-09-10'),
  (2, 1002, 'banana', 2.00, '2025-09-10'),
  (3, 1003, 'orange', 4.20, '2025-09-11');

INSERT OVERWRITE

次の例では、パーティション化された Paimon プライマリキーテーブルを作成し、パーティションを上書きします。

-- パーティション化された Paimon プライマリキーテーブルを作成
CREATE EXTERNAL TABLE <ext_database_name>.<ext_schema_name>.<ext_table_name> (
  id     BIGINT,
  name   TEXT,
  amount FLOAT8,
  dt     TEXT
)
LOGICAL PARTITION BY LIST(dt)
WITH (
  "table_format" = 'paimon',
  "file_format"  = 'parquet',
  "merge-engine" = 'deduplicate',
  "bucket-key"   = 'id',
  "bucket"       = '2'
);

-- データを上書き
INSERT OVERWRITE <ext_database_name>.<ext_schema_name>.<ext_table_name>
  (id, name, amount, dt)
VALUES
  (1, 'Alice', 100.50, '2025-09-01'),
  (2, 'Bob',   200.00, '2025-09-01'),
  (3, 'Carol', 150.75, '2025-09-02');

UPDATE

UPDATE は、マージエンジンが deduplicate または partial-update に設定されている Paimon プライマリキーテーブルでのみサポートされます。プライマリキー列の更新はサポートされていません。

UPDATE <ext_database_name>.<ext_schema_name>.<ext_table_name>
SET amount = 5
WHERE id = 1;

DELETE

DELETE は、マージエンジンが deduplicate または partial-update に設定されている Paimon プライマリキーテーブルでのみサポートされます。partial-update テーブルの場合、DELETE を実行する前に partial-update.remove-record-on-delete プロパティを有効にする必要があります。

DELETE FROM <ext_database_name>.<ext_schema_name>.<ext_table_name>
WHERE id = 2;