YQL queries to topics

For reading and writing messages to topics, the familiar YQL constructs are used: SELECT for reading and INSERT for writing.

Local and external topics

YQL queries to topics work the same regardless of whether the topic is in the current database or in another database YDB. The source and destination of messages can be either a topic in the same database where the query is executed, or a topic in another database.

Local topics

Local topics are topics created in the same database YDB as the query being executed.

In the query text, they are referred to by a short name — just like a table in the current database:

SELECT * FROM input_topic WITH (FORMAT = json_each_row, SCHEMA = (...));

INSERT INTO output_topic SELECT ...;

External topics

External topics are topics located in another database YDB.

Access to them is performed only through a pre-created external data source with the YDB source type.

After creating a source, for example named ext_source, access to topic input_topic in the external database is written as follows:

SELECT * FROM ext_source.input_topic WITH (FORMAT = json_each_row, SCHEMA = (...));

The name ext_source in the documentation is conventional — in your database, the source may be named differently; it is important that it matches in CREATE EXTERNAL DATA SOURCE and in the prefix before the topic name.

Reading from a topic

Reading from a topic can be limited to only the topic's current data, or it can wait for new written messages.

Reading current data

In this mode, reading is performed from the first to the last offset stored in the topic at the time the query is started. If data continues to be written to the topic, the query will stop after reaching the last offset known at startup. Specifying filters by service fields speeds up reading, as reading is performed only over the specified ranges.

SELECT
    Data    -- message body
FROM
    input_topic  -- local topic; for external: ext_source.input_topic
LIMIT 10;

Reading with data waiting

To wait for new messages, use the WITH (STREAMING = "TRUE") option. Reading starts from the current moment and continues until the number of messages specified in the LIMIT expression is read. The LIMIT parameter is required, without it the query will not finish, as it will wait for new messages indefinitely.

SELECT
    Data
FROM
    ext_source.input_topic  -- external topic; for local: input_topic
WITH (STREAMING = "TRUE")
LIMIT 10;

For continuous processing of incoming data, use streaming queries.

Message format and schema

When reading from a topic, the message body can be obtained in two ways: raw data and formatted data.

Raw data

Use it when the message content does not need to be parsed; it is enough to read the body as is.

SELECT
    Data
FROM
    input_topic  -- local topic; for external: ext_source.input_topic
WITH (
    FORMAT = raw,
    SCHEMA = (
        Data String
    )
)
LIMIT 10;

As a result, only the Data column is available: the message body in its original form.

The same result can be obtained without the WITH block: see table reading.

Formatted data

Use it when messages are serialized in a known format (JSON, CSV, etc.). The FORMAT parameter sets the parsing method, and SCHEMA sets the names and types of fields that will appear in the SELECT result:

SELECT
    Id,
    Name
FROM
    input_topic  -- local topic; for external: ext_source.input_topic
WITH (
    FORMAT = json_each_row,
    SCHEMA = (
        Id Uint64 NOT NULL,
        Name Utf8 NOT NULL
    )
);

Fields from SCHEMA are available in SELECT by name, like table columns.

For more details about supported formats: Data formats for reading/writing from topics.

Using the reader

A consumer is a named subscription to a topic that stores the current read position.

A consumer is created via the CLI or when creating a topic using CREATE TOPIC. The consumer name is specified in the query text with a pragma:

PRAGMA pq.Consumer="my_consumer";

If a consumer is not specified, reading from the topic is performed without one. Specifying a consumer allows tracking the read position and lag from the topic side, for example via the CLI.

Moving data from a topic to a table via UPSERT

Data from a topic can be moved to a table via UPSERT INTO:

UPSERT INTO
    table_name
SELECT
    Data            -- you can use any transformations
FROM
    ext_source.input_topic;  -- external topic; for local: input_topic

Service fields

When reading, you can request service fields and user-defined message attributes:

Example of using service fields:

SELECT
    Data,                                                   -- message body
    __ydb_create_time AS CreateTime,                        -- message creation time
    __ydb_write_time AS WriteTime,                          -- message write time
    __ydb_offset AS Offset,                                 -- message offset in topic
    __ydb_partition_id AS Partition,                        -- partition number
    __ydb_message_group_id AS MessageGroupId,               -- message group identifier
    __ydb_seq_no AS SeqNo                                   -- sequence number within partition
FROM
    input_topic  -- local topic; for external: ext_source.input_topic
LIMIT 10;

Filters on service fields are evaluated before reading data from the topic and significantly reduce the volume of messages read. Comparison operators (=, <>, <, <=, >, >=, IN), logical conditions (AND, OR), and fields partition_id, write_time, offset are supported. Predicates on other service fields do not limit the read volume.

SELECT
    Data
FROM
    ext_source.input_topic  -- external topic; for local: input_topic
WHERE
    __ydb_partition_id = 42
        AND __ydb_offset >= 1000
        AND __ydb_offset <= 1100
        AND __ydb_write_time > CurrentUtcTimestamp() - Interval("PT2H");

Example of using user attributes:

SELECT
    COUNT(*) AS ErrorCount
FROM
    input_topic  -- local topic; for external: ext_source.input_topic
WHERE
    __ydb_user_attributes["type"] = "log"
        AND __ydb_user_attributes["level"] = "error";

Writing to a topic

Writing a single message

INSERT INTO
    output_topic  -- local topic; for external: ext_source.output_topic
SELECT
    "my_message";       -- message body

Writing data from a table

To write a table row with multiple columns to a topic, create a JSON object. The TableRow function creates a structure from all columns, Yson::From converts it to Yson, Yson::SerializeJson serializes it to a JSON string, and ToBytes converts the result to the String type required for writing to the topic:

INSERT INTO
    ext_source.output_topic  -- external topic; for local: output_topic
SELECT
    ToBytes(Unwrap(Yson::SerializeJson(Yson::From(TableRow()))))
FROM
    table_name;

Limitations

See also

• CREATE TOPIC
• ALTER TOPIC
• DROP TOPIC
• Streaming queries