> ## Documentation Index
> Fetch the complete documentation index at: https://docs.bloklake.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Reading Arrow

> Decode /query responses in Python, TypeScript and Rust, with complete, tested clients.

> [!TIP]
>
> The [client libraries](https://docs.bloklake.com/clients.md) for TypeScript, Python and Rust do everything on this page for
> you and return JSON rows. Read on if you want Arrow tables (for Parquet, pandas, Polars, DuckDB)
> or use another language.

`POST /query` returns `application/vnd.apache.arrow.stream`. Any Apache Arrow library can read
it. There is one thing to get right: the body is **several Arrow IPC streams back to back**, one
per row type, not one stream.

## The response layout

* One complete Arrow IPC stream per row type you asked for, always in the order
  **transactions, then logs, then blocks** (whatever order you listed them in).
* Each stream has its own schema, its record batches, and its own end-of-stream marker.
* Every stream's schema carries the same `bl.*` metadata. `bl.streams` lists the streams in
  order (`"transactions,logs"`) and `bl.row_type` names the current one.
* The first schema arrives as soon as the query is planned, before any rows.

A correct reader:

1. Opens a stream reader on the body and reads it to its end-of-stream marker.
2. Reads `bl.streams` from its schema, then opens the next stream on the **same** body, until it
   has read that many.
3. Checks that the total rows received equal `bl.objects`.

> [!WARNING]
>
> **Check for truncation.** If a query fails after the server has started sending, the server
> can't send an error body any more: it closes the connection. Most Arrow readers raise an error
> when a body ends mid-batch, but **a body cut exactly between two batches can look like a valid,
> shorter stream.** `bl.objects` is in the schema, sent before any row, and equals the number of
> rows the response carries. Compare it with what you received, and treat a mismatch as a failed
> request. The readers below do this.

## Complete readers

Each of these is a small, complete client that returns Arrow tables: it sends the query, decodes every stream, checks for
truncation and reports API errors with their code. Copy one into your project.

> [tab] Python
>
> Needs `pyarrow` and `requests` (`pip install pyarrow requests`). `query()` also retries
> rate limits, busy servers and broken connections; see
> [Errors and retries](https://docs.bloklake.com/production/errors-and-retries.md).
>
> ```python bloklake_arrow.py theme={null}
> """A small, complete Bloklake client: POST /query, decode every Arrow stream, retry what's retryable."""
> import os
> import time
>
> import pyarrow as pa
> import pyarrow.ipc as ipc
> import requests
>
> BASE = os.environ.get("BLOKLAKE_URL", "https://api.bloklake.com/v1/chains/eth")
> KEY = os.environ["BLOKLAKE_API_KEY"]
> session = requests.Session()
> session.headers["Authorization"] = f"Bearer {KEY}"
>
>
> class QueryError(Exception):
>     def __init__(self, status: int, code: str, message: str, retry_after: float | None = None):
>         super().__init__(f"{status} {code}: {message}")
>         self.status, self.code, self.retry_after = status, code, retry_after
>
>
> def _query_once(body: dict) -> tuple[dict[str, pa.Table], dict[str, str]]:
>     resp = session.post(f"{BASE}/query", json=body, stream=True, timeout=(10, 120))
>     if resp.status_code != 200:
>         try:
>             err = resp.json()["error"]
>         except Exception:
>             err = {"code": "http_error", "message": resp.text[:200]}
>         ra = resp.headers.get("Retry-After")
>         raise QueryError(resp.status_code, err["code"], err["message"], float(ra) if ra else None)
>     resp.raw.decode_content = True
>     tables: dict[str, pa.Table] = {}
>     meta: dict[str, str] = {}
>     try:
>         while True:
>             reader = ipc.open_stream(resp.raw)       # one complete Arrow IPC stream
>             table = reader.read_all()                # ends at this stream's end-of-stream marker
>             meta = {k.decode(): v.decode() for k, v in reader.schema.metadata.items()}
>             tables[meta["bl.row_type"]] = table
>             if len(tables) == len(meta["bl.streams"].split(",")):
>                 break
>     except (pa.ArrowInvalid, OSError, requests.exceptions.RequestException) as e:
>         # the server aborts the connection if a query fails after it started sending
>         raise QueryError(0, "truncated", f"response ended early: {e}") from e
>     # A body cut exactly between two batches reads as a shorter, valid stream:
>     # bl.objects (sent before any row) is the number of rows the response carries.
>     received = sum(t.num_rows for t in tables.values())
>     if received != int(meta["bl.objects"]):
>         raise QueryError(0, "truncated", f"got {received} of {meta['bl.objects']} rows")
>     return tables, meta
>
>
> def query(body: dict, attempts: int = 5) -> tuple[dict[str, pa.Table], dict[str, str]]:
>     """Run a query, retrying rate limits, busy servers and broken connections."""
>     for attempt in range(attempts):
>         try:
>             return _query_once(body)
>         except QueryError as e:
>             retryable = e.status in (429, 503) or e.status >= 500 or e.code == "truncated"
>             if not retryable or attempt == attempts - 1:
>                 raise
>             time.sleep(e.retry_after or min(2 ** attempt, 30))
>     raise AssertionError("unreachable")
> ```
>
> ```python theme={null}
> from bloklake_arrow import query
>
> tables, meta = query({
>     "fromBlock": 19426000, "toBlock": 19426009,
>     "logs": [{"address": ["0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"]}],
>     "joinTransactions": True,
>     "fields": {"log": {"blockNumber": True, "logIndex": True}, "transaction": {"hash": True}},
> })
> print({name: t.num_rows for name, t in tables.items()})   # {'transactions': 62, 'logs': 85}
> print(meta["bl.next_block"])                              # 19426010
> ```

> [tab] TypeScript
>
> Needs `apache-arrow` (`npm install apache-arrow`). Works in Node 18+, Bun and Deno.
> `RecordBatchReader.readAll` yields one reader per stream in the body.
>
> ```typescript bloklake-arrow.ts theme={null}
> import { RecordBatchReader, Table } from "apache-arrow";
>
> const BASE = process.env.BLOKLAKE_URL ?? "https://api.bloklake.com/v1/chains/eth";
> const KEY = process.env.BLOKLAKE_API_KEY!;
>
> export class QueryError extends Error {
>   constructor(public status: number, public code: string, message: string, public retryAfter?: number) {
>     super(`${status} ${code}: ${message}`);
>   }
> }
>
> /** POST /query: every Arrow stream of the response as a Table, keyed by row type, plus the bl.* metadata. */
> export async function query(body: unknown): Promise<{ tables: Record<string, Table>; meta: Map<string, string> }> {
>   const res = await fetch(`${BASE}/query`, {
>     method: "POST",
>     headers: { "content-type": "application/json", authorization: `Bearer ${KEY}` },
>     body: JSON.stringify(body),
>   });
>   if (!res.ok) {
>     const err = await res.json().then((j) => j.error).catch(() => ({ code: "http_error", message: res.statusText }));
>     const ra = res.headers.get("retry-after");
>     throw new QueryError(res.status, err.code, err.message, ra ? Number(ra) : undefined);
>   }
>   const tables: Record<string, Table> = {};
>   let meta = new Map<string, string>();
>   try {
>     // The body is several complete Arrow IPC streams back to back: one reader per stream.
>     for await (const reader of RecordBatchReader.readAll(res.body!)) {
>       const batches = [];
>       for await (const batch of reader) batches.push(batch);
>       meta = reader.schema.metadata;
>       tables[meta.get("bl.row_type")!] = new Table(reader.schema, batches);
>     }
>   } catch (e) {
>     throw new QueryError(0, "truncated", `response ended early: ${e}`);
>   }
>   // bl.objects (sent before any row) is how many rows the response carries.
>   const received = Object.values(tables).reduce((n, t) => n + t.numRows, 0);
>   if (received !== Number(meta.get("bl.objects"))) throw new QueryError(0, "truncated", `got ${received} of ${meta.get("bl.objects")} rows`);
>   return { tables, meta };
> }
> ```
>
> ```typescript theme={null}
> import { query } from "./bloklake-arrow.ts";
>
> const { tables, meta } = await query({
>   fromBlock: 19426000, toBlock: 19426009,
>   logs: [{ address: ["0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48"] }],
>   fields: { log: { blockNumber: true, data: true } },
> });
> const row = tables.logs!.get(0)!;
> row.block_number;  // 19426000n: UInt64 columns are bigint in JavaScript
> row.data;          // Uint8Array
> ```

> [tab] Rust
>
> Needs `arrow` (feature `ipc`), `reqwest` (feature `json`), `serde_json` and `tokio`.
>
> ```toml Cargo.toml theme={null}
> [dependencies]
> arrow = { version = "59", default-features = false, features = ["ipc"] }
> reqwest = { version = "0.13", features = ["json"] }
> serde_json = "1"
> tokio = { version = "1", features = ["full"] }
> ```
>
> ```rust src/bloklake_arrow.rs theme={null}
> use arrow::ipc::reader::StreamReader;
> use arrow::record_batch::RecordBatch;
> use std::collections::HashMap;
> use std::io::Cursor;
>
> type Error = Box<dyn std::error::Error + Send + Sync>;
>
> /// The decoded response: record batches per row type, and the `bl.*` metadata.
> pub struct Response {
>     pub tables: HashMap<String, Vec<RecordBatch>>,
>     pub meta: HashMap<String, String>,
> }
>
> /// POST /query and decode every Arrow stream of the response.
> pub async fn query(client: &reqwest::Client, base: &str, key: &str, body: &serde_json::Value) -> Result<Response, Error> {
>     let resp = client.post(format!("{base}/query")).bearer_auth(key).json(body).send().await?;
>     let status = resp.status();
>     if !status.is_success() {
>         return Err(format!("{status}: {}", resp.text().await?).into());
>     }
>     // A response carries about 8 MB of row data at most, so buffering it is fine.
>     let mut cursor = Cursor::new(resp.bytes().await?);
>     let mut tables = HashMap::new();
>     let meta = loop {
>         // One complete IPC stream; the reader stops at its end-of-stream marker.
>         let reader = StreamReader::try_new(&mut cursor, None)?;
>         let meta = reader.schema().metadata().clone();
>         let batches = reader.collect::<Result<Vec<_>, _>>()?;
>         tables.insert(meta["bl.row_type"].clone(), batches);
>         if tables.len() == meta["bl.streams"].split(',').count() {
>             break meta;
>         }
>     };
>     // bl.objects (sent before any row) is how many rows the response carries.
>     let received: usize = tables.values().flatten().map(|b| b.num_rows()).sum();
>     if received.to_string() != meta["bl.objects"] {
>         return Err(format!("truncated response: got {received} of {} rows", meta["bl.objects"]).into());
>     }
>     Ok(Response { tables, meta })
> }
> ```

> [!NOTE]
>
> In any other language, the recipe is the same: open an Arrow IPC stream reader on the response
> body, read it to the end, read the next one from the same position, repeat `bl.streams` times,
> then compare the row count with `bl.objects`. Go's `ipc.NewReader`, Java's
> `ArrowStreamReader` and C++'s `RecordBatchStreamReader` all work this way.

## Into a dataframe

Arrow tables convert to dataframes without copying the data.

```python Polars theme={null}
import polars as pl

logs = pl.from_arrow(tables["logs"])
# topics are 32 bytes; an address in a topic is the last 20
logs = logs.with_columns(
    ("0x" + pl.col("topic1").bin.slice(12).bin.encode("hex")).alias("from"),
    ("0x" + pl.col("topic2").bin.slice(12).bin.encode("hex")).alias("to"),
)
```

```python pandas theme={null}
df = tables["logs"].to_pandas()
df["from"] = df["topic1"].map(lambda b: "0x" + b[12:].hex())
```

```python DuckDB theme={null}
import duckdb

txs = tables["transactions"]           # DuckDB reads Arrow tables in place
duckdb.sql("select count(*), sum(gas_used) from txs").fetchall()
```

See [Data types](https://docs.bloklake.com/responses/data-types.md) for how hashes, addresses and 128/256-bit amounts are
encoded, with helpers to convert them.

## Reading incrementally

The readers above hold one response in memory at a time. A response carries about 8 MB of row
data at most, so that's fine for almost everyone, and the [pagination loop](https://docs.bloklake.com/querying/pagination.md)
keeps memory flat across a whole backfill.

If you want to process rows while a response is still arriving (lower latency to the first
row), iterate the record batches instead of collecting them (`for batch in reader` in pyarrow,
`for await (const batch of reader)` in arrow-js). Keep reading promptly: a response that isn't
read for 60 seconds, or is read slower than 50 KB/s, is
[cut off](https://docs.bloklake.com/production/errors-and-retries.md#slow-and-idle-clients).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.