YQL queries to topics

To read and write messages to topics, 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 YDB database. The source and receiver 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 YDB database 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 YDB database.

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, accessing topic input_topic in an 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 conditional — 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 performed in table and streaming modes (not to be confused with streaming queries).

Table reading

In table 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 on Service fields speeds up reading, as reading occurs only over the specified ranges.

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

Streaming reading

To read new messages, use the WITH (STREAMING = "TRUE") option — see more in the Streaming reading of data from a topic section. 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 complete, 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 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 when messages are serialized in a known format (JSON, CSV, etc.). The FORMAT parameter specifies the parsing method, and SCHEMA specifies 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 on supported formats: Data formats for reading/writing from topics.

Using a 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.

Transferring 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            -- any transformations can be used
FROM
    ext_source.input_topic;  -- external topic; for local: input_topic

Service fields

When reading, you can request 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. Supported are comparison operators (=, <>, <, <=, >, >=, IN), logical conditions (AND, OR), and fields partition_id, write_time, offset. 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");

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 a 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

Warning

Reading and writing user attributes are not supported.