PRQL Query Reference

CSIRT-Pro uses PRQL (Pipelined Relational Query Language) as its query language. PRQL offers an intuitive pipeline-style syntax that is compiled to the query dialect of the underlying columnar log analytics store before execution.

Basic Syntax

PRQL queries flow from top to bottom through a pipeline.

from `<pipeline-name>`
filter <condition>
select {<fields>}
group {<grouping>} (aggregate {<aggregation>})
sort {<ordering>}
take <limit>

Table Names

The table name in from corresponds to the Pipeline name (enclosed in backticks). The organization UUID prefix is added automatically, so users do not need to specify it.

`from`: Select Table

from `firewall_logs`

Specify the Pipeline name in backticks.

from `firewall_logs`
join side:left `auth_logs` (==src_ip)

Use join to combine with another table.

`filter`: Filter Conditions

Comparison Operators

Operator	Description	Example
`==`	Equal	`filter status == 200`
`!=`	Not equal	`filter status != 404`
`>`	Greater than	`filter bytes > 1000`
`<`	Less than	`filter response_time < 500`
`>=`	Greater or equal	`filter status >= 400`
`<=`	Less or equal	`filter status <= 499`
`~=`	Regex match	`filter message ~= "error\\|fail"`

Logical Operators

# AND (chain multiple filter statements)
from `firewall_logs`
filter action == "DENY"
filter src_ip == "192.168.1.100"

# OR (use ||)
from `auth_logs`
filter (status == "failed" || status == "locked")

Time Range Filters

from `firewall_logs`
filter __time > @2025-01-01
filter __time < @2025-01-31

__time Field

__time is the timestamp field present in every table. The UI time range selector adds time filters automatically.

`select`: Choose Fields

from `firewall_logs`
select {__time, src_ip, dst_ip, action, bytes}

Aliases

from `firewall_logs`
select {
  timestamp = __time,
  source = src_ip,
  destination = dst_ip,
  transferred = bytes
}

`derive`: Computed Fields

from `web_access`
derive {
  response_kb = bytes / 1024,
  is_error = status >= 400
}

`group` and `aggregate`: Aggregation

Basic Aggregation

from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {
    count = count this
  }
)
sort {-count}
take 10

Aggregation Functions

Function	Description	Example
`count this`	Count	`count = count this`
`sum <col>`	Sum	`total = sum bytes`
`avg <col>`	Average	`avg_time = avg response_time`
`min <col>`	Minimum	`first = min __time`
`max <col>`	Maximum	`last = max __time`

Aggregation Performance

Fields are extracted from the raw log at query time (schema-on-read). Large GROUP BY aggregations therefore scan the raw message column, so they run slower than they would against a pre-typed columnar layout. This is the deliberate trade for schema flexibility and fast ingestion. Point lookups remain fast.

`sort`: Ordering

# Descending (- prefix)
from `firewall_logs`
sort {-__time}

# Ascending
from `firewall_logs`
sort {bytes}

`take`: Limit Results

from `firewall_logs`
take 100

`join`: Table Joins

from `firewall_logs`
join side:left `auth_logs` (==src_ip)
select {firewall_logs.__time, src_ip, auth_logs.username, action}

Join Type	Description
`join`	INNER JOIN
`join side:left`	LEFT JOIN
`join side:right`	RIGHT JOIN
`join side:full`	FULL OUTER JOIN

Writing Join Conditions

When both sides share the same column name, use the (==column) shorthand.

join `auth_logs` (==src_ip)

When the column names differ, refer to the left side as this and the joined side as that.

from `firewall_logs`
join `indicators` (this.src_ip == that.value)

`window` — Moving Averages, Running Totals, Ranking

window evaluates an aggregate or ranking over a range of rows (e.g. the preceding N rows, or everything from the start up to the current row). It maps to SQL window functions (OVER (...)). Ordering comes from the preceding sort; placing it inside a group computes per partition.

Frame specification

Spec	Description
`rows:-2..0`	Current row plus the 2 preceding rows (moving frame)
`rolling:3`	3-row rolling window
`expanding:true`	Cumulative, from the first row to the current row

Running total (expanding)

from `firewall_logs`
sort {__time}
derive {one = 1}
window expanding:true (
  derive {running_total = sum one}
)

Moving frame (rows / rolling)

from `metrics`
sort {__time}
window rows:-2..0 (
  derive {moving_avg = average value}
)

Ranking (row_number / lag)

from `firewall_logs`
group {src_ip} (
  sort {-bytes}
  window rows:0..0 (
    derive {rank_in_src = row_number this}
  )
)

Functions available inside a window:

Function	Description
`row_number this`	Sequential number within the partition (1, 2, 3, …)
`lag <n> <col>` / `lead <n> <col>`	Value from n rows before / after
`sum` / `average` / `min` / `max` / `count`	Aggregate over the frame

loop (recursion) is not supported

PRQL provides a loop transform that applies a pipeline repeatedly (equivalent to a recursive CTE in SQL). It is not available in the current CSIRT-Pro analytics engine (verified on a live instance — it returns a server error). Use the gen_series built-in function to generate sequences, and lag / lead in window (above) to reference neighboring rows.

Built-in Functions

CSIRT-Pro provides built-in functions that expose common analytics-store functions for convenient use in PRQL queries. These functions are defined and provided by CSIRT-Pro, ready to call directly from PRQL.

Map and JSON Operations

Function	Description	Example
`get_map col key`	Get value from Map field	`derive {val = get_map tags "severity"}`
`json_extract_string col key`	Extract string from JSON	`derive {name = json_extract_string raw "'name'"}`
`array_expand field`	Expand JSON array to rows	`derive {item = array_expand items}`
`make_json_kv k v`	Create JSON from key-value	`derive {j = make_json_kv "key" value}`
`make_json_kv2 k1 v1 k2 v2`	Create JSON from 2 KV pairs	`derive {j = make_json_kv2 "ip" ip "port" port}`
`to_json_array col`	Convert array to JSON string	`derive {json = to_json_array arr}`

Array Operations

Function	Description	Example
`array_join col`	Expand array elements to rows	`derive {ip = array_join ip_list}`
`array_join_newline arr`	Join array with newlines	`derive {text = array_join_newline messages}`
`group_array col`	Aggregate values into array	`aggregate {ips = group_array src_ip}`
`group_uniq_array col`	Aggregate unique values into array	`aggregate {ips = group_uniq_array src_ip}`

Time and Interval Operations

Function	Description	Example
`now`	Current timestamp	`derive {current = now}`
`to_interval_minute n`	N-minute interval	`filter __time > (now) - (to_interval_minute 30)`
`to_interval_hour n`	N-hour interval	`filter __time > (now) - (to_interval_hour 24)`
`to_interval_day n`	N-day interval	`filter __time > (now) - (to_interval_day 7)`
`to_interval_week n`	N-week interval	`filter __time > (now) - (to_interval_week 4)`
`to_interval_month n`	N-month interval	`filter __time > (now) - (to_interval_month 3)`
`to_interval_year n`	N-year interval	`filter __time > (now) - (to_interval_year 1)`

Network Analysis

Function	Description	Example
`check_is_private_ip ip`	Check if IP is private (RFC 1918)	`derive {priv = check_is_private_ip src_ip}`
`network_graph tbl src dst`	Generate network graph JSON	Used for visualizations

Version Comparison

Function	Description	Example
`is_older_version v1 v2`	Compare version strings	`filter (is_older_version version "2.0.0")`
`to_ver_array v`	Convert version to numeric array	`derive {ver = to_ver_array version}`

Type Conversion

Function	Description	Example
`to_int64_or_null col`	Convert string to Int64 (NULL on failure)	`derive {n = to_int64_or_null port}`

Utilities

Function	Description	Example
`gen_series n`	Generate sequence 0 to n-1	`derive {seq = gen_series 10}`
`make_map k v`	Create Map from key-value	`derive {m = make_map "action" action}`
`make_map2 k1 v1 k2 v2`	Create Map from 2 KV pairs	`derive {m = make_map2 "ip" ip "port" port}`

Practical Query Examples

Top 10 Blocked Source IPs

from `firewall_logs`
filter action == "DENY"
group {src_ip} (
  aggregate {count = count this}
)
sort {-count}
take 10

Error Logs in Last Hour

from `app_logs`
filter __time > (now) - (to_interval_hour 1)
filter level == "ERROR"
sort {-__time}

Extract Severity from Tags

from `alerts`
derive {severity = get_map tags "severity"}
filter severity == "high"
sort {-__time}
take 50

Identify External IPs

from `firewall_logs`
derive {is_internal = check_is_private_ip src_ip}
filter is_internal == false
group {src_ip} (
  aggregate {count = count this}
)
sort {-count}
take 20

Security Constraints

Constraint	Description
Table Access Control	Only your organization's Pipelines (or shared ones via SharedAccess) are accessible
s-string Restriction	Raw SQL injection is prevented; `s"..."` is allowed only for `now()` and `toInterval*()`
Forbidden Functions	File-reading functions (`read`) are blocked
UUID Auto-Prefix	Table names are automatically prefixed with your organization UUID

Response Format

Search API responses are returned in a columnar binary format for efficient data transfer. The example below reads that stream into a dataframe.

import pyarrow as pa
import requests

response = requests.post(
    "https://<domain>/api/user/<user_id>/search/",
    headers={"X-API-Key": "<api_key>"},
    json={"expr": "from `firewall_logs` take 10"}
)

reader = pa.ipc.open_stream(response.content)
table = reader.read_all()
df = table.to_pandas()