> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ocient.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Transactional Data Pipelines

export const AWS = "Amazon® Web Services℠ (AWS℠)";

export const Ocient = "Ocient®";

Transactional data pipelines in {Ocient} load data in batches and automatically roll back all uncommitted data when a failure occurs. The `CREATE TRANSACTIONAL PIPELINE` SQL statement supports a `START FOREGROUND` clause that creates a pipeline, immediately starts it, and blocks the SQL session until the pipeline reaches an inactive state — either `COMPLETED` or `FAILED`. This approach provides a single-command batch loading from within a SQL session.

> **Info:** The `START FOREGROUND` clause is only valid for `TRANSACTIONAL` pipelines. The Ocient System rejects this clause on `CONTINUOUS` and `BATCH` pipelines.

# CREATE TRANSACTIONAL PIPELINE

The `CREATE TRANSACTIONAL PIPELINE` SQL statement creates a new transactional data pipeline in the current database or replaces an existing one when you specify the `OR REPLACE` keywords. When you include the `START FOREGROUND` clause, the statement also starts the pipeline immediately and blocks until the pipeline reaches an inactive state.

```sql theme={null}
CREATE [OR REPLACE] TRANSACTIONAL PIPELINE pipeline_name
    SOURCE source_type source_options
    EXTRACT extract_options
    INTO table_name
    SELECT column_list
    [START FOREGROUND
        [ERROR (LIMIT n, FILE_ERROR {FAIL | SKIP_MISSING_FILE | TOLERATE})]
        [USING LOADERS loader_name [, ...]]
        [ON COMPLETION {NO_FLUSH | FLUSH_AND_WAIT | FLUSH_AND_RETURN}]
    ]
;
```

| Parameter         | Type       | Description                                                                                                                            |
| ----------------- | ---------- | -------------------------------------------------------------------------------------------------------------------------------------- |
| `pipeline_name`   | Identifier | The name of the pipeline to create. The name must be distinct from any existing pipeline names in the current database.                |
| `source_type`     | Identifier | The type of data source (for example, `S3` or `HDFS`). See [ CREATE PIPELINE ](/data-pipelines#CREATE-PIPELINE) for supported values.  |
| `source_options`  | Varies     | Source-specific connection and configuration settings. See [ CREATE PIPELINE ](/data-pipelines#CREATE-PIPELINE) for details.           |
| `extract_options` | Varies     | Format and parsing settings for the source data. See [ CREATE PIPELINE ](/data-pipelines#CREATE-PIPELINE) for details.                 |
| `table_name`      | Identifier | The target table for loading data.                                                                                                     |
| `column_list`     | String     | The list of columns for the load.                                                                                                      |
| `n`               | INTEGER    | The maximum number of errors the system permits before the pipeline transitions to the `FAILED` status.                                |
| `loader_name`     | Identifier | The name of a Loader Node to use for this pipeline run. Separate multiple Loader Node names with commas in the `USING LOADERS` clause. |

The `ERROR` clause controls how the system behaves with errors. The `LIMIT` keyword limits the number of errors. The `FILE_ERROR` keyword denotes the behavior when a source file is missing or unreadable. The `FAIL` value causes the pipeline to fail immediately. The `SKIP_MISSING_FILE` value silently skips missing files. And the `TOLERATE` value records the error and continues loading.

The `ON COMPLETION` clause controls segment generation after the pipeline completes. The default value is `NO_FLUSH`. The `FLUSH_AND_WAIT` value denotes that the system flushes segments synchronously before returning. The `FLUSH_AND_RETURN` value denotes that the system initiates a flush and returns immediately.

## START FOREGROUND Clause

The `START FOREGROUND` clause is not valid in combination with the `IF NOT EXISTS` clause. When you omit the `START FOREGROUND` clause, the `CREATE TRANSACTIONAL PIPELINE` statement creates the pipeline without starting it, and all existing behavior is unchanged.

The `ERROR`, `USING LOADERS`, and `ON COMPLETION` sub-clauses within the `START FOREGROUND` clause behave identically to the same clauses as in the `START PIPELINE` SQL statement.

## Blocking Behavior

When you execute the `CREATE TRANSACTIONAL PIPELINE ... START FOREGROUND` SQL statement, the SQL session blocks for the full duration of pipeline execution. The session does not return until the pipeline reaches an inactive state (`COMPLETED` or `FAILED`).

If you disconnect from the SQL session during pipeline execution, for example, by pressing Ctrl+C at the command line, the pipeline continues running independently. The disconnection does not stop or cancel the pipeline. After the pipeline reaches an inactive state, the Ocient System records the final status and all associated events in the `sys.pipelines` and `sys.pipeline_events` system catalog tables.

## Queuing Behavior

If all pipeline concurrency slots are occupied when you execute the statement, the system places the pipeline in a `QUEUED` state. The SQL session remains blocked while the pipeline is queued and continues to block after a concurrency slot opens, until the pipeline itself reaches an inactive state. You can monitor the pipeline status in the `sys.pipelines` system catalog table from a separate session during this time.

## Error Behavior

When the pipeline reaches the `COMPLETED` state, the statement returns successfully. When the pipeline reaches the `FAILED` or `STOPPED` state, the statement throws a `SQLException` error that contains the status message of the pipeline.

The `ERROR LIMIT` and `FILE_ERROR` keywords control failure thresholds in the same way as in the `START PIPELINE` statement. When the number of errors exceeds the `LIMIT` value, the pipeline transitions to the `FAILED` state.

## Transactional Rollback

On failure, the system automatically rolls back the transactional storage scope. The system does not commit any data from the failed pipeline run to the target table. This rollback behavior is identical to any other transactional pipeline failure and requires no special handling.

## Pipeline Lifecycle

Pipelines started with the `START FOREGROUND` clause follow the same lifecycle as any other pipeline. The system records all lifecycle events, including `CREATED`, `RUNNING`, `COMPLETED`, and `FAILED` states, in the `sys.pipeline_events` table. The `STARTED` event includes the start options used for the run, providing a complete audit trail. After the pipeline reaches an inactive state, the pipeline object persists in the `sys.pipelines` system catalog table with full metadata and metrics.

## Examples

### Create a Transactional Pipeline with an Error Limit

This statement creates a transactional pipeline that loads CSV files from {AWS} S3, tolerates up to 100 errors, and blocks until loading completes or the system exceeds the error limit.

```sql theme={null}
CREATE OR REPLACE TRANSACTIONAL PIPELINE my_pipeline
    SOURCE S3
        ENDPOINT 'https://s3.us-east-1.amazonaws.com'
        BUCKET 'ocient-docs'
        FILTER 'metabase_samples/csv/orders.csv'
        ENABLE_PATH_STYLE_ACCESS true
    EXTRACT FORMAT CSV
        FIELD_DELIMITER ','
        NUM_HEADER_LINES 1
    INTO my_schema.my_table
    SELECT
        $1 as id,
        $2 as user_id,
        $3 as product_id
    START FOREGROUND
        ERROR LIMIT 100, FILE_ERROR TOLERATE
        ON COMPLETION FLUSH_AND_WAIT;
```

### Create a Transactional Pipeline Using a Specific Loader Node

This statement creates a transactional pipeline using a specific Loader Node and returns as soon as the pipeline completes or fails, with no segment flush.

```sql theme={null}
CREATE OR REPLACE TRANSACTIONAL PIPELINE sales_pipeline
    SOURCE S3
        ENDPOINT 'https://s3.us-east-1.amazonaws.com'
        BUCKET 'ocient-docs'
        FILTER 'metabase_samples/csv/orders.csv'
        ENABLE_PATH_STYLE_ACCESS true
    EXTRACT FORMAT CSV
        FIELD_DELIMITER ','
        NUM_HEADER_LINES 1
    INTO sales.transactions
    SELECT
        $1 as id,
        $2 as user_id,
        $3 as product_id
    START FOREGROUND
        USING LOADERS loader_node_1
        ON COMPLETION NO_FLUSH;
```

# Related Links

[ START PIPELINE ](/data-pipelines#START-PIPELINE)
[ CREATE PIPELINE ](/data-pipelines#CREATE-PIPELINE)
[ DROP PIPELINE ](/data-pipelines#DROP-PIPELINE)
[ sys.pipelines ](/system-catalog#sys-pipelines)
[ sys.pipeline\_events ](/system-catalog#sys-pipeline_events)
[ Transactions ](/transactions)