Azure HorizonDB 用 pg_durable による耐久性機能 (プレビュー)

pg_durable は、Azure HorizonDB 内の永続実行エンジンです。これにより、実行時間の長いマルチステップ SQL ワークフロー (パイプライン、ETL ジョブ、AI 呼び出し、スケジュールされたジョブ、承認フローの埋め込み) を定義し、Postgres を離れることなく、Durable Functions などの専用オーケストレーターから期待されるのと同じ信頼性保証で実行できます。

pg_durable は、永続的な AI パイプラインの下の実行レイヤーでもあります。 AI パイプラインを使用しているなら、クラッシュしても耐え、障害時に再試行し、最後に完了したステップから再開できるようにするのが pg_durable です。

Note

pg_durable は プレビュー段階です。

"永続的" とは

pg_durableの永続的な関数は、すべての手順でディスクに保持されます。これにより、プレーンな BEGIN ... COMMIT ブロックまたは cron ジョブから取得できない特定の保証セットが提供されます。

データベースのクラッシュと再起動が発生しても存続します。 完了した手順は、サーバーがバックアップされたときに再実行されません。進行中の手順は、最後のチェックポイントから再開されます。保留中の手順は、ワーカーがオンラインに戻ったときに実行されます。
長時間の待機に耐えます。 ワークフローは、何時間もスリープ状態にしたり、cron スケジュールを待ったり、外部信号をブロックしたりして、中断したところから再開することができます。
障害が発生しても存続します。 失敗したステップは、関数全体を再実行せずに自動的に再試行できます。
ID をキャプチャします。 関数は、バックグラウンドワーカーの特権ではなく、起動したユーザーの特権を使用して実行されます。マルチテナントワークロードは分離された状態を維持します。
SQL から監視可能な状態を維持します。 HorizonDB の他のすべてに使用するのと同じインターフェイス ( SELECT ステートメント) を使用して、状態、履歴、実行数、および出力を検査できます。

耐久性が自動的には行わないこと: これだけでは、非べき等な外部操作を単独で再試行しても安全とはなりません。ステップが料金の発生する外部 API を呼び出す場合は、ステップが冪等になるように設計してください（たとえば、冪等性キーを渡します）。

pg_durableを使用する場合

次の作業を行う必要がある場合は、 pg_durable を使用します。

途中で失敗するまでにかなり時間がかかることがあります（数百万行に対する埋め込み生成、複数ステップのETLジョブ、バックフィルなど）。
既に成功した部分をやり直さずに、障害発生時に再試行する必要があります。
スケジュールに従って実行する必要があります (毎時間、平日の午前 9 時)。
外部イベント (承認、Webhook、別のシステムからのシグナル) を待機する必要があります。
分岐、結合、または競合実行を用いて、複数のステップを連携させます。
現在、外部オーケストレーター + Postgres データベースとして実装されています。ほとんどの作業はデータベースパーツです。

ワークロードが単一の短いトランザクションステートメントである場合は、 pg_durable必要はありません。通常の INSERT / UPDATEを使用します。

どのように機能するのか

永続的関数は、SQL DSL を使用して構築し、 df.start()で送信する手順のグラフです。グラフは永続化され、バックグラウンドワーカーによって実行されます。

2 つの重要なアイデア:

関数グラフと実行状態は、HorizonDB 自体の df および duroxide スキーマに格納されます。バックアップ、ポイントインタイムリストア、高可用性はすべて、ワークフローの状態に自動的に適用されます。別個のオーケストレーター状態を管理する必要はありません。
バックグラウンドワーカーは、 shared_preload_librariesによって開始されます。 CREATE EXTENSION後に拡張機能を検出し、関数の実行を開始します。データベースが再起動すると、ワーカーは実行中のインスタンスに再アタッチして再開します。

Note

pg_durable 内の実行エンジンは、Rust (Durable Task Framework とテンポラルから着想を得た) Microsoftのオープンソースの永続的な実行ランタイムである Duroxide 上に構築されています。 duroxide スキーマ名は、Duroxide がオーケストレーション履歴、関連付け ID、および再生状態を保持する場所です。 pg_durable から得られる決定論的リプレイ、相関イベントID、永続タイマーの保証は、Duroxide に直接由来します。

pg_durableを有効にする

Azure HorizonDB で pg_durable を有効にするには、最初にパラメーターグループを構成してから、各データベースに拡張機能を作成します。

次のセットアップ記事を使用します。

サーバーのパラメーターグループを作成します。
shared_preload_librariesを含むようにpg_durableを設定します。
azure.extensionsを含むようにpg_durableを設定します。
パラメーターグループをサーバーに適用します。
各ターゲットデータベースに接続し、次のコマンドを実行します。

使用する各データベースに拡張機能を作成します。

CREATE EXTENSION IF NOT EXISTS pg_durable;

CREATE EXTENSION では、 df スキーマ (関数グラフと監視ビュー) と duroxide スキーマ (実行状態) がプロビジョニングされます。バックグラウンドワーカーは数秒で拡張機能を検出し、関数を実行する準備ができています。

最初の永続的な関数

-- Start a one-step durable function
SELECT df.start('SELECT ''Hello, durable world!''');
-- Returns an 8-character instance ID, for example: a1b2c3d4

-- Check status
SELECT df.status('a1b2c3d4');

-- Get the result
SELECT df.result('a1b2c3d4');

1 ステップ関数でも永続的です。 df.start() 後、ワーカーが選択する前にデータベースが再起動しても、関数は引き続き実行されます。

Note

df.start() はワークフローを非同期的に送信し、すぐに返します。複数ステップのワークフローの場合は、 df.list_instances()、 df.instance_info()、 df.status()、または df.result() を使用して、副作用を検証する前に完了を確認します。

プログラムモデル

永続的関数は、ステップ、演算子、および組み込み関数から構築されたグラフです。プレーン SQL 文字列は自動ラップされるため、 df.sql() を明示的に呼び出す必要はありません。

Operators

Operator	意味	例
`~>`	シーケンス - まず左、次に右を実行	`'SELECT 1' ~> 'SELECT 2'`
`&`	Join - 並列で実行し、すべてが完了するまで待機	`'SELECT 1' & 'SELECT 2'`
`\|`	レース - 並列実行し、先に完了した方が勝つ	`fast_query \| df.sleep(30)`
`?>` `!>`	if /else - ブール条件での分岐	`cond ?> then_branch !> else_branch`
`@>`	ループ - 永続的に繰り返す (プレフィックス演算子)	`@> body`
`\|=>`	名前 - ステップの結果をキャプチャする	`'SELECT id FROM users LIMIT 1' \|=> 'user_id'`

便利な組み込み機能

Function	Purpose
`df.sleep(seconds)`	N 秒間一時停止します。再起動後も状態が維持されます。
`df.wait_for_schedule(cron)`	次に cron 式が一致するまで待ちます。
`df.wait_for_signal(name, timeout)`	外部 `df.signal()` が到着するまでブロックします。
`df.http(url, method, body, headers, timeout)`	一時的な障害時に再試行して、永続的なアクティビティとして HTTP 呼び出しを行います。
`df.if(cond, then, else)`	条件分岐。
`df.loop(body, cond)`	SQL 条件が正しい間は繰り返します。
`df.join(a, b)` / `df.race(a, b)`	並列実行と競合実行。
`df.join3(a, b, c)`	3 方向の並列実行の場合。
`df.start(body, label, database)`	永続関数を送信し、そのインスタンス ID を返します。
`df.cancel(id, reason)`	実行中のインスタンスを取り消します。
`df.status(id)` / `df.result(id)`	結果を検査します。
`df.explain(input)`	視覚エフェクト用の関数グラフをレンダリングします。

すべての pg_durable機能の詳細をご覧ください。

変数

|=> は、ステップの結果を名前でキャプチャします。後の手順では、 $nameとして参照します。

SELECT df.start(
    'SELECT 100 AS amount' |=> 'total'
    ~> 'SELECT $total * 2 AS doubled'
);

使用例

再試行を含むマルチステップ ETL

クリーンアップ、読み込み、インデックス、ログを行う毎日の ETL:

SELECT df.start(
    'DELETE FROM target WHERE loaded_at < now() - INTERVAL ''1 day'''
    ~> 'INSERT INTO target SELECT * FROM staging'
    ~> 'REINDEX TABLE target'
    ~> 'INSERT INTO etl_log (job, finished_at) VALUES (''nightly'', now())',
    'nightly-etl'
);

DELETEとINSERTの間でデータベースが再起動すると、ワーカーはINSERTから再開します。DELETEは再実行されません。

スケジュール済みジョブ (cron)

平日の午前 9 時にメンテナンスタスクを実行します。

SELECT df.start(
    @> (
        df.wait_for_schedule('0 9 * * 1-5')
        ~> 'CALL refresh_materialized_views()'
    ),
    'weekday-refresh'
);

このジョブを停止する場合は、 cancel 関数を実行できます。

SELECT df.cancel('a1b2c3d4', 'stop test cron job');

タイムアウトを含む承認ワークフロー

外部承認シグナルを最大 24 時間待ってから、コミットまたは拒否します。

SELECT df.start(
    'SELECT order_id, total FROM orders WHERE id = 1' |=> 'order'
    ~> df.wait_for_signal('approval', 86400) |=> 'sig'
    ~> df.if(
        'SELECT NOT ($sig::jsonb->>''timed_out'')::boolean
            AND ($sig::jsonb->''data''->>''approved'')::boolean',
        'UPDATE orders SET status = ''approved'' WHERE id = $order_id',
        'UPDATE orders SET status = ''rejected'' WHERE id = $order_id'
    ),
    'order-approval'
);

-- Later, approve from anywhere
SELECT df.signal('a1b2c3d4', 'approval',
                 '{"approved": true, "approver": "jane@contoso.com"}');

Durable HTTP 呼び出し

df.http() は、永続的なアクティビティとして外部呼び出しを行います。5xx 応答、ネットワークエラー、タイムアウトは自動的に再試行されます。

SELECT df.start(
    df.http('https://api.example.com/users/123', 'GET') |=> 'user'
    ~> 'INSERT INTO users_cache (data) VALUES (($user::jsonb->>''body'')::jsonb)',
    'fetch-user'
);

許可されている HTTP セキュリティの詳細については、 pg_durableを参照してください。

監視と操作

すべてが SQL からクエリ可能です。学習する別の UI やサービスはありません。

-- All instances
SELECT * FROM df.list_instances();

-- Filter by status
SELECT * FROM df.list_instances() WHERE status = 'Running';
SELECT * FROM df.list_instances() WHERE status = 'Failed';

-- Detail for one instance
SELECT * FROM df.instance_info('a1b2c3d4');

-- Execution history (useful for retried or looped functions)
SELECT * FROM df.instance_executions('a1b2c3d4', 20);

-- The function graph as it ran
SELECT * FROM df.instance_nodes('a1b2c3d4');

-- System-wide metrics
SELECT * FROM df.metrics();

ワーカーが稼働しているかを確認する方法:

SELECT epoch_id, last_seen_at, now() - last_seen_at AS time_since_last_heartbeat
FROM df._worker_epoch;

15 秒以下の time_since_last_heartbeat は、ワーカーが正常であることを意味します。それより大きい値、または行がまったく存在しない場合は、ワーカーが停止しているか、まだ初期化が完了していないことを意味します。

Visual Studio Codeでワークフローを監視する

Visual Studio Code 用の PostgreSQL 拡張機能には、Pipelines & Workflows ビューに Workflows タブがあり、エディターから pg_durable ワークフローインスタンスを確認し、実行状態を監視できます。

[ワークフロー] ウィンドウを開く

Visual Studio Codeで、PostgreSQL 拡張機能を開きます。
オブジェクトエクスプローラーで、データベースを右クリックします。
[パイプライン] と [ワークフロー] を選択します。
[ ワークフロー ] タブを選択します。

左側のウィンドウには PG Durable Runs が一覧表示され、中央のウィンドウには選択したワークフローインスタンスの詳細が表示されます。

ワークフローの実行を検査する

ワークフロー実行を選択したら、概要を確認して次の内容を検証します。

状態: completed、 running、または failed。
実行ID: インスタンスの一意の識別子。
開始時刻と期間: 実行の進行状況とパフォーマンスを追跡します。
詳細パネル: 追加の実行メタデータ。

使用可能なタブを使用して、詳細を確認します。

グラフ: ワークフロー構造とステップフローを示す、ステップバイステップの実行ビューを視覚的に表示します。
タイミング: パフォーマンス分析とボトルネックの識別のための期間に重点を置いたビュー。
結果: ワークフロー実行からの出力と結果指向の詳細。

AI パイプラインに関連するワークフローの場合、 パイプライン定義の表示 アクション (使用可能な場合) を使用すると、ワークフロー実行からパイプライン定義にリンクできます。これは、実行全体の動作の比較や回帰の調査に役立ちます。

ID と分離

Durable Functions は、ワーカーの権限ではなく、それらを送信したユーザーの権限で実行されます。 pg_durable は送信時に session_user と current_user の両方をキャプチャするため、 SET ROLE コンテキストで送信された関数は、その有効なロールで実行されます。

これは、以下のようなことを意味します。

ユーザーは、既にアクセス許可を持っているデータのみを表示および変更します。
非スーパーユーザーは、durable function を送信することで権限昇格を行うことはできません。
マルチテナントワークロードは、ロールと許可モデルが正しい限り分離された状態を維持します。

レプリカ、バックアップ、および PITR との対話

バックアップと PITR。 関数グラフ (df スキーマ) と実行状態 (duroxide スキーマ) は通常のテーブルに格納され、HorizonDB バックアップに含まれます。ポイントインタイムリストアでは、両方が復元されます。
レプリカの読み取り。 バックグラウンドワーカーはプライマリでのみ実行されます。読み取りレプリカでは、 df.* 監視ビューに対してクエリを実行できますが、関数は実行されません。
フェールオーバー。 フェールオーバー後、新しいプライマリ上のワーカーは、古いプライマリが中断したところから処理を引き継ぎます。実行中のインスタンスは、最後のチェックポイントから再開されます。

外部オーケストレーターと比較した場合

特徴	外部オーケストレーター	`pg_durable`
デプロイメント	独立したサービス、独立した ID、独立した状態ストア	1 つのデータベース
状態の持続性	オーケストレーターのストレージレイヤー	データと同様のバックアップ、HA、PITRを提供します
アイデンティティ	作業者はサービス ID で実行されます	関数は、送信ユーザーとして実行されます
障害モード	オーケストレーターとデータベース間のネットワーク	なし - 同じプロセス
最適な用途	多くのサービスに対応するシステム間オーケストレーション	処理の大半が Postgres 内またはその周辺で行われるワークロード

pg_durable は、システム間パイプラインの外部オーケストレーターを置き換えようとしていません。ほとんどの作業がデータベース作業 (埋め込み、変換、AI 呼び出し、スケジュールされたメンテナンス) であり、別のサービスを追加する方がメリットよりもコストが高い場合は、これが適切な選択です。

プレビュー期間中の制限事項

df.http() 5xx およびネットワークエラーの再試行。 4xx 応答がワークフローに返され、処理されます。自動的には再試行されません。
バックグラウンドワーカーは、インスタンスごとに 1 つのデータベースを提供します。複数データベースのファンアウトは、ワーカーのデータベースで実行されている関数からの df.start(..., database => 'other_db') によってサポートされます。
関数の定義と実行状態は、pg_durable期間中ののメジャーバージョン間では移植できません。アップグレードする前に、実行中のインスタンスをドレインまたはキャンセルします。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-06-13

Azure HorizonDB 用 pg_durable による耐久性機能 (プレビュー)

"永続的" とは

pg_durableを使用する場合

どのように機能するのか

pg_durableを有効にする

最初の永続的な関数

プログラム モデル

Operators

便利な組み込み機能

変数

使用例

再試行を含むマルチステップ ETL

スケジュール済みジョブ (cron)

タイムアウトを含む承認ワークフロー

Durable HTTP 呼び出し

監視と操作

Visual Studio Codeでワークフローを監視する

[ワークフロー] ウィンドウを開く

ワークフローの実行を検査する

ID と分離

レプリカ、バックアップ、および PITR との対話

外部オーケストレーターと比較した場合

プレビュー期間中の制限事項

関連するコンテンツ

フィードバック

その他のリソース

プログラムモデル