Building a pipeline

New pipeline asks only for a name (“e.g. Nightly sales sync”) and takes you straight to the editor — the pipeline isn’t created until its flow validates, so drafts never clutter the list.

The editor

Three panes: an Add step palette on the left, the canvas in the middle, and a config drawer for the selected step on the right. Drag between steps to connect them — an edge means runs after, and that’s how you express ordering, fan-in and fan-out. A DAG holds up to 100 steps.

Extract

Pulls rows from a source. The config drawer offers six tabs — Table, SQL, CDC, Files, REST and SaaS — and a step uses exactly one of them.

Table

Pick a Source (any connected data source) and a table (schema.table). The whole table is read, unless you set a cursor — see Incremental below.

SQL

Pick a Source and write a query. Use this to filter, join or pre-aggregate at the source, where the database is already good at it.

CDC

Consume a CDC stream — inserts, updates and deletes captured off a Postgres source by managed Debezium. You name the stream id, not a connection: Debezium already holds the source credentials.

Pair it with a Merge (upsert by key) load to apply the changes.

Files

Read objects out of a store — S3, Azure Blob, ADLS Gen2, GCS, or plain HTTPS.

Connect your storage in the connection wizard under Data Sources (or create it via the REST API).

Field	What it does
Path	a prefix or glob inside the store: raw/orders/, raw/orders/*.csv
Format	auto (by file extension), csv, json or parquet
Compression	auto, none, gzip or zstd
Union by name	on by default — files whose columns differ by name still stack, so schema drift across a folder of daily exports just works
New files only	only objects newer than the last successful run (off by default)
Max files	one run’s intake bound — default 1000, max 10 000

CSV files take the usual extras (delimiter, header, quote, escape, null string, skipped rows, explicit column types, date/timestamp formats).

REST

Pull a paginated JSON HTTP API. It needs a REST API connection — the URL and any auth headers live encrypted on the connection, and the step schema refuses an inline secret rather than letting it ride along in the DAG.

Field	What it does
Records path	where the record array lives, JSONPath-lite: $.data.items
Columns	project each record into named columns by path ($.customer.id)
Pagination	none, offset, page, cursor, or link_header (RFC 5988)
Page size	up to 10 000; max pages defaults to 1000
Rate limit	space requests out, up to 100 per second
Incremental	send the last watermark back as a query param on the next run

REST incremental is its own thing — a query param plus the record field the watermark is read from, with an optional first-run value.

Pagination owns its query params

A param the pagination mode claims (offset, limit, page, your cursor param) is stamped onto every page request. Naming it again as a filter or a watermark param is refused at save, rather than being silently overwritten at 3am.

SaaS

Pull a SaaS app through a curated, sandboxed connector: Salesforce, HubSpot, Stripe, Google Analytics 4, Shopify or Jira. Pick the provider, its connection (credentials stored encrypted), and optionally the resources to pull — leave it empty for the provider’s standard set (e.g. Stripe: charges, customers, invoices, subscriptions).

The connector runs inside a hardened container — pinned image, non-root, capability-dropped, memory/time capped, credentials on a RAM-only mount that vanishes with the container. You never supply a URL: each connector is pinned to its vendor’s own API domain, and the only identifiers you provide (a shop name, a site name, a property id) are strictly character-gated.

Incremental is on by default: each resource keeps a cursor watermark (updated_at, SystemModstamp, created, …) that advances only when the run succeeds — a failed run never skips rows.

Six connectors, honestly counted

This is a curated start, not a 700-connector catalogue. The generic REST tab reaches most other JSON APIs today; more curated connectors land provider by provider.

Schema drift (wire extracts)

Sources change mid-life: a column gets added, dropped, or retyped. On the Table and SQL extracts you can opt into a managed drift policy with schemaDrift:

• allow — a new column flows through; a removed column keeps its place, null-filled, so downstream shapes stay stable instead of breaking the night a source column disappears; a retyped column takes the new type with a loud warning in the step log.
• strict — any drift fails the step, naming the exact columns. The governance mode for shapes that must never move silently. (Strict is checked on the single-stream path, so it can’t be combined with partition.)

The baseline is a per-step schema snapshot that advances only when the part actually lands. Leaving schemaDrift unset keeps the legacy behavior: whatever the source has now is what lands.

{ "sourceId": "8f14e45f-…", "table": "public.orders", "schemaDrift": "allow" }

vs ADF

This is ADF-grade drift without the Hive-partition quirk where drifted columns get dropped on partitioned sinks — here the drifted column is in the Parquet, full stop. Files already tolerate drift via union-by-name; REST, SaaS and CDC carry their own schema semantics.

Object-store connections (API only)

S3, Azure Blob, ADLS Gen2 and Google Cloud Storage connections can’t be created in the connection wizard yet — it doesn’t offer them a form. Create one with the REST API and it appears in every picker that wants a store: the Files extract tab, the Files load target, and the file-arrival trigger.

curl -X POST https://<your-host>/api/sources \
  -H "Authorization: Bearer $DS_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Data lake (S3)",
    "type": "s3",
    "config": {
      "bucket": "my-data-lake",
      "region": "eu-west-1",
      "accessKeyId": "AKIA…",
      "secretAccessKey": "…"
    }
  }'

The type is s3, azure_blob, adls_gen2 or gcs, and the token needs the source:write scope. Each takes its own config:

Type	Config
s3	bucket, accessKeyId, secretAccessKey; region defaults to us-east-1
azure_blob / adls_gen2	account, container, plus either accountKey or sasToken
gcs	bucket and credentials — the service-account key JSON, as a string

A store is a credential, not a queryable source

An object-store connection exists so pipelines can read and write it. It is deliberately not a SQL source — a bucket isn’t a database, so it won’t appear in the model or SQL layers, and asking it a SQL-shaped question answers: read it with a pipeline (Extract → Files). That’s by design, not a missing feature.

What Extract can pull from

Family	Engines	How it reads
Postgres wire	Postgres, Redshift, CockroachDB, TimescaleDB, Neon, Supabase, AlloyDB, YugabyteDB, Azure/RDS/Cloud SQL Postgres, and more (17)	streaming cursor
MySQL wire	MySQL, MariaDB, TiDB, SingleStore, PlanetScale, Azure/RDS/Cloud SQL MySQL, and more (11)	streaming cursor
TDS	SQL Server, Azure SQL, Azure Synapse, Fabric Warehouse	streaming cursor
Bulk unload	Snowflake, BigQuery	the warehouse’s own COPY INTO / EXPORT DATA into your staging bucket
Object stores	S3, Azure Blob, ADLS Gen2, GCS, HTTPS	the Files tab
HTTP	any paginated JSON API	the REST tab

Redshift row-streams over the Postgres wire by default and upgrades to a bulk UNLOAD when the connection has staging configured — see bulk staging.

Incremental extract

Set a cursor column (“Cursor column (empty = full pull)”) and a Start from value, and each run pulls only a bounded window: rows greater than the last watermark and less than or equal to a freshly probed maximum. Two properties matter:

• The cursor advances only when the whole run succeeds — a failure halfway through never skips rows on the next run.
• An idle window is a 0-row success, not an error.

Not available on bulk sources

Incremental extract is refused on Snowflake, BigQuery and staged-Redshift sources — they unload in bulk. Use the Files or CDC shape, or filter in the SQL tab instead.

Partitioned extract

A single cursor is one connection and one thread. Split the key range and pull it in parallel:

{
  "id": "extract_events",
  "type": "extract",
  "config": {
    "sourceId": "8f14e45f-…",
    "table": "public.events",
    "partition": { "column": "id", "count": 8 }
  }
}

count is 2–64 (the worker also caps real concurrency, default 4). Split by an orderable, roughly-uniform column — typically the primary key; a heavily skewed column just produces unbalanced partitions.

API only, Postgres wire only

There is no UI for partitioned extract — set it through the API or the DAG JSON. It works on the Postgres wire only (refused on MySQL and SQL Server), and is refused on bulk-unload warehouses, files and REST — each of those already parallelizes its own way.

Transform (SQL)

Fully enabled. A transform step is either a declarative op chain or raw SQL — exactly one of the two.

Both compile through one shared compiler, so the SQL you preview is byte-for-byte the SQL the worker runs. Every identifier is quoted per dialect and every literal is typed; only a derive expression is free-form, and it’s guarded.

The 21 ops

Up to 50 ops per chain, grouped in the drawer:

Group	Ops
Rows	Filter · Sort · Limit · Deduplicate · Assert (quality gate)
Shape	Select (keep/drop/rename) · Derived column · Cast · Aggregate · Pivot · Unpivot · Surrogate key
Analytics	Window (running totals, lag/lead, first/last per partition) · Rank
Combine	Join · Union · Lookup · Exists
Semi-structured	Flatten · Parse JSON · Stringify

As DAG JSON, a chain is just a list:

[
  { "op": "filter", "where": { "column": "status", "op": "eq", "value": "paid" } },
  { "op": "derive", "column": "net", "expression": "amount - discount" },
  {
    "op": "aggregate",
    "groupBy": ["region"],
    "measures": [{ "fn": "sum", "column": "net", "as": "revenue" }]
  },
  { "op": "sort", "by": [{ "column": "revenue", "direction": "desc" }] }
]

Several of them exist to avoid classic footguns:

• Lookup dedupes the reference side first (firstMatchOnly, on by default), so a duplicated key cannot multiply your rows. That’s what makes a lookup a lookup rather than a join.
• Exists with negate compiles to LEFT JOIN … IS NULL, sidestepping the NOT IN NULL trap that quietly returns nothing.
• Window’s last_value always sees the whole partition — without that, the SQL default frame stops at the current row and last_value just echoes it.
• Pivot takes an explicit value list (values: [{ "value": "Q1" }, …]) — static SQL can’t conjure columns it hasn’t seen, so you name the columns you want, exactly like a warehouse PIVOT … IN (…). A cell with no rows is an honest NULL, and each cell aggregates with the fn you pick (sum by default; count needs no column).
• Surrogate key numbers rows from startAt (default 1). Give it an orderBy to make the numbering deterministic — without one, engine row order decides.

Assert — the quality gate

An assert holds up to 32 named checks: rules (any structured predicate — not-null, ranges, formats) and unique keys (rows whose key appears more than once all fail — keeping one would be Deduplicate’s job). Checks are NULL-strict: a comparison that can’t be proven fails the check; add or is_null to a rule if NULLs are acceptable. One assert per chain.

What happens to failing rows is the onFail policy:

• fail (default) — any failing row fails the step with per-check counts (amount_positive (2), unique_id (1)), before anything is published. The load is blocked.
• quarantine — failing rows leave the chain at the gate and land on a quarantine port on the step (amber handle on the canvas), each row stamped with _ds_check naming the first check it failed. Clean rows flow on as the step’s normal output. Connect the quarantine port to a load (or any step) to inspect or replay the bad rows — a port left unconnected still writes its part, so nothing is ever silently dropped.

{
  "op": "assert",
  "onFail": "quarantine",
  "checks": [
    { "name": "amount_positive", "where": { "column": "amount", "op": "gt", "value": 0 } },
    { "name": "unique_id", "unique": ["id"] }
  ]
}

The preview’s SQL tab shows both statements a quarantining step runs — the pass filter and the quarantine complement.

Raw SQL

Each upstream step is a view named after it, so raw SQL reads like SELECT * FROM extract_orders JOIN extract_customers USING (customer_id). It’s the escape hatch for anything the op set doesn’t express.

Preview

Preview runs the step against the previous run’s data and shows the compiled SQL plus rows (50 by default, 500 max).

Preview needs a prior run

It reads the last run’s Parquet, so the pipeline must be saved and have run at least once. Non-transform steps — including route — refuse preview.

AI authoring

Describe the transform in plain language and get back a validated declarative op chain — never raw SQL, never a black box. You see the ops, you can edit them, and they validate like any hand-built chain (one automatic repair round if the first attempt doesn’t validate). If the AI gateway is unreachable, you get a clean “AI unavailable” error, not a broken step.

Run in parallel across workers

Tick “Run in parallel across workers” and set Partitions (2–64) to fan a transform across the worker fleet. The rules are strict because correctness is:

• Row-wise ops only — filter, select, derive, cast, parse, stringify, flatten. Anything that has to see rows together (aggregate, dedupe, join, union, lookup, exists, sort, limit) is refused by name.
• A windowed derive (an expression containing OVER) is refused — a window reads other rows.
• Raw SQL can never be partitioned — it can’t be proven row-wise.
• The step needs exactly one upstream (it slices that input’s parts).

Large aggregates and joins don’t need this: at scale they’re pushed down into the warehouse instead.

Where a transform runs

You don’t pick. Small inputs run in-process on DuckDB; past a size threshold a declarative chain is pushed down into ClickHouse, reading Parquet straight from the lake. Same compiler, same results.

Flowlets

A chain you’ll want again — “dedupe by key”, “standardize an address” — can be saved as a flowlet: a reusable, parameterized, versioned fragment you instantiate into other pipelines.

Code (Python)

For what SQL can’t express. Your script runs in a hardened sandbox: a digest-pinned image, no network, non-root, all Linux capabilities dropped, with hard memory and wall-clock caps.

The contract is I/O-shaped: Parquet in → Parquet out. Read from DS_IN (one directory per upstream step, named by step id) and write one or more *.parquet files into DS_OUT.

import os, glob
import pandas as pd

src = os.path.join(os.environ['DS_IN'], 'extract_orders', '*.parquet')
df = pd.concat(pd.read_parquet(f) for f in glob.glob(src))

df = df[df['amount'] > 100]

df.to_parquet(os.path.join(os.environ['DS_OUT'], 'big_orders.parquet'), index=False)

duckdb, pyarrow and pandas are pre-installed. There’s no network, so there is no pip install at run time — that’s a security property, not a gap.

Setting	Range
Code	up to 1 MiB (bulk data ships through the lake, not the script)
Timeout	10–3600 s (default 600)
Memory	256–16 384 MB (default 1024)

Nothing the script prints is trusted as a result — the worker validates that the outputs actually parse as Parquet. The step needs the sandbox executor enabled server-side; if it isn’t, it refuses cleanly.

Route (split)

Send rows down named ports by condition. First match wins, and rows matching nothing land on the default port (rest unless you rename it) — so every input row lands on exactly one port. 1–16 routes.

{
  "id": "split_by_region",
  "type": "route",
  "dependsOn": ["extract_orders"],
  "config": {
    "routes": [
      { "port": "emea", "when": { "column": "region", "op": "in", "values": ["DE", "FR", "ES"] } },
      { "port": "amer", "when": { "column": "region", "op": "eq", "value": "US" } }
    ],
    "defaultPort": "rest"
  }
}

A downstream step names the port it consumes:

{
  "id": "load_emea",
  "type": "load",
  "dependsOn": ["split_by_region"],
  "fromPorts": { "split_by_region": "emea" },
  "config": { "datasetName": "orders_emea", "mode": "replace" }
}

A route splits exactly one incoming step, and a consumer that doesn’t name its port is refused at save.

Load

Three target kinds, on tabs — Dataset, Files, Database:

• Dataset — a managed dataset in your ClickHouse. Modes: Replace, Append, or Merge (upsert by key).
• Database — write straight back into your own database or warehouse.
• Files — Parquet, CSV or JSON into an object store, optionally hive-partitioned.

Each is covered properly — with merge semantics, CDC deletes, and multi-sink fan-out — in Where pipelines can write.

Retry

Every step carries its own retry policy: maxAttempts 1–10 (default 3, counting the first attempt), an initial interval, and a backoff coefficient.

"retry": { "maxAttempts": 5, "initialIntervalMs": 2000, "backoffCoefficient": 2 }

Validation

The editor validates continuously: a “✓ flow is valid” chip, or an issue count with a Fix before saving list — unconnected steps, missing config, cycles, a route consumer with no port named.

Create pipeline / Save commits. If someone else edited the pipeline while you had it open, you’ll be asked to reload — last write doesn’t silently win.

Schedules

Schedule sets the cadence:

Preset	Fires
No schedule (manual only)	never — you run it
Hourly	every hour
Daily (02:00)	once a day
Weekly (Mon 02:00)	once a week
Micro-batch (every N seconds)	every N seconds — 15-second floor, max 86 400
Custom (cron)	a full cron expression (e.g. 30 6 * * 1-5)

Cron schedules take a timezone (default UTC). A schedule is either cron or every-N-seconds, never both, plus an Enabled toggle.

Schedules fire. A poller ticks every ~10 seconds and starts every due schedule; each row is claimed atomically, so multiple API instances never double-fire. After downtime it fires once for the latest missed occurrence and re-arms — it doesn’t stampede through every window you were down for.

Triggers

A schedule answers when; a trigger answers because of what. Open Trigger in the editor toolbar and pick a When. A pipeline can have both a schedule and a trigger.

When files arrive in a folder

Watch a folder on S3, Azure Blob, ADLS Gen2 or GCS and run when new objects land. (The storage picker lists your connected object-store sources created under Data Sources.)

Field	Meaning
Storage + folder/pattern	what to watch
Wait for quiet (seconds)	don’t fire while the newest object is younger than this — 0–3600, default 60
Wait for at least	fire only once N new files are waiting (default 1)

Two things make this safe:

• It baselines on first observation, so turning it on never retro-fires the folder’s existing history.
• The quiet period is the point. A multipart upload lands as several objects over several seconds; firing on the first one reads a half-written batch. This is the single most common event-trigger bug, and the default guards against it.

A watcher, not a subscription

The poller lists the folder each tick. There’s nothing to configure in your cloud account — no Event Grid, no bucket notification, no IAM for a push endpoint — and it works the same on every store. The honest trade: latency is the poll interval (seconds), not milliseconds.

After another pipeline finishes

Pick the upstream pipeline and whether to chain only if it succeeded (the default) or whenever it finishes, even on failure. A pipeline can’t follow itself — that’s refused at save.

When an external system calls a URL

A URL you paste into whatever already knows the data is ready: a CI job, a vendor’s “callback URL” box, a curl at the end of someone else’s script.

1. Add the webhook trigger to the pipeline and save.

2. Ask for the URL. It’s a write-grade act — handing over a credential — so this call is authenticated:

   curl -s https://<your-host>/api/pipelines/<pipeline-id>/webhook \
     -H "Authorization: Bearer $DS_TOKEN"

   {
     "enabled": true,
     "url": "https://<your-host>/api/pipelines/<pipeline-id>/webhook?token=9f2c…",
     "token": "9f2c…"
   }

3. Have the external system POST to it — no body needed. You get 201 and the new run row back:

   curl -X POST "https://<your-host>/api/pipelines/<pipeline-id>/webhook?token=9f2c…"

   The token can ride a header instead, if the caller prefers:

   curl -X POST https://<your-host>/api/pipelines/<pipeline-id>/webhook \
     -H "x-datasquares-token: 9f2c…"

The token is the credential

The POST endpoint is unauthenticated by design — the token is the credential, so treat that URL like a password. The token is derived (HMAC), never stored, so there’s nothing to steal from the database. To revoke it, remove the webhook trigger — the URL dies with it.

A wrong token answers identically for a real and a nonexistent pipeline, so the endpoint can’t be used to probe which pipeline ids exist.

When a trigger can’t fire

Triggers use the same ladder as a manual run: the product must be licensed, a concurrency slot must be free, the plan allowance must not be exhausted, and a worker must be live. A trigger can never skip a rung that a schedule respects. If one is failing, the editor toolbar shows “Trigger error” and the dialog says “This trigger is failing.”

A file-arrival trigger pointed at a database connection, a connection from another workspace, or a pipeline following itself is refused at save (422). Changing a trigger resets its cursor.

Related

• Where pipelines can write
• Runs, backfill & monitoring
• Change data capture
• Flowlets
• Connector catalog — what Extract can pull from.