CREATE TABLE

CREATE TABLE defines a table that is persisted in durable storage.

In Materialize, you can create:

Read-write tables. With read-write tables, users can read (SELECT) and write to the tables (INSERT, UPDATE, DELETE).
Read-only tables from sources that use the new syntax: PostgreSQL, MySQL, SQL Server, and Kafka/Redpanda. Users cannot write (INSERT, UPDATE, DELETE) to these tables. These tables are populated by data ingestion from a source. You must be on v26+ to use the new syntax.

Syntax summary

To create a new read-write table (i.e., users can perform SELECT, INSERT, UPDATE, and DELETE operations):

CREATE [TEMP|TEMPORARY] TABLE [IF NOT EXISTS] <table_name> (
  <column_name> <column_type> [NOT NULL][DEFAULT <default_expr>]
  [, ...]
)
[WITH (
  PARTITION BY (<column_name> [, ...]) |
  RETAIN HISTORY [=] FOR <duration>
)]
;

For details, see CREATE TABLE: Read-write table.

To create a read-only table from a source connected (via native connector) to an external PostgreSQL:

CREATE TABLE [IF NOT EXISTS] <table_name> FROM SOURCE <source_name> (REFERENCE <upstream_table>)
[WITH (
    TEXT COLUMNS (<column_name> [, ...])
  | EXCLUDE COLUMNS (<column_name> [, ...])
  | PARTITION BY (<column_name> [, ...])
  [, ...]
)]
;

Creating the tables from sources starts the snapshotting process. Snapshotting syncs the currently available data into Materialize. Because the initial snapshot is persisted in the storage layer atomically (i.e., at the same ingestion timestamp), you are not able to query the table until snapshotting is complete.

NOTE: During the snapshotting, the data ingestion for the existing tables for the same source is temporarily blocked. As such, if possible, you can resize the cluster to speed up the snapshotting process and once the process finishes, resize the cluster for steady-state. You can monitor the snapshot progress on the overview page for the source in the Materialize console.

For details, see CREATE TABLE: PostgreSQL source table.

To create a read-only table from a source connected (via native connector) to an external MySQL database:

CREATE TABLE [IF NOT EXISTS] <table_name> FROM SOURCE <source_name> (REFERENCE <upstream_schema>.<upstream_table>)
[WITH (
    TEXT COLUMNS (<column_name> [, ...])
  | EXCLUDE COLUMNS (<column_name> [, ...])
  | PARTITION BY (<column_name> [, ...])
  [, ...]
)]
;

For details, see CREATE TABLE: MySQL source table.

To create a read-only table from a source connected (via native connector) to an external SQL Server database:

CREATE TABLE [IF NOT EXISTS] <table_name> FROM SOURCE <source_name> (REFERENCE <upstream_table>)
[WITH (
    TEXT COLUMNS (<column_name> [, ...])
  | EXCLUDE COLUMNS (<column_name> [, ...])
  | PARTITION BY (<column_name> [, ...])
  [, ...]
)]
;

For details, see CREATE TABLE: SQL Server source table.

Materialize can decode Avro messages by integrating with a schema registry to retrieve a schema, and automatically determine the columns and data types to use in the table.

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT AVRO
    USING CONFLUENT SCHEMA REGISTRY CONNECTION <csr_connection_name>
      [KEY STRATEGY <key_strategy>]
      [VALUE STRATEGY <value_strategy>]
  | USING AWS GLUE SCHEMA REGISTRY CONNECTION <glue_connection_name> (
      SCHEMA NAME = '<schema_name>'
    )
[INCLUDE
    KEY [AS <name>]
  | PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE
    NONE
  | DEBEZIUM
  | UPSERT [ ( VALUE DECODING ERRORS = INLINE [AS <name>] ) ]
];

Materialize can decode JSON messages into a single column named data with type jsonb. Refer to the jsonb type documentation for the supported operations on this type.

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT JSON
[INCLUDE
    PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE NONE];

Materialize can parse new-line delimited data as plain text, or read raw bytes without applying any formatting or decoding. Text-formatted tables have a single column, by default named text. Raw byte-formatted tables have a single column, by default named data.

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT TEXT | BYTES
[INCLUDE
    PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE NONE];

Materialize can parse CSV-formatted data. The data in CSV tables is read as text.

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT CSV WITH <n> COLUMNS [DELIMITED BY <char>]
[INCLUDE
    PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE NONE];

Materialize can decode Protobuf messages by integrating with a schema registry or parsing an inline schema to retrieve a .proto schema definition. It can then automatically define the columns and data types to use in the table.

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
FORMAT PROTOBUF USING CONFLUENT SCHEMA REGISTRY CONNECTION <csr_connection_name>
  | FORMAT PROTOBUF MESSAGE '<message_name>' USING SCHEMA '<schema_bytes>'
[INCLUDE
    KEY [AS <name>]
  | PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE
    NONE
  | UPSERT [ ( VALUE DECODING ERRORS = INLINE [AS <name>] ) ]
];

By default, the message key is decoded using the same format as the message value. However, you can set the key and value encodings explicitly using KEY FORMAT … VALUE FORMAT.

CREATE TABLE [IF NOT EXISTS] <table_name>
FROM SOURCE <src_name>
KEY FORMAT <key_format> VALUE FORMAT <value_format>
-- <key_format> and <value_format> can be:
-- AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION <conn_name>
--     [KEY STRATEGY <strategy>]
--     [VALUE STRATEGY <strategy>]
-- | AVRO USING AWS GLUE SCHEMA REGISTRY CONNECTION <glue_conn_name> (SCHEMA NAME = '<schema_name>')
-- | CSV WITH <num> COLUMNS DELIMITED BY <char>
-- | JSON | TEXT | BYTES
-- | PROTOBUF USING CONFLUENT SCHEMA REGISTRY CONNECTION <conn_name>
-- | PROTOBUF MESSAGE '<message_name>' USING SCHEMA '<schema_bytes>'
[INCLUDE
    KEY [AS <name>]
  | PARTITION [AS <name>]
  | OFFSET [AS <name>]
  | TIMESTAMP [AS <name>]
  | HEADERS [AS <name>]
  | HEADER '<key>' AS <name> [BYTES]
  [, ...]
]
[ENVELOPE
    NONE
  | DEBEZIUM
  | UPSERT [(VALUE DECODING ERRORS = INLINE [AS name])]
];

For details, see CREATE TABLE: Kafka source table.

CREATE TABLE

Syntax summary

Related pages