Streaming SQL reference

How-to guideApplies to: Unified engine deployments

Use these commands through Workbench, Spark Connect SQL, or the SQL operations API on a Laminar instance with Streaming enabled. Syntax templates below describe command forms. The examples can be copied into Workbench.

Register a connector

CREATE TABLE connector_name [(column_name data_type, ...)]
WITH (connector = 'connector_type', type = 'source_or_sink', ...);

The connector option identifies a Streaming connector declaration. This registers a native Streaming table definition. It does not create a Unified catalog table or a Unified read connection. Use a column list for fixed-schema Kafka JSON input. An Iceberg sink can omit its columns when infer_schema = 'true'. Connector-specific options are parsed by the existing Streaming connectors. See Sources and Sinks for complete declarations and connection options.

Create a pipeline

CREATE STREAMING PIPELINE pipeline_name
[WITH (
  parallelism = positive_integer,
  checkpoint_interval = 'number unit',
  start_paused = true_or_false
)]
AS INSERT INTO sink_name
SELECT ... FROM source_name ...;

The transformation must be an INSERT INTO ... SELECT ... statement accepted by the Streaming planner. Register the referenced source and sink first. A WITH name AS (...) query step can appear after the INSERT INTO target.

Pipeline option	Default	Accepted value
`parallelism`	`1`	A positive integer; the planner applies operator constraints
`checkpoint_interval`	`'10 seconds'`	Whole-number duration from 1 second through 1 day
`start_paused`	`false`	`true` saves the initial job as stopped; `false` makes it eligible to start

Supported duration units are ms, millisecond, milliseconds, s, second, seconds, m, minute, and minutes. Use a space between the whole number and the unit. For example, '1000 ms', '10 seconds', and '1 minute' are valid. Use lowercase unit names. Values such as '0 seconds', '1 ms', and '1.5 seconds' are rejected. Unknown pipeline options are rejected. Put execution options in the pipeline’s WITH clause. Workbench’s bounded compute selection is not the configuration for this Streaming job. Creation validates and compiles the transformation, then saves the native program and job. It starts automatically unless start_paused = true. Success confirms acceptance, rather than worker startup or output publication.

List pipelines

SHOW PIPELINES;

Returns one JSON metadata value per saved native pipeline in the result column. An instance with no pipelines returns an empty result.

Describe a pipeline

DESCRIBE PIPELINE orders_to_iceberg;

DESC PIPELINE orders_to_iceberg is also accepted. Replace the name with an existing pipeline. A missing name reports a not-found error. The result contains pipeline_id, name, job_id, state, stop_mode, query, and checkpoint_interval_micros. Manage pipelines explains these fields.

Stop a pipeline

STOP PIPELINE orders_to_iceberg;

Requests a checkpoint stop. Acceptance is asynchronous. Inspect the pipeline until its state reaches Stopped.

Start a stopped pipeline

START PIPELINE orders_to_iceberg;

Clears the stop request on the existing pipeline. This also starts a pipeline that was created with start_paused = true. Jobs in Failed or Finished state require the existing native pipeline controls.

Statement boundaries and retries

Send one statement per SQL operation. Workbench Run all can submit a complete script in order. If a statement fails, later statements are not run. Earlier successful operations remain accepted. A script is not a transaction, and no BEGIN ... END wrapper is required. Use SQL literals in these definitions. Bound SQL arguments are not supported. Names follow the existing Streaming registry rules. Choose unused names when creating new definitions. For a lost response, retain the same session and operation ID. A matching retry returns the same accepted resource. Reusing the identity with different content fails. See Retry a request.

Current command limits

This SQL interface does not provide ALTER STREAMING PIPELINE, DROP PIPELINE, or connector deletion. Use the existing native controls for those actions. Connector declarations do not create external Kafka topics or provision an Iceberg catalog or warehouse. Streaming transformations use the existing Streaming planner and connector behavior. A general Unified query, or a DataFrame writeStream definition, is not a replacement for the pipeline definition shown here. Operational metrics and service visibility use the shared Observability docs.

Transformation Engine

Learn Laminar

Workloads

SQL reference

HTTP API

Recipes

Operations

Streaming SQL reference

Register a connector

Create a pipeline

List pipelines

Describe a pipeline

Stop a pipeline

Start a stopped pipeline

Statement boundaries and retries

Current command limits

​Register a connector

​Create a pipeline

​List pipelines

​Describe a pipeline

​Stop a pipeline

​Start a stopped pipeline

​Statement boundaries and retries

​Current command limits

Register a connector

Create a pipeline

List pipelines

Describe a pipeline

Stop a pipeline

Start a stopped pipeline

Statement boundaries and retries

Current command limits