cap query results and sample random order (#121)

Author: philcunliffe

skills/collectivus-query/SKILL.md

@@ -16,7 +16,7 @@ Use `ctvs query` to inspect local Collectivus recordings. It reads local JSONL r
    - **Missing partitions still error.** Run the exact `ctvs query refresh …` command the CLI prints, or rerun the target query with `--refresh always`.
    - Broad manual refreshes are explicit: `ctvs query refresh --all [dataset]`. Do not run a broad refresh when the printed file-targeted command is enough.
    - Pass `--strict-freshness` only when the user explicitly needs the pre-1.7 strict mode (e.g., scheduled checks that must never read stale data); it turns stale partitions back into a hard error.
-4. Prefer structured output for analysis: use `--format json` for follow-up reasoning, `--format markdown` when showing a table to the user, and `--limit` to keep output bounded.
+4. Prefer structured output for analysis: use `--format json` for follow-up reasoning and `--format markdown` when showing a table to the user. Query output is hard-capped at 100 rows (`--limit` defaults to 100 and cannot exceed 100), so use filters, `COUNT(*)`, aggregations, `ORDER BY`, and `OFFSET`/narrower predicates when you need more than the first page. For random samples, use `ORDER BY RANDOM() LIMIT n`; top-level random ordering is reservoir-sampled.
 5. Use high-level query commands before custom SQL. Switch to `ctvs query sql` only when the built-in commands cannot answer the question.
 6. For unfamiliar SQL tables, run `ctvs query schema <table> --format json` before querying. It works for built-in recording tables and tables registered with `ctvs collect`.
 
@@ -115,6 +115,7 @@ Use `JSON_VALUE(<col>, '$.path')` to extract scalars from the `attributes` / `st
 - Do not paste `--config` into every command by habit. Use it when discovery shows the service is not using `~/.hyp/collectivus.json`.
 - Do not read arbitrary Parquet or Iceberg files directly for `ctvs query sql`; the CLI resolves SQL table names and injects only known query tables.
 - Keep SQL read-only and use only query tables from `ctvs query catalog`: built-ins (`logs`, `traces`, `metrics`, `proxy_messages`, `gascity_messages`) and registered collection tables.
+- `ctvs query sql` never returns more than 100 rows, even if the SQL text asks for a larger top-level `LIMIT`. Treat table-shaped results as samples unless the query is an aggregate/count that proves completeness.
 - Use UTC dates with `--date YYYY-MM-DD`; repeat `--date` when the user wants a union across multiple date partitions.
 - Use `--service`, `--gateway-id`, `--from`, `--to`, or `--since` to narrow broad investigations.
 

skills/collectivus-query/references/query-cli.md

@@ -35,7 +35,7 @@ Commands default to `~/.hyp/collectivus.json`. If the running gateway or OTEL co
 - `--date <YYYY-MM-DD>`: Restrict to one UTC date partition. Repeat it to query or refresh multiple days.
 - `--gateway-id <id>`: Restrict to one gateway id.
 - `--service <name>`: Restrict `serviceName` for logs, traces, and metrics.
-- `--limit <n>`: Maximum rows to render. Default `100`, maximum `1000`.
+- `--limit <n>`: Maximum rows to return. Default `100`, maximum `100`.
 - `--format <fmt>`: `table`, `json`, `jsonl`, or `markdown`.
 - `--refresh <mode>`: `never` or `always`. Default `never`.
 - `--all`: Refresh all matching source files for `ctvs query refresh`.
@@ -140,3 +140,7 @@ ctvs query sql "select model, sum(cast(JSON_VALUE(attributes, '\$.usage.input_to
 ```
 
 SQL must be a read-only `select` over known query tables. Table names are resolved from the SQL AST and may be built-ins (`logs`, `traces`, `metrics`, `proxy_messages`, `gascity_messages`) or registered collection tables from `ctvs query catalog`.
+
+`ctvs query sql` hard-caps top-level result sets at 100 rows. If the SQL omits a top-level `LIMIT`, the CLI applies `LIMIT 100`; if the SQL asks for a larger top-level limit, the CLI clamps it to 100. Use aggregates/counts for complete summaries, or add filters and `OFFSET` to page through wider table-shaped results.
+
+Top-level `ORDER BY RANDOM() LIMIT n` uses reservoir sampling instead of sorting the full result set. It is still capped at 100 rows by the normal top-level limit rules.

src/cli/query.js

@@ -71,7 +71,7 @@ Shared options:
   --date <YYYY-MM-DD>            Restrict to one UTC date partition; repeat for multiple days
   --gateway-id <id>              Restrict to one gateway id
   --service <name>               Restrict serviceName for OTLP datasets
-  --limit <n>                    Max rows to render (default: 100, max: 1000)
+  --limit <n>                    Max rows to return (default/max: 100)
   --format <fmt>                 table, json, jsonl, markdown
   --refresh <mode>               never or always (default: never).
                                  Stale partitions query with a stderr warning;
@@ -82,7 +82,7 @@ Shared options:
   --help, -h                     Show this help`
 
 const DEFAULT_LIMIT = 100
-const MAX_LIMIT = 1000
+const MAX_LIMIT = 100
 const DATE_PATTERN = /^\d{4}-\d{2}-\d{2}$/
 
 /**

src/query/random-sample.js

@@ -0,0 +1,155 @@
+import { executeSql, parseSql } from 'squirreling'
+
+/**
+ * @import { AsyncDataSource, AsyncRow, QueryResults, SqlPrimitive, Statement } from 'squirreling'
+ */
+
+/**
+ * @typedef {Record<string, SqlPrimitive>[]} SqlRows
+ */
+
+/**
+ * @typedef {{
+ *   limit: number,
+ *   query: Statement,
+ * }} RandomSamplePlan
+ */
+
+/**
+ * @param {Statement} statement
+ * @returns {Statement}
+ */
+function cloneStatement(statement) {
+  return JSON.parse(JSON.stringify(statement))
+}
+
+/**
+ * @param {Statement} statement
+ * @returns {(import('squirreling').SelectStatement & { orderBy: Array<{ expr?: { type?: string, funcName?: string } }> }) | undefined}
+ */
+function topLevelSelect(statement) {
+  if (statement.type === 'select') return statement
+  if (statement.type === 'with') return topLevelSelect(statement.query)
+  return undefined
+}
+
+/**
+ * @param {ReturnType<typeof topLevelSelect>} select
+ * @returns {boolean}
+ */
+function hasRandomOrder(select) {
+  if (!select || select.orderBy.length !== 1) return false
+  const expr = select.orderBy[0]?.expr
+  return expr?.type === 'function' && expr.funcName?.toUpperCase() === 'RANDOM'
+}
+
+/**
+ * @param {string | Statement} query
+ * @returns {RandomSamplePlan | undefined}
+ */
+export function getRandomSamplePlan(query) {
+  const statement = typeof query === 'string' ? parseSql({ query }) : query
+  const select = topLevelSelect(statement)
+  if (!hasRandomOrder(select) || select?.limit === undefined || select.limit <= 0 || select.offset !== undefined) {
+    return undefined
+  }
+
+  const cloned = cloneStatement(statement)
+  const clonedSelect = topLevelSelect(cloned)
+  if (!clonedSelect) return undefined
+  clonedSelect.orderBy = []
+  clonedSelect.limit = undefined
+
+  return {
+    limit: select.limit,
+    query: cloned,
+  }
+}
+
+/**
+ * @param {AsyncRow} row
+ * @returns {AsyncRow}
+ */
+function memoizeAsyncRow(row) {
+  /** @type {Map<string, Promise<SqlPrimitive>>} */
+  const cellCache = new Map()
+  return {
+    ...row,
+    cells: Object.fromEntries(Object.entries(row.cells).map(([column, accessor]) => [column, () => {
+      if (!cellCache.has(column)) cellCache.set(column, accessor())
+      return cellCache.get(column)
+    }])),
+  }
+}
+
+/**
+ * @param {AsyncRow[]} rows
+ */
+function shuffleRows(rows) {
+  for (let i = rows.length - 1; i > 0; i--) {
+    const j = Math.floor(Math.random() * (i + 1))
+    ;[rows[i], rows[j]] = [rows[j], rows[i]]
+  }
+}
+
+/**
+ * @param {{
+ *   tables: Record<string, AsyncDataSource | SqlRows>,
+ *   query: string | Statement,
+ *   plan?: RandomSamplePlan,
+ * }} args
+ * @returns {AsyncGenerator<AsyncRow>}
+ */
+async function* executeRandomSampleRows(args) {
+  const { plan } = args
+  if (!plan) {
+    yield* executeSql({ tables: args.tables, query: args.query }).rows()
+    return
+  }
+
+  /** @type {AsyncRow[]} */
+  const sample = []
+  let seen = 0
+  for await (const row of executeSql({ tables: args.tables, query: plan.query }).rows()) {
+    seen++
+    if (sample.length < plan.limit) {
+      sample.push(row)
+      continue
+    }
+
+    const replacementIndex = Math.floor(Math.random() * seen)
+    if (replacementIndex < plan.limit) {
+      sample[replacementIndex] = row
+    }
+  }
+
+  shuffleRows(sample)
+  for (const row of sample.map(memoizeAsyncRow)) {
+    yield row
+  }
+}
+
+/**
+ * Execute SQL, replacing top-level `ORDER BY RANDOM() LIMIT n` with reservoir
+ * sampling so the executor does not need to materialize and sort every row.
+ *
+ * @param {{
+ *   tables: Record<string, AsyncDataSource | SqlRows>,
+ *   query: string | Statement,
+ * }} params
+ * @returns {QueryResults}
+ */
+export function executeSqlWithRandomSample(params) {
+  const plan = getRandomSamplePlan(params.query)
+  const baseResults = executeSql({
+    tables: params.tables,
+    query: plan?.query ?? params.query,
+  })
+
+  return {
+    columns: baseResults.columns,
+    rows: () => executeRandomSampleRows({ ...params, plan }),
+    numRows: baseResults.numRows,
+    maxRows: baseResults.maxRows,
+  }
+}

src/query/sql.js

@@ -1,5 +1,5 @@
 import fs from 'node:fs'
-import { asyncRow, collect, executeSql, extractTables, parseSql } from 'squirreling'
+import { asyncRow, collect, extractTables, parseSql } from 'squirreling'
 import { parquetReadObjects } from 'hyparquet'
 import { compressors } from 'hyparquet-compressors'
 import { icebergDataSource, loadLatestFileCatalogMetadata } from 'icebird'
@@ -19,19 +19,20 @@ import {
 import { readCacheCursor } from './iceberg/cursor.js'
 import { createLocalIcebergIO } from './iceberg/resolver.js'
 import { queryCacheTableExists } from './iceberg/store.js'
+import { executeSqlWithRandomSample } from './random-sample.js'
 
 /**
- * @import { AsyncDataSource, AsyncRow, ScanOptions, ScanResults, Statement } from 'squirreling'
+ * @import { AsyncDataSource, AsyncRow, ScanOptions, ScanResults, SelectStatement, SetOperationStatement, Statement } from 'squirreling'
  * @import { CachePartition, QueryDataset, QueryPaths, QueryResultSet, QueryScope, ResolvedQueryTableInfo, ResolvedQueryTables } from './types.js'
  * @import { Lister, Resolver, TableMetadata } from 'icebird/src/types.js'
  */
 
 /**
  * @param {string} sql
- * @param {number} defaultLimit
+ * @param {number} rowLimit
  * @returns {{ statement: Statement, tableNames: string[] }}
  */
-export function prepareReadOnlySql(sql, defaultLimit) {
+export function prepareReadOnlySql(sql, rowLimit) {
   const trimmed = sql.trim()
   if (trimmed.length === 0) throw new Error('SQL query is required')
   /** @type {Statement} */
@@ -42,7 +43,7 @@ export function prepareReadOnlySql(sql, defaultLimit) {
     throw new Error(`SQL must be a single read-only SELECT statement: ${formatError(err)}`)
   }
   const tableNames = uniqueStrings(extractTables(statement))
-  applyDefaultLimit(statement, defaultLimit)
+  applyResultLimit(statement, rowLimit)
   return { statement, tableNames }
 }
 
@@ -62,7 +63,7 @@ export async function executeLogicalSql(args) {
     : args.datasets && args.datasets.length > 0
       ? await buildTables(args.paths, { ...args.scope, datasets: args.datasets })
       : {}
-  const results = executeSql({ tables, query: args.statement })
+  const results = executeSqlWithRandomSample({ tables, query: args.statement })
   const rows = await collect(results)
   return { columns: results.columns, rows }
 }
@@ -347,9 +348,15 @@ async function* scanBuiltinIcebergRows(dataset, sourcesPromise, columns, scope,
   /** @type {Set<string>} */
   const seenRowIds = new Set()
   const sources = await sourcesPromise
+  const innerLimit = builtinInnerScanLimit(scope, options, sources.length)
   for (const { partition, source } of sources) {
     if (options.signal?.aborted) return
-    const scan = source.scan({ columns: innerColumns, where: options.where, signal: options.signal })
+    const scan = source.scan({
+      columns: innerColumns,
+      where: options.where,
+      ...(innerLimit === undefined ? {} : { limit: innerLimit }),
+      signal: options.signal,
+    })
     for await (const sourceRow of scan.rows()) {
       if (options.signal?.aborted) return
       const row = await resolveAsyncRow(sourceRow)
@@ -380,13 +387,19 @@ async function* scanCollectionIcebergRows(table, sourcesPromise, columns, scope,
   /** @type {Set<string>} */
   const seenRowIds = new Set()
   const sources = await sourcesPromise
+  const innerLimit = collectionInnerScanLimit(scope, options, sources.length)
   for (const { source, meta } of sources) {
     if (options.signal?.aborted) return
     const innerColumns = scanColumnsWithPrivateColumns(
       requestedColumns,
       collectionScopeColumns(meta, scope)
     )
-    const scan = source.scan({ columns: innerColumns, where: options.where, signal: options.signal })
+    const scan = source.scan({
+      columns: innerColumns,
+      where: options.where,
+      ...(innerLimit === undefined ? {} : { limit: innerLimit }),
+      signal: options.signal,
+    })
     for await (const sourceRow of scan.rows()) {
       if (options.signal?.aborted) return
       const row = await resolveAsyncRow(sourceRow)
@@ -524,6 +537,30 @@ function canUseCollectionRowCount(scope) {
   return !scope.service && !scope.date && !scope.dates && !scope.from && !scope.to
 }
 
+/**
+ * @param {QueryScope} scope
+ * @param {ScanOptions} options
+ * @param {number} sourceCount
+ * @returns {number | undefined}
+ */
+function builtinInnerScanLimit(scope, options, sourceCount) {
+  if (sourceCount !== 1 || options.limit === undefined || options.offset !== undefined || options.where) return undefined
+  if (scope.service || scope.from || scope.to) return undefined
+  return options.limit
+}
+
+/**
+ * @param {QueryScope} scope
+ * @param {ScanOptions} options
+ * @param {number} sourceCount
+ * @returns {number | undefined}
+ */
+function collectionInnerScanLimit(scope, options, sourceCount) {
+  if (sourceCount !== 1 || options.limit === undefined || options.offset !== undefined || options.where) return undefined
+  if (scope.gatewayId || scope.service || scope.date || scope.dates || scope.from || scope.to) return undefined
+  return options.limit
+}
+
 /**
  * @param {Record<string, unknown>} row
  * @param {CachePartition} partition
@@ -706,9 +743,9 @@ function collectionColumns(partitions) {
  * @param {Statement} statement
  * @param {number} limit
  */
-function applyDefaultLimit(statement, limit) {
+function applyResultLimit(statement, limit) {
   const target = topLevelStatement(statement)
-  if (target && 'limit' in target && target.limit === undefined) {
+  if (target && isLimitableStatement(target) && (target.limit === undefined || target.limit > limit)) {
     target.limit = limit
   }
 }
@@ -718,10 +755,18 @@ function applyDefaultLimit(statement, limit) {
  * @returns {Statement | undefined}
  */
 function topLevelStatement(statement) {
-  if (statement.type === 'with') return statement.query
+  if (statement.type === 'with') return topLevelStatement(statement.query)
   return statement
 }
 
+/**
+ * @param {Statement} statement
+ * @returns {statement is SelectStatement | SetOperationStatement}
+ */
+function isLimitableStatement(statement) {
+  return statement.type === 'select' || statement.type === 'compound'
+}
+
 /**
  * @param {unknown} err
  * @returns {string}