Ingestion — EL pipes

Pipes are how data lands in ObeliskDB: declarative extract-and-load jobs that pull external sources into governed tables an agent can query.

Obelisk's connector library loads external data extract-and-load style, incremental or full. DuckDB is the extraction kernel (files, blob storage, databases); the REST source is declarative. All state — cursor watermarks, per-file load history, run history — lives in the catalog, so every run is auditable and every table an agent reads has a known provenance.

A pipe

A pipe = source → target table + write mode:

Mode	Behavior
`append`	add rows (new micro-partitions)
`merge`	upsert on `primary_key` via a staging table and a real MERGE
`full_refresh`	replace the table's contents with this run's data

Three ways to create one:

- UI — Ingestion tab → New pipe (templates per source type)
- Python — session.ingest.create_pipe(name, source={…}, target="DB.SCHEMA.T", mode="merge", primary_key=["ID"])
- CLI — obelisk ingest create MY_PIPE --file pipe.json

Run with the UI's ▶ button, obelisk ingest run MY_PIPE, session.ingest.run(...), or from a task: EXECUTE PIPE MY_PIPE.

Sources

file — local files, file shares, or a remote URL

{ "type": "file", "path": "landing/*.csv" }

Formats: csv/tsv, parquet, json/jsonl (+ .gz), inferred from the extension or forced with "format". Incremental is per-file: a file is loaded once per (name, mtime, size) within a 64-day window — the same dedup contract as the Snowflake COPY INTO it mirrors. Remote URLs (https://…, s3://…) load each run.

database — via DuckDB ATTACH

{ "type": "database", "db_type": "postgres",
  "connection": "host=localhost dbname=app user=${PGUSER} password=${PGPASSWORD}",
  "table": "public.orders", "cursor_column": "updated_at" }

db_type: postgres, mysql, sqlite, duckdb. With cursor_column, extraction is incremental: WHERE cursor > <last watermark>; the new watermark is the max value seen. Use query instead of table for custom SQL.

rest — declarative APIs

{ "type": "rest",
  "url": "https://api.example.com/v1/items",
  "headers": { "Authorization": "Bearer ${API_TOKEN}" },
  "records_path": "data.items",
  "pagination": { "type": "offset", "offset_param": "offset",
                  "limit_param": "limit", "page_size": 100 },
  "incremental": { "cursor_field": "updated_at", "param": "updated_since",
                   "initial": "1970-01-01T00:00:00Z" } }

- ${ENV_VAR} expands from the environment (secrets stay out of the catalog).
- records_path is a dot path to the record list ("" = the response root).
- Pagination: none, offset, or cursor (next_path + cursor_param).
- Nested objects/arrays arrive JSON-serialized into VARCHAR columns.

custom — any Python callable

{ "type": "custom", "callable": "my_module:my_source" }

The callable receives the state dict (mutate it to persist a watermark) and returns an iterable of dict records or Arrow tables.

Schema handling

The target table is created from the first load's Arrow schema if missing. On later loads, source columns missing from the target load as NULL; extra source columns raise a loud schema-drift error (use full_refresh or add the column). Drift fails loud rather than silently reshaping a table under an agent's queries.

Operations

obelisk ingest list            # pipes + last run
obelisk ingest run NAME
obelisk ingest runs NAME       # run history with watermarks
obelisk ingest drop NAME

Schedule a pipe with a task: CREATE TASK load_x SCHEDULE = '5 MINUTES' AS EXECUTE PIPE X; then ALTER TASK load_x RESUME — the UI server's scheduler runs it. See Streams & tasks for the scheduling model.

Loading a file from the UI

Sidebar → ⬆ load: pick a local file (csv/tsv/parquet/json/jsonl, .gz ok), name the target table (created from the file's schema if missing), choose append / merge / full refresh. The upload flows through the same loader as pipes — same type inference, same COW commit.

Streaming ingest (real-time)

Push rows continuously; Obelisk buffers per target and commits a micro-batch when either threshold trips (5000 rows or 2 seconds):

POST /api/stream/DEMO.SALES.EVENTS
{"rows": [{"EVENT_ID": 1, "KIND": "click"}, {"EVENT_ID": 2, "KIND": "view"}]}

session.ingest.streamer.insert("DEMO.SALES.EVENTS", rows)
session.ingest.streamer.flush()          # force-commit

Each commit is a normal versioned append — streams see it instantly and WHEN-gated tasks fire on the next scheduler tick: webhook → stream → task → transform, end to end in seconds. Point your webhooks at it, and a standing question an agent registered keeps answering itself as data arrives.