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.).

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.