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

# Transformations

> 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(first-p

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](https://docs.e6data.com/transformation/streaming/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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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](https://docs.e6data.com/transformation/streaming/first-pipeline#4-verify-the-iceberg-output). 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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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](https://docs.e6data.com/transformation/streaming/manage-pipelines) explains lifecycle commands. [Streaming SQL reference](https://docs.e6data.com/transformation/streaming/sql-reference) lists the accepted syntax and execution options.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.