> **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. # Arrow Flight [Arrow Flight](https://arrow.apache.org/docs/format/Flight.html) is Apache Arrow's protocol for sending tables over gRPC. It's the only protocol Amp's query server speaks, on port `1602`, and `ampsql`, the SDKs, and ADBC speak it for you. This page is for writing your own client, in any language with an Arrow Flight library, and for what every client sends and receives: the request headers, and the block ranges that come with each result. ## Prerequisites * A running Amp to [connect](/querying/connecting) to. * To run the examples, Python 3 with pyarrow and pandas (`pip install pyarrow pandas`), and the [example datasets](/querying/connecting#example-datasets). ## How a query runs A query takes two calls: 1. **`GetFlightInfo`**, with the SQL. Amp parses and plans the query, and returns the result's schema and a ticket. The ticket holds the query, whether it streams, its resume cursor, and the dataset versions its [aliases](/querying/dataset-aliases) resolved to. 2. **`DoGet`**, with the ticket. Amp plans the query again, runs it, and sends the result as a series of `FlightData` messages: Arrow record batches, each with its [block ranges](/querying/arrow-flight#block-ranges). Of the mistakes in a query, `GetFlightInfo` only catches a bad [`stream` setting](/querying/sql#settings) and what stops Amp from working out the result's schema: invalid SQL, and tables, columns, or functions that don't exist. The others come up in `DoGet`, like a [query that can't stream](/querying/streaming-queries#what-streaming-queries-support) or [columns with the same name](/querying/arrow-flight#mistakes-that-return-internal), so a client has to handle errors from both calls. The SQL goes in `GetFlightInfo`'s descriptor, as in [Arrow Flight SQL](https://arrow.apache.org/docs/format/FlightSql.html): a descriptor of type `CMD`, whose command is a `CommandStatementQuery` with the SQL in its `query` field, packed in a protobuf `Any`. Amp supports nothing else: | You send | Amp responds | | -------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------- | | The SQL itself as the command, without a `CommandStatementQuery` | `failed to decode Any command envelope` | | Another Flight SQL command, like `CommandGetTables` | `unsupported flight descriptor command`, followed by the command's type | | A descriptor of type `PATH` | `unsupported flight descriptor type: 1` | | `ListFlights`, `GetSchema`, `PollFlightInfo`, `DoPut`, `DoAction`, `DoExchange`, `Handshake` | gRPC status `UNIMPLEMENTED` | ## Write a client `flight_query.py` runs a query and prints each part of its result as it arrives, with its block ranges. It takes the SQL, then any headers to send, written `name: value`: ```python [flight_query.py] import json import sys import pyarrow.flight as flight def statement_query(sql: str) -> bytes: """Wraps SQL in a Flight SQL CommandStatementQuery, packed in a protobuf Any.""" def field(number: int, value: bytes) -> bytes: # A protobuf field of bytes: its tag, its length as a varint, then the bytes length, size = b"", len(value) while size > 0x7F: length += bytes([size & 0x7F | 0x80]) size >>= 7 return bytes([number << 3 | 2]) + length + bytes([size]) + value type_url = b"type.googleapis.com/arrow.flight.protocol.sql.CommandStatementQuery" return field(1, type_url) + field(2, field(1, sql.encode())) sql, *headers = sys.argv[1:] options = flight.FlightCallOptions(headers=[tuple(h.encode() for h in header.split(": ", 1)) for header in headers]) client = flight.connect("grpc://localhost:1602") info = client.get_flight_info(flight.FlightDescriptor.for_command(statement_query(sql)), options) for chunk in client.do_get(info.endpoints[0].ticket, options): metadata = json.loads(chunk.app_metadata.to_pybytes()) if chunk.app_metadata else {} ranges = ", ".join(f"{r['network']} {r['numbers']['start']}-{r['numbers']['end']}" for r in metadata["ranges"]) if metadata["ranges_complete"]: print(f"[{ranges}] complete") else: print(f"[{ranges}] {chunk.data.num_rows} rows") print(chunk.data.to_pandas().to_string(index=False)) ``` * pyarrow has no Flight SQL client, so `statement_query` builds the `CommandStatementQuery` itself. * It sends the headers with both calls, which works for every header (see [Headers](/querying/arrow-flight#headers)). * Its `for` loop stops at a stream's first [keep-alive message](/querying/arrow-flight#keep-alive-messages). That keeps the script short, and lets the examples on these pages finish on their own. * For an encrypted connection, like a hosted Amp's, connect to `grpc+tls://` followed by the host and port. For a server with a self-signed certificate, pass the certificate of the authority that signed it, in PEM format: `flight.connect(url, tls_root_certs=open("ca.pem", "rb").read())`. Otherwise, the first call fails with `certificate verify failed: unable to get local issuer certificate`. :::terminal ```bash python flight_query.py 'SELECT block_num, gas_used FROM "acme/eth_mainnet".blocks ORDER BY block_num LIMIT 3' ``` ```text [mainnet 24000000-24000099] 3 rows block_num gas_used 24000000 57860235 24000001 27660983 24000002 17602635 [mainnet 24000000-24000099] complete ``` ::: A batch query's result comes with the block ranges of the [snapshot](/querying/batch-queries#what-a-batch-query-reads) it read: here, all 100 blocks, even though the query returns three. The same client runs [streaming queries](/querying/streaming-queries). In other languages, an Arrow Flight SQL client builds the command for you: its `execute` method calls `GetFlightInfo` with a `CommandStatementQuery`, and returns the ticket to pass to `DoGet`. Arrow has Flight SQL clients for C++, Go, Java, and Rust, among others. ## Headers A request carries its options in gRPC metadata headers: | Header | Read by | Value | Description | | ------------------ | --------------- | --------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------ | | `amp-stream` | `GetFlightInfo` | `true`, `1`, `false`, or `0` | Runs the query as a streaming or a batch query, whatever its `SETTINGS` say. See [streaming queries](/querying/streaming-queries#start-a-stream). | | `amp-resume` | `GetFlightInfo` | A cursor, in JSON | Resumes a stream after the cursor. See [resume a stream](/querying/streaming-queries#resume-a-stream). | | `amp-alias` | `GetFlightInfo` | A JSON object of aliases and references | Chooses the dataset behind each name in the SQL. See [dataset aliases](/querying/dataset-aliases). | | `authorization` | `DoGet` | `Bearer` followed by a JWT | Identifies the user, for [per-user limits](/querying/limits#per-user-limits). See [authentication](/querying/connecting#authentication). | | `amp-capabilities` | Both | JSON | What the request may read and use. Set by gateways: see [namespace access](/querying/arrow-flight#namespace-access) and [per-user limits](/querying/limits#per-user-limits). | | `cf-ray` | Both | Any text | An ID for the request, which Amp's logs include. Cloudflare sets it. Amp makes one up when it's missing. | Amp checks every header on both calls: a malformed one fails either call with gRPC status `FAILED_PRECONDITION` and ``malformed `amp-stream` header``, or the name of the header at fault. A call ignores the headers it doesn't read, so a client can send the same headers with both, as `flight_query.py` does. The `authorization` header must hold a JWT whose `sub` claim is the user's ID: any other token fails with ``malformed `authorization` header``. Amp reads the claim but doesn't verify the token: a gateway in front of it does that. ### Namespace access A gateway that serves several tenants can limit the namespaces a request reads, with a `namespace.read` capability in `amp-capabilities`: ```json { "capabilities": [{ "name": "namespace.read", "scope": ["acme", "analytics"] }] } ``` * A dataset in another namespace is hidden: a query that names it, in its SQL or an [alias](/querying/dataset-aliases), fails as if it didn't exist, with `table '…' not found`. That holds for batch and streaming queries alike. * A reference that pins a manifest hash only resolves if the manifest belongs to the dataset it names. * An empty scope, `[]`, hides every namespace. A request without a `namespace.read` capability reads every namespace. * A request with both an `authorization` header and `amp-capabilities` must also have a `resource.limit` capability. Without one, every query fails with ``missing `resource.limit` capability in `amp-capabilities` header`` (`FAILED_PRECONDITION`). See [Per-user limits](/querying/limits#per-user-limits). Amp trusts `amp-capabilities` as it comes. A gateway that uses it must set it itself, and drop any value a client sends. ## Block ranges Each message of a result carries, in its `app_metadata`, the block ranges its rows come from, in JSON. Here's the message that ends the second microbatch of a stream with `microbatch_max_interval = 40`: ```json { "ranges": [ { "numbers": { "start": 24000040, "end": 24000079 }, "network": "mainnet", "hash": "0x855e4c9340c45dec30805076ac1c86ea4de27ef5f0653c6d808fe1693af0375f", "prev_hash": "0xad842a6b22dd88e054abc5358b7df70dc39396dc6168f1180bba77ba442da490", "timestamp": 1765585319 } ], "ranges_complete": true } ``` | Field | Description | | -------------------- | ------------------------------------------------------------------------------------------------------- | | `ranges` | One range for each chain the query reads. Empty for a query that reads no dataset, like `SELECT 1`. | | `ranges[].network` | The chain, as its [provider](/providers) names it, like `mainnet`. | | `ranges[].numbers` | The range's first and last block, both included. | | `ranges[].hash` | The hash of the last block. | | `ranges[].prev_hash` | The hash of the block before the first. | | `ranges[].timestamp` | The last block's time, in seconds since 1970 (UTC). Missing from data that older versions of Amp wrote. | | `ranges_complete` | `true` on the message that ends the range, and `false` on the messages before it. | For each range, a streaming query's microbatch or a batch query's snapshot, you receive: 1. One or more messages with rows, and `ranges_complete: false`. 2. An empty message with the same ranges, and `ranges_complete: true`. You now have all of the range's rows. The last block of each chain, its number and its hash, makes the cursor to [resume a stream](/querying/streaming-queries#resume-a-stream) from. Without rows, the first step changes. A streaming query's range skips it. A batch query still sends one message there, with no rows and `ranges_complete: false`, then the second as usual. So an empty message doesn't mean that the range is complete, or that anything went wrong: only `ranges_complete: true` means that you have all of its rows. The `prev_hash` above is the `hash` of the stream's first microbatch, which ended at block 24,000,039. Block numbers can have gaps, as on Solana, which skips slots, so a range doesn't always start right after the previous one, but its `prev_hash` still matches the previous `hash`. ### Keep-alive messages So that the connection doesn't time out while a stream waits for new blocks, Amp sends an empty message every [`keep_alive_interval`](/running-amp/configuration-reference#serverkeep_alive_interval) seconds, 30 by default, counted from the start of the stream. It keeps that schedule whether blocks arrive or not, even in the middle of a microbatch, so a keep-alive message doesn't mean that a range is complete. The message has no `app_metadata`, no `data_header`, and no `data_body`. It isn't part of the result, so clients skip it, but with nothing in it to read, some Flight libraries trip on it. :::warning pyarrow reports each keep-alive message as the end of the stream, but the stream stays open. `read_chunk()`, which a `for` loop over the reader calls, raises `StopIteration`, as it does at a real end, and the next `read_chunk()` returns the next data. `read_all()` and `to_reader()` stop there for good, and lose the rest of the stream, without an error. ::: So `flight_query.py` stops at the first keep-alive message, 30 seconds after it starts, even while new blocks keep arriving, as on Base, which produces one every 2 seconds. To read on, tell the two apart: at a real end, pyarrow calls a client middleware's `call_completed` before it raises `StopIteration`, and on a keep-alive message, it doesn't. In `flight_query.py`, connect with this middleware: ```python class CallEnd(flight.ClientMiddlewareFactory, flight.ClientMiddleware): """Records whether the latest call has ended.""" ended = False def start_call(self, info): self.ended = False return self def call_completed(self, exception): self.ended = True call = CallEnd() client = flight.connect("grpc://localhost:1602", middleware=[call]) ``` Then replace the `for` line with these, and leave the loop's body as it is: ```python reader = client.do_get(info.endpoints[0].ticket, options) while True: try: chunk = reader.read_chunk() except StopIteration: if call.ended: break continue # A keep-alive message ``` The script now prints each new block's rows until you press Ctrl+C, which takes effect when the next message arrives: up to 30 seconds later. You can also reconnect after each keep-alive message with the cursor of the last complete range (see [resume a stream](/querying/streaming-queries#resume-a-stream)), but that restarts the query every time. ## Detect reorgs When a [reorg](/get-started/concepts/reorgs) replaces blocks a stream has already sent, the stream sends the new blocks' rows, in a range that starts at or before the end of the previous one. To detect it, compare each new range with the last range you received for the same chain. A range equal to it is part of the same microbatch. Otherwise: | The new range starts | Its `prev_hash` | Which means | | ------------------------------------- | ----------------------------------- | ---------------------------------------------------------- | | After the previous range's end | Matches the previous range's `hash` | No reorg: the chain continues. | | At or before the previous range's end | Doesn't match the previous `hash` | A reorg: the blocks from the new range's start on changed. | | Anything else | | Amp broke the protocol. Stop the stream and report it. | On a reorg: 1. Discard the rows of every range that ends at or after the new range's start. 2. If the new range starts after the first block of a range you're discarding, you've also discarded rows that are still valid, before the reorg. Reconnect with `amp-resume` set to the cursor of the last range you kept, so that Amp sends them again. 3. Otherwise, keep going: apply the new range's rows. The SDKs do this for you. ## Errors A call that fails returns a gRPC status and a message. Together, they say whether retrying can help: | Status | Means | Retry | | --------------------- | ----------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ | | `INVALID_ARGUMENT` | The request is wrong: invalid SQL, an unknown table or alias, a bad setting, a query that can't stream. | No: fix the request. | | `FAILED_PRECONDITION` | A header is malformed. | No: fix the header. | | `UNAVAILABLE` | Amp can't answer yet: the tables have no synced blocks in common, or the dataset registry can't be reached. | Yes, after a while. | | `RESOURCE_EXHAUSTED` | The query ran out of memory, or its user out of budget. | See [Limits](/querying/limits#errors). | | `DEADLINE_EXCEEDED` | A batch query ran past the server's deadline. | See [Limits](/querying/limits#errors). | | `INTERNAL` | Amp failed to run the query, or the query has [one of these mistakes](/querying/arrow-flight#mistakes-that-return-internal). | No for those mistakes: fix the query. Otherwise, once. If it fails again, report it. | | `UNIMPLEMENTED` | A call Amp doesn't support. | No. | The messages for each are on the pages for [SQL](/querying/sql#errors), [streaming queries](/querying/streaming-queries#what-streaming-queries-support), [dataset aliases](/querying/dataset-aliases#errors), and [limits](/querying/limits#errors). ### Mistakes that return `INTERNAL` Some mistakes in a query return `INTERNAL`, and fail the same way every time. They come from `DoGet`, after `GetFlightInfo` succeeded. If the message contains one of these, fix the query instead of retrying: | Message contains | Cause | Fix | | ---------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------------- | | `Duplicate field names detected in plan schema` | The result has two columns with the same name, as with `SELECT *` over a join. | Select [the columns you need](/querying/joins#column-names), with `AS` to rename the ones that clash. | | ``selecting `_ts` from a multi-table context``, or the same for `_block_num` | A [streaming join](/querying/joins#blocks-of-joined-rows) selects `*`, one table's `*`, or `_block_num`. | Select columns by name, and use `block_num()` and `ts()`. | | `query contains non-incremental operation: Sort` | A streaming `DISTINCT ON` has an `ORDER BY`. With any other streaming query, this message comes with `INVALID_ARGUMENT`. | Remove the `ORDER BY`. | | `Cannot infer common argument type` | A comparison between types that can't be compared, like a timestamp and a number. | Change the value, or `CAST` it to the column's type. | | `Unsupported CAST from` | A `CAST` between types Amp can't convert, like an address to an integer. | Cast to a type it can convert to. | | `Cannot cast string` | A value that doesn't convert to the type, in a `CAST` or a comparison. | Fix the value, or use `TRY_CAST`, which returns `NULL` for it. | | `Divide by zero error` | A division by a value that's zero on some row. | Skip those rows, or divide by `NULLIF(value, 0)`, which returns `NULL` for them. | | `is declared as non-nullable but contains null values` | `evm_decode_type` got data it can't decode. | [Decode inside a `CASE`](/querying/functions/evm#decode-data-that-may-not-match). | | `failed to infer SQL output schema`, with `Unsupported type` | `arrow_cast` got a type name Arrow doesn't know. | Use a name [`arrow_typeof`](/querying/sql#types) returns. |