# PureStream pipeline document — operation reference

A document is `{version, source, operations, sink, quarantine?}`. Every
operation names the column it acts on: nothing here infers a column's
meaning from its name, and nothing infers a number or date format from
its values. State both.

Every operation also takes `id` (yours, unique in the document) and
`type` (the heading below). An operation marked *Can reject rows* also
takes `on_error`, and a document containing one needs a `quarantine`
sink or the run is refused.

A source, and each entry of `inputs`, may declare `column_types`: an
object from column name to `string`, `int64`, `float64` or `bool`.
`string` keeps the column as its text on every dataset. The other three
also convert when the input is read and accept one spelling only, the
JSON one (`10`, `"10"` and `10.0` are int64 10; `"007"`, `"+5"` and
`"1,000"` are not). A row that does not load is quarantined before the
first operation, so a typed declaration needs a `quarantine` sink. For any
other notation, declare `string` and use `null_if` and a `parse_*` step.

`cast`, `json_parse` and a typed declaration never change a value
silently: a number with a fraction does not become an integer, and a
number float64 cannot hold does not become infinity or zero. The row
fails instead: `on_error` routes it, and a declaration quarantines it.
`cast` refuses to turn a number or a boolean into a date or datetime, or
a date or datetime into a number: nothing states the unit. A `default`
that `cast`, `parse_number` or `arithmetic` cannot hold in its result
type is refused.

`json_parse` takes each value only as the JSON type its shape declares:
a JSON number for `int64` and `float64`, `true` or `false` for `bool`,
a JSON string for `string`. A quoted number, a number in a `string`
field or `true` in an `int64` field is a row error. Unlike a declaration,
which reads a column of text and so takes `"10"` as int64 10, it sees
the JSON type. `arithmetic` that overflows float64 is a row error, and
`concat_columns` gives null for a row whose every column is null.

A `union` reads each input with that input's own types. Where inputs
disagree on a column and it becomes text, a list or an object reads as
JSON written from its own input's types: `[1, 2.5]`, read on its own as
decimals, reads `[1.0, 2.5]`. A decimal number reads `0.00001` and
`NaN`, unless it sits inside a list or object that is rebuilt as text,
or in an object with such a field, where it reads `1e-05`. For text
identical to one dataset's, declare the column `string` on every input.

Operations run in the order given. A row-reducing operation may not sit
between two that can reject rows — put reducers last.

JSON Schema: `pipeline-document.schema.json`. 43 operations.

## `arithmetic`

Phase `derive`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `operator` | `"add" or "subtract" or "multiply" or "divide"` | required |
| `with_column` | `string or null` | default `null` |
| `with_value` | `integer or number or null` | default `null` |
| `into` | `string` | required |
| `default` | `integer or number or null` | default `null` |

## `cast`

Phase `parse`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `to` | `"int64" or "float64" or "string" or "date" or "datetime"` | required |
| `default` | `string or integer or number or null` | default `null` |

## `coalesce`

Phase `nulls`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[string]` | required |
| `into` | `string` | required |

## `concat_columns`

Phase `derive`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[string]` | required |
| `into` | `string` | required |
| `separator` | `string` | default `""` |
| `ignore_nulls` | `boolean` | default `true` |

## `date_part`

Phase `derive`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `part` | `DatePart` | required |
| `into` | `string` | required |

## `deduplicate`

Phase `reduce`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[string]` | default `[]` |
| `keep` | `"first" or "last"` | default `"first"` |

## `drop`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[string]` | required |

## `drop_nulls`

Phase `reduce`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[string]` | default `[]` |

## `expect`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `predicate` | `Predicate` | required |
| `error_code` | `string` | default `"EXPECTATION_FAILED"` |
| `message` | `string or null` | default `null` |

## `expect_schema`

Phase `validate`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[ExpectedField]` | required |
| `allow_extra` | `boolean` | default `true` |

## `expect_unique`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `columns` | `array[string]` | required |

## `explode`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |

## `extract`

Phase `text`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `pattern` | `string` | required |
| `group` | `integer` | default `1` |
| `into` | `string` | required |

## `extract_path`

Phase `structural`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `path` | `Path` | required |
| `into` | `string` | required |

## `fill_null`

Phase `nulls`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `value` | `string or integer or number or boolean or null` | default `null` |
| `from_column` | `string or null` | default `null` |

## `filter`

Phase `reduce`.

| Field | Type | Requirement |
| --- | --- | --- |
| `where` | `Predicate` | required |

## `flatten`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `separator` | `string` | default `"_"` |

## `format_date`

Phase `derive`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `format` | `string` | required |
| `into` | `string` | required |

## `group_by`

Phase `reduce`.

| Field | Type | Requirement |
| --- | --- | --- |
| `by` | `array[string]` | default `[]` |
| `aggregates` | `array[Aggregate]` | required |

## `join`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `with` | `string` | required |
| `on` | `array[string]` | required |
| `how` | `"inner" or "left" or "semi" or "anti"` | default `"left"` |
| `cardinality` | `"one_to_one" or "many_to_one" or "many_to_many"` | required |
| `key_cast` | `"string" or null` | default `null` |
| `suffix` | `string or null` | default `null` |
| `select` | `array[string] or null` | default `null` |

## `json_parse`

Phase `structural`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `shape` | `FieldSchema` | required |

## `json_stringify`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |

## `limit`

Phase `reduce`.

| Field | Type | Requirement |
| --- | --- | --- |
| `count` | `integer` | required |

## `lowercase`

Phase `text`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |

## `null_if`

Phase `nulls`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `values` | `array[string]` | required |
| `case_sensitive` | `boolean` | default `true` |

## `parse_bool`

Phase `parse`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `true_values` | `array[string]` | default `["true", "t", "yes", "y", "1"]` |
| `false_values` | `array[string]` | default `["false", "f", "no", "n", "0"]` |
| `case_sensitive` | `boolean` | default `false` |

## `parse_date`

Phase `parse`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `to` | `"date" or "datetime"` | default `"date"` |
| `format` | `string` | default `"%Y-%m-%d"` when `to` is `"date"`, `"%Y-%m-%dT%H:%M:%S"` when `to` is `"datetime"` |

## `parse_number`

Phase `parse`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `to` | `"int64" or "float64"` | required |
| `decimal_separator` | `"." or ","` | default `"."` |
| `group_separator` | `string or null` | default `","` |
| `strip_characters` | `string` | default `"$\u20ac\u00a3\u00a5% \t "` |
| `default` | `integer or number or null` | default `null` |

## `rename`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `to` | `string` | required |

## `replace`

Phase `text`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `pattern` | `string` | required |
| `replacement` | `string` | default `""` |
| `literal` | `boolean` | default `false` |
| `replace_all` | `boolean` | default `true` |

## `round`

Phase `derive`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `decimals` | `integer` | default `0` |

## `select`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `columns` | `array[string]` | required |

## `sort`

Phase `reduce`.

| Field | Type | Requirement |
| --- | --- | --- |
| `by` | `array[SortKey]` | required |

## `split_column`

Phase `derive`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `separator` | `string` | required |
| `into` | `array[string]` | required |
| `literal` | `boolean` | default `true` |

## `trim`

Phase `text`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `characters` | `string or null` | default `null` |

## `union`

Phase `structural`.

| Field | Type | Requirement |
| --- | --- | --- |
| `with` | `array[string]` | required |

## `uppercase`

Phase `text`.

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |

## `validate_currency`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `default` | `string or null` | default `null` |

## `validate_email`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `default` | `string or null` | default `null` |

## `validate_phone`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `default` | `string or null` | default `null` |

## `validate_url`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `default` | `string or null` | default `null` |
| `schemes` | `array[string]` | default `["http", "https"]` |

## `validate_uuid`

Phase `validate`. Can reject rows.

| Field | Type | Requirement |
| --- | --- | --- |
| `on_error` | `ErrorPolicy` | default `"reject_row"` |
| `column` | `string` | required |
| `default` | `string or null` | default `null` |

## `window`

Phase `derive`.

| Field | Type | Requirement |
| --- | --- | --- |
| `partition_by` | `array[string]` | default `[]` |
| `order_by` | `array[string]` | default `[]` |
| `descending` | `boolean` | default `false` |
| `nulls_last` | `boolean` | default `true` |
| `function` | `"count" or "n_unique" or "sum" or "mean" or "median" or "std" or "var" or "min" or "max" or "first" or "last" or "list" or "row_number" or "rank" or "dense_rank"` | required |
| `column` | `string or null` | default `null` |
| `into` | `string` | required |

## Shared types

### `Aggregate`

| Field | Type | Requirement |
| --- | --- | --- |
| `function` | `"count" or "n_unique" or "sum" or "mean" or "median" or "std" or "var" or "min" or "max" or "first" or "last" or "list"` | required |
| `column` | `string or null` | default `null` |
| `into` | `string` | required |

### `DatePart`

`"year" or "month" or "day" or "hour" or "minute" or "second" or "weekday" or "ordinal_day" or "week"`

### `ErrorPolicy`

One of: "null" or "keep_original" or "default" or "reject_row" or "fail".

### `ExpectedField`

| Field | Type | Requirement |
| --- | --- | --- |
| `name` | `string` | required |
| `physical` | `PhysicalType or null` | default `null` |
| `logical` | `LogicalType or null` | default `null` |

### `FieldSchema`

| Field | Type | Requirement |
| --- | --- | --- |
| `name` | `string` | required |
| `physical` | `PhysicalType` | required |
| `logical` | `LogicalType or null` | default `null` |
| `nullable` | `boolean` | default `true` |
| `description` | `string or null` | default `null` |
| `inner` | `array[FieldSchema] or null` | default `null` |

### `Path`

| Field | Type | Requirement |
| --- | --- | --- |
| `segments` | `array[Segment]` | required |

### `Predicate`

`Compare or IsNull or Contains or Matches or InSet or AllOf or AnyOf or Not`

### `SortKey`

| Field | Type | Requirement |
| --- | --- | --- |
| `column` | `string` | required |
| `descending` | `boolean` | default `false` |
| `nulls_last` | `boolean` | default `false` |

### `AllOf`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"all_of"` | required |
| `of` | `array[Predicate]` | required |

### `AnyOf`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"any_of"` | required |
| `of` | `array[Predicate]` | required |

### `Compare`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"compare"` | required |
| `column` | `string` | required |
| `op` | `"eq" or "ne" or "gt" or "ge" or "lt" or "le"` | required |
| `value` | `LiteralValue` | required |

### `Contains`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"contains"` | required |
| `column` | `string` | required |
| `substring` | `string` | required |
| `case_sensitive` | `boolean` | default `true` |

### `InSet`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"in_set"` | required |
| `column` | `string` | required |
| `values` | `array[LiteralValue]` | required |

### `IsNull`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"is_null"` | required |
| `column` | `string` | required |
| `negated` | `boolean` | default `false` |

### `LogicalType`

One of: "email" or "url" or "uuid" or "phone" or "currency".

### `Matches`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"matches"` | required |
| `column` | `string` | required |
| `pattern` | `string` | required |

### `Not`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"not"` | required |
| `of` | `Predicate` | required |

### `PhysicalType`

One of: "null" or "bool" or "int64" or "float64" or "string" or "date" or "datetime" or "list" or "struct".

### `Segment`

`Key or Index`

### `Index`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"index"` | required |
| `position` | `integer` | required |

### `Key`

| Field | Type | Requirement |
| --- | --- | --- |
| `kind` | `"key"` | required |
| `name` | `string` | required |

### `LiteralValue`

`string or integer or number or boolean`