Help us shape the future of Go! Take the 2026 Go Developer Survey

clickhouse

package
v0.20.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 27, 2026 License: MIT Imports: 24 Imported by: 0

README

weft/obsdb/clickhouse

The hosted backend of weft/obsdb (ADR 0024, S3.6): the observability schema as ClickHouse tables that a stock OTel Collector can also feed.

import (
    "github.com/weftgo/weft/obsdb"
    "github.com/weftgo/weft/obsdb/clickhouse"
)

db, err := clickhouse.Open("clickhouse://default:pass@localhost:9000/weft",
    clickhouse.TTL(30*24*time.Hour, 90*24*time.Hour)) // content, spans+runs
// db implements obsdb.DB: Write, Runs, Run, Events, Transcript,
// RunSpans, Trace, Sessions, Session, ResolvePublicID, SaveExperiment,
// Experiments, Experiment, Close.

ClickHouse 25.6 or later. The schema's TTLs are on DateTime64(9) columns (migration 0001), which a server accepts from 25.6: older ones — 23.3, 23.8, 24.8, 25.3, 25.4 and 25.5 tried — refuse 0001 with "TTL expression result column should have DateTime or Date type, but has DateTime64(9)". CI pins 25.8 (the recipe below).

Open creates and versions the schema in obsdb_migrations (the SQLite backend's numbering rule), writes go through async inserts (async_insert=1, wait_for_async_insert=1), and the weft tables are maintained by materialized views:

  • otel_traces, otel_logs — column-compatible with the OTel Collector ClickHouse exporter pinned at v0.162.0 (the version named in migrations/0001_init.sql): we keep every column its INSERT names, and add the weft identity as materialized columns with bloom skip indexes on SessionId, PublicId, Agent and TraceId. A stock collector pinned to that version, configured with create_schema: false, writes into the same database.
  • weft_records — ReplacingMergeTree(InsertTime) ORDER BY (RunId, Kind, Pos) (the transport idempotency key), filled from otel_logs where weft.record IN ('event', 'messages') — never delta, never heartbeat. Exact reads use FINAL.
  • weft_runs — AggregatingMergeTree ORDER BY RunId filled from both otel_logs (heartbeats included, for last-seen) and otel_traces; Sessions groups it by SessionId.

Options: TTL(content, meta) overrides the retention windows (content 30 days, spans and runs 90 by default; values ≤ 0 keep a class's default), KeepDeltas() stores delta rows in weft_deltas for debugging (counted, never stored by default). A TTL drops an expired row only when ClickHouse merges its part, so every read that lists or opens runs cuts at weft_runs' own window itself (the table's TTL LastSeen + toIntervalSecond(n) clause, read from system.tables at most once a minute): Runs and Search (every filter), Sessions (rows, Total, turn facts, the tag and Failed filters, the summarizer usage), Session and its runs, Run and its children, ChildRuns, the run tree (RunTreeSpans: its rows and the spans of the runs it keeps — a run whose parent is cut is out with it, whether its row names that parent or only core's id rule does; a cut run, a derived one too, not found as in Run) and ResolvePublicID. A resume past it is no resumer, in the same call's instant: its parked run reads parked — no resumed_by, kept by ?status=parked and SessionQuery.Awaiting, its session's pending calls, resumes and outcomes agreeing — as SQLite, which cuts nothing, reads a parked run whose resume DeleteRun removed. A run past it is neither counted nor paged nor found, merged away or not, Total always matches the rows, and no read keeps a turn another has cut; a run's own records and spans, read by id, stay until the merge drops them. The boundary is taken once per call — one instant for every statement of it — so a run crossing it between Sessions' count, page, turn facts and summarizer read is in all of them or in none. It is on the server's clock, the one TTL merges use: the window's read takes the server's now64(9) beside it, and the boundary is the host's now plus that offset minus the window, so a host clock that drifts from the server's moves no run across it. A failed system.tables read keeps the last window known for a minute (logged at Debug), a read the caller cancelled holds nothing; with none known yet, or a window past about 100 years (DateTime's range), the reads cut nothing, as before the cut existed.

The tag filters (SessionQuery.Meta, RunQuery.Meta) narrow their row scan on idx_meta (migration 0026), an ngram bloom filter over Meta: each pair whose key and value are printable ASCII without ", \ or / also asks Meta LIKE '%"<k>":"<v>"%' (the pair's JSON text — a superset; the exact JSONExtractString match decides), so the granules holding none of it are skipped; any other pair reads every granule's Meta, as before.

Deletion (DeleteRun, DeleteSession, Sweep — Studio's DELETE verbs and studio.Retain) is a lightweight DELETE FROM … WHERE on otel_traces, otel_logs, weft_records, weft_deltas and weft_runs (lightweight deletes, GA since ClickHouse 23.3, below the schema's own 25.6 floor): rows are masked at once and leave the parts at the next merge; Vacuum (weft vacuum) forces that with OPTIMIZE TABLE … FINAL on the same tables — heavy on a large database. Each chunk of a deletion (400 runs, 200 span keys) issues up to seven synchronous (mutations_sync = 2) lightweight DELETE mutations, so a large sweep is many mutations. The runs' own spans go after the non-weft spans below them, and weft_runs last: a deletion cut short leaves the run listed and a retry finishes it. The TTLs stay as they are: Studio's retention is additive.

The read-time status parked (RunQuery.Status = obsdb.StatusParked, /api/runs?status=parked) is an anti-join: a run with pending calls whose id no row's ResumedFrom names. That NOT IN (SELECT ResumedFrom … WHERE ResumedFrom <> '') set is built from the ResumedFrom column of every weft_runs row — idx_resumed_from (a bloom filter) serves a lookup of given ids, not "every non-empty value" — so the parked filter reads one String column across the whole table per query (the resumes only are kept in the set, small next to the runs). Under the retention cut the set holds the kept resumes alone: the resumes are collapsed per run (their newest LastSeen) and cut at the call's one boundary, which adds a primary-key read of the resumes' own rows to that scan; SessionQuery.Awaiting reads the same set. On a large table expect that column scan; the other status words read only the grouped rows the order index walks.

DSN forms: clickhouse://user:pass@host:9000/database (native protocol, the port is 9000 not 8123), clickhouses://... or ?secure=1 for TLS, extra query parameters become clickhouse-go settings.

Running the tests

Unit tests run offline. The conformance table (obsdbtest.Run), the collector-shape insert test and the option tests are gated on WEFT_CLICKHOUSE_DSN and need a server; each test creates and drops its own database.

Container recipe (one line)
docker run -d --name weft-clickhouse -p 127.0.0.1:9000:9000 \
  -e CLICKHOUSE_PASSWORD=weft --ulimit nofile=262144:262144 \
  clickhouse/clickhouse-server:25.8-alpine

WEFT_CLICKHOUSE_DSN='clickhouse://default:weft@127.0.0.1:9000/default' \
  go test -race ./...
docker-compose
services:
  clickhouse:
    image: clickhouse/clickhouse-server:25.8-alpine
    environment:
      CLICKHOUSE_PASSWORD: weft
    ports:
      - "127.0.0.1:9000:9000"
    ulimits:
      nofile: 262144:262144
CI job (service container)

The job below is what .github/workflows/ci.yml runs for this module — clickhouse/clickhouse-server as a service container, the DSN pointing at the port the service maps onto the runner. A service container takes docker create flags through options: (GitHub Actions has no ulimits: key — one makes the whole workflow file invalid), and the health check holds the steps until the server answers a query on the native port. A gated test that skips fails the job, so a DSN the tests cannot read never reads green; the same DSN un-gates cmd/weft's --db clickhouse:// test.

clickhouse:
  runs-on: ubuntu-latest
  services:
    clickhouse:
      image: clickhouse/clickhouse-server:25.8-alpine
      env:
        CLICKHOUSE_PASSWORD: weft
      ports:
        - 9000:9000
      options: >-
        --ulimit nofile=262144:262144
        --health-cmd "clickhouse-client --password weft --query 'SELECT 1'"
        --health-interval 5s
        --health-timeout 5s
        --health-retries 20
  env:
    WEFT_CLICKHOUSE_DSN: clickhouse://default:weft@127.0.0.1:9000/default
  steps:
    - uses: actions/checkout@v7
    - uses: actions/setup-go@v7
      with:
        go-version: stable
    - name: test (obsdb/clickhouse, gated suite)
      working-directory: obsdb/clickhouse
      run: |
        set -o pipefail
        go test -race -count=1 -v ./... | tee "$RUNNER_TEMP/clickhouse.log"
        if grep -E '^\s*--- SKIP' "$RUNNER_TEMP/clickhouse.log"; then
          echo "gated tests skipped: WEFT_CLICKHOUSE_DSN did not reach them" >&2
          exit 1
        fi
    - name: test (cmd/weft, gated --db clickhouse:// test)
      working-directory: cmd/weft
      run: |
        set -o pipefail
        go test -race -count=1 -v -run '^TestNewServerClickhouse$' ./... | tee "$RUNNER_TEMP/weft-cmd.log"
        grep -q -- '--- PASS: TestNewServerClickhouse' "$RUNNER_TEMP/weft-cmd.log"

Documentation

Overview

Package clickhouse — see doc.go for the package documentation.

Package clickhouse is obsdb's hosted backend (ADR 0024, S3.6): the observability schema as ClickHouse tables that a stock OTel Collector can also feed.

The stance: column-compatible, not DDL-verbatim

otel_traces and otel_logs keep every column name and type the OTel Collector's ClickHouse exporter INSERT expects — pinned at exporter v0.162.0, the version recorded in migrations/0001_init.sql — with the weft identity materialized beside them (skip-indexed bloom filters on SessionId, PublicId, Agent and TraceId, S3.6). The exporter's README tells production users to manage their own schema (create_schema: false); ours owns the engines, the TTLs and the weft columns, and a stock collector pinned to that version can write into the same database — the collector-shape test inserts with its exact column lists. Attribute values in the collector's map columns are strings by that schema's design; the package writes the typed maps into its own JSON columns (WeftAttrs and friends, defaulted empty so collector inserts never name them) and reads them back exactly, falling back to the map columns for collector-written rows.

The weft tables

weft_records is a ReplacingMergeTree(InsertTime) ORDER BY (RunId, Kind, Pos) — the transport idempotency key (I4) — filled by a materialized view from otel_logs where weft.record IN ('event', 'messages'): never 'delta' (D3: counted in weft_runs, never stored) and never 'heartbeat' (no position; it only moves last-seen). It carries its own content TTL, because a source table's TTL does not cascade through a view. Exact reads use FINAL or LIMIT 1 BY on the dedup key; weft records are immutable by construction, so dedup only ever collapses transport duplicates.

weft_runs is an AggregatingMergeTree ORDER BY RunId filled by two materialized views, from otel_logs (heartbeats included, for last-seen) and otel_traces: min for started, max for finished, last-seen, identity strings, the terminal flags and usage — only one non-empty value exists per run, so max never invents one, and a retried batch's identical bytes cannot move a max. Event and message counts are derived at read from weft_records (uniqExact over the dedup key), which a retry cannot inflate. Sessions are a GROUP BY SessionId over weft_runs. A run derived from stock GenAI spans (docs/record-contract.md §9) has no view: Write derives it in Go from the trace's stored spans (obsdb.DeriveStockRuns) and inserts its row with Derived = 1 (migration 0009 says why a view cannot).

Open, writes, reads

Open(dsn, opts...) returns an obsdb.DB over the DSN's database, creating and versioning the schema in obsdb_migrations (the SQLite backend's numbering rule). Writes go through async inserts (async_insert=1, wait_for_async_insert=1), batched per Write call. Lists read through GROUP BY RunId with the same aggregates; single run pages read FINAL. Status is never stored: every read derives it through obsdb.DeriveStatus. KeepDeltas() turns delta storage on for debugging (rows into weft_deltas, nothing else feeds it); TTL(...) overrides the retention windows (content 30 days, spans and runs 90 by default) with ALTER TABLE ... MODIFY TTL.

Index

Examples

Constants

View Source 
const (
	DefaultContentTTL = 30 * 24 * time.Hour
	DefaultMetaTTL    = 90 * 24 * time.Hour
)

DefaultContentTTL and DefaultMetaTTL are S3.6's retention defaults: content-bearing rows (otel_logs, weft_records, weft_deltas, weft_session_events) 30 days, spans and runs 90 days.

Variables

View Source 
var ErrNewerSchema = errors.New("clickhouse: database schema is newer than this obsdb")

ErrNewerSchema is returned by Open when the database's obsdb_migrations is ahead of this binary's highest migration: a database written by a newer weft fails loudly instead of silently ignoring tables it does not know.

Functions

func Open 

func Open(dsn string, opts ...Option) (obsdb.DB, error)

Open connects using a clickhouse-go DSN (clickhouse://user:pass@host:9000/database?...; clickhouses:// or secure=1 for TLS), brings the schema up to date, applies TTL overrides, and returns an obsdb.DB over the DSN's database. The connection carries async_insert=1 and wait_for_async_insert=1 so batched writes return once the server has flushed them.

Example

Opening the backend: a clickhouse-go DSN — the database in the path is the one the schema lands in — plus options composed at call time, here a one-week spans-and-runs window with the 30-day content default kept, and delta storage on for a debugging session. Open brings the schema up to date and returns the obsdb.DB every backend shares, so the reads are the same calls the SQLite backend takes.

This example carries no // Output: because it needs a server (the README has the one-line container recipe); go test compiles it but does not execute it. 

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/weftgo/weft/obsdb"
	"github.com/weftgo/weft/obsdb/clickhouse"
)

func main() {
	db, err := clickhouse.Open(
		"clickhouse://default:weft@127.0.0.1:9000/weft?dial_timeout=10s",
		clickhouse.TTL(0, 7*24*time.Hour),
		clickhouse.KeepDeltas(),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer func() { _ = db.Close() }()

	page, err := db.Runs(context.Background(), obsdb.RunQuery{Agent: "greeter"})
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("greeter runs:", page.Total)
}

Types

type DB 

type DB struct {
	// contains filtered or unexported fields
}

DB is the ClickHouse backend. Create it with Open; the zero value is not usable. It is safe for concurrent use: the driver pools connections, writers batch per Write.

func (*DB) BreakpointSet added in v0.20.0 

func (d *DB) BreakpointSet(ctx context.Context, runtimeID string) (_ obsdb.BreakpointSet, _ bool, err error)

BreakpointSet answers runtimeID's stored set with its provenance (obsdb.BreakpointProvenanceStore).

func (*DB) Breakpoints added in v0.15.0 

func (d *DB) Breakpoints(ctx context.Context, runtimeID string) ([]string, bool, error)

Breakpoints answers runtimeID's stored set.

func (*DB) Catalogs added in v0.10.0 

func (d *DB) Catalogs(ctx context.Context, runID string) (_ []obsdb.ToolsRecord, err error)

func (*DB) ChildRuns added in v0.17.0 

func (d *DB) ChildRuns(ctx context.Context, parentIDs []string) (_ map[string][]obsdb.RunRow, err error)

ChildRuns is obsdb.TreeReader's: the parents' children, merged rows (FINAL) narrowed by primary key to the runs some row of which names one of the parents — idx_parent_run (0020) skips the granules none does — one statement per idChunk of parent ids, grouped in Go; the counts filled as Run's children are. Each statement reads at most what is left of obsdb.ChildRunsCap and one more, the one more answering obsdb.ErrChildRunsCapped.

func (*DB) Close 

func (d *DB) Close() error

Close closes the connection. Further calls return obsdb.ErrClosed.

func (*DB) Commands added in v0.15.0 

func (d *DB) Commands(ctx context.Context, q obsdb.CommandQuery) (_ []obsdb.StoredCommand, err error)

Commands answers the rows q selects, newest first.

func (*DB) Compactions added in v0.10.0 

func (d *DB) Compactions(ctx context.Context, runID string) (_ []obsdb.Compaction, err error)

Compactions reads the run's compaction records (ADR 0028 §8) from otel_logs, the one table that keeps their attributes: weft_records holds a view's Reason but not its range or hash, and thread's session marker (kind compaction) is outside weft_records_mv's kinds. The scan is bounded by the run's time window from weft_runs. otel_logs is a plain MergeTree: a retried batch's duplicate is dropped here, by (kind, index) — by hash for a marker.

func (*DB) CountLogs added in v0.16.0 

func (d *DB) CountLogs(ctx context.Context, runID string, minSeverity int) (_ obsdb.LogCounts, err error)

CountLogs counts a run's app log records at or above minSeverity (obsdb.CountOtherLogs), the candidates narrowed by SeverityNumber in the query.

func (*DB) DeleteRun added in v0.14.0 

func (d *DB) DeleteRun(ctx context.Context, id string) (err error)

DeleteRun removes a run and its subagent children (obsdb.DB).

func (*DB) DeleteRuntime added in v0.19.0 

func (d *DB) DeleteRuntime(ctx context.Context, runtimeID string) (err error)

DeleteRuntime removes every stored version of one runtime's row and of its breakpoint set (a lightweight DELETE by key, synchronous). The caller serialises it with its own writes of that runtime, as it does PruneCommands.

func (*DB) DeleteSession added in v0.14.0 

func (d *DB) DeleteSession(ctx context.Context, id string) (err error)

DeleteSession removes every run of a session, children included (obsdb.DB).

func (*DB) ErrorCounts added in v0.19.0 

func (d *DB) ErrorCounts(ctx context.Context, q obsdb.SearchQuery) (_ []obsdb.ErrorCount, _ bool, err error)

ErrorCounts answers obsdb.Aggregator over Search's match.

func (*DB) Events 

func (d *DB) Events(ctx context.Context, runID string, after int64, limit int) (_ obsdb.EventPage, err error)

func (*DB) Experiment 

func (d *DB) Experiment(ctx context.Context, id string) (_ obsdb.Experiment, err error)

Experiment returns one experiment; ErrNotFound otherwise.

func (*DB) Experiments 

func (d *DB) Experiments(ctx context.Context) (_ []obsdb.Experiment, err error)

Experiments lists the saved definitions, newest update first.

func (*DB) Manifest added in v0.16.0 

func (d *DB) Manifest(ctx context.Context, hash string) (_ obsdb.StoredManifest, _ bool, err error)

Manifest answers the manifest stored under hash: its earliest row.

func (*DB) OtherLogs added in v0.10.0 

func (d *DB) OtherLogs(ctx context.Context, runID string, q obsdb.LogQuery) (_ obsdb.LogPage, err error)

OtherLogs reads a run's app log records from otel_logs: Write stores every record there, and an app's own is one with no weft.run.id that is no session event (a thread record or compaction marker filed under no run with a weft.session.id is the session's: weft_session_events reads it; one with no session id is an app log, as SQLite's writer files it) — the selection SQLite's writer makes into other_logs (obsdb.ReadOtherLogs: the attribution through the run's spans, the order, the paging). The scan is bounded by the run's time window (otel_logs' order key leads with time) and the run's trace ids (its bloom filter), the first obsdb.MaxLogCandidates by time. otel_logs keeps the collector's string map only, so attribute values read back as strings.

func (*DB) Prompt added in v0.10.0 

func (d *DB) Prompt(ctx context.Context, runID, hash string) (_ obsdb.PromptRecord, err error)

func (*DB) PruneCommands added in v0.15.0 

func (d *DB) PruneCommands(ctx context.Context, before time.Time, keep int) (_ int, err error)

PruneCommands applies the store's retention. Each rule selects its keys over the FINAL rows and, in the same query, every stored version of them (key, InsertTime); the DELETE then names exactly those versions. A Put landing between the select and the DELETE writes a version the select never saw, so its row survives with the newest write — the runtime that re-registered, the route a live run's ack just named. The rule's predicate is not what the DELETE filters by: a lightweight DELETE filters per version, so a predicate would leave an older version of a pruned row standing (a finished command's queued version) for FINAL to read back. The prune is still not one transaction (obsdb.CommandStore.PruneCommands): a version of a row the select saw that commits with the same nanosecond InsertTime is not told apart.

func (*DB) PutBreakpointSet added in v0.20.0 

func (d *DB) PutBreakpointSet(ctx context.Context, runtimeID string, bp obsdb.BreakpointSet) (err error)

PutBreakpointSet replaces runtimeID's set whole, with its provenance (obsdb.BreakpointProvenanceStore).

func (*DB) PutBreakpoints added in v0.15.0 

func (d *DB) PutBreakpoints(ctx context.Context, runtimeID string, tools []string) error

PutBreakpoints replaces runtimeID's set whole.

func (*DB) PutCommand added in v0.15.0 

func (d *DB) PutCommand(ctx context.Context, c obsdb.StoredCommand) (err error)

PutCommand upserts one command row whole (an insert; FINAL reads the newest).

func (*DB) PutManifest added in v0.16.0 

func (d *DB) PutManifest(ctx context.Context, m obsdb.StoredManifest) (err error)

PutManifest stores m unless its hash is stored. Two racing writers may both insert; the rows' Ver (2^63 - FirstNs) and Manifest's argMin keep the earlier one, and the body is the same by construction.

func (*DB) PutRoute added in v0.15.0 

func (d *DB) PutRoute(ctx context.Context, r obsdb.RunRoute) (err error)

PutRoute upserts one run → runtime route.

func (*DB) PutRuntime added in v0.15.0 

func (d *DB) PutRuntime(ctx context.Context, r obsdb.StoredRuntime) (err error)

PutRuntime upserts one runtime's registration whole.

func (*DB) Requests added in v0.10.0 

func (d *DB) Requests(ctx context.Context, runID string, q obsdb.RequestQuery) (_ []obsdb.RequestRecord, err error)

func (*DB) ResolvePublicID 

func (d *DB) ResolvePublicID(ctx context.Context, publicID string) (_ string, err error)

func (*DB) Route added in v0.15.0 

func (d *DB) Route(ctx context.Context, runID string) (_ obsdb.RunRoute, _ bool, err error)

Route answers runID's route.

func (*DB) Run 

func (d *DB) Run(ctx context.Context, id string) (_ obsdb.RunDetail, err error)

func (*DB) RunSpans 

func (d *DB) RunSpans(ctx context.Context, runID string) (_ []obsdb.Span, err error)

func (*DB) RunTreeSpans added in v0.17.0 

func (d *DB) RunTreeSpans(ctx context.Context, q obsdb.TreeQuery) (_ obsdb.RunTree, err error)

RunTreeSpans is obsdb.TreeReader's. A weft run's subtree is found from the run rows before a span is read: the run and the rows core's id rule names its descendants (treeIDRows: weft_runs' primary key, RunId = ? OR startsWith), the rows whose ParentRunID is one of them (treeChildRows: idx_parent_run, 0020, the same set as a subquery), and one more level per non-Subagent child found (a literal list); obsdb.TreeMembers places a row whose parent chain does not reach the run by the id rule. Then the members' spans are read by RunId (idx_weft_run_id, 0016), in any trace, capped (treeSpans, obsdb.CapTree): the run's trace is never read whole. Every span with a weft.run.id has its run's row (weft_runs_traces_mv), so the rows name every run a span of the tree can belong to. A derived run (no run id on its spans) is cut from its trace (traceSpans), as RunSpans reads it. A Subagent-only tree costs four statements: the run's row, the id-rule rows, their children, the spans; a deeper chain of non-Subagent children one statement more per level, bounded by the tree's own depth. A derived run reads its trace's first obsdb.TreeTraceCap spans (stockTreeSpans). Every row statement and the id-rule span read take the call's retention cut (retention.go), so the tree holds the runs Run's children, ChildRuns and the cost reads hold: a child past weft_runs' TTL is out of all of them alike, merged away or not, and a run past it — a derived one too — is ErrNotFound here as in Run, checked before any other read. A run whose parent was cut is out too: the walk follows kept rows. The id-rule row read alone is uncut, its cut rows kept as obsdb.TreeCutParent, so the id rule stops at a cut run: its descendants stay out whether their rows name it or only their ids do.

func (*DB) Runs 

func (d *DB) Runs(ctx context.Context, q obsdb.RunQuery) (_ obsdb.RunPage, err error)

func (*DB) Runtimes added in v0.15.0 

func (d *DB) Runtimes(ctx context.Context) (_ []obsdb.StoredRuntime, err error)

Runtimes answers every stored runtime, most recently seen first.

func (*DB) SaveExperiment 

func (d *DB) SaveExperiment(ctx context.Context, e obsdb.Experiment) (err error)

SaveExperiment upserts one experiment by id (an insert; FINAL reads the newest).

func (*DB) Search added in v0.14.0 

func (d *DB) Search(ctx context.Context, q obsdb.SearchQuery) (_ obsdb.RunPage, err error)

Search answers obsdb.Searcher in one query (plus a count only when the page is past the match): the run list Runs reads, narrowed by the cheap filters, joined once to the search fields of exactly those runs — prompt and answer from one pass over their messages records, the tool names from one pass over their tool_start events — and every term tested against the joined row.

func (*DB) Session 

func (d *DB) Session(ctx context.Context, id string) (_ obsdb.SessionDetail, err error)

func (*DB) SessionEvents added in v0.18.0 

func (d *DB) SessionEvents(ctx context.Context, q obsdb.SessionEventQuery) (_ obsdb.SessionEventPage, err error)

SessionEvents pages a session's thread records and compaction markers from weft_session_events (obsdb.SessionEventReader; migration 0025): one session (its public id's under q.PublicID), FINAL and one row per (run, kind, BodyHash) so a retried batch's duplicate reads once, in (time, run, kind, body) order — the SQLite reader's order, body the tie-break two records of one time, run and kind need; the table's key (SessionId first) makes the scan a range read of the session's prefix.

func (*DB) Sessions 

func (d *DB) Sessions(ctx context.Context, q obsdb.SessionQuery) (_ obsdb.SessionPage, err error)

Sessions is a GROUP BY SessionId over weft_runs' top-level, non-playground runs (S3.6): turns, first/last seen, summed usage, and the newest turn's derived status — argMax on (Turn, Started) picks the same row SQLite's ORDER BY ... LIMIT 1 does.

The runs collapse per RunId first (the same aggregates the Runs list uses) because unmerged AggregatingMergeTree parts would otherwise double-count usage in the outer sum; the session level's aliases are prefixed s* so no ClickHouse alias substitution can shadow the argMax arguments (aliases are query-global).

func (*DB) Stats added in v0.19.0 

func (d *DB) Stats(ctx context.Context, q obsdb.StatsQuery) (_ obsdb.StatsPage, err error)

Stats answers obsdb.Aggregator.

func (*DB) Sweep added in v0.14.0 

func (d *DB) Sweep(ctx context.Context, before time.Time) (n int, err error)

Sweep deletes the runs not running whose LastSeen is before the cutoff, with their children that are not running (obsdb.DB): a candidate whose parent is a candidate too is reached through it. The candidates are read once (one row of two strings each); the deletion goes in passes of obsdb.SweepBatch roots, so a large backlog is a bounded round of mutations at a time rather than one storm.

func (*DB) Tools added in v0.10.0 

func (d *DB) Tools(ctx context.Context, runID, hash string) (obsdb.ToolsRecord, error)

func (*DB) Trace 

func (d *DB) Trace(ctx context.Context, traceID string) (_ []obsdb.Span, err error)

Trace returns any trace's spans — weft and non-weft alike (the polyglot promise). An unknown trace id returns no rows and no error, like the SQLite backend.

func (*DB) Traces added in v0.14.0 

func (d *DB) Traces(ctx context.Context, q obsdb.TraceQuery) (_ obsdb.TracePage, err error)

Traces groups otel_traces by TraceId (obsdb.DB.Traces), every column an aggregate over the trace's span rows — idempotent under a retried batch's duplicate rows (uniqExact counts each SpanId once; argMin, min, max and groupUniqArray ignore a repeat) — and filters the groups, never the rows: a trace's Start is min(Timestamp) over all its spans. Times bind as integer nanoseconds (Runs' reason: a positional time.Time renders at Seconds scale).

func (*DB) Transcript 

func (d *DB) Transcript(ctx context.Context, runID string) ([]json.RawMessage, error)

func (*DB) TranscriptBatches added in v0.10.0 

func (d *DB) TranscriptBatches(ctx context.Context, runID string) (_ []obsdb.TranscriptBatch, err error)

func (*DB) Vacuum added in v0.14.0 

func (d *DB) Vacuum(ctx context.Context) (err error)

Vacuum runs OPTIMIZE TABLE … FINAL on the tables deletions mutate (obsdb.DB): every part is merged, which is what drops rows a lightweight DELETE masked. Heavy on a large database — the sweep never calls it; `weft vacuum` does, on the operator's word.

func (*DB) Write 

func (d *DB) Write(ctx context.Context, b obsdb.Batch) (err error)

Write stores one batch: spans into otel_traces and records into otel_logs — both in the collector's own column shape, so the row a stock collector would have written and the row weft writes are the same shape (the materialized views cannot tell them apart). Deltas ride otel_logs (the runs view counts them; the records view filters them) and, with KeepDeltas, are additionally inserted into weft_deltas directly — the option lives in Write, not the schema, because a view cannot be option-gated.

Idempotence (I4) is the engines' job: otel_logs tolerates duplicate rows, weft_records' ReplacingMergeTree collapses the (RunId, Kind, Pos) duplicates at merge (exact reads use FINAL), and weft_runs' min/max aggregates are unchanged by identical values, so a retried transport's duplicate is a no-op once merged and never inflates a count at read time (uniqExact over the key dedups pre-merge).

A stock GenAI span (no weft.run.id, obsdb.StockCandidate) re-derives its trace's derived runs after the spans are stored (deriveStock, docs/record-contract.md §9): no view can, migration 0009 says why.

type Option 

type Option interface {
	// contains filtered or unexported methods
}

Option configures Open.

func KeepDeltas 

func KeepDeltas() Option

KeepDeltas turns delta storage on for debugging (Q4, closed): deltas are counted in weft_runs' DeltaCount high-water mark but never stored by default — the weft_records view filters them out by construction. With this option Write also inserts each delta into weft_deltas (ReplacingMergeTree on (RunId, Pos)), which nothing else feeds; the table stays empty without the option.

func TTL 

func TTL(content, meta time.Duration) Option

TTL overrides the retention windows. content applies to otel_logs, weft_records and weft_deltas; meta applies to otel_traces and weft_runs. Values ≤ 0 keep the default for that class (DefaultContentTTL / DefaultMetaTTL), so TTL(0, 7*24*time.Hour) changes only the spans-and-runs window. Open applies non-default windows with ALTER TABLE ... MODIFY TTL after the migrations (a no-op on the server when the clause is unchanged). An Open with both classes at their defaults alters nothing: the tables keep whatever windows they already have — the migration's defaults on a fresh database, or an earlier Open's override — so a process that opens the database without TTL (the studio binary) never undoes the windows the writer set. Windows are whole seconds, rounded up.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.