> **Can't find what you're looking for?** Use `search_docs` on the docs MCP server at `https://docs.ampdb.io/api/mcp` to find what you need. # Streaming Queries A streaming query turns SQL into a feed. Amp runs it over the blocks synced so far, then again over each new range of blocks as its datasets sync them, and sends only the new rows. It doesn't end on its own. Use it to keep an app, a database, or a cache up to date with a chain. Streaming queries run the same SQL as [batch queries](/querying/batch-queries), except for what can't be computed one range of blocks at a time, like sorting the whole result (see [what they support](/querying/streaming-queries#what-streaming-queries-support)). ## Prerequisites * A [connection](/querying/connecting) to a query server, with a client that can stream: the [TypeScript](/sdks/typescript/streaming) or [Python](/sdks/python/streaming) SDK, or your own [Arrow Flight client](/querying/arrow-flight#write-a-client). `ampsql` only runs batch queries. * To run the examples, the [example datasets](/querying/connecting#example-datasets), and the Python client from [Arrow Flight](/querying/arrow-flight#write-a-client), saved as `flight_query.py`. It prints each part of the stream as it arrives. ## Start a stream End the query with `SETTINGS stream = true`. Here, for every 20th block: :::terminal ```bash python flight_query.py 'SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks WHERE block_num % 20 = 0 SETTINGS stream = true' ``` ```text [mainnet 24000000-24000099] 5 rows block_num gas_used 24000000 57860235 24000020 31081726 24000040 57144543 24000060 17614864 24000080 59943261 [mainnet 24000000-24000099] complete ``` ::: The stream stays open: the example datasets stop at their last block, but a dataset that's still syncing would send each new block's rows as it arrives. `flight_query.py` stops reading at Amp's first [keep-alive message](/querying/arrow-flight#keep-alive-messages), 30 seconds after it starts, even while new blocks keep arriving, because pyarrow reports that message as the end of the stream. That section shows how to read on. You can also leave the SQL as it is and send the `amp-stream` header: `true` or `1` streams the query, and `false` or `0` runs it as a batch query, whatever its `SETTINGS` say. Any other value fails with ``malformed `amp-stream` header``. :::warning Amp [ignores settings it doesn't know](/querying/sql#settings), without an error. If a query you meant to stream ends once its results have arrived, check the spelling: `SETTINGS steam = true` runs a batch query. ::: ## How results arrive Amp runs a stream in microbatches: it runs the query over a range of blocks, sends the new rows, then moves to the next range. For each microbatch, the client receives: 1. The rows, in one or more Arrow record batches. Each one carries the microbatch's block range, marked incomplete. 2. An empty batch carrying the same block range, marked complete. From then on, the range's rows are all delivered. A microbatch with no rows only sends the second: its range is complete, with nothing in it. A query whose rows are rare can send many of these in a row. The rows of a microbatch come in block order. Each range has, for each chain, its first and last block, the hash of the last block, and the hash of the block before the first. [Arrow Flight](/querying/arrow-flight#block-ranges) describes the exact format. The SDKs read it for you. Once a stream has caught up, Amp waits for new blocks. So that the connection doesn't time out while it waits, Amp sends a message with no data every [`keep_alive_interval`](/running-amp/configuration-reference#serverkeep_alive_interval) seconds (30 by default, and at least 30), counted from the start of the stream. It keeps that schedule whether blocks arrive or not, even in the middle of a microbatch. A stream only ends when the client disconnects, or on an error. ### Microbatch size A microbatch reads at most `microbatch_max_interval` blocks: `100000` by default, or what the query server's [`microbatch_max_interval`](/running-amp/configuration-reference#servermicrobatch_max_interval) setting says. A query can set its own, higher or lower, in its `SETTINGS`: :::terminal ```bash python flight_query.py 'SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks WHERE block_num % 20 = 0 SETTINGS stream = true, microbatch_max_interval = 40' ``` ```text [mainnet 24000000-24000039] 2 rows block_num gas_used 24000000 57860235 24000020 31081726 [mainnet 24000000-24000039] complete [mainnet 24000040-24000079] 2 rows block_num gas_used 24000040 57144543 24000060 17614864 [mainnet 24000040-24000079] complete [mainnet 24000080-24000099] 1 rows block_num gas_used 24000080 59943261 [mainnet 24000080-24000099] complete ``` ::: `0` counts as `1`, so each microbatch reads one block: it doesn't fail, turn the limit off, or fall back to the server's setting. [Settings](/querying/sql#settings) lists the values that do fail. Smaller microbatches deliver rows sooner when a stream catches up on history, and use less memory, but take longer overall. When the query server [limits each query's memory](/querying/limits#memory) (`query_max_mem_mb`), Amp sizes each microbatch from the memory the previous ones used, aiming for 85% of the limit, and `microbatch_max_interval` is only a ceiling. A microbatch that still runs out of memory fails the query. ### Where a stream starts A new stream starts at the first block of the tables it reads. If the query filters on `block_num()` or `_block_num`, Amp works out the first block that can match, and starts there instead of reading the earlier ranges for nothing: ```sql SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks WHERE block_num() >= 24000097 SETTINGS stream = true ``` ```text [Output] [mainnet 24000097-24000099] 3 rows block_num gas_used 24000097 52877862 24000098 36233422 24000099 10751338 [mainnet 24000097-24000099] complete ``` Amp only skips ahead when the filter applies to every row of the result, and it can work out the first block from it. Otherwise, the stream starts at the first block, as it does when: * The filter is inside a `GROUP BY`, `DISTINCT ON`, or `UNION ALL` query, which combines rows after filtering them. * The filter is on the `block_num` column, like `WHERE block_num >= 24000097`, rather than `block_num()` or `_block_num`. * The filter computes on the block number, like `block_num() + 1 >= 24000098`, or combines it with another column using `OR`. `AND` is fine. * The stream reads several chains. There, a filter on `_block_num` doesn't skip ahead, and one on `block_num()` makes the stream fail: see [Several chains](/querying/streaming-queries#several-chains). The stream still works through the earlier ranges. It sends each one even when no rows match, complete and with no rows. With a lower bound like `block_num() >= 24000097`, that's every range before the one that holds the block. In a `GROUP BY`, for example: ```sql SELECT block_num(), COUNT(*) AS tx_count FROM "acme/eth_mainnet".transactions WHERE block_num() >= 24000097 GROUP BY block_num() SETTINGS stream = true, microbatch_max_interval = 40 ``` ```text [Output] [mainnet 24000000-24000039] complete [mainnet 24000040-24000079] complete [mainnet 24000080-24000099] 3 rows block_num() tx_count 24000097 166 24000098 165 24000099 122 [mainnet 24000080-24000099] complete ``` To skip those ranges, filter the groups with `HAVING` instead, which applies to every row of the result: ```sql SELECT block_num(), COUNT(*) AS tx_count FROM "acme/eth_mainnet".transactions GROUP BY block_num() HAVING block_num() >= 24000097 SETTINGS stream = true, microbatch_max_interval = 40 ``` ```text [Output] [mainnet 24000097-24000099] 3 rows block_num() tx_count 24000097 166 24000098 165 24000099 122 [mainnet 24000097-24000099] complete ``` A resumed stream starts after its cursor. ## Resume a stream To pick up where a stream left off, after a disconnection for example, send a cursor in the `amp-resume` header. A cursor is the last block of a complete range, for each chain the stream reads: its number and its hash. ```json { "mainnet": { "number": 24000097, "hash": "0x326546b40bf50a0f05070e92f1764d8266591850c00d5c92587cd13d84312332" } } ``` The stream continues with the next block: :::terminal ```bash python flight_query.py 'SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks SETTINGS stream = true' \ 'amp-resume: {"mainnet": {"number": 24000097, "hash": "0x326546b40bf50a0f05070e92f1764d8266591850c00d5c92587cd13d84312332"}}' ``` ```text [mainnet 24000098-24000099] 2 rows block_num gas_used 24000098 36233422 24000099 10751338 [mainnet 24000098-24000099] complete ``` ::: * Only save a cursor from a range marked complete. Rows from a range that wasn't complete may be partial: resuming after it would skip the rest. * If the cursor's block is no longer on the chain, because of a reorg, Amp resends the blocks from where the chains diverge. * A cursor that isn't valid JSON fails with ``malformed `amp-resume` header``. The SDKs save cursors and resume for you. ## Reorgs A [reorg](/get-started/concepts/reorgs) replaces blocks a chain had already produced. When a stream has already sent rows from the replaced blocks, Amp sends the rows of the new blocks, and the client must discard the old ones. Amp doesn't send a separate message for it: the next range starts at or before the end of the previous one, and the hash of the block before it doesn't match the previous range's last hash. When that happens, discard the rows you received from that range's first block on, then apply the new rows. The new range may start before the fork, so some blocks you receive again are unchanged. [Arrow Flight](/querying/arrow-flight#detect-reorgs) has the exact rules, and the SDKs apply them for you. A few kinds of datasets order blocks by number only, without hashes, for sources whose blocks are already final. Their streams never signal a reorg. ## What streaming queries support A streaming query can only do what Amp can compute one block range at a time, without keeping state between ranges. | Supported | Not supported | | ------------------------------------------------------------------------------------ | -------------------------------------------------------- | | Selecting and computing columns, filters, functions | `ORDER BY` over the whole result | | `UNION ALL`, subqueries, `WITH` clauses, `IN` and `EXISTS` | `LIMIT` | | `INNER`, cross, and semi joins, with [restrictions](/querying/joins#streaming-joins) | Outer and anti joins | | `GROUP BY`, when its first key is `block_num()` | Other aggregations, like `COUNT(*)` over the whole table | | `DISTINCT ON`, when its first expression is `block_num()`, without `ORDER BY` | `DISTINCT`, window functions, and recursive queries | | Raw and derived datasets | [Static datasets](/datasets/static-datasets) | Grouping by block works because each block's rows all arrive in the same microbatch: ```sql SELECT block_num(), COUNT(*) AS tx_count FROM "acme/eth_mainnet".transactions WHERE block_num % 10 = 0 GROUP BY block_num() SETTINGS stream = true, microbatch_max_interval = 50 ``` ```text [Output] [mainnet 24000000-24000049] 5 rows block_num() tx_count 24000000 232 24000010 464 24000020 186 24000030 92 24000040 228 [mainnet 24000000-24000049] complete [mainnet 24000050-24000099] 5 rows block_num() tx_count 24000050 72 24000060 179 24000070 153 24000080 68 24000090 132 [mainnet 24000050-24000099] complete ``` A query Amp can't stream fails before it sends any rows, with gRPC status `INVALID_ARGUMENT` and a message naming the operation: ```text query is not supported in streaming mode | Caused by: query contains non-incremental operation: Aggregate ``` A query reading a static dataset fails the same way, with `static datasets are not supported in streaming queries; use a bounded (batch) query`. Run it as a batch query instead. These errors come from `DoGet`, the call that fetches the results. `GetFlightInfo`, the call before it, accepts the query and returns a ticket, so a successful `GetFlightInfo` doesn't mean a query can stream (see [how a query runs](/querying/arrow-flight#how-a-query-runs)). ## Several chains A stream can read datasets of several chains, in a `UNION ALL` or a [join](/querying/joins#streaming-joins). Block numbers of different chains can't be compared, so Amp lines the chains up by time instead: * Each microbatch is a window of time, and its range has each chain's blocks in that window. * `microbatch_max_interval` is in seconds, not blocks. `0` counts as `1` here too. * The stream only advances as far as every chain has synced, so it waits for the one furthest behind. Here, with every fourth Ethereum block and every 24th Base block, so that each chain has a block every 48 seconds: :::terminal ```bash python flight_query.py "SELECT 'ethereum' AS chain, block_num, timestamp FROM \"acme/eth_mainnet\".blocks WHERE block_num % 4 = 0 UNION ALL SELECT 'base', block_num, timestamp FROM \"acme/base_mainnet\".blocks WHERE block_num % 24 = 0 SETTINGS stream = true, microbatch_max_interval = 120" ``` ```text [base 39397512-39397572, mainnet 24000000-24000010] 6 rows chain block_num timestamp ethereum 24000000 2025-12-13 00:06:11+00:00 base 39397512 2025-12-13 00:06:11+00:00 ethereum 24000004 2025-12-13 00:06:59+00:00 base 39397536 2025-12-13 00:06:59+00:00 ethereum 24000008 2025-12-13 00:07:47+00:00 base 39397560 2025-12-13 00:07:47+00:00 [base 39397512-39397572, mainnet 24000000-24000010] complete [base 39397573-39397638, mainnet 24000011-24000021] 6 rows chain block_num timestamp ethereum 24000012 2025-12-13 00:08:35+00:00 base 39397584 2025-12-13 00:08:35+00:00 ethereum 24000016 2025-12-13 00:09:23+00:00 base 39397608 2025-12-13 00:09:23+00:00 ethereum 24000020 2025-12-13 00:10:11+00:00 base 39397632 2025-12-13 00:10:11+00:00 [base 39397573-39397638, mainnet 24000011-24000021] complete … ``` ::: :::warning In a stream over several chains, filter and select `block_num` rather than `block_num()`: `block_num()` currently makes the stream fail, with gRPC status `CANCELLED` and `Received RST_STREAM with error code 8`, which doesn't name the cause. ::: A [derived dataset](/datasets/derived-datasets) whose SQL reads several chains runs the same way. Its tables must include a `_ts` column, and its deployment can't have an end block.