Ingestion & SmartIngest#

Relata supports multiple ingestion paths. All writes are governed (ACL, provenance, WAL-durable) and bi-temporal (every row carries valid_from/to + system_from/to), regardless of which door the data enters through. Every path funnels into the same governed_upsert_* write pipeline, so SmartIngest identity detection, registered ingest pipelines, and audit logging apply uniformly.

CSV / JSON ingest#

# CLI
relata ingest data.csv --type Person
 
# HTTP — body format is auto-detected by first byte (no Content-Type change needed):
#   CSV (default) · NDJSON (leading '{') · JSON array (leading '[')
curl -X POST http://localhost:9090/ingest \
  -H "Authorization: Bearer $RELATA_BEARER_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"object_type":"Person","data":[{"name":"Ada","email":"ada@example.com"}]}'

The purpose parameter on /ingest is validated against ^[a-zA-Z_][a-zA-Z0-9_]{0,63}$ — use underscores, not hyphens (e.g. threat_intel, not threat-intel); max 64 chars.

Document ingest (auto-chunking + embedding)#

curl -X POST http://localhost:9090/ingest/document \
  -H "Authorization: Bearer $RELATA_BEARER_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"source":"report.pdf","content":"...base64...","auto_chunk":true}'

Documents are automatically chunked, embedded, and indexed for vector + BM25 search. Since v1.1 the ingest hot path no longer embeds text rows — embeddings are either caller-supplied (_emb_text in the row payload) or populated asynchronously by the embedder sidecar via the media-worker drain. See LLM & Embedding configuration.

Media ingest#

Images, audio, and video are processed through modality-specific embedders and stored as governed blobs. Use POST /ingest/media?modality=image|audio|video (base64 body) or include _emb_* vectors inline. Perceptual-hash dedup is applied automatically. See LLM & Embedding configuration for the sidecar contract.

Kafka adapter#

A pure-Rust Kafka consumer (KafkaIngestAdapter in relata-storage::kafka) — no rdkafka / librdkafka C FFI, so no unsafe on the poll path (repository invariants forbid unsafe entirely). It speaks the Kafka binary wire protocol directly for three API keys: Metadata (partition-leader discovery), Fetch (record batches), and OffsetCommit (consumer-group offset durability → at-least-once).

# Configure via env (the adapter runs in the server process)
RELATA_KAFKA_BOOTSTRAP=kafka:9092 \
RELATA_KAFKA_TOPIC=events \
RELATA_KAFKA_GROUP=relata-ingest \
RELATA_KAFKA_PARTITION=0 \
relata serve

Each consumed record flows through the same governed_upsert_many batch path as /ingest/bulk, so registered ingest pipelines run on every Kafka record. The legacy in-memory StreamingAdapter is retained for tests; KafkaIngestAdapter is the production path.

Ingest pipelines#

Pipelines define pre-write field transforms that run before a row is committed — keeping raw-to-structured normalization inside Relata's governance boundary rather than an external pre-processor. Every write path — single-row governed_upsert_durable, batch governed_upsert_many (the path behind /ingest/bulk, /ingest/cdr, /ingest/logs, /ingest/metrics, and the Kafka drain) — runs matching pipelines before validation/persist.

Five built-in processors: dissect, grok, date, fingerprint, community_id.

POST /pipelines
Authorization: Bearer <token>
Content-Type: application/json
 
{
  "name": "firewall-flow",
  "target_type": "NetworkFlow",
  "on_failure": "keep",
  "processors": [
    {"kind": "dissect", "field": "raw", "pattern": "%{src_ip}:%{src_port} -> %{dst_ip}:%{dst_port}"},
    {"kind": "fingerprint", "fields": ["src_ip", "dst_ip"]},
    {"kind": "community_id"}
  ]
}

A pipeline can only add fields the caller didn't already send — it never overwrites. on_failure is keep (default: log + continue) or drop (abort + drop the row). List with GET /pipelines.

Pipelines are in-memory only — lost on restart (persist-on-shutdown is a tracked follow-up). DEFINE PIPELINE SQL does not exist; POST /pipelines is the only registration path.

CDR fast path#

A typed fast path for telco call-detail records — the highest-volume telco workload. CdrRecord is a flat struct (~80 bytes/row vs ~200 bytes for the generic HashMap row) with MSISDNs as u64 and timestamps as i64 ns, behind a non-blocking IngestQueue.

curl -X POST "http://localhost:9090/ingest/cdr?purpose=investigation" \
  -H "Authorization: Bearer $RELATA_BEARER_TOKEN" \
  -H "Content-Type: text/csv" \
  --data-binary @cdrs.csv
 
# or via the CLI helper
relata cdr ingest calls.csv [--purpose law_enforcement]
relata cdr analyze +919876543210      # common-contact hand-off analysis
relata cdr timeline +919876543210     # most-recent-calls timeline

purpose is required. Column names are case-insensitive and accept aliases (caller/a_number/a_msisdn → caller_msisdn; start/call_start/timestamp → call_start_ns; etc.). MSISDNs are normalised (non-digit chars stripped; >15 digits = parse error). A timestamp < 10^13 is treated as epoch seconds and scaled to ns; >= 10^13 as ns already.

relata import --from (migration connectors)#

Migrate an existing database straight into the governed identity fabric — no CSV export step. Postgres is a real, wired connector; neo4j/mongo/clickhouse are honest stubs (each prints the CSV/NDJSON workaround and exits non-zero).

relata import --from postgres \
  --dsn "postgresql://user:pass@localhost:5432/appdb" \
  --table users --type Person \
  --batch 500 --on-conflict overwrite \
  --token "$RELATA_BEARER_TOKEN" --url "http://localhost:9090"

The connector opens a read-only transaction and a server-side cursor, FETCHes --batch-sized pages (streaming, bounded memory), and POSTs each page as one /ingest/bulk request. Type-faithful JSON: numeric/decimal kept as exact text (no precision loss on money columns). The source PK maps to _pk so --on-conflict overwrite updates existing rows. Use --dry-run to preview ≤5 mapped rows without writing. See Connectors & Extensions for the full migration story.

Embedder-sidecar lifecycle#

The ingest hot path does not embed. Vectors are either caller-supplied (_emb_text/_emb_image/... in the row payload) or populated asynchronously by the embedder sidecar: set RELATA_ACCEL_ENDPOINT and the media-worker drain cycle populates _emb_* off the request thread (typically sub-second). The built-in CPU embedder (128-dim, deterministic) is query-side only — it embeds the recall() search query when no sidecar is configured. Circuit breaker opens after 3 consecutive failures ("embedder circuit open", 60 s cooldown). Full contract + reference Python sidecar: LLM & Embedding configuration.

SmartIngest identity pipeline#

SmartIngest runs on every ingest automatically — no configuration beyond registering canonical types and optional enrichment tables:

1. Identity extraction — emails, phone numbers, IBANs, MMSIs, VINs, IMEIs are detected from free-text fields using 76 canonical-type validators (eager regex/pattern phase; lazy detection deferred to materialized views).
2. Entity matching — extracted identifiers are matched against the existing entity graph to detect duplicates or known entities.
3. Entity merge — if a match is found, the new data is linked to the existing entity rather than creating a duplicate (FUSE_IDENTITIES).
4. Enrichment — registered enrichment lookup tables (CSV-backed) can augment rows at ingest time.

Tune detection latency with RELATA_ENRICH_MODE / RELATA_LAZY_TYPES (eager inline vs lazy background job). See DETECT_IDENTITIES in the SQL reference.

OTLP ingest (traces/logs/metrics)#

OpenTelemetry-compatible ingest endpoints:

Enrichment lookup tables#

Register a CSV as a query-time enrichment table:

curl -X POST http://localhost:9090/lookup/register \
  -H "Authorization: Bearer $RELATA_BEARER_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"name":"country_codes","data":[{"code":"US","name":"United States"},...]}'

Reference in SQL — see REGISTER LOOKUP / LOOKUP in the SQL reference and the worked example in the Query cookbook.

See also#

• Connectors & Extensions — Connector trait, dbt adapter, migration connectors, packs
• LLM & Embedding configuration — sidecar contract, _emb_* lifecycle
• Identity — canonical types and entity resolution
• Search & Retrieval — how ingested data becomes searchable
• Configuration — ingest-related env vars