Snowflake-managed Apache Iceberg™ table における STREAM 機能を試してみた

## 概要

Snowflake-managed Apache Iceberg™ テーブルでサポートされている STREAM 機能について検証し、結果と検証コードを共有します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7a6c95f3-b4f9-4b29-9754-a8998aacfafb.png)

出所：[Apache Iceberg™ テーブル | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/tables-iceberg)

Snowflake のドキュメントでは「ほとんどのシナリオにおいて CDC（変更データキャプチャー）の記録ソースとして適切」とされていますが、厳密には CDC を実現するための機能ではないことに注意してください。

> ほとんどのシナリオで CDC 記録の適切なソースです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b578423e-297d-4a88-bc6b-6bbba0ede207.png)

出所：[ストリームの紹介 | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/streams-intro#changes-clause-read-only-alternative-to-streams)

なお、STREAM からテーブルへ DML 処理を実行した際に STREAM のデータが消失する（これを「データの消費」と呼ぶ）仕様である点も押さえておきましょう。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/28ecc479-d7c8-47e1-a8d2-30fcf1ffca50.png)

出所：[ストリームの紹介 | Snowflake Documentation](https://docs.snowflake.com/ja/user-guide/streams-intro)

本記事は「SnowflakeにおけるApache Iceberg機能の完全ガイド」シリーズの一部です。  
Apache Iceberg機能の全体像やSnowflake上での活用方法について詳しく知りたい方は、以下の記事をご覧ください。

- [SnowflakeにおけるApache Icebergの機能完全ガイド：Apache Iceberg on Snowflake 活用の第一歩 #Spark - Qiita](https://qiita.com/manabian/items/7ba563b614e8ae016cca)

## 事前準備

以下の記事で紹介している環境構築手順を事前に実施してください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5a727271-1090-4694-bf8c-b2a493fd0e56.png)

出所：[Snowflake Catalog における Iceberg テーブルの基本的な操作手順 #Spark - Qiita](https://qiita.com/manabian/items/69761bde791d0f014246)

## データベース、テーブル、および STREAM の作成

### データベースを作成

```sql
CREATE OR REPLACE DATABASE iceberg_stream_db;
USE DATABASE iceberg_stream_db
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/9e2c13da-a4eb-483c-9d13-5c989621103d.png)

### STREAM 元のテーブルを作成

```sql
-- stream 元のテーブルを作成
CREATE OR REPLACE ICEBERG TABLE source_table (
  id    INT,
  name  VARCHAR,
  status VARCHAR
)
    CATALOG = 'SNOWFLAKE'
    EXTERNAL_VOLUME = 'my_azure_sf_volume'
    BASE_LOCATION = 'source_table';
SHOW TABLES IN DATABASE iceberg_stream_db;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/29428d44-4b9d-4e20-845b-6ed2106cccdc.png)

### STREAM を作成

```sql
CREATE OR REPLACE STREAM source_table__stream
    ON TABLE source_table;
SHOW STREAMS IN DATABASE iceberg_stream_db;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/a19f8a30-3e94-4af6-8d17-9bcafe42dd31.png)

### STREAM の書き込み先のテーブルを作成

STREAM から `CTAS` で直接テーブルを作成しようとすると、Iceberg テーブルのカラム名の制約によってエラーが発生しました。

```sql
-- STREAM の書き込み先のテーブルを作成
CREATE OR REPLACE ICEBERG TABLE source_table__change
    CATALOG = 'SNOWFLAKE'
    EXTERNAL_VOLUME = 'my_azure_sf_volume'
    BASE_LOCATION = 'source_table__change'
    AS
    SELECT * FROM source_table__stream;
```

>  091372: SQL compilation error: 'METADATA$ACTION, METADATA$ISUPDATE, METADATA$ROW_ID' cannot be used as column name(s) for Iceberg tables. METADATA$ and SYSTEM$ column name prefixes are reserved.

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1a0382df-0124-476f-acda-df0856f1c7d0.png)

そこで、`$` を `__` に置き換えてテーブルを作成しました。

```sql
CREATE OR REPLACE ICEBERG TABLE source_table__change
    CATALOG = 'SNOWFLAKE'
    EXTERNAL_VOLUME = 'my_azure_sf_volume'
    BASE_LOCATION = 'source_table__change'
    AS
    SELECT
        * EXCLUDE (
            METADATA$ACTION,
            METADATA$ISUPDATE,
            METADATA$ROW_ID        
        ),
        METADATA$ACTION AS METADATA__ACTION,
        METADATA$ISUPDATE AS METADATA__ISUPDATE,
        METADATA$ROW_ID AS METADATA__ROW_ID
        FROM 
            source_table__stream;
```

```sql
SHOW TABLES IN DATABASE iceberg_stream_db;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ef9888f0-87c5-4d70-86cb-cd6f9b21e1aa.png)

## 検証コードと結果

### STREAM の初期状態を確認

```sql
-- STREAM の初期状態をチェック
SELECT
    *
    FROM
        source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/dd53c43c-3114-455c-88a8-26ff20217226.png)

### INSERT 時の動作確認

まず、INSERT したレコードが全て STREAM に格納されることを確認します。

```sql
INSERT INTO source_table (id, name, status) VALUES
  (1,'Alice','active'),
  (2,'Bob','active'),
  (3,'Charlie','active'),
  (4,'Daisy','active'),
  (5,'Ethan','active');
SELECT 
    *
    FROM source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ab093e5b-cf16-4186-bb95-31d3f27581d6.png)

次に、STREAM のデータを保管用テーブルに INSERT して「消費」します。

```sql
-- データを消費
INSERT INTO source_table__change
(
    ID,
    NAME,
    STATUS,
    METADATA__ACTION,
    METADATA__ISUPDATE,
    METADATA__ROW_ID
)
SELECT
    * EXCLUDE (
        METADATA$ACTION,
        METADATA$ISUPDATE,
        METADATA$ROW_ID        
    ),
    METADATA$ACTION AS METADATA__ACTION,
    METADATA$ISUPDATE AS METADATA__ISUPDATE,
    METADATA$ROW_ID AS METADATA__ROW_ID
    FROM 
        source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/04386766-9dd7-474a-9083-9a9e9a0013f0.png)

### UPDATE 時の動作確認

UPDATE 後のレコードが STREAM に格納されることを確認します。

```sql
-- 2人のステータスを変更（更新）
UPDATE source_table
    SET status = 'inactive' WHERE id IN (2, 3);
SELECT 
    *
    FROM source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/60372ef6-74a5-4491-b60e-986509fd6855.png)

更新後のレコードも同様に保管用テーブルへ INSERT して消費します。

```sql
-- データを消費
INSERT INTO source_table__change
(
    ID,
    NAME,
    STATUS,
    METADATA__ACTION,
    METADATA__ISUPDATE,
    METADATA__ROW_ID
)
SELECT
    * EXCLUDE (
        METADATA$ACTION,
        METADATA$ISUPDATE,
        METADATA$ROW_ID        
    ),
    METADATA$ACTION AS METADATA__ACTION,
    METADATA$ISUPDATE AS METADATA__ISUPDATE,
    METADATA$ROW_ID AS METADATA__ROW_ID
    FROM 
        source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/7cb45cf2-dfd7-42ab-b980-70b322b28fe2.png)

### DELETE 時の動作確認

DELETE したレコードが STREAM に格納されることを確認します。

```sql
DELETE FROM source_table
    WHERE id = 5;
SELECT 
    *
    FROM source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/888ce2a0-5d41-429e-a876-fc35a7935bc3.png)

同様に、保管用テーブルへ INSERT して消費します。

```sql
-- データを消費
INSERT INTO source_table__change
(
    ID,
    NAME,
    STATUS,
    METADATA__ACTION,
    METADATA__ISUPDATE,
    METADATA__ROW_ID
)
SELECT
    * EXCLUDE (
        METADATA$ACTION,
        METADATA$ISUPDATE,
        METADATA$ROW_ID        
    ),
    METADATA$ACTION AS METADATA__ACTION,
    METADATA$ISUPDATE AS METADATA__ISUPDATE,
    METADATA$ROW_ID AS METADATA__ROW_ID
    FROM 
        source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/91453842-608e-4b1e-adcb-1500fc0d3de9.png)

### MERGE 時の動作確認

MERGE（UPSERT）されたレコードが STREAM に格納されることを確認します。

```sql
MERGE INTO source_table AS tgt
USING (
  SELECT 1 AS id, 'Alice_updated' AS name, 'inactive' AS status
  UNION ALL
  SELECT 6 AS id, 'Frank' AS name, 'active' AS status
) AS src
ON tgt.id = src.id
WHEN MATCHED THEN
  UPDATE SET tgt.name = src.name, tgt.status = src.status
WHEN NOT MATCHED THEN
  INSERT (id, name, status)
  VALUES (src.id, src.name, src.status);
SELECT 
    *
    FROM
        source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b13de9b5-a412-4ad1-b68d-18d6313e0345.png)

その後、保管用テーブルへ消費します。

```sql
-- データを消費
INSERT INTO source_table__change
(
    ID,
    NAME,
    STATUS,
    METADATA__ACTION,
    METADATA__ISUPDATE,
    METADATA__ROW_ID
)
SELECT
    * EXCLUDE (
        METADATA$ACTION,
        METADATA$ISUPDATE,
        METADATA$ROW_ID        
    ),
    METADATA$ACTION AS METADATA__ACTION,
    METADATA$ISUPDATE AS METADATA__ISUPDATE,
    METADATA$ROW_ID AS METADATA__ROW_ID
    FROM 
        source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/308d4c0d-15d3-4c12-a247-9adce99fd305.png)

### INSERT と DELETE を実行した時の動作確認

INSERT した後に DELETE を行ったレコードが、STREAM に格納されるケースを確認します。

```sql
INSERT INTO source_table (id, name, status) VALUES
  (6,'Franky','active');
SELECT 
    *
    FROM source_table__stream;
-- 1人のレコードを削除
DELETE FROM source_table WHERE id = 6;
SELECT 
    *
    FROM source_table__stream;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/36268db7-7a38-4970-aed6-da96ac2cb0a9.png)

:::note warn
前日に同様のコードを実行した際は、STREAM に書き込まれないケースがありました。どちらが正式な仕様であるかが不明ですが、最新の結果を検証結果としました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/64fe1747-259d-409f-8ebb-70a40fdc6805.png)
:::

保管用テーブルへデータを INSERT して消費します。

```sql
-- データを消費
INSERT INTO source_table__change
(
    ID,
    NAME,
    STATUS,
    METADATA__ACTION,
    METADATA__ISUPDATE,
    METADATA__ROW_ID
)
SELECT
    * EXCLUDE (
        METADATA$ACTION,
        METADATA$ISUPDATE,
        METADATA$ROW_ID        
    ),
    METADATA$ACTION AS METADATA__ACTION,
    METADATA$ISUPDATE AS METADATA__ISUPDATE,
    METADATA$ROW_ID AS METADATA__ROW_ID
    FROM 
        source_table__stream;
```

### CTAS の動作確認

CTAS（`CREATE TABLE AS SELECT`）を実行すると、STREAM にデータは書き込まれず、以下のエラーが発生しました。

```sql
CREATE OR REPLACE ICEBERG TABLE source_table
    EXTERNAL_VOLUME = 'my_azure_sf_volume'
    CATALOG = 'SNOWFLAKE'
    BASE_LOCATION = 'source_table'
    AS
    SELECT 1 AS id, 'Alice' AS name, 'active' AS status;
SELECT 
    *
    FROM source_table__stream;
```

> 091901: Base table 'ICEBERG_STREAM_DB.PUBLIC.SOURCE_TABLE' dropped, cannot read from stream 'SOURCE_TABLE__STREAM' in line 3 position 9.

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b8684ed4-3daa-4ca3-b7ff-0856d5b7fabf.png)