DQL
1.x
Docs/DQL/Operators
Open

Reading22 min
Updated4 Aug 2026
Sourcev1/operators.mdx

DQL pipe operators

Every operator in the pipe catalog, generated from the catalog the engine registers, so it cannot describe an operator that does not ship.

A pipe is an ordered chain of these operators applied to a stream of rows. They appear below in catalog order, which follows the shape of a typical pipe: source, filter, transform, aggregate, sort, side effect.

Column key01

ColumnMeaning
Live-safePure and deterministic with default config, so it can run on a live-updating result set.
PushableCan fold into the SQL prefix. The planner makes the final call from the field types.
RequiresHost services the operator needs in its OpContext. Blank means it works anywhere.

Index02

OperatorSummaryLive-safePushableRequires
filterKeep rows that match a predicateeval
projectSelect / alias columns
renameRename columns
dropRemove columns
computeAdd a computed columneval, formula
transformCompute multiple columns in one stageeval
castConvert column types
dropNullsDrop rows with nulls in named columns
fillNullsReplace nulls in named columns
flattenUnnest an array column
unnestObjectSpread an object column into root keys
nestGroup rows into nested arrays
pivotRows → columns
unpivotColumns → rows
distinctDeduplicate rows
dedupeKeep one row per identity key
sampleRandom subset of rows
assertFail the query when an expression is falseeval
timeBucketBucket rows into fixed time intervals
gapfillFill missing time intervals
sortOrder rows
limitCap row count
skipDrop the first N rows
topPerGroupTop N rows per partition
groupBySet keys for the next aggregate
aggregateFold rows into groups
histogramBucket numeric values into equal-width bins
windowPer-row computation within a partition
lookupHash-join with another datasetclassic
asofJoinJoin on the closest timestampclassic
crossJoinCartesian product with another datasetclassic
intersectRows present in every sourceclassic
exceptRows in left but not in rightclassic
callFunctionInvoke a registered DTL functionfunctionRegistry
callAppInvoke a managed appappCaller
algoRun a registered algorithmalgorithms
branchPer-row conditional sub-pipeseval
mergeFan-out then concat
tapPass-through with a debug label

filter03

Keep rows that match a predicate

Live-safe by default: yes    Pushable: yes    Requires: eval

Filters rows by a WHERE clause. Plain field/op/value forms push down to SQL; DTL expr forms run in-memory.

FieldTypeRequiredDescription
whereanyyesConditions a row must satisfy to be kept. Use simple {field, op, value} shape for SQL-pushable filters, or {expr: "…"} for DTL expressions evaluated in-memory.

Plain comparison

{
  "op": "filter",
  "where": {
    "field": "level",
    "op": "==",
    "value": "ERROR"
  }
}

DTL expression

{
  "op": "filter",
  "where": {
    "expr": "ts > now() - duration(\"1h\")"
  }
}

project04

Select / alias columns

Live-safe by default: yes    Pushable: yes

Reduces each row to a chosen subset of columns, optionally renaming them.

FieldTypeRequiredDescription
selectarrayColumns to keep, in order. Each entry is {field, as?} — set as to rename.
droparray of stringAlternative to select — instead of listing what to keep, list what to remove. Mutually exclusive with select.

Pick fields

{
  "op": "project",
  "select": [
    {
      "field": "id"
    },
    {
      "field": "name"
    }
  ]
}

rename05

Rename columns

Live-safe by default: yes    Pushable: yes

Rewrites column keys in-place. The map is from -> to.

FieldTypeRequiredDescription
mapobjectyesObject whose keys are the existing column names and whose values are the new names. Pairs not listed are left untouched. (input: column-rename-map)

id → key

{
  "map": {
    "id": "key"
  },
  "op": "rename"
}

drop06

Remove columns

Live-safe by default: yes    Pushable: no

Deletes the named columns from each row.

FieldTypeRequiredDescription
columnsarray of stringyesNames of columns to remove from every row.

Drop secrets

{
  "columns": [
    "password",
    "secret"
  ],
  "op": "drop"
}

compute07

Add a computed column

Live-safe by default: yes    Pushable: no    Requires: eval, formula

Evaluates a DTL expression (kind:"expr") or Excel-style formula (kind:"formula") per row and stores the result under as.

FieldTypeRequiredDescription
asstringyesColumn to write the computed value into. Overwrites any existing column with the same name. (input: column-output)
kindstringWhich language expr / formula is in. Defaults to DTL expr. Default: expr.
exprstringconditional {"kind":["expr",""]}DTL expression evaluated against each row. Used when kind is expr. (input: dtl-expression)
formulastringconditional {"kind":["formula"]}Spreadsheet formula evaluated against each row. Used when kind is formula. (input: formula)

DTL expr

{
  "as": "doubled",
  "expr": "value * 2",
  "op": "compute"
}

Formula

{
  "as": "tax",
  "formula": "price * 0.1",
  "kind": "formula",
  "op": "compute"
}

transform08

Compute multiple columns in one stage

Live-safe by default: yes    Pushable: no    Requires: eval

Like compute but takes a list. Entries evaluate in order and can reference earlier results. Use from for plain column copies and replace: true to drop everything not produced.

FieldTypeRequiredDescription
computearray of objectyesOrdered list of column definitions. Each entry uses expr (DTL) or from (column copy).
droparray of stringOptional. Columns to remove from the output after computation.
replacebooleanWhen true, drop every column not produced by compute. Useful for reshape-and-rename in one step. Default: false.

Add several columns

{
  "compute": [
    {
      "as": "total",
      "expr": "a + b"
    },
    {
      "as": "ratio",
      "expr": "a / b"
    }
  ],
  "op": "transform"
}

Reshape

{
  "compute": [
    {
      "as": "id",
      "from": "_id"
    },
    {
      "as": "score",
      "expr": "wins / matches"
    }
  ],
  "op": "transform",
  "replace": true
}

cast09

Convert column types

Live-safe by default: yes    Pushable: no

Coerce columns to int / float / bool / string / timestamp. onError chooses null (default) | skip | fail.

FieldTypeRequiredDescription
castsarray of objectyesList of {field, to, onError?} rules applied in order.

dropNulls10

Drop rows with nulls in named columns

Live-safe by default: yes    Pushable: yes

When any is true (default) drops rows where ANY listed column is null; when false drops only when EVERY listed column is null.

FieldTypeRequiredDescription
columnsarray of stringColumns inspected for null. Empty means every column on each row.
anybooleanWhen true, drop a row if ANY listed column is null. When false, drop only when EVERY listed column is null. Default: true.

fillNulls11

Replace nulls in named columns

Live-safe by default: yes    Pushable: no

Methods: zero, value, lastValue, nextValue, mean. lastValue/nextValue require partitionBy + orderBy for deterministic forward/backward fill.

FieldTypeRequiredDescription
methodstringvalue uses the literal value; zero uses 0; lastValue carries the previous non-null forward; nextValue carries the next non-null backward; mean uses the column mean. Default: value.
valueanyconditional {"method":["value"]}Used when method is value. Any JSON scalar.
columnsarray of stringColumns whose null cells are replaced. Empty means every column on each row.
partitionByarray of stringconditional {"method":["lastValue","nextValue"]}Required for lastValue / nextValue. Within each partition, ordering and carry semantics apply independently.
orderByarrayconditional {"method":["lastValue","nextValue"]}Required for lastValue / nextValue to define what "previous" / "next" means within a partition.

flatten12

Unnest an array column

Live-safe by default: yes    Pushable: no

Emits one row per array element. When as is set, the element is stored under that key and the original field is preserved.

FieldTypeRequiredDescription
fieldstringyesColumn whose array values are exploded into separate rows. (input: column)
asstringOptional. Column name to store each element under. When unset, the element replaces the array column on each row. (input: column-output)
indexAsstringOptional. Column name to store the element's 0-based index in the source array. (input: column-output)
preserveEmptybooleanWhen true, rows whose array is empty or null still produce one output row (with the element column nil). Default drops them. Default: false.

Unnest tags

{
  "field": "tags",
  "op": "flatten"
}

unnestObject13

Spread an object column into root keys

Live-safe by default: yes    Pushable: no

Useful for JSONB/metadata blobs whose nested keys you want as top-level columns.

FieldTypeRequiredDescription
fieldstringyesColumn whose object value's keys are spread onto the root row. (input: column)
prefixstringOptional. Prepended to every spread key (e.g. m_ produces m_owner). (input: prefix)
dropbooleanWhen true, the original object column is removed after spreading. Default: false.

nest14

Group rows into nested arrays

Live-safe by default: yes    Pushable: no

Inverse of flatten: emits one row per group with the remaining columns collected into an array under into.

FieldTypeRequiredDescription
byarray of stringyesPartition keys; one output row is emitted per unique combination.
intostringyesName of the output column where the per-group array of nested records is stored. (input: column-output)
includearray of stringOptional. Columns to include in each nested record. Defaults to every column not in by.

pivot15

Rows → columns

Live-safe by default: yes    Pushable: no

Each unique value in columnKey becomes a column. aggregate controls how to combine duplicates (sum/avg/count/min/max/first/last).

FieldTypeRequiredDescription
rowKeysarray of stringColumns whose distinct combinations identify each output row.
columnKeystringyesColumn whose distinct values become the names of new columns. (input: column)
valueFieldstringyesColumn whose values populate the cells. (input: column)
aggregatestringHow to combine multiple input rows that map to the same cell. Default: first.
fillValueanyValue to use for output cells with no source row. Defaults to null.
prefixstringOptional string prepended to every generated column name. (input: prefix)

unpivot16

Columns → rows

Live-safe by default: yes    Pushable: no

Melts named columns into name/value pairs (one output row per input row × valueCol).

FieldTypeRequiredDescription
idColsarray of stringColumns kept verbatim on every output row. Empty means every column not in valueCols.
valueColsarray of stringColumns whose values are unstacked into rows. Empty means every column not in idCols.
nameAsstringyesOutput column that records the source column name. (input: column-output)
valueAsstringyesOutput column that records the cell value. (input: column-output)

distinct17

Deduplicate rows

Live-safe by default: yes    Pushable: no

When by is empty, uniqueness uses every column; otherwise only the named keys.

FieldTypeRequiredDescription
byarray of stringColumns whose combined values define a row's identity. Leave empty to deduplicate using every column.

dedupe18

Keep one row per identity key

Live-safe by default: yes    Pushable: no

Different from distinct (full-row equality): pick first or last row per key, with optional ordering.

FieldTypeRequiredDescription
byarray of stringyesColumns whose combined values define a row's identity.
keepstringWhether to keep the first row encountered per identity (first) or the last (last). Combined with orderBy for deterministic results. Default: first.
orderByarrayOptional. Sort applied within each identity group before applying keep.

sample19

Random subset of rows

Live-safe by default: yes    Pushable: no

Either n (target size) or ratio (0..1). seed makes output deterministic. Method: random (reservoir) | systematic (every k-th).

FieldTypeRequiredDescription
methodstringrandom uses reservoir sampling; systematic keeps every k-th row. Default: random.
nintegerSample size in rows. Mutually exclusive with ratio.
rationumberFraction of input rows to keep, between 0 and 1. Mutually exclusive with n.
seedintegerOptional integer seed for deterministic sampling. Same seed + same input ⇒ same output.

assert20

Fail the query when an expression is false

Live-safe by default: yes    Pushable: no    Requires: eval

Runtime guardrail. scope=row evaluates per row; scope=overall checks once with count available.

FieldTypeRequiredDescription
scopestringrow evaluates the expression per row; overall evaluates once at the end with the row count available. Default: row.
exprstringyesDTL expression that must be truthy. Falsy results fail the query. (input: dtl-expression)
messagestringCustom error message returned when the assertion fails.

timeBucket21

Bucket rows into fixed time intervals

Live-safe by default: yes    Pushable: no

Adds a column with the start of the bucket containing each row's timestamp. Pair with groupBy + aggregate for time-series rollups.

FieldTypeRequiredDescription
fieldstringyesColumn carrying the row's timestamp. (input: column)
intervalstringyesDuration string for the bucket width — "5m", "1h", "1d". (input: duration)
asstringyesColumn to write the bucket-start timestamp into. (input: column-output)
tzstringOptional IANA timezone (e.g. America/New_York) used to align bucket boundaries. Defaults to UTC. (input: timezone)
originstringOptional RFC3339 anchor that defines where buckets start. Defaults to the Unix epoch. (input: timestamp)

5-minute buckets

{
  "as": "bucket",
  "field": "ts",
  "interval": "5m",
  "op": "timeBucket"
}

gapfill22

Fill missing time intervals

Live-safe by default: yes    Pushable: no

Emits synthetic rows for time buckets missing from the input. Useful before charting sparse sensor streams.

FieldTypeRequiredDescription
fieldstringyesColumn carrying the row's timestamp. (input: column)
intervalstringyesDuration string ("5m", "1h") describing the gap between expected timestamps. (input: duration)
methodstringHow to populate value columns on synthetic rows: zero, null, lastValue (carry forward), or value (use the literal value field). Default: null.
valueanyconditional {"method":["value"]}Used when method is value. Any JSON scalar.
fromstringOptional RFC3339 lower bound. Defaults to the earliest timestamp in the input. (input: timestamp)
tostringOptional RFC3339 upper bound. Defaults to the latest timestamp in the input. (input: timestamp)
groupByarray of stringOptional. Treat the input as multiple independent time series, gapfilling each partition separately.
carryarray of stringconditional {"method":["lastValue"]}Columns whose value is copied from the most recent real row when method is lastValue.

sort23

Order rows

Live-safe by default: yes    Pushable: yes

Sorts by one or more keys. Direction defaults to asc.

FieldTypeRequiredDescription
byarrayyesOrdered list of {field, dir} keys. Earlier entries are the primary sort; later entries break ties.

Top 10 by ts

{
  "by": [
    {
      "dir": "desc",
      "field": "ts"
    }
  ],
  "op": "sort"
}

limit24

Cap row count

Live-safe by default: yes    Pushable: yes

FieldTypeRequiredDescription
nintegeryesKeep at most this many rows. Set to 0 to drop everything.

skip25

Drop the first N rows

Live-safe by default: yes    Pushable: yes

FieldTypeRequiredDescription
nintegeryesNumber of rows to discard from the start of the input. Combined with limit gives offset/page semantics.

topPerGroup26

Top N rows per partition

Live-safe by default: yes    Pushable: no

Keeps the highest- (or lowest-) ranked N rows per group. With no partition, equivalent to top N overall.

FieldTypeRequiredDescription
nintegeryesHow many rows to keep per partition (1 = top-1).
partitionByarray of stringOptional. Columns that split the input into partitions. Empty = single partition (top-N overall).
byarrayyesSort keys used to rank rows within each partition. The first key is primary; later keys break ties.

groupBy27

Set keys for the next aggregate

Live-safe by default: yes    Pushable: yes

On its own this is a pass-through. The planner pairs it with the immediately-following aggregate op.

FieldTypeRequiredDescription
keysarray of stringyesColumns whose unique combinations define each group. The downstream aggregate folds rows within each combination.

aggregate28

Fold rows into groups

Live-safe by default: yes    Pushable: yes

Computes COUNT, SUM, AVG, MIN, MAX over groups defined by a preceding groupBy (or the whole stream when standalone).

FieldTypeRequiredDescription
keysarray of stringOptional. Same shape as groupBy.keys — when set, the aggregate uses these keys directly without a preceding groupBy stage.
aggsarrayyesOrdered list of {fn, field, as} clauses (e.g. {fn: "SUM", field: "amount", as: "total"}).

histogram29

Bucket numeric values into equal-width bins

Live-safe by default: yes    Pushable: no

Replaces input rows with one row per bin: {binStart, binEnd, count}. Provide explicit min/max or let the op derive them.

FieldTypeRequiredDescription
fieldstringyesColumn whose numeric values are bucketed. (input: column)
binsintegeryesNumber of equal-width bins to produce.
minnumberOptional lower bound. Defaults to the minimum value seen in the input.
maxnumberOptional upper bound. Defaults to the maximum value seen in the input.
asCountstringOutput column for the per-bin count. (input: column-output) Default: count.
asStartstringOutput column for each bin's lower edge. (input: column-output) Default: binStart.
asEndstringOutput column for each bin's upper edge. (input: column-output) Default: binEnd.

window30

Per-row computation within a partition

Live-safe by default: yes    Pushable: no

row_number, rank, dense_rank, lag, lead, first_value, last_value. Output preserves input row order.

FieldTypeRequiredDescription
fnstringyesWhich windowing function to compute per row.
partitionByarray of stringColumns that split the input into independent windows. Within each partition, the function is computed in orderBy order.
orderByarraySort keys applied within each partition before the function runs. Required for ranking/lag/lead semantics.
fieldstringconditional {"fn":["lag","lead","first_value","last_value"]}Column the window function reads. Required for lag/lead/first_value/last_value; ignored for ranking functions. (input: column)
offsetintegerRow offset for lag/lead. Defaults to 1 row.
defaultanyValue to emit when the window function falls outside the partition (e.g. lag(1) on the first row).
asstringyesColumn to write the window value into. (input: column-output)

lookup31

Hash-join with another dataset

Live-safe by default: yes    Pushable: no    Requires: classic

Fetches the right side via a secondary classic query. Use cacheTtlMs to avoid re-fetching on every live update.

FieldTypeRequiredDescription
datasetstringyesDataset name to fetch the join's right side from. (input: dataset)
onobjectyesColumn names on each side that must equal for a match.
modestringleft keeps every left row; inner drops left rows with no match. Default: left.
asstringOptional. When set, the matched right-side row is nested under this column instead of being flattened into the left row. (input: column-output)
selectarray of stringOptional. Subset of right-side columns to merge in. Empty means all columns.
whereanyOptional WHERE clause applied to the right-side dataset before the join.
limitintegerOptional. Cap the number of right-side rows fetched (useful when dataset is large).
cacheTtlMsintegerCache the right-side fetch for this many milliseconds. 0 disables caching. Useful when the same lookup runs many times under live mode. Default: 0.

asofJoin32

Join on the closest timestamp

Live-safe by default: yes    Pushable: no    Requires: classic

For each left row, finds the right row with the closest timestamp on a matching key. Direction backward (default) | forward | nearest, with optional tolerance.

FieldTypeRequiredDescription
datasetstringyesDataset to fetch the right side from. (input: dataset)
leftTimestringyesColumn on the streaming (left) side carrying the row's timestamp. (input: column)
rightTimestringyesColumn on the looked-up (right) side carrying the row's timestamp. (input: dataset-column)
leftKeystringOptional. Column on the left whose value must equal rightKey for a match. (input: column)
rightKeystringOptional. Column on the right whose value must equal leftKey for a match. (input: dataset-column)
directionstringbackward finds the latest right row at or before the left row; forward the earliest at or after; nearest the absolute closest. Default: backward.
tolerancestringOptional duration ("1m", "30s"). Drops matches further than this from the left row's timestamp. (input: duration)
asstringOptional. When set, the matched right row is nested under this column. (input: column-output)
selectarray of stringOptional subset of right-side columns to merge in. Empty means all columns.
whereanyOptional WHERE clause applied to the right-side dataset before the join.
limitintegerOptional cap on right-side rows fetched.

crossJoin33

Cartesian product with another dataset

Live-safe by default: yes    Pushable: no    Requires: classic

Output size is N × M — use sparingly. Right-side rows can be filtered with where and capped with limit.

FieldTypeRequiredDescription
datasetstringyesDataset to fetch the cartesian-product right side from. (input: dataset)
asstringOptional. When set, each right row is nested under this column instead of being flattened into the left row. (input: column-output)
selectarray of stringOptional subset of right-side columns to merge in. Empty means all columns.
whereanyOptional WHERE clause applied to the right-side dataset before the cross-join.
limitintegerOptional cap on right-side rows. Strongly recommended to bound output size.

intersect34

Rows present in every source

Live-safe by default: yes    Pushable: no    Requires: classic

Set intersection across N sub-pipes, identified by by keys (or full-row equality).

FieldTypeRequiredDescription
sourcesarray of objectyesTwo or more sub-pipes. Output rows are those present in every sub-pipe (under the chosen identity).
byarray of stringOptional. When set, intersection compares only these columns. Empty falls back to full-row equality.

except35

Rows in left but not in right

Live-safe by default: yes    Pushable: no    Requires: classic

FieldTypeRequiredDescription
leftobjectyesSource whose rows are kept when not present in right.
rightobjectyesSource whose rows are subtracted from left.
byarray of stringOptional. When set, set difference compares only these columns. Empty falls back to full-row equality.

callFunction36

Invoke a registered DTL function

Live-safe by default: no    Pushable: no    Requires: functionRegistry

Calls a function from the function extension. pure: true declares the function side-effect-free and makes it live-safe.

FieldTypeRequiredDescription
namestringyesFully qualified DTL function name (e.g. geo::lookup). (input: function-name)
modestringperRow calls the function once per row; batch passes the whole row stream to the function once. Default: perRow.
purebooleanWhen true, asserts the function has no side effects so live mode can re-run it on each update. Default: false.
argsobjectObject passed to the function. Values may reference row columns via $columnName.
literalArgsbooleanWhen true, $column-style refs in args are passed verbatim instead of resolved against the row.
asstringColumn to store the function result under. (input: column-output)

Per-row pure

{
  "args": {
    "x": "$value"
  },
  "name": "math::abs",
  "op": "callFunction",
  "pure": true
}

callApp37

Invoke a managed app

Live-safe by default: no    Pushable: no    Requires: appCaller

Calls an external app via the runtime extension. Always non-live-safe; live subscriptions require dryRun: true.

FieldTypeRequiredDescription
appIdstringyesIdentifier of the managed app to invoke. (input: app-id)
methodstringApp method to call. Defaults to transform. Default: transform.
capabilitystringCapability namespace the app must expose. Defaults to pipe_query. Default: pipe_query.
batchbooleanWhen true, the entire row stream is sent to the app once. When false, the app is invoked per row. Default: true.
payloadobjectObject merged into the app invocation payload alongside the rows.
datasetstringOptional dataset name passed to the app for context. (input: dataset)

algo38

Run a registered algorithm

Live-safe by default: no    Pushable: no    Requires: algorithms

Invokes a native or external algorithm from the shared catalog (e.g. minmax_scale, kmeans, robust_zscore). params are forwarded verbatim to the algorithm — see its descriptor for accepted keys. Live-safety depends on the chosen algorithm: pure ones (most ETL transforms) are live-safe; external/stateful ones are not.

FieldTypeRequiredDescription
namestringyesIdentifier of a registered algorithm (e.g. minmax_scale, kmeans). Must exist in the algorithm catalog or the stage fails at build time.
paramsobjectAlgorithm-specific options passed through unchanged. Each algorithm documents its own keys in its catalog descriptor (e.g. minmax_scale takes column and as).

Min-max scale a column

{
  "name": "minmax_scale",
  "op": "algo",
  "params": {
    "as": "v_scaled",
    "column": "v"
  }
}

branch39

Per-row conditional sub-pipes

Live-safe by default: yes    Pushable: no    Requires: eval

Routes each row through then or else based on a DTL predicate. Live-safety propagates from the children.

FieldTypeRequiredDescription
whenstringyesDTL expression evaluated per row. Truthy results take the then branch. (input: dtl-expression)
thenarrayyesStages applied to rows where the predicate is truthy.
elsearrayOptional. Stages applied to rows where the predicate is falsy. Empty drops the row.

merge40

Fan-out then concat

Live-safe by default: yes    Pushable: no

Runs each sub-pipe against a clone of the input and concatenates outputs.

FieldTypeRequiredDescription
sourcesarray of objectyesList of sub-pipes. Each receives a clone of the input rows; outputs are concatenated in order.

tap41

Pass-through with a debug label

Live-safe by default: yes    Pushable: no

Records the row count against a label and returns rows unchanged.

FieldTypeRequiredDescription
labelstringLabel that surfaces in stats.pipe[].label so this checkpoint is identifiable in the response.