Operations reference ==================== Reference for pipeline error log functions. For guide-style documentation, see :ref:`Pipeline error log ` . Error log functions ------------------- Each pipeline maintains an error log table at ``{source_schema}.pipeline_{pipeline_name}_errors`` . These functions provide a consistent interface for querying and managing that log. -------------- aidb.get_error_logs ^^^^^^^^^^^^^^^^^^^ Returns error log entries for a pipeline, with optional filters. Parameters ^^^^^^^^^^ .. csv-table:: :header: Parameter,Type,Default,Description :widths: 15,12,12,20 :align: left :class: longtable `p_pipeline_name`,TEXT,Required,Name of the pipeline. `p_source_id`,TEXT,NULL,Filter to a specific source record ID. `p_pipeline_step`,SMALLINT,NULL,Filter to a specific pipeline step number. `p_error_category`,:ref:`aidb.ErrorBlocking` ,NULL,"Filter by category: `RecordTemporary`, `RecordPermanent`, `PipelineTemporary`, or `PipelinePermanent`." `p_limit`,INTEGER,NULL,Maximum number of rows to return. `p_offset`,INTEGER,NULL,Number of rows to skip (for pagination). Returns ^^^^^^^ .. csv-table:: :header: Column,Type,Description :widths: 12,15,25 :align: left :class: longtable `id`,bigint,Error log entry ID `source_id`,text,Source record that failed `part_ids`,bigint[],Part identifiers (for chunked records) `pipeline_step`,smallint,Step number where the failure occurred `step_operation`,:ref:`aidb.PipelineStepOperation` ,"Operation name (for example, `ParsePdf`, `ChunkText`)" `error_message`,text,Error description `error_category`,:ref:`aidb.ErrorBlocking` ,Error category `failed_at`,timestamptz,Timestamp of the failure `retry_count`,integer,Number of retry attempts `last_retry_at`,timestamptz,Timestamp of the last retry `last_seen_at`,timestamptz,Timestamp of the most recent occurrence of this error. Updates on every repeat; `failed_at` stays fixed at the first occurrence. Examples ^^^^^^^^ .. code:: sql - - All errors for a pipeline SELECT * FROM aidb.get_error_logs(my_pipeline); - - Filter by source record SELECT * FROM aidb.get_error_logs(my_pipeline, p_source_id => doc_42); - - Filter by step and category SELECT * FROM aidb.get_error_logs( my_pipeline, p_pipeline_step => 1::SMALLINT, p_error_category => RecordPermanent ); - - Paginate SELECT * FROM aidb.get_error_logs(my_pipeline, p_limit => 20, p_offset => 40); -------------- aidb.get_error_log_summary ^^^^^^^^^^^^^^^^^^^^^^^^^^ Returns aggregated error counts for a single pipeline, grouped by step, operation, and category. .. _parameters-1: Parameters ^^^^^^^^^^ .. csv-table:: :header: Parameter,Type,Description :widths: 12,15,25 :align: left :class: longtable `p_pipeline_name`,TEXT,Name of the pipeline. .. _returns-1: Returns ^^^^^^^ .. csv-table:: :header: Column,Type,Description :widths: 12,15,25 :align: left :class: longtable `pipeline_step`,smallint,Step number `step_operation`,:ref:`aidb.PipelineStepOperation` ,Operation name `error_category`,:ref:`aidb.ErrorBlocking` ,Error category `error_count`,bigint,Number of errors in this group `latest_failed_at`,timestamptz,Most recent failure timestamp Example ^^^^^^^ .. code:: sql 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) -------------- aidb.get_all_error_summaries ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Returns the same rollup as ``aidb.get_error_log_summary()`` but across all pipelines at once. .. _returns-2: Returns ^^^^^^^ .. csv-table:: :header: Column,Type,Description :widths: 12,15,25 :align: left :class: longtable `pipeline_name`,text,Pipeline name `pipeline_step`,smallint,Step number `step_operation`,:ref:`aidb.PipelineStepOperation` ,Operation name `error_category`,:ref:`aidb.ErrorBlocking` ,Error category `error_count`,bigint,Number of errors in this group `latest_failed_at`,timestamptz,Most recent failure timestamp .. _example-1: Example ^^^^^^^ .. code:: sql 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 pdf_ingester | 1 | ParsePdf | PipelinePermanent | 1 | 2026-04-10 15:00:12.000+00 (2 rows) -------------- aidb.clear_error_logs ^^^^^^^^^^^^^^^^^^^^^ Deletes specific error log entries by ID. Review entries with ``aidb.get_error_logs()`` before clearing. .. _parameters-2: Parameters ^^^^^^^^^^ .. csv-table:: :header: Parameter,Type,Description :widths: 12,15,25 :align: left :class: longtable `p_pipeline_name`,TEXT,Name of the pipeline. `p_error_ids`,BIGINT[],Array of error log entry IDs to delete. .. _returns-3: Returns ^^^^^^^ ``BIGINT`` — the number of rows deleted. .. _example-2: Example ^^^^^^^ .. code:: sql SELECT aidb.clear_error_logs(my_pipeline, ARRAY[1, 3]); __OUTPUT__ clear_error_logs - ----------------- 2 (1 row) -------------- aidb.requeue_pipeline_errors ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Introduced in 7.5.0. Deletes the targeted record-level error log entries and marks their source records dirty on the pipeline’s state surface. The next normal run re-processes them; the call itself does no processing. See :ref:`Requeuing failed records ` for the workflow. .. _parameters-3: Parameters ^^^^^^^^^^ .. csv-table:: :header: Parameter,Type,Description :widths: 12,15,25 :align: left :class: longtable `p_pipeline_name`,TEXT,Name of the pipeline. `p_error_ids`,BIGINT[],Array of error log entry IDs to requeue. .. _returns-4: Returns ^^^^^^^ One row per input ID, ordered by ascending ``error_id`` . .. csv-table:: :header: Column,Type,Description :widths: 12,15,25 :align: left :class: longtable `error_id`,bigint,Error log entry ID from the input array. `action`,text,"`requeued`, `skipped_pipeline_level`, or `not_found` (see below)." - ``requeued`` : record-level row deleted and source marked dirty. Table sources enqueue a state event; volume sources set a one-shot ``retry_requested_at`` marker. - ``skipped_pipeline_level`` : entry has a NULL ``source_id`` and is left in place. Fix the underlying issue and re-run, or dismiss with ``aidb.clear_error_logs`` . - ``not_found`` : ID not present in the error log, or the pipeline has no error log table. Concurrent requeues on overlapping IDs serialize on per-row ``FOR UPDATE`` locks; non-overlapping ID sets proceed in parallel. Background pipelines drain requeued rows automatically on the next worker tick. Disabled or Live pipelines need a follow-up ``aidb.run_pipeline`` call. .. _example-3: Example ^^^^^^^ .. code:: sql 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 categories ---------------- .. csv-table:: :header: Category,Scope,Blocks pipeline?,Description :widths: 15,12,12,20 :align: left :class: longtable `RecordTemporary`,Single record,No,Transient failure for one record. May succeed on retry. `RecordPermanent`,Single record,No,"Permanent failure for one record (for example, corrupt data)." `PipelineTemporary`,Entire step,Yes,"Transient failure that stopped the step (for example, service outage). May resolve on retry." `PipelinePermanent`,Entire step,Yes,"Permanent failure that stopped the step (for example, invalid model config)."