Pipeline error log#
各パイプラインにはエラーログテーブルがあります。レコードが任意のステップで失敗すると、ログは、ソースレコードID、失敗したステップ、操作タイプ、およびエラーメッセージをキャプチャします。バッチの残りの部分は処理を継続します。
その仕組み#
レコードに失敗すると、ソースID、失敗したステップ番号、操作タイプ、エラーメッセージとともに
{source_schema}.pipeline_{pipeline_name}_errorsログに記録されます。単一の不良レコードはパイプラインをブロックしません。バッチ内の他のレコードは、残りのすべてのステップを続行します。
エラーログテーブルはパイプラインで作成され、パイプラインが削除されるときに削除されます。 7.2.0より前に作成されたパイプラインは、アップグレード中にテーブルを取得します。
書き込みは失敗したステップのトランザクション内で発生するため、成功したステップが孤立したログ行を残すことはなく、障害がレコードを失うことはありません。
ログに記録された各エラーは、 Postgres
WARNINGも生成します。 promoteb.pipeline_error_warnings で切り替えます ;エラーは関係なくログに残ります。queryおよびsummaryファンクションは、エラーログテーブルが欠落しているパイプラインに対して空の結果を返します。
エラーログテーブルを手動で削除しないでください。 7.4.0の時点で、ソース
DELETEハンドラーとステップ後クリーンアップは、テーブルが見つからない場合にエラーを発生させます。エラーは、4つの
error_category値のいずれかを取得します helpb.ErrorBlocking を参照してください。
Category |
Scope |
Blocks pipeline? |
Description |
|---|---|---|---|
RecordTemporary |
Record |
No |
単一レコードの一時的な障害。再試行で成功する場合があります。 |
RecordPermanent |
Record |
No |
単一のレコードの永続的な障害たとえば、破損したデータ。 |
PipelineTemporary |
Pipeline |
Yes |
ステップ全体を停止した一時的な障害たとえば、サービスの停止。再試行で解決される場合があります。 |
PipelinePermanent |
Pipeline |
Yes |
ステップ全体を停止した永続的な障害たとえば、無効なモデル構成。 |
レコードレベルのエラーはパイプラインをブロックしません。他のレコードは処理を続けます。パイプラインレベルのエラーは、ステップ全体をブロックします。一時的なエラーは、再試行すると解決される場合があります。永続的なエラーは、不正なデータまたは構成ミスを示しています。具体的には
RecordTemporary
は現在モデルレート制限エラーの場合にのみ起動し、RecordPermanent
は不正な入力データたとえば、破損したPDF、不正なHTML、無効なUTF-8をカバーし、PipelineTemporary
は一時的な外部障害たとえば、オブジェクトストアの停止、モデルサービスタイムアウトをカバーします)、PipelinePermanent
は、一般的な構成の誤りたとえば、削除されたモデル、無効なステップ構成などです。
エラーのクエリー#
aidb.get_error_logs()
を使用して、パイプラインのエラーログを検査します。
SELECT * FROM aidb.get_error_logs(my_pipeline);
__OUTPUT__
id | source_id | part_ids | pipeline_step | step_operation | error_message | error_category | failed_at | retry_count | last_retry_at
- ---+-----------+----------+---------------+----------------+----------------------------------------+-------------------+----------------------------+-------------+---------------
3 | doc_99 | {0,2} | 2 | ChunkText | text segment exceeded maximum length | RecordPermanent | 2026-04-10 14:32:01.123+00 | 0 |
2 | doc_42 | {0} | 1 | ParsePdf | failed to extract text from page 0 | RecordPermanent | 2026-04-10 14:31:58.456+00 | 0 |
1 | doc_7 | | 1 | ParsePdf | corrupt PDF header | RecordPermanent | 2026-04-10 14:31:55.789+00 | 0 |
(3 rows)
ソースレコード、ステップ番号、またはエラーカテゴリでフィルタリングします。
SELECT * FROM aidb.get_error_logs(
my_pipeline,
p_source_id => doc_42
);
SELECT * FROM aidb.get_error_logs(
my_pipeline,
p_pipeline_step => 1::SMALLINT,
p_error_category => RecordPermanent
);
p_limit およびp_offset を使用したページ付け
SELECT * FROM aidb.get_error_logs(my_pipeline, p_limit => 20, p_offset => 40);
完全なパラメーターと戻り型の詳細については、 Operations reference を参照してください。
エラーの概要#
1つのパイプラインのステップ、オペレーション、およびカテゴリ別にグループ化されたエラー数を取得します。
SELECT * FROM aidb.get_error_log_summary(my_pipeline);
__OUTPUT__
pipeline_step | step_operation | error_category | error_count | latest_failed_at
- --------------+----------------+-------------------+-------------+----------------------------
1 | ParsePdf | RecordPermanent | 2 | 2026-04-10 14:31:58.456+00
2 | ChunkText | RecordPermanent | 1 | 2026-04-10 14:32:01.123+00
(2 rows)
または、すべてのパイプラインで同じロールアップを一度に取得します。
SELECT * FROM aidb.get_all_error_summaries();
__OUTPUT__
pipeline_name | pipeline_step | step_operation | error_category | error_count | latest_failed_at
- --------------+---------------+----------------+-------------------+-------------+----------------------------
my_pipeline | 1 | ParsePdf | RecordPermanent | 2 | 2026-04-10 14:31:58.456+00
my_pipeline | 2 | ChunkText | RecordPermanent | 1 | 2026-04-10 14:32:01.123+00
pdf_ingester | 1 | ParsePdf | PipelinePermanent | 1 | 2026-04-10 15:00:12.000+00
(3 rows)
完全なパラメーターと戻り型の詳細については、 Operations reference および promoteb.get_all_error_summaries を参照してください。
失敗したレコードの再キューイング#
5.0の新機能。
aidb.requeue_pipeline_errors()は、ターゲットのレコードレベルのエラー行を削除し、パイプラインのステートサーフェスでソースレコードをダーティとしてマークします。次の通常の実行で再処理されます。呼び出し自分自身は処理を行いません。
SELECT * FROM aidb.requeue_pipeline_errors(my_pipeline, ARRAY[1, 2, 99]);
__OUTPUT__
error_id | action
- ---------+------------------------
1 | requeued
2 | requeued
99 | not_found
(3 rows)
結果はerror_id
の昇順で並べられます。各行は、3つのアクションのいずれかをレポートします。
requeuedレコードレベルの行が削除され、ソースがダーティとしてマークされました。テーブルソースは状態イベントをエンキューします。ボリュームソースは、状態行にワンショットのretry_requested_atマーカーを設定します。skipped_pipeline_levelエントリーにはNULLsource_id以下を参照してください、そして適切な場所に残されます。not_foundIDがエラーログになかったか、パイプラインにエラーログテーブルがありません古いパイプラインまたはSemantic KBのようなEmptyソース。
バックグラウンドパイプラインは、次のワーカーティックに再キューイングされた行を自動的にドレインします。無効またはライブパイプラインを実際に再処理するには、フォローアップのaidb.run_pipeline('my_pipeline')
が必要です。
リキューイングは遅延マークダーティ操作であり、同期再試行ではありません。呼び出し自分自身は処理を行わないため、レコードごとの成功または失敗は、フォローアップの実行後にのみ表面化します。再処理後に何が起こったのかを確認するには、
aidb.get_error_logs を再照会します。
重複するIDのコンカレント・リキューは、行ごとのFOR UPDATE
ロックでシリアル化されます。重複しないIDセットは並列的に処理されます。
Requeueは、 aidb.run_pipeline
またはバックグラウンドワーカーをブロックしません。
パイプラインレベルのエラー#
パイプラインレベルのエラーNULL source_id
は、特定のレコードに関連付けられていないバッチ障害を記録します。レコードごとに再度キューイングすることはできないため、
aidb.requeue_pipeline_errors はaction='skipped_pipeline_level'
を返し、ログにエントリを残します。
SELECT * FROM aidb.requeue_pipeline_errors(my_pipeline, ARRAY[7]);
__OUTPUT__
error_id | action
- ---------+------------------------
7 | skipped_pipeline_level
(1 row)
根本的な問題たとえば、モデルの可用性、構成、削除されたモデルなどを修正し、パイプラインを再実行します。次のステップがエラーなしで実行されると、NULL行は自動的にクリアされます。再実行せずに終了するには、
aidb.clear_error_logs を使用します。
無効モードのボリュームに関する注意事項#
ボリュームソースを使用する無効モードのパイプラインでは、
aidb.requeue_pipeline_errors
はマークされたファイルにretry_requested_at
マーカーを設定しますが、ドレインをスケジュールできないため、
aidb.run_pipeline を指すNOTICE
を発行します。手動aidb.run_pipeline
は、マークされたファイルのみではなくボリューム全体を再読み取ります。これにより、大規模なボリュームのフルボリュームスキャンがコストがかかります。
大規模な無効モードボリュームの場合、パイプラインをバックグラウンドモードに切り替え、aidb.requeue_pipeline_errors
を呼び出し、ワーカーにマークされたファイルをドレインさせて、パイプラインを無効に戻します。
完全なパラメーターと戻り型の詳細については、 Operations reference を参照してください。
エラーのクリア#
IDで特定のエラーログエントリを削除します。
SELECT aidb.clear_error_logs(my_pipeline, ARRAY[1, 3]);
__OUTPUT__
clear_error_logs
- -----------------
2
(1 row)
削除された行数を返します。決済はIDベースです。 get_error_logs()
を使用してエントリを削除する前に確認してください。
完全なパラメーターと戻り型の詳細については、 Operations reference を参照してください。
自動クリーンアップ#
エラーログエントリは、3つの場合、aidb.clear_error_logs
またはaidb.requeue_pipeline_errors なしでクリアされます。
ソースレコードが削除されました。ライブ
DELETEトリガー、バックグラウンドワーカーDELETE処理、およびボリュームの古い状態のクリーンアップはすべて、宛先行とともに一致するエラーログエントリを削除します。ソースエンベロープは、後の実行で宛先に到達します。パイプラインエグゼキューターは、
(source_id, part_ids)によってスコープされるエントリをクリアするため、部分的粒度の成功によって兄弟エラーがワイプされません。そのステップが次に正常に実行された後、ステップのパイプラインレベルの行NULL
source_id。その実行中に同時に失敗したエグゼキューターによって書き込まれたエラーは保存されます。
パイプラインを削除すると、エラーログテーブル全体が削除されます。
パイプラインステータスの統合#
エラー数は promoteb.pipeline_metrics ビューにも表示されます。パイプラインステータスはエラーの重大度を反映します。
PartialErrors一部のレコードは失敗しました。残りは通常に処理されました。BlockingErrorsパイプラインレベルのエラーにより、ステップが完全に停止しました。
パイプラインの状態のモニタリングの詳細については、 モニタリングとオブザーバビリティ を参照してください。
エンドツーエンドの例#
3つのドキュメントソースを使用した短いウォークスルー。1つの行には破損したPDFがあり、1つは解析しますが、特大のチャンクを生成し、3番目は埋め込みステップでモデルレート制限に達します。
パイプラインを実行します。 3つの障害はいずれもバッチを中止しません。次に、結果のエラーを検査します。
SELECT aidb.run_pipeline(my_pipeline);
SELECT id, source_id, error_category
FROM aidb.get_error_logs(my_pipeline);
__OUTPUT__
id | source_id | error_category
- ---+-----------+-----------------
3 | doc_200 | RecordTemporary
2 | doc_42 | RecordPermanent
1 | doc_7 | RecordPermanent
(3 rows)
パイプラインはPartialErrors を報告し、数を表示します。
SELECT pipeline, "Status", "count(record errors)", "count(blocking errors)"
FROM aidb.pipeline_metrics WHERE pipeline = my_pipeline;
__OUTPUT__
pipeline | Status | count(record errors) | count(blocking errors)
- -------------+---------------+----------------------+------------------------
my_pipeline | PartialErrors | 3 | 0
(1 row)
上流のレート制限の問題を修正した後、レート制限された行を再度キューに入れます。これは、ソースがダーティであることを示します。自分自身は再処理しません。
SELECT * FROM aidb.requeue_pipeline_errors(my_pipeline, ARRAY[3]);
__OUTPUT__
error_id | action
- ---------+----------
3 | requeued
(1 row)
再度キューに入れられた行をドレインします。バックグラウンドパイプラインがこれを自動的に取得します。無効またはライブパイプラインには明示的な呼び出しが必要です。トリアージ後に2つの永続的な障害を無視します。
SELECT aidb.run_pipeline(my_pipeline);
SELECT aidb.clear_error_logs(my_pipeline, ARRAY[1, 2]);
__OUTPUT__
clear_error_logs
- -----------------
2
(1 row)
ステータスはUpToDate に戻ります。
SELECT pipeline, "Status" FROM aidb.pipeline_metrics WHERE pipeline = my_pipeline;
__OUTPUT__
pipeline | Status
- -------------+----------
my_pipeline | UpToDate
(1 row)