Iceberg テーブルエンジン

このエンジンは、Amazon S3、Azure、HDFS、およびローカルに保存された Apache Iceberg テーブルに対するデータインテグレーションを提供します。

明示的にスキーマを指定しない場合、Icebergテーブルはストレージ上にあらかじめ存在している必要があります。書き込み可能なバックエンドに新しいスタンドアロンのIcebergテーブルを作成するには、CREATE TABLEステートメントでスキーマを指定します。

CREATE TABLE iceberg_table_s3
    ENGINE = IcebergS3(url, [, NOSIGN | access_key_id, secret_access_key, [session_token]], format, [,compression], [,extra_credentials])

CREATE TABLE iceberg_table_azure
    ENGINE = IcebergAzure(connection_string|storage_account_url, container_name, blobpath, [account_name, account_key, format, compression])

CREATE TABLE iceberg_table_hdfs
    ENGINE = IcebergHDFS(path_to_table, [,format] [,compression_method])

CREATE TABLE iceberg_table_local
    ENGINE = IcebergLocal(path_to_table, [,format] [,compression_method])

引数の説明は、それぞれ S3、AzureBlobStorage、HDFS、File エンジンの引数の説明に対応しています。 format は、Icebergテーブル内のデータファイルのフォーマットを表します。

IcebergS3 では、オプションの extra_credentials パラメータを使用して、ClickHouse Cloud でロールベースアクセスを行うための role_arn を渡すことができます。設定手順については、Secure S3 を参照してください。

エンジンパラメータは、Named Collections を使用して指定できます

CREATE TABLE iceberg_table ENGINE=IcebergS3('http://test.s3.amazonaws.com/clickhouse-bucket/test_table', 'test', 'test')

named collections を使用する場合:

<clickhouse>
    <named_collections>
        <iceberg_conf>
            <url>http://test.s3.amazonaws.com/clickhouse-bucket/</url>
            <access_key_id>test</access_key_id>
            <secret_access_key>test</secret_access_key>
        </iceberg_conf>
    </named_collections>
</clickhouse>

CREATE TABLE iceberg_table ENGINE=IcebergS3(iceberg_conf, filename = 'test_table')

Iceberg テーブルエンジンは、disk 設定からストレージバックエンドを自動検出し、それに応じて IcebergS3、IcebergAzure、または IcebergLocal を使い分けます。disk が指定されていない場合は、デフォルトで IcebergS3 実装が使用されます。

次の表は、スキーマ推論時に (読み取り用として) Iceberg のデータ型が ClickHouse のデータ型にどのように対応付けられるかを示しています。

上記のデータ型マッピングは読み取り時に適用されます。ClickHouse で Iceberg スキーマを作成または変更し、データを書き込む際には、以下の制限事項があります。

• ClickHouse では、Bool、Decimal、FixedString、Int8、UInt8、Int16、UInt16 を含む Iceberg スキーマを作成したり、カラムをこれらの型のいずれかとして追加または変更したりすることはできません。この操作は、サポートされていない型の例外により失敗します。ただし、既存の Iceberg カラムに Bool または Decimal の値を挿入することは可能です。
• ClickHouse で Iceberg スキーマを作成または変更する場合、すべての DateTime および DateTime64 カラムは、マイクロ秒精度でタイムゾーンを持たない Iceberg の timestamp にマッピングされます。ClickHouse では timestamptz、timestamp_ns、timestamptz_ns のスキーマ型を生成できないため、より高い精度やタイムゾーンのセマンティクスは Iceberg スキーマに反映されません。
• ClickHouse では、decimal、fixed、timestamp_ns、timestamptz_ns のカラムを直接パーティションフィールドとして使用する Iceberg テーブルには書き込めません。この操作は、サポートされていない型の例外により失敗します。
• Bool や Decimal など、ClickHouse が境界値をシリアル化できない型を含むデータファイルでは、ClickHouse は Iceberg マニフェストエントリからすべてのカラムの下限値と上限値を省略します。カラムサイズと null カウントは引き続き含まれ、データの正確性も維持されますが、リーダーはこれらのファイルに対してマニフェストレベルの min-max プルーニングを使用できません。

ClickHouse は、時間の経過とともにスキーマが変化した Iceberg テーブル の読み取りをサポートしています。これには、カラムが追加、削除、並べ替えされたテーブルや、カラムが required から Nullable に変更されたテーブルが含まれます。さらに、次の型変換がサポートされています。

• int -> long
• float -> double
• decimal(P, S) -> decimal(P', S) where P' > P.

現在のところ、ネストされた構造や、Array および Map 内の要素の型を変更することはできません。

動的なスキーマ推論を使用して作成したテーブルで、作成後にスキーマが変更されたものを読み取るには、テーブルの作成時に allow_dynamic_metadata_for_data_lakes = true を設定します。

ClickHouse は、Iceberg テーブルに対する SELECT クエリでパーティションプルーニングをサポートしており、無関係なデータファイルをスキップすることでクエリパフォーマンスを最適化できます。パーティションプルーニングを有効にするには、use_iceberg_partition_pruning = 1 を設定します。Iceberg のパーティションプルーニングの詳細については、https://iceberg.apache.org/spec/#partitioning を参照してください

ClickHouse は Iceberg テーブルでのタイムトラベルをサポートしており、特定のタイムスタンプまたはスナップショット ID を指定して過去のデータをクエリできます。

時間の経過とともに、Iceberg テーブルへの頻繁な書き込みによって、現在のスナップショットのマニフェストリストに多数の小さなマニフェストファイルが蓄積されることがあります。マニフェストリストが長くなると、データファイルを見つけるためにすべてのマニフェストファイルを読み取る必要があるため、クエリプランの作成が遅くなります。ClickHouse では、OPTIMIZE TABLE ... MANIFEST ステートメントを使用して、これらのマニフェストファイルを、より少数で大きなものにコンパクションできます。

OPTIMIZE TABLE example_table MANIFEST SETTINGS allow_experimental_iceberg_compaction = 1;

これにより、新しいスナップショット (replace 操作) が生成され、統合された一連のマニフェストファイルを介して同じデータファイルを参照するようになります。データファイルは書き換えられず、行の追加・削除・重複排除も行われません。再編成されるのはマニフェスト層だけです。

• この機能は Experimental であり、allow_experimental_iceberg_compaction 設定で制御されます。設定が有効になっていない場合、このステートメントは例外をスローします。
• コンパクション は、current スナップショット のマニフェストリスト内のマニフェストファイル数が iceberg_manifest_min_count_to_compact 設定で指定された threshold (デフォルトは 30) を超えた場合にのみ試行されます。現在の数が threshold 以下の場合、コンパクション はスキップされ、新しい スナップショット は作成されません。より積極的に コンパクション を行うには、threshold を低く設定してください。
• OPTIMIZE TABLE ... MANIFEST は Iceberg tables でのみサポートされています。これを他の テーブルエンジン に対して実行すると、例外をスローします。
• OPTIMIZE TABLE ... MANIFEST は Iceberg format-version 2 の table でのみサポートされています。これを format-version 1 の table に対して実行すると例外をスローし、format-version 3 の table に対して実行した場合も同様です。これは、v3 の row-lineage first_row_id メタデータが manifest rewrite を通じてまだ往復変換されないためです。
• OPTIMIZE TABLE ... MANIFEST は、data files にファイルごとの key_metadata を含む暗号化された Iceberg tables ではサポートされていません。manifest rewrite の過程でこの暗号化メタデータを保持する機能はまだ実装されていないため、このステートメントは NOT_IMPLEMENTED 例外をスローします。

ClickHouse は、以下の削除方式を使用する Iceberg テーブルの読み取りをサポートしています。

• 位置削除
• 等価削除 (バージョン 25.8+ でサポート)

以下の削除方式は サポートされていません。

• 削除ベクトル (v3 で導入)

 SELECT * FROM example_table ORDER BY 1
SETTINGS iceberg_timestamp_ms = 1714636800000

 SELECT * FROM example_table ORDER BY 1
SETTINGS iceberg_snapshot_id = 3547395809148285433

注: 同一のクエリで iceberg_timestamp_ms パラメータと iceberg_snapshot_id パラメータを同時に指定することはできません。

• スナップショット は通常、次の場合に作成されます:

  • 新しいデータがテーブルに書き込まれたとき
  • 何らかのデータのコンパクションが実行されたとき

• スキーマ変更では通常、スナップショットは作成されません - このため、スキーマ進化を経たテーブルでタイムトラベルを使用する場合には、重要な挙動の違いが生じます。

これらのシナリオでは、外部の Iceberg ライターによるスキーマ変更を Spark を使って説明します。

次の一連の操作を考えてみましょう。

-- Create a table with two columns
CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example (
order_number int,
product_code string
  )
USING iceberg
OPTIONS ('format-version'='2')

-- Insert data into the table
INSERT INTO spark_catalog.db.time_travel_example VALUES
    (1, 'Mars')

  ts1 = now() // A piece of pseudo code

-- Alter table to add a new column
ALTER TABLE spark_catalog.db.time_travel_example ADD COLUMN (price double)

ts2 = now()

-- Insert data into the table
INSERT INTO spark_catalog.db.time_travel_example VALUES (2, 'Venus', 100)

ts3 = now()

-- Query the table at each timestamp
  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts1;

+------------+------------+
|order_number|product_code|
+------------+------------+
|           1|        Mars|
+------------+------------+
  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts2;

+------------+------------+
|order_number|product_code|
+------------+------------+
|           1|        Mars|
+------------+------------+

  SELECT * FROM spark_catalog.db.time_travel_example TIMESTAMP AS OF ts3;

+------------+------------+-----+
|order_number|product_code|price|
+------------+------------+-----+
|           1|        Mars| NULL|
|           2|       Venus|100.0|
+------------+------------+-----+

異なるタイムスタンプでのクエリ結果:

• ts1 と ts2: 元の 2 つのカラムのみが表示されます
• ts3: 3 つのカラムすべてが表示され、最初の行の price は NULL になります

現在時点でのタイムトラベルクエリでは、現在のテーブルとは異なるスキーマが表示されることがあります。

-- Create a table
CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example_2 (
order_number int,
product_code string
  )
USING iceberg
OPTIONS ('format-version'='2')

-- Insert initial data into the table
INSERT INTO spark_catalog.db.time_travel_example_2 VALUES (2, 'Venus');

-- Alter table to add a new column
ALTER TABLE spark_catalog.db.time_travel_example_2 ADD COLUMN (price double);

ts = now();

-- Query the table at a current moment but using timestamp syntax

  SELECT * FROM spark_catalog.db.time_travel_example_2 TIMESTAMP AS OF ts;

    +------------+------------+
    |order_number|product_code|
    +------------+------------+
    |           2|       Venus|
    +------------+------------+

-- Query the table at a current moment
  SELECT * FROM spark_catalog.db.time_travel_example_2;
    +------------+------------+-----+
    |order_number|product_code|price|
    +------------+------------+-----+
    |           2|       Venus| NULL|
    +------------+------------+-----+

これは、ALTER TABLE では新しいスナップショットが作成されず、現在のテーブルについては Spark がスナップショットではなく最新のメタデータファイルから schema_id の値を取得するためです。

2 つ目は、タイムトラベルを行っても、テーブルにまだ一度もデータが書き込まれていない時点の状態は取得できないということです:

-- Create a table
CREATE TABLE IF NOT EXISTS spark_catalog.db.time_travel_example_3 (
order_number int,
product_code string
  )
USING iceberg
OPTIONS ('format-version'='2');

ts = now();

-- Query the table at a specific timestamp
  SELECT * FROM spark_catalog.db.time_travel_example_3 TIMESTAMP AS OF ts; -- Finises with error: Cannot find a snapshot older than ts.

ClickHouse では、この挙動は Spark と共通です。Spark の Select クエリを ClickHouse の Select クエリに置き換えて考えても、同じように機能します。

ClickHouse で Iceberg テーブルエンジンを使用する場合、システムは Iceberg テーブルの構造を記述した適切な metadata.json ファイルを特定する必要があります。この解決処理は、次のように行われます。

1. パスの直接指定:

• iceberg_metadata_file_path を設定すると、システムはこの値を Iceberg テーブル の directory path と組み合わせて、その正確な path を使用します。
• この設定が指定されている場合、他のすべての解決設定は無視されます。

2. table UUID の照合:

• iceberg_metadata_table_uuid が指定されている場合、システムは次のように動作します:
  • metadata directory 内の .metadata.json ファイルのみを対象にします
  • 指定した UUID と一致する table-uuid フィールドを含むファイルに絞り込みます (大文字と小文字は区別しません)

3. デフォルトの検索:

• 上記のどちらの設定も指定されていない場合、metadata directory 内のすべての .metadata.json ファイルが候補になります

上記のルールを使用して候補ファイルを特定した後、システムはその中から最も新しいものを判定します。

• iceberg_recent_metadata_file_by_last_updated_ms_field が有効な場合:

  • last-updated-ms の値が最も大きいファイルが選択されます

• それ以外の場合:

  • バージョン番号が最も大きいファイルが選択されます
  • (バージョンは、V.metadata.json または V-uuid.metadata.json 形式のファイル名では V として表されます)

注: 特に明記されていない限り、ここで言及している設定はすべて engine レベルの設定であり、以下のように table の作成時に指定する必要があります。

CREATE TABLE example_table ENGINE = Iceberg(
    's3://bucket/path/to/iceberg_table'
) SETTINGS iceberg_metadata_table_uuid = '6f6f6407-c6a5-465f-a808-ea8900e35a38';

注: Icebergカタログは通常メタデータの解決を担いますが、ClickHouse の Iceberg テーブルエンジンは S3 に保存されたファイルを Icebergテーブルとして直接解釈するため、これらの解決ルールを理解しておくことが重要です。

Iceberg テーブルエンジンおよびテーブル関数は、S3、AzureBlobStorage、HDFS ストレージと同様にデータキャッシュをサポートしています。こちらを参照してください。

Iceberg テーブルエンジンとテーブル関数は、マニフェストファイル、マニフェストリスト、metadata JSON の情報を保存するメタデータキャッシュをサポートしています。キャッシュはメモリ上に保存されます。この機能は設定 use_iceberg_metadata_files_cache で制御されており、デフォルトで有効です。

非同期メタデータプリフェッチは、iceberg_metadata_async_prefetch_period_ms を設定することで、Iceberg テーブルの作成時に有効化できます。0 (デフォルト) に設定されている場合、またはメタデータキャッシュが有効になっていない場合は、非同期プリフェッチは無効になります。 この機能を有効にするには、0 以外のミリ秒単位の値を指定する必要があります。これは、プリフェッチサイクル間の間隔を表します。

有効にすると、サーバーはリモートのカタログを定期的に確認し、新しいメタデータバージョンを検出するバックグラウンド処理を実行します。続いてそれを解析し、スナップショットを再帰的にたどりながら、アクティブなマニフェストリストファイルとマニフェストファイルを取得します。 メタデータキャッシュにすでにあるファイルは、再度ダウンロードされません。各プリフェッチサイクルの終了時点で、最新のメタデータスナップショットがメタデータキャッシュで利用可能になります。

CREATE TABLE example_table ENGINE = Iceberg(
    's3://bucket/path/to/iceberg_table'
) SETTINGS
    iceberg_metadata_async_prefetch_period_ms = 60000;

読み取り操作における非同期メタデータ先読みを最大限に活用するには、iceberg_metadata_staleness_ms パラメータをクエリまたはセッションのパラメータとして指定する必要があります。デフォルトでは (0 - 未指定) 、各クエリのコンテキストで、サーバーはリモートカタログから最新のメタデータを取得します。 メタデータの古さに対する許容値を指定すると、サーバーはリモートカタログにアクセスせずに、キャッシュされたメタデータスナップショットを使用できるようになります。cache にメタデータのバージョンがあり、それが指定した古さの window 内にダウンロードされていれば、そのバージョンがクエリの処理に使用されます。 そうでない場合は、リモートカタログから最新バージョンが取得されます。

SELECT count() FROM icebench_table WHERE ...
SETTINGS iceberg_metadata_staleness_ms=120000

注記: 非同期メタデータの先読みは ICEBERG_SCEDULE_POOL で実行されます。これは、アクティブな Iceberg テーブルに対するバックグラウンド処理用のサーバー側 threadpool です。この threadpool のサイズは、iceberg_background_schedule_pool_size サーバー設定パラメータ (デフォルトは 10) で制御されます。

注記: 現時点では、非同期先読みが有効な場合、メタデータキャッシュのサイズは、すべてのアクティブなテーブルの最新のメタデータスナップショット全体を保持できるだけ十分であることが想定されています。

• Iceberg テーブル関数