dlt でロードしたデータを SQLMesh で扱ってみる

表題の作業を実際に行ってみました。問題点や、その一部の解消方法をメモしました。

## はじめに

### SQLMesh とは

SQLMeshは「dbt に代わるデータ変換ツール」として注目されています。
Tobiko Data社を中心にOSSとして開発されているほか、[Tobiko Cloud](https://tobikodata.com/product.html)としてマネージドサービスも提供しています。

SQLMeshの主な特徴は以下の通りです。
参考： [Tobiko - dbt tag](https://tobikodata.com/tag/dbt.html)

#### virtual data environments

開発環境、検証環境などを view の切り替えを使って、実体データの複製を最小限に止める仕組みです。
（[Copy-On-Write](https://ja.wikipedia.org/wiki/%E3%82%B3%E3%83%94%E3%83%BC%E3%82%AA%E3%83%B3%E3%83%A9%E3%82%A4%E3%83%88)に似た、「変更が生じたテーブルのみ新しく作成し、その他は既存のテーブルを参照させる方法」をとっているようです）
開発環境で動作確認したものを本番環境に適用する場合や、切り戻しをする場合も、view の参照先を切り替えるだけなので、短時間で行えます。

「本物のデータを使わないと検証にならない」というのはよくある状況だと思いますが、モデルの開発者には個人情報へのアクセスを制限する必要があるなど、セキュリティやガバナンス観点で環境分離を行なっている場合も多いと思います。
virtual data environment による環境分離だけでモデル開発を進めるかどうか、組織のポリシー次第になるのではないでしょうか。

SQLMesh でも gateway という接続先を分離することで複数の環境を分けることも可能で、その辺りの手順がこちらのページに解説されています。

[Isolated systems guide - SQLMesh](https://sqlmesh.readthedocs.io/en/stable/guides/isolated_systems/)


#### column-level lineage

SQLGlot というライブラリを内部で利用しており、クエリの内容をAST（抽象構文木）レベルで解釈できるので、カラムレベルでの系統情報を扱うことができます。
また、モデルの変更を行う際に、その変更が下流のモデルに影響するのかどうかをシステムが推定し、必要なものだけを再作成することができます。下流のモデルをすべて再作成するのに比べて、計算のコストを低減することができます。


#### state awareness

dbt の incremental なマテリアライズ方式では、「既存のデータに含まれる最新の日時以降に、上流で増加したレコード」を取り込むことができます。SQLMeshではモデルごとに処理対象の時間間隔や、実行の頻度を指定し（interval approach）、「Modelでどの時間区分を処理したか」というstateを追跡管理します。
差分更新のモデルであっても、過去のデータを部分的に再処理したり、最初に大きなデータを取り込む場合にも分割して処理することができます。


### dlt とは

検索すると、Databricks の Delta Live Tables の結果が混じってくるので厄介ですが、"data load tool" の略称で、データロードを行うオープンソースのPythonライブラリです。

[Introduction | dlt Docs](https://dlthub.com/docs/intro)

「Pythonのライブラリ」なので、Notebook上でも動かせます。また、スクリプトを書けば、さまざまなデータソースを簡単に取り込むことができます。

データの構造や型から、ロードするテーブルのスキーマを自動的に作成してくれます。



## [dlt](https://dlthub.com/docs/intro) でのデータロード

下記のようなサンプルのデータロードスクリプトを用意しました。
duckdb と snowflake の両方に、同じデータをロードします。

```python
import dlt


DUCKDB_PATH = "./data/dialect_check_data.duckdb"

pipeline_snow = dlt.pipeline(
    pipeline_name="dialect_check_snowflake",
    destination="snowflake",
    dataset_name="dia_check_raw",
    dev_mode=False,
)

duckdb_destination = dlt.destinations.duckdb(
        DUCKDB_PATH,
        enable_dataset_name_normalization=False)
#breakpoint()
pipeline_duck = dlt.pipeline(
    pipeline_name="dialect_check_duckdb",
    destination=duckdb_destination,
    dataset_name="dia_check_raw",
    dev_mode=False,
)

data = [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]

if __name__ == "__main__":
    for pipeline in [pipeline_snow, pipeline_duck]:
    #for pipeline in [pipeline_snow, ]:
        load_info = pipeline.run(
            data,
            table_name="items",
            write_disposition={
                "disposition": "append",
            },
            refresh="drop_resources"
        )
        print(load_info)
```

dlt の接続情報は、`project_root/.dlt/secrets.toml` に保存します。

この結果、duckdb 側では dataset name である `dia_check_raw` が小文字のままカタログ名として利用されます。
![duckdb_catalog_name.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/564576/75707eec-a024-4a3d-972a-cdc97b4922cb.jpeg)

一方、snowflake 側では 大文字の `DIA_CHECK_RAW` がスキーマ名として利用されます。
![snowflake_schema_name.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/564576/9dad9f80-ead3-4cd0-9ba8-4f9fc3cb2b23.jpeg)

SQLMesh では、sqlglotを利用して、1つの方言（dialect）で書かれたModelファイルで複数のDWHを管理することができる [^1]  設計になっているのですが、Modelで読みに行く先のカタログ名/スキーマ名のCaseが異なる、という事象に対応する必要が生じます。

[^1]: [Tobiko - Automated Data Warehouse Migrations](https://tobikodata.com/automated_data_warehouse_migrations.html)

## [SQLMesh](https://sqlmesh.com/) によるモデルの構築

パイプライン名を確認します
```shell
% pdm run dlt pipeline --list-pipelines
11 pipelines found in /Users/m-nakamura/.dlt/pipelines
dialect_check_snowflake
dialect_check_duckdb

...

```

パイプライン名を指定して、SQLMesh の modelを自動生成することができます。
参考： [dlt - SQLMesh](https://sqlmesh.readthedocs.io/en/stable/integrations/dlt/?h=dlt)

以下の例は、`raw_csv_import_dev` という dlt pipeline をもとに、duckdb dialect の SQLMesh モデルを生成するコマンドです。

```shell
% pdm run sqlmesh init -t dlt --dlt-pipeline raw_csv_import_dev duckdb
```

以下のようなファイルが生成されます
```
.
├── audits
├── config.yaml
├── macros
├── models
│   ├── incremental__dlt_loads.sql
│   └── incremental_items.sql
├── seeds
└── tests
```

（incremental__dlt_loads.sql は dlt で利用するメタデータなので不要かもしれません）

incremental_items.sql  は以下のような内容です。
kind が差分更新になっています。

参考： [Model kinds - SQLMesh](https://sqlmesh.readthedocs.io/en/stable/concepts/models/model_kinds/#incremental_by_time_range)

```sql
MODEL (
  name dia_check_raw_sqlmesh.incremental_items,
  kind INCREMENTAL_BY_TIME_RANGE (
    time_column _dlt_load_time,
  ),
);

SELECT
  CAST(c.id AS BIGINT) AS id,
  CAST(c.name AS TEXT) AS name,
  CAST(c._dlt_load_id AS TEXT) AS _dlt_load_id,
  CAST(c._dlt_id AS TEXT) AS _dlt_id,
  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) as _dlt_load_time
FROM
  dia_check_raw.items as c
WHERE
  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
```

SQLMesh で利用する接続の情報は、`project_root/config.yaml` に保存します。

SQLMesh では、複数の独立したシステム（本番環境 / 検証環境 のようなもの）に対応することができます。「仮想環境」という用語と混乱を招く恐れがあるため、SQLMeshのドキュメントの中では 'isolated system' と呼ばれています。
[Isolated systems guide - SQLMesh](https://sqlmesh.readthedocs.io/en/stable/guides/isolated_systems/)
具体的には、`config.yaml` のなかに複数の `gateway` を定義し、切り替えて利用することができます。

作成する Model  には SQL の dialect を設定することができ、gateway で指定する接続先のタイプに応じて Model の SQL が sqlglot によって変換され、実行される、というコンセプトになっているようです。
[tobymao/sqlglot: Python SQL Parser and Transpiler](https://github.com/tobymao/sqlglot)

## 複数gatewayの罠

複数のgatewayを利用するには、いくつか問題がありました。

### 大文字で作成される Snowflake 上のスキーマ名が見つからない問題

dlt で作成されるソーステーブルは、Snowflake上では大文字で作成されるのですが、sqlglot で変換されるクエリでは小文字のスキーマ名を探しに行くため、クエリが失敗してしまいます。

この問題は、「gateway ごとの model_defaults.dialect で normalization_strategy を指定できる機能」によって回避できます。

具体的には、下記のようなオプションを `config.yaml` に指定します。

```
gateways:
  duckdb:
    connection:
      type: duckdb
      database: /path/to/dialect_check_data.duckdb
  snowflake:
    connection:
      type: snowflake
      account: xxxx
      user: xxxx
      private_key: |
        -----BEGIN PRIVATE KEY-----
        some_private_key
        ==
        -----END PRIVATE KEY-----
      database: xxxx
      warehouse: xxxx
      role: xxxxxx
    model_defaults:
      #see  https://github.com/tobymao/sqlglot/blob/main/sqlglot/dialects/dialect.py#L92
      # uppercase means "Unquoted identifiers are uppercased."
      dialect: duckdb, normalization_strategy=uppercase  <- ココ

default_gateway: duckdb

model_defaults:
  dialect: duckdb
  start: 2025-02-24
```

[Feat: Add model defaults support per gateway configuration by VaggelisD · Pull Request #3888 · TobikoData/sqlmesh](https://github.com/TobikoData/sqlmesh/pull/3888)


### SHA2 / SHA256 など、SQLエンジンによって名前が異なる関数がある問題

```python
Python 3.13.1 (main, Dec  3 2024, 17:59:52) [Clang 16.0.0 (clang-1600.0.26.4)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import sqlglot
>>> sqlglot.__version__
'26.7.0'
>>> sqlglot.parse_one("SELECT sha256(col1)", dialect="duckdb").sql(dialect="snowflake")
'SELECT SHA256(col1)' #<- 本来は SHA2(col1)になるべき
```

これは、SQLGlotのduckdb dialect の Parser に問題があったことが原因でした。

```python
>>> sqlglot.parse_one("SELECT sha256(col1)", dialect="duckdb")
Select(
  expressions=[
    Anonymous(   #<-  ここがSHA2として認識されないと、snowflake dialect として正しく変換されない
      this=sha256,
      expressions=[
        Column(
          this=Identifier(this=col1, quoted=False))])])
```

以下のチケットで現在は修正されています。
Parser 
チケット： [The sha256 function is not recognized correctly in duckdb dialect. · Issue #4815 · tobymao/sqlglot](https://github.com/tobymao/sqlglot/issues/4815)


### TO_TIMESTAMP() の引数が、duckdbだとdoubleを受け付けるが、snowflakeだとintしか受け付けない問題

Model の中で使われている `TO_TIMESTAMP()`関数の引数が、duckdb だと double だが、snowflake だと int という差異がありました。

そのため、modelに下記のような記述があると、snowflake側で実行エラーになってしまいました。

```sql
TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) as _dlt_load_time
```

snowflake dialect の Generator では、TO_TIMESTAMP は [以下のように書かれており](https://github.com/tobymao/sqlglot/blob/v26.6.0/sqlglot/dialects/snowflake.py#L986)、単にrenameだけしているようでした。

```python
exp.UnixToTime: rename_func("TO_TIMESTAMP"),
```

この問題は、duckdbのTO_TIMESTAMP() 関数が int も受け付けることがわかったため、以下のようにmodelのクエリを変更することで解消できました

```sql
TO_TIMESTAMP(CAST(c._dlt_load_id AS INT)) as _dlt_load_time
```
（おそらく、sqlglot の snowflake dialect のGeneratorを変更しても対応可能だったと思います）
参考：
* [Timestamp with Time Zone Functions – DuckDB](https://duckdb.org/docs/stable/sql/functions/timestamptz.html#to_timestampdouble)
* [TO_TIMESTAMP / TO_TIMESTAMP_* | Snowflake Documentation](https://docs.snowflake.com/ja/sql-reference/functions/to_timestamp)

## 現時点での所感

「sqlglot を使い、複数のSQLエンジンを一つのModelファイルでサポートする」というコンセプトは、（dialectの対応でハマる恐れはあるものの）ローカルのduckdbをモデルの開発に使うことができるので非常に魅力的だと感じました。
また、[sqlmesh の仮想環境](https://sqlmesh.readthedocs.io/en/stable/concepts/plans/#plan-application) は、一見複雑そうに感じましたが、「仮想環境の切り替えはView の定義の切り替えに相当する」と考えれば、コントロール可能である、という印象をもちました。