Data pipelines enable you to load data from your chosen source into the Ocient Hyperscale Data Warehouse™. 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. You can 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. Also, you can administer privileges for data pipelines and data pipeline functions. For details, see Data Control Language (DCL) Statement Reference.

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.

You must have the ALTER privilege on the pipeline. For details, see Data Control Language (DCL) Statement Reference.

Syntax

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 ]
kafka_source ::=
    KAFKA
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic_name
        [ WRITE_OFFSETS write_offsets ]
        [ CONFIG config_option ]
        [ AUTO_OFFSET_RESET (
            ( 'largest' | 'latest' ) | ( 'earliest' | 'smallest' ) | 'error' )
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 (
            ( 'largest' | 'latest' ) | ( 'earliest' | 'smallest' ) | 'error' )
        [ 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 ]
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 ( 'sample_file' | 'all_files' ) ]
        } ]
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 SQL statement.

Update a Pipeline

You can update a pipeline with the OR REPLACE clause in the CREATE PIPELINE SQL statement. Use this clause when you want to continue loading from the current place in a continuous load, but you need to modify transformations or other settings. If you specify the OR REPLACE clause and the pipeline already exists, the database replaces the original pipeline object with the options specified in the new CREATE OR REPLACE PIPELINE statement. When you replace an existing pipeline, the pipeline retains its current position in the source data so that data is not duplicated when the pipeline is resumed with a START PIPELINE SQL statement. First, you must stop a pipeline before executing the CREATE OR REPLACE 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.
• Apache® 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.

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.

• SOURCE Options
• EXTRACT Options
• Data Formats for Data Pipelines
• Transform Data in Data Pipelines

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.

...
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.

...
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.

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

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-examples 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

CREATE PIPELINE orders_pipeline
    SOURCE S3
        ENDPOINT 'https://s3.us-east-1.amazonaws.com'
        BUCKET 'ocient-examples'
        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.

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
;

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.

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.

• * 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 | 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 | Optional.
    
    The method the Ocient System uses to decompress file data. The only supported option is gzip.
    To load uncompressed data, omit the COMPRESSION_METHOD option.
    
    ℹ️The COMPRESSION_METHOD option is only applicable to file-based sources. Kafka pipelines automatically decompress data based on the Kafka topic configuration. | | SORT_BY | filename | STRING | 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. | | SORT_DIRECTION | ASC | STRING | Optional.
  
  The sort direction, either ASC ascending or DESC descending, 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. | | SORT_REWRITE | NULL | STRING | 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 | 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 | 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 | 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 | 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 | 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 | 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.

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.

FILESYSTEM Source Options

No options exist specific to the file system source (SOURCE FILESYSTEM) except for general file-based source options.

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.

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 Apache® Hadoop® Distributed File System (HDFS) data sources (SOURCE HDFS).

For advanced options, see Data Pipeline Load of JSON Data from HDFS.

KAFKA Source Options

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

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.

General File Monitor Options

SQS Monitor Options

Use these options when you use MONITOR sqs for Amazon SQS.

Kafka Monitor Options

Use these options when you use MONITOR kafka for Kafka.

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.

EXTRACT Options

General Extract Options

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

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.

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 loads, the schema configuration must include either the URL option or the INLINE option, but not both.

For file-based loads, the schema configuration can include:

• 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.

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.

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.

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.

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.

Example

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

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

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.

DROP PIPELINE

DROP PIPELINE removes an existing pipeline in the current database. You cannot remove a pipeline that is running.

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

Syntax

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

Examples

Remove Existing Data Pipeline

Remove an existing pipeline named ad_data_pipeline.

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.

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.

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 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.
• The preview of a continuous data pipeline for loading data from files is not supported.

Syntax

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 ]

SQL Statement Options

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.

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.

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

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

Delete the previewload table.

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.

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.

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

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

Delete the previewload table.

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.

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.

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

  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.

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

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.

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;

  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

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.

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 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

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) ]

SQL Statement Options

Examples

Start an existing pipeline named ad_data_pipeline with default settings.

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.

START PIPELINE ad_data_pipeline
    ERROR LIMIT 10;

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

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. 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

STOP PIPELINE pipeline_name

Example

Stop an existing pipeline named ad_data_pipeline.

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 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.

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

Syntax

ALTER PIPELINE [ IF EXISTS ] pipeline_original_name RENAME TO pipeline_new_name

Example

Rename an existing pipeline named ad_data_pipeline to renamed_pipeline.

ALTER PIPELINE ad_data_pipeline RENAME TO renamed_pipeline;

EXPORTPIPELINE

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.

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

EXPORT PIPELINE pipeline_name

Example

Export an existing pipeline in the database ad_data_pipeline.

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.

Syntax

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

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.

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.

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.

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.

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

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

Examples

Remove the Existing Pipeline Function

Remove an existing pipeline function named sort_function.

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.

DROP PIPELINE FUNCTION IF EXISTS sort_function;

Related Links

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