Data Platform - KaireonAI

Connector status. 80 connector types are registered, but only four can currently move data in a pipeline: Amazon S3, Google Cloud Storage, Azure Blob Storage, and SFTP. Those four are the union of the only paths that exist — the pipeline source enum, and campaign file delivery. Every other type is shown greyed as Coming soon in the New Connector picker and cannot be selected, because storing credentials for a connector nothing can read is a liability rather than a feature.The connectors documented on this page describe the registry entry and its configuration form. The remaining coming-soon types are the W16 expansion entries documented on Connectors Expanded. Coming-soon connectors are visibly badged and disabled in the connector picker.

Overview

The Data module is the foundation of KaireonAI. Everything the platform decides on — offers, scores, journeys — runs on clean, structured data flowing in from your systems. There is one workflow you’ll repeat for every entity you bring in:

Connect (Connector) → Define (Schema) → Build & Run (Pipeline) → Inspect (Aggregates)

Step	What you do	Page
1. Connect	Create a Connector with credentials + connection details (one connector serves many pipelines).	Connector Reference
2. Define	Create a Schema. The platform auto-creates the matching `ds_*` PostgreSQL table; adding fields runs ALTER TABLE. Pick a primary key (single or composite).	Data Model
3. Build & Run	Create a Pipeline pointing at the Connector + Schema. The visual editor’s Add-Node toolbar (transform / validate / enrich / branch) edits an in-memory draft; Save Changes versions it; Publish flips the live version the scheduler runs.	Flow Getting Started → Flow Editor UI → Transforms
4. Inspect	Inspect rows in Customer Viewer, traced events in Interaction History, and per-run metrics in Pipeline Runs.	Flow Lineage

Navigate to Data in the sidebar to access Connectors, Schemas, Pipelines, Pipeline Runs, Sources, Segments, and Aggregates (Customer Viewer + Interaction History).

One connector → many pipelines. The connector holds the bucket / host / credentials only. Each pipeline source node holds its own path, file mask, and format — so you can ingest different files into different schemas with different schedules from a single set of credentials. See the connector-reuse note on the Data → Connectors page.

Pipeline node behavior — current build. The standalone join node is functional: it performs a real in-memory relational join (inner / left / right / full) of its two upstreams and materializes the merged rows. A validate node carrying row-level rules now actually rejects failing rows from the load — only the passing rows reach the target (failing rows are dropped, and optionally quarantined to the DLQ table). The pipeline enrich node, however, is not yet wired: its providers (llm_tag, geocode, ml_score) need tenant credentials that aren’t implemented in this build, so the node now errors at run time instead of silently passing rows through. Remove the enrich node from a pipeline before running it. (This pipeline enrich node is distinct from the Decision Flow Enrich stage described under Schemas & Enrichment below, which is wired.)

Connectors

Connectors define how KaireonAI reaches your external data. 80 connector types are registered across 8 categories; four of them (S3, GCS, Azure Blob, SFTP) can currently be used in a pipeline, and the rest are shown as Coming soon. Each connector has its own dynamic configuration form with typed fields, multiple authentication methods, and a Test Connection button to verify connectivity before saving.

“Registered” vs. “runnable as a pipeline source.” A connector being registered (and connection-testable) is distinct from being wired as a pipeline source node. Today the flow runtime’s source executor ingests from file / object-store connectors — Amazon S3, Google Cloud Storage, Azure Blob Storage, SFTP, HTTP Pull, and Local Filesystem. The other registered connectors (warehouses, databases, CRM, CDP, messaging, streaming, and most API connectors) are usable for connectivity and, in several cases, for outbound/delivery flows, but their pipeline-source ingestion is not yet wired in this build. The Status column in each table below reflects the connector registry, not source-executor coverage.

Object Storage

Connector	Status	Auth Methods	Key Config Fields
Amazon S3	Ready (runtime ingestion + Test Connection)	IAM Role, Access Key	Bucket, path prefix, AWS region (30 regions), custom endpoint for S3-compatible storage
Google Cloud Storage	Ready (runtime ingestion + Test Connection)	Service Account JSON, Access Token, Application Default Credentials	Bucket, path prefix, GCP project ID
Azure Blob Storage	Ready (runtime ingestion + Test Connection)	Connection String, Account Key	Storage account, container name, blob prefix
SFTP	Ready (runtime ingestion + Test Connection)	Username/Password, SSH Key	Host, port, remote path
HTTP Pull	Ready (runtime ingestion via REST API connector)	API Key, Bearer Token	Source URL in `node.config.path`; SSRF-guarded by `validateAndResolve`
Local Filesystem	Ready (runtime ingestion)	None	Mount path (host-relative)

All five cloud backends and local_fs share a common object-store abstraction shipped in Phase 6.2, so a pipeline written against one backend behaves the same way against any of the others. Object storage connectors support file format selection: CSV, JSON, JSON Lines, Parquet, Avro, ORC, TSV, and XML — read end-to-end through the same format-parser layer regardless of source backend.

Streaming

Connector	Status	Auth Methods	Key Config Fields
Apache Kafka	Registered · streaming-gated	Username/Password (SASL), API Key	Bootstrap servers, topic, consumer group, security protocol, schema registry URL
Confluent Cloud	Registered · streaming-gated	API Key	Bootstrap server, topic, schema registry URL + credentials
Amazon Kinesis	Coming soon (connection test works)	IAM Role, Access Key	Stream name, AWS region, start position (LATEST or TRIM_HORIZON)

Streaming ingestion is gated and not available in the default build. The pipeline source node ingests from file / object-store connectors only (its IR connector enum is s3, gcs, azure_blob, sftp, ftp, local_fs, http_pull). Kafka, Kinesis, and Pulsar are streaming source kinds gated behind the FLOW_STREAMING_ENABLED=true environment variable on a self-hosted worker; when the flag is off (the default, and every hosted playground deployment) they raise a clear remediation error instead of ingesting. The connectors are registered and their connection test may work, but a Kafka/Kinesis/Pulsar pipeline source cannot run in the default build. For near-real-time ingestion on the playground, land data in object storage and schedule a Batch / Micro-Batch pipeline on a short cron. See streaming-runtime.

Data Warehouses

Connector	Status	Auth Methods	Key Config Fields
Snowflake	Ready	Username/Password, OAuth/Key-Pair Token	Account identifier, warehouse, database, schema, role, Source Table, Row Limit
Google BigQuery	Ready	Service Account JSON	GCP project ID, dataset, location, Source Table, Row Limit
Amazon Redshift	Ready	Username/Password, IAM Role	Cluster endpoint, port, database, schema
Databricks	Ready	Token, OAuth2	Workspace URL, HTTP path, Unity Catalog, schema

Snowflake and BigQuery row limits. Both connectors require a sourceTable and accept an optional rowLimit (default 100,000 rows — demo-safe). Set rowLimit to 0 to remove the cap entirely; only do this once you have sized the target database and pipeline run budget for a full-table read. The executor reads from the configured source table and caps the row count at rowLimit (or returns every row when rowLimit is 0).

Databases

Connector	Auth Methods	Key Config Fields
PostgreSQL	Username/Password, Connection String	Host, port, database, schema, SSL mode (prefer/require/verify-ca/verify-full/disable)
MySQL	Username/Password, Connection String	Host, port, database, SSL mode
MongoDB	Connection String, Username/Password	Database, default collection (Atlas or self-hosted)

CRM

Connector	Auth Methods	Key Config Fields
Salesforce	OAuth2, Username/Password + Security Token	Instance URL, API version, objects to sync (Contact, Account, Opportunity, etc.)
HubSpot	Private App Token, OAuth2	Portal ID, objects to sync (contacts, companies, deals)

Customer Data Platforms

Connector	Status	Auth Methods	Key Config Fields
Segment	Ready	API Key	Workspace slug, source ID

Messaging

Connector	Status	Auth Methods	Key Config Fields
Braze	Coming soon	API Key	REST API endpoint (US-01, US-02, US-03, EU-01)

APIs and Direct Upload

Connector	Status	Auth Methods	Key Config Fields
REST API	Ready	API Key, OAuth2, Token, Username/Password	Base URL, HTTP method (GET/POST), pagination type (none/offset/cursor/link), custom headers
Webhook	Ready	API Key, Bearer Token	Webhook path, max batch size
CSV File Upload	Ready	None	Delimiter (comma/tab/semicolon), header row detection
Shopify	Ready	Admin API Access Token	Shop domain
Stripe	Ready	Secret Key (sk_live_ or sk_test_)	No additional config required
Mailchimp	Ready	API Key	Server prefix, list/audience ID

Coming-soon connectors (amazon_kinesis, braze) are visibly badged and disabled in the connector picker. You can still view the form definitions in the registry, but creating a pipeline against them will no-op at run time until ingestion is implemented. Connection Test works today for Amazon Kinesis.

Security

Connector credentials are encrypted at rest using the platform encryption layer. The authConfig field is never returned in API responses — the GET endpoint explicitly excludes it from the select clause. Only the connection metadata (name, type, status, last tested timestamp) is exposed.

Schemas

Schemas define your entity structures. Unlike metadata-only schema systems, KaireonAI schemas are backed by real PostgreSQL tables. Creating a schema executes a create table statement. Adding a field runs alter table add column. Deleting a schema drops the table along with its dependent objects.

Entity Types

Each schema is assigned an entity type that describes what it models:

Entity Type	Typical Use
`customer`	Customer profiles, demographics, preferences
`account`	Account records, membership details
`transaction`	Purchase history, payment records
`product`	Product catalog, inventory
`event`	Behavioral events, clickstream, interactions
`interaction`	Customer-agent interactions, support tickets
`segment`	Pre-computed audience segments
`campaign`	Campaign metadata and performance
`subscription`	Subscription plans and status
`custom`	Any other entity type

Field Types and PostgreSQL Mapping

Every field you define maps to a concrete PostgreSQL column type:

Abstract Type	PostgreSQL Type	Notes
`varchar`	varchar (length-bounded)	Default length 255 if not specified
`text`	`text`	Unlimited length strings
`integer`	`integer`	32-bit signed integer
`bigint`	`bigint`	64-bit signed integer
`smallint`	`smallint`	16-bit signed integer
`numeric` / `decimal`	numeric (precision, scale)	Arbitrary precision; specify precision and scale
`float` / `real`	`real`	32-bit floating point
`double`	`double precision`	64-bit floating point
`boolean`	`boolean`	true/false
`date`	`date`	Calendar date
`timestamp`	`timestamp`	Date and time without timezone
`timestamptz`	`timestamptz`	Date and time with timezone
`json`	`jsonb`	Binary JSON for nested/dynamic data
`uuid`	`uuid`	Universally unique identifier

DDL Behavior

When you create a schema through the API or UI, the following happens:

Metadata record is created in the platform’s data-schema registry with field definitions.
A safe create table if not exists statement is executed against PostgreSQL with an auto-generated table name prefixed with ds_ (e.g., schema “customers” becomes table ds_customers).
Every table automatically gets created_at (timestamptz) and updated_at (timestamptz) columns.
Primary key handling: when no field is marked isPrimaryKey, an auto-generated id BIGSERIAL PRIMARY KEY column is added. When any field is marked isPrimaryKey: true (e.g. via the schema-create form’s “Custom primary key column” pane, or by passing isPrimaryKey: true in the API fields[] payload), the auto-id column is skipped and your column becomes the table’s PK.
Your defined fields are added as additional columns with their mapped PostgreSQL types, nullability, uniqueness constraints, and default values.

If the DDL fails, the metadata record is rolled back to prevent orphaned metadata without a backing table.

Schema operations execute real DDL statements against your database. Creating a schema creates a table, adding a field alters the table, and deleting a schema drops the table with CASCADE. These operations are not reversible through the UI.

Field Constraints

Each field supports the following constraints:

Constraint	DDL Effect
`isPrimaryKey`	Primary-key constraint on the column
`isUnique`	Uniqueness constraint (skipped if the column is already the primary key)
`isNullable: false`	Not-null constraint
`defaultValue`	Default-value clause (validated against safe literal patterns)

Default values are validated to prevent SQL injection — only literals ('text', 123, null, true, false), the current-timestamp function, and current_timestamp are permitted.

CSV Column Inference

When uploading CSV files, the platform can automatically infer column types from sample data:

Detected Pattern	Inferred Type
UUID format	`uuid`
true/false/yes/no/0/1	`boolean`
Integer values	`integer` (or `bigint` if > 2,147,483,647)
Decimal values	`numeric` with auto-sized precision and scale
ISO date strings	`date` or `timestamptz` (if time component present)
JSON objects/arrays	`json`
Short strings (≤255 chars)	`varchar` with auto-sized length
Long strings	`text`

Entity Types & Relationships

A schema’s kind is set by a single field, entityType, which determines how it participates in decisioning:

Entity Type	Purpose	Key Behavior
`customer`	Core customer profile data (demographics, preferences, scores)	Loaded automatically as the base `customer.*` namespace in every decision. One row per customer.
`account`	Account-level entity data (accounts, subscriptions, holdings)	Joined into decisions via a Schema Join.
`custom`	Any other entity (orders, transactions, addresses, catalog data) — the default	Joined into decisions via a Schema Join.

Related-entity data via Schema Joins: To bring a related entity (orders, addresses, accounts) into a decision, create a Schema Join (Data → Schema Joins): the customer schema is the primary side, the related schema is the foreign side, and aggregations (sum/count/avg/min/max/any/all/first) roll one-to-many rows up to the customer level at decision time. Joins with autoEnrich: true are included in the decision context automatically — see Schema Joins.

Schema References

Schemas are referenced throughout the platform:

Enrichment stages in Decision Flows load customer data from schema tables at decision time
Computed values reference schema data via the customer.* namespace in formulas
Pipelines use schemas as target destinations for ETL workflows
Segments define customer cohorts using schema field conditions with SQL-like filters

Pipelines

Pipelines are visual ETL workflows built with a drag-and-drop flow editor powered by React Flow. Each pipeline connects a source connector to a target schema through a chain of transform nodes.

Pipeline Structure

A pipeline consists of:

Connector — The source data connection (one of the four selectable types: S3, GCS, Azure Blob, SFTP)
Schema — The target destination table
Nodes — Visual nodes in the flow editor. The IR supports eight node kinds: source, transform, validate, branch, join, enrich, target, and outcomes (the inbound response-file sink — see the pipeline-node behavior note above for which are runnable today)
Edges — Connections between nodes defining data flow direction

Transform Types

KaireonAI provides 19 built-in transform types. Sixteen are implemented in the runtime executor and transform your data row-by-row. Three — vector embedding, geo resolution, and summarize — are IR-validated but currently pass rows through unchanged (they need tenant-supplied provider credentials / a materialization job that is not wired in this build). Sentiment scoring and language detection run in the built-in rule-based mode; sentiment scoring’s LLM mode is also passthrough until an LLM provider is wired. The most commonly used ops are detailed below:

Rename Field

Rename columns to standardize naming conventions across data sources. Configure the source field name and the desired target field name.Config: sourceField, targetField

Cast Type

Convert a field to a different data type. Supported target types: string, integer, bigint, float, numeric, boolean, date, timestamp, json, uuid.Config: field, targetType

Expression

Compute a new field using SQL-like expressions. Includes a function picker with 50+ built-in functions across 5 categories:

Category	Functions
String	`upper`, `lower`, `trim`, `substring`, `replace`, `concat`, `split_part`, `left`, `right`, `lpad`, `length`, `regexp_replace`
Numeric	`abs`, `round`, `ceil`, `floor`, `mod`, `greatest`, `least`, `coalesce`, `nullif`
Date/Time	`date_trunc`, `extract`, `age`, `to_char`, `to_date`, `to_timestamp`, `date + interval`, `date - interval`, `now`, `current_date`
Type Cast	Cast expressions to integer, bigint, numeric, text, boolean, date, timestamp, or uuid
Conditional	`case when`, `coalesce`, `nullif`, `if null`

Config: outputField, expression

Runtime coverage. The editor’s function picker lists the full SQL surface, but the in-process runtime currently evaluates a subset: the single-column string functions LOWER / UPPER / TRIM / LENGTH, plus the safe formula engine’s arithmetic, comparison, ternary, and built-ins (min, max, round, abs, coalesce, concat). Expressions the runtime can’t evaluate leave the row unchanged (no error), so validate the output of complex expressions before relying on them.

Filter Rows

Keep only rows matching specified conditions. Supports a visual condition builder with operators:

Operator	Applies To
`=`, `!=`	string, integer, numeric, boolean
`>`, `<`, `>=`, `<=`	integer, numeric, date, timestamp
pattern match (`like`), starts with, ends with	string
`in`, `not in`	string, integer
`is null`, `is not null`	all types
`between`	integer, numeric, date, timestamp

Config: field, operator, value

Runtime coverage. The in-process runtime evaluates the comparison operators (=, !=, >, <, >=, <=) against a single column and literal. Set-membership, pattern, null, and between predicates are accepted by the builder but are not yet enforced at run time — a predicate the runtime can’t parse conservatively keeps the row rather than dropping it.

Drop Field

Remove unwanted columns from the data flow. Select one or more fields to exclude from downstream processing.Config: fields (array)

Add Field

Add a new column with a name, data type, and default value or computed expression.Config: fieldName, fieldType, defaultValue

Map Values

Replace field values using a JSON lookup table. Useful for code-to-label mapping.Example: {"M": "Male", "F": "Female", "O": "Other"}Config: field, mappings (JSON object), defaultValue (for unmatched values)

Split Field

Split a single field into multiple output fields by a separator character. For example, split a full name into first and last name fields.Config: sourceField, separator, outputFields (array)

Merge Fields

Concatenate multiple columns into a single field with an optional separator string.Config: sourceFields (array), separator, outputField

Deduplicate

Remove duplicate rows based on one or more key columns. Keeps the first occurrence when duplicates are found.Config: keyFields (array)

Aggregate

Group by one or more columns and apply aggregate functions: sum, count, avg, min, max.Config: groupByFields (array), aggregations (array of {field, function, alias})

Lookup / Join

LEFT JOIN with another schema table to enrich data. Specify the lookup schema, join key, and which fields to pull from the lookup table.Config: lookupSchema, joinField, lookupField, selectFields (array)

Hash

Apply cryptographic hashing to field values. Supports SHA-256 and MD5 algorithms. Used for anonymization or generating deduplication keys.Config: field, algorithm (sha256 or md5), outputField

Mask PII

Detect and mask personally identifiable information. Supports partial masking patterns:

SSN: ***-**-1234
Email: j***@example.com
Phone: ***-***-5678
Credit card: ****-****-****-1234

Config: field, maskType, preserveLength

Summarize

Intended to aggregate related-entity data into a customer-level summary table at pipeline time (not decision time), so it does not add latency to the Recommend API.

Not yet materialized at runtime. The summarize op is IR-validated but the runtime executor currently passes rows through unchanged — it does not populate a summary table. No summary-table materialization code ships in this build. For customer-level rollups of related-entity data, use Schema Join aggregations at decision time, or an Aggregate transform writing to a dedicated schema.

Config: customerKey, windowDays

Keyboard Shortcuts

Shortcut	Action
Ctrl/Cmd + Z	Undo last action
Ctrl/Cmd + Shift + Z	Redo last undone action

Undo/redo buttons are also available in the editor toolbar.

Execution Config

Pipelines support the following execution modes with configurable resource allocation:

Mode	Status	Description	Use Case
Batch	Ready	Process all records in configurable batch sizes	Full loads, daily syncs
Micro-Batch	Ready	Small frequent batches scheduled via cron	Near real-time with controlled throughput
Streaming	Self-host	Continuous record-by-record processing via long-lived consumer	Real-time event streams — Kafka, Kinesis, Pulsar

Streaming mode spawns a long-lived consumer per pipeline and is gated behind the FLOW_STREAMING_ENABLED=true environment variable on the worker service. The gate exists because a streaming consumer requires a persistent worker container (separate from the request-driven API) that is part of the self-hosted deployment topology, not the hosted playground. Selecting Streaming on the playground returns a clear error pointing to streaming-runtime. For near-real-time ingestion on the hosted playground, use Batch or Micro-Batch with a short cron cadence against a batch-polling connector (e.g., Kafka).

Configuration Options

Setting	Default	Description
Batch Size	10,000	Records per batch (batch/micro-batch modes)
Parallelism	1	Concurrent workers (1 to 16)
Partitioning	None	Partition key for distributed processing
Error Handling	Fail	`skip` (continue on error), `fail` (stop pipeline), `dlq` (route to dead-letter queue)
Scheduling	Manual	Cron expression for automatic runs

Loading Strategies

When a pipeline writes to a target schema, you choose how incoming data merges with existing rows. The loading strategy is configured per pipeline and applies at execution time. The target executor supports six load modes:

Strategy	Behavior	Use Case
Append	Insert new rows. Existing rows are untouched. This is the default.	Incremental event streams, transaction logs
Truncate & Load	Truncate the target table, then insert all rows in one transaction. Brief downtime while the table is empty. Guarded by an empty-source check (`failOnEmptySource`, default on) plus a pre-load backup.	Full refresh of small-to-medium lookup tables
Upsert	Insert with on-conflict-do-update using a configurable key column (`upsertKey`). Rows with matching keys are updated; new keys are inserted.	Incremental updates where source rows may change (e.g., customer profiles)
Incremental Watermark	Loads only rows past a stored high-water mark. Requires a `watermarkColumn` (timestamp / integer / bigint); the watermark is persisted per pipeline, and the first run falls back to `MAX(column)`.	Append-only sources where you only want new rows each run
Blue-Green Swap	Loads all rows into a `<target>_new` table (`CREATE TABLE … LIKE … INCLUDING ALL`), then performs an atomic rename triple (in one transaction) to swap it into production. The production table is untouched until the swap succeeds. If loading fails, `<target>_new` is dropped and production is never affected.	Zero-downtime full refresh for production-critical tables. Recommended for high-availability deployments.
CDC Mirror	Applies the batch rows as a change feed: an op column (`cdcOpColumn`) on each row drives upsert vs delete by the CDC key (`upsertKey`). Unknown op values fail the run before any DML.	Applying CDC extracts (e.g., Debezium-shaped files) so the target mirrors the source table

Blue-Green Swap is the safest strategy for production data. The atomic rename means readers see either the old table or the new table — never a partially loaded state.

Row Validation

Every pipeline run validates incoming rows before writing to the target table. Validation catches type mismatches, null violations, and length overflows before they hit the database.

Mode	Behavior
Strict	Fail the entire pipeline on the first validation error. No rows are written. Use when data quality is critical.
Skip	Log the error and skip the invalid row. Valid rows are still written. Use when partial loads are acceptable.
Coerce	Attempt to fix the value (e.g., cast `"123"` to integer, trim overlength strings). If coercion fails, the row is skipped.

Validation checks include:

Type checking — Does the value match the target column’s data type?
Null validation — Is a NOT NULL column receiving a null value?
Length limits — Does a VARCHAR(n) value exceed its maximum length?

Validation errors are surfaced in the UI with row number, column name, and error description, so you can trace exactly which source rows failed and why.

Progress Tracking

Pipeline executions display real-time progress in the UI:

Progress bar with percentage complete based on rows processed vs. estimated total
Row counters showing loaded, failed, and skipped counts updated in real time
Validation error accordion that expands to show individual row-level errors with field and reason
Run history table showing each execution’s loading strategy, duration, row counts, and error summary

Run History

Each pipeline tracks its execution history. A run’s status moves through pending → running and settles on completed, completed_with_errors, or failed, alongside timestamps and error details. The pipeline list view shows the last run status and timing at a glance.

Supported File Formats

When reading from file-based sources (local_fs, s3, gcs, azure_blob, sftp, http_pull — all share the same format-parser layer via the common object-store abstraction shipped in Phase 6.2):

Format	Extension	Notes
CSV	`.csv`	RFC-4180 quoting/escaping via papaparse; header row → object keys; default delimiter `,`
JSON	`.json`	Array → one row per element; bare object → single row
JSON Lines	`.jsonl`	One JSON object per line; parse error reports line number
Parquet	`.parquet`	hyparquet reader (pure JS); up to ~50 MB per file
Avro	`.avro`	avsc reader; schema-embedded container files only
ORC	`.orc`	Self-host via the `@kaireonai/orc-native` add-on (not yet published). Workaround: convert ORC → parquet upstream.
TSV	`.tsv`	papaparse with `delimiter: "\t"`; same quoting/escaping as CSV
XML	`.xml`	fast-xml-parser; configurable `recordPath` (default: every direct child of root); attribute keys prefixed with `@_`

Field Reference

Connector Fields

Field	Type	Description
`id`	string	Auto-generated unique identifier
`name`	string	Display name for the connector
`type`	enum	One of 80 registered connector types; only `aws_s3`, `gcs`, `azure_blob`, `sftp` are usable in a pipeline
`category`	enum	`object_storage`, `streaming`, `warehouse`, `database`, `crm`, `cdp`, `messaging`, `api`
`authMethod`	enum	`iam_role`, `access_key`, `service_account_json`, `connection_string`, `oauth2`, `api_key`, `username_password`, `ssh_key`, `token`, `none`
`authConfig`	object	Encrypted credentials (never returned in API responses)
`config`	object	Connection-specific settings (bucket, host, topic, etc.)
`status`	enum	`connected`, `disconnected`, `error`, `testing`
`lastTestedAt`	datetime	Timestamp of most recent connection test
`lastError`	string	Error message from last failed test

Schema Fields

Field	Type	Description
`id`	string	Auto-generated unique identifier
`name`	string	Machine-safe name (lowercase, underscored) used for table naming
`displayName`	string	Human-readable name shown in the UI
`description`	string	Optional description
`entityType`	enum	`customer`, `account`, `transaction`, `product`, `event`, `interaction`, `segment`, `campaign`, `subscription`, `custom`
`tableName`	string	Auto-generated PostgreSQL table name (prefixed with `ds_`)
`fields`	array	Ordered list of field definitions (name, dataType, constraints)
`status`	enum	`active`, `draft`, `archived`

Pipeline Fields

Field	Type	Description
`id`	string	Auto-generated unique identifier
`name`	string	Display name for the pipeline
`description`	string	Optional description
`connectorId`	string	Reference to the source connector
`schemaId`	string	Reference to the target schema
`irVersion`	string	IR document version, always `"1.0"`. The node graph is stored in the versioned pipeline IR (JSONB), not in per-node/per-edge columns
`publishedIrVersion`	int	The IR version currently published/active; the scheduler loads the published-or-latest IR
`schedule`	string	Cron expression for scheduled execution (null for manual)
`executionConfig`	object	Batch size, parallelism, partitioning, error handling
`status`	enum	`draft`, `active`, `paused`, `archived`
`lastRunAt`	datetime	Timestamp of most recent execution
`lastRunStatus`	string	Status of the most recent run: `running`, `completed`, `completed_with_errors`, or `failed`

Worked Example

This example walks through creating a customer data pipeline end-to-end: define a schema, connect to S3, build a pipeline with transforms, and execute.

Step 1: Create a Customer Schema

curl -X POST /api/v1/schemas \
  -H "Content-Type: application/json" \
  -d '{
    "name": "customers",
    "displayName": "Customer Profiles",
    "description": "Core customer entity for decisioning",
    "entityType": "customer",
    "fields": [
      { "name": "name", "dataType": "varchar", "length": 100, "isNullable": false },
      { "name": "email", "dataType": "varchar", "length": 255, "isNullable": false, "isUnique": true },
      { "name": "credit_score", "dataType": "integer", "isNullable": true },
      { "name": "balance", "dataType": "numeric", "precision": 12, "scale": 2, "isNullable": true }
    ]
  }'

This creates a PostgreSQL table named ds_customers with the following DDL:

CREATE TABLE IF NOT EXISTS "ds_customers" (
  "id" BIGSERIAL PRIMARY KEY,
  "created_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  "updated_at" TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  "name" VARCHAR(100) NOT NULL,
  "email" VARCHAR(255) NOT NULL UNIQUE,
  "credit_score" INTEGER,
  "balance" NUMERIC(12,2)
);

Step 2: Create an S3 Connector

curl -X POST /api/v1/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Customer Data Lake",
    "type": "aws_s3",
    "config": {
      "bucket": "my-data-lake",
      "prefix": "customers/daily/",
      "region": "us-east-1"
    },
    "authMethod": "iam_role",
    "authConfig": {
      "roleArn": "arn:aws:iam::123456789012:role/KaireonDataAccess"
    }
  }'

Step 3: Build a Pipeline with Transforms

Create a pipeline that reads customer CSVs from S3, filters out zero-balance records, renames a field, and masks PII before loading into the schema: The create endpoint is IR-native: the body carries { irVersion: "1.0", ir: <Pipeline IR> } alongside name, connectorId, and schemaId. The ir document holds the full node graph (a legacy nodes/edges body is rejected). The transform node chains its ops in a single node’s ops array:

curl -X POST /api/v1/pipelines \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Customer Daily Load",
    "connectorId": "<connector-id>",
    "schemaId": "<schema-id>",
    "irVersion": "1.0",
    "ir": {
      "kind": "pipeline",
      "version": "1.0",
      "id": "customer-daily-load",
      "metadata": { "name": "Customer Daily Load", "description": "Ingest customer data with PII masking" },
      "nodes": [
        {
          "id": "src",
          "kind": "source",
          "connector": "s3",
          "config": {
            "bucket": "my-data-lake",
            "path": "customers/daily/",
            "pattern": { "type": "glob", "value": "*.csv" },
            "ordering": "fifo_by_name",
            "waitPolicy": { "maxRetries": 3, "intervalMinutes": 5, "onMissAction": "alert" },
            "atomicity": { "stagingFolder": ".staging", "successFolder": ".archive", "failureFolder": ".failed" },
            "format": "csv"
          }
        },
        {
          "id": "clean",
          "kind": "transform",
          "input": "src",
          "ops": [
            { "type": "filter", "predicate": "balance > 0" },
            { "type": "rename", "from": "email_address", "to": "email" },
            { "type": "mask_pii", "field": "ssn", "maskType": "ssn" }
          ]
        },
        {
          "id": "tgt",
          "kind": "target",
          "input": "clean",
          "schema": "public.ds_customers",
          "loadMode": "append"
        }
      ],
      "errorHandling": {
        "dlq": { "enabled": true },
        "retry": { "maxAttempts": 3, "backoff": "exponential" }
      }
    }
  }'

This pipeline:

Reads CSV files from the customers/daily/ prefix in S3
Filters to rows where balance > 0 (the runtime evaluates a single <column> <op> <literal> comparison per filter op)
Renames email_address to email to match the schema field name
Masks the ssn field, keeping only the last four digits
Appends the transformed rows into the ds_customers table

API Quick Reference

Connectors

Operation	Method	Endpoint
List connectors	GET	`/api/v1/connectors`
Create connector	POST	`/api/v1/connectors`
Update connector	PUT	`/api/v1/connectors`
Delete connector	DELETE	`/api/v1/connectors?id={id}`

Schemas

Operation	Method	Endpoint
List schemas	GET	`/api/v1/schemas`
Create schema (also creates the backing table)	POST	`/api/v1/schemas`
Delete schema (also drops the backing table)	DELETE	`/api/v1/schemas?id={id}`

Pipelines

Operation	Method	Endpoint
List pipelines	GET	`/api/v1/pipelines`
Create pipeline	POST	`/api/v1/pipelines`
Update pipeline	PUT	`/api/v1/pipelines`
Delete pipeline	DELETE	`/api/v1/pipelines?id={id}`

All endpoints require authentication and tenant context. Responses use cursor-based pagination with limit and cursor parameters.

For complete API request/response schemas, see the Connectors API, Schemas API, and Pipelines API.

Decision Flows

Use Enrich stages to load schema data at decision time for real-time personalization.

Computed Values

Write formulas that reference customer.* fields from your schema tables.

Core Concepts

Understand how data connects to decisioning and delivery across the platform.

​Overview

​Connectors

​Object Storage

​Streaming

​Data Warehouses

​Databases

​CRM

​Customer Data Platforms

​Messaging

​APIs and Direct Upload

​Security

​Schemas

​Entity Types

​Field Types and PostgreSQL Mapping

​DDL Behavior

​Field Constraints

​CSV Column Inference

​Entity Types & Relationships

​Schema References

​Pipelines

​Pipeline Structure

​Transform Types

​Keyboard Shortcuts

​Execution Config

​Configuration Options

​Loading Strategies

​Row Validation

​Progress Tracking

​Run History

​Supported File Formats

​Field Reference

​Connector Fields

​Schema Fields

​Pipeline Fields

​Worked Example

​Step 1: Create a Customer Schema

​Step 2: Create an S3 Connector

​Step 3: Build a Pipeline with Transforms

​API Quick Reference

​Connectors

​Schemas

​Pipelines

​Related

Decision Flows

Computed Values

Core Concepts

Overview

Connectors

Object Storage

Streaming

Data Warehouses

Databases

CRM

Customer Data Platforms

Messaging

APIs and Direct Upload

Security

Schemas

Entity Types

Field Types and PostgreSQL Mapping

DDL Behavior

Field Constraints

CSV Column Inference

Entity Types & Relationships

Schema References

Pipelines

Pipeline Structure

Transform Types

Keyboard Shortcuts

Execution Config

Configuration Options

Loading Strategies

Row Validation

Progress Tracking

Run History

Supported File Formats

Field Reference

Connector Fields

Schema Fields

Pipeline Fields

Worked Example

Step 1: Create a Customer Schema

Step 2: Create an S3 Connector

Step 3: Build a Pipeline with Transforms

API Quick Reference

Connectors

Schemas

Pipelines

Related