CREATE TABLE ... FLOW (pipelines)

Use the CREATE TABLE ... FLOW statement to create a managed table in a pipeline, written by one or more flows.

Syntax

CREATE TABLE
  table_name
  [ table_specification ]
  [ table_clauses ]
  [ flow_clause ]

table_specification
  ( { column_identifier column_type [column_properties] } [, ...]
    [ CONSTRAINT expectation_name EXPECT (expectation_expr)
        [ ON VIOLATION { FAIL UPDATE | DROP ROW } ] ] [, ...] )

table_clauses
  { PARTITIONED BY (col [, ...]) |
    CLUSTER BY clause |
    LOCATION path |
    COMMENT table_comment |
    TBLPROPERTIES clause |
    WITH { ROW FILTER clause } } [ ... ]

flow_clause
  FLOW [ IMPORT CHECKPOINT checkpoint_path ] INSERT [ONCE] BY NAME query

To fan multiple sources into one managed table, declare several flows that target it with CREATE FLOW (pipelines):

CREATE FLOW flow_name AS INSERT INTO table_name BY NAME query

Parameters

• table_name

  The name of the managed table to create. If the name is not qualified, the table is created in the pipeline's target schema. The name must not already belong to a streaming table.

• table_specification

  Optionally defines the columns, their types, properties, and descriptions. If omitted, the schema is inferred from the flow query.

• CONSTRAINT expectation_name EXPECT (expectation_expr) [ ON VIOLATION { FAIL UPDATE | DROP ROW } ]

  Adds data quality expectations to the managed table. These data quality expectations can be tracked over time and accessed through the pipeline's event log. A FAIL UPDATE expectation causes the processing to fail when both creating the table as well as refreshing the table. A DROP ROW expectation causes the entire row to be dropped if the expectation is not met. See Manage data quality with pipeline expectations.

  expectation_expr can be composed of literals, column identifiers within the table, and deterministic, built-in SQL functions or operators except:

  • Aggregate functions
    • Analytic window functions
    • Ranking window functions
    • Table valued generator functions

  Also expectation_expr must not contain any subquery.

• PARTITIONED BY (col [, ...])

  Optionally partitions the table by a subset of columns.

• CLUSTER BY clause

  Optionally enables liquid clustering on the table. You cannot combine PARTITIONED BY and CLUSTER BY.

• LOCATION path

  An optional storage location for the table data.

• COMMENT table_comment

  A STRING literal describing the table.

• TBLPROPERTIES clause

  Optionally sets one or more user-defined table properties.

• WITH ROW FILTER clause

  Adds a row filter function to the table. Future queries for that table receive a subset of the rows for which the function evaluates to TRUE.

• FLOW INSERT [ONCE] BY NAME query

  Defines an append flow that inserts the result of query into the table, matching result columns to table columns by name. query can reference batch or streaming sources. ONCE runs the flow a single time (for example, for a backfill) rather than on every update. Each named flow processes its input exactly once per pipeline update, identical to FLOW INSERT BY NAME on a streaming table.

• IMPORT CHECKPOINT checkpoint_path

  A STRING literal with the path to an existing Structured Streaming checkpoint to import into the flow, so a migrated stream resumes from its last committed offset instead of reprocessing the source. Importing a checkpoint is in Beta. See Migrate a Structured Streaming checkpoint.

Migrate a Structured Streaming checkpoint

Use IMPORT CHECKPOINT to migrate an existing Structured Streaming workload to a pipeline without reprocessing the source. Set checkpoint_path to the checkpointLocation your Structured Streaming query used, which can be a cloud storage, Unity Catalog volume, or DBFS path. On the first pipeline update, the flow clones that checkpoint into the pipeline's managed storage. The flow then resumes from the last committed offset with its state (such as aggregations, deduplication keys, and watermarks) intact. Subsequent pipeline updates use the flow's cloned checkpoint; the original checkpoint is not modified.

Recreate the Structured Streaming query as a pipeline flow that continues writing to the table the query wrote to, and import its checkpoint:

-- Use the existing table your Structured Streaming query writes to; CREATE TABLE adopts it.
CREATE TABLE streaming_table
-- Continue from the imported checkpoint instead of reprocessing the source.
FLOW IMPORT CHECKPOINT '/Volumes/my_catalog/my_schema/checkpoints/my_stream'
INSERT BY NAME
  -- The same source your original query read from.
  SELECT * FROM STREAM(source_table);

The checkpoint is imported only on the first pipeline update. Later updates ignore IMPORT CHECKPOINT. A full refresh does not re-import the checkpoint. It starts from a new, empty checkpoint and reprocesses the source.

Limitations

• Managed tables do not support CDC change flows. AUTO CDC INTO (SQL) or apply_changes / apply_changes_from_snapshot (Python) against a managed table fails with MANAGED_TABLE_DOES_NOT_SUPPORT_CDC. For CDC targets, use a CREATE STREAMING TABLE (pipelines) with a flow defined by CREATE FLOW (pipelines).
• Managed tables do not support FLOW ... REPLACE WHERE. Only FLOW INSERT BY NAME is supported. To recompute and overwrite a targeted subset of a table, create a streaming table and define a REPLACE WHERE flow with CREATE FLOW (pipelines). See Batch processing with REPLACE WHERE flows.
• Managed tables are supported only in pipelines with Unity Catalog. The Hive metastore is not supported.
• You cannot reuse the name of an existing streaming table for a managed table. Drop the streaming table first, or the statement fails with CANNOT_SWITCH_STREAMING_TABLE_TO_MANAGED_TABLE.
• IMPORT CHECKPOINT is supported only on FLOW INSERT BY NAME.

Examples

-- Create a managed table populated by an inline append flow from a streaming table
CREATE TABLE output
FLOW INSERT BY NAME SELECT * FROM STREAM(samples.tpch.orders);

-- Create a managed table that ingests files with schema inference and evolution
CREATE TABLE raw_data
FLOW INSERT BY NAME
  SELECT * FROM STREAM read_files('abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/base/path');

-- Create a partitioned managed table from a streaming source
CREATE TABLE events
PARTITIONED BY (bucket)
FLOW INSERT BY NAME
  SELECT id, bucket FROM STREAM read_files('abfss://my_path', format => 'json');

-- Create a managed table with liquid clustering
CREATE TABLE orders_clustered
CLUSTER BY (order_date, customer_id)
FLOW INSERT BY NAME
  SELECT
    o_orderkey   AS order_id,
    o_custkey    AS customer_id,
    o_orderdate  AS order_date,
    o_totalprice AS total_price
  FROM STREAM(samples.tpch.orders);

-- Create a managed table with a data quality expectation that drops violating rows
CREATE TABLE valid_events
  (CONSTRAINT positive_id EXPECT (id > 0) ON VIOLATION DROP ROW)
FLOW INSERT BY NAME
  SELECT id FROM STREAM read_files('s3://bucket/path', format => 'json');

Additional resources

• CREATE STREAMING TABLE (pipelines)
• CREATE FLOW (pipelines)
• CREATE MATERIALIZED VIEW (pipelines)
• Pipeline SQL language reference
• Lakeflow pipelines Python language reference