datasqrl/sqrl/manage-connector

Creating External Connectors

Use CREATE TABLE statements with WITH clause to connect external data sources and sinks:

CREATE TABLE MyTable (
    column1 TYPE,
    column2 TYPE,
    event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
    WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND
) WITH (
    'connector' = 'connector-name',
    'option1' = 'value1',
    ...
);

Formatting rules — always follow, never compress onto a single line:

• CREATE TABLE name ( on its own line; each column definition on its own line with 2-space indent; ) WITH ( on its own line
• Each connector property goes on its own line inside WITH (, indented 4 spaces, with spaces around = (e.g. 'connector' = 'kafka-safe')
• Connector options that apply to only one runtime mode (streaming vs. batch) go inside conditional mustache blocks — {{#is_batch}} … {{/is_batch}} for batch-only options, {{^is_batch}} … {{/is_batch}} for streaming-only ones; {{#is_batch}} / {{/is_batch}} each go on their own line with no leading indent
• Separate each CREATE TABLE block with a blank line when multiple tables appear in the same file

Connector Sources

Before writing or editing a `WITH (...)` clause, every time, fully read the linked documentation of the connector you are about to use for the connector specific configuration options, even when you are copying a `CREATE TABLE` that already exists in the project.

• kafka and kafka-safe: Read or write append-only streams from Kafka-compatible data sources. Prefer the -safe version which adds DLQ and smart watermark support.
• upsert-kafka and upsert-kafka-safe: Read or write change streams for Kafka-compatible data sources. Prefer the -safe version which adds DLQ support.
• filesystem: Read or write data from local and cloud storage systems.
• iceberg: Read or write data from Apache Iceberg.
• jdbc: Write data via JDBC or use for lookup joins.
• datagen: Generate synthetic data for testing.
• print: Print output to stdout for debugging.
• blackhole: Discards all output (useful for testing).

CDC Connectors:

• mysql-cdc, postgres-cdc, and sqlserver-cdc: Read CDC change streams from MySQL, PostgreSQL, and Microsoft SQL Server. These Flink CDC connectors are included by default.

If the user requires connection to other data systems, instruct the user to contact customer support. Never suggest connectors not listed above.

Connector Formats

Always fully read the linked documentation of the format you are about to use for the format specific configuration options as well. The format specific options (*.timestamp-format.standard, nesting support, delimiters, schema-registry settings) appear only on the format page. Connectors and formats are based on Flink 2.3.x but DataSQRL extends those with useful features (like the -safe kafka connector variants).

The following formats are supported.

Popular formats:

• flexible-json and json: JSON-line format. Prefer flexible-json since it supports nested-json in export and is otherwise identical to json.
• avro: Avro format.
• avro-confluent: Avro with Confluent schema registry.
• csv: Delimited file format. Best for test data — Flink's CSV support is limited; prefer parquet/avro/flexible-json for production where practical.
• parquet: Apache Parquet columnar format.
• raw: Raw byte-based format for single column values.
• protobuf: Protocol Buffers format.
• orc: Apache ORC columnar format.

CDC formats:

• canal: Canal CDC format for streaming changes from MySQL.
• debezium: Debezium CDC format for streaming changes from MySQL, PostgreSQL, Oracle, SQL Server.
• maxwell: Maxwell CDC format for streaming changes from MySQL.
• ogg: Oracle GoldenGate CDC format for real-time data replication.

Connector Organization

Best Practice:

• Place all CREATE TABLE statements for one data source in a single .sqrl file
• Store in connectors/ folder (e.g., connectors/kafka-source.sqrl)
• Import into main script: IMPORT connectors.kafka-source.*;
• Reuse what already exists: if a source or sink is already defined as a .sqrl script in the project, import that script — never copy its CREATE TABLE into a new file. Treat an existing definition as correct and complete unless a compile or test run actually fails on it.
• Shared connectors: source/sink definitions reused across projects (e.g. a shared data catalog) live in a separate SQRL project declared under configuration file( See how to set configure file to use shared project by invoking /configure-sqrl skill and /implement-sqrl skill for IMPORT Statement section in sqrl-language spec). Invoke the /build-catalog skill when building such a catalog.

Environment Variables in Connector Config

Connector properties may reference environment variables. There are two forms, and the difference is security-relevant:

• '${VARIABLE_NAME}' is a non-secret variable. sqrl compile resolves it when it is available and writes the value into the build/ artifacts (a missing one is left as a placeholder for the runtime). Use it for bootstrap servers, hostnames, ports, connection strings (e.g. '${CUSTOMER_DATA_KAFKA_BROKERS}'), the consumer group / deployment id ('${DEPLOYMENT_ID}') and the test-data path ('${DATA_PATH}').
• '${{VARIABLE_NAME}}' is a secret variable. It is never resolved at compile time; the compiler rewrites it to ${VARIABLE_NAME} in the generated artifacts so the value is read only when the pipeline runs. Use it for every credential: passwords, tokens, API keys, SASL/JAAS strings (e.g. 'password' = '${{POSTGRES_PASSWORD}}'). A secret written as ${VAR} ends up in plain text in the build output, whatever the variable is called.

Use variables only for values that genuinely differ per deployment.

Hardcode logical names — Kafka topics, table names, stream names (e.g. 'topic' = 'customer_master.customer') are normally identical across environments and are not security sensitive, so a variable obfuscates the value.

Critical Requirements

Event-Time Processing:

• ALWAYS define watermarks on external sources used for streaming projects. Prefer event-time; use processing time only for the file-based-source exception below.
• Internal tables carry no `WATERMARK` definition. An internal table is a table with an engine hint like /*+engine(kafka) */ and no 'connector' option. DataSQRL manages that table and generates its watermark. Write a WATERMARK on such a table only when it is requested. Invoke the /implement-sqrl skill (CREATE TABLE internal vs external) for the full comparison.
• Watermark timestamps must be monotonically increasing with some bounded out-of-orderedness. If no such timestamp exists on source data, define an additional column ingestion_time AS now() and watermark on it. This can be useful for file-based sources.
• Define watermark: WATERMARK FOR ts AS ts - INTERVAL '1' SECOND where the interval is an upper limit of out-of-orderedness.
• Add a primary key (PRIMARY KEY (...) NOT ENFORCED) and/or a partition key only if they genuinely apply to the data source/sink.

Watermarks on production file-based sources (filesystem and Iceberg sources): use a data/metadata timestamp column as the WATERMARK only if BOTH (1) it is a quasi-monotonically increasing timestamp and (2) new files arrive frequently and deterministically (every few minutes up to ~1 hour). Otherwise — a non-monotonic column (e.g. S3 last_modified, which makes the watermark jump backwards) or infrequent/irregular file drops (which make it freeze) — add a computed ingestion-time column and watermark that instead:

ingestion_time AS now(),
WATERMARK FOR ingestion_time AS ingestion_time - INTERVAL '1' SECOND

This is the one sanctioned exception to "prefer event time"; it applies only to file-based sources whose data timestamp fails the requirements (1) or (2) defined above. Kafka and other streaming sources keep using their source/metadata timestamp.

This rule selects the watermark of a production source. A test connector reads from the filesystem but stands in for the production source, so it does NOT pick its watermark from this rule — it mirrors whatever the production connector does (see Test connectors mirror production below).

Table Types:

• STREAM: Append-only (e.g., filesystem, kafka-safe, iceberg). The planner defaults an unclassified connector to this type for imports.
• VERSIONED_STATE: Retraction stream (e.g., upsert-kafka-safe, mysql-cdc, postgres-cdc, sqlserver-cdc).
• LOOKUP: External lookup table (e.g., jdbc, jdbc-sqrl).

Kafka source watermarks: engines.kafka.use-source-watermark and use-transaction-source-watermark apply only to mutation tables with a timestamp metadata column and require kafka-safe or upsert-kafka-safe. They generate SOURCE_WATERMARK(); use a normal table watermark for other connector sources.

Entity Data: ingest entities as a stream of updates (append-only connector). Converting to versioned state is a downstream decision made per consumer: deduplicate with DISTINCT where a consumer needs a single version per entity, either the current one or the one valid at an event's timestamp in a temporal join, and read the STREAM where a consumer needs the changes themselves, such as a history view or an aggregation over every update. A source that already declares a PRIMARY KEY with upsert semantics is versioned state already, so it is read directly with no DISTINCT. Invoke the /implement-sqrl skill (DISTINCT operator) for the decision.

Connector Definitions

File layout

<project>/
└── connectors/
    ├── sources.sqrl          # schema tables ONLY — one `_X_schema` per source, shared by every environment
    ├── sources-test.sqrl     # connector tables for tests — filesystem over test-data/
    ├── sources-local.sqrl    # connector tables for local runs — e.g. DataSQRL-managed Kafka
    ├── sources-prod.sqrl     # connector tables for production — Kafka, Iceberg, JDBC, …
    └── test-data/
        └── <table>.jsonl     # records the test connectors read, as '${DATA_PATH}/<table>.jsonl'

• `sources.sqrl` declares only schema tables (_X_schema): the physical payload columns and their doc-strings. No connector, no WITH (...).
• `sources-<env>.sqrl` opens with IMPORT sources.*; and declares one connector table per source, each with its own WITH (...) LIKE _X_schema that defines the connector and extends the schema table. The table name is identical in every environment (CustomerUpdates in test, local and prod alike, only the connector behind it differs).
• The main script imports exactly one of these files: IMPORT connectors.sources-{{environment}} AS sources;, where {{environment}} comes from script.config.environment in that environment's package.json overlay.
• `test-data/` holds the records the test connectors read. Invoke the /test-sqrl skill for what those records must contain.

A source's payload columns are written once, in _X_schema, and shared by every environment. What each environment file adds is its own WITH (...) config, and its own event-time watermark declaration where the timestamp may come from a different place.

Column ownership