pg_lake・pg_incremental を使って Snowflake Postgres の Iceberg テーブルへの増分同期を試してみた

[Snowflake Postgres]
  通常テーブル
       ↓ pg_incremental（time_interval ごとに増分を書き出し）
  Iceberg テーブル（S3 ※Snowflake 管理）
       ↓ カタログ統合（REFRESH_INTERVAL_SECONDS ごとにメタデータを更新）
[Snowflake]
  Iceberg テーブル（AUTO_REFRESH = TRUE）※読み取り専用

export PGHOST=<エンドポイント>.postgres.snowflake.app
export PGPORT=5432
export PGUSER=snowflake_admin
export PGPASSWORD=<パスワード>
export PGDATABASE=postgres

psql -c "SELECT version();"
                                                     version                                                     
-----------------------------------------------------------------------------------------------------------------
 PostgreSQL 18.4 on aarch64-unknown-linux-gnu, compiled by gcc (GCC) 11.5.0 20240719 (Red Hat 11.5.0-14), 64-bit

SELECT name, default_version, installed_version
FROM pg_available_extensions
WHERE name IN ('pg_lake', 'pg_incremental', 'pg_cron');
      name      | default_version | installed_version 
----------------+-----------------+-------------------
 pg_incremental | 1.5             | 
 pg_lake        | 3.3             | 
 pg_cron        | 1.6             | 

CREATE TABLE orders (
  id          BIGSERIAL PRIMARY KEY,
  customer_id INT           NOT NULL,
  amount      NUMERIC(10,2) NOT NULL,
  status      TEXT          NOT NULL DEFAULT 'pending',
  created_at  TIMESTAMPTZ   NOT NULL DEFAULT NOW(),
  updated_at  TIMESTAMPTZ   NOT NULL DEFAULT NOW()
);

INSERT INTO orders (customer_id, amount, status)
SELECT
  (random() * 1000)::INT,
  (random() * 10000)::NUMERIC(10,2),
  (ARRAY['pending','processing','completed','cancelled'])[floor(random()*4+1)]
FROM generate_series(1, 10000);

-- pg_lake 有効化（CASCADE で依存拡張も一括インストール）
CREATE EXTENSION pg_lake CASCADE;

-- Iceberg テーブル作成（orders の同期先）
CREATE TABLE orders_iceberg (
  id          BIGINT,
  customer_id INT,
  amount      NUMERIC(10,2),
  status      TEXT,
  created_at  TIMESTAMPTZ,
  updated_at  TIMESTAMPTZ
) USING iceberg;

\dE
                     List of relations
 Schema |      Name      |     Type      |      Owner
--------+----------------+---------------+-----------------
 public | orders_iceberg | foreign table | snowflake_admin

-- pg_cron / pg_incremental 有効化
CREATE EXTENSION pg_cron;
CREATE EXTENSION pg_incremental;

-- パイプライン作成（1分ごとに updated_at ベースで増分コピー）
SELECT incremental.create_time_interval_pipeline(
  pipeline_name     := 'sync_orders_to_iceberg',
  time_interval     := '1 minute',
  source_table_name := 'orders',
  start_time        := (SELECT MIN(updated_at) FROM orders),
  command           := $$
    INSERT INTO orders_iceberg
    SELECT id, customer_id, amount, status, created_at, updated_at
    FROM orders
    WHERE updated_at >= $1 AND updated_at < $2
  $$
);

-- パイプライン一覧
SELECT * FROM incremental.pipelines;

-- time_interval パイプラインの詳細
SELECT * FROM incremental.time_interval_pipelines;
     pipeline_name      | time_interval | batched | min_delay |  last_processed_time   
------------------------+---------------+---------+-----------+------------------------
 sync_orders_to_iceberg | 00:01:00      | t       | 00:00:30  | 2026-06-28 06:36:00+00

CREATE OR REPLACE CATALOG INTEGRATION postgres_iceberg_integration
  CATALOG_SOURCE    = SNOWFLAKE_POSTGRES
  TABLE_FORMAT      = ICEBERG
  CATALOG_NAMESPACE = 'public'
  REST_CONFIG = (
    POSTGRES_INSTANCE      = '<インスタンス名>'
    CATALOG_NAME           = 'postgres'
    ACCESS_DELEGATION_MODE = VENDED_CREDENTIALS
  )
  ENABLED = TRUE;

CREATE DATABASE IF NOT EXISTS pglake;
USE DATABASE pglake;

CREATE OR REPLACE ICEBERG TABLE orders_from_postgres
  CATALOG            = 'postgres_iceberg_integration'
  CATALOG_TABLE_NAME = 'orders_iceberg'
  AUTO_REFRESH = TRUE;

SELECT COUNT(*) FROM orders_from_postgres;
SELECT * FROM orders_from_postgres LIMIT 5;

-- 【Postgres】INSERT（例：10,000 行）
\timing on

> INSERT INTO orders (customer_id, amount, status, created_at, updated_at)
SELECT
  (random() * 1000)::INT,
  (random() * 10000)::NUMERIC(10,2),
  'batch_10k',
  NOW(),
  NOW()
FROM generate_series(1, 10000);
INSERT 0 10000
Time: 34.695 ms

-- 【Postgres】INSERT 完了時刻を記録（updated_at = NOW() で挿入したため、レコードの updated_at と一致する）
> SELECT NOW() AS insert_time, COUNT(*) FROM orders WHERE status = 'batch_10k';
          insert_time          | count 
-------------------------------+-------
 2026-06-28 16:17:17.087108+09 | 10000

-- 【Snowflake】反映確認
> SELECT COUNT(*) AS sf_count FROM orders_from_postgres WHERE status = 'batch_10k';
+----------+
| SF_COUNT |
|----------|
|    10000 |
+----------+

> SELECT CURRENT_TIMESTAMP() AS visible_time;
+-------------------------------+
| VISIBLE_TIME                  |
|-------------------------------|
| 2026-06-28 16:19:27.165 +0900 |
+-------------------------------+