API Introduction

Berserk exposes two query protocols: gRPC (native streaming) and HTTP (ADX v2 REST). Both execute KQL queries against the same engine — they differ in how results are delivered.

gRPC Streaming Protocol

The gRPC API returns a server-side stream of typed frames. Each frame carries one payload:

Schema → RowBatch → RowBatch → Progress → RowBatch → Metadata → Completion

Frame Types

Result Iterations

For long-running queries, the engine delivers incremental results as data is processed. Each update is identified by a result_iteration_id — a UUID that changes with every new set of results.

When the iteration ID changes, discard all previously accumulated rows and start fresh. This is how the engine signals "here is a more complete result set that supersedes the previous one."

Iteration "abc-123":
  RowBatch(table="PrimaryResult", rows=[...], is_iteration_complete=false)
  RowBatch(table="PrimaryResult", rows=[...], is_iteration_complete=true)

Iteration "def-456":          ← new ID → clear all previous rows
  RowBatch(table="PrimaryResult", rows=[...], is_iteration_complete=true)

RowBatch Concatenation

A single result iteration may produce multiple RowBatch frames for the same table because gRPC messages are limited to ~4 MiB and batches are capped at 100 rows.

Concatenate all RowBatch frames that share the same table_name and result_iteration_id:

Batch 1: table="PrimaryResult", iteration="abc", rows=[row0..row99], is_iteration_complete=false
Batch 2: table="PrimaryResult", iteration="abc", rows=[row100..row150], is_iteration_complete=true
→ Result: 151 rows

The is_iteration_complete flag on the last batch signals that no more batches will arrive for this iteration. Clients should:

1. Show the first batch immediately (fast feedback)
2. Accumulate subsequent batches
3. Render the full table once is_iteration_complete=true
4. On new iteration ID, clear and restart

Multi-Table Results

Some queries produce multiple tables (e.g., fork queries or queries with extra diagnostic tables):

• PrimaryResult — the main result table (always first)
• ExtraTable_0, ExtraTable_1, ... — additional tables
• Fork queries use branch names (e.g., "Totals", "TopTwo")

Schema frames are sent once per table name. RowBatch frames reference their table by table_name. When an iteration ID changes, all tables are cleared together — not individually.

Schema(name="PrimaryResult", columns=[...])
Schema(name="Totals", columns=[...])
RowBatch(table="PrimaryResult", iteration="abc", rows=[...])
RowBatch(table="Totals", iteration="abc", rows=[...])

Progress Frames

Progress frames are sent periodically during execution, independent of row delivery. They contain:

• Scan statistics: rows processed, chunks total/scanned/skipped
• Skip reasons: range, bloom filter, shar (hash), required fields
• Timing: chunk scan time, query time, merge time (nanoseconds)
• Bin progress: per-bin completion for summarize ... by bin() queries
• Planning progress: segment planning completion count
• Operator diagnostics: per-operator key-value telemetry

Use the latest progress frame — each one represents cumulative statistics.

Metadata Frame

Sent after the primary table schema, the metadata frame carries:

• Visualization: type and properties from the render operator (e.g., timechart, piechart)
• Warnings: execution warnings like "summarize memory limit reached" or "result truncated"
• Partial failures: segments that couldn't be read, with IDs and error messages

Headers

Example: Processing a gRPC Stream

use tokio_stream::StreamExt;

let mut stream = client.execute_query(request).await?.into_inner();
let mut tables: HashMap<String, Vec<Row>> = HashMap::new();
let mut current_iteration: Option<String> = None;

while let Some(frame) = stream.next().await {
    let frame = frame?;
    match frame.payload {
        Some(Payload::Schema(s)) => { /* store column definitions */ }
        Some(Payload::Batch(b)) => {
            // New iteration? Clear everything.
            if current_iteration.as_ref() != Some(&b.result_iteration_id) {
                tables.clear();
                current_iteration = Some(b.result_iteration_id.clone());
            }
            // Append rows to the correct table
            tables.entry(b.table_name).or_default().extend(b.rows);
        }
        Some(Payload::Progress(p)) => { /* update stats display */ }
        Some(Payload::Error(e)) => { /* handle error */ }
        Some(Payload::Done(_)) => break,
        _ => {}
    }
}

for {
    frame, err := stream.Recv()
    if err == io.EOF { break }

    switch p := frame.Payload.(type) {
    case *ExecuteQueryResultFrame_Schema:
        // Store column definitions
    case *ExecuteQueryResultFrame_Batch:
        if currentIteration != p.Batch.ResultIterationId {
            tables = map[string][]Row{}  // clear all
            currentIteration = p.Batch.ResultIterationId
        }
        tables[p.Batch.TableName] = append(tables[p.Batch.TableName], p.Batch.Rows...)
    case *ExecuteQueryResultFrame_Progress:
        // Update stats display
    case *ExecuteQueryResultFrame_Error:
        // Handle error
    case *ExecuteQueryResultFrame_Done:
        break
    }
}

async for frame in stub.ExecuteQuery(request, timeout=60):
    payload = frame.WhichOneof("payload")

    if payload == "schema":
        # Store column definitions
    elif payload == "batch":
        if current_iteration != frame.batch.result_iteration_id:
            tables.clear()  # new iteration — discard old rows
            current_iteration = frame.batch.result_iteration_id
        tables[frame.batch.table_name].extend(frame.batch.rows)
    elif payload == "progress":
        # Update stats display
    elif payload == "error":
        raise QueryError(frame.error.code, frame.error.message)
    elif payload == "done":
        break

HTTP REST Protocol (ADX v2)

The HTTP API implements the Kusto v2 frame format, making it compatible with Azure Data Explorer tooling (Grafana ADX plugin, Kusto SDKs, etc.).

The examples below hit the query service directly (:9510, unauthenticated, in-cluster). Through the public gateway the same paths are mounted at the endpoint root and require authentication:

curl -X POST https://<gateway>/v2/rest/query \
  -H 'Authorization: Bearer <token>' \
  -H 'Content-Type: application/json' \
  -d '{"db": "default", "csl": "my_table | take 10"}'

The gateway authenticates the bearer token (CLI access token or service-principal token) and injects the trusted identity before forwarding. Anonymous calls get 401 with a WWW-Authenticate: Bearer challenge — never a login redirect. A non-empty db is required.

Non-Progressive Mode (Default)

Returns all frames in a single JSON array:

curl -X POST http://localhost:9510/v2/rest/query \
  -H 'Content-Type: application/json' \
  -d '{"csl": "my_table | take 10"}'

Response:

[
  {"FrameType": "DataSetHeader", "IsProgressive": false, "Version": "v2.0"},
  {"FrameType": "DataTable", "TableKind": "QueryProperties", "TableName": "@ExtendedProperties", ...},
  {"FrameType": "DataTable", "TableKind": "PrimaryResult", "TableName": "PrimaryResult", "Columns": [...], "Rows": [...]},
  {"FrameType": "DataTable", "TableKind": "QueryCompletionInformation", ...},
  {"FrameType": "DataSetCompletion", "HasErrors": false}
]

Progressive Mode

Enabled by setting results_progressive_enabled: true in request properties. Streams frames via chunked HTTP transfer:

curl -X POST http://localhost:9510/v2/rest/query \
  -H 'Content-Type: application/json' \
  -d '{
    "csl": "my_table | summarize count() by bin(timestamp, 1m)",
    "properties": {"Options": {"results_progressive_enabled": true}}
  }'

Progressive frames use replace semantics — each TableFragment supersedes all previous fragments for the same table:

{"FrameType": "DataSetHeader", "IsProgressive": true, "IsFragmented": true}
{"FrameType": "TableHeader", "TableId": 0, "TableName": "PrimaryResult", "Columns": [...]}
{"FrameType": "TableFragment", "TableId": 0, "TableFragmentType": "DataReplace", "Rows": [[...]]}
{"FrameType": "TableProgress", "TableId": 0, "TableProgress": 45.0}
{"FrameType": "TableFragment", "TableId": 0, "TableFragmentType": "DataReplace", "Rows": [[...]]}
{"FrameType": "TableProgress", "TableId": 0, "TableProgress": 100.0}
{"FrameType": "TableCompletion", "TableId": 0, "RowCount": 150}
{"FrameType": "DataSetCompletion", "HasErrors": false}

Errors

Both modes return HTTP 4xx with a JSON error body for query-level failures (parse errors, execution errors). The error is not wrapped in v2 frames:

{
  "error": {
    "code": "General_BadRequest",
    "message": "Syntax error: ...",
    "@type": "Kusto.Data.Exceptions.KustoBadRequestException"
  }
}

Headers

Column Types

Both protocols use the same type system:

Dynamic Values (gRPC)

In the gRPC protocol, cell values are encoded as TTDynamic protobuf messages with a oneof for each scalar type plus arrays and property bags. See the proto definition for the full schema.

In the HTTP protocol, values are JSON-native — strings, numbers, booleans, nulls, arrays, and objects.