datasqrl/sqrl/implement-sqrl

Implement SQRL scripts

SQRL is an extension of Flink SQL that adds support for table functions and convenience syntax to build reactive data processing and serving applications. The “R” in SQRL stands for Reactive and Relationships.

This document focuses only on features unique to SQRL; when SQRL accepts Flink SQL verbatim we simply refer to the upstream spec.

Reference Templates

Review project examples in /opt/datasqrl-examples/README.md before implementing new features to identify related projects and copy features from those reference implementations.

Script Structure

A SQRL script is an ordered list of statements separated by semicolons (;). Only one statement is allowed per line, but a statement may span multiple lines.

Typical order of statements:

IMPORT ...            -- import other SQRL scripts
CREATE TABLE ...      -- define internal & external sources
MyTable := SELECT ... -- define tables or functions (with hints)
EXPORT MyTable TO ... -- write table data to sinks

At compile time the statements form a directed-acyclic graph (DAG). Each node is then assigned to an enabled execution engine according to the optimizer and the compiler generates the data processing code for that engine. See Where a table runs.

Flink SQL

SQRL inherits full Flink SQL grammar for

• CREATE {TABLE | VIEW | FUNCTION | CATALOG | DATABASE}
• SELECT queries inside any of the above
• USE ...
• INSERT INTO

SQRL currently tracks Flink 2.3; use its compatible SQL syntax. Refer to the Flink SQL documentation for the detailed specification.

Type System

In SQRL, every table and function has a type based on how the table represents data. The type determines the semantic validity of queries against tables and how data is processed by different engines.

SQRL assigns one of the following types to tables based on the definition:

• STREAM: Represents a stream of immutable records with an assigned timestamp (often referred to as the "event time"). Streams are append-only. Stream tables represent events or actions over time.
• VERSIONED_STATE: Contains records with a natural primary key and a timestamp, tracking changes over time to each record, thereby creating a change-stream.
• STATE: Similar to VERSIONED_STATE but without tracking the history of changes. Each record is uniquely identified by its natural primary key.
• LOOKUP: Supports lookup operations using a primary key against external data systems but does not allow further processing of the data.
• STATIC: For data that does not change over time, such as constants.

The table type determines what operators a table supports and how those operators are applied.

IMPORT Statement

Imports another SQRL script into the current script.

IMPORT qualifiedPath (AS identifier)?;
IMPORT qualifiedPath.*;             -- wildcard

Path resolution

A qualifiedPath resolves in exactly one of three ways:

• Relative (default) — against the directory of the importing script: IMPORT my.custom.script maps to ./my/custom/script.sqrl.
• `root.` (reserved) — resolves the rest of the path against the build root of the top-level project being compiled, no matter how deeply nested the importing script is. SQRL dot-paths have no .., so a script in a subfolder cannot climb upward relatively; root. is the only way for it to reach something at the project root (IMPORT root.data_catalog.identity.customers;, IMPORT root.functions.my_func;).
• `stdlib.` (reserved) — resolves to compiler built-ins (e.g. IMPORT stdlib.math.*;), not the file system. Import a whole library with .* or a single function by name.

• The path may continue one segment past a SQRL script to import a single `CREATE TABLE` from it. Standard-library paths may similarly select a function, e.g. IMPORT stdlib.math.hypot AS hypotenuse. Hyphenated segments are written as they are (IMPORT connectors.kafka-source.*); backticking a segment is also accepted (` IMPORT data_catalog.sources ). A {{...}} template variable in the path is substituted from script.config before the path is resolved (IMPORT connectors.sources-{{environment}} AS sources;` — see Config Templating).
• Imports that end in .* are imported inline which means that the statement from that script are executed verbatim in the current script.

Otherwise, imports are available within a namespace that's equal to the name of the script or the optional AS identifier.

Examples

• IMPORT my.custom.script.*: All table definitions from the script are imported inline and can be referenced directly as MyTable in FROM clauses.
• IMPORT my.custom.script: Tables are imported into the script namespace and can be referenced as script.MyTable
• IMPORT my.custom.script AS myNamespace: Tables are imported into the myNamespace namespace and can be referenced as myNamespace.MyTable
• IMPORT connectors.sources-{{environment}} AS sources: Imports the connector file selected by the environment config variable (sources-test.sqrl, sources-prod.sqrl, …) into the sources namespace
• IMPORT functions.* / IMPORT functions.my_func: Imports Java UDFs from the project's functions/ folder (invoke the /implement-udf skill); IMPORT stdlib.math.* imports built-in functions (see functions.md)

Import style controls API exposure

Because a wildcard import executes the statements verbatim in the current script (above), its tables become definitions of this script and follow the normal exposure rules (see Interfaces). A namespaced import keeps its tables private: they can be referenced in FROM clauses as namespace.Table, but they are not exposed in the interface directly.

IMPORT connectors.sources AS sources;   -- private: sources.Customer usable in FROM, absent from the API directly
IMPORT connectors.enablement.*;         -- inline: these tables are exposed in the API

To publish only what the API needs, import the sources aliased and re-declare the tables that should be exposed:

IMPORT connectors.sources AS sources;
/** Purchases with computed line total. */
Purchase := SELECT *, quantity * unit_price AS total_price FROM sources.Purchase;

When a single table must be a mutation, put it in its own connector file and wildcard-import that file, while the read-only sources stay aliased.

Included projects / multi-project (script.include)

A project can import scripts from another SQRL project, e.g. a shared data catalog or a metrics layer reused by several pipelines, declared under script.include in the project's configuration file. To include another SQRL project, invoke the /configure-sqrl skill for configuration. What matters for the language is how the imports resolve:

• At compile time the included project (the directory holding the top-level package file, together with all its subfolders, every .sqrl in them is loaded) is copied into `build/<namespace>/`, alongside the consumer's own scripts, where <namespace> is the key of its script.include entry in the configuration file. A script reaches it with a plain relative import, like IMPORT <namespace>.sources AS sources;, and that is the preferred form. Name include namespaces with underscores (data_catalog), because a dash has a different meaning in SQL. When the importing script is in a subdirectory (e.g. ./my-module/module-script.sqrl), use IMPORT root.<namespace>.sources; (see Path resolution). Below the namespace the path follows the included project's own directory structure at any depth. Example:IMPORT data_catalog.customer.customer_data.customer_master-{{environment}} AS cs;.
• {{...}} in the consumer's import statement is substituted from the consumer's own script.config; {{...}} inside the included project's scripts (e.g. a catalog's {{environment}} or a {{^is_batch}}…{{/is_batch}} section) is rendered with the script.config of the package file named by package, overridden by script.include.<namespace>.config.
• Inside an included project, reference sibling scripts with *relative* imports, so they travel with the copied files. A root. import inside an included project breaks once it is consumed: root then points at the consumer's build root, not the included project's.

CREATE TABLE (internal vs external)

SQRL understands the complete Flink SQL CREATE TABLE syntax, but distinguishes between internal and external source tables. External source tables are standard Flink SQL tables that connect to an external data source (e.g. database or Kafka cluster). Internal tables are managed by SQRL through an engine hint (kafka or iceberg). Kafka internal sources are exposed for data ingestion in the interface.

Example (internal): An internal table managed by DataSQRL as a Kafka topic carries the engine(kafka) hint (see the /manage-connector skill):

/*+engine(kafka) */
CREATE TABLE Customer (
  customerid BIGINT,
  email STRING,
  ts TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',
  PRIMARY KEY (customerid) NOT ENFORCED
);

The internal table does not need WATERMARK declaration. SQRL generates the watermark for it. Declare a WATERMARK explicitely on an internal table only when the user asks for one.

With engine(kafka), PARTITIONED BY (col, ...) on an internal table without a PRIMARY KEY keys each record by those columns, both the records the server writes for mutations and the records the pipeline exports into the table, so records with the same key stay on one Kafka partition in order. The key columns can be uuid or auth metadata columns, but not the timestamp metadata column. The Flink source ignores that key and reads every column from the message value, so adding PARTITIONED BY to an existing internal table is safe: records written before the table had a key are still read, also when the topic is read again from the earliest offset. This relies on the default kafka-safe connector. Do not switch internal Kafka tables to the plain kafka connector, which drops records that have no key. Use PARTITIONED BY when records only need a Kafka key for partitioning and ordering. A PRIMARY KEY also keys the records, but it makes the table an upsert table (VERSIONED_STATE), and its upsert source drops records that were written without a key.

Example (external): An external table connects to a data system through a WITH (...) clause and declares its own WATERMARK:

CREATE TABLE kafka_json_table (
  user_id    INT,
  name       STRING,
  event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
  WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND
) WITH (
  'connector' = 'kafka-safe',
  'topic'     = 'users',
  'format'    = 'flexible-json'
);