Transformations

How-to guideApplies to: Unified engine deployments

The INSERT INTO ... SELECT ... in a pipeline defines how incoming records become output rows. Use the registered orders_source and paid_orders_sink from Your first pipeline.

Choose one alternative to run. Each example below creates a paused pipeline. They write to the same example destination. Start one at a time when comparing them, so several pipelines do not append copies of the same source records.

Filter and calculate

Keep paid orders with positive amounts. Select only the output columns you need and name the calculated value.

CREATE STREAMING PIPELINE filtered_orders
WITH (start_paused = true)
AS INSERT INTO paid_orders_sink
SELECT
  order_id,
  customer_id,
  amount,
  amount * 1.10 AS amount_with_tax
FROM orders_source
WHERE status = 'paid' AND amount > 0;

For the walkthrough’s four input messages, orders 1 and 3 pass the filter. Their output amounts with tax are 22 and 44. The pipeline remains Stopped until you start it. The sink uses order_id, customer_id, amount, and amount_with_tax as its output column names. Keep these names and types consistent when writing to the same Iceberg table.

Handle missing amounts and apply a rule

COALESCE supplies a value when an amount is NULL. CASE applies tax only when an order is paid. This alternative keeps all records.

CREATE STREAMING PIPELINE priced_orders
WITH (start_paused = true)
AS INSERT INTO paid_orders_sink
SELECT
  order_id,
  customer_id,
  COALESCE(amount, 0.0) AS amount,
  CASE
    WHEN status = 'paid' THEN COALESCE(amount, 0.0) * 1.10
    ELSE COALESCE(amount, 0.0)
  END AS amount_with_tax
FROM orders_source;

For order 2 from the walkthrough, the output keeps amount 30 and sets amount_with_tax to 30 because its status is pending. A record with "amount": null produces zero in both amount columns. Supply NULL as a JSON null value in the source record. Unlike the filter example, this query also keeps negative amounts. Add a WHERE condition when your business rule should exclude them.

Build the query in steps

A common table expression, or CTE, names a step inside one query. Place WITH after the INSERT INTO target. The pipeline’s earlier WITH (...) contains execution options; the query’s WITH name AS (...) contains a named query step.

CREATE STREAMING PIPELINE staged_paid_orders
WITH (
  parallelism = 2,
  checkpoint_interval = '10 seconds',
  start_paused = true
)
AS INSERT INTO paid_orders_sink
WITH paid AS (
  SELECT order_id, customer_id, amount
  FROM orders_source
  WHERE status = 'paid' AND amount > 0
)
SELECT
  order_id,
  customer_id,
  amount,
  amount * 1.10 AS amount_with_tax
FROM paid;

The CTE does not create a stored table or a separate pipeline. This query produces the same rows and columns as the first example.

Start and check one example

Choose the pipeline you want to run. This block starts the filter example.

START PIPELINE filtered_orders;
DESCRIBE PIPELINE filtered_orders;

Check that its state reaches Running. Inspect output through the destination’s Iceberg catalog connection, as shown in the walkthrough. The source’s earliest setting can replay retained records for a new pipeline, so use a fresh destination when comparing exact row counts.

Keep the transformation inside the pipeline

Standalone reads of a Streaming connector name are not Unified catalog queries. The source and sink names are resolved by the Streaming planner inside the pipeline definition. These examples cover row selection, calculations, conditions, missing values, and CTEs. Stateful transformations such as aggregations and joins have additional Streaming rules. Do not assume that every query from the Unified SQL reference can be used unchanged in a pipeline.

Clean up

Stop the pipeline you started, then check its state.

STOP PIPELINE filtered_orders;
DESCRIBE PIPELINE filtered_orders;

Use the name of the example you actually started. Once it is Stopped, remove the example pipeline definitions through the existing pipeline controls. Keep shared connector definitions while any other pipeline still uses them.

Continue

Manage pipelines explains lifecycle commands. Streaming SQL reference lists the accepted syntax and execution options.

Transformation Engine

Learn Laminar

Workloads

SQL reference

HTTP API

Recipes

Operations

Filter and calculate

Handle missing amounts and apply a rule

Build the query in steps

Start and check one example

Keep the transformation inside the pipeline

Clean up

Continue

​Filter and calculate

​Handle missing amounts and apply a rule

​Build the query in steps

​Start and check one example

​Keep the transformation inside the pipeline

​Clean up

​Continue

Filter and calculate

Handle missing amounts and apply a rule

Build the query in steps

Start and check one example

Keep the transformation inside the pipeline

Clean up

Continue