Data Pipelines - Ocient Documentation

Data pipelines enable you to load data from your chosen source into the OcientAIQ™ Unified Data Platform. You can preview a data pipeline using the PREVIEW PIPELINE SQL statement to test the creation statement and any required data transformation. Then, you can create a data pipeline using the CREATE PIPELINE statement. To manage the load, use the START PIPELINE statement to start the load and the STOP PIPELINE statement to stop it. Or, you can use the QUIESCE PIPELINE statement to stop the execution of the data pipeline gracefully. To evolve the schema of a data pipeline, you can use the ALTER PIPELINE statement. You can also rename a data pipeline using the ALTER PIPELINE RENAME statement. To see the full definition of a created pipeline, use the EXPORT PIPELINE statement. When you finish with the load, you can use the DROP PIPELINE statement to remove the data pipeline. You can create user-defined data pipeline functions using the CREATE PIPELINE FUNCTION statement and remove the function using the DROP PIPELINE FUNCTION statement. To see the full definition of a created pipeline function, use the EXPORT PIPELINE FUNCTION statement. Also, you can administer privileges for data pipelines and data pipeline functions. For details, see Data Control Language (DCL) Statement Reference.

As an alternative to writing DDL statements, you can create data pipelines using the Pipeline Builder in the Management UI. The Pipeline Builder provides a guided workflow to configure sources, formats, and transformations without writing SQL.

CREATE PIPELINE

CREATE PIPELINE defines a data pipeline that you can execute with the START PIPELINE SQL statement. Specify the type of load, data source, and data format. To replace an existing pipeline, specify the OR REPLACE keywords. For a comparison with the ALTER PIPELINE SQL statement, see Update a Pipeline. You must have the ALTER privilege on the pipeline. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

CREATE [ OR REPLACE ] [ BATCH | CONTINUOUS | TRANSACTIONAL ] PIPELINE [ IF NOT EXISTS ] pipeline_name
    [ <advanced_pipeline_options> ]
    [ BAD_DATA_TARGET <kafka_bad_data_target> ]

    SOURCE ( <s3_source> | <filesystem_source> | <hdfs_source> | <kafka_source> )
        [ MONITOR ( <sqs_monitor> | <kafka_monitor> ) ]
    [ LOOKUP lookup_source <lookup_options> ]  [ ... ]

    EXTRACT
        ( <delimited_extract_options> | <asn1_extract_options> |
        <avro_extract_options> | <json_extract_options> |
        <parquet_extract_options> | <binary_extract_options> |
        <xml_extract_options> )
        [ <general_extract_options> ]

    [ INSERT ] INTO destination_table_name

    SELECT
        expression as col_alias_target,
        expression2 as col_alias_target2,
        ...
    [ WHERE filter_expression ]

    [ [ INSERT ] INTO destination_table_name_n
    SELECT
        expression_n as col_alias_target_n,
        expression2_n as col_alias_target2_n,
        ...
    [ WHERE filter_expression_n ] ] [ ... ]

/********************/
/* ADVANCED OPTIONS */
/********************/
advanced_pipeline_options ::=
    [ CORES processing_cores ]
    [ PARTITIONS file_partitions ]
    [ BATCH_SIZE number_of_rows ]
    [ RECORD_NUMBER_FORMAT record_number_format ]

/***************************/
/* BAD_DATA_TARGET OPTIONS */
/***************************/
kafka_bad_data_target ::=
    KAFKA
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic_name
        [ CONFIG config_option ]

/******************/
/* SOURCE OPTIONS */
/******************/
s3_source ::=
    S3
        <file_based_source_options>
        BUCKET bucket_name
        ( FILTER | FILTER_GLOB | FILTER_REGEX | OBJECT_KEY ) specifiers
        [ REGION region ]
        [ ENDPOINT endpoint ]
        [ ENABLE_PATH_STYLE_ACCESS enable_path_style_access ]
        [ ACCESS_KEY_ID access_key_credentials ]
        [ SECRET_ACCESS_KEY secret_key_credentials ]
        [ SESSION_TOKEN session_token ]
        [ ROLE_ARN role_arn ]
        [ ASSUME_ROLE_CONFIG assume_role_config]
        [ MAX_CONCURRENCY parallel_connections ]
        [ READ_TIMEOUT num_seconds ]
        [ REQUEST_DEPTH num_requests ]
        [ REQUEST_RETRIES num_retries ]
        [ HEADERS headers ]

filesystem_source ::=
    FILESYSTEM
        <file_based_source_options>
        ( FILTER | FILTER_GLOB | FILTER_REGEX ) specifiers

hdfs_source ::=
    HDFS
        <file_based_source_options>
        ( FILTER | FILTER_GLOB | FILTER_REGEX ) specifiers
        ENDPOINT endpoint
        [ CONFIG hdfs_config ]
        [ PRINCIPAL principal ]
        [ KEYTAB_FILE keytab_file ]
        [ TOKEN token | TOKEN_FILE token_file ]
        [ RENEW_INTERVAL renew_interval ]

kafka_source ::=
    KAFKA
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic_name
        [ GROUP_ID group_id ]
        [ WRITE_OFFSETS write_offsets ]
        [ CONFIG config_option ]
        [ AUTO_OFFSET_RESET ( 'latest' | 'earliest' ) ]

file_based_source_options ::=
    [ PREFIX prefix ]
    [ COMPRESSION_METHOD 'gzip' ]
    [ SORT_BY ( 'filename' | 'created' | 'modified' )
        [ SORT_DIRECTION ( ASC | DESC ) ] ]
    [ SORT_REWRITE sort_rewrite ]
    [ START_FILENAME start_filename ]
    [ END_FILENAME end_filename ]
    [ START_CREATED_TIMESTAMP start_created_timestamp ]
    [ END_CREATED_TIMESTAMP end_created_timestamp ]
    [ START_MODIFIED_TIMESTAMP start_modified_timestamp ]
    [ END_MODIFIED_TIMESTAMP end_modified_timestamp ]

/******************/
/* MONITOR OPTIONS */
/******************/
sqs_monitor ::=
    SQS
        <general_monitor_options>
        QUEUE_URL queue_url
        [ REGION region ]
        [ ENDPOINT endpoint ]
        [ ACCESS_KEY_ID access_key_id ]
        [ SECRET_ACCESS_KEY secret_access_key ]

kafka_monitor ::=
    KAFKA
        <general_monitor_options>
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic
        AUTO_OFFSET_RESET ( 'latest' | 'earliest' )
        [ GROUP_ID group_id ]
        [ MESSAGE {
            FILENAME file_selector
            TIMESTAMP timestamp_selector
            SIZE size_selector
        } ]
        [ CONSUMER {
            MAX_MESSAGES max_messages
            TIMEOUT consumer_timeout
        } ]

general_monitor_options ::=
    [ POLLING_INTERVAL polling_interval ]
    [ BATCH {
        MAX_FILES max_files
        TIMEOUT batch_timeout
        LOOKBACK lookback
     } ]

/*******************/
/* LOOKUP OPTIONS  */
/*******************/
lookup_options ::=
    CONNECTION_TYPE connection_type
    CONNECTION_STRING connection_string
    LOOKUP_SCHEMA lookup_schema
    LOOKUP_TABLE lookup_table
    [ REFRESH_PERIOD refresh_period ]
    [ CONFIG config_json ]

/*******************/
/* EXTRACT OPTIONS */
/*******************/
general_extract_options ::=
    [ CHARSET_NAME charset_name ]
    [ COLUMN_DEFAULT_IF_NULL column_default_if_null ]
    [ NULL_STRINGS null_strings ]
    [ TRIM_WHITESPACE trim_whitespace ]
    [ VALIDATE_CHARACTERS validate_characters ]
    [ REPLACE_INVALID_CHARACTERS replace_invalid_characters ]
    [ REPLACEMENT_CHARACTER replacement_character ]

delimited_extract_options ::=
    FORMAT ( 'delimited' | 'csv' )
        [ COMMENT_CHAR comment_char ]
        [ EMPTY_FIELD_AS_NULL empty_field_as_null ]
        [ ESCAPE_CHAR escape_char ]
        [ FIELD_DELIMITER field_delimiter ]
        [ FIELD_OPTIONALLY_ENCLOSED_BY enclosure_char ]
        [ HEADERS delimited_headers ]
        [ NUM_HEADER_LINES num_header_lines ]
        [ NUM_FOOTER_LINES num_footer_lines ]
        [ RECORD_DELIMITER record_delimiter ]
        [ SKIP_EMPTY_LINES skip_empty_lines ]
        [ OPEN_ARRAY open_array_char ]
        [ CLOSE_ARRAY close_array_char ]
        [ ARRAY_ELEMENT_DELIMITER array_delimiter ]
        [ OPEN_OBJECT open_object_char ]
        [ CLOSE_OBJECT close_object_char ]
        [ STRIP_ARRAY_ELEMENT_QUOTES strip_array_element_quotes ]
        [ STRIP_FIELD_QUOTES strip_field_quotes ]
        [ TRIM_ARRAY_ELEMENTS trim_array_elements]

asn1_extract_options ::=
    FORMAT 'asn.1'
        SCHEMA {
            URL url_asn1
            RECORD_TYPE record_type
        }

avro_extract_options ::=
    FORMAT 'avro'
        [ SCHEMA {
            [ INLINE inline_string ]
            [ INFER_FROM infer_from ]
            [ URL url_avro ]
            [ CONFIG config_avro ]
            [ SUBJECT subject ]
            [ SCHEMA_REGISTRY_ID_LOCATION ( 'header_or_value' | 'value' | 'none' ) ]
        } ]

binary_extract_options ::=
    FORMAT 'binary'
        RECORD_LENGTH length_in_bytes
        [ ENDIANNESS ( 'big' | 'little' ) ]
        [ AUTO_TRIM_PADDING auto_trim_padding ]
        [ PADDING_CHARACTER padding_character ]

json_extract_options ::=
    FORMAT 'json'

parquet_extract_options ::=
    FORMAT 'parquet'
        SCHEMA {
            INFER_FROM infer_from
        }

xml_extract_options ::=
    FORMAT 'xml'

Pipeline Identity and Naming

The name of a data pipeline is unique in an Ocient® System. Reference the pipeline in other SQL statements like START PIPELINE, STOP PIPELINE, and DROP PIPELINE using the name of the pipeline. The SQL statement throws an error if a pipeline with the same name already exists unless you specify the IF NOT EXISTS option. You can rename a pipeline with the ALTER PIPELINE RENAME SQL statement.

SQL identifiers, including pipeline names, table names, and column aliases in the SELECT clause, are case-insensitive. The Ocient System normalizes unquoted identifiers to lowercase. To preserve the original case of an identifier, enclose it in double quotes. For details, see Identifiers.

Update a Pipeline

You can update an existing pipeline in two ways: with the OR REPLACE clause in the CREATE PIPELINE SQL statement or with the ALTER PIPELINE SQL statement. Both statements require the pipeline to be inactive, which means that it has the STOPPED, QUIESCED, COMPLETED, or FAILED status. Both statements keep the name of the pipeline. Otherwise, the statements behave differently. The CREATE OR REPLACE PIPELINE statement replaces the pipeline by dropping the existing pipeline object and creating a new one. The new pipeline has these characteristics:

A replaced file-based pipeline cannot perform deduplication against data that the original pipeline loaded. A replaced Kafka streaming pipeline starts its loading from the last committed offset.
The pipeline starts with a fresh state, including target files, raft data, the durability horizon, metrics, and system catalog entries. This state does not include Kafka consumer group offsets, which the Kafka Broker stores.
The name of the pipeline remains the same. Because the system catalog and metrics entries of the dropped pipeline persist under that name, a query by name can return entries for both the original and the new pipeline identifier. Filter by pipeline identifier to distinguish them.

The ALTER PIPELINE statement updates most of the configuration and transformations of an existing pipeline. The statement performs an in-place modification that preserves the existing state of the pipeline. The statement provides these guarantees:

No change to the pipeline identifier.
No impact to the durability horizon, which means no data loss or duplication.
No impact to metrics, except for querying by name after a rename.

Where the Load Resumes

The statement that you choose does not by itself determine where the load resumes. The resume point depends on the type of the source.

For a file source, such as S3, Apache® Hadoop® Distributed File System (HDFS), or a file system, the ALTER PIPELINE statement keeps the existing list of files and the position of the load. The CREATE OR REPLACE PIPELINE statement builds a new list of files, so the load starts over and reloads files that the original pipeline already loaded.
For a Kafka source, the resume point depends on the consumer group identifier, not on the statement. If the group identifier does not change, the pipeline resumes from the last offset that it committed to the Kafka Broker. If the group identifier changes, the pipeline starts according to the auto.offset.reset policy, which depends on whether the new group previously committed offsets for the topic. For details, see Kafka Offsets and Consumer Group Identifiers.

If you do not specify the GROUP_ID option, the consumer group identifier of a Kafka pipeline derives from the name of the pipeline. Renaming the pipeline changes that identifier and the resume point of the load. For details, see ALTER PIPELINE RENAME.

Use the ALTER PIPELINE statement for most updates, such as when you rotate credentials, tune pipeline options, or evolve the schema of the pipeline. Use the CREATE OR REPLACE PIPELINE statement when you want a new pipeline object, such as when you want to reset the durability horizon and metrics or restart a file-based load from the beginning. For details, see the ALTER PIPELINE SQL statement.

Data Pipeline Modes

You can define pipelines in either batch BATCH, continuous CONTINUOUS, or transactional TRANSACTIONAL mode.

File sources (e.g., s3, filesystem) support BATCH and CONTINUOUS modes. File-based loads default to BATCH mode if you do not specify this keyword.
Kafka only supports CONTINUOUS mode. Loads with a Kafka source default to CONTINUOUS mode if you do not specify this keyword.
When you execute the START PIPELINE SQL statement using a data pipeline in the BATCH mode, the system creates a static list of files with the PENDING status in the sys.pipeline_files system catalog table. With the CONTINUOUS mode, the monitor appends new incoming files to the list of files in the sys.pipeline_files system catalog table.
With the CONTINUOUS mode, you can apply filters for the data pipeline. The system only uses files in the consumed messages if the filenames match the filters. For this mode, these options are invalid:
- PREFIX
- SORT_BY
- SORT_DIRECTION
- SORT_REWRITE
- START_CREATED_TIMESTAMP
- END_CREATED_TIMESTAMP
- START_MODIFIED_TIMESTAMP
- END_MODIFIED_TIMESTAMP
The TRANSACTIONAL mode enables the data pipeline to roll back the interim results of the load if an error occurs during load execution. The system makes all data visible only after the load completes successfully. This mode supports all options that the BATCH mode does. For details, see Transactional Data Pipelines.

This diagram shows the interaction between the source and storage for a batch and continuous pipeline. For a batch pipeline, the data pipeline automatically finds loading targets and loads them. Whereas for a continuous data pipeline, the monitor searches for new files and provides the loading target and the pipeline loads the targets.

Batch pipeline with a source and Kafka streaming pipeline with a source and monitor.

Required Options

You must define certain options in every pipeline. For these options, there is no default value. Optional options need to be present only if you use specific functions. A pipeline must contain the SOURCE, EXTRACT FORMAT, and INTO table_name ... SELECT statements.

These options depend on each other: IF NOT EXISTS and OR REPLACE SQL statements are mutually exclusive.

SELECT Statement and Data Transformation

Use the INTO table_name ... SELECT SQL statement in the CREATE PIPELINE SQL statement to specify how to transform the data. The SELECT statement includes a set of one or more expressions and a target column name in the form expression as column_name. The expression part in the statement contains a source field reference (e.g., $1 or $my_field.subfield) and, optionally, the transformation function you want to apply. If your data is nested in arrays, you can use special transformation functions, such as the EXPLODE_OUTER function, to expand the data into individual rows. For details about data transformation and supported transformation functions, see Transform Data in Data Pipelines. For details about data types and casting, see Data Types for Data Pipelines and Data Formats for Data Pipelines. You can also specify metadata values, such as filename, to load in the SELECT SQL statement. For details, see Load Metadata and File-Based Partitioned Data in Data Pipelines. This is an example of a transformation statement snippet.

SQL

...
FORMAT json
INTO public.orders
SELECT
    TIMESTAMP(BIGINT($created_timestamp)) as ordertime,
    metadata('filename') as source_filename,
    $order_number as ordernumber,
    $customer.first_name as fname,
    LEFT($customer.middle_initial,1) as minitial,
    $customer.last_name as lname,
    $postal_code as postal_code,
    $promo_code as promo_code,
    $order_total as ordertotal,
    DECIMAL($tax,8,2) as tax,
    CHAR[]($line_items[].product_name) as product_names,
    CHAR[]($line_items[].sku) as skus

Optionally, you can filter the load by using the WHERE clause in the form WHERE filter_expression. This expression should evaluate to a Boolean value or NULL and can include more than one filter expression. The system loads rows that contain data matching the filter criteria when the expression evaluates to true. The system does not load the rows when the expression evaluates to false or NULL. You can include any transform in the WHERE clause as in the SELECT clause. This is an example of a filter snippet that loads order data for customers whose last name starts with the letter a. Use the COALESCE function to return only non-NULL values.

SQL

...
FORMAT json
INTO public.orders
SELECT
    TIMESTAMP(BIGINT($created_timestamp)) as ordertime,
    metadata('filename') as source_filename,
    $order_number as ordernumber,
    $customer.first_name as fname,
    LEFT($customer.middle_initial,1) as minitial,
    $customer.last_name as lname,
    $postal_code as postal_code,
    $promo_code as promo_code,
    $order_total as ordertotal,
    DECIMAL($tax,8,2) as tax,
    CHAR[]($line_items[].product_name) as product_names,
    CHAR[]($line_items[].sku) as skus
WHERE
    STARTSWITH(COALESCE($customer.last_name, ''),'a')

NULL and Default Value Handling

The Ocient System has specific ways to handle NULL and default values in the data pipeline. Consider this information when you prepare data to load and write the CREATE TABLE and CREATE PIPELINE SQL statements. To load NULL values, insert the NULL value in the data. In this case, if you specify NOT NULL for the target column in the CREATE TABLE statement, the data pipeline fails to load. If you omit a column in the SELECT SQL statement, the data pipeline loads the default value for the column. If you do not specify the default value, the pipeline loads a NULL value. If you specify NOT NULL for the target column, the data pipeline also fails to load. You can use the DEFAULT keyword to load a default value. In this case, if the column does not have a default value, the pipeline fails to load. If the pipeline loads a NULL into a column with a specified default value, you can use COALESCE(<value>, DEFAULT) to insert the default value instead of the NULL value, where <value> is the NULL column. You can modify the load of one column at a time in this way. This table describes the data pipeline behavior for NULL or omitted column values.

Value in the Data Pipeline	Nullable Target Column	Default Value in Target Column	Resulting Data Pipeline Behavior
NULL	No	Value might or might not be set.	The pipeline uses the default value if the default exists and you specify the `COLUMN_DEFAULT_IF_NULL` option. Otherwise, the pipeline fails.
NULL	Yes	Value might or might not be set.	The pipeline uses the default value if the default exists and you specify the `COLUMN_DEFAULT_IF_NULL` option. Otherwise, the pipeline uses the NULL value.
Omitted column in the `SELECT` SQL statement	NULL value might or might not be set.	Yes	The pipeline uses the default value.
Omitted column in the `SELECT` SQL statement	No	No	The pipeline fails.
Omitted column in the `SELECT` SQL statement	Yes	No	The pipeline uses the NULL value.

Required Privileges

You must have the CREATE PIPELINE privilege on the underlying database and the VIEW privilege on each table in the pipeline definition to execute the CREATE PIPELINE SQL statement. The table must already exist. See the START PIPELINE SQL statement for the required privileges to execute a pipeline. For details, see Data Control Language (DCL) Statement Reference.

Examples

Load JSON Data from Kafka

This example loads JSON data from Kafka using the CREATE PIPELINE SQL statement. Use the bootstrap server 192.168.0.1:9092 using the Kafka topic orders. Load data into the public.orders table. Specify the data to load as these JSON selectors:

Identifier $id
User identifier $user_id
Product identifier $product_id
Subtotal amount $subtotal
Tax $tax
Total amount $total
Discount amount $discount
Created time $created_at
Quantity $quantity

SQL

CREATE PIPELINE orders_pipeline
    SOURCE KAFKA
        BOOTSTRAP_SERVERS '192.168.0.1:9092'
        TOPIC 'orders'
    EXTRACT
        FORMAT json
INTO public.orders
SELECT
    $id as id,
    $user_id as user_id,
    $product_id as product_id,
    $subtotal as subtotal,
    $tax as tax,
    $total as total,
    $discount as discount,
    $created_at as created_at,
    $quantity as quantity;

For a complete tutorial, see Data Pipeline Load of JSON Data from Kafka.

Load Delimited Data from S3

This example loads delimited data in CSV format from Amazon® Web Services℠ (AWS℠) S3. Use the https://s3.us-east-1.amazonaws.com endpoint with the ocient-docs bucket and path metabase_samples/csv/orders.csv. Denote path-style access using the ENABLE_PATH_STYLE_ACCESS option set to true. Specify one header line with the NUM_HEADER_LINES option. Load data into the public.orders table. Specify the data to load using the column numbers:

Idenfier
User identifier
Product identifier
Subtotal amount
Tax
Total amount
Discount amount
Created time
Quantity

SQL

CREATE PIPELINE orders_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
        NUM_HEADER_LINES 1
INTO public.orders
SELECT
    $1 as id,
    $2 as user_id,
    $3 as product_id,
    $4 as subtotal,
    $5 as tax,
    $6 as total,
    $7 as discount,
    $8 as created_at,
    $9 as quantity;

For a complete tutorial, see Data Pipeline Load of CSV Data from S3.

Continuous File Load of Delimited Data in CSV Files

Create a data pipeline that uses continuous file loading with delimited data in CSV files. Specify to use 32 partitions and 16 cores with the PARTITIONS and CORES options, respectively. The source is S3 with the http://endpoint.ocient.com endpoint and cs_data bucket. Specify the filter '*.csv' to find all files with a filename that matches a glob pattern without subdirectories, for example, data.csv. The system filters filenames with subdirectories such as data/data_sample.csv. For continuous file loading, specify the MONITOR option to use the Kafka monitor. Specify the test-broker:9092 bootstrap server, cfl_kafka_ten_adtech_flat_small Kafka topic, and reset the offset to the smallest offset using the earliest value of the AUTO_OFFSET_RESET option. Set the client group identifier to 84079bf1-bdc4-4b10-ba12-41ba6b17dffe. The format is csv with the record delimiter as the newline character \n. Load data in the public.ad_sessions table. The SELECT statement identifies the columns to load by number. There are 39 fields in the CSV data. For each column, transform each column using cast functions. For details, see Scalar Data Conversion Functions for each function. See Date and Time Functions for the TO_TIMESTAMP function.

SQL

CREATE CONTINUOUS PIPELINE test_continuous_pipeline
    PARTITIONS 32
    CORES 16
    SOURCE S3
        ENDPOINT 'http://endpoint.ocient.com/'
        BUCKET 'cs_data'
        FILTER '*.csv'
        MONITOR kafka
            BOOTSTRAP_SERVERS 'test-broker:9092'
            TOPIC 'cfl_kafka_ten_adtech_flat_small'
            AUTO_OFFSET_RESET 'earliest'
            GROUP_ID '84079bf1-bdc4-4b10-ba12-41ba6b17dffe'
    EXTRACT
        FORMAT csv
        RECORD_DELIMITER '\n'
    INTO public.ad_sessions
SELECT
    TO_TIMESTAMP(CHAR($1), 'yyyy-MM-dd HH:mm:ss.SSSSSS', 'java') AS event_date_time,
    CHAR($2)                    AS device_model,
    TINYINT($8)                 AS device_user_age,
    BOOLEAN($10)                AS device_ad_tracking_disabled,
    BINARY($11)                 AS device_mac,
    INT($14)                    AS ip_zip,
    FLOAT($19)                  AS ip_zip_latitude,
    DOUBLE($20)                 AS ip_zip_longitude,
    BIGINT($21)                 AS session_id,
    SMALLINT($32)               AS session_response_latency,
    DECIMAL($34, 10, 1)         AS session_transaction_revenue,
    CHAR($39)                   AS session_app_name
;

Continuous File Load of CSV Files with a Unique Message Format

Create a data pipeline that uses continuous file loading when the Kafka monitor consumes messages that do not use the default message format. By default, the monitor expects each message to match the structure of an s3:ObjectCreated:* event. When your messages use a different structure, specify the MESSAGE syntax to map a JSON selector to the filename, the last modification timestamp, and the file size of each incoming file. For details about the event structure, see S3 Notification Content Structure. In this example, the monitor consumes messages in this JSON format.

JSON

{"filename": "/tmp/pipelines/temp.csv", "timestamp": "2025-10-15T12:00:00Z", "size": 12}

Specify to use 32 partitions and 16 cores with the PARTITIONS and CORES options, respectively. The source is S3 with the http://endpoint.ocient.com/ endpoint and cs_data bucket. Specify the filter 'adtech_flat/small/1576515918-0000049496*' to load only the files in the adtech_flat/small directory with a filename that begins with 1576515918-0000049496. For continuous file loading, specify the MONITOR option to use the Kafka monitor with these options:

BOOTSTRAP_SERVERS — The three Kafka brokers test-broker1:9092, test-broker2:9092, and test-broker3:9092.
TOPIC — The cfl_kafka Kafka topic.
AUTO_OFFSET_RESET — The earliest value, which resets the offset to the smallest offset.
POLLING_INTERVAL — The interval of 10 seconds at which the monitor consumes the event topic.
BATCH — A maximum of 10 pending files, a wait of 15 seconds before the system starts a batch, and a lookback period of 10 hours for file deduplication.
CONSUMER — A maximum of 10 messages for each poll and an operation timeout of 1000 milliseconds.
MESSAGE — The $filename, $timestamp, and $size JSON selectors that match the fields in the incoming message.
GROUP_ID — The client group identifier ci__5aaa443e618a.

For details about these options, see Continuous File Loading Source Options. The format is csv. Load data in the public.ad_sessions table. The SELECT statement identifies the columns to load by number. For each column, transform each column using cast functions. The NULL_IF function loads a NULL value when the source field contains an empty string. For details, see Scalar Data Conversion Functions for each cast function. See Date and Time Functions for the TO_TIMESTAMP function and Other Functions and Expressions for the NULL_IF function.

SQL

CREATE CONTINUOUS PIPELINE kafka_message_pipeline
    PARTITIONS 32
    CORES 16
    SOURCE S3
        BUCKET 'cs_data'
        ENDPOINT 'http://endpoint.ocient.com/'
        FILTER 'adtech_flat/small/1576515918-0000049496*'
        MONITOR kafka
            BOOTSTRAP_SERVERS 'test-broker1:9092,test-broker2:9092,test-broker3:9092'
            TOPIC 'cfl_kafka'
            AUTO_OFFSET_RESET 'earliest'
            POLLING_INTERVAL 10 seconds
            BATCH { MAX_FILES 10 TIMEOUT 15 seconds LOOKBACK 10 hours }
            CONSUMER { MAX_MESSAGES 10 TIMEOUT 1000 }
            MESSAGE {
                FILENAME $filename
                TIMESTAMP $timestamp
                SIZE $size
            }
            GROUP_ID 'ci__5aaa443e618a'
    EXTRACT
        FORMAT csv
    INTO public.ad_sessions
SELECT
    TO_TIMESTAMP(CHAR($1), 'yyyy-MM-dd HH:mm:ss.SSSSSS', 'java') AS event_date_time,
    CHAR(NULL_IF($2, ''))       AS device_model,
    TINYINT($8)                 AS device_user_age,
    BOOLEAN($10)                AS device_ad_tracking_disabled,
    NULL_IF($11, '')            AS device_mac,
    INT($14)                    AS ip_zip,
    FLOAT($19)                  AS ip_zip_latitude,
    DOUBLE($20)                 AS ip_zip_longitude,
    BIGINT($21)                 AS session_id,
    SMALLINT($32)               AS session_response_latency,
    DECIMAL($34, 10, 1)         AS session_transaction_revenue,
    CHAR($39)                   AS session_app_name
;

Transactional Load of Delimited Data in CSV Files

Create a data pipeline that uses transactional file loading with delimited data in CSV files. The source is S3 with the http://endpoint.ocient.com endpoint and cs_data bucket. Specify the filter '*.csv' to find all files with a filename that matches a glob pattern without subdirectories, for example, data.csv. The system filters filenames with subdirectories such as data/data_sample.csv. The format is csv. Load data in the public.ad_sessions table. The SELECT statement identifies the columns to load by number. For each column, transform each column using cast functions. For details, see Scalar Data Conversion Functions for each function. See Date and Time Functions for the TO_TIMESTAMP function.

SQL

CREATE TRANSACTIONAL PIPELINE test_continuous_pipeline
    SOURCE S3
        ENDPOINT 'http://endpoint.ocient.com/'
        BUCKET 'cs_data'
        FILTER '*.csv'
    EXTRACT
        FORMAT csv
    INTO public.ad_sessions
SELECT
    TO_TIMESTAMP(CHAR($1), 'yyyy-MM-dd HH:mm:ss.SSSSSS', 'java') AS event_date_time,
    CHAR($2)                    AS device_model,
    TINYINT($8)                 AS device_user_age,
    BOOLEAN($10)                AS device_ad_tracking_disabled,
    BINARY($11)                 AS device_mac,
    INT($14)                    AS ip_zip,
    FLOAT($19)                  AS ip_zip_latitude,
    DOUBLE($20)                 AS ip_zip_longitude,
    BIGINT($21)                 AS session_id,
    SMALLINT($32)               AS session_response_latency,
    DECIMAL($34, 10, 1)         AS session_transaction_revenue,
    CHAR($39)                   AS session_app_name
;

SOURCE Options

File-Based Source Options

Options that apply to both the S3, FILESYSTEM, and HDFS sources.

Option Key	Default	Data Type	Option Configurability	Description
FILTER \| FILTER_GLOB \| FILTER_REGEX	None	STRING	Yes	The expression for filtering files in the directory or child directories to load. If you do not specify the `PREFIX` option, this pattern applies to the full path to the file, except for the Bucket for S3-compatible file sources. If you specify the `PREFIX` option, this pattern applies to the file path portion after the specified prefix value. Paths include a leading forward slash. For `FILESYSTEM` sources, this option is required. For `S3` sources, one of the `FILTER`, `FILTER_GLOB`, `FILTER_REGEX`, or `OBJECT_KEY` keys is required. Specify one of these options: * `FILTER_GLOB` or `FILTER` — Regular Unix® filename patterns. Supported options include: - `` indicates unlimited wildcard characters except for the path character, `/`. - `` indicates wildcard characters, including path separators. - `?` indicates a single wildcard character. `FILTER_REGEX` — Regular expression patterns to filter files. Some common patterns include: - `.` matches any character. - `` matches any 0 or more of the preceding character. - `+` matches a single character. - `[135]` matches any one character in the set. - `[1-5]` matches any one character in the range. - `(a\|b)` matches a or b. ℹ️ When you use the FILTER or FILTER_GLOB options with a continuous data pipeline, the system supports only basic globbing, and extended, range globbing is not supported. FILTER_GLOB Examples:* List all CSV files in subdirectories named `2024` that are in any subdirectory of the `trades` directory, which is in the root of the bucket. `FILTER_GLOB = '/trades//2024/.csv'` List all CSV files in the bucket in all subdirectories. `FILTER_GLOB = '*/.csv'` FILTER_REGEX Examples: List all `json.gz` files in the bucket that contain the name `orders` anywhere in the path. `FILTER_REGEX = '.orders.\.json\.gz'` List all files in the bucket in the root path `metrics` or `values` where the date string on the filename is 2000 to 2004. `FILTER_REGEX = '/(metrics\|values)/.200[0-4].'`
PREFIX	`NULL`	STRING or ARRAY OF STRINGS	No	Optional. Specify a prefix within which to apply the filter. When you specify a list of prefixes, the system applies the filter to each one, and the results are unioned together. Prefixes must be paths to directories, meaning they end with a forward slash. For `FILESYSTEM` sources, paths should be absolute, meaning they begin with a forward slash. For `S3` sources, use prefixes to reduce the search space when listing objects within the specified `BUCKET`. For maximum performance, use the `PREFIX` option whenever possible, especially when combined with the `FILTER_REGEX` option. ℹ️This option is not available for continuous data pipelines. Examples: List all CSV files in the `2024` subdirectory of `files`. `PREFIX '/files/2024/' FILTER '*/.csv'` List all JSON files of subdirectory `2024/09/` in the `orders` directory. `PREFIX '/data/orders/2024/09/' FILTER_REGEX './orders/.json'` List all JSON files in multiple subdirectories. `PREFIX ['/data/orders/2024/09/', '/data/orders/2024/10/']`
COMPRESSION_METHOD	`NULL`	STRING	No	Optional. The method the Ocient System uses to decompress file data. To load uncompressed data, omit the COMPRESSION_METHOD option. ℹ️The COMPRESSION_METHOD option is only applicable to file-based sources. Kafka streaming pipelines automatically decompress data based on the Kafka topic configuration. Value Range: `gzip`.
SORT_BY	`filename`	STRING	Yes	Optional. The sort criteria for sorting the file list before the load. Supported options are: * `filename` — Sort files lexicographically by the filename. * `created` — Sort files based on the file created timestamp. * `modified` — Sort files based on the file modified timestamp. Sorting files such that the Ocient System sorts the data within the files according to the TimeKey® column in the target table can lead to better query performance. ℹ️This option is not available for continuous data pipelines. Value Range: `filename`, `created`, or `modified`.
SORT_DIRECTION	`ASC`	STRING	Yes	Optional. The sort direction determines the sort order. If you specify this option, you must also specify the SORT_BY option. ℹ️This option is not available for continuous data pipelines. Value Range: `ASC` (ascending) or `DESC` (descending).
SORT_REWRITE	`NULL`	STRING	Yes	Optional. This option is a sort comparator, allowing files to be renamed during the sort operation using capture groups from the filter regular expression. If you specify this option, you must also specify the `FILTER_REGEX` option. If you do not specify this option, the system uses the sorting specified by the SORT_BY option. ℹ️This option is not available for continuous data pipelines. Example: `FILTER_REGEX '(\d+)_(\d+)_(\d+)-order.json'` `SORT_REWRITE '\3-\1-\2.json'` This specification renames the file from `5_6_1999-orders.json` to `1999_5_6.json` by switching the order of the date fields.
START_FILENAME	`NULL`	STRING	Yes	Optional. The filename string that is the lower bound to filter for files lexicographically for batch pipelines. (inclusive) Use the full path of the file, such as `'/dir/load_file.json'`. If the file is located in the top-most directory, start the path with a slash `/`, such as `'/load_file.json'`. You can use this option without the END_FILENAME option. If you specify the END_FILENAME option, the value of the START_FILENAME option must be smaller than the value for the END_FILENAME option lexicographically.
END_FILENAME	`NULL`	STRING	Yes	Optional. The filename string that is the upper bound to filter for files lexicographically for batch pipelines. (inclusive) Use the full path of the file, such as `'/dir/load_file.json'`. If the file is located in the top-most directory, start the path with a slash `/`, such as `'/load_file.json'`. You can use this option without the START_FILENAME option. If you specify the END_FILENAME option, the value of the START_FILENAME option must be smaller than the value for the END_FILENAME option lexicographically.
START_CREATED_TIMESTAMP	`NULL`	TIMESTAMP-formatted STRING	Yes	Optional. The ISO-8601-compliant date or date time that is used as the lower bound (inclusive) to filter files by created timestamp for batch pipelines. The time zone should match the file metadata. If you specify both, the value for the START_CREATED_TIMESTAMP option must be before the value for the END_CREATED_TIMESTAMP option. ℹ️This option is not available for continuous data pipelines.
END_CREATED_TIMESTAMP	`NULL`	TIMESTAMP-formatted STRING	Yes	Optional. The ISO-8601-compliant date or date time that is used as the lower bound (inclusive) to filter files by created timestamp for batch pipelines. The time zone should match the file metadata. If you specify both, the value for the START_CREATED_TIMESTAMP option must be before the value for the END_CREATED_TIMESTAMP option. ℹ️This option is not available for continuous data pipelines.
START_MODIFIED_TIMESTAMP	`NULL`	TIMESTAMP-formatted STRING	Yes	Optional. The ISO-8601-compliant date or date time that is used as the lower bound (inclusive) to filter files by modified timestamp for batch pipelines. The time zone should match the file metadata. If you specify both, the value for the START_MODIFIED_TIMESTAMP option must be before the value for the END_MODIFIED_TIMESTAMP option. ℹ️This option is not available for continuous data pipelines.
END_MODIFIED_TIMESTAMP	`NULL`	TIMESTAMP-formatted STRING	Yes	Optional. The ISO-8601-compliant date or date time that is used as the lower bound (inclusive) to filter files by modified timestamp for batch pipelines. The time zone should match the file metadata. If you specify both, the value for the START_MODIFIED_TIMESTAMP option must be before the value for the END_MODIFIED_TIMESTAMP option. ℹ️This option is not available for continuous data pipelines.

S3 Source Options

You can apply these options to data sources of the SOURCE S3 type, which include S3 and S3-compatible services.

Option Key	Default	Data Type	Option Configurability	Description
BUCKET	None	STRING	No	The name of the bucket in AWS S3.
OBJECT_KEY	`NULL`	STRING or ARRAY OF STRINGS	No	Optional. Specify an S3 object key(s) to load. You cannot use this option if you specify `PREFIX`, `FILTER`, `FILTER_GLOB`, or `FILTER_REGEX` keys are specified. If you do not specify one of the `FILTER`, `FILTER_GLOB`, or `FILTER_REGEX` keys, this option is required. Object keys do not include a leading forward slash. Specifying individual objects can be faster than filtering, because the Ocient System avoids listing all the files in a large directory. Object keys must not include any of the special characters: `?{}[]` ℹ️This option is not available for continuous data pipelines. Examples:* Load a single object from the designated bucket at the specified object key. `OBJECT_KEY 'order_data/jsonl/orders_20251101.jsonl'` Load a list of objects from the designated bucket at the specified object keys. `OBJECT_KEY ['order_data/jsonl/orders_20251101.jsonl', 'order_data/jsonl/orders_20251201.jsonl']`
ACCESS_KEY_ID	`''`	STRING	Yes	Optional. The access key identification for AWS credentials. If you specify this option, you must also specify the `SECRET_ACCESS_KEY` option. The Ocient System uses anonymous credentials when you specify an empty value for this option.
SECRET_ACCESS_KEY	`''`	STRING	Yes	Optional. The secret key for AWS credentials. If you specify this option, you must also specify the ACCESS_KEY_ID option. The Ocient System uses anonymous credentials when you specify an empty value for this option.
SESSION_TOKEN	`NULL`	STRING	Yes	Optional. Temporary credentials can be made with a combination with the existing access key identifier and secret key, and an additional session token. You must specify the ACCESS_KEY_ID and SECRET_ACCESS_KEY options when you use this option.
ROLE_ARN	`NULL`	STRING	Yes	Optional. The Amazon® Resource Name (ARN) of the role to retrieve temporary credentials. The STS client used to retrieve the temporary credentials specifies the region loaded from `DefaultAwsRegionProviderChain` and credentials loaded from `DefaultCredentialsProvider`. You cannot use this option with the ACCESS_KEY_ID and SECRET_ACCESS_KEY options.
ASSUME_ROLE_CONFIG	`{}`	STRING	Yes	Optional. Additional configuration used to configure the `StsAssumeRoleCredentialsProvider` or `AssumeRoleRequest` used to retrieve and refresh temporary credentials with the ROLE_ARN option. This option is a JSON-formatted string that you can only specify at the same time as the ROLE_ARN option. The supported configuration options are: `"asyncCredentialUpdateEnabled"`, `"prefetchTimeSeconds"`, `"staleTimeSeconds"` for `StsAssumeRoleCredentialsProvider` and `"durationSeconds"`, `"externalId"`, `"policy"`, `"roleSessionName"`, `"serialNumber"`, `"tokenCode"` for `AssumeRoleRequest`.
REGION	`'us-east-1'`	STRING	No	Optional. The region that the Ocient System uses for AWS access. If you specify the `ENDPOINT` option, the system ignores this option.
ENDPOINT	`NULL`	STRING	No	Optional. The endpoint URI for the S3-compatible service API. (e.g., `https://s3.us-east-2.amazonaws.com`) When unspecified, this option defaults to `https://s3.REGION.amazonaws.com`. If you provide this option, the Ocient System ignores the `REGION` option.
ENABLE_PATH_STYLE_ACCESS	`NULL`	BOOLEAN	No	Optional. Whether to use path-style access, where the path includes the bucket name in the URL. For example, `https://s3.us-east-1.amazonaws.com/bucket-name/key-name`. The `false` value denotes virtual-hosted style access, where the URL contains the bucket name as part of the domain name. For example, `https://bucket-name.s3.us-east-1.amazonaws.com/key-name`. When unspecified, this option defaults to `true` if the `ENDPOINT` option is specified and `false` otherwise.
HEADERS	`NULL`	STRING	Yes	Optional. The headers to send with every request. This option is a JSON-formatted string. Represent the chosen header names as keys with corresponding values as scalars or lists of scalars. The system converts scalars that are not strings to strings. During the load, the system maps each element in a list to the header name represented by the corresponding key. Examples: `HEADERS '{"x-amz-request-payer": "requester"}'` returns this header for each request: `x-amz-request-payer: requester`. `HEADERS '{"header-name": ["list", "of", "values"]}'` returns this header for each request: `header-name: list, of, values`.
MAX_CONCURRENCY	50	INTEGER	Yes	Optional. Determines the number of parallel connections the Ocient System uses to communicate with the AWS S3 service. ⚠️ This option does not require modification in most cases. Contact Ocient Support to modify these values. Value Range: 1 to 100,000.
READ_TIMEOUT	0 (unlimited timeout)	TIME INTERVAL	Yes	Optional. The time until a read operation times out. ⚠️ This option does not require modification in most cases. Contact Ocient Support to modify these values. Value Range: 0 milliseconds to 12 hours.
REQUEST_DEPTH	500	INTEGER	Yes	Optional. The upper boundary of requests that the Ocient System handles concurrently. ⚠️ This option does not require modification in most cases. Contact Ocient Support to modify these values. Value Range: 1 to 100,000.
REQUEST_RETRIES	10	INTEGER	Yes	Optional. Number of times the AWS SDK retries failing requests before the Ocient System throws an error. ⚠️ This option does not require modification in most cases. Contact Ocient Support to modify these values. Value Range: 1 to 1,000.

S3 Credentials Hierarchy The S3 source configuration supports this hierarchy to obtain S3 credentials:

Level 1: Data pipeline configuration (set by using the ACCESS_KEY_ID and SECRET_ACCESS_KEY options)
Level 2: AWS SDK Default Credential Provider Chain (set by using the instructions provided in the web page)
Level 3: Anonymous access (set by default)

If the Ocient System does not obtain the credentials at a level, the system tries a lower level.

You can choose the level to store the credentials, where Level 1 is the highest. Higher levels take precedence for credential storage.

FILESYSTEM Source Options

No options exist specific to the file system source (SOURCE FILESYSTEM) except for general file-based source options. Files must be located in these directories:

Always allowed:
- /tmp
- config streamloader.extractorEngineParameters.tempDir (default is /var/opt/ocient/tmp)
- config streamloader.extractorEngineParameters.configurationOption.filesystem.access.directories
  - The default is empty.
  - You can specify a list of directories.
Not allowed:
- /etc
- /bin
- /sbin
- /lib
- /lib64
- /usr
- /boot
- /proc
- /sys
- /run
- /root
- /var/lib
- /var/log
- /var/run
- /var/cache

When you load data using SOURCE FILESYSTEM, the files must be addressable from all of your Loader Nodes. The Ocient System uses the specified path in the pipeline PREFIX and FILTER options to select the files to load.A shared view of the files you want to load must be available to all Loader Nodes involved in a pipeline. For example, you can use a Network File System (NFS) mount available to all Loader Nodes at a common path on each node.

Example This example CREATE PIPELINE SQL statement snippet contains a FILESYSTEM source and filters to all CSV files in the /tmp/sample-data/ directory on each of the Loader Nodes.

SQL

CREATE PIPELINE ...
    SOURCE FILESYSTEM
        PREFIX '/tmp/sample-data/'
        FILTER '**/*.csv'
    EXTRACT
        FORMAT delimited
INTO public.orders
SELECT
    $1 as username,
    $2 as subtotal,
    ...

HDFS Source Options

You can specify these options for HDFS data sources (SOURCE HDFS). HDFS sources support the same file formats as FILESYSTEM and S3 sources, including delimited, csv, json, binary, asn.1, avro, parquet, and xml. For advanced options, see Data Pipeline Load of JSON Data from HDFS. The HDFS source also supports Kerberos authentication using a keytab file or a delegation token. For details, see Kerberos Authentication for HDFS Sources. For examples, see “Load Data from HDFS with Kerberos Keytab Authentication” and “Load Data from HDFS with Kerberos Delegation Token Authentication” below.

Option Key	Default	Data Type	Option Configurability	Description
ENDPOINT	None	STRING	No	Required. The hostname and port number of the namenode server. For example, `'my-hdfs-server:1234'`. The namenode server manages the file system and regulates access to data.
CONFIG	None	JSON-formatted STRING	Yes	Optional. The JSON string that contains HDFS configuration properties to use during the load. The properties are primarily for authentication. You can also use these properties to configure advanced properties such as connection timeouts, parallelism, and network chunk sizes. Example: `CONFIG '{"dfs.client.use.datanode.hostname": true, "dfs.client.socket-timeout": 300000}'`
KEYTAB_FILE	None	STRING	Yes	Optional. The file path to the keytab file that stores one or more Kerberos principals and their encrypted keys. The file must exist at the specified path on all Loader Nodes. Use this option for keytab file authentication. If you specify either the `PRINCIPAL` or `KEYTAB_FILE` option, you must specify both. You cannot specify this option with the `TOKEN` or `TOKEN_FILE` option.
PRINCIPAL	None	STRING	Yes	Optional. The unique user identity within a Kerberos realm, such as `'loaduser@EXAMPLE.COM'`. Use this option for keytab file authentication. If you specify either the `PRINCIPAL` or `KEYTAB_FILE` option, you must specify both. You cannot specify this option with the `TOKEN` or `TOKEN_FILE` option.
RENEW_INTERVAL	10 MINUTES	TIME INTERVAL	Yes	Optional. The frequency at which the loading process renews the delegation token. You can specify this option only when you specify the `TOKEN` or `TOKEN_FILE` option. Value Range: 5 seconds to 24 hours.
TOKEN	None	STRING	Yes	Optional. The delegation token as a URL-formatted string. Use this option for delegation token authentication. You can specify only one of the `TOKEN` and `TOKEN_FILE` options. You cannot specify this option with the `PRINCIPAL` or `KEYTAB_FILE` option.
TOKEN_FILE	None	STRING	Yes	Optional. The file path to the delegation token storage file. The file must exist at the specified path on all Loader Nodes. The system checks that the file exists when you create the data pipeline. Use this option for delegation token authentication. You can specify only one of the `TOKEN` and `TOKEN_FILE` options. You cannot specify this option with the `PRINCIPAL` or `KEYTAB_FILE` option.

Examples These examples show different ways to use Kerberos authentication for loading HDFS source data. Load Data from HDFS with Kerberos Keytab Authentication This example loads GZIP-compressed CSV files from an HDFS source that requires Kerberos authentication with a keytab file. The pipeline uses the hdfs-host:9000 endpoint and filters on the path /data/scores/2026/*.csv.gz. Configure Kerberos authentication by specifying these options:

The PRINCIPAL option with the Kerberos principal loaduser@EXAMPLE.COM.
The KEYTAB_FILE option with the path of a keytab file that exists on all Loader Nodes.
The CONFIG option with the dfs.namenode.kerberos.principal and dfs.data.transfer.protection configuration properties so that the loading process can identify the Kerberos principal of the namenode server.

Load the data into the public.student_scores table. The SELECT statement identifies the columns to load by number. For details about Kerberos authentication, see Kerberos Authentication for HDFS Sources.

SQL

CREATE PIPELINE student_scores_keytab_pipeline
    SOURCE HDFS
        ENDPOINT 'hdfs-host:9000'
        FILTER '/data/scores/2026/*.csv.gz'
        PRINCIPAL 'loaduser@EXAMPLE.COM'
        KEYTAB_FILE '/etc/security/keytabs/loaduser.keytab'
        CONFIG '{"dfs.data.transfer.protection": "authentication", "dfs.namenode.kerberos.principal": "hdfs/hdfs-host@EXAMPLE.COM"}'
        COMPRESSION_METHOD 'gzip'
    EXTRACT
        FORMAT csv
    INTO public.student_scores
SELECT
    $1 AS id,
    $2 AS name,
    $3 AS score,
    $4 AS active;

Load Data from HDFS with Kerberos Delegation Token Authentication This example loads the same GZIP-compressed CSV files as the previous example with delegation token authentication instead of a keytab file. Specify the TOKEN option with the delegation token string. The loading process renews the token every five minutes as specified by the RENEW_INTERVAL option. For details about Kerberos authentication, see Kerberos Authentication for HDFS Sources.

SQL

CREATE PIPELINE student_scores_token_pipeline
    SOURCE HDFS
        ENDPOINT 'hdfs-host:9000'
        FILTER '/data/scores/2026/*.csv.gz'
        TOKEN 'HAAHbG9hZHVzZXItdG9rZW4tc3RyaW5n'
        RENEW_INTERVAL 5 MINUTES
        COMPRESSION_METHOD 'gzip'
    EXTRACT
        FORMAT csv
    INTO public.student_scores
SELECT
    $1 AS id,
    $2 AS name,
    $3 AS score,
    $4 AS active;

KAFKA Source Options

You can apply these options to Kafka data sources (SOURCE KAFKA).

For compression, you do not need to specify a compression option in Kafka streaming pipelines, because the Ocient System handles the compression type automatically. Records produced to the Kafka broker with a compression.type setting or with the compression.type set on the topic automatically decompress when the loading process consumes the records. The loading process uses built-in headers in Kafka to determine the required decompression during extraction.

Option Key	Default	Data Type	Option Configurability	Description
BOOTSTRAP_SERVERS	None	STRING	No	A comma-delimited list of `IP:port` pairs that contain the IP addresses and the associated port numbers of the Kafka Brokers. You can also use a hostname instead of the IP address. Example: `BOOTSTRAP_SERVERS = '198.51.100.1:9092,198.51.100.2:9092'`
TOPIC	None	STRING	No	The name of the Kafka topic indicates where to consume records.
WRITE_OFFSETS	`true`	BOOLEAN	Yes	Optional. Indicates whether the Kafka consumer should write its durably-made record offsets to the Kafka Broker.
CONFIG	`'{` `"group.id": "<systemId>__<databaseName>__<pipelineName>"` `}'`	JSON-formatted STRING	Yes	Optional. The consumer configuration that the Kafka consumers should use. Frequently specified configurations include `group.id` and `security.protocol`. These configurations are fixed and cannot be overridden: `{` `"enable.auto.commit": false,` `"key.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer",` `"value.deserializer": "org.apache.kafka.common.serialization.ByteArrayDeserializer"` `}`
AUTO_OFFSET_RESET	`'latest'`	STRING	Yes	Optional. Determine which action to take for the Kafka configuration when there is no initial offset in the offset store or the specified offset is out of range. Supported values are: `'earliest'` — Automatically reset the offset value to the smallest value. `'latest'` — Automatically reset the offset value to the largest value.
GROUP_ID	`'<systemId>__<databaseName>__<pipelineName>'`	STRING	No	Optional. Client group identifier for the Kafka source configuration.

For the consumer configuration, to create a secure connection from the Kafka consumer to a Kafka Broker, set the "security.protocol" key along with any SSL or SASL keys.If a certificate file is required, you must add it to the truststore used by the JVM® on all Loader Nodes. The truststore path must be identical on all Loader Nodes. The Kafka configuration can reference this truststore path.

If you specify the ssl.certificate.location or ssl.ca.location consumer configuration, you must specify both of these configurations. Otherwise, the system throws an error. For example:

CONFIG '{"auto.offset.reset": "earliest","ssl.certificate.location":"/etc/blab/file1.txt","ssl.ca.location":"/etc/blab/file2.txt"}'

Continuous File Loading Source Options

Specify these options for data pipelines that use FILESYSTEM or S3 sources with the CONTINUOUS mode. For a Kafka source, do not use these options.

You are responsible for setting up, maintaining, and publishing messages to the external tooling (e.g., Kafka, Amazon® SQS). The data pipeline does not generate messages automatically.

General File Monitor Options

Option Key	Default	Data Type	Option Configurability	Description
MONITOR	None	STRING	No	The type of monitor. Valid values are: `sqs` or `kafka`.
POLLING_INTERVAL	10 SECONDS	TIME INTERVAL	Yes	Optional. Unit of time to consume the event topic. Value Range: 10 seconds to 2 minutes.
MAX_FILES	100	INTEGER	Yes	Optional. In the `BATCH` syntax, the maximum number of pending files to collect before starting a batch. Value Range: 1 to 2,000.
TIMEOUT	1 MINUTE	TIME INTERVAL	Yes	Optional. In the `BATCH` syntax, the time to wait before starting a batch. Value Range: 10 seconds to 30 minutes.
LOOKBACK	1 DAY	TIME INTERVAL	Yes	Optional. In the `BATCH` syntax, the unit of time to look back for file deduplication. Value Range: 0 milliseconds to 2 days.

SQS Monitor Options Use these options when you use MONITOR sqs for SQS.

Option Key	Default	Data Type	Option Configurability	Description
QUEUE_URL	None	STRING	Yes	The URL of the target queue. For example: `http://localhost:32769/000000000000/queue1`
REGION	`NULL`	STRING	Yes	Optional. The region for the SQS operation for signing requests. When unspecified, the region is inferred from `QUEUE_URL` by looking for `sqs.REGION` or `sqs-fips.REGION`. If the region cannot be inferred, it defaults to `us-east-1`.
ENDPOINT	`NULL`	STRING	Yes	Optional. The endpoint URL for the client. For example: `http://localhost:32769` When unspecified, defaults to `https://sqs[-fips].REGION.amazonaws.com` if `REGION` was inferred or specified. Otherwise, this option is required.
ACCESS_KEY_ID	`''`	STRING	Yes	Optional. Access key identifier for SQS authentication. If you specify this option, you must also specify the `SECRET_ACCESS_KEY` option. The Ocient System uses anonymous credentials when you specify an empty value for this option.
SECRET_ACCESS_KEY	`''`	STRING	Yes	Optional. The secret access key for SQS authentication.

Kafka Monitor Options Use these options when you use MONITOR kafka for Kafka.

Option Key	Default	Data Type	Option Configurability	Description
BOOTSTRAP_SERVERS	None	STRING	Yes	The bootstrap servers for the Kafka consumer configuration. The string contains a list of brokers as a comma-separated list of the broker hostnames or broker hostnames and port number combinations, in the format `hostname:port`.
TOPIC	None	STRING	Yes	The name of the Kafka topic indicates where to consume records.
AUTO_OFFSET_RESET	`'latest'`	STRING	Yes	Optional. Action to take for the Kafka configuration when there is no initial offset in the offset store or the chosen offset is out of range: `'earliest'` — Automatically reset the offset to the smallest offset. `'latest'` — Automatically reset the offset to the largest offset.
GROUP_ID	`'<systemId>__<databaseName>__<pipelineName>_monitor'`	STRING	Yes	Optional. Client group identifier for the Kafka configuration.
CONFIG	`'{ "enable.auto.commit": false,` `"group.id": "<systemId>__<databaseName>__<pipelineName>_monitor",` `"max.poll.interval.ms": "600000",` `"heartbeat.interval.ms": "6000",` `"session.timeout.ms": "600000" }'`	JSON-formatted STRING	Yes	Optional. The consumer configuration that the Kafka consumers should use. This option is a JSON-formatted string. Certain values within this configuration are fixed, whereas the Ocient System provides other values with a default value that you can modify.
MAX_MESSAGES	100	INTEGER	Yes	Optional. In the `CONSUMER` syntax, the maximum number of messages to poll each time. Value Range: 1 to 999.
TIMEOUT	1000 MILLISECONDS	TIME INTERVAL	Yes	Optional. In the `CONSUMER` syntax, the operation timeout that controls how long the consume request waits for the response. Value Range: 0 milliseconds to 1 hour.
FILENAME	`$"Records"[1].s3.object.key` corresponds to a `s3:ObjectCreated:Put` event.	SELECTOR	Yes	Optional. In the `MESSAGE` syntax, defines the JSON selector for the filename of an incoming message in a monitor. The default value works for `s3:ObjectCreated:*` events. If you override this default configuration, you can parse custom JSON messages. The format must still adhere to S3 standards. For details, see Amazon Simple Storage Service Event message structure. The message must be in the JSON format. If you specify this option, then you must specify the other two options in the `MESSAGE` syntax.
TIMESTAMP	`$"Records"[1]."eventTime"` corresponds to a `s3:ObjectCreated:Put` event.	SELECTOR	Yes	Optional. In the `MESSAGE` syntax, defines the JSON selector for the last modification timestamp of the file of an incoming message in a monitor. The default value works for `s3:ObjectCreated:*` events. If you override this default configuration, you can parse custom JSON messages. The format must still adhere to S3 standards. For details, see Amazon Simple Storage Service Event message structure. The message must be in the JSON format. If you specify this option, then you must specify the other two options in the `MESSAGE` syntax.
SIZE	`$"Records"[1].s3.object.size` corresponds to a `s3:ObjectCreated:Put` event.	SELECTOR	Yes	Optional. In the `MESSAGE` syntax, defines the JSON selector for the file size of an incoming message in a monitor. The default value works for `s3:ObjectCreated:*` events. If you override this default configuration, you can parse custom JSON messages. The format must still adhere to S3 standards. For details, see Amazon Simple Storage Service Event message structure. The message must be in the JSON format. If you specify this option, then you must specify the other two options in the `MESSAGE` syntax.

The same Kafka CONFIG option override considerations apply. For details, see KAFKA Source Options.

LOOKUP Options

You can optionally look up data in a table from an external database. You can include this table in the SELECT statement of the CREATE PIPELINE SQL statement and perform join operations on its columns. To use the LOOKUP keyword, specify the external source name lookup_source. You can look up data in multiple external databases. In this case, use the LOOKUP keyword with the source name and corresponding options for each database. You must provide the appropriate JDBC JAR file for the JDBC connection to external databases. For tables in the Ocient System, this JAR file is not needed.

The Ocient System executes the JDBC code as trusted code. Only provide trusted JAR files to the loading process. The streamloader.extractorEngineParameters.configurationOption.engine.external.jdbc.jarRootDirectory configuration parameter specifies the location of the JAR file. For details, see Configuration Settings for Data Pipelines.

Option Key	Default	Data Type	Option Configurability	Description
CONNECTION_TYPE	None	STRING	Yes	The type of connection. Only `'jdbc'` is supported.
CONNECTION_STRING	None	STRING	Yes	The string for connection to the external database. For example: `'jdbc:sqlite:/path/to/file'`
LOOKUP_SCHEMA	None	STRING	Yes	The schema of the table in the external database.
LOOKUP_TABLE	None	STRING	Yes	The name of the table in the external database.
REFRESH_PERIOD	30 SECONDS	TIME INTERVAL	Yes	Optional. The time between lookup cache refreshes. Value Range: 1 second to 12 hours.
CONFIG	`NULL`	JSON-formatted STRING	Yes	Optional. The JSON string that contains additional configuration options. For example: `'{"force_refresh_if_null": false}'` Supported JSON keys are: `force_refresh_if_null`, specified as a Boolean that defaults to `true`. When you set this value to `true`, the system triggers a lookup cache refresh for each value missing from the lookup table (to account for a concurrent load). This cache refresh can have a noticeable performance impact. As such, if you expect that some values cannot result in successful lookups, then set this key to the `false` value. `load_timeout_seconds`, specified as a long integer with no default value. This option is the timeout, in seconds, for a lookup cache refresh. `driver_class`, specified as a string with no default value. This option is the fully qualified JDBC driver class name. If you specify this option, the system throws an error when the system cannot find this driver or the driver cannot load into the process. If the JAR file for the connection is a JDBC 4 file, the system automatically loads the driver class. Otherwise, you must specify the fully qualified class name (from `Class::getName` in Java®) using this option, such as `'{"driver_class": "org.test.example.Driver"}'`.

EXTRACT Options

General Extract Options

You can specify these options on any of the allowed FORMAT types.

Option Key	Default	Data Type	Option Configurability	Description
FORMAT	None	STRING	No	Specifies the format of the files to load. Supported values are: `delimited` `csv` `json` `binary` `asn.1` `avro` `parquet` `xml`
CHARSET_NAME	Binary format: `ibm1047` All other formats: `utf-8`	STRING	No	Optional. Specifies the character set for decoding data from the source records into character data. Use this character set when you load data into VARCHAR columns or when you apply `CHAR` transformation functions. Defaults to `utf-8` for all formats except BINARY, which defaults to `ibm1047`. You can configure the default value for `BINARY` formatted data using a SQL statement such as: `ALTER SYSTEM ALTER CONFIG SET 'sql.pipelineParameters.extract.binary.defaultCharset' = 'ibm1047'`
COLUMN_DEFAULT_IF_NULL	`false`	BOOLEAN	Yes	Optional. Specifies whether pipelines should load the column default value when the result of a series of transforms is NULL. If you set this option to `false` (the default), the pipeline loads NULL values into columns. If you set this option to `true`, the pipeline loads the defined default value of the column when the result of the execution of the transformation on the column is NULL.
NULL_STRINGS	Delimited or CSV format: `['null', 'NULL']` JSON format: `[]`	ARRAY OF STRINGS	Yes	Optional. Specifies string values that should represent a NULL value when extracted from the source records. Use this in `csv`, `delimited`, and `json` formats to convert specific values to NULL instead of requiring individual transform function calls to `NULL_IF` with those values. This option applies to source data immediately after extraction. If the result of transformations is one of these strings, you must use `NULL_IF` to transform to NULL with the specified string values.
TRIM_WHITESPACE	`false`	BOOLEAN	Yes	Optional. Specifies whether to trim whitespace from the beginning and end of each string field during extraction. In this case, whitespace is a space, tab, carriage return, or linefeed character. If the trimmed field value matches one of the values specified by the `NULL_STRINGS` option, the system converts the value in the field to `NULL`. If you set this value to `true` and the `EMPTY_FIELD_AS_NULL` option to `true`, the system converts any field that contains only whitespace to `NULL`. This option does not trim arrays of strings. To trim elements in arrays, you can use the `TRIM` transform function or the `TRANSFORM` function with a Lambda expression that uses the `TRIM` function.
VALIDATE_CHARACTERS	`false`	BOOLEAN	Yes	Optional. Specifies whether to detect invalid characters based on the encoding type during extraction. If the system detects invalid characters, the system throws a file-level error for file loads or a record-level error for Kafka streaming pipelines. This option is not supported for ASN.1, Apache® Parquet™, and binary formats.
REPLACE_INVALID_CHARACTERS	`false`	BOOLEAN	Yes	Optional. Specifies whether to replace invalid characters based on the encoding type with the replacement character `U+FFFD`. This option is not supported for ASN.1, Parquet, and binary formats.
REPLACEMENT_CHARACTER	`�`	STRING	Yes	Optional. Specifies the character that indicates invalid data when you set the REPLACE_INVALID_CHARACTERS option to `true`. The default value is the replacement character `U+FFFD`. This option is not supported for ASN.1, Parquet, and binary formats.

For ASN.1, Apache® Avro™, and Parquet extract options, the SCHEMA option is a MAP type with a NULL default value. This option is optionally specified depending on the format. SCHEMA specifies the syntax of options for schema retrieval.

ASN.1 Extract Options

You can specify these options for ASN.1 data record extraction (FORMAT ASN.1). For details about ASN.1-formatted data, see Load ASN.1 Data.

Option Key	Default	Data Type	Option Configurability	Description
URL	None	STRING	Yes	In the `SCHEMA` syntax, specify the URL to the ASN.1 schema file in the `.asn` format.
RECORD_TYPE	None	STRING	Yes	In the `SCHEMA` syntax, specify the fully qualified name of the record type to parse. The name is the root-level ASN.1 type to extract from each DER-encoded record. (e.g., `Example.PersonnelRecord`)

Avro Extract Options

You can specify these options for Avro data record extraction. The SCHEMA options are all optional for object container files (OCF). For Kafka streaming pipelines, you must specify a schema configuration that includes either the URL option or the INLINE option (for manual specification), but not both. For file-based loads, the system can infer the schema from the file structure, so a schema configuration is not required. However, you can specify the schema configuration with:

Either the INLINE option or the INFER_FROM option, but not both of these options
Neither the INLINE nor the INFER_FROM options

For details about loading Avro-formatted data, see Load Avro Data.

Option Key	Default	Data Type	Option Configurability	Description
URL	None	STRING	Yes	The URL of the schema registry.
CONFIG	None	STRING	Yes	The configuration string that includes authentication for accessing the schema registry.
SUBJECT	`<topic_name>-value`, where `<topic_name>` is the name of the source Kafka topic	STRING	Yes	The custom subject for the schema registry.
SCHEMA_REGISTRY_ID_LOCATION	`header_or_value`	STRING	Yes	For Kafka streaming pipelines, the location to store the schema identifier for the schema registry. Supported values are: `'header_or_value'` — Specifies that the system searches for the schema Universally Unique IDentifier (UUID) in the `__value_schema_id` Kafka header. If the system finds the UUID, the system uses that identifier. Otherwise, the system derives the identifier from the starting bytes of the Kafka message value. `'value'` — Specifies that the system derives a schema identifier or UUID from the starting bytes of the Kafka message value. `'none'` — For inline schemas, specifies that there is no embedded schema identifier to skip. Value Range: `header_or_value`, `value`, or `none`.
INLINE	None	STRING	Yes	The JSON specification of the schema declaration as a string. Specify either a JSON string or object.
INFER_FROM	`sample_file`	STRING or ARRAY OF STRINGS	Yes	In the `SCHEMA` syntax, specify how to infer the Avro schema. Supported values are: `'sample_file'` — Use a sample file. list of files — Infer from a list of files with filenames that have valid full paths and do not include a leading forward slash. Examples: `SCHEMA { INFER_FROM` sample_file `}` `SCHEMA { INFER_FROM` [‘infer_from/examples/fileA.avro’, ‘infer_from/examples/fileB.avro’]` }

BINARY Extract Options

You can apply these options to binary data record extraction (FORMAT BINARY). For details about BINARY-formatted data, see Load Binary Data.

The general option CHARSET_NAME has a different default value for FORMAT BINARY.

Option Key	Default	Data Type	Option Configurability	Description
RECORD_LENGTH	None	INTEGER	Yes	Specifies the fixed size in bytes of each record in the source data. The Ocient System splits the binary data into binary chunks according to this length value and processes them individually.
ENDIANNESS	`big`	STRING	Yes	Optional. Specifies the endianness used to interpret multi-byte sequences in various transforms of binary data. Value Range: `big` or `little`.
AUTO_TRIM_PADDING	`true`	BOOL	Yes	Optional. Determines if padding characters should be trimmed after decoding the binary data into string data. If you set this option to `TRUE`, the Ocient System trims all instances of the `PADDING_CHARACTER` value from the end of a string after the system decodes the string from `BINARY` type.
PADDING_CHARACTER	(space)	STRING	Yes	Optional. The padding character from the string after the system decodes the string from `BINARY` type. You can change the default padding character for `BINARY` formatted data using a SQL statement such as: `ALTER SYSTEM ALTER CONFIG SET 'sql.pipelineParameters.extract.binaryFormat.defaultPaddingCharacter' = '*'`

The Ocient System trims the default padding character of a space from the end of the text data in binary data.

Delimited and CSV Extract Options

You can specify these options for delimited and CSV format data record extraction (FORMAT DELIMITED or FORMAT CSV data formats, which are aliases). For details about working with delimited and CSV data, see Load Delimited and CSV Data.

Option Key	Default	Data Type	Option Configurability	Description
FIELD_DELIMITER	`,`	STRING or ARRAY OF STRINGS	Yes	Optional. Specifies a character or list of possible characters for delimiting fields. The default value sets the field delimiter to only a comma. The Ocient System automatically interprets the values you specify as C-style escaped strings. This means you do not need to specify an escape string (`e'some value'`) is not required to specify control characters. This differs from the default string behavior in Ocient. For details, see String Literals and Escape Sequences. `FIELD_DELIMITER` and `FIELD_DELIMITERS` are aliases. Examples: Use a tab character: `FIELD_DELIMITER = '\t'` Use a pipe character: `FIELD_DELIMITER = '\|'` Use either a pipe or a comma character: `FIELD_DELIMITER = [',', '\|']` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
RECORD_DELIMITER	`['\r\n', '\n']`	STRING or ARRAY OF STRINGS	Yes	Optional. Specifies the string or an array of strings for delimiting records. The file is split into individual records using this character during processing. Common values include `'\r\n'` and `'\n'`. The value must be one or two bytes. The Ocient System automatically interprets the values you specify as C-style escaped strings. An escape string (`e'some value'`) is not required to specify control characters. This differs from the default strings behavior in Ocient. The system chooses the first specified delimiter and uses that delimiter for the rest of the data. Data with mixed delimiters is not supported. For details, see String Literals and Escape Sequences. `RECORD_DELIMITER` and `RECORD_DELIMITERS` are aliases. Examples: Use a linefeed character: `RECORD_DELIMITER = '\n'` Use a carriage return and linefeed character sequence: `RECORD_DELIMITER = '\r\n'`
NUM_HEADER_LINES	`0`	INTEGER	Yes	Optional. Specifies the number of header lines, typically 0 or 1. The Ocient System skips this number of lines and does not load them as data during file processing. Use this option when your data includes a row of header values.
NUM_FOOTER_LINES	`0`	INTEGER	Yes	Optional. Specifies the number of footer lines, typically 0 or 1. The Ocient System skips this number of lines starting from the end of the file and does not load them as data during file processing. Use this option when your data includes a row of footer values.
OPEN_ARRAY	`[`	STRING	Yes	Optional. Specifies the character that indicates the start of an array in a `csv` or `delimited` field. Use this option to parse array data types. Specify the `CLOSE_ARRAY` option also when using this option. Set this option to NULL or `''` to turn off the detection of these control characters. If you set this option to either of these characters, the system also turns off the detection of these characters for the `CLOSE_ARRAY` option. Example: Convert source data such as `val1,"[1,2,3]",val2` to an array when referenced as `$2[]`. `OPEN_ARRAY '['` `CLOSE_ARRAY ']'` `ARRAY_ELEMENT_DELIMITER ','` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
CLOSE_ARRAY	`]`	STRING	Yes	Optional. Specifies the character that indicates the end of an array in a `csv` or `delimited` field. Use this option to parse array data types. Specify the `OPEN_ARRAY` option also when using this option. Set this option to NULL or `''` to turn off the detection of these control characters. If you set this option to either of these characters, the system also turns off the detection of these characters for the `OPEN_ARRAY` option. Example: Convert source data such as `val1,"{1,2,3}",val2` to an array when referenced as `$2[]`. `CLOSE_ARRAY '}'` `OPEN_ARRAY '{'` `ARRAY_ELEMENT_DELIMITER ','` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
ARRAY_ELEMENT_DELIMITER	`,`	STRING	Yes	Optional. Specifies the character that separates values in an array within a `csv` or `delimited` field. Use this option to parse array data types. Set this option to NULL or `''` to turn off the detection of these control characters. Example: Convert source data such as `val1,"[1;2;3]",val2` to an array when referenced as `$2[]`. `ARRAY_ELEMENT_DELIMITER = ';'` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
OPEN_OBJECT	`{`	STRING	Yes	Optional. Specifies the character that indicates the start of a tuple in a field. Use this option to parse tuple data types. Specify the `CLOSE_OBJECT` option also when using this option. Set this option to NULL or `''` to turn off the detection of these control characters. If you set this option to either of these characters, the system also turns off the detection of these characters for the `CLOSE_OBJECT` option. Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
CLOSE_OBJECT	`}`	STRING	Yes	Optional. Specifies the character that indicates the end of a tuple in a field. Use this option to parse tuple data types. Specify the `OPEN_OBJECT` option also when using this option. Set this option to NULL or `''` to turn off the detection of these control characters. If you set this option to either of these characters, the system also turns off the detection of these characters for the `OPEN_OBJECT` option. Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
EMPTY_FIELD_AS_NULL	`true`	BOOLEAN	Yes	Optional. Specifies whether the Ocient System should extract an empty source field as NULL or a missing value. When this option is set to `true`, the Ocient System treats empty fields as NULL. Otherwise, the system treats fields as a missing value. For string-type fields, a missing value is equivalent to an empty string. Define an empty field as two consecutive delimiters (e.g., the second field is empty in `abc,,xyz`). If a field is explicitly an empty string as indicated by quote characters, the Ocient System treats the field as an empty string, not an empty field (e.g., the second field is an empty string in `abc,"",xyz`). The `CHAR($1)` transformation function handles both NULLs and empty strings and passes them through. ⚠️ Beware that the `NULL_IF($1, '')` transformation function directly loads a NULL for both NULLs and missing values. This transformation can override the behavior of `EMPTY_FIELD_AS_NULL`.
FIELD_OPTIONALLY_ENCLOSED_BY	`"`	STRING	Yes	Optional. Also known as the “quote character,” this option specifies the character for optionally enclosing fields. Fields enclosed by this character can include delimiters, the enclosure character, or the escape character. Set this option to NULL or `''` to turn off the detection of these control characters. Examples: Use a double quote character: `FIELD_OPTIONALLY_ENCLOSED_BY = '"'` Use a single quote character: `FIELD_OPTIONALLY_ENCLOSED_BY = ''''` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
ESCAPE_CHAR	`"`	STRING	Yes	Optional. Specifies the escape character within fields enclosed by the `FIELD_OPTIONALLY_ENCLOSED_BY` option. Use this option to escape the enclosure character or escape character. Set this option to NULL or `''` to turn off the detection of these control characters. Examples: Use a double quote as the escape character. `ESCAPE_CHAR = '"'` Use a single quote as the escape character. When you specify the escape character, you often have to use an escape sequence. This action follows standard SQL rules. `ESCAPE_CHAR = ''''` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
SKIP_EMPTY_LINES	`false`	BOOLEAN	Yes	Optional. Specifies whether or not to skip empty lines.
COMMENT_CHAR	`NULL`	STRING	Yes	Optional. Specifies the character used to comment out a record in the source file. The load skips records where the first character of a record is equal to this character. Set this option to NULL or `''` to turn off the detection of these control characters. Example: `COMMENT_CHAR '#'` Value Range: A single one-byte character, `NULL`, `''`, or an array of single one-byte characters.
HEADERS	`NULL`	ARRAY OF STRINGS	Yes	Optional. Specifies the header labels associated with each column in a delimited file. This array of values corresponds to the columns in order from left to right. Use these labels in the `CREATE PIPELINE SELECT` SQL statement to refer to column values. For example, if you specify `HEADERS ['col1', 'col2', col3']`, you can refer to the first column as `$"col1"` instead of `$1` in the `SELECT` statement.
STRIP_ARRAY_ELEMENT_QUOTES	`false`	BOOLEAN	Yes	Optional. When you set this option to `true`, the system removes quote characters from array elements, so that, for example, `["str"]` becomes `[str]`.
STRIP_FIELD_QUOTES	`true`	BOOLEAN	Yes	Optional. When you set this option to `true`, the system removes quote characters from the string, so that, for example, `"str"` becomes `str`.
TRIM_ARRAY_ELEMENTS	`false`	BOOLEAN	Yes	Optional. When you set this option to `true`, the system removes whitespace characters from the array elements, so that, for example, `" str"` becomes `"str"`.

JSON Extract Options

No options exist for JSON data record extraction (FORMAT JSON). For details about JSON-formatted data, see Load JSON Data.

PARQUET Extract Options

You can specify these options for Parquet data record extraction (FORMAT ``PARQUET). For details about Parquet-formatted data, see Load Parquet Data. You can extract Parquet files from S3, FILESYSTEM, and HDFS sources.

When you use the FORMAT PARQUET option with an AWS S3 Source, the ENDPOINT option is required.

Option Key	Default	Data Type	Option Configurability	Description
SCHEMA INFER_FROM	`sample_file`	STRING or ARRAY OF STRINGS	Yes	In the `SCHEMA` syntax, specify how to infer the Parquet schema. The supported values are: `sample_file` — Infer from a random file. list of files — Infer from a list of files with filenames that have valid full paths and do not include a leading forward slash. Examples: `SCHEMA { INFER_FROM` sample_file `} <br/>`SCHEMA

XML Extract Options

No options exist for the XML format extraction (FORMAT XML). For details about XML-formatted data, see Load XML Data.

Bad Data Targets

Bad data represents records that the Ocient System could not load due to errors in the transformations or invalid data in the source records. You can provide options for a bad data target that the Ocient System uses during pipeline execution to capture the records that are not loaded. The original bytes that the pipeline tried to load are captured in the bad data target along with the metadata about the error, such as the error message or source. Kafka is the only supported bad data target.

Kafka Bad Data Target

When you use Kafka as a bad data target, the Ocient System produces the original bytes of the source record into the Kafka topic of your choice. The Ocient System includes the metadata about the record in the header of the record as it is sent to Kafka. You can configure the Kafka topic on your Kafka Brokers using the retention and partition settings of your choice. You can use a Kafka bad data target with any data source (e.g., S3, local file, and so on).

If the Kafka Broker is unreachable when the Ocient System attempts to produce a bad data record to the bad data target, the system logs an error on the Loader Node, and the pipeline continues.

Example This example CREATE PIPELINE SQL statement snippet contains a bad data target definition using the BAD_DATA_TARGET option.

SQL

CREATE PIPELINE ...
    BAD_DATA_TARGET
        KAFKA
        TOPIC 'orders_errors'
        BOOTSTRAP_SERVERS '111.11.111.1:9092,111.11.111.2:9092'
        CONFIG '{"compression.type": "gzip"}'
    SOURCE
        ...
    EXTRACT
        ...
INTO public.orders
SELECT
    $order.billing.name as username,
    $order.subtotal as subtotal,
    ...

Kafka Bad Data Target Options

Option Key	Default	Data Type	Option Configurability	Description
BOOTSTRAP_SERVERS	None	STRING	Yes	A comma-delimited list of `IP:port` pairs that contain the IP addresses of the Kafka Brokers and the associated port number. Example: `BOOTSTRAP_SERVERS = '111.11.111.1:9092,111.11.111.2:9092'`
TOPIC	None	STRING	Yes	The name of the Kafka topic where the Ocient System should produce bad data records.
CONFIG	`'{` `"compression.type": "none"` `}'`	JSON-formatted STRING	Yes	Optional. The producer configuration that the Kafka producer should use.

Advanced Pipeline Tuning Options

You can use pipeline tuning options to control the parallelism or batching dynamics of your pipelines. This tuning can throttle the resources used on a pipeline or increase parallel processing across Loader Nodes. These options are advanced settings that might require a detailed understanding of the underlying mechanics of the loading infrastructure in the Ocient System to employ. Due to the inherent nature of each source type, the behavior of these options can differ between file-based and Kafka-based loads. All these options are optional. | Option Key | Default | Data Type | Option Configurability | Description | | --- | --- | --- | --- | --- | --- | | CORES | The maximum number of CPU cores available on each Loader Node. | INTEGER | Yes | Maximum number of processing threads that the Ocient System uses during execution on each Loader Node. The system creates this number of threads on each Loader Node.
The Ocient System automatically determines the default value by finding the number of cores of a Loader Node. You can use this option for performance tuning.
The calculation for maximum parallelism of a pipeline is: number_of_loaders * CORES.
About Kafka Partitions and Parallelism
For Kafka streaming pipelines, this option determines the number of Kafka Consumers created on each Loader Node.
For Kafka Pipelines, the recommendation is that number_of_loaders * CORES equals the number of Kafka topic partitions.
If this number exceeds the number of Kafka topic partitions, the work might spread unevenly across Loader Nodes.
If this number is less than the number of Kafka topic partitions, some Kafka Consumers might receive uneven amounts of work. In this case, use a value for number_of_loaders * CORES that is an even divisor of the number of Kafka topic partitions to avoid a skew in the rates of processing across partitions.

Value Range: 1 to 1,024. | | PARTITIONS | Equal to the value of CORES. | INTEGER | Yes | Specifies the number of partitions over which to split the file list. Not applicable to Kafka streaming pipelines.
The Ocient System automatically sets a default value based on the configured value for the CORES option. You can use this option for performance tuning.
The number of partitions determines how many buckets of work the Ocient System generates for each batch of files processed on a Loader Node. The pipeline processes this number of partitions in parallel using the specified number of cores.
If you specify fewer partitions than cores, some cores are not fully utilized, and resources are wasted. If you specify more partitions than cores, the Ocient System divides partitions in a round-robin fashion over the available cores.

Value Range: 1 to 2,048. | | BATCH_SIZE | A dynamic value, determined by the Ocient System for each pipeline to maximize performance. | INTEGER | Yes | Number of rows in the batch to load at one time.
The Ocient System automatically calculates a dynamic value depending on the table columns and the utilization of internal buffers to transfer records to the database backend. You can use this option to turn off the dynamic adjustments for performance tuning.
⚠️ Only change this setting in rare cases where loading performance is slower than expected, and you have a large record size. If this setting is improperly set, pipelines might fail with out-of-memory exceptions.
You can configure the default value (for the batch payload target) using a SQL statement such as:

ALTER SYSTEM ALTER CONFIG SET 'streamloader.extractorEngineParameters.configurationOption.osc.batch.payload.target' '65536'

Value Range: 1 to 1,073,741,824. | | RECORD_NUMBER_FORMAT | For file loads that do not use the EXPLODE_OUTER function, the default is `[19, 45, 0]`.
For Kafka streaming pipelines that do not use the `EXPLODE_OUTER` function, the default is `[0, 64, 0]`.
For loads that use the `EXPLODE_OUTER` function, the default is the specified load-type-specific default value with 13 subtracted from the record index bits. The system adds these bits to the bits for rows within a record.
For example, the default for file loads that use this function is `[19, 32, 13]`. | ARRAY | Yes | The 64-bit record number for each record of the load. This number uniquely identifies a row within its partition.
The format is an array with three values in the format`[, , ]`
The file index bits “ value is the number of bits used to represent the file index within a partition.
The record index bits “ value is the number of bits used to represent the record index within a file.
The rows per record index bits “ is the number of bits used to represent the row within a record. The system uses this value with the `EXPLODE_OUTER` function.
These three values must sum to 64.
**Example**
`RECORD_NUMBER_FORMAT= [10, 54, 0]`
Set the number of file index bits to `10` and the number of record index bits to `54`, allowing up to 2^10 files and 2^54 records per file. The system does not support the `EXPLODE_OUTER` function in this configuration because the rows per record index bits are `0`. | |

DROP PIPELINE

DROP PIPELINE removes an existing pipeline in the current database. You cannot remove a pipeline that is in the RUNNING or QUIESCING status. To remove a pipeline that is running, first execute the STOP PIPELINE or QUIESCE PIPELINE SQL statement. You must have the DROP privilege on the pipeline to execute this SQL statement. For details, see Data Control Language (DCL) Statement Reference.

When you drop a pipeline, the Ocient System also removes the associated system catalog information, such as pipeline errors, events, files, partitions, and metrics.

Syntax

SQL

DROP PIPELINE [ IF EXISTS ] pipeline_name [, ...]

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified pipeline to remove. You can drop multiple pipelines by specifying additional pipeline names, separated by commas.

Examples Remove Existing Data Pipeline Remove an existing pipeline named ad_data_pipeline.

SQL

DROP PIPELINE ad_data_pipeline;

Remove Existing Data Pipeline by Checking for Existence Remove an existing pipeline named ad_data_pipeline or return a warning if the Ocient System does not find the pipeline in the database.

SQL

DROP PIPELINE IF EXISTS ad_data_pipeline;

PREVIEW PIPELINE

PREVIEW PIPELINE enables you to view the results of loading data for a specific CREATE PIPELINE SQL statement without creating a whole data pipeline and without storing those results in the target table. Using this SQL statement, you can iterate quickly and modify the syntax as needed to achieve your expected results. After you confirm your expected results, you can use the same syntax in the body of the CREATE PIPELINE statement with the appropriate source. A table must exist in the database to serve as the target of your PREVIEW PIPELINE statement. This table ensures the pipeline matches the column types of the target table. However, the execution of this statement does not load data into the target table.

Preview Sources

The SOURCE INLINE source type is available only for the PREVIEW PIPELINE SQL statement. You cannot create a data pipeline with inline source data. The inline source data limit is 1,000 rows. Other source types defined in the CREATE PIPELINE statement, S3, KAFKA, and FILESYSTEM, are compatible with the PREVIEW PIPELINE statement. The extract options vary by the source type to mirror the CREATE PIPELINE statement. The database returns 10 records by default.

Preview Error Handling

Pipeline-level errors cause the PREVIEW PIPELINE SQL statement to fail. The Ocient System returns an error and no result set. However, the Ocient System accumulates record-level errors that occur during the execution of this statement in a single warning that the system returns along with the result set. Each line of the warning describes a record-level error in human-readable form or as a JSON blob, depending on the value of the SHOW_ERRORS_AS_JSON option. Rows or columns that encounter record-level errors have NULL values in the result set.

Preview Limitations

Limitations of this SQL statement are:

Before executing a PREVIEW PIPELINE SQL statement, you must create a table for the Ocient System to have context for the preview.
The maximum number of rows a PREVIEW PIPELINE SQL statement can return is 1,000 rows.
The COLUMN_DEFAULT_IF_NULL option from the CREATE PIPELINE SQL statement has no effect on the PREVIEW PIPELINE SQL statement.
The PREVIEW PIPELINE SQL statement does not honor the assignment of a service class based on text matching.
These source options are not supported:
- START_FILENAME
- END_FILENAME
When you execute two duplicate PREVIEW PIPELINE statements for a specific Kafka topic, the two statements share a consumer group. If the topic is small, one or both of the result sets might only be a partial result.
For multiple tables, you can preview only one table at a time. You must specify the name of the table you want to preview using the FOR keyword.
Previewing a continuous data pipeline is not supported.

Syntax

SQL

PREVIEW PIPELINE pipeline_name
    [ MODE mode ]
    [ SHOW_ERRORS_AS_JSON show_errors_as_json ]
    SOURCE [ INLINE ] (inline_string | <s3_source> | <filesystem_source> | <kafka_source>)
        [ LIMIT limit ]
    EXTRACT
        FORMAT csv
        RECORD_DELIMITER record_delimiter
        FIELD_DELIMITERS ['delim1', 'delim2', ...]
        [ INTERMEDIATE_VALUES intermediate_values ]
[ INSERT ] INTO created_tablename
SELECT
    preview_column_formula AS preview_column_name, ...
[ WHERE filter_expression ]
[ [ INSERT ] INTO created_tablename_n
SELECT
    preview_column_formula AS preview_column_name, ...
[ WHERE filter_expression ] ] [ ,... ]
[ FOR created_tablename_n ]

Though this syntax shows the CSV format, you can also use the PREVIEW PIPELINE statement with the other formats.

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified data pipeline for the preview.
`created_tablename`	identifier	The identifier for the name of the table that you create before executing the `PREVIEW PIPELINE` SQL statement.
`created_tablename_n`	identifier	For multiple tables, the identifier for the name of another table that you create before executing the `PREVIEW PIPELINE` SQL statement. Use the `FOR` keyword to specify which table content to preview.
`preview_column_formula`	identifier	The identifier for the formula of the data to load. For example, for the data in the first field of the inline source, use `$1`. If you need to add a transformation, you can use functions to transform data, such as `CONCAT($1, $2)`, to load the concatenation of the first two fields in the inline source data.
`preview_column_name`	identifier	The name of the column in the target table.

SQL Statement Options

Option Key	Default	Data Type	Description
SOURCE INLINE	Required	STRING	The string that contains data for the preview of the data pipeline load. For example, source data can be `'oci,ent,ocient\|ware,house,warehouse'`, where `\|` is the record delimiter and `,` is the field delimiter. For special characters, such as `\t`, use an escape sequence such as `e'oci,ent,oci\tent\|ware,house,ware\thouse'`. This option supports these text-based formats: CSV, JSON, and XML.
MODE	`'transform'`	STRING	Indicates whether to perform a validation of the PREVIEW PIPELINE SQL statement. Valid values are: `'validate'` and `'transform'`. Set this option to `'validate'` to check whether the creation of the data pipeline succeeds. If the pipeline is valid, the statement produces no output. Otherwise, the statement returns an error. A valid data pipeline has the correct syntax for the source and extract options, as well as any transformation functionality. After you have a valid pipeline with the `PREVIEW PIPELINE` statement, you can simply replace the `PREVIEW` keyword with `CREATE` to create the pipeline. Set this option to `'transform'` to retrieve a preview of the results of the pipeline.
SHOW_ERRORS_AS_JSON	`false`	BOOLEAN	Indicates whether to show errors. Values are `true` or `false`. If the value is `true`, the Ocient System returns record-level errors as JSON blobs rather than human-readable messages.
LIMIT	10	INTEGER	The number of rows, specified as an integer, to return in the preview results for sources with many rows. The default value is 10 rows.
INTERMEDIATE_VALUES	`false`	BOOLEAN	Indicates whether to capture intermediate values during a transformation sequence. Values are `true` or `false`. If the value is `true`, the Ocient System appends an extra column to the result set. Each value in the column contains a JSON blob that describes the intermediate values processed for each column after each transformation.

You must specify at least one column name in the SELECT part of the syntax. The name of the specified column must match the name of the column in the created table. The number of columns in the SELECT part can be less than those in the created table. With this mismatch, the PREVIEW PIPELINE SQL statement might succeed even if the CREATE PIPELINE statement fails.

For definitions of other extract options, see the CREATE PIPELINE SQL statement options in CREATE PIPELINE. Examples Preview Pipeline Using CSV Format Preview the load of two rows of data. First, create a table to serve as the context for the load. The previewload table contains three columns with these data types: string, integer, and Boolean.

SQL

CREATE TABLE previewload (col1 VARCHAR, col2 INT, col3 BOOLEAN);

Create the preview pipeline testpipeline with this data: 'hello,2,true|bye,3,false'. Specify the CSV extract format, | record delimiter, and the , field delimiter. Load the data without transformation.

SQL

PREVIEW PIPELINE testpipeline
    SOURCE INLINE 'hello,2,true|bye,3,false'
    EXTRACT
        FORMAT CSV
        RECORD_DELIMITER '|'
        FIELD_DELIMITERS [',']
INTO previewload
SELECT
    $1 AS col1,
    $2 AS col2,
    $3 AS col3;

Output

Text

col1                                         col2       col3
\--------------------------------------------------------------
hello                                        2          true
bye                                          3          false

Fetched 2 rows

Delete the previewload table.

SQL

DROP TABLE previewload;

Preview Pipeline Using CSV Format with Escape Characters Preview the load of two rows of data. First, create a table to serve as the context for the load. The previewload table contains three columns with these data types: string, integer, and Boolean.

SQL

CREATE TABLE previewload (col1 VARCHAR, col2 INT, col3 BOOLEAN);

Create the preview pipeline testpipeline with this data: 'hello\tworld,2,true|bye\tworld,3,false'. Specify the CSV extract format, | record delimiter, and , field delimiter. Load the data without transformation. In this case, the data contains the special character \t. You must escape the character by using the escape sequence e.

SQL

PREVIEW PIPELINE testpipeline
    SOURCE INLINE e'hello\tworld,2,true|bye\tworld,3,false'
    EXTRACT
        FORMAT CSV
        RECORD_DELIMITER '|'
        FIELD_DELIMITERS [',']
    INTO previewload
    SELECT
        $1 AS col1,
        $2 AS col2,
        $3 AS col3;

Output

Text

col1                                         col2       col3
\--------------------------------------------------------------
hello   world                                  2        true
bye     world                                  3        false

Fetched 2 rows

Delete the previewload table.

SQL

DROP TABLE previewload;

Preview Pipeline Using CSV Format with Transformation Create a table to serve as the context for the load. The previewload table contains three string columns.

SQL

CREATE TABLE previewload (col1 VARCHAR, col2 VARCHAR, col3 VARCHAR);

Create the preview pipeline testpipeline with this data: 'hello,world|bye,world'. Specify the CSV extract format, | record delimiter, and the , field delimiter. Load the data with a transformation to concatenate the two strings and return the result in the third column.

SQL

PREVIEW PIPELINE testpipeline
    SOURCE INLINE 'hello,world|bye,world'
    EXTRACT
        FORMAT CSV
        RECORD_DELIMITER '|'
        FIELD_DELIMITERS [',']
    INTO previewload
    SELECT
        $1 AS col1,
        $2 AS col2,
        CONCAT($1,$2) AS col3;

Output

Text

col1                                         col2                                         col3
\---------------------------------------------------------------------------------------------------------------------------------------
hello                                        world                                        helloworld
bye                                          world                                        byeworld

Fetched 2 rows

The third column contains the concatenated result of the first two columns. Delete the previewload table.

SQL

DROP TABLE previewload;

Preview Pipeline Using the Kafka Source Create the previewload table with these columns:

id — Non-NULL integer
salut — Non-NULL string
name — Non-NULL string
surname — Non-NULL string
zipcode — Non-NULL integer
age — Non-NULL integer
rank — Non-NULL integer

SQL

CREATE TABLE previewload (
    id INT NOT NULL,
    salut VARCHAR(3) NOT NULL,
    name VARCHAR(10) NOT NULL,
    surname VARCHAR(10) NOT NULL,
    zipcode INT NOT NULL,
    age INT NOT NULL,
    rank INT NOT NULL);

Create the preview pipeline test_small_kafka_simple_csv. Specify the ddl_csv topic. Indicate that the Kafka consumer should not write its durably-made record offsets to the Kafka Broker by using the WRITE_OFFSETS option set to false. Specify the bootstrap server as servername:0000 and configuration options as "auto.offset.reset": "earliest" by using the BOOTSTRAP_SERVERS and CONFIG options, respectively. Limit the returned results to three rows by using the LIMIT option. Specify the CSV extract format and \n record delimiter by using the FORMAT and RECORD_DELIMITER extract options, respectively.

SQL

PREVIEW PIPELINE test_small_kafka_simple_csv
    SOURCE KAFKA
        TOPIC 'ddl_csv'
        WRITE_OFFSETS false
        BOOTSTRAP_SERVERS 'servername:0000'
        CONFIG '{"auto.offset.reset": "earliest"}'
        LIMIT 3
    EXTRACT
        FORMAT csv
        RECORD_DELIMITER '\n'
    INTO previewload
    SELECT
        INT($1) AS id,
        CHAR($2) AS salut,
        CHAR($3) AS name,
        CHAR($4) AS surname,
        INT($5) AS zipcode,
        INT($6) AS age,
        INT($7) AS rank;

Text

id         salut                                        name                                         surname                                      zipcode    age        rank
\-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
105        Mr                                           Jmhsuxofspx                                  Uaofgayjugb                                  85573      29         2
101        Mr                                           Ijmmtbddkyh                                  Yqbxqnkgidp                                  52393      43         1
109        Mr                                           Bigohpwfwmr                                  Qcxgakpkoeu                                  74420      1          3

Fetched 3 rows

QUIESCE PIPELINE

QUIESCE PIPELINE stops the execution of the specified data pipeline gracefully. When you execute this SQL statement, the pipeline transitions to the QUIESCING status. The Ocient System creates no new work units and waits for all in-flight work units to finish. After the in-flight work finishes, the pipeline transitions to the QUIESCED status and the statement returns. A quiesced pipeline provides a clean breakpoint with no partially loaded files:

For a batch pipeline, every file that belongs to the quiesced pipeline in the sys.pipeline_files system catalog table is in the LOADED, FAILED, LOADED_WITH_ERRORS, or PENDING status.
For continuous pipelines with monitors and Kafka streaming pipelines, the pipeline commits all consumed messages to the external sources and stops consuming before it transitions to the QUIESCED status. Every file that belongs to the quiesced pipeline in the sys.pipeline_files system catalog table is in the LOADED, FAILED, LOADED_WITH_ERRORS, or PENDING status.

Because all loaded data is durable and the pipeline retains its load position, you can safely execute the ALTER PIPELINE, DROP PIPELINE, or START PIPELINE SQL statement on a quiesced pipeline without the risk of missing or duplicate data. In contrast, the STOP PIPELINE SQL statement stops the pipeline as soon as possible and can leave files in intermediate statuses. For details, see STOP PIPELINE. The QUIESCE PIPELINE SQL statement has these behaviors:

You can quiesce a pipeline only in the RUNNING status. You cannot quiesce a pipeline in the COMPLETED, FAILED, STOPPED, or QUIESCED status, or a pipeline that has never started.
Consecutive QUIESCE PIPELINE statements on the same pipeline are idempotent. Only the first statement signals the pipeline to shut down.
To interrupt a graceful stop that takes too long, execute the STOP PIPELINE SQL statement on the pipeline in the QUIESCING status. The Ocient System interrupts the drain and stops the pipeline as soon as possible.
If the in-flight work units reach the error limit of the pipeline during the quiesce, the pipeline transitions to the FAILED status.
The QUIESCE PIPELINE SQL statement returns after the pipeline reaches one of the inactive statuses: STOPPED, QUIESCED, COMPLETED, or FAILED.
Kafka streaming pipelines that set the VERSION source option to 1 do not support the QUIESCE PIPELINE SQL statement. To stop these pipelines, use the STOP PIPELINE statement.

The Ocient System logs the QUIESCING and QUIESCED events in the sys.pipeline_events system catalog table. You can monitor the status of the pipeline in the information_schema.pipeline_status view. Required Privileges You must have the EXECUTE privilege on the pipeline to execute this SQL statement. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

QUIESCE PIPELINE pipeline_name

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified data pipeline.

Example Quiesce an existing pipeline named ad_data_pipeline.

SQL

QUIESCE PIPELINE ad_data_pipeline;

After the statement returns, the pipeline is in the QUIESCED status. To resume the pipeline, execute the START PIPELINE SQL statement.

START PIPELINE

START PIPELINE begins the execution of the specified data pipeline that extracts data and loads it into the target tables specified by the CREATE PIPELINE SQL statement. You can execute START PIPELINE when the pipeline is in the STOPPED, FAILED, or QUIESCED status. A newly created pipeline has the STOPPED status. The pipeline transitions to the QUEUED status while it waits for the Ocient System to schedule it, and then to the RUNNING status after it begins to process data. When you execute the START PIPELINE SQL statement, the Ocient System creates a static list of files in the sys.pipeline_files system catalog table and marks them with the PENDING status. After the system assigns a file to an underlying task, the system marks the file as QUEUED. After the system verifies that the file exists, the system marks the file as LOADING to signify that a Loader Node has started reading the source data. Finally, upon successfully loading the file, the system transitions the status of the file to the terminal status LOADED. A Kafka streaming pipeline never enters the COMPLETED state in the information_schema.pipeline_status view. Instead, the pipeline remains running after you start the pipeline until you stop it or the pipeline reaches the specified error limit using the ERROR LIMIT option. You must have the EXECUTE privilege on the pipeline and the INSERT privilege on any table that is a target in the pipeline. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

START PIPELINE pipeline_name
    [ ERROR
        [ LIMIT <integer_value> ]
        [ FILE_ERROR (FAIL | SKIP_MISSING_FILE | TOLERATE) ] ]
    [ USING LOADERS <loader_names> ]
    [ ON COMPLETION (NO_FLUSH | FLUSH_AND_WAIT | FLUSH_AND_RETURN) ]

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified data pipeline.

SQL Statement Options

Option Key	Default	Data Type	Description
ERROR LIMIT `<integer_value>`	0	INTEGER	Error log option that determines the number of record-level errors that can occur during the execution of a pipeline that the load tolerates before the whole pipeline execution fails. `<integer_value>` is a number greater than or equal to -1. When you set `<integer_value>` to -1, the load tolerates an unlimited number of record-level errors. By default, continuous pipelines tolerate an unlimited number of record-level errors, whereas batch pipelines tolerate zero errors. With multiple target tables, record-level errors are specific to a table. For example, if a transformation for loading one table succeeds but another transformation for loading another table fails, the row corresponding to the successful transformation loads to the first table but not the other.
ERROR FILE_ERROR `<error_action>`	`FAIL`	STRING	For pipelines that load data from S3 or local file sources, this error configuration option determines how to treat unrecoverable file-level errors. Examples of unrecoverable file-level errors are: The file is listed when the pipeline starts, but is missing later during the load. The Gzip file is corrupted and cannot be decompressed. The file cannot be downloaded from the source. Record-level error that is not tolerable occurs when tokenizing or transforming data in the file. `<error_action>` can be one of these keywords: `FAIL` — Fail the whole pipeline because of a file-level error. `SKIP_MISSING_FILE` — Only tolerate errors that occur due to missing files. If a file exists in the list when the pipeline starts but is missing later during the load, skip the file and continue with the next file. `TOLERATE` — Tolerate all unrecoverable file-level errors. In this mode, the load also tolerates an unlimited number of record-level errors. The `FAILED`, `SKIPPED`, and `LOADED_WITH_ERRORS` file statuses appear in the `sys.pipeline_files` system catalog tables, respectively, and indicate how the pipeline handled the file error. Value Range: `FAIL`, `SKIP_MISSING_FILE`, or `TOLERATE`.
USING LOADERS `loader_names`	`NULL`	LIST OF STRINGS	Specify one or more names of Loader Nodes as a comma-separated list for executing the `START PIPELINE` SQL statement. If you do not use this option, the Ocient System uses all of the Loader Nodes that are active to execute the pipeline. You can find node names in the `sys.nodes` system catalog table.
ON COMPLETION `<completion_mode>`	`NO_FLUSH`	STRING	Completion type option that specifies the behavior when the pipeline finishes loading. This option determines when the remaining pages are converted into Segments. `NO_FLUSH` — Do not force a flush of pages. Rely on watermarks and timeouts to trigger final conversion to Segments. `FLUSH_AND_WAIT` — Trigger a flush of pages, initiating final conversion to Segments. The pipeline blocks and waits for the conversion to Segments to complete before marking the pipeline as `COMPLETED`. `FLUSH_AND_RETURN` — Trigger a flush of pages, initiating the final conversion to Segments. The Ocient System marks the pipeline as `COMPLETED` immediately following the flush without waiting for conversion to Segments to complete.

For the query to execute successfully, the specified node names must identify nodes that have:

ACTIVE operational status
streamloader role

When you execute the START PIPELINE SQL statement, the Ocient System creates a static list of files only for batch pipelines and a dynamic list for continuous pipelines in the sys.pipeline_files system catalog table.

Examples Start an existing pipeline named ad_data_pipeline with default settings.

SQL

START PIPELINE ad_data_pipeline;

Start an existing pipeline named ad_data_pipeline with error tolerance (tolerate 10 errors before aborting the pipeline). For details about error tolerance, see Error Tolerance in Data Pipelines.

SQL

START PIPELINE ad_data_pipeline
    ERROR LIMIT 10;

Data pipelines log a message for each pipeline error to the sys.pipeline_errors system catalog table, even if you do not specify the ERROR option. Use BAD_DATA_TARGET settings to capture the original source data.

Start an existing pipeline named ad_data_pipeline using the Loader Node named stream-loader1.

SQL

START PIPELINE ad_data_pipeline
    USING LOADERS "stream-loader1";

To resume a pipeline with file loading, see Data Pipeline Behavior Considerations. To restart a Kafka data pipeline, see Data Pipeline Behavior Considerations. For pipeline dependencies, see Data Pipeline Behavior Considerations.

STOP PIPELINE

STOP PIPELINE stops the execution of the pipeline and its associated tasks as soon as possible. The Ocient System interrupts in-flight work, and the pipeline transitions to the STOPPED status. Files can remain in intermediate statuses, such as LOADING or QUEUED. To stop a pipeline gracefully with no partially loaded files, use the QUIESCE PIPELINE SQL statement instead. You can also execute STOP PIPELINE on a pipeline in the QUIESCING status to interrupt a graceful stop. After you stop a pipeline, you can execute the START PIPELINE SQL statement on the pipeline to run the pipeline again. Regardless, the load deduplicates any records previously loaded in the same pipeline. You must have the EXECUTE privilege on the pipeline to execute this SQL statement. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

STOP PIPELINE pipeline_name

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified data pipeline.

Example Stop an existing pipeline named ad_data_pipeline.

SQL

STOP PIPELINE ad_data_pipeline;

You can see the status of the parent tasks in the sys.tasks system catalog table and see the status of the child tasks in the sys.subtasks system catalog table.

ALTER PIPELINE

ALTER PIPELINE updates the configuration and transformations of an existing data pipeline. The statement performs an in-place modification that preserves the existing state of the pipeline, including its identifier, durability horizon, and metrics. You can rename the pipeline, update the advanced tuning options, bad data target, source, monitor, lookup sources, and extraction options, and evolve the data pipeline schema in the SELECT clause. For a comparison with the CREATE OR REPLACE PIPELINE SQL statement, see Update a Pipeline. For details about schema evolution, see Load Avro Data.

To execute this SQL statement, the data pipeline must be inactive, which means that it has the STOPPED, QUIESCED, COMPLETED, or FAILED status. You can use the QUIESCE PIPELINE SQL statement to stop the execution of the data pipeline gracefully, or the STOP PIPELINE statement to stop it immediately, before you alter it.

All sections of the statement are optional, but you must specify at least one section. The statement changes only the sections that you specify. Other parts of the pipeline remain unchanged. Specified sections must adhere to the rules defined in SOURCE Options, LOOKUP Options, EXTRACT Options, Bad Data Targets, and Advanced Pipeline Tuning Options. For pipelines with multiple tables, specify each SELECT clause for tables in the existing pipeline. Each WHERE clause must match the corresponding clause of the existing pipeline. Use the FORCE keyword to change a non-backward-compatible schema or to add, modify, or remove WHERE clauses. When you specify the FORCE keyword, the statement replaces all WHERE clauses of the existing pipeline with the WHERE clauses in the ALTER PIPELINE statement. If the existing pipeline contains multiple WHERE clauses and you specify only a subset of them, the statement drops the WHERE clauses that you omit. Required Privileges You must have the ALTER privilege on the pipeline to execute this SQL statement. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

ALTER PIPELINE [ IF EXISTS ] pipeline_name
    [ RENAME TO pipeline_new_name ]
    [ <advanced_pipeline_options> ]
    [ BAD_DATA_TARGET <kafka_bad_data_target> ]

    [ SOURCE ( <s3_source> | <filesystem_source> | <hdfs_source> | <kafka_source> )
        [ MONITOR ( <sqs_monitor> | <kafka_monitor> ) ] ]

    [ LOOKUP lookup_source <lookup_options> ]  [ ... ]

    [ EXTRACT
        ( <delimited_extract_options> | <asn1_extract_options> |
        <avro_extract_options> | <json_extract_options> |
        <parquet_extract_options> | <binary_extract_options> |
        <xml_extract_options> )
        [ <general_extract_options> ] ]

    [ [ INSERT ] INTO destination_table_name

    SELECT
        expression as col_alias_target,
        expression2 as col_alias_target2,
        ...
    [ WHERE filter_expression ]

    [ [ INSERT ] INTO destination_table_name_n
    SELECT
        expression_n as col_alias_target_n,
        expression2_n as col_alias_target2_n,
        ...
    [ WHERE filter_expression_n ] ] [ ... ] ]

    [ FORCE ]

For the definitions of the syntax subsections, see the CREATE PIPELINE SQL statement.

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified pipeline to alter.
`pipeline_new_name`	string	The new name of the data pipeline.

Limitations The ALTER PIPELINE SQL statement has these limitations:

If the pipeline has the COMPLETED status, you can only rename the pipeline.
You cannot alter the pipeline type, source type, or monitor type.
You cannot alter the target tables. If the pipeline has multiple target tables, the statement must include all transformations for all tables. You cannot alter the list or the order of the target tables.
If an option is not configurable, the FORCE keyword does not make the option configurable. The Option Configurability column in each options table indicates whether you can alter the option.
Altering options can affect loading results.
The statement must not contain a FROM clause. If you include a FROM clause, the system rejects the statement.
When you configure a section, you must specify the section name to avoid ambiguity. The section names are BAD_DATA_TARGET, SOURCE, MONITOR, LOOKUP, EXTRACT, and INTO. If the section name has an associated value, you must also specify the value. For example, to alter the SECRET_ACCESS_KEY option for an S3 source and an SQS monitor, you must specify SOURCE S3 and MONITOR SQS.

Altering options requires the pipeline to have the QUIESCED status. If the pipeline has the FAILED or STOPPED status, you can use the FORCE keyword, but do so with caution. Altering a pipeline with the FAILED status can cause data duplication.

Examples Rename a Data Pipeline Rename an existing pipeline named x to y without changing any other part of the pipeline.

SQL

ALTER PIPELINE x RENAME TO y;

Rename a Data Pipeline and Change the Source Connection Configuration Rename an existing pipeline named x to y and update the configuration of its HDFS source.

SQL

ALTER PIPELINE x RENAME TO y SOURCE HDFS CONFIG '{"dfs.client.socket-timeout": 300000}';

Rename a Data Pipeline and Change the Extraction and Transformations Rename an existing pipeline named x to y, update the extraction schema inference to use the new_file.avro file, and update the transformation for the t table to load the new_field source field into the new_col column.

SQL

ALTER PIPELINE x RENAME TO y EXTRACT INFER_FROM ['new_file.avro'] INTO t SELECT $new_field AS new_col;

Rename a Data Pipeline and Update the Source Credentials Rename an existing pipeline named orders_pipeline to sales_pipeline and update the access credentials of its S3 source.

SQL

ALTER PIPELINE orders_pipeline
    RENAME TO sales_pipeline
    SOURCE S3
        ACCESS_KEY_ID 'new_access_key_id'
        SECRET_ACCESS_KEY 'new_secret_access_key';

Update Credentials for an S3 Source and an SQS Monitor Update the secret access key of the S3 source and the SQS monitor for an existing pipeline named orders_pipeline. Even though the statement alters only the SECRET_ACCESS_KEY options, you must specify the SOURCE S3 and MONITOR SQS section names.

SQL

ALTER PIPELINE orders_pipeline
    SOURCE S3
        SECRET_ACCESS_KEY 'new_secret_access_key'
    MONITOR SQS
        SECRET_ACCESS_KEY 'new_monitor_secret_access_key';

Schema Evolution Examples The remaining examples demonstrate schema evolution and use this table and data pipeline definition. You must stop the execution of the data pipeline before performing schema evolution operations. Create the users table with these columns:

id — Universally Unique IDentifier (UUID) of the user
firstname — First name of the user
lastname — Last name of the user
birthyear — Year of birth
groups — List of groups where the user belongs

SQL

CREATE TABLE users(
    id         UUID NOT NULL,
    firstname  VARCHAR(255) NOT NULL,
    lastname   VARCHAR(255) NOT NULL,
    birthyear  INT,
    groups     VARCHAR(255)[] NOT NULL DEFAULT 'char[]'
);

Assume you have user data in Avro format in multiple files in the /data/users directory. Create the users_pipeline data pipeline for the Avro files containing user data *.avro. The schema configuration instructs the system to infer from a random file using the INFER_FROM option.

SQL

CREATE PIPELINE users_pipeline
    SOURCE filesystem
        FILTER '/data/users/*.avro'
    EXTRACT
        FORMAT avro
        SCHEMA {
            INFER_FROM sample_file
        }
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear;

Start the data pipeline.

SQL

START PIPELINE users_pipeline;

Schema Evolution with a Column Addition Stop the data pipeline.

SQL

STOP PIPELINE users_pipeline;

Change the schema of an existing pipeline named users_pipeline to add the groups column. Access the array of strings within the column.

SQL

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
    $groups[]   AS groups;

Start the data pipeline again.

SQL

START PIPELINE users_pipeline;

Check Whether the Data Pipeline to Change Exists for Column Addition Stop the data pipeline.

SQL

STOP PIPELINE users_pipeline;

Change the schema of an existing pipeline named users_pipeline to add the groups column. Use the IF EXISTS keywords to check whether the data pipeline exists. Access the array of strings within the column.

SQL

ALTER PIPELINE IF EXISTS users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
    $groups[]   AS groups;

Start the data pipeline again.

SQL

START PIPELINE users_pipeline;

Schema Evolution with a Column Removal Stop the data pipeline.

SQL

STOP PIPELINE users_pipeline;

Change the schema of an existing pipeline named users_pipeline to remove the birthyear column.

SQL

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname;

Start the data pipeline again.

SQL

START PIPELINE users_pipeline;

Schema Evolution with a Column Data Type Modification Stop the data pipeline.

SQL

STOP PIPELINE users_pipeline;

Change the schema of an existing pipeline named users_pipeline to narrow the INT data type to a SMALLINT type for the birthyear column using the SMALLINT casting function.

SQL

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    SMALLINT($birthyear)  AS birthyear;

Start the data pipeline again.

SQL

START PIPELINE users_pipeline;

Schema Evolution with the Addition of a Filter Condition Stop the data pipeline.

SQL

STOP PIPELINE users_pipeline;

Change the schema of an existing pipeline named users_pipeline to filter the year of birth to be greater than 1950 using the WHERE clause. Add the FORCE keyword to instruct the data pipeline to evolve the schema with the filter condition.

SQL

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear
WHERE
    $birthyear > 1950
FORCE;

Start the data pipeline again.

SQL

START PIPELINE users_pipeline;

ALTER PIPELINE RENAME

ALTER PIPELINE RENAME TO SQL statement changes the name of the pipeline object, while retaining its identifier, options, and other metadata. The Ocient System reflects this change in the sys.pipelines system catalog table. Then, you must use the new name when you refer to the pipeline in SQL statements.

If you rename a Kafka streaming pipeline that does not specify the GROUP_ID option, the consumer group identifier changes with the name, because the default identifier derives from the name of the pipeline. The renamed pipeline does not resume from the last committed offset of the original consumer group. Instead, it starts according to the auto.offset.reset policy, which can reload data from the beginning of the topic or skip to the end of it. To preserve the position of the load across a rename, specify the GROUP_ID option when you create the pipeline.

You must have the ALTER privilege on the pipeline to execute this SQL statement. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

ALTER PIPELINE [ IF EXISTS ] pipeline_original_name RENAME TO pipeline_new_name

Parameter	Data Type	Description
`pipeline_original_name`	string	The name of the existing data pipeline.
`pipeline_new_name`	string	The new name of the data pipeline.

Example Rename an existing pipeline named ad_data_pipeline to renamed_pipeline.

SQL

ALTER PIPELINE ad_data_pipeline RENAME TO renamed_pipeline;

EXPORT PIPELINE

EXPORT PIPELINE returns the CREATE PIPELINE SQL statement used to create the pipeline object. You can use the output of this statement to recreate an identical pipeline when you remove the original pipeline.

The execution of this statement censors sensitive S3 values like ACCESS_KEY_ID and SECRET_ACCESS_KEY and Kafka Consumer Configuration password-type fields. The database replaces them with *.

To execute this statement, you must have the VIEW privilege on the pipeline and any table the pipeline targets. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

EXPORT PIPELINE pipeline_name

Parameter	Data Type	Description
`pipeline_name`	string	The name of the specified data pipeline.

Example Export an existing pipeline in the database ad_data_pipeline.

SQL

EXPORT PIPELINE ad_data_pipeline;

CREATE PIPELINE FUNCTION

CREATE PIPELINE FUNCTION enables you to define a function for loading data. Define the function behavior using the Apache® Groovy™ language. For details bout this language, see The Apache Groovy Programming Language. Function arguments and output are strongly typed and immutable. You can test the execution of your function using the PREVIEW PIPELINE SQL statement.

The Ocient System does not support the overload of function names.

Syntax

SQL

CREATE [ OR REPLACE ] PIPELINE FUNCTION [ IF NOT EXISTS ]
    function_name( input_argument [, ...] )
    LANGUAGE GROOVY
    RETURNS output_argument_definition
    IMPORTS [ library_name [, ...] ]
    AS $$
        groovy_declaration
    $$

Parameter	Type	Description
`function_name`	string	A unique identifier for the data pipeline function.
`input_argument`	string	The name of one or more input arguments of the function. Specify data types for input arguments according to the support data types defined in Data Types for Data Pipelines. For the data type declaration, use `NOT NULL` where applicable for maximum performance.
`output_argument_definition`	string	The type definition of the output from the function.
`library_name`	string	The name of one or more Java libraries. You can include libraries by using the `IMPORTS` clause or specifying the fully-qualified class (e.g., `java.lang.Integer`) path in the source definition.
`groovy_declaration`	string	The Groovy definition of the function.

Install and Enable Third-Party Libraries

You can use the default list of supported third-party libraries or additional third-party libraries that you install.

Supported Libraries

Data pipeline functions can import classes from the default list of supported third-party libraries. This table provides the resources for each supported library package.

Library Package	Resource
`java.lang.*`	java.lang
`java.util.*`	java.util
`java.nio.ByteBuffer.*`	ByteBuffer
`groovy.json.*`	groovy.json
`groovy.xml.*`	groovy.xml
`groovy.yaml.*`	groovy.yaml
`org.apache.groovy.datetime.extensions.*`	org.apache.groovy.datetime.extensions
`org.apache.groovy.dateutil`	org.apache.groovy.dateutil.extensions
`com.ocient.streaming.data.types.*`	Data Types for User-Defined Data Pipeline Functions

Additional Libraries

You can install and enable additional third-party libraries to import for use in your data pipeline functions. You must install the JAR package on all Loader Nodes in the /opt/ocient/current/lib/extractorengine_udt folder. Then, add the fully qualified class name in the function import list as part of the library_name parameter. For example, to reference the ByteBuffer class from the com.fastbuffer package, specify com.fastbuffer.ByteBuffer in the library_name parameter and use the class in the Groovy definition as var x = new com.fastbuffer.ByteBuffer().

Groovy Data Type Mapping

For the Groovy definition, the Ocient System maps its SQL data type to the corresponding Groovy data type. Your Groovy code should use the Groovy data type defined in this table for any input arguments and output.

SQL Data Type	Groovy Data Type
`BIGINT`	`java.lang.Long`
`BINARY(N) or HASH(N)`	`byte[]`
`BOOLEAN`	`java.lang.Boolean`
`CHAR(N) or VARCHAR(N)`	`java.lang.String`
`DATE`	`java.time.LocalDate`
`DECIMAL(P,S)`	`com.ocient.streaming.data.types.Decimal`
`DOUBLE`	`java.lang.Double`
`INT`	`java.lang.Integer`
`IPV4`	`java.net.Inet4Address`
`IP`	`java.net.Inet6Address`
`ST_POINT`	`com.ocient.streaming.data.types.gis.STPoint`
`ST_LINESTRING`	`com.ocient.streaming.data.types.gis.STLinestring`
`ST_POLYGON`	`com.ocient.streaming.data.types.gis.STPolygon`
`FLOAT`	`java.lang.Float`
`SMALLINT`	`java.lang.Short`
`TIME`	`com.ocient.streaming.data.types.Time`
`TIMESTAMP`	`com.ocient.streaming.data.types.Timestamp`
`BYTE`	`java.lang.Byte`
`TUPLE<<type1, type2, …>>`	`com.ocient.streaming.data.types.OcientTuple`
`TYPE[]`	`java.util.List<TYPE>`
`UUID`	`java.util.UUID`
`VARBINARY(N)`	`byte[]`
`VARCHAR(N)`	`java.lang.String`

Example Create the sort_function data pipeline function to sort an array of integers. The function has two input arguments: value, a non-NULL array of integers, and ascending, the sort order. The function returns a non-NULL array of integers. If value is empty, the function throws an error. The function imports these Java libraries:

java.lang.Integer
java.util.ArrayList
java.util.Collections
java.util.Comparator
java.util.List

Define the Groovy code. Because the input arguments do not change, the example Groovy code first copies the value argument, sorts the copied list according to the sort order, and returns the sorted array.

SQL

CREATE PIPELINE FUNCTION
    sort_function( value INT[] NOT NULL, ascending BOOLEAN NOT NULL)
    LANGUAGE GROOVY
    RETURNS INT[] NOT NULL
    IMPORTS [
        'java.lang.Integer',
        'java.util.ArrayList',
        'java.util.Collections',
        'java.util.Comparator',
        'java.util.List'
        ]
    AS $$
        /* Throw an error if the array is empty */
        if (value.isEmpty()){
            throw new PipelineFunctionException("Unexpected empty array");
        }

        /* Make a copy of the list. */
        List<Integer> sorted = new ArrayList<>((List<Integer>)value);

        /* Sort the array elements according to the specified order. */
        sorted.sort(ascending ? Comparator.naturalOrder() : Comparator.reverseOrder());

        /* Return the sorted array. */
        return sorted;
    $$;

View the creation information about the sort_function function using the sys.pipeline_functions system catalog table. This statement returns the function name, return type, argument names, data types of the arguments, and the list of imported libraries.

SQL

SELECT name, return_type, argument_names, argument_types, imported_libraries
FROM sys.pipeline_functions;

DROP PIPELINE FUNCTION

DROP PIPELINE FUNCTION removes an existing pipeline function. You must have the DROP privilege on the pipeline function to execute this SQL statement. For details, see Data Control Language (DCL) Statement Reference. Syntax

SQL

DROP PIPELINE FUNCTION [ IF EXISTS ] function_name [, ...]

Parameter	Data Type	Description
`function_name`	string	The name of the specified data pipeline function to remove. You can drop multiple pipelines by specifying additional function names, separated by commas.

Examples Remove the Existing Pipeline Function Remove an existing pipeline function named sort_function.

SQL

DROP PIPELINE FUNCTION sort_function;

Remove an Existing Pipeline Function by Checking for Existence Remove an existing pipeline function named sort_function or return a warning if the Ocient System does not find the function in the database.

SQL

DROP PIPELINE FUNCTION IF EXISTS sort_function;

EXPORT PIPELINE FUNCTION

EXPORT PIPELINE FUNCTION returns the CREATE PIPELINE FUNCTION SQL statement used to create the pipeline function. You can use the output of this statement to recreate an identical pipeline function when you remove the original function. Required Privileges To execute this statement, you must have the VIEW privilege on the pipeline function. For details, see the Data Control Language (DCL) Statement Reference. Syntax

SQL

EXPORT PIPELINE FUNCTION function_name

Parameter	Data Type	Description
`function_name`	string	The name of the specified data pipeline function.

Example Return the CREATE PIPELINE FUNCTION SQL statement for the existing pipeline function named sort_function.

SQL

EXPORT PIPELINE FUNCTION sort_function;

Output

Text

export                                       
---------------------------------------------
CREATE PIPELINE FUNCTION "clamp_function"("value" INT NOT NULL, "low" INT NOT NULL, "high" INT NOT NULL)
    LANGUAGE groovy
    RETURNS INT NOT NULL
    IMPORTS []
    AS $$
/* Reject an invalid range. */
        if (low > high) {
            throw new PipelineFunctionException("Invalid range: low > high");
        }

        /* Clamp the value to the inclusive range [low, high]. */
        if (value < low) {
            return low;
        }
        if (value > high) {
            return high;
        }

        /* The value is already within range. */
        return value;
    $$;

Fetched 1 row

Load Data Data Pipeline Load of CSV Data from S3 Data Pipeline Load of Parquet Data from S3 Transform Data in Data Pipelines Data Pipeline Behavior Considerations Load Metadata and File-Based Partitioned Data in Data Pipelines Data Control Language (DCL) Statement Reference Identifiers

​CREATE PIPELINE

​Pipeline Identity and Naming

​Update a Pipeline

​Where the Load Resumes

​Data Pipeline Modes

​Required Options

​SELECT Statement and Data Transformation

​NULL and Default Value Handling

​Required Privileges

​Examples

​Load JSON Data from Kafka

​Load Delimited Data from S3

​Continuous File Load of Delimited Data in CSV Files

​Continuous File Load of CSV Files with a Unique Message Format

​Transactional Load of Delimited Data in CSV Files

​SOURCE Options

​File-Based Source Options

​S3 Source Options

​FILESYSTEM Source Options

​HDFS Source Options

​KAFKA Source Options

​Continuous File Loading Source Options

​LOOKUP Options

​EXTRACT Options

​General Extract Options

​ASN.1 Extract Options

​Avro Extract Options

​BINARY Extract Options

​Delimited and CSV Extract Options

​JSON Extract Options

​PARQUET Extract Options

​XML Extract Options

​Bad Data Targets

​Kafka Bad Data Target

​Kafka Bad Data Target Options

​Advanced Pipeline Tuning Options

​DROP PIPELINE

​PREVIEW PIPELINE

​Preview Sources

​Preview Error Handling

​Preview Limitations

​Output

​Output

​Output

​QUIESCE PIPELINE

​START PIPELINE

​STOP PIPELINE

​ALTER PIPELINE

​ALTER PIPELINE RENAME

​EXPORT PIPELINE

​CREATE PIPELINE FUNCTION

​Install and Enable Third-Party Libraries

​Supported Libraries

​Additional Libraries

​Groovy Data Type Mapping

​DROP PIPELINE FUNCTION

​EXPORT PIPELINE FUNCTION

​Output

​Related Links

CREATE PIPELINE

Pipeline Identity and Naming

Update a Pipeline

Where the Load Resumes

Data Pipeline Modes

Required Options

SELECT Statement and Data Transformation

NULL and Default Value Handling

Required Privileges

Examples

Load JSON Data from Kafka

Load Delimited Data from S3

Continuous File Load of Delimited Data in CSV Files

Continuous File Load of CSV Files with a Unique Message Format

Transactional Load of Delimited Data in CSV Files

SOURCE Options

File-Based Source Options

S3 Source Options

FILESYSTEM Source Options

HDFS Source Options

KAFKA Source Options

Continuous File Loading Source Options

LOOKUP Options

EXTRACT Options

General Extract Options

ASN.1 Extract Options

Avro Extract Options

BINARY Extract Options

Delimited and CSV Extract Options

JSON Extract Options

PARQUET Extract Options

XML Extract Options

Bad Data Targets

Kafka Bad Data Target

Kafka Bad Data Target Options

Advanced Pipeline Tuning Options

DROP PIPELINE

PREVIEW PIPELINE

Preview Sources

Preview Error Handling

Preview Limitations

Output

Output

Output

QUIESCE PIPELINE

START PIPELINE

STOP PIPELINE

ALTER PIPELINE

ALTER PIPELINE RENAME

EXPORT PIPELINE

CREATE PIPELINE FUNCTION

Install and Enable Third-Party Libraries

Supported Libraries

Additional Libraries

Groovy Data Type Mapping

DROP PIPELINE FUNCTION

EXPORT PIPELINE FUNCTION

Output

Related Links