Aggregations

Count, group, sort, and paginate your results.

count()

Returns the number of records matching the pipeline:

from("users") | count()
  from("orders") | filter(amount > 100) | count()

group()

Group records by a field and compute aggregations on each group:

from("orders") | group(status, count(), sum(amount))

Supported aggregations:

• count() — number of records in each group
• sum(field) — sum of field values
• avg(field) — average of field values
• min(field) — minimum field value
• max(field) — maximum field value

from("sensors") | group(zone, avg(temperature), max(temperature))

window()

Aggregate records over sliding time windows. The window duration is specified in milliseconds:

// Count records per 10-second window
  from("events") | window(10000, count)
  // Average value per minute
  from("sensors") | window(60000, average)
  // Sum values per 5-minute window
  from("transactions") | window(300000, sum)

sort()

Sort results by a field in ascending or descending order:

from("orders") | sort(amount, desc)
  from("orders") | sort(timestamp, asc)

limit()

Cap the number of records returned. Combine with sort() to get top-N results:

from("logs") | limit(100)
  from("orders") | sort(amount, desc) | limit(10)

Pagination

Two pagination strategies are available:

Offset Pagination

from("users") | page(1, 50)
  from("users") | page(2, 50)

Cursor Pagination

Cursor-based pagination avoids drift from concurrent writes. Use a record key or timestamp as the cursor:

from("events") | page(cursor("2026-05-30T11:42:00Z"), 50)