1. Apache Icebergのクエリプランニング実装の概要と重要性 [20]

  1. 講演の主題と焦点:

    1. 本講演は、Apache Icebergのプランニング実装が実際にどのように機能するかについての極めて技術的な概要であり、特にJava実装に焦点を当てている [20]。
    2. プランニングが重要である理由は、読み込むデータ量を最小限に抑えることが、クエリを高速化する最速の方法であるためである [30]。
    3. これを実現する最も迅速な方法は、ファイルに関連付けられたメトリクスを使用して、ファイルがクエリに関連するかどうかを判断することである [32]。

  2. 従来のファイルレベルでのメトリクス利用:

    1. Apache Parquetのようなほとんどのカラム型ファイル形式は、ファイルのフッターにメトリクス（例：列ごとの最小値と最大値の範囲）を持っている [34]。
    2. クエリの述語（例: X=2, Y>40, ZがBAZで始まる）とこれらのメトリクスを比較することで、ファイルを読み込む必要があるかどうかを判断できる [36]。
    3. 例として、Yが40より大きいかという述語に対して、ファイルのYの範囲が0から10であれば、そのファイルを読み込む必要がないと確実に判断できる [37]。
    4. IN句や範囲指定（例: X<50）のクエリでも同様のロジックが適用され、ファイルに一致する可能性があれば読み込みが必要と判断される [42]。

  3. クラウド環境での課題とテーブルフォーマットへの移行:

    1. ファイルレベルでのメトリクスチェックは、HDFSのようなファイルシステムでは非常に安価であった [51]。
    2. しかし、S3やGoogle Cloudのようなクラウドファイルシステムに移行すると、APIコールごとに課金されるため、ファイルのリスト化やフッターチェックが非常に高価になる [56]。
    3. このコスト増大のため、テーブルフォーマットの世界に移行し、Apache Icebergはその一つである [59]。

  4. 本講演の目標:

    1. Apache Icebergがどのようにメトリクスを取得し、ファイル形式を読み込まずに読み取れる形式にし、そのメタデータ情報でクエリを計画するかを説明する [60]。

2. Icebergプランニングの段階的プロセスとアーキテクチャ [61]

  1. プランニングの段階的概要:

    1. Icebergの概要: Icebergのアーキテクチャとプランニングに必要な主要機能について説明する [62]。
    2. クエリの分解: SparkSQLなどのクエリがIcebergライブラリに渡される要素にどのように分解されるかを説明する [63]。
    3. Manifest Listフィルタリング: SQLクエリから抽出された述語が、Manifest Listのフィルタリングにどのように適用されるかを説明する [65]。
    4. Manifestフィルタリング: Manifestファイルがどのようにフィルタリングされるかを説明する [66]。
    5. 削除ファイルの計画: 比較的新しい概念である削除ファイルが、データファイルと共にどのように計画されるかを説明する [67]。
    6. 将来の展望: Apache Iceberg V4で取り組んでいる構造的な変更について簡単に触れる [69]。

  2. プランニングの三つの領域:

    1. エンジン固有の領域: Spark、Trino、Snowflakeなどのエンジンに依存し、クエリをIcebergが扱えるエンティティに変換する [71]。
    2. Icebergコードベース内: 変換された構成要素を使用して、Icebergコードベース内でクエリを実行する [76]。
    3. ファイルスキャンタスク: 最終的にディスクからファイルを読み取るためのファイルスキャンタスクを取得する [77]。
    4. この後、さらに行グループ（Row Group）のフィルタリング層があるが、これはファイル形式固有であるため、本講演では省略される [78]。

3. Apache Icebergの基本概念とメタデータ構造 [83]

  1. Icebergの定義:

    1. Icebergはオープンテーブルフォーマットであり、多数のファイルをリレーショナルデータベースのように扱う方法を提供する [89]。
    2. 目的は、コンピューティングとストレージを分離し、ACIDトランザクションを維持し、スキーマ進化やパーティショニングなどの概念を提供することである [90]。
    3. 特にクラウドストレージ向けに最適化され、定義自体がオープンコンセプトであることが重要である [91]。

  2. メタデータ構造:

    1. Icebergは、データファイルの上にメタデータのセットを構築する [93]。
    2. Icebergは、このメタデータを一連のファイルとして格納することを選択している [98]。
    3. この構造は、Apache Iceberg仕様（Spec）で定義されており、ベンダーやエンジンに依存しない形でルールが記述されている [99]。
    4. IcebergのJavaライブラリはこのルールに従うが、他の言語やエンジンからの実装も互換性を保てるように設計されている [105]。

  3. 階層的メタデータ構造:

    1. メタデータファイル (metadata.json): テーブルのJSON記述であり、スキーマやテーブルの一般的な形状を記述する [107]。
    2. スナップショット (Snapshots): メタデータファイルがリストするものであり、それぞれがManifest Listを指す [111]。
    3. Manifest List: Manifest Fileのリストである [112]。
    4. Manifest File: データファイル（例: Parquetファイル）と削除ファイルのリストである [113]。
    5. 階層的フィルタリング: この階層構造により、下位層を読み込む必要があるかを判断するために、現在のファイルの統計情報を確認できる [118]。これにより、プランニングのために実際に読み込む必要があるファイル数が大幅に削減される [119]。

  4. コミットの仕組みとカタログの役割:

    1. Icebergの構造は階層的なツリーであり、カタログ（例: Hive Metastore、DynamoDBなど）がテーブル識別子をmetadata.jsonファイルに結びつける [122]。
    2. コミットの実行: 新しい状態に移行するには、新しいツリーを構築する [128]。
    3. 新しいmetadata.jsonファイルを作成し、新しいManifest Listを指す [129]。
    4. 新しいManifest Listは、古いManifestだけでなく、追加された新しいデータファイルを含む新しいManifestも指す [131]。
    5. これにより、カタログがmetadata.json間のポインタを切り替えることで、ツリーが横に成長し続ける [132]。
    6. タイムトラベル: 以前のスナップショットも利用可能であるため、新しいManifest Listは古いManifestも指すことができる [129]。

  5. Iceberg RESTプロトコルへの移行 (将来の方向性):

    1. 従来のカタログプラグインでは、クライアントがツリー全体を生成し、カタログにポインタの切り替えを要求していた [134]。
    2. Iceberg RESTプロトコルでは、クライアントはテーブルの変更セットをカタログに送信するだけでよく、カタログが実際にmetadata.jsonファイルを書き込み、ポインタを交換する [137]。
    3. 将来的には、クライアントは書き込んだデータファイルのみを知り、コミット処理のロジック全体をカタログに依存させることを目指している [140]。

  6. クエリプランニングにおけるコミットの重要性:

    1. クエリプランニングで最初に行うべきことの一つは、metadata.jsonが現在のスナップショットと過去のスナップショットの両方を持っているため、どのツリー（スナップショット）に入るかを決定することである [143]。

4. Icebergの主要機能: スナップショット、スキーマ進化、パーティショニング [145]

4.1. スナップショットとスキーマの関連性 [146]

  1. スナップショットの構造:
    1. metadata.json内のスナップショットはJSONオブジェクトのリストであり、それぞれがManifest Listのパスを持つ [149]。
    2. 各スナップショットは親を指しており、新しい子を追加しても以前のスナップショットは変更されない [154]。
    3. これにより、新しいスナップショットを作成し続けると、親を指す構造ができあがる [157]。
    4. スナップショット間でブランチを作成することも可能である [159]。

4.2. フィールドIDによるスキーマ進化の実現 [161]

  1. 従来の課題:

    1. 従来のシステムでは、スキーマ・オン・リードパターンが使用され、ファイルがテーブルのスキーマを決定していた [163]。
    2. この方式では、列を削除してから同じ名前の列を再度追加すると、以前のデータが復活してしまう問題があった [165]。これは、テーブルのフィールド名とファイル内のフィールド名が強く結びついているためである [174]。

  2. Icebergの解決策: フィールドID:

    1. Icebergでは、スキーマを定義する際に、すべてのフィールドに名前（SQLから見える）とフィールドID（データファイルへのアクセスに使用）を持たせる [176]。
    2. データファイル自体には、フィールドIDと列名のマッピングが書き込まれる [185]。
    3. クエリで列名（例: ID）を選択すると、まずテーブルスキーマで対応するフィールドID（例: 2）を検索し、そのIDを使用してデータファイル内のフィールドにアクセスする [187]。
    4. 名前の変更: テーブル内の列名を変更しても、基になるファイルは変更されない。変更されるのはフィールド名だけであり、フィールドIDはそのまま維持される [191]。
    5. 列の削除と再追加: 列を削除してから同じ名前で再追加した場合、新しい列には新しいフィールドIDが付与される [198]。これにより、古いファイル内のデータは新しい列と一致しないため、データの復活を防ぐことができる [199]。

  3. スキーマ制約と型定義:

    1. 現在の制約: Icebergの初期仕様（V3）では、requiredとoptionalのみが許可されていた [203]。
    2. 新しい機能: 最近、デフォルト値（defaults）の機能が追加された [207]。
    3. 将来の展望: 将来的には、列の制約（例: 一意性）の追加が検討されているが、分散システムでの一意性の実装は非常に複雑でコストが高いため、慎重である [212]。
    4. 型定義: Iceberg内のすべての型は、エンジン定義ではなくIcebergによって定義されている [221]。これにより、異なるエンジン間で整数型などのバイト長や最小/最大値が異なっていても、Icebergの定義に変換することで互換性を確保している [224]。

4.3. 隠されたパーティショニングとパーティション変換 [237]

  1. Icebergのパーティショニング:

    1. Hiveのようなシステムは、ディレクトリ構造（例: column_name=value）でパーティショニングを行う [240]。
    2. Icebergはこれとは異なり、ファイルを書き込む際に、パーティション値のタプルをファイルに関連付ける [243]。
    3. このタプルは、パーティション仕様（Partition Spec）に基づいており、これはファイル内の異なる列に対する変換（Transforms）の集合である [244]。

  2. パーティション変換の例:

    1. バケット変換（Bucketing Transform）: 値のハッシュを取り、モジュロを適用して整数値を得る [245]。
    2. この変換は、特定のソースID（フィールドID）を入力として使用し、特定のフィールドIDで定義される [247]。
    3. データファイルライターは、この変換を使用してパーティションタプル（例: 1または2）を生成し、ファイルに関連付ける [249]。
    4. このタプルは、ファイルの物理的な場所とは独立している [255]。

  3. 名前の独立性:

    1. パーティション変換も、列名ではなくフィールドID（ソースID）に結びつけられている [257]。
    2. これにより、フィールド名を変更しても、パーティション変換を変更する必要はない [262]。

  4. 変換の標準化:

    1. Icebergは、truncate、bucketなど、様々な変換を提供している [264]。
    2. これらの変換は、すべてIceberg仕様内で明示的に定義されている [268]。
    3. 例として、バケット変換は、特定のMurmur3実装とモジュロを使用して定義されており、すべてのエンジンが同じ関数を使用して同じ結果を得ることを保証している [271]。

5. SparkSQLからIcebergへの述語プッシュダウンとスキーマプルーニング [275]

5.1. SparkからIcebergへの変換プロセス [275]

  1. データソースV2 API:

    1. SparkSQLでクエリが実行されると、多くの内部処理を経て、最終的にIceberg Sparkプラグイン内のSparkScanBuilderに入る [275]。
    2. SparkScanBuilderは、Sparkがデータソースと通信し、「クエリについて知っている情報」を渡し、「行の束」を返すように要求する場所である [277]。

  2. 述語のプッシュダウン (Push Predicates):

    1. Sparkは、pushPredicates関数を通じて、Spark述語のリストをIcebergに渡す [279]。
    2. 変換: まず、SparkのデータソースV2式言語からIcebergの式言語に変換される（convert関数） [285]。
    3. バインド: 変換された述語は、最初はアンバウンド述語（名前のみを持つ）である [293]。
    4. Icebergは、この述語を現在のスキーマにバインドしようとする [297]。
    5. スキーマ内で述語の名前（例: X）に対応するフィールドIDを検索する [299]。
    6. フィールドIDが式に結びつけられると、その述語はIceberg API内でプッシュダウンできると判断される [299]。
    7. Sparkへのフィードバック: Icebergは、実際に処理できた述語（pushableFilters）と処理できなかった述語（postScanFilters）をSparkに伝え返す [302]。
    8. Icebergが完全に処理した述語については、Sparkは計算層で再適用しないという最適化を行うことができる [304]。

  3. タイムトラベルとスキーマの選択:

    1. クエリにAS OF句が含まれる場合、APIはその情報を解析し、スキャンビルダーにプッシュダウンする [307]。
    2. Icebergは、選択されたスナップショットに関連付けられたスキーマIDを使用して、そのスナップショット時点の正しいスキーマにバインドする [310]。

5.2. スキーマプルーニング (Prune Columns) [321]

  1. プルーニングの実行:
    1. prune関数は、Sparkが要求する列の定義とテーブルのIcebergスキーマを受け取り、要求された列に一致するIcebergスキーマを出力する [323]。
    2. 述語列の包含: フィルタ式（述語）に含まれる列は、クエリで明示的に射影されていなくても、読み取りのためにスキーマに含める必要がある [325]。
    3. 例として、SELECT X WHERE Z > 5というクエリでは、Zは述語の評価のためだけに必要であり、読み取りのために射影されるわけではない [326]。
    4. 最終的に、IcebergはフィールドIDを含むIcebergスキーマを返し、残りのプランニングで使用する [330]。

6. Manifest ListとManifestファイルのフィルタリングロジック [332]

6.1. スナップショットの読み込みとManifest Groupの初期化 [332]

  1. スナップショットの読み込み:
    1. Iceberg固有のオブジェクト（式とスキーマ）が準備された後、まずスナップショットを読み込む必要がある [331]。
    2. スナップショットは、特定のManifest Listに関連付けられており、Manifest ListにはデータManifestと削除Manifestが含まれる [333]。
    3. Manifest Listファイルを読み込み、データManifestと削除Manifestの二つのオブジェクトを取得する [335]。
    4. これらのオブジェクトは、プランニングの実際の作業が行われるManifestGroup.javaに渡される [340]。

6.2. Manifest Listのフィルタリング (パーティションサマリーに基づく) [343]

  1. Manifest Listの目的:

    1. Manifest List内の各Manifestが、クエリに一致する可能性のあるエントリを含むかどうかを判断する必要がある [343]。
    2. Manifest Listのエントリでチェックできる唯一のものは、パーティションサマリーである [345]。

  2. Manifest Listエントリの情報:

    1. パーティション仕様ID (Partition Spec ID): このManifest内のすべてのデータファイルに使用されたパーティション変換のセットを示す [349]。
    2. パーティション統計 (Partition Statistics): このManifest内のタプル（パーティション値）の最小値と最大値の境界を示す [350]。

  3. 述語の変換とバインド:

    1. Manifest Listのフィルタリングでは、クエリのIceberg式を、Manifestに使用された特定のパーティション仕様にバインドし、射影する必要がある [351]。
    2. 例として、タイムスタンプに基づく式（例: timestamp < 2008/11/05）は、パーティション変換（例: day変換）が適用された後、パーティションサマリーのフィールドに対する式に変換される [353]。
    3. 変換後の式は、day_timestamp < days_from_epic_version_of_that_timestampのようになる [361]。

  4. Visitorパターンによる評価:

    1. 変換された式は、Visitorパターンを使用して評価される [363]。
    2. ANDノードに到達した場合、すべての子ノードが真であることを確認する [367]。
    3. Less Thanノードのような特定の操作に到達した場合、そのノード固有の実装が実行される [377]。
    4. 統計情報との比較: 統計構造内の下限（Lower Bound）がリテラル値より小さい場合、一致する行がないと判断される [378]。
    5. この評価により、実際に開く必要があるManifestファイルが決定される [385]。

6.3. Manifestファイル内のデータファイルフィルタリング (列メトリクスに基づく) [386]

  1. データファイルエントリの情報:

    1. Manifestファイルを開くと、データファイルのリストが含まれており、各エントリにはManifest Listよりもはるかに多くの情報が含まれている [386]。
    2. パーティション: データファイル固有のパーティションタプル（上限/下限なし） [391]。
    3. 列メトリクス:
    • 値の総数（value counts） [393]。
    • Null値の数（null value counts） [394]。
    • NaN値の数（nan value count） [395]。
    • 下限と上限（Lower and Upper Bound） [397]。
    1. これらの統計は、Icebergテーブルの書き込み時に、Parquetファイルなどのフッターから集約され、Manifestファイル内の行として保存される [399]。

  2. 列メトリクスの格納形式と課題:

    1. 列メトリクスは、フィールドIDからシリアライズされたバイナリ表現へのマップとして保持される [408]。
    2. 課題1: 型情報がない: バイナリ表現から元の型を判断することは難しく、型が昇格（Promotion）された場合、表現が変わる可能性がある [409]。
    • Iceberg仕様では、バイナリの長さを確認して型を判断できる型拡張（Type Widening）のみが許可されている [413]。
    1. 課題2: ワイドテーブルでの非効率性: メトリクスがマップとして格納されているため、特定の列のメトリクスのみを読み取るために、すべてのキーと値を読み取る必要があり、カラム型ファイル形式でのプッシュダウンが利用できない [414]。

  3. Manifest Readerによる評価:

    1. Manifest Groupは、ManifestReader.javaを使用してManifestの読み取りを開始する [421]。
    2. entries関数は、以下の二つの評価器に基づいて各データファイルエントリをフィルタリングする [424]。
    • パーティション評価器 (Evaluator.java): パーティションを評価する [425]。
    • メトリクス評価器 (InclusiveMetricsEvaluator): 列メトリクス全体を評価する [426]。
    1. 評価器は、常に「行が一致する可能性がある」または「行が一致しない」のいずれかを返す [432]。

  4. パーティション評価の仕組み:

    1. 再びVisitorパターンを使用し、Iceberg式を特定のパーティション構造にバインドする [436]。
    2. 評価器は、構造体から適切な部分を抽出し、リテラル値と比較する [440]。

  5. 列メトリクス評価の仕組み:

    1. データファイルから統計オブジェクトを取り出し、Iceberg式に対して様々な統計をチェックする [443]。
    2. Null/NaNチェック: まず、NullまたはNaNのみが含まれているかをチェックする [446]。
    3. 上限のチェック: 次に、上限（Upper Bound）をチェックする [449]。
    4. T upper = upperBound(term)の行では、式に適用する必要がある関数（例: 生成された列や実際の列）を考慮して、上限を比較可能な形式に変換する [450]。
    5. 最終的に、ファイル構造内の上限と式の上限を比較し、行が一致するかどうかを判断する [455]。

7. 削除ファイル（Delete Files）の処理とスキャンタスクの構築 [457]

  1. 削除ファイルの必要性:

    1. 削除ファイルは、既存のデータファイル内の特定の行を「存在しない」とマークするために使用される [458]。
    2. クエリでデータファイルを読み取る際、エンジンはこれらの削除を適用する必要がある [459]。

  2. スキャンタスクの構築:

    1. スキャンタスク（Scan Task）は、読み取る必要があるデータファイルと、それに適用する必要がある削除ファイルを組み合わせたものである [460]。
    2. これはManifestGroup内で実行され、これまでのフィルタリング作業の結果（処理が必要なデータファイルのリスト）を使用する [463]。

  3. 削除ファイルインデックスの構築:

    1. 削除ファイルは、データファイルとは別に、Manifest Groupの初期化時に読み込まれた削除Manifestから処理される [466]。
    2. データManifestに対して行われたフィルタリング（上限/下限チェック、パーティションチェック）と全く同じことが削除Manifestにも行われる [475]。
    3. データファイルのプランニングを開始する直前に、削除ファイルインデックス（Delete File Index）が構築される [479]。
    4. このインデックスは、パーティションごとに、Equality DeletesとPosition Deletesのマップを構築する [481]。
    5. delete file indexは、特定のデータファイルに対して適切な削除ファイルのセットを決定するために、これらのマップを検索する [485]。

  4. 削除ファイルの検索ロジック:

    1. Equality Deletesの検索:
    • すべてのデータファイルに適用されるグローバルなEquality Deletes、または特定のパーティションに適用されるEquality Deletesを検索する [487]。
    • ファイルが追加された時期を示すLamport's Clockであるシーケンス番号を使用して、削除ファイルが現在のスナップショットに適用されるかどうかを判断する [491]。
    1. Delete Vectorsの検索:
    • V3で導入された新しい機能であり、1つのファイルに対して1つのDelete Vectorのみを持つことができるため、検索は非常に迅速である [492]。
    1. Position Deletesの検索:
    • Delete Vectorがない場合、Position Deletesをチェックする必要がある [496]。
    • Position Deletesは、削除する行のリストとパスを持つParquetファイルであり、データファイルとパスを関連付け、削除する行識別子を取得する [497]。
    • Delete Vectorsは、削除する行を示すビットセットである [501]。

  5. 最終的なタスクの計画:

    1. 削除ファイルが決定された後、最終ステップとしてplanTasks（TableScanUtil内）が実行される [504]。
    2. これは、定義された分割サイズ（Split Size）に基づいて、スキャンタスクを分割または結合する方法である [505]。
    3. データファイルが大きすぎる場合は分割し、小さすぎる場合は結合して、特定のエンジンタスクでより多くのデータを読み込むようにする [508]。
    4. この後、タスクはSparkにシリアライズされ、Spark固有のロジックでメモリ形式に読み込まれる [510]。
    5. Icebergはすべてのプランニングを完了しており、Sparkに渡されたファイルは読み込まれる必要がある（ただし、LIMITなどの場合は途中で終了する可能性がある） [511]。

8. Apache Iceberg V4での将来の変更点とコミュニティ活動 [513]

  1. 現在の実装の課題とV4での変更計画:

    1. 列統計情報の拡張:
    • 現在のマップ形式は、カラム型に適しておらず、型情報がないため、型拡張以外の型昇格が事実上不可能である [519]。
    • V4では、列統計情報の格納方法を拡張する予定である [518]。
    1. Manifest Listの廃止:
    • Manifest Listで行っているロジックのほとんどが、データファイルに対して行っているロジックの拡張版である [523]。
    • 提案されているのは、Manifest List自体を別の種類のManifestにし、ロジックを普遍化することである [524]。
    1. メタデータへの削除ベクトルの適用:
    • 現在のセットアップでは、子Manifestを変更するたびに、ファイル全体を削除して新しい行で書き直す必要がある [527]。
    • 代わりに、メタデータにも削除ベクトルを適用できるようにし、ファイルを削除・書き換えせずにメタデータからエントリを削除できるようにする [527]。
    1. 評価ロジックの維持: これらの変更にもかかわらず、Manifest評価の仕組み自体は基本的に同じままであり、その概念がより多くのものに拡張されると予想される [529]。

  2. コミュニティ活動と将来の機能:

    1. オープンソースコミュニティ: Apache Icebergはオープンソースであり、誰でも参加できる [533]。
    2. ワイドテーブルへの対応:
    • AIの重要性が増すにつれて、フィーチャセットを格納するためのワイドテーブルが増加している [593]。
    • Parquetなどのファイル形式では、同じ行グループのすべての列ベクトルを同じファイルに保持する必要があるため、列数が非常に多いと、ファイルが非常にワイドになり、カラム型の利点が失われる問題がある [594]。
    • シスターファイル（Sister Files）のような概念（メインファイルとは別のファイルに追加の列を格納する）が検討されている [599]。
    • Icebergコミュニティでは、ファイル形式の部分をプラグイン可能にするための主要な提案が進行中であり、これによりこれらのアイデアを試す道が開かれる [604]。
    1. 互換性への注力: Icebergコミュニティは、エンジン間の互換性を最優先している [609]。
    • 他のフォーマット（例: Hoodie）に見られる一部の機能（例: 読み取り時の上書きを伴うシスターファイル）は、一部のエンジンでの実装が困難であると判断されるため、Icebergでは採用を控えている場合がある [606]。

  3. 質疑応答からの追加情報:

    1. 述語プッシュダウンとパーティションチェック: Icebergは、述語がパーティションに対して機能するように変換可能であれば、常にそうする [553]。
    • 例として、タイムスタンプに対する述語は、パーティションメトリクスに合わせて変換される [557]。
    • これはIcebergの隠されたパーティショニングの一部であり、ユーザーはテーブルがどのようにパーティショニングされているかを知る必要がない [559]。
    1. Puffin Files: Puffin Filesは、Iceberg仕様への比較的最近の追加であり、現時点では主に削除ベクトルを格納する場所である [575]。
    • また、異なる値の数など、テーブルの他の側面を記述する追加のメトリクスを格納でき、これらは他のAPIを通じてエンジンに提供される [577]。
    1. Polaris: Polarisは、Iceberg仕様の実装ではなく、Iceberg RESTカタログの実装である [587]。