service

package
v1.0.0-alpha.13 Latest Latest
Warning

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

 Go to latest Published: Oct 15, 2026 License: MIT Imports: 43 Imported by: 0

Documentation

Overview

Package service provides the orchestration layer for pulse operations.

Index

Constants

View Source 
const (
	// RegroupIncomingFlattened: the archive is ungrouped (0x01) and the
	// arriving shard was grouped; it was stored as its logical (0x01)
	// twin. Only the arriving shard was rewritten.
	RegroupIncomingFlattened = "incoming_flattened"
	// RegroupIncomingRegrouped: the archive is grouped and the arriving
	// shard declared a different group layout (none, or other groups);
	// it was re-encoded to the archive's layout. Only the arriving shard
	// was rewritten.
	RegroupIncomingRegrouped = "incoming_regrouped"
	// RegroupConstantPromoted: the arriving shard disagreed with a
	// CONSTANT group's one value, so that group became an indexed group
	// across the WHOLE archive — every shard re-encoded (each row gains a
	// 4-byte index).
	RegroupConstantPromoted = "constant_promoted"
)

Reasons a GroupReconciliation records.

Variables

This section is empty.

Functions

func MapRequestTimeout 

func MapRequestTimeout(ctx context.Context, err error) error

MapRequestTimeout turns err into PULSE_LIMIT_EXCEEDED {limit: request_timeout, configured, observed: elapsed ns} when ctx's RequestTimeout deadline fired; any other error — a caller's context.DeadlineExceeded / context.Canceled included — is returned unchanged. A nil err stays nil: a call that finished is a success even if the deadline passed after its last poll.

func SplitAnchorPath 

func SplitAnchorPath(path string) (string, string, bool)

SplitAnchorPath is the exported form of splitAnchorPath, used by the pulse facade (Predict, Inspect) so anchor paths resolve consistently across every entry point. See splitAnchorPath for semantics.

func WithExecInfo 

func WithExecInfo(ctx context.Context, ei *ExecInfo) context.Context

WithExecInfo returns ctx carrying ei. The facade calls it only when observability is on.

Types

type AddShardResult 

type AddShardResult struct {
	// Archive is the archive path that was rewritten.
	Archive string `json:"archive"`
	// Added is the source path of the shard that was appended.
	Added string `json:"added"`
	// ShardCount is the archive's shard count after the add.
	ShardCount int `json:"shard_count"`
	// Widened carries one entry per set field promoted to a wider rung;
	// empty on the ordinary append.
	Widened []SetWidening `json:"widened,omitempty"`
	// Regrouped carries one entry per parent-group layout change the add
	// made (see GroupReconciliation); empty on the ordinary append,
	// including an append whose group dictionary merely union-merged.
	Regrouped []GroupReconciliation `json:"regrouped,omitempty"`
	// Warnings carry every non-fatal diagnostic the add produced —
	// PULSE_SHARD_DESCRIPTION_DIVERGENCE from cohesion, one
	// PULSE_SHARD_SET_WIDENED per entry in Widened and one
	// PULSE_SHARD_GROUPS_REWRITTEN per entry in Regrouped. Shaped like the
	// descriptor envelope's warning entries so a caller lifts them
	// straight onto `warnings` without reshaping.
	Warnings []encx.CohesionWarning `json:"warnings"`
}

AddShardResult is what AddShard reports.

AddShard returns a result rather than a bare error specifically so the widen warning cannot be dropped. Auto-widening is a whole-archive byte rewrite: it is far cheaper than the full re-import it replaces, but it is not free, and a caller who never learns it happened cannot tell an O(1) append from an O(archive) rewrite. The warning is mandatory, so the channel carrying it is mandatory too.

type BuildIndexResult 

type BuildIndexResult struct {
	IndexPath string
	Index     *encx.Index

	// ManifestPath is the keyless sidecar index manifest this build
	// upserted its entry into (encoding.IndexManifestPath). Additive —
	// it names the ONE path a caller can read to discover this index's
	// key tuple without knowing the tuple already, which the sidecar's
	// own hash-derived filename makes impossible.
	ManifestPath string
}

BuildIndexResult is the outcome of a successful point-lookup sidecar index build: the derived sidecar path it was written to (via encoding.SidecarIndexPath) plus the in-memory encoding.Index that was serialized there, in case a caller wants to inspect it (or run assertions against it in a test) without a round-trip read.

type ChildStart 

type ChildStart func(ctx context.Context, index int, req *types.Request) (context.Context, func(error))

ChildStart starts a child operation — slot index of a Compose, stage index of a ProcessChain — running req under the parent carrier's context. It returns the context the child runs under (carrying the child's own carrier) and the function that ends it with the child's error. The facade supplies it (SetChildStart) for operations that fan out; the service calls it through startChild.

type Cohort 

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

Cohort represents an opened .pulse file with its parsed schema. For shard archives Shards is populated in central-directory order (which equals shard insertion order). For single-file cohorts Shards is empty.

func (*Cohort) OpenRecordReader 

func (c *Cohort) OpenRecordReader() (*CohortRecordReader, error)

OpenRecordReader opens an index-addressed record reader over the cohort: a single-file cohort (ungrouped 0x01 and grouped 0x02), a whole shard archive, or one anchored shard.

func (*Cohort) RecordCount 

func (c *Cohort) RecordCount() (int64, error)

RecordCount returns the number of records in the cohort file. It reads the file and counts records based on the schema's per-record byte size.

func (*Cohort) Records 

func (c *Cohort) Records(_ context.Context) (processing.RecordIterator, error)

Records returns a streaming iterator over records in the cohort. Records are decoded lazily from disk — the full file is not materialized.

func (*Cohort) Schema 

func (c *Cohort) Schema() *encoding.Schema

Schema returns the cohort's schema.

func (*Cohort) Shards 

func (c *Cohort) Shards() []ShardEntry

Shards returns the shard manifest for an archive-backed cohort. Returns an empty slice for single-file cohorts. The returned slice is a defensive copy; callers may mutate freely.

type CohortRecordReader 

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

CohortRecordReader reads records of an opened cohort by index, returning each as a freshly allocated []any in schema LOGICAL field order holding the EXACT stored value (see exactValue for the per-type Go mapping). A grouped (0x02) cohort decodes identically to its ungrouped twin.

A single-file cohort is one segment. A shard archive is one segment per shard in archive (central-directory) order, addressed through a global index: prefix sums over the shard record counts locate record i's shard by binary search, and the shard's records decode against the archive's CANONICAL schema, exactly as the shard iterator does. An anchored shard (`archive.pulse#shard.pulse`) is a one-segment archive read and also decodes against the canonical schema.

RecordAt is safe for concurrent use: every call drives its own io.SectionReader (private cursor) over one shared io.ReaderAt, so no seek cursor is shared. Close waits for in-flight reads and makes every later RecordAt fail with SERVICE_RESOURCE.

func (*CohortRecordReader) Close 

func (r *CohortRecordReader) Close() error

Close releases the file handle. It is idempotent.

func (*CohortRecordReader) Len 

func (r *CohortRecordReader) Len() int64

Len returns the number of records the reader addresses (for an archive, the sum over its shards).

func (*CohortRecordReader) RecordAt 

func (r *CohortRecordReader) RecordAt(i int64) ([]any, error)

RecordAt decodes record i. An index outside [0, Len()) is SERVICE_VALIDATION; a call after Close is SERVICE_RESOURCE.

func (*CohortRecordReader) Schema 

func (r *CohortRecordReader) Schema() *encoding.Schema

Schema returns the cohort schema the reader decodes against; row slots are indexed by Schema().Fields. For an archive or an anchored shard it is the archive's canonical schema.

type ComposeOptions 

type ComposeOptions struct {
	// MaxWorkers caps concurrent in-flight Process calls. Zero means
	// runtime.GOMAXPROCS(0). Negative values are clamped to 1.
	MaxWorkers int

	// PerRequestTimeout, if positive, derives a context.WithTimeout for
	// each request. Zero means no per-request timeout (the parent ctx's
	// deadline still applies).
	PerRequestTimeout time.Duration

	// FailFast cancels in-flight siblings on the first request error.
	// Default is true: surface errors quickly. Set false to collect every
	// request's outcome (errors aggregated into a single CodedError with
	// per-index detail).
	FailFast bool
}

ComposeOptions controls parallel execution of a ComposedRequest.

Order of responses is always preserved — slot-by-index — regardless of MaxWorkers or completion order.

type CreateShardArchiveResult 

type CreateShardArchiveResult struct {
	// Archive is the archive path that was written.
	Archive string `json:"archive"`
	// ShardCount is the number of shards the archive holds.
	ShardCount int `json:"shard_count"`
	// Widened carries one entry per set field promoted to a wider rung
	// while the archive was being assembled; empty on the ordinary
	// create.
	Widened []SetWidening `json:"widened,omitempty"`
	// Regrouped carries one entry per parent-group layout change the
	// create made while seeding (see GroupReconciliation).
	Regrouped []GroupReconciliation `json:"regrouped,omitempty"`
	// Warnings carry every non-fatal diagnostic the create produced —
	// PULSE_SHARD_DESCRIPTION_DIVERGENCE from cohesion, one
	// PULSE_SHARD_SET_WIDENED per entry in Widened and one
	// PULSE_SHARD_GROUPS_REWRITTEN per entry in Regrouped.
	Warnings []encx.CohesionWarning `json:"warnings"`
}

CreateShardArchiveResult is what CreateShardArchive reports.

Like AddShardResult it is a RESULT rather than a bare error for one reason: the seed path auto-widens too, and the PULSE_SHARD_SET_WIDENED warning is mandatory wherever a widen happens. A create that silently re-lays-out the shards it was handed is the same failure as a silent add — the caller cannot tell a straight zip of their inputs from a rewrite of them, and the `.pulse` bytes now in the archive are not the `.pulse` bytes on disk beside it.

One rule, no asymmetry: create and add widen on the same conditions, through the same planner and executor, and report through the same SetWidening / CohesionWarning shapes.

type DecodeCallback 

type DecodeCallback func(rec *processing.Record) error

DecodeCallback is invoked once per decoded record by a parallel decode worker. Implementations MUST be safe to call from a single goroutine — each worker is given its own DecodeCallback via the DecodeCallbackFactory, so the same callback instance is never shared across workers.

Returning a non-nil error aborts the worker's decode loop; the errgroup propagates the error and cancels every other worker.

type DecodeCallbackFactory 

type DecodeCallbackFactory func(workerIdx, startRecord, recordCount int) DecodeCallback

DecodeCallbackFactory constructs a per-worker DecodeCallback. Called once per worker before that worker begins decoding. The factory MUST be safe to call from the goroutine that owns workerIdx; the returned callback is consumed only by that same goroutine.

workerIdx is the worker's 0-based index (0..workers-1) and is stable for the lifetime of the parallel run — partial-state implementations can key per-worker accumulators on it. startRecord is the ABSOLUTE cohort record index of the segment's first record (a multiple of linalg.MergeBlockSize), so the callback's i-th record is record startRecord+i — what a block-keyed reducer keys its blocks on. Each record also arrives stamped with that position (processing.Record.SetMergePosition(0, startRecord+i)), which is what a processing.BlockMerger reads. recordCount is the number of records the worker will receive (not the cohort total), useful when the callback wants to pre-size a slice or accumulator. The factory is never called for an empty segment (more workers than blocks), so recordCount is always >= 1.

type ExecInfo 

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

ExecInfo is the per-request execution-facts carrier the facade's observe helper hands the service through ctx (WithExecInfo) when observability is on. The service stamps it at the existing decision points — the execution arm (fused crosstab, streaming vs buffered, shard-parallel, parallel decode, join), the projection, worker and shard counts — and adds the run's row and byte counters.

It is independent of Response.Components: the row counters come from Response.Metadata (always built by the engine, before any `return` shaping), so they are set with DisableComponents and with a `return` block that drops components.run.

Off (no carrier in ctx) every stamp is a nil-receiver no-op: the only off-path cost is a ctx.Value lookup, which never allocates.

Phases: the carrier also keeps the operation's phase clock. Phases are contiguous laps — Lap(ph) attributes the time since the previous boundary (the carrier's creation, a Lap or a Mark) to ph — so a boundary costs one clock read and nothing runs per row. A phase that runs in several pieces (plan before and after the cohort open) sums.

Child operations: a Compose slot or a ProcessChain stage runs under its OWN carrier (startChild), so each child reports its own arm and counters; the facade folds every finished child into its parent with Absorb.

Concurrency: shard / decode workers read through one counting file system and ComposeParallel children absorb into the parent from their goroutines, so the counters are atomics and the plan fields and the phase clock sit behind a mutex.

func ExecInfoFrom 

func ExecInfoFrom(ctx context.Context) *ExecInfo

ExecInfoFrom returns the carrier in ctx, or nil when observability is off. The facade reads it to time its own shape phase.

func NewExecInfo 

func NewExecInfo(start time.Time) *ExecInfo

NewExecInfo returns a carrier whose phase clock starts at start (the operation's start).

func (*ExecInfo) Absorb 

func (e *ExecInfo) Absorb(c ExecSnapshot)

Absorb folds a finished child operation's snapshot into this (the parent's) carrier: row and byte counters sum, Workers / Shards / ProjectedFields keep the maximum, and the arm is the children's arm when every absorbed child ran the same one, else empty. Phases are not absorbed — each child reports its own.

func (*ExecInfo) Lap 

func (e *ExecInfo) Lap(ph observe.Phase)

Lap closes the current phase segment: the time since the previous boundary is added to ph, and the boundary moves to now. Nil-safe — a no-op (and no clock read) when observability is off.

func (*ExecInfo) Mark 

func (e *ExecInfo) Mark()

Mark moves the phase boundary to now without attributing the elapsed time to any phase — used after a stretch that belongs to child operations (their own carriers time it) or to no phase.

func (*ExecInfo) SetChildStart 

func (e *ExecInfo) SetChildStart(f ChildStart)

SetChildStart installs the child-operation starter.

func (*ExecInfo) Snapshot 

func (e *ExecInfo) Snapshot() ExecSnapshot

Snapshot copies the carrier's current state.

type ExecSnapshot 

type ExecSnapshot struct {
	Arm observe.Arm
	// Workers is the resolved worker count of the arm (1 for a serial
	// arm); Shards the shard count of an archive-backed cohort (0 for a
	// single file).
	Workers int
	Shards  int
	// ProjectedFields is the retained-field count when a decode
	// projection was installed; 0 means full decode.
	ProjectedFields int

	RowsScanned int64
	RowsMatched int64
	RowsOut     int64
	BytesRead   int64

	// Phases are the phases that ran, in execution order, each with its
	// summed lap time.
	Phases []observe.PhaseTiming
}

ExecSnapshot is a point-in-time copy of an ExecInfo.

type GroupReconciliation 

type GroupReconciliation struct {
	// Reason is one of "incoming_flattened", "incoming_regrouped",
	// "constant_promoted".
	Reason string `json:"reason"`
	// ArchiveRewritten is true when every shard already in the archive
	// was re-encoded (constant_promoted), false when only the arriving
	// shard was.
	ArchiveRewritten bool `json:"archive_rewritten"`
	// Groups labels the groups involved: the promoted groups for
	// constant_promoted, the archive's groups otherwise.
	Groups []string `json:"groups,omitempty"`
	// ArchiveFormatVersion / IncomingFormatVersion are the .pulse format
	// versions the archive and the arriving shard declared (1 = no
	// groups, 2 = grouped).
	ArchiveFormatVersion  int `json:"archive_format_version"`
	IncomingFormatVersion int `json:"incoming_format_version"`
	// ShardsRewritten / RecordsRewritten size the rewrite.
	ShardsRewritten  int   `json:"shards_rewritten"`
	RecordsRewritten int64 `json:"records_rewritten"`
}

GroupReconciliation records one parent-group layout change a CreateShardArchive or AddShard made to fit a shard into the archive, with the size of the rewrite it cost. Typed for the same reason SetWidening is: an embedder adding shards on a schedule must be able to act on it without parsing prose. The mandatory PULSE_SHARD_GROUPS_REWRITTEN warning carries the same figures.

A plain dictionary union — an arriving shard with the archive's group layout whose tuples are renumbered into the canonical dictionary — is NOT reported here, exactly as a categorical dictionary union is not: it rewrites only the arriving shard and is the ordinary cost of an add.

type IndexFreshnessReason 

type IndexFreshnessReason string

IndexFreshnessReason names why Service.VerifyIndex reached its Fresh verdict. All three values are reachable from VerifyIndex's size fast-path / full-hash decision tree — see VerifyIndex's doc comment for the full contract. 

const (
	// IndexFreshnessReasonStatMismatch indicates the current cohort
	// file's SIZE no longer matches the sidecar's recorded
	// Index.SourceSize. The cohort has definitely changed since the
	// index was built — the fingerprint is taken over the whole file, so
	// a file of a different length cannot hash to the recorded digest —
	// and this verdict is therefore reached WITHOUT computing the
	// content hash (the size fast-path).
	//
	// A modification time that moved while the size held no longer
	// reaches this reason: mtime resolution is not preserved across
	// filesystems, so that case escalates to the fingerprint and lands
	// on IndexFreshnessReasonFingerprintMatch / _Mismatch with
	// ModTimeDrift set. See VerifyIndex's doc comment.
	IndexFreshnessReasonStatMismatch IndexFreshnessReason = "stat_mismatch"

	// IndexFreshnessReasonFingerprintMatch indicates the size matched
	// the sidecar's recorded snapshot and the full content hash
	// (recomputed as confirmation, never skipped) is byte-equal to the
	// sidecar's embedded Fingerprint. Reached whether or not the
	// modification time also matched — VerifyIndexResult.ModTimeDrift
	// reports which.
	IndexFreshnessReasonFingerprintMatch IndexFreshnessReason = "fingerprint_match"

	// IndexFreshnessReasonFingerprintMismatch indicates the size matched
	// the sidecar's recorded snapshot, but the full content hash
	// recomputed for confirmation differs from the sidecar's embedded
	// Fingerprint — either a same-size rewrite the size check cannot
	// rule out, or a same-size rewrite that also moved the mtime
	// (ModTimeDrift true).
	IndexFreshnessReasonFingerprintMismatch IndexFreshnessReason = "fingerprint_mismatch"
)

type IndexInfo 

type IndexInfo struct {
	IndexPath      string
	Keys           []string
	DistinctKeys   int
	IndexedRecords int
}

IndexInfo describes one sidecar point-lookup index discovered alongside a cohort by Service.ListIndexes: the sidecar's on-disk path (encoding.SidecarIndexPath's derived shape), its ordered key column names, and the same distinct-key / indexed-record summary Service.BuildIndex's result carries. Never re-derived by re-scanning the cohort: the counts come from the keyless index manifest when one exists (which is what lets discovery cost no sidecar read at all) and otherwise from the sidecar's own bucket table (encoding.ReadIndexFile).

type MaterializeSink 

type MaterializeSink interface {
	Begin(cohort *Cohort, columns []descx.MaterializeColumn) error
	Row(source int64, derived []any) error
}

MaterializeSink receives a materialize run. Begin is called once, after every pre-flight check passed and before any record is decoded, with the opened source cohort and the derived columns in output order; an error from it ends the run. Row is then called once per filter-passing source record — in source order, or in the request's sort order when it carries one — with the record's 0-based source index (global over a shard archive, in Cohort.Shards order — the CohortRecordReader index) and one value per derived column (float64 or nil). Row is never called when Begin failed.

type MaterializeWarner 

type MaterializeWarner interface {
	Warn(warnings []*types.ResponseWarning)
}

MaterializeWarner is a MaterializeSink that also receives the run's Response-level warnings — a fitted vector attribute's singular or insufficient fit — once, after the last Row.

type RealPather 

type RealPather interface {
	RealPath(name string) (string, error)
}

RealPather is the capability interface a filesystem implements to declare that the virtual path it serves maps to a regular on-disk file the kernel can mmap. The returned path MUST satisfy:

  • It is an absolute or process-CWD-relative path resolvable by os.Open at the moment of the call.
  • The file is a regular file (not a device, FIFO, or socket).
  • The returned bytes, when read, are byte-equal to what the filesystem's Open(name).Read would yield.

External afero.Fs implementations (e.g. a CopyOnRead overlay that caches GCS objects to a local tempdir) should satisfy this interface to opt their files into the iterator's mmap fast path. The signature intentionally matches afero.BasePathFs.RealPath so that wrapper is detected without further glue.

Returning a non-nil error declines the fast path for that call and the iterator falls back to afero.ReadFile.

type Row 

type Row = map[string]any

Row is a single result row in a processing stream. Aliased so callers can write `service.Row` without leaking the underlying map type.

type RowIter 

type RowIter interface {
	Next(ctx context.Context) (Row, bool, error)
	Close() error
	// Metadata returns the run metadata (total/filtered row counts,
	// cohort filename). Available after the iterator is exhausted; may
	// return nil before then. Streaming consumers that need metadata
	// before draining should call Process instead.
	Metadata() *types.ResponseMetadata
	// Components returns the per-operator constituent-parts metadata
	// emitted by the run — same shape as Response.Components on a
	// buffered Process call. Available after the iterator is exhausted;
	// may return nil before then or when the run produced no components
	// payload. Streaming consumers that surface per-chunk component
	// deltas read this on terminal flush to expose the buffered-only
	// payload for non-mergeable operators (AGG_MEDIAN, AGG_PERCENTILE).
	Components() *types.ResponseComponents
}

RowIter is a pull-based iterator over a processing result. Next reports (row, true, nil) for each available row, then (nil, false, nil) on exhaustion. Close releases any underlying file handles.

Implementations are NOT goroutine-safe. A single consumer per iterator.

type Service 

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

Service is the orchestration layer connecting filesystem, encoding, and processing.

func New 

func New(fsConfig *fs.Config) *Service

New creates a new Service with the given filesystem configuration. Smart-defaults resolution runs by default; call DisableDefaults to opt out per instance (or pass pulse.Options{DisableDefaults: true} via the facade).

func (*Service) AddShard 

func (s *Service) AddShard(ctx context.Context, archivePath, shardPath string) (*AddShardResult, error)

AddShard appends shardPath to the archive at archivePath. The incoming shard is validated against the archive's canonical schema via ValidateStructuralCohesion + MergeDictUnion. If the union-merge extends the canonical dictionary, the rewritten `_schema.pulse` payload reflects the extension.

When a merged dictionary outgrows a set_* field's bitmask, the field is WIDENED to the narrowest rung that holds it — across the canonical schema and every shard payload in the archive — rather than the add being refused. Refusing forces a full re-import (re-read the source, re-infer, rebuild), which is strictly more expensive than the mechanical re-stride it would be avoiding. Widening emits a MANDATORY PULSE_SHARD_SET_WIDENED warning on the result: an expensive whole-archive rewrite that happens silently is indistinguishable from a cheap append. A union above the widest rung has nowhere to go and stays fatal (PULSE_SHARD_DICT_WIDTH_OVERFLOW).

v1 strategy: read the full archive into memory, append the new shard entry, rewrite the central directory + EOCD, and write the resulting bytes back via the atomic temp+fsync+rename pattern (§7.1). That atomicity is what makes the widen safe without locking — concurrency is caller-owned by contract, and a mid-rewrite failure leaves the original archive byte-identical and openable rather than half-widened (which would open, and decode every post-widen field at the wrong offset).

func (*Service) AutoLabels 

func (s *Service) AutoLabels() []*types.LabelBinding

AutoLabels returns the configured default bindings. Exposed for tests.

func (*Service) BindFilterModels 

func (s *Service) BindFilterModels(ctx context.Context, refs []types.ModelRef) (context.Context, error)

BindFilterModels returns ctx carrying refs loaded — every file read and decoded under the Request.Models rule (model.CheckRefs / LoadFiles, the instance's registered populations as the outer set) — marked bound, so a FilterToFile* call under it compiles its predicate's coef() against them. No refs binds the populations alone.

func (*Service) BoundRequest 

func (s *Service) BoundRequest(ctx context.Context) (context.Context, context.CancelFunc)

BoundRequest bounds ctx by the instance's RequestTimeout for one top-level call. It returns ctx unchanged (and a no-op release) when RequestTimeout is unlimited — the default. The caller defers release and maps its error through MapRequestTimeout(ctx, err) with the RETURNED ctx. A nested entry (a Compose slot's Process, chain stage 0, the facade's ProcessStreamResult around ProcessStream) bounds again with the same duration, starting later: its deadline never precedes the outer one, so context.WithTimeoutCause keeps the outer deadline and cause — the outermost call owns the budget.

func (*Service) BuildIndex 

func (s *Service) BuildIndex(ctx context.Context, path string, keyFields []string) (*BuildIndexResult, error)

BuildIndex scans the single-file cohort at path exactly once, resolves each record's ordered-tuple composite key across every column named in keyFields to its exact on-wire bytes (processing.CompositeKeyFieldOnWireBytes — the ordered concatenation of each key column's on-wire bytes; a single-element keyFields is the degenerate 1-tuple case and byte-identical to the E1 single-key path), groups row-ids by key into hash buckets (encoding.BucketIndex / encoding.HashKey — the same hash the lookup probe uses), computes the source .pulse file's content-hash fingerprint (encoding.ComputeFingerprint), and writes the sidecar via encoding.WriteIndexFile at the path encoding.SidecarIndexPath derives. keyFields may name any number of columns >= 1; key column ORDER is significant — building on ["a","b"] and ["b","a"] produces two distinct sidecar files at two distinct derived paths, with distinct composite key byte layouts.

Row-ids are the record's 0-based sequential position in the cohort (row 0 is the first record after the header + schema), matching encoding.RecordLocator's Offset(i) addressing — lookup feeds a bucket's row-id straight into RecordLocator without any translation.

Rows where ANY key column's value is null are skipped entirely — not indexed, not a build error (processing.CompositeKeyFieldOnWireBytes returns ok=false when any component is null; a composite key with a null component is unindexable, so the whole row is skipped rather than partially indexed).

Shard archive cohorts (detected via the cheap leading-magic-bytes dispatch service.Service.Open already performs — encoding.ZipMagic vs encoding.MagicBytes, no added scan) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED. Row-id addressing (RecordLocator.Offset) only has meaning for a single contiguous record region, which a multi-shard archive does not present. Sharded point-lookup support is out of scope for v1.

Key field types are gated by processing.IsIndexKeyableFieldType — see that predicate's doc comment for the full, settled keyable-type policy (E2-S3). Rejected types fail fast here, before any record is scanned: set_* with PROCESSING_CONFIG and processing.IndexKeyRejectionMessage's type-specific explanation, a vec_f32 / vec_f64 field with PULSE_VECTOR_FIELD_NOT_SCALAR (vectors.FieldNotScalar, slot "index key", that explanation under details.reason).

func (*Service) CohortCounts 

func (s *Service) CohortCounts(ctx context.Context, path string) (descx.CohortCounts, error)

CohortCounts opens the cohort at path and returns the header facts the limit pre-flight reads for it (cohortCounts) whether or not a memory limit is set — predict's PredictOptions.CohortCounter, so a Compose slot's limit findings carry the runtime's own figures. No record is read.

func (*Service) CompactShardArchive 

func (s *Service) CompactShardArchive(ctx context.Context, archivePath string) error

CompactShardArchive rewrites the archive at archivePath to eliminate orphaned bytes from prior in-place mutations and refresh the canonical schema-doc metadata (aggregate_record_count + shard_count). The rewrite reads every shard payload (excluding the reserved `_schema.pulse` entry) in central-directory order, re-emits the canonical schema with the recomputed metadata, and writes the result via the standard atomic temp+rename pattern (§7.1).

v1 note: AddShard / RemoveShard already use temp+rename for the whole-archive rewrite, so v1 archives never accumulate orphan bytes from Pulse-issued mutations. Compact still has a job in v1: it refreshes the canonical _schema.pulse metadata that may have drifted (for example, if the archive was edited by a third-party zip tool outside Pulse) and is the explicit reclaim path for archives containing orphan bytes from any source. Future PRs may introduce true in-place AddShard atop lower-level zip primitives; Compact's contract remains identical — produce a clean archive byte-for-byte.

Errors:

  • PULSE_ARCHIVE_MAGIC_INVALID / PULSE_ARCHIVE_CORRUPT — surfaced when the source archive cannot be opened.
  • PULSE_SHARD_HEADER_INVALID — surfaced when a shard payload's header cannot be parsed during record-count recomputation.
  • SERVICE_RESOURCE — I/O failure on read, temp-write, or rename.

func (*Service) Compose 

func (s *Service) Compose(ctx context.Context, composed *types.ComposedRequest) (*types.ComposedResponse, error)

Compose executes multiple requests, returning a response for each.

Before dispatching any slot, the validate phase synthesizes a default Label of `request_<index+1>` (1-based) on every slot whose caller- supplied Label is empty, against an in-memory clone of each *Request so the caller's pointer is not mutated. Two slots resolving to the same final Label are rejected with PULSE_COMPOSE_LABEL_COLLISION. Compose-only overlay kinds resolve sibling references by final Label so the names must be unique across the batch.

func (*Service) ComposeParallel 

func (s *Service) ComposeParallel(
	ctx context.Context,
	composed *types.ComposedRequest,
	opts ComposeOptions,
) (*types.ComposedResponse, error)

ComposeParallel runs every request in composed concurrently across a bounded worker pool. Responses are returned in the same order as composed.Requests; per-request errors are surfaced according to opts.

Registry factories return fresh stateful instances per request, so concurrent Process calls do not share aggregator/attribute state. Geo and decimal aggregators dispatch through buffered code paths that are also safe for concurrent invocation (no shared mutable state).

func (*Service) ComposeParallelResolved 

func (s *Service) ComposeParallelResolved(
	ctx context.Context,
	composed *types.ComposedRequest,
	opts ComposeOptions,
) (*types.ComposedResponse, []*types.Request, error)

ComposeParallelResolved is ComposeParallel that also returns the slot requests the responses ran (see ComposeResolved).

func (*Service) ComposeResolved 

func (s *Service) ComposeResolved(ctx context.Context, composed *types.ComposedRequest) (*types.ComposedResponse, []*types.Request, error)

ComposeResolved is Compose that also returns the slot requests the responses ran — the label- and defaults-resolved clones, index-aligned with ComposedResponse.Responses (nil on error). The facade resolves each slot's `return` plan from them (descx.ResolveReturn reads the defaults-resolved request for its precision exemptions) and shapes the finished, overlay-folded response; the service never shapes.

func (*Service) CountRecords 

func (s *Service) CountRecords(ctx context.Context, path string) (uint64, error)

CountRecords returns the total record count for the cohort at path without decoding any payload bytes. Three resolution paths:

  • Single-file `.pulse`: stat the file, stream the header + schema, and derive count = (file_size − header − schema) / record_stride. Bytes read is bounded by header + schema size — O(1) in the record count.
  • Shard archive (zip magic): read the archive's central directory
  • the reserved `_schema.pulse` entry. The SHRD trailer's AggregateRecordCount is the cached sum of every shard's record count; cohesive archives keep it in lockstep with the per-shard headers. Cost is O(N shards) for the directory walk, not O(total records).
  • Anchor path `archive#shard`: opens the archive's central directory and uses the named shard's header to derive count. Cost is O(shard payload size) today; the v1 contract is "no full-archive decode," not "header-only" for anchors.

Embedders that want strict O(1) for anchors can call pulse.Inspect(archive) and read the per-shard counts from the returned Shards slice.

func (*Service) CreateShardArchive 

func (s *Service) CreateShardArchive(ctx context.Context, archivePath string, shardPaths []string) (*CreateShardArchiveResult, error)

CreateShardArchive builds a fresh Pulse shard archive at archivePath from the constituent shardPaths. The first shard seeds the canonical schema; every other shard is validated via encoding.ValidateStructuralCohesion + encoding.ValidateDictPrefixRule. Dictionary growth across shards accumulates into the canonical schema. The archive is written to a temp path on the same filesystem and Rename'd over the destination — partial writes never appear at archivePath (atomic temp+rename, §7.1).

Set-width AUTO-WIDEN applies here exactly as it does on AddShard: a set column whose merged dictionary outgrows its declared bitmask, or whose rung differs between two of the shards being seeded, is promoted to the narrowest rung that holds every member, across the canonical schema and every shard accumulated so far. Refusing at create while auto-widening at add was an asymmetry with no defence — the same two files produce an archive or an error depending only on whether they arrived together or one after the other.

Widening emits a MANDATORY PULSE_SHARD_SET_WIDENED warning on the result, which is why this returns a *CreateShardArchiveResult rather than a bare error. A silent whole-archive rewrite is exactly the failure class the byte-layout contract exists to prevent: the shards stored in the archive are no longer byte-identical to the `.pulse` files the caller named, and nothing else would say so.

A union above the widest rung has nowhere to go and stays fatal (PULSE_SHARD_DICT_WIDTH_OVERFLOW).

Errors:

  • PULSE_SHARD_RESERVED_NAME — any shardPath has basename `_schema.pulse`.
  • PULSE_SHARD_NAME_COLLISION — two shardPaths share a basename.
  • PULSE_SHARD_SCHEMA_MISMATCH / PULSE_SHARD_DICT_DIVERGENCE — propagated from cohesion validators.
  • PULSE_SHARD_DICT_WIDTH_OVERFLOW — a set union past the widest rung.
  • SERVICE_VALIDATION — empty shardPaths.
  • SERVICE_RESOURCE — I/O failure on any input/output.

func (*Service) CrosstabFusionDisabled 

func (s *Service) CrosstabFusionDisabled() bool

CrosstabFusionDisabled reports whether the fused-crosstab dispatch is switched off (SetDisableCrosstabFusion / pulse.Options.DisableCrosstabFusion).

func (*Service) DecodeWorkers 

func (s *Service) DecodeWorkers() int

DecodeWorkers returns the configured cap. Exposed for tests and the buffered-decode orchestrator.

func (*Service) DedupPreflight 

func (s *Service) DedupPreflight(_ context.Context, path string) error

DedupPreflight performs the layout dispatch a retro-dedup needs before the byte work starts (io.DedupJob, which speaks only the single-file layout). It mirrors WidenSetField's dispatch exactly:

  • SERVICE_VALIDATION — path is an anchored shard (`archive.pulse#shard.pulse`): a shard inside an archive cannot be regrouped alone, because every shard shares the canonical schema.
  • SERVICE_VALIDATION — path is a shard archive (zip magic). details.layout = "shard_archive". Shard archives DO carry grouped shards (shard create/add union-merge their dictionaries), but an archive-wide dedup is a different operation from a per-file one: the viability gate, constant elision and group suggestion would have to be decided over the union of every shard (a cross-shard observation pass), and every shard rewritten atomically to one layout. That is not built; the supported route is to dedup each source shard (or `shard extract` + dedup) and `shard create` the results, which reconciles them to one layout.
  • SERVICE_RESOURCE — the cohort does not exist / cannot be opened.

A nil return means path is a single-file cohort the job may read.

func (*Service) DefaultMultiplicity 

func (s *Service) DefaultMultiplicity() *types.Multiplicity

DefaultMultiplicity returns a copy of the installed default multiplicity block, or nil.

func (*Service) DefaultTimeZone 

func (s *Service) DefaultTimeZone() string

DefaultTimeZone returns the installed default zone name ("" = UTC).

func (*Service) DefaultWeight 

func (s *Service) DefaultWeight() *types.WeightSpec

DefaultWeight returns a copy of the installed default weight, or nil.

func (*Service) DefaultsDisabled 

func (s *Service) DefaultsDisabled() bool

DefaultsDisabled reports SetDisableDefaults, so the facade's predict validates the request the runtime will execute.

func (*Service) DisableComponents 

func (s *Service) DisableComponents() bool

DisableComponents reports the engine-level setting.

func (*Service) DropIndex 

func (s *Service) DropIndex(ctx context.Context, path string, keyFields []string) error

DropIndex removes the sidecar point-lookup index built for keyFields against the cohort at path (encoding.SidecarIndexPath derives the exact path). Non-interactive by design — there is no confirmation prompt (a prompt would block a scripted/automated caller); callers that want a confirm-before-delete UX build it on top of ListIndexes + DropIndex themselves. Sidecars are cheap, deterministic rebuild artifacts (see encoding.IndexFormatVersion's doc comment), so dropping one is never destructive of anything Service.BuildIndex cannot regenerate.

The cohort's keyless sidecar index manifest (encoding.IndexManifestSuffix) has the corresponding entry pruned in the same call, so a dropped index stops being discoverable as well as stopping being readable.

Returns PULSE_INDEX_MISSING (not a silent no-op) when neither a sidecar NOR a manifest entry exists at the derived path for keyFields — a caller who asked to drop a specific key-set's index gets a clear, coded signal rather than an ambiguous success, matching Service.VerifyIndex's contract for the same "no sidecar found" case. An ORPHANED manifest entry (the entry is present, the file is gone — an `rm` behind Pulse's back) is a successful drop rather than PULSE_INDEX_MISSING: that state is what PULSE_INDEX_MANIFEST_STALE reports, and this is the call its fixup points at. Shard archive cohorts are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED, matching Service.BuildIndex / Service.VerifyIndex / Service.ListIndexes's v1 single-file-only contract.

func (*Service) EchoRequest 

func (s *Service) EchoRequest() bool

EchoRequest reports the current echo setting.

func (*Service) Extensions 

func (s *Service) Extensions() *processing.ExtensionRegistry

Extensions returns the installed ExtensionRegistry, or nil when no extensions are registered. Exposed for descriptor/manifest emission and MCP schema-binding paths.

func (*Service) ExtensionsSnapshot 

func (s *Service) ExtensionsSnapshot() *descx.ExtensionsSnapshot

ExtensionsSnapshot returns the descriptor-side projection of the registered (visible) extensions, or nil when none are installed.

func (*Service) ExtractShard 

func (s *Service) ExtractShard(ctx context.Context, archivePath, shardBasename string) (io.ReadCloser, error)

ExtractShard returns an io.ReadCloser over the named shard's standalone single-file `.pulse` bytes. The closer releases the archive resources; callers must invoke Close after draining.

func (*Service) Facet 

func (s *Service) Facet(ctx context.Context, path string, field string) ([]string, error)

Facet returns distinct values for the named field in the cohort.

For categorical fields the dictionary is the authoritative source:

  • Single-file cohorts return the schema's field dictionary directly.
  • Shard archives return the canonical dictionary carried by `_schema.pulse`, which is guaranteed to be the union of every shard's per-shard dictionary by the append-only prefix rule enforced at insert time (sharding design contract §3.2). No record streaming is required.

For numeric fields the cohort is streamed once and the set of distinct values observed is returned in encounter order. For shard archives the stream spans every shard in central-directory (insertion) order — the union semantics specified in §5.4. The streamed distinct-value set is bounded by MaxGroups, and a set MaxEstimatedMemory refuses the streamed path before any record is read (FacetLimitRefusal); the categorical fast path returns the schema's dictionary, which is already resident, reads no record and mints nothing, so neither limit applies to it.

func (*Service) FacetSchema 

func (s *Service) FacetSchema(ctx context.Context, req *types.FacetRequest) (*types.FacetResult, error)

FacetSchema is the rich multi-field facet implementation. The pulse.FacetSchema facade is a thin wrapper around this method.

Algorithm: one streaming pass over the cohort applies the base filterers, then for each requested field updates the appropriate accumulator (discrete dictionary count, online numeric stats, online histogram bin). Numeric percentiles force a buffered second-stage sort over the per-field non-null value slices. Additive contribution counts run a parallel discrete accumulator with the additive field's own filter clauses stripped from the base filter.

func (*Service) FilterToFile 

func (s *Service) FilterToFile(ctx context.Context, src, dst, filterExpr string) (int64, error)

FilterToFile reads the .pulse cohort at src, evaluates filterExpr against every record using FILTER_EXPRESSION semantics, and writes a new .pulse cohort at dst containing only records that pass. Custom expr functions and lookup tables registered via pulse.Options.Extensions are honored.

Three input shapes are supported, dispatched on the first four bytes:

  • Single-file cohort (magic "PULSE\x00\x00\x00"): the dst is a single-file cohort whose header + schema bytes are copied byte-for-byte from src; only the record payload differs. A grouped (0x02) source therefore yields a 0x02 destination carrying the source's group descriptors and dictionaries UNCHANGED — surviving physical rows are copied by reader position, so every row's entry index stays valid. Entries no surviving row references are carried, not pruned (a deliberate v1 choice: pruning costs a pass and renumbers indices); inspect's ratio is then surviving rows / carried entries. Re-deduping the output with the same groups prunes. A filter that keeps nothing writes the preamble alone: a valid, empty 0x02 cohort. Records are handed their row's entry indices (RecordReader.GroupIndex), so a filter over one group's members takes the per-entry precompute.

  • Shard archive (magic "PK\x03\x04"): the dst is a new shard archive preserving the per-shard layout — one input shard maps to one output shard at the same basename, in central-directory (insertion) order. Per-shard header + schema + categorical dictionary bytes are copied byte-for-byte (records inside each shard reference that shard's local dictionary, which is preserved under the prefix-rule invariant). Shards with zero surviving records are written as empty payloads (header + schema + zero record bytes). The canonical `_schema.pulse` trailer is refreshed: aggregate_record_count = sum of surviving rows across shards, shard_count = unchanged.

  • Anchor syntax `archive.pulse#shard.pulse`: resolves to a single shard inside an archive; dst is a single-file cohort built from that shard alone.

Returns the number of records written to dst (sum across shards for archive inputs).

func (*Service) FilterToFileBySetAndExpr 

func (s *Service) FilterToFileBySetAndExpr(ctx context.Context, src, dst, includeField string, set processing.MemberSet, filterExpr string) (int64, error)

FilterToFileBySetAndExpr applies an include-set predicate (membership test against an externally-supplied list of values) and optionally a filter expression to every record in src, writing the survivors to dst. At least one of (set, filterExpr) must be supplied — if both, the set is tested first and the expression evaluated only on surviving rows (cheap-check-first short-circuit).

Input-shape dispatch matches FilterToFile (single-file, shard archive, anchor). includeField names the field whose value is tested for membership in set; the set must have been built against the same canonical schema as src (the field is resolved per-call to keep the predicate schema-coherent across shards).

Returns the number of records written to dst.

func (*Service) FilterToFileFilterers 

func (s *Service) FilterToFileFilterers(ctx context.Context, src, dst string, filterers []*types.Filterer, filterExpr string) (int64, error)

FilterToFileFilterers is FilterToFile for a structured predicate: filterExpr is the expression the filterers translate to (the facade's FilterToFileRequest.canonicalExpression), and the filterers themselves are judged by the shared field rule (internal/descriptor.FieldRefRefusals, filterer half) against the cohort schema — the canonical one for a shard archive — before the expression compiles, so an unknown or empty Field is refused with the code, message and details Process and predict use rather than an expression compile error.

func (*Service) InstanceSnapshot 

func (s *Service) InstanceSnapshot() *descx.InstanceSnapshot

InstanceSnapshot returns the installed instance snapshot. A nil result is unscoped: every feature enabled, no extensions.

func (*Service) Limits 

func (s *Service) Limits() limits.Limits

Limits returns the effective resource limits — the built-in defaults when SetLimits was never called. The result is a copy.

func (*Service) ListIndexes 

func (s *Service) ListIndexes(ctx context.Context, path string) ([]IndexInfo, error)

ListIndexes enumerates every sidecar point-lookup index built against the cohort at path, and is the answer to "what key tuples can I look this cohort up by" — a question the sidecar naming scheme otherwise makes unanswerable, since encoding.SidecarIndexPath derives each sidecar's filename from a HASH of its ordered key tuple, so the file cannot be opened without already knowing its contents.

TWO SOURCES, MANIFEST FIRST:

  1. The keyless manifest (encoding.IndexManifestSuffix, "cohort.pulse.indexes.json") when one exists. It carries every entry's ordered key tuple AND its distinct-key / indexed-record counts, so this path reads NO sidecar bucket table at all: one read of the document plus one existence probe per entry. That is what makes discovery work on object storage, which cannot list a directory (and is why Pulse writes the manifest at all).
  2. The directory listing (afero.ReadDir, never raw os), keeping entries whose basename matches the SidecarIndexPath shape for this cohort and reading each match's key spec + bucket table (encoding.ReadIndexFile).

The two UNION when both are available: the manifest is the guaranteed floor, and a sidecar on disk the manifest does not name (built by a Pulse predating the manifest, then never rebuilt) is still listed rather than disappearing the day a manifest appears beside it. When a manifest exists, a failure to list the directory is NOT an error — the manifest already answered, and a backend with no listing is the case it was written for.

A manifest entry naming a sidecar that is not present is PULSE_INDEX_MANIFEST_STALE rather than a silently skipped entry; see listIndexesFromManifest. A malformed manifest is PULSE_INDEX_MANIFEST_INVALID rather than a silent fall back to the directory listing, which would succeed locally and answer "no indexes" on a bucket.

Results are sorted by IndexPath for a deterministic, rebuild-stable listing order (a directory listing's raw order is not guaranteed across afero backends).

No sidecar indexes is not an error — ListIndexes returns an empty (non-nil) slice. Shard archive cohorts (same leading-magic-bytes dispatch Service.Open already performs) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED, matching Service.BuildIndex / Service.VerifyIndex's v1 single-file-only contract.

func (*Service) ListShards 

func (s *Service) ListShards(ctx context.Context, archivePath string) ([]ShardEntry, error)

ListShards opens the archive and returns its ShardEntry slice in central-directory order (reuses Service.Open which peeks per-shard headers to populate RecordCount). Single-file cohorts return an empty slice.

func (*Service) Lookup 

func (s *Service) Lookup(ctx context.Context, req *types.LookupRequest) (*types.LookupResult, error)

Lookup resolves a point lookup against the cohort named in req.Cohort, keyed on the ordered tuple req.KeyComponents() returns (either req.Keys verbatim, for a composite/multi-column key, or the req.Field/req.Value single-key convenience path — see types.LookupRequest.KeyComponents): derive the sidecar index path for the ordered key-field names (the same derivation Service.BuildIndex used to write it), read only the sidecar's cheap bucket-data-free prefix (encoding.ReadIndexMeta) to validate the request's key-component count and run the read-path staleness check, resolve every component's literal to on-wire key bytes and concatenate them in order, seek directly to and parse the ONE hash bucket the key resolves to (encoding.ReadBucketByKey) for a byte-equal entry, and — on a hit — read the matching record(s) via encoding.RecordLocator .ReadRecordAt with a single reused DecodePlan projected to req.ReturnColumns, seeking straight to each matched record's byte offset rather than decoding the cohort from the start.

This is an O(1) operation in the cohort's record count: neither the sidecar index nor the cohort payload is ever read in full. Only the bucket-data-free index prefix, the one resolved bucket, and the matched record(s) (plus the fixed header+schema prefix each file carries) are read.

Key column order is significant: KeyComponents in a different order derives a different sidecar path and a different composite key byte layout, even when the same set of columns/values is supplied.

Multiplicity: when the matched bucket entry carries more than one row-id (a duplicate key value in the source cohort), req.Multiplicity selects the behavior — see types.LookupMultiplicity's doc comment. The zero value (unset) and LookupMultiplicityAssertUnique both mean "assert unique": Lookup fails with PULSE_LOOKUP_AMBIGUOUS when more than one row-id matches. LookupMultiplicityFirst takes the lowest row-id deterministically; LookupMultiplicityAll returns every matching row. Row order (both "first"'s pick and "all"'s slice order) is always ascending row-id — the sidecar bucket's RowIDs slice is not itself guaranteed sorted, so Lookup sorts a copy before use.

Shard archive cohorts (detected via the cheap leading-magic-bytes dispatch Service.Open already performs) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED — sharded point-lookup is out of scope for v1. See Service.BuildIndex's doc comment for why.

func (*Service) Materialize 

func (s *Service) Materialize(ctx context.Context, req *types.Request, sink MaterializeSink) error

Materialize runs req's record-level pipeline — features, filters, attributes, windows — over its cohort and hands every filtered record's derived values to sink. It runs Process's own pre-flight in Process's order (cohort required, the instance slot gate, smart defaults, zones, the field-reference rule with weight resolution, the limits pre-flight) so a request Process refuses is refused with the same error, plus the materialize gates: a slot that does not produce one value per record (PULSE_MATERIALIZE_SLOT_UNSUPPORTED), no derived slot (PULSE_MATERIALIZE_NO_DERIVED) and a feature whose outputs cannot be named (PULSE_MATERIALIZE_OUTPUT_UNDECLARED) — all before sink.Begin.

A sort reorders the Row calls only (processing.DeriveRecords): the derived values are those of the unsorted run. The scan always decodes every field (no projection) and holds every record. The memory pre-flight already judges that buffered arm: with aggregations and matrices refused, no accepted request is streamable. Bound by Limits.RequestTimeout like Process.

func (*Service) Open 

func (s *Service) Open(ctx context.Context, path string) (*Cohort, error)

Open reads a .pulse file and returns a Cohort with the parsed schema. Dispatches on the file's leading magic bytes:

  • "PULSE\x00\x00\x00" (single-file cohort) — reads header + schema directly from the file. Shards is empty.
  • "PK\x03\x04" (zip archive) — parses the zip central directory, reads the canonical schema from the reserved `_schema.pulse` entry, and populates Shards with every other entry in central-directory order. RecordCount is populated from `_schema.pulse` metadata or shard headers.

Returns PULSE_ARCHIVE_MAGIC_INVALID when the file matches neither magic, PULSE_ARCHIVE_CORRUPT when the zip EOCD or central directory is unreadable, and PULSE_SHARD_MISSING when an archive lacks the reserved `_schema.pulse` entry.

func (*Service) OpenAnchor 

func (s *Service) OpenAnchor(_ context.Context, archivePath, entry string) (*Cohort, error)

OpenAnchor opens a single shard inside a Pulse shard archive as if it were a standalone single-file `.pulse` cohort. The schema is read from the shard's own header (NOT the archive's canonical schema) so anchor-addressed shards stand alone for the purposes of facade methods. Shards is empty on the returned Cohort.

Errors:

  • PULSE_ARCHIVE_MAGIC_INVALID — archivePath is not a Pulse shard archive.
  • PULSE_SHARD_MISSING — the named entry is not present in the archive.
  • PULSE_SHARD_RESERVED_NAME — entry is the reserved `_schema.pulse`.
  • PULSE_SHARD_HEADER_INVALID / ENCODING_INVALID — the shard payload is malformed.

func (*Service) PrecheckShardCohesion 

func (s *Service) PrecheckShardCohesion(ctx context.Context, archivePath string, schema *encoding.Schema) error

PrecheckShardCohesion answers, before a single row exists, whether a shard laid out by schema could be appended to the archive at archivePath — the up-front check an anchored CohortBuilder runs at construction so a schema AddShard is certain to refuse fails there, not after every row has been spooled.

It runs AddShard's own merge over a ZERO-ROW shard of schema: the same set-rung plan before strict structural cohesion (reconcileFlat), then the same dictionary union (encx.MergeDictUnion), on the LOGICAL schemas when either side is grouped, exactly as mergeShard and mergeGroupedShard do. A refusal therefore carries AddShard's code, message and details unchanged (PULSE_SHARD_SCHEMA_MISMATCH, PULSE_SHARD_DICT_WIDTH_OVERFLOW, …).

It never refuses what AddShard would accept: set widening, a dictionary union and a group re-layout are all legal here, and every verdict it can reach is monotone in the rows still to come — a builder's dictionaries only grow from their pre-seeded entries, so a union that already overflows or diverges stays that way. What only rows decide (the group re-encode, a key violation, an overflow the appended labels cause) is left to AddShard at Close, which stays the authority: the archive may change between the two. Nothing is written.

func (*Service) Process 

func (s *Service) Process(ctx context.Context, req *types.Request) (*types.Response, error)

Process executes a single request against the specified cohort. Records are streamed from disk — the full file is never held in memory as raw bytes alongside the decoded records.

func (*Service) ProcessChain 

func (s *Service) ProcessChain(ctx context.Context, req *types.ChainRequest) (*types.ChainResponse, error)

ProcessChain executes a source-rooted linear chain of Process requests. The first stage runs against the cohort identified by req.Cohort; each subsequent stage receives the previous stage's rows as its input, materialised through an in-memory SliceIterator against a synthesised schema.

All stages must pass processing.ChainRefusal (mergeable + scalar-emitting aggregators only — internal/mergegate, the rule the chain validator shares). The executor returns PULSE_CHAIN_NOT_MERGEABLE with the offending stage index in details on first failure, allowing callers to fall back to per-stage Process; a stage >= 1 carrying Joins is PULSE_CHAIN_STAGE_JOIN. Stage 0's Request.Cohort is replaced with req.Cohort; any Request.Cohort on stages >= 1 is ignored.

func (*Service) ProcessStream 

func (s *Service) ProcessStream(ctx context.Context, req *types.Request) (RowIter, error)

ProcessStream executes a request and returns a pull-based row iterator over the result. The semantics match Process — same gates, same errors, same metadata — but the consumer can drain rows incrementally instead of receiving the whole []map[string]any up front.

Today the iterator buffers internally (calls Process and walks Data). The signature is forward-compatible with a future true-streaming path driven by the streaming Processor; consumers that adopt it now will pick up that change without a code change on their side.

func (*Service) ProjectBufferedFields 

func (s *Service) ProjectBufferedFields() bool

ProjectBufferedFields reports the current setting. Read by callers that need to know whether to compute NeededFields ahead of time.

func (*Service) ReadModelFile 

func (s *Service) ReadModelFile(path string) ([]byte, error)

ReadModelFile reads a model file's bytes at path (relative to the instance filesystem root) — the read every `models` load uses, and the facade's predict loader (PredictOptions.ModelLoader).

func (*Service) RemoveShard 

func (s *Service) RemoveShard(ctx context.Context, archivePath, shardBasename string) error

RemoveShard rewrites the archive at archivePath omitting the named shard. The canonical schema is preserved (orphaned dictionary entries are retained — canonical only grows). Returns PULSE_SHARD_MISSING when the named shard is not present.

func (*Service) ResolveCanonicalSchema 

func (s *Service) ResolveCanonicalSchema(_ context.Context, src string) (*encoding.Schema, error)

ResolveCanonicalSchema reads the canonical schema for src without streaming the record payload. Single-file cohorts return their own schema; shard archives return the canonical `_schema.pulse` schema; anchored shards return the anchored shard's local schema. Used by callers (notably the CLI cohort-filter include-from path) that need schema introspection to build an include-set before invoking FilterToFileBySetAndExpr.

func (*Service) Sample 

func (s *Service) Sample(ctx context.Context, path string, n int) ([]map[string]any, error)

Sample returns up to n rows from the cohort as maps of field name to value. Streams from disk — stops reading as soon as n rows are collected. For shard archives the rows span the union of shards in central-directory (insertion) order; offset+limit apply globally across the union, never per-shard (sharding design contract §5.4).

func (*Service) SampleWithRequest 

func (s *Service) SampleWithRequest(ctx context.Context, req *types.SampleRequest, path string) ([]map[string]any, []processing.ResolverWarning, error)

SampleWithRequest is the labelled variant. The req struct carries the cohort path (via Cohort.Filename), the row cap (N), and any LabelBinding entries that translate categorical values to display labels in the returned rows. Path resolution mirrors the rest of the facade — Cohort.Filename is resolved against the Service's data-dir / fs. Returns the rows plus the resolver-side warnings (PULSE_LABEL_COLLISION / PULSE_LABEL_LOOKUP_MISS) so callers can fold them into the response envelope at their boundary.

func (*Service) SetAutoLabels 

func (s *Service) SetAutoLabels(bindings []*types.LabelBinding)

SetAutoLabels installs the default LabelBindings injected into read requests before validation. Pass nil to clear. See the autoLabels field and applyAutoLabels for the schema-aware filtering applied per request. Matches pulse.Options.AutoLabels.

func (*Service) SetDecodeWorkers 

func (s *Service) SetDecodeWorkers(n int)

SetDecodeWorkers configures the per-cohort parallel decode worker pool cap used by the buffered Process path. Zero falls back to runtime.NumCPU() at dispatch time when the cohort exceeds parallelDecodeRecordThreshold; 1 disables parallelism (strictly serial path). Negative values are not rejected here — pulse.New() performs that validation at the public API boundary.

func (*Service) SetDefaultMultiplicity 

func (s *Service) SetDefaultMultiplicity(m *types.Multiplicity)

SetDefaultMultiplicity installs pulse.Options.DefaultMultiplicity (nil = none), the multiple-comparison block every test, post-test and overlay inherits field by field after its own block and its request's. pulse.New validates it first. The block is copied.

func (*Service) SetDefaultWeight 

func (s *Service) SetDefaultWeight(spec *types.WeightSpec)

SetDefaultWeight installs pulse.Options.DefaultWeight (nil = none), the weight every weight-bearing slot inherits when neither the slot nor the request names one. pulse.New validates it first. The spec is copied.

func (*Service) SetDisableComponents 

func (s *Service) SetDisableComponents(disabled bool)

SetDisableComponents toggles the engine-level default for Response.Components emission. When true, every Process / Compose / ProcessChain / Facet response on this service skips the Components block unless the per-request Request.DisableComponents pointer forces the opposite decision. See pulse.Options.DisableComponents for the full contract.

func (*Service) SetDisableCrosstabFusion 

func (s *Service) SetDisableCrosstabFusion(disabled bool)

SetDisableCrosstabFusion toggles the fused-crosstab dispatch in processCrosstab. When true, every crosstab request takes the buffered RunCrosstab path even when CanFuseCrosstab would accept it. Exposed so the equivalence test in internal/service/crosstab_fused_test.go can drive the same request through both paths and assert byte-equal output without relying on internals.

func (*Service) SetDisableDefaults 

func (s *Service) SetDisableDefaults(disabled bool)

SetDisableDefaults toggles the smart-defaults pass. Predict still computes and reports DefaultsApplied independently; this flag governs only what the runtime mutates before Process / Compose execution.

func (*Service) SetEchoRequest 

func (s *Service) SetEchoRequest(enabled bool)

SetEchoRequest toggles per-stage normalized-request capture in ProcessChain. When true, ChainResponse carries a NormalizedRequest snapshot built during execution; when false, that field is nil. Matches pulse.Options.EchoRequest.

func (*Service) SetExtensions 

func (s *Service) SetExtensions(r *processing.ExtensionRegistry)

SetExtensions installs an ExtensionRegistry containing embedder- registered operator overlays. The registry is read-only after this call; pass nil to clear. The processor consults this registry before falling through to built-in factories. The installed instance snapshot's hidden names are applied on top (either call order works).

func (*Service) SetExtensionsSnapshot 

func (s *Service) SetExtensionsSnapshot(snap *descx.ExtensionsSnapshot)

SetExtensionsSnapshot installs an extension projection with NO feature scoping (descx.UnscopedInstanceSnapshot). For callers that wire a service by hand; pulse.New uses SetInstanceSnapshot.

func (*Service) SetInstanceSnapshot 

func (s *Service) SetInstanceSnapshot(snap *descx.InstanceSnapshot)

SetInstanceSnapshot installs the instance's no-execute view — its extension projection plus its resolved feature set and digest. pulse.New populates this alongside SetExtensions. Pass nil to clear.

func (*Service) SetLimits 

func (s *Service) SetLimits(l limits.Limits)

SetLimits installs the instance's effective (resolved, validated) resource limits. pulse.New resolves and validates them; this setter stores them as given.

func (*Service) SetPopulations 

func (s *Service) SetPopulations(set *model.Set)

SetPopulations installs the instance's registered populations (model.RegisteredSet; nil for none): the outermost layer of every request's bound-model set. Called once, by pulse.New.

func (*Service) SetProjectBufferedFields 

func (s *Service) SetProjectBufferedFields(enabled bool)

SetProjectBufferedFields enables buffered-decode field projection. When enabled the streaming iterator computes the set of fields a request actually reads (NeededFields) and skips map writes for fields outside that set. Per-record memory drops proportional to the projection ratio; decode CPU is unchanged. Default is false; pulse.Options{ProjectBufferedFields: true} opts in.

Extension operators without a registered FieldInputs hook widen the projection to "every field" automatically — projection is always a strict superset of the fields actually consumed, so it can never produce a wrong answer.

func (*Service) SetShardWorkers 

func (s *Service) SetShardWorkers(n int)

SetShardWorkers configures the per-shard parallel worker pool cap. Zero falls back to runtime.NumCPU() at dispatch time; 1 disables parallelism (strictly serial path). Negative values are not rejected here — pulse.New() performs that validation at the public API boundary.

func (*Service) SetStrict 

func (s *Service) SetStrict(strict bool)

SetStrict toggles the strict request-validation flag: warnings the runtime raises (weighting, vectors) become errors.

func (*Service) SetTimeZones 

func (s *Service) SetTimeZones(defaultZone string, cache *temporal.Cache)

SetTimeZones installs the engine-wide default zone name (pulse.Options.DefaultTimeZone; empty means UTC) and the zone cache every request-time resolution loads through. A nil cache loads uncached through temporal.LoadZone.

func (*Service) ShardWorkers 

func (s *Service) ShardWorkers() int

ShardWorkers returns the configured cap. Exposed for tests and the shard-reduce orchestrator.

func (*Service) Strict 

func (s *Service) Strict() bool

Strict reports the current strict-validation setting.

func (*Service) VerifyIndex 

func (s *Service) VerifyIndex(ctx context.Context, path string, keyFields []string) (*VerifyIndexResult, error)

VerifyIndex reports whether the sidecar point-lookup index built for keyFields against the cohort at path is still fresh, using a size fast-path before paying for a full content-hash recompute:

  1. Load the sidecar's metadata only (encoding.ReadIndexMetaFile at the path encoding.SidecarIndexPath derives — never the bucket data, which freshness never needs). PULSE_INDEX_MISSING if absent.
  2. Stat the CURRENT cohort file and compare its SIZE against the sidecar's recorded Index.SourceSize (captured by Service.BuildIndex at build time). If it differs, the cohort has definitely changed — the fingerprint covers the whole file, so a different length cannot produce the recorded digest — so report Fresh=false / Reason=IndexFreshnessReasonStatMismatch / FastPath=true WITHOUT computing the content hash.
  3. If the size matches, that alone is not sufficient proof of freshness (a pathological same-size rewrite admits a false negative), so fall through to a full encoding.ComputeFingerprint recompute and compare against the sidecar's embedded Fingerprint for a conclusive verdict (IndexFreshnessReasonFingerprintMatch / IndexFreshnessReasonFingerprintMismatch, both FastPath=false). A modification time that moved while the size held takes this same path and sets ModTimeDrift: mtime resolution is NOT preserved across filesystems (the snapshot is Unix nanoseconds, an S3 LastModified is whole seconds, and every copy and cache hop truncates somewhere), so drift alone is not evidence of mutation and the fingerprint is what answers.

VerifyIndex never rebuilds the sidecar and never falls back to a full cohort scan — it only reports freshness. Rebuilding stays an explicit, caller-initiated Service.BuildIndex call (surfaced as `pulse index build`). This method underpins the `pulse index verify` CLI leaf (a later story).

Shard archive cohorts (detected via the cheap leading-magic-bytes dispatch Service.Open already performs) are rejected with PULSE_INDEX_UNSUPPORTED_SHARDED — sharded point-lookup is out of scope for v1. See Service.BuildIndex's doc comment for why.

func (*Service) VerifyShardArchive 

func (s *Service) VerifyShardArchive(ctx context.Context, archivePath string) (*VerifyResult, error)

VerifyShardArchive walks the archive at archivePath and re-validates every shard against the canonical schema. For every real shard (excluding `_schema.pulse`) it:

  1. Calls encoding.Archive.PeekShardHeader to check magic + format_version, raising PULSE_SHARD_HEADER_INVALID on mismatch.
  2. Parses the shard's full header + schema block.
  3. Calls encoding.ValidateStructuralCohesion against the canonical schema. Structural mismatches surface as PULSE_SHARD_SCHEMA_MISMATCH errors; per-field description divergence emits PULSE_SHARD_DESCRIPTION_DIVERGENCE warnings. Structural cohesion includes the parent-group layout: a shard whose groups differ from the canonical schema's (a 0x01 shard in a 0x02 archive, or the reverse) has a different physical stride and is refused.
  4. Calls encoding.ValidateDictPrefixRule. Any prefix-rule violation — categorical, set, or a parent group's entries — surfaces as PULSE_SHARD_DICT_DIVERGENCE.
  5. Reports per-set-field width headroom and per-group index headroom on the result, so an impending `shard add` auto-widen, a constant promotion or a u32 overflow is foreseeable.
  6. Re-peeks the shard's record count and sums across the archive. A drift against the canonical AggregateRecordCount emits a warning (the live per-shard sum is authoritative per the design contract).

Returns a non-nil error only when the archive itself cannot be opened (PULSE_ARCHIVE_MAGIC_INVALID, PULSE_ARCHIVE_CORRUPT, SERVICE_RESOURCE). Per-shard cohesion failures are accumulated into VerifyResult.Errors so the caller renders the full diagnosis — partial failures should not mask later shards' issues.

func (*Service) WidenSetField 

func (s *Service) WidenSetField(_ context.Context, path, field string, target encoding.FieldType) (*encoding.WidenReport, error)

WidenSetField widens the set column named field in the single-file cohort at path to the wider set rung target, rewriting the cohort in place, and returns the report describing what moved.

The byte work is entirely encoding.WidenSetFieldFile: it builds the re-strided cohort into a temp file beside the original, fsyncs it and renames it over the original, so a failure at any point leaves the original byte-identical. What this layer adds is the LAYOUT DISPATCH a `.pulse` path needs and the engine deliberately does not perform — the extension covers two formats (single-file, PULSE magic; shard archive, zip magic PK\x03\x04) and the widen engine speaks only the first. An archive handed to it is refused by ReadHeader before a byte is written, so the archive survives, but the caller is told the file is not a cohort rather than told what is actually true: that widening an archive is a different operation, performed across the canonical `_schema.pulse` and every shard payload at once. This method names that distinction instead.

Refusals are coded errors, each reusing an existing code:

  • SERVICE_RESOURCE — the cohort does not exist / cannot be opened.
  • SERVICE_VALIDATION — the path is a shard archive, or an anchored shard inside one (`archive.pulse#shard.pulse`).
  • ENCODING_INVALID — no field of that name in the schema.
  • ENCODING_TYPE_MISMATCH — the field is not a set, the target is not wider, or the field is already at the widest rung (encoding.CheckSetWiden owns that whole policy).

Sidecars are not rebuilt; see encoding.WidenSetFieldFile for why a widened cohort invalidates them through their own fingerprints.

func (*Service) ZoneLoader 

func (s *Service) ZoneLoader() descx.ZoneLoader

ZoneLoader returns the loader request-time zone resolution uses, so the facade's predict resolves through the same cache.

type SetWidening 

type SetWidening struct {
	// Field is the widened set field's name.
	Field string `json:"field"`
	// From is the rung the archive declared before the rewrite.
	//
	// From == To is the one-shard case: the arriving shard declared a
	// NARROWER rung than the archive and was promoted to meet it, so no
	// existing shard moved. Callers deciding whether an archive-wide
	// rewrite happened compare From against To — never the presence of
	// the entry, which covers both directions.
	From string `json:"from"`
	// To is the rung every shard in the archive declares after the
	// rewrite. Never narrower than From or IncomingFrom.
	To string `json:"to"`
	// IncomingFrom is the rung the ARRIVING shard declared. It differs
	// from From exactly when the two sides disagreed on the rung, which
	// is the case a re-import of a new period produces: the source
	// inferred the rung its own data needed.
	IncomingFrom string `json:"incoming_from"`
	// DictEntries is the merged dictionary size the two schemas imply.
	DictEntries int `json:"dict_entries"`
	// ShardsRewritten counts every shard payload actually re-laid-out.
	// An archive widen rewrites the archive's existing shards plus the
	// incoming one, because after it no shard carries the old stride; a
	// promotion of a narrow incoming shard rewrites exactly that one.
	ShardsRewritten int `json:"shards_rewritten"`
	// RecordsRewritten is the total number of records re-laid-out
	// across those shards.
	RecordsRewritten int64 `json:"records_rewritten"`
}

SetWidening records one set field promoted to a wider rung while a shard was being added, together with the size of the rewrite it cost.

It is TYPED rather than only narrated in the warning message because an embedder that adds shards on a schedule needs to act on the fact (rebuild the point-lookup index, re-export, alert) without parsing prose. The mandatory warning carries the same figures for a human or an LLM reading the `--json` envelope.

func (SetWidening) ArchiveWidened 

func (w SetWidening) ArchiveWidened() bool

ArchiveWidened reports whether this widening promoted the ARCHIVE — the canonical schema and every shard already stored in it — rather than only promoting the arriving shard to meet a rung the archive already declared. The two cost wildly different amounts and the warning wording differs accordingly.

type ShardEntry 

type ShardEntry struct {
	// Filename is the basename of the shard inside the archive
	// (e.g. "20190101.pulse"). Equals the zip entry name.
	Filename string

	// RecordCount is the number of records carried by the shard.
	// Populated either from the `_schema.pulse` aggregate metadata or
	// from a first-read peek of the shard's own header.
	RecordCount int64
}

ShardEntry is one shard inside a Pulse shard archive. Populated on a Cohort opened against a `.pulse` artifact whose first four bytes match the zip magic; empty for single-file cohorts.

type VerifyIndexResult 

type VerifyIndexResult struct {
	IndexPath string
	Fresh     bool
	Reason    IndexFreshnessReason

	// FastPath is true exactly when Reason ==
	// IndexFreshnessReasonStatMismatch — the size comparison alone was
	// conclusive and no content hash was computed. A fresh=true verdict
	// is NEVER reached via the fast path alone: a matching size is
	// treated as inconclusive on its own and always falls through to a
	// full Fingerprint recompute before VerifyIndex will report
	// Fresh=true.
	FastPath bool

	// ModTimeDrift reports that the cohort file's modification time no
	// longer matches the snapshot taken at index build while its size
	// does. On its own that is NOT staleness — mtime resolution is not
	// preserved across filesystems (an S3 LastModified is whole
	// seconds against the index's recorded nanoseconds), so the verdict
	// comes from the fingerprint and this flag is how a caller sees
	// that the cheap stat pair disagreed and the hash overruled it.
	// Always false alongside IndexFreshnessReasonStatMismatch, which is
	// a size verdict.
	ModTimeDrift bool
}

VerifyIndexResult is the outcome of Service.VerifyIndex: whether the sidecar index at IndexPath is still fresh against its source cohort, why, and whether that verdict was reached via the size fast-path alone (FastPath=true) or required a full content-hash recompute (FastPath=false).

type VerifyResult 

type VerifyResult struct {
	Errors   []*errors.CodedError   `json:"errors"`
	Warnings []encx.CohesionWarning `json:"warnings"`

	// SetWidthHeadroom reports, per set_* field in the canonical
	// schema, how much bitmask capacity is left and which rung a widen
	// would promote it to. It is not a verdict — an archive with zero
	// headroom is perfectly valid — it is FORESIGHT: `shard add` widens
	// a set field across every shard payload when the merged dictionary
	// outgrows the rung, and without this the only way to discover the
	// field was one member from its ceiling is to pay for the rewrite.
	// Nil for archives with no set fields.
	SetWidthHeadroom []encoding.SetWidthHeadroom `json:"set_width_headroom,omitempty"`

	// GroupIndexHeadroom reports, per parent group in the canonical
	// schema (format 0x02), how many entries its dictionary holds against
	// its ceiling — the u32 index space for an indexed group, one entry
	// for a constant group. Foresight in the same sense as
	// SetWidthHeadroom: a union past the u32 space is fatal
	// (PULSE_SHARD_DICT_WIDTH_OVERFLOW), and a constant group's next
	// distinct value promotes it archive-wide. Nil for ungrouped archives.
	GroupIndexHeadroom []encx.GroupIndexHeadroom `json:"group_index_headroom,omitempty"`
}

VerifyResult carries the structured outcome of VerifyShardArchive. Errors aggregates every fatal cohesion failure discovered while walking the archive's shards; Warnings carry non-fatal divergences (per-field description drift, aggregate-record-count drift). An empty Errors slice means the archive is structurally sound.

The shape parallels descriptor.Envelope's {errors, warnings} so callers can lift entries straight into the --json envelope without reshaping. encoding.CohesionWarning is the canonical warning carrier (shared with insert-time cohesion validators).

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.