--- title: "Query topics as tables" slug: "dev-event-broker-guides-querying-topics" description: "Learn how to read Kafka events as VAST DataBase tables using Python SDK, SQL, Trino, and Spark, and understand their structure and limitations." updated: 2026-10-30T18:18:46Z published: 2026-10-30T18:18:46Z canonical: "docs.vastdata.com/dev-event-broker-guides-querying-topics" --- > ## Documentation Index > Fetch the complete documentation index at: https://docs.vastdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Query topics as tables *Applies to VAST Cluster 5.5. Examples use the VAST DataBase Python SDK 2.1.0.* Every VAST Event Broker topic is also a VAST DataBase table. You can read the events that Kafka producers wrote with the VAST DataBase Python SDK, with SQL through the VAST Query Engine, or with Trino and Spark, without a Kafka consumer. This guide shows where the tables are, what their columns hold, and how to read them. It also covers what you can't do with them. Run the code blocks in order in one Python session. Below each block is the output you should see. ## Where topics live When your platform team creates a Kafka view, the cluster creates a VAST DataBase database for it: the view's bucket. Each topic is a table in that bucket's `kafka_topics` schema, with the same name as the topic ([Managing Event Topics](/docs-5-5/docs/publishing-events-to-vast-event-broker#managing-event-topics)). So topic `orders` on a Kafka view with bucket `events` is the table: | Client | Name | | --- | --- | | Python SDK | `tx.bucket("events").schema("kafka_topics").table("orders")` | | VAST Query Engine (ADBC) | `"events/kafka_topics"."orders"` | | Trino | `vast."events/kafka_topics"."orders"` | Each partition is stored separately inside the table, but you query the table as a whole. ![The same topic table can be read two ways. A Kafka consumer reads each partition in order, from its group's committed offset. The VAST DataBase SDK, the VAST Query Engine, Trino or Spark read the topic as a table: filters such as a time window on timestamp_broker and column lists run on the cluster, and aggregations run in the engine. key and value are bytes; there are no partition or offset columns. For payload fields, blob expansion writes typed columns to a separate table.](https://cdn.us.document360.io/89b91113-5dc6-49db-86ec-57b16dc8ce56/Images/Documentation/eb-two-ways.svg?sv=2026-02-06&spr=https&st=2026-11-08T08%3A55%3A56Z&se=2026-11-08T09%3A55%3A56Z&sr=c&sp=r&sig=vVcppmSZVXWCkV%2BB4UDYG6pBKGs%2BxBUGkM7CN4Hsze4%3D) *The same topic, read by a Kafka consumer or queried as a table.* ## Columns Each row of a topic table is one event. The columns are: | Column | Type | Holds | | --- | --- | --- | | `key` | binary | The record key, as the producer sent it | | `value` | binary | The record value (the payload) | | `timestamp_broker` | timestamp (ms) | When the broker stored the event | | `timestamp_producer` | timestamp (ms) | The timestamp the producer set on the record | | `headers` | list of struct (`key` string, `value` binary) | The record headers, in order. Keys can repeat. | | `external_offset`, `compaction_process_seq_num`, `extra_metadata` | int64, int64, binary | Used by the cluster. Empty (null) for events produced by Kafka clients. | `key` and `value` are raw bytes. The cluster doesn't parse them, so decode them in your code. To get typed columns from JSON values, set up [blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion). There's no partition column and no offset column: you can't query a topic by partition ([Kafka Protocol Support](/docs-5-5/docs/publishing-events-to-vast-event-broker#kafka-protocol-support)). The cluster uses an event's row ID in its partition as the Kafka offset, but the row ID alone doesn't tell you the partition. If you need the partition and offset as columns, blob expansion adds them to its target table. ## Before you start You need: - A topic to read, and the bucket of its Kafka view. This guide creates its own topic with a Kafka client; see [Topics](/develop/docs/dev-event-broker-guides-topics). - For the Kafka client: the bootstrap servers, and credentials if the view uses SASL/PLAIN. See [Setup](/develop/docs/dev-event-broker-operations-setup). - For the SDK: an S3 endpoint and an access key pair for a user with read access (the `s3:Tabular*` read actions) on the Kafka view's bucket. See [setup prerequisites](/develop/docs/dev-database-operations-setup-prerequisites). - For SQL: a Query Engine endpoint and `s3:QueryEngine*`. See [VAST Query Engine (ADBC)](/develop/docs/dev-database-integrations-adbc#before-you-start). Set these environment variables: ``` export KAFKA_BOOTSTRAP=":9092" export KAFKA_USERNAME="" # only if the view uses SASL/PLAIN export KAFKA_PASSWORD="" # only if the view uses SASL/PLAIN export AWS_S3_ENDPOINT_URL="http://" export AWS_ACCESS_KEY_ID="" export AWS_SECRET_ACCESS_KEY="" export EVENTS_BUCKET="" export QE_ENDPOINT="http://" ``` Install the clients: ``` pip install "vastdb[adbc]==2.1.0" "confluent-kafka>=2.4,<2.9" ``` ## Set up Create a topic, produce three JSON events to it, and connect the SDK: ``` import json import os from datetime import datetime, timedelta, timezone import vastdb from confluent_kafka import Producer from confluent_kafka.admin import AdminClient, NewTopic EVENTS_BUCKET = os.environ["EVENTS_BUCKET"] TOPIC = "docs-querying-topics-orders" def kafka_config(): cfg = {"bootstrap.servers": os.environ["KAFKA_BOOTSTRAP"]} if os.environ.get("KAFKA_USERNAME"): cfg.update({ "security.protocol": "SASL_PLAINTEXT", "sasl.mechanism": "PLAIN", "sasl.username": os.environ["KAFKA_USERNAME"], "sasl.password": os.environ["KAFKA_PASSWORD"], }) return cfg admin = AdminClient(kafka_config()) admin.create_topics([NewTopic(TOPIC, num_partitions=3, replication_factor=1)])[TOPIC].result() producer = Producer({**kafka_config(), "compression.type": "none", "enable.idempotence": False, "acks": "all"}) orders = [ {"order_id": 1, "status": "new", "amount": 19.5}, {"order_id": 2, "status": "new", "amount": 7.25}, {"order_id": 1, "status": "paid", "amount": 19.5}, ] for order in orders: producer.produce(TOPIC, key=f"order-{order['order_id']}", value=json.dumps(order), headers=[("source", b"docs")]) producer.flush() session = vastdb.connect() print("ready") ``` ``` ready ``` `flush()` returns once the broker has acknowledged every record. The broker acknowledges a produce request after its events are committed, so they're readable as table rows straight away. ## Read with the Python SDK Open the topic table like any other table, and look at its columns: ``` with session.transaction() as tx: topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC) for field in topic.columns(): print(field.name, field.type) ``` ``` key binary value binary timestamp_broker timestamp[ms] timestamp_producer timestamp[ms] headers list not null> external_offset int64 compaction_process_seq_num int64 extra_metadata binary ``` Read the events and decode the payloads. As with any SDK query, call `read_all()` before the transaction ends: ``` with session.transaction() as tx: topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC) events = topic.select(columns=["key", "value", "timestamp_broker", "headers"]).read_all() for row in sorted(events.to_pylist(), key=lambda r: (r["key"], r["value"])): print(row["key"].decode(), json.loads(row["value"]), [(h["key"], h["value"]) for h in row["headers"]]) ``` ``` order-1 {'order_id': 1, 'status': 'new', 'amount': 19.5} [('source', b'docs')] order-1 {'order_id': 1, 'status': 'paid', 'amount': 19.5} [('source', b'docs')] order-2 {'order_id': 2, 'status': 'new', 'amount': 7.25} [('source', b'docs')] ``` Rows don't come back in any guaranteed order: the cluster reads the table partition by partition. Sort on the client if order matters. ## Filter on the cluster Filters and column lists are applied on the cluster, so only matching events come back. A common filter is a time window on `timestamp_broker`: ``` since = datetime.now(timezone.utc) - timedelta(hours=1) with session.transaction() as tx: topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC) recent = topic.select(columns=["key", "value"], predicate=topic["timestamp_broker"] >= since).read_all() print(recent.num_rows, "events in the last hour") ``` ``` 3 events in the last hour ``` You can't filter on fields inside `value`, because the cluster sees it as bytes. To filter or aggregate on payload fields, use [blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion) and query its target table. The DB guide [Querying](/develop/docs/dev-database-guides-querying) covers predicates and tuning. ## Query with the VAST Query Engine The Query Engine runs SQL on the cluster. Connect with the ADBC driver, and name the table as `"<bucket>/kafka_topics"."<topic>"`. Quote the topic name, since topic names often contain `-` or `.`: ``` import adbc_driver_manager.dbapi as dbapi import adbc_driver_vastdb conn = dbapi.connect( driver=adbc_driver_vastdb.get_driver_path(), db_kwargs={ "vast.db.endpoint": os.environ["QE_ENDPOINT"], "vast.db.access_key": os.environ["AWS_ACCESS_KEY_ID"], "vast.db.secret_key": os.environ["AWS_SECRET_ACCESS_KEY"], }, autocommit=True, ) TABLE = f'"{EVENTS_BUCKET}/kafka_topics"."{TOPIC}"' with conn.cursor() as cur: cur.execute(f"SELECT count(*) AS events FROM {TABLE} WHERE key = 'order-1'::BLOB") print("events for order-1:", cur.fetchone()[0]) cur.execute(f"SELECT key, value FROM {TABLE} ORDER BY value DESC LIMIT 2") for k, value in cur.fetchall(): print(k.decode(), value.decode()) ``` ``` events for order-1: 2 order-2 {"order_id": 2, "status": "new", "amount": 7.25} order-1 {"order_id": 1, "status": "paid", "amount": 19.5} ``` The aggregation and the filter run on the cluster. `key` and `value` are binary, so compare them with a binary value: `'order-1'::BLOB`, or `CAST(key AS VARCHAR) = 'order-1'`. Don't use the SQL-standard hex literal `X'6f72…'`: it runs without an error but matches nothing. The example decodes the results in Python. In VAST Cluster 5.5 the Query Engine doesn't support `GROUP BY`, and accepts `ORDER BY` only with a `LIMIT`: `GROUP BY` fails with "The operator 'HASH_GROUP_BY' ... is currently not supported", and `ORDER BY` without `LIMIT` with "The operator 'ORDER_BY' ... is currently not supported for this usage". To group events, use Trino or Spark, or read the columns with the SDK and group them with PyArrow (`table.group_by()`). See [VAST Query Engine (ADBC)](/develop/docs/dev-database-integrations-adbc) for the driver, and the [SQL reference](/develop/docs/dev-database-reference-sql-index) for the SQL you can use. ## Query with Trino or Spark The VAST DataBase connectors for Trino and Spark see topic tables like any other table. In Trino the topics schema is `<bucket>/kafka_topics`: for example `kafka2/kafka_topics` for a Kafka view whose bucket is `kafka2` ([Querying Events from Topics using VAST Database Connectors](/docs-5-5/docs/publishing-events-to-vast-event-broker#querying-events-from-topics-using-vast-database-connectors)). In Trino, decode the binary payload with `from_utf8()`, and read JSON fields with the JSON functions: ``` SELECT from_utf8(key) AS order_key, json_extract_scalar(from_utf8(value), '$.status') AS status, timestamp_broker FROM vast."events/kafka_topics"."docs-querying-topics-orders" ORDER BY timestamp_broker; ``` Parsing JSON in every query costs CPU on the Trino workers. For payload fields you query often, [blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion) gives you typed columns that the cluster can filter. See [Trino](/develop/docs/dev-database-integrations-trino) and [Spark](/develop/docs/dev-database-integrations-spark) to set up the connectors. ## What you can't do with a topic table Topic tables are read-only through the VAST DataBase API. You write events with a Kafka producer. The cluster refuses inserts, updates, deletes, imports and column changes on a topic table: ``` import pyarrow as pa try: with session.transaction() as tx: topic = tx.bucket(EVENTS_BUCKET).schema("kafka_topics").table(TOPIC) topic.insert(pa.table({"key": [b"order-9"], "value": [b"{}"]})) except Exception as e: print(type(e).__name__) ``` ``` BadRequest ``` Also: - You can't create a topic through the VAST DataBase API. Create it with a Kafka client or VMS; see [Topics](/develop/docs/dev-event-broker-guides-topics). - If a topic's maximum message size is set above 5 MB, database queries on the topic aren't supported ([Creating a Topic in VAST Web UI](/docs-5-5/docs/publishing-events-to-vast-event-broker#creating-a-topic-in-vast-web-ui)). - Retention and compaction remove rows in the background, so a query only sees what the topic still holds. See [Topics](/develop/docs/dev-event-broker-guides-topics). **Warning: dropping a topic table deletes the topic.** `table.drop()` in the SDK, or `DROP TABLE` in SQL, on a table in `kafka_topics` deletes the Kafka topic and all its events, as if you'd deleted it with a Kafka admin client. Producers and consumers of that topic start getting errors. ## Reads see committed events only A query sees the events of produce requests that have committed, and none of a request that's still in progress. That's the same view a Kafka consumer gets. Each Produce request is one database transaction, so a query never sees part of one. See [architecture](/develop/docs/dev-event-broker-concepts-architecture). ## Clean up Delete the topic with the Kafka admin client, and close the connections: ``` conn.close() admin.delete_topics([TOPIC])[TOPIC].result() print("cleaned up") ``` ``` cleaned up ``` ## See also - [Blob expansion](/develop/docs/dev-event-broker-guides-blob-expansion): turn JSON values into typed columns as events arrive. - [Consuming](/develop/docs/dev-event-broker-guides-consuming): read the same events with a Kafka consumer. - [Architecture](/develop/docs/dev-event-broker-concepts-architecture): how topics are stored as tables. - [Querying](/develop/docs/dev-database-guides-querying), [VAST Query Engine (ADBC)](/develop/docs/dev-database-integrations-adbc) and [Trino](/develop/docs/dev-database-integrations-trino) in the VAST DataBase docs.