APPLY PIPELINE

The Apply Pipeline component allows you to group multiple pipeline components as a unit, which can be bypassed as a group, and optionally in groups of N records at a time using the BATCH option.

There are two types of APPLY PIPELINE components: Inner and After. Options differ depending on which kind is used.

Inner Pipeline

An inner Apply Pipeline is an ETL block that happens as data is read from the source, in batches of records, and is part of the SELECT SQL CDC operation -- not a separate command. Only the BREAK_ON_EMPTY option is supported.

-- As part of the SELECT operation:
SELECT * FROM DB [sql] (SELECT * FROM sys.tables) APPLY PIPELINE (...);

After Pipeline

An After Pipeline is a separate ETL block after data has been read or loaded. It supports the BREAK_ON_EMPTY and BATCH options inside a Read Pipeline, and the additional UPSERTS and DELETES options for Write Pipelines.

SELECT * FROM DB [sql] (SELECT * FROM sys.tables); 
-- As a separate block:
APPLY PIPELINE (...);

The APPLY PIPELINE operation is not supported before the first SELECT or LOAD operation. Nesting an APPLY PIPELINE block within another one is not supported.

Executes a pipeline block comprised of one or more pipeline components on the entire data set.

APPY PIPELINE
	(...)	
	{ BREAK_ON_EMPTY }
	{ BATCH N }

	-- Applies to Target Pipelines only:
	{ UPSERTS } 
	{ DELETES }
;

BATCH	Executes the pipeline block N records at a time, as many times as needed, and aggregates the results
BREAK_ON_EMPTY	When specified, the pipeline will exit as soon as any of its ETL component returns 0 records
DELETES	Applies this pipeline to the Delete Stream only of the Writer
UPSERTS	Applies this pipeline to the Upsert Streams only of the Writer

Example 1

-- Get data from an RSS Feed
SELECT * FROM HTTP [Rss Feed] (GET /econi.xml) APPLY TX (//item);

-- Run a pipeline block that can be bypassed as a whole
-- This transformation block happens after all the data 
-- has been read from the HTTP source

APPLY PIPELINE (
	
	-- Execute a statement which returns 0 records; the pipeline will exit immediately as a result
	EXEC ON DB [mysqldb] (
	  SELECT * FROM @pipelinedata() WHERE 1=0;
	);

	-- will not execute because the BREAK_ON_EMPTY option is specified
	ADD COLUMN 'dateFound' WITH EXPRESSION '#utcnow()' IGNORE_IF_EXISTS;

	-- will not execute because the BREAK_ON_EMPTY option is specified
	APPLY SCHEMA (
		datetime dateFound
		string(36) guid
		string(255) link
		_controlId|new column with a default value|#rndguid()
	)
	BREAK_ON_EMPTY
	;

);