DDL

CREATE TABLE

Ingest Engine’s CREATE TABLE statements come in three flavors: Create Table As, Memory Table and Connection Table.

CREATE TABLE AS

This command creates a table from the query included in it, e.g.

CREATE TABLE orders AS SELECT customer_id, order_id FROM orders;

The schema is inferred from the query and other queries within the same context can select from the new table. For example, given a source table raw_orders with these records:

customer_id	order_id	product	amount
1	101	widget	25.00
2	102	gadget	49.99
1	103	widget	25.00

The following statement:

CREATE TABLE customer_totals AS
SELECT customer_id, count(*) as order_count, sum(amount) as total_spent
FROM raw_orders
GROUP BY customer_id, TUMBLE(INTERVAL '1' HOUR);

Produces a table customer_totals with the inferred schema (customer_id INT, order_count BIGINT, total_spent FLOAT) and contents:

customer_id	order_count	total_spent
1	2	50.00
2	1	49.99

CREATE TABLE (In-Memory)

CREATE TABLE statements without any connection info are presumed to be in memory. It can be written to within the same query context and then read from. A memory table may be used in only one INSERT INTO statement, but can then be selected from multiple times. For example, you could create an orders table with a statement like

CREATE TABLE orders (customer_id INT, order_id INT);

CREATE VIEW is simply an alias for creating a memory table.

CREATE TABLE (Connection)

Connection tables allow Ingest Engine to read and write to external systems like Kafka clusters. Connection tables may be used as sources or sinks depending on the type of connection. For details on all of the supported connectors, see the connector docs. Connection tables can be created via the Connections tabs of the Web UI, or directly in SQL via the CREATE TABLE statement. Connection tables are created via special CREATE TABLE statements that include a WITH clause. The WITH clause specifies the connector, the format that the data is encoded with, and various other options that are specific to the connector, as documented on the individual connector pages. The general form of the statement is:

CREATE TABLE <table name> (
  [<field name> <field type>,]*
  [WATERMARK FOR <field_name> [AS <expression>]]
) WITH (
  connector = '<connector name>',
  format = '<format name>',
  [format options]
  [connector options]
)

where connector is one of the supported connectors and format is one of the supported formats. For example, to create a Kafka source for the topic order_topic:

CREATE TABLE orders (
  customer_id INT,
  order_id INT,
  date_string TEXT
) WITH (
  connector = 'kafka',
  format = 'json',
  type = 'source',
  bootstrap_servers = 'localhost:9092',
  topic = 'order_topic'
);

For full details on how to create connection tables, see the connector docs.

Schema inference

When creating a connection table, you can specify the schema explicitly by listing fields in the CREATE TABLE statement, or you can let Ingest Engine infer the schema from how it’s used. This is mostly relevant for sinks, where the schema can be inferred from the query that writes to the table. For example, a Kafka sink could be created like this

CREATE TABLE results WITH (
  connector = 'kafka',
  format = 'json',
  type = 'sink',
  bootstrap_servers = 'localhost:9092',
  topic = 'results'
);

and written to like this

INSERT INTO results
SELECT customer_id as customer_id, count(*) as count
FROM orders
GROUP BY customer_id, TUMBLE(INTERVAL '1' HOUR);

That will result in records like this being written to the results topic

{"customer_id": 1, "count": 10}
{"customer_id": 2, "count": 5}

To illustrate the full flow: given these records arriving on the order_topic source:

customer_id	order_id	date_string
1	101	2024-01-15
1	102	2024-01-15
2	201	2024-01-15

The INSERT INTO results query above infers the sink schema as (customer_id INT, count BIGINT) and writes:

customer_id	count
1	2
2	1

Schema-inferred sinks can also be created in the Web UI by selecting the “infer schema” option when creating the table. Note that when relying on schema inference, the column names will be determined by the query, so you will generally want to alias them using as to ensure they are what you expect.

Options

Connection tables allow you to configure a number of options that specify and modify the behavior. They are specified via the WITH clause, with an unquoted key and a single or double-quoted value. The following options are supported across all connections. Specific connections have their own options. To see all of the supported options, refer to the connector docs.

Option	required	Description
`connector`	yes	The name of the connector to use.
`format`	if connector does not have a built-in format	The format of the data to be deserialized.
`idle_micros`	no	The number of microseconds to wait before considering a source idle. Defaults to `30000000` (30 seconds). Set to `-1` to disable source idleness.

Virtual Fields

Virtual fields can be created within the CREATE TABLE statement. These are done using the GENERATED ALWAYS AS (expression) syntax. expression must be a valid Ingest Engine SQL expression that only depends on non-virtual fields within the table. For example

CREATE TABLE events (
  id TEXT,
  event_type TEXT,
  user_id TEXT,
  key TEXT GENERATED ALWAYS AS (concat(id, '-', user_id))
) WITH (
  connector = 'kafka',
  format = 'json',
  type = 'source',
  bootstrap_servers = 'localhost:9092',
  topic = 'events'
);

Given these raw events from Kafka:

id	event_type	user_id
evt1	click	alice
evt2	purchase	bob
evt3	pageview	alice

The virtual key field is computed automatically:

id	event_type	user_id	key
evt1	click	alice	evt1-alice
evt2	purchase	bob	evt2-bob
evt3	pageview	alice	evt3-alice

INSERT INTO

Ingest Engine supports INSERT INTO statements for both memory and connection tables. In line with standard SQL the insertion will happen column-wise, attempting coercion to the SQL types. For example, if you have a memory table orders with columns customer_id and order_id you could insert into it with a statement like

INSERT INTO orders SELECT customer, order FROM source_table;

If the table is a connection table this will result in a sink, otherwise it will be a memory table that can then be read from.

Ingest Engine

Get started

Connectors

Streaming SQL

Operate

Reference

Concepts

CREATE TABLE

CREATE TABLE AS

CREATE TABLE (In-Memory)

CREATE TABLE (Connection)

Schema inference

Options

Virtual Fields

INSERT INTO

​CREATE TABLE

​CREATE TABLE AS

​CREATE TABLE (In-Memory)

​CREATE TABLE (Connection)

​Schema inference

​Options

​Virtual Fields

​INSERT INTO

CREATE TABLE

CREATE TABLE AS

CREATE TABLE (In-Memory)

CREATE TABLE (Connection)

Schema inference

Options

Virtual Fields

INSERT INTO