# Quick Reference for ksqlDB for Confluent Platform This quick reference summarizes ksqlDB SQL statements, keywords, and operators in condensed form for fast lookup. For detailed descriptions of each statement and keyword, see the [ksqlDB API reference](overview.md#ksqldb-api-index). For details on SQL syntax, see the [ksqlDB syntax reference](../../reference/sql/lexical-structure.md#ksqldb-lexical-structure). ## ADVANCE BY Specify the duration of a “hop” in a HOPPING window. For more information, see [Hopping window](../../concepts/time-and-windows-in-ksqldb-queries.md#ksqldb-time-and-windows-hopping-window). ```sql SELECT [...], aggregate_function WINDOW HOPPING (SIZE , ADVANCE BY ) [...] ``` ## ALTER SYSTEM Change a system-level property value (only available in Confluent Cloud). ```sql ALTER SYSTEM 'auto.offset.reset'='earliest'; ``` ## ALTER STREAM Add new columns to a stream. This is not supported for streams defined using queries (`CREATE STREAM ... AS`). ```sql ALTER STREAM stream_name ADD [COLUMN] column_name data_type ADD [COLUMN] ... ... ... ``` ## ALTER TABLE Add new columns to a table. This is not supported for tables defined using queries (`CREATE TABLE ... AS`) ```sql ALTER TABLE stream_name ADD [COLUMN] column_name data_type ADD [COLUMN] ... ... ... ``` ## AND / OR Logical AND/OR operators in a WHERE clause. For more information, see [SELECT](select-push-query.md#ksqldb-reference-select-push-query-examples). ```none SELECT column_name(s) FROM stream_name | table_name WHERE condition AND|OR condition ``` ## AS Alias a column, expression, or type. For more information, see [Create a table](create-table-as-select.md#ksqldb-reference-create-table-as-select-examples). ```none SELECT column_name AS column_alias FROM stream_name | table_name ``` ## ASSERT Assert values, stream, table, or tombstones. ```sql ASSERT NULL VALUES sourceName (columns)? KEY values ``` ## ASSERT SCHEMA Assert the existence or non-existence of a schema. ```sql ASSERT (NOT EXISTS)? SCHEMA (SUBJECT subjectName)? (ID id)? (TIMEOUT timeout)?; ``` ## ASSERT TOPIC Assert the existence or non-existence of a topic. ```sql ASSERT (NOT EXISTS)? TOPIC topicName (WITH properties)? (TIMEOUT timeout)?; ``` ## BETWEEN Constrain a value to a specified range in a WHERE clause. ```sql WHERE expression [NOT] BETWEEN start_expression AND end_expression; ``` The BETWEEN operator is used to indicate that a certain value must be within a specified range, including boundaries. ksqlDB supports any expression that resolves to a numeric or string value for comparison. The following push query uses the BETWEEN clause to select only records that have an `event_id` between 10 and 20. ```sql SELECT event FROM events WHERE event_id BETWEEN 10 AND 20 EMIT CHANGES; ``` ## CASE Select a condition from one or more expressions. ```sql SELECT CASE WHEN condition THEN result [ WHEN … THEN … ] … [ WHEN … THEN … ] [ ELSE result ] END FROM stream_name | table_name; ``` ksqlDB supports a `searched` form of CASE expression. In this form, CASE evaluates each boolean `condition` in WHEN clauses, from left to right. If a condition is true, CASE returns the corresponding result. If none of the conditions is true, CASE returns the result from the ELSE clause. If none of the conditions is true and there is no ELSE clause, CASE returns null. The schema for all results must be the same, otherwise ksqlDB rejects the statement. The following example push query uses a CASE expression. ```sql SELECT CASE WHEN orderunits < 2.0 THEN 'small' WHEN orderunits < 4.0 THEN 'medium' ELSE 'large' END AS case_result FROM orders EMIT CHANGES; ``` ## CAST Change the type of an expression to a different type. ```sql CAST (expression AS data_type); ``` You can cast an expression’s type to a new type by using CAST. The following example query converts a numerical count, which is a BIGINT, into a suffixed string, which is a VARCHAR. For example, the integer `5` becomes `5_HELLO`. ```sql SELECT page_id, CONCAT(CAST(COUNT(*) AS VARCHAR), '_HELLO') FROM pageviews_enriched WINDOW TUMBLING (SIZE 20 SECONDS) GROUP BY page_id; ``` ## CREATE CONNECTOR Create a new connector in the Kafka Connect cluster. For more information, see [CREATE CONNECTOR](create-connector.md#ksqldb-reference-create-connector). ```sql CREATE SOURCE | SINK CONNECTOR connector_name WITH( property_name = expression [, ...]); ``` ## CREATE STREAM Register a stream on a Kafka topic. For more information, see [CREATE STREAM](create-stream.md#ksqldb-reference-create-stream). ```sql CREATE STREAM stream_name ( { column_name data_type [KEY] } [, ...] WITH ( property_name = expression [, ...] ); ``` ## CREATE STREAM AS SELECT Create a new materialized stream and corresponding Kafka topic, and stream the result of the query into the topic. For more information, see [CREATE STREAM AS SELECT](create-stream-as-select.md#ksqldb-reference-create-stream-as-select). ```sql CREATE STREAM stream_name [WITH ( property_name = expression [, ...] )] AS SELECT select_expr [, ...] FROM from_stream [[ LEFT | FULL | INNER ] JOIN [join_table | join_stream] [WITHIN [ | ( , )] [GRACE PERIOD ]] ON join_criteria]* [ WHERE condition ] [PARTITION BY new_key_expr [, ...]] EMIT CHANGES; ``` ## CREATE TABLE Register a stream on a Kafka topic. For more information, see [CREATE TABLE](create-table.md#ksqldb-reference-create-table). ```sql CREATE TABLE table_name ( { column_name data_type (PRIMARY KEY) } [, ...] ) WITH ( property_name = expression [, ...] ); ``` ## CREATE TABLE AS SELECT Create a new materialized table and corresponding Kafka topic, and stream the result of the query as a changelog into the topic. For more information, see [CREATE TABLE AS SELECT](create-table-as-select.md#ksqldb-reference-create-table-as-select-examples). ```sql CREATE TABLE table_name [WITH ( property_name = expression [, ...] )] AS SELECT select_expr [, ...] FROM from_stream | from_table [[ LEFT | FULL | INNER ] JOIN [join_table | join_stream] ON join_criteria]* [ WINDOW window_expression ] [ WHERE condition ] [ GROUP BY grouping_expression ] [ HAVING having_expression ] EMIT CHANGES; ``` ## CREATE TYPE Alias a complex type declaration. For more information, see [CREATE TYPE](create-type.md#ksqldb-reference-create-type). ```sql CREATE TYPE AS ; ``` ## DEFINE Defines a variable. For more information, see [DEFINE](define.md#ksqldb-reference-define). ```sql DEFINE = ''; ``` ## DESCRIBE List columns in a stream or table along with their data types and other attributes. For more information, see [DESCRIBE](describe.md#ksqldb-reference-describe). ```sql DESCRIBE [EXTENDED] (stream_name | table_name); ``` ## DESCRIBE CONNECTOR List details about a connector. For more information, see [DESCRIBE CONNECTOR](describe-connector.md#ksqldb-reference-describe-connector). ```sql DESCRIBE CONNECTOR connector_name; ``` ## DESCRIBE FUNCTION List details about a function, including input parameters and return type. For more information, see [DESCRIBE FUNCTION](describe-function.md#ksqldb-reference-describe-function). ```sql DESCRIBE FUNCTION function_name; ``` ## DROP CONNECTOR Delete a connector from the Connect cluster. For more information, see [DROP CONNECTOR](drop-connector.md#ksqldb-reference-drop-connector). ```sql DROP CONNECTOR connector_name; ``` ## DROP STREAM Drop an existing stream and optionally mark the stream’s source topic for deletion. For more information, see [DROP STREAM](drop-stream.md#ksqldb-reference-drop-stream). ```sql DROP STREAM [IF EXISTS] stream_name [DELETE TOPIC]; ``` ## DROP TABLE Drop an existing table and optionally mark the table’s source topic for deletion. For more information, see [DROP TABLE](drop-table.md#ksqldb-reference-drop-table). ```sql DROP TABLE [IF EXISTS] table_name [DELETE TOPIC]; ``` ## DROP TYPE Remove a type alias from ksqlDB. For more information, see [DROP TYPE](drop-type.md#ksqldb-reference-drop-type). ```sql DROP TYPE [IF EXISTS] AS ; ``` ## EMIT CHANGES Specify a push query with a continuous output refinement in a SELECT statement. For more information, see [Push Queries](../../concepts/queries.md#ksqldb-concepts-queries-push). ```sql CREATE STREAM stream_name AS SELECT select_expr [, ...] FROM from_stream EMIT CHANGES; ``` ## EMIT FINAL Specify a push query with a suppressed output refinement in a SELECT statement on a windowed aggregation. For more information, see [Push Queries](../../concepts/queries.md#ksqldb-concepts-queries-push). ```sql CREATE TABLE table_name AS SELECT select_expr_with_aggregation [, ...] FROM from_stream [ WINDOW window_expression ] [ GROUP BY grouping_expression ] EMIT FINAL; ``` ## EXPLAIN Show the execution plan for a SQL expression or running query. For more information, see [EXPLAIN](explain.md#ksqldb-reference-explain). ```sql EXPLAIN (sql_expression | query_id); ``` ## FULL JOIN Select all records when there is a match in the left stream/table *or* the right stream/table records. Equivalent to FULL OUTER JOIN. For more information, see [Join streams and tables](../joins/join-streams-and-tables.md#ksqldb-joins-join-event-streams). ```none SELECT column_name(s) FROM stream_name1 | table_name1 FULL JOIN stream_name2 | table_name2 ON .column_name=.column_name ``` ## GRACE PERIOD Allow events to be accepted for a time period after a window ends. For more information, see [Out-of-order events](../../concepts/time-and-windows-in-ksqldb-queries.md#ksqldb-time-and-windows-out-of-order-events) ```none SELECT orderzip_code, TOPK(order_total, 5) FROM orders WINDOW TUMBLING (SIZE 1 HOUR, GRACE PERIOD 2 HOURS) GROUP BY order_zipcode EMIT CHANGES; ``` ## GROUP BY Group records in a window. Required by the WINDOW clause. Windowing queries must group by the keys that are selected in the query. For more information, see [Time and Windows in ksqlDB](../../concepts/time-and-windows-in-ksqldb-queries.md#ksqldb-time-and-windows-in-queries). ```none SELECT column_name, aggregate_function(column_name) FROM table_name WHERE column_name operator value GROUP BY column_name ``` ## HAVING Extract records from an aggregation that fulfill a specified condition. ```none SELECT column_name, aggregate_function(column_name) FROM table_name WHERE column_name operator value GROUP BY column_name HAVING aggregate_function(column_name) operator value ``` ## HEADER Populate a column with the Kafka record’s last header that matches the key. ```sql CREATE STREAM S (column_name BYTES HEADER('key')) WITH (kafka_topic='s', format='json'); ``` ## HEADERS Populate a column with the full list of the Kafka record’s headers. ```sql CREATE STREAM S (column_name ARRAY> HEADERS) WITH (kafka_topic='s', format='json'); ``` ## HOPPING Group input records into fixed-sized, possibly overlapping windows, based on the timestamps of the records. For more information, see [HOPPING](select-push-query.md#ksqldb-reference-select-push-query-hopping-window). ```none SELECT WINDOWSTART, WINDOWEND, aggregate_function FROM from_stream WINDOW HOPPING window_expression EMIT CHANGES; ``` ## IF EXISTS Test whether a stream or table is present in ksqlDB. ```sql DROP STREAM [IF EXISTS] stream_name [DELETE TOPIC]; DROP TABLE [IF EXISTS] table_name [DELETE TOPIC]; ``` ## IN Specifies multiple `OR` conditions. ```none SELECT select_expr [, ...] FROM from_stream | from_table WHERE exp IN (exp0, exp1, exp2); ``` The above is equivalent to: ```none SELECT select_expr [, ...] FROM from_stream | from_table WHERE exp = exp0 OR exp = exp1 OR exp = exp2; ``` ## INNER JOIN Select records in a stream or table that have matching values in another stream or table. For more information, see [Join streams and tables](../joins/join-streams-and-tables.md#ksqldb-joins-join-event-streams). ```none SELECT column_name(s) FROM stream_name1 | table_name1 INNER JOIN stream_name2 | table_name2 ON .column_name=.column_name ``` ## INSERT INTO Stream the result of a SELECT query into an existing stream and its underlying Kafka topic. For more information, see [INSERT INTO](insert-into.md#ksqldb-reference-insert-into). ```sql INSERT INTO stream_name SELECT select_expr [, ...] FROM from_stream [ LEFT | FULL | INNER ] JOIN [join_table | join_stream] [WITHIN [ | ( , )] [GRACE PERIOD ]] ON join_criteria [ WHERE condition ] [ PARTITION BY new_key_expr [, ...] ] EMIT CHANGES; ``` ## INSERT VALUES Produce a row into an existing stream or table and its underlying Kafka topic based on explicitly specified values. For more information, see [INSERT VALUES](insert-values.md#ksqldb-reference-insert-values). ```sql INSERT INTO stream_name|table_name [(column_name [, ...]])] VALUES (value [,...]); ``` ## LEFT JOIN Select all records from the left stream/table and the matched records from the right stream/table. For more information, see [Join streams and tables](../joins/join-streams-and-tables.md#ksqldb-joins-join-event-streams). ```none SELECT column_name(s) FROM stream_name1 | table_name1 LEFT JOIN stream_name2 | table_name2 ON .column_name=.column_name ``` ## LIKE Match a string with the specified pattern. ```none SELECT select_expr [, ...] FROM from_stream | from_table WHERE exp LIKE pattern_string; ``` The LIKE operator is used for prefix or suffix matching. ksqlDB supports the `%` wildcard, which represents zero or more characters. The following push query uses the `%` wildcard to match any `user_id` that starts with “santa”. ```sql SELECT user_id FROM users WHERE user_id LIKE 'santa%' EMIT CHANGES; ``` ## NULLIF Returns NULL if two expressions are equal, otherwise it returns the first expression. ```none SELECT NULLIF(col1, col2) FROM from_stream | from_table EMIT CHANGES; ``` ## PARTITION BY Repartition a stream. For more information, see [Partition Data to Enable Joins](../joins/partition-data.md#ksqldb-joins-partition-data). ```none CREATE STREAM stream_nam WITH ([...,] PARTITIONS=number_of_partitions) AS SELECT select_expr [, ...] FROM from_stream PARTITION BY new_key_expr [, ...] EMIT CHANGES; ``` ## PAUSE Pause a persistent query. For more information, see [PAUSE](pause.md#ksqldb-reference-pause). ```sql PAUSE query_id; ``` ## PRINT Print the contents of Kafka topics to the ksqlDB CLI. For more information, see [PRINT](print.md#ksqldb-reference-print). ```sql PRINT topicName [FROM BEGINNING] [INTERVAL interval] [LIMIT limit] ``` ## RESUME End a paused persistent query. For more information, see [RESUME](resume.md#ksqldb-reference-resume). ```sql RESUME query_id; ``` ## RIGHT JOIN Select all records from the right stream/table and the matched records from the left stream/table. For more information, see [Join streams and tables](../joins/join-streams-and-tables.md#ksqldb-joins-join-event-streams). ```none SELECT column_name(s) FROM stream_name1 | table_name1 RIGHT JOIN stream_name2 | table_name2 ON .column_name=.column_name ``` ## RUN SCRIPT Execute predefined queries and commands from a file. For more information, see [RUN SCRIPT](run-script.md#ksqldb-reference-run-script). ```sql RUN SCRIPT ; ``` ## SELECT (pull query) Pull the current value from a materialized table and terminate. For more information, see [SELECT (Pull Query)](select-pull-query.md#ksqldb-reference-select-pull-query). ```sql SELECT select_expr [, ...] FROM from_item [ WHERE where_condition ] [ AND window_bounds ] [ LIMIT count ]; ``` ## SELECT (push query) Push a continuous stream of updates to a stream or table. For more information, see [SELECT (Push Query)](select-push-query.md#ksqldb-reference-select-push-query). ```sql SELECT select_expr [, ...] FROM from_item [[ LEFT | FULL | INNER ] JOIN join_item [WITHIN [ | ( , )] [GRACE PERIOD ]] ON join_criteria]* [ WINDOW window_expression ] [ WHERE condition ] [ GROUP BY grouping_expression [, ...] ] [ HAVING having_expression ] EMIT CHANGES [ LIMIT count ]; ``` ## SESSION Group input records into a session window. For more information, see [SELECT (Push Query)](select-push-query.md#ksqldb-reference-select-push-query-session-window). ```none SELECT WINDOWSTART, WINDOWEND, aggregate_function FROM from_stream WINDOW SESSION window_expression EMIT CHANGES; ``` ## SET property Assign a property value. ```sql SET 'auto.offset.reset'='earliest'; ``` ## SHOW CONNECTORS List all connectors in the Connect cluster. For more information, see [SHOW CONNECTORS](show-connectors.md#ksqldb-reference-show-connectors). ```sql SHOW | LIST CONNECTORS; ``` ## SHOW FUNCTIONS List available scalar and aggregate functions available. For more information, see [SHOW FUNCTIONS](show-functions.md#ksqldb-reference-show-functions). ```sql SHOW | LIST FUNCTIONS; ``` ## SHOW PROPERTIES List the [configuration settings](../../reference/server-configuration.md#ksqldb-reference-server-configuration) that are currently in effect. For more information, see [SHOW PROPERTIES](show-properties.md#ksqldb-reference-show-properties). ```sql SHOW PROPERTIES; ``` ## SHOW QUERIES List queries that are currently running in the cluster. For more information, see [SHOW QUERIES](show-queries.md#ksqldb-reference-show-queries). ```sql SHOW | LIST QUERIES [EXTENDED]; ``` ## SHOW STREAMS List the currently defined streams. For more information, see [SHOW STREAMS](show-streams.md#ksqldb-reference-show-streams). ```sql SHOW | LIST STREAMS [EXTENDED]; ``` ## SHOW TABLES List the currently defined tables. For more information, see [SHOW TABLES](show-tables.md#ksqldb-reference-show-tables). ```sql SHOW | LIST TABLES [EXTENDED]; ``` ## SHOW TOPICS List the available topics in the Kafka cluster that ksqlDB is configured to connect to. For more information, see [SHOW TOPICS](show-topics.md#ksqldb-reference-show-topics). ```sql SHOW | LIST [ALL] TOPICS [EXTENDED]; ``` ## SHOW TYPES List all custom types and their type definitions. For more information, see [SHOW TYPES](show-types.md#ksqldb-reference-show-types). ```sql SHOW | LIST TYPES; ``` ## SHOW VARIABLES List all defined variables. ```sql SHOW VARIABLES; ``` ## SIZE Specify the duration of a HOPPING or TUMBLING window. For more information, see [Time and Windows in ksqlDB](../../concepts/time-and-windows-in-ksqldb-queries.md#ksqldb-time-and-windows). ```none SELECT WINDOWSTART, WINDOWEND, aggregate_function FROM from_stream WINDOW TUMBLING (SIZE ) EMIT CHANGES; ``` ## SPOOL Store issued commands and their results in a file. For more information, see [SPOOL](spool.md#ksqldb-reference-spool). ```sql SPOOL ``` ## TERMINATE End a query. For more information, see [TERMINATE](terminate.md#ksqldb-reference-terminate). ```sql TERMINATE query_id; ``` ## TUMBLING Group input records into fixed-sized, non-overlapping windows based on the timestamps of the records. For more information, see [TUMBLING](select-push-query.md#ksqldb-reference-select-push-query-tumbling-window). ```none SELECT WINDOWSTART, WINDOWEND, aggregate_function FROM from_stream WINDOW TUMBLING window_expression EMIT CHANGES; ``` ## UNDEFINE Undefines a variable. ```sql UNDEFINE name; ``` ## UNSET property Unassign a property value. ```sql UNSET 'auto.offset.reset'; ``` ## WHERE Extract records that fulfill a specified condition. For more information, see [SELECT](select-push-query.md#ksqldb-reference-select-push-query-examples). ```none SELECT column_name(s) FROM from_stream | from_table WHERE column_name operator value ``` ## WINDOW Group input records that have the same key into a window, for operations like aggregations and joins. For more information, see [WINDOW](select-push-query.md#ksqldb-reference-select-push-query-window). ```none SELECT WINDOWSTART, WINDOWEND, aggregate_function FROM from_stream WINDOW window_expression EMIT CHANGES; ``` ## WINDOWSTART / WINDOWEND Specify the beginning and end bounds of a window. For more information, see [WINDOW](select-push-query.md#ksqldb-reference-select-push-query-window). ```none SELECT WINDOWSTART, WINDOWEND, aggregate_function FROM from_stream WINDOW window_expression EMIT CHANGES; ```