Queries and views
Keep a queryable table of a topic and read it with filters, sorting, and paging
A view is a table that stays up to date with a topic. You say which fields to extract, and the managed plane writes a row for every new message. Use views when agents or services need to look up, filter, or count what happened without reading the whole log. Views need Laser Stack or LaserData Cloud.
Quick example
Register a projection with the fields to extract, bind the readings topic to it, publish a reading, and query the view.
import {
ContentType,
Laser,
ProjectionBindingBuilder,
ProjectionBuilder,
queryResultValueText
} from "@laserdata/laser-sdk"
const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
await using laser = await Laser.connectEnv()
const readings = laser.stream("telemetry").topic("readings")
await readings.ensure(1)
await laser.projections().register(
new ProjectionBuilder("readings_v1.v1")
.name("readings_v1")
.contentType(ContentType.Json)
.fields(["host", "cpu", "status"])
.build()
)
await laser.bindings().apply(
new ProjectionBindingBuilder()
.source("telemetry", "readings")
.allow("readings_v1.v1")
.defaultProjection("readings_v1.v1")
.index("readings_v1")
.build()
)
// The binding is applied asynchronously. Queries fail until the index exists.
const indexReady = () => laser.query("readings_v1").fetch().then(() => true, () => false)
while (!(await indexReady())) await sleep(200)
await readings.publish().json({ host: "node-2", cpu: 91, status: "degraded" }).send()
// The row lands shortly after the publish.
const findDegraded = () => laser.query("readings_v1").whereEq("status", "degraded").fetch()
let degraded = await findDegraded()
while (degraded.rows.length === 0) degraded = await sleep(200).then(findDegraded)
for (const row of degraded.rows) console.log(queryResultValueText(degraded, row, "host"))use laser_sdk::prelude::full::*;
use std::time::Duration;
#[tokio::main]
async fn main() -> Result<(), LaserError> {
let laser = Laser::connect_env().await?;
let readings = laser.stream("telemetry").topic("readings");
readings.ensure(1).await?;
let projection = Projection::builder("readings_v1.v1")
.name("readings_v1")
.content_type(ContentType::Json)
.fields(["host", "cpu", "status"])
.build();
laser.projections().register(projection).await?;
let binding = ProjectionBinding::builder()
.source("telemetry", "readings")
.allow("readings_v1.v1")
.default_projection("readings_v1.v1")
.index("readings_v1")
.build();
laser.bindings().apply(binding).await?;
// The binding is applied asynchronously. Queries fail until the index exists.
while laser.query("readings_v1").fetch().await.is_err() {
tokio::time::sleep(Duration::from_millis(200)).await;
}
let reading = serde_json::json!({"host": "node-2", "cpu": 91, "status": "degraded"});
readings.publish().json(&reading)?.send().await?;
// The row lands shortly after the publish.
loop {
let degraded = laser
.query("readings_v1")
.where_eq("status", "degraded")
.fetch()
.await?;
if !degraded.rows.is_empty() {
for row in °raded.rows {
println!("{:?}", degraded.value_text(row, "host"));
}
return Ok(());
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
}import asyncio
import laser_sdk as ls
async def main() -> None:
async with await ls.Laser.connect_env() as laser:
readings = laser.stream("telemetry").topic("readings")
await readings.ensure(1)
await laser.projections().register(
ls.ProjectionBuilder("readings_v1.v1")
.name("readings_v1")
.content_type("json")
.fields(["host", "cpu", "status"])
.build()
)
await laser.bindings().apply(
ls.ProjectionBindingBuilder()
.source("telemetry", "readings")
.allow("readings_v1.v1")
.default_projection("readings_v1.v1")
.index("readings_v1")
.build()
)
# The binding is applied asynchronously. Queries fail until the index exists.
while True:
try:
await laser.query("readings_v1").fetch()
break
except ls.LaserError:
await asyncio.sleep(0.2)
await readings.publish({"host": "node-2", "cpu": 91, "status": "degraded"}).send()
# The row lands shortly after the publish.
degraded = await laser.query("readings_v1").where_eq("status", "degraded").fetch()
while not degraded.rows:
await asyncio.sleep(0.2)
degraded = await laser.query("readings_v1").where_eq("status", "degraded").fetch()
for row in degraded.rows:
print(degraded.value_text(row, "host"))
asyncio.run(main())The index name readings_v1 is separate from the topic name, so you can version the view without renaming the topic.
Filter and sort
where_eq matches an indexed field exactly. The filter_* family compares any indexed field, and chained filters combine with AND. TypeScript comparison filters take a typed value object.
const hot = await laser
.query("readings_v1")
.filterGte("cpu", { kind: "long", value: 80n })
.orderDesc("cpu")
.limit(20)
.fetch()
console.log(`${hot.rows.length} hot hosts, more pages: ${hot.page.hasMore}`)let hot = laser
.query("readings_v1")
.filter_gte("cpu", 80_i64)
.order_desc("cpu")
.limit(20)
.fetch()
.await?;
println!("{} hot hosts, more pages: {}", hot.rows.len(), hot.page.has_more);hot = await laser.query("readings_v1").filter_gte("cpu", 80).order_desc("cpu").limit(20).fetch()
print(f"{len(hot.rows)} hot hosts, more pages: {hot.page.has_more}")Page through rows
A page holds 50 rows by default and at most 1,000. To read more, chain max_rows(n) (TypeScript: maxRows(n)) and walk the rows with rows(), which fetches pages as you go and stops at the ceiling. rows() without a ceiling fails with an invalid error. Walk many rows has the code in all three languages.
Count and group
Call count(), sum(field), or avg(field) and add group_by([...]) to get one row per group instead of a list of rows. window(field, every_micros) buckets counts by time. Queries and views in depth has the full list.
Know when new data landed
Bind with notify() and read the change feed to query only when the view moved. When one query must see your own earlier writes, chain read_your_writes() (TypeScript: readYourWrites()) on a deployment that supports it.
Good to know
- Registration and binding apply asynchronously. Register the projection before the binding, and wait for a query to succeed before you publish records you expect to see.
- Rows appear shortly after a publish, not at once. Queries are eventually consistent by default.
- Only declared fields are queryable. The projection builder also keeps a copy of each payload with its row, which typed reads such as
fetch_typedneed.index_only()turns that off. - A client with a default stream scopes index and projection names to that stream. Use the same client setup to bind and to query.
- Row retention is set on the binding and is independent of topic retention. By default rows go when the log drops their messages.
- On standalone Apache Iggy, queries return an unsupported error.