extend

type Aggregator ¶

type Aggregator interface {
	// Aggregate computes the aggregation over rows for the named field.
	// Implementations decide how nulls contribute.
	Aggregate(rows Rows, field string) (float64, error)
}

Aggregator computes one aggregate value over a set of rows. It is the base interface every extension aggregator implements; the buffered engine path calls Aggregate once with every row that passed the request's filters.

type AggregatorFactory ¶

type AggregatorFactory func(spec *types.Aggregation, schema *encoding.Schema) (Aggregator, error)

AggregatorFactory builds a fresh Aggregator for one aggregation slot of one request. It receives the slot's spec and the cohort schema; during probe-validation at pulse.New it is called once with a spec carrying only the operator name and an empty schema, and must tolerate both.

type AttributeComputer ¶

type AttributeComputer interface {
	Compute(rows Rows, field string) ([]float64, error)
}

AttributeComputer derives one value per row. It is the base interface every extension attribute implements and the whole contract of a registration declaring Mode "buffered": the engine calls Compute once with every row that passed the request's filters.

Compute returns exactly one value per row, aligned with rows (entry i belongs to rows.At(i)).

type AttributeFactory ¶

type AttributeFactory func(spec *types.Attribute, schema *encoding.Schema) (AttributeComputer, error)

AttributeFactory builds a fresh AttributeComputer for one attribute slot of one request. During probe-validation at pulse.New it is called once with a spec carrying only the operator name and an empty schema, and must tolerate both.

type FeatureComputer ¶

type FeatureComputer interface {
	Compute(rows Rows, field string) (map[string]FeatureOutput, error)
}

FeatureComputer derives one or more columns from the rows of one buffered call, keyed by output column name. Every FeatureOutput must be aligned with rows (len(Values) == rows.Len()); a misaligned output is refused with PROCESSING_INTERNAL.

type FeatureFactory ¶

type FeatureFactory func(spec *types.Feature, schema *encoding.Schema) (FeatureComputer, error)

FeatureFactory builds a fresh FeatureComputer for one feature slot of one request. The schema is the cohort schema before any feature output is added.

type FeatureOutput ¶

type FeatureOutput struct {
	Values []float64
	Nulls  []bool
}

FeatureOutput is one derived column: Values has one entry per input row and, when Nulls is non-nil, Nulls has the same length with Nulls[i] == true meaning row i is null (Values[i] is then ignored). The engine writes the column; a feature operator never mutates a Record.

type FilterFunc ¶

type FilterFunc func(rec Record) (bool, error)

FilterFunc reports whether one row passes the filter. It runs on the streaming and buffered paths alike, once per row, and must give the same answer on both. The Record is valid only for the call.

type FiltererBuilder ¶

type FiltererBuilder interface {
	Build(spec *types.Filterer, schema *encoding.Schema) (FilterFunc, error)
}

FiltererBuilder compiles one filter slot of one request into a FilterFunc. Build is called once per request with the slot's spec and the cohort schema.

type FiltererFactory ¶

type FiltererFactory func() FiltererBuilder

FiltererFactory returns a fresh FiltererBuilder. It takes no arguments; during probe-validation at pulse.New it is called once and must return a non-nil builder.

type Grouper ¶

type Grouper interface {
	Group(rows Rows, field string) (map[string][]int, error)
}

Grouper partitions the rows of one buffered call into named buckets. It is the base interface every extension grouper implements.

Group returns, per bucket key, the INDICES into rows (0 <= i < rows.Len()) of the rows that land in that bucket. A row may appear in more than one bucket only when the registration declares FansOut. An out-of-range index is refused with a PROCESSING_INTERNAL coded error; it never panics.

type GrouperFactory ¶

type GrouperFactory func(spec *types.Group, schema *encoding.Schema) (Grouper, error)

GrouperFactory builds a fresh Grouper for one group slot of one request. During probe-validation at pulse.New it is called once with a spec carrying only the operator name and an empty schema, and must tolerate both.

type MatrixFactory ¶

type MatrixFactory func(spec *types.MatrixSpec, schema *encoding.Schema) (MatrixOperator, error)

MatrixFactory builds a fresh MatrixOperator for one matrix slot of one request. It receives the slot's spec and the cohort schema; it should decode and validate spec.Params here so a bad parameter fails before any record is read.

type MatrixInput ¶

type MatrixInput struct {
	// Name is the result name (types.MatrixResult.Name).
	Name string

	// CoMoment is the slot's merged co-moment state over Members, in
	// the missing-value mode the request chose (CoMoment.Mode). It is
	// READ-ONLY: never call Add or Merge on it — the engine may share it
	// with the slot's components floor. Read the figures through Cov,
	// Corr, Mean, N, W, NEff, PairN, PairW, PairW2, PairNEff and the raw
	// per-pair co-moment PairM2.
	CoMoment *linalg.CoMoment

	// Rows is the admitted row set, present only when the registration
	// declares Streamable: false (the buffered tier, which runs the
	// request buffered and serially); nil otherwise.
	Rows MatrixRows

	// Members are the resolved member columns, in axis order: the
	// CoMoment's variable i is Members[i].
	Members []string

	// Labels are the display labels per member when the vector declares
	// explicit labels; nil otherwise.
	Labels []string

	// Params is the spec's raw params object (nil when absent).
	Params json.RawMessage

	// WeightKind is the kind of the row weight folded into CoMoment
	// (and Rows), or "" when the slot is unweighted. A registration
	// that does not declare WeightAware always sees "".
	WeightKind types.WeightKind

	// Encoding is the Values layout the request asked for; Square
	// applies it.
	Encoding types.MatrixEncoding

	// WantAuxiliary, WantVectors and WantScalars report whether the
	// request renders that part of the result (a `return` exclusion
	// skips it). An operator should not compute a part it was told to
	// skip; one it returns anyway is dropped.
	WantAuxiliary, WantVectors, WantScalars bool
}

MatrixInput is what MatrixOperator.Finalize reads. It is valid only for the duration of the call (see the package doc's reuse contract).

func (in *MatrixInput) Rect(columns []string, at func(r, c int) float64) *types.MatrixValues

func (in *MatrixInput) Square(at func(r, c int) float64) *types.MatrixValues

type MatrixOperator ¶

type MatrixOperator interface {
	Finalize(in *MatrixInput) (MatrixOutput, error)
}

MatrixOperator is an extension MAT_* operator. It is FINALIZE-ONLY: the engine owns accumulation — the per-block co-moments, the fixed merge tree over 4096-row blocks, grouped buckets, missing-value modes and row weights — and hands the operator the merged state once per matrix result (once per group bucket on a grouped request). Because the operator never folds rows itself, its result inherits the matrix bit contract: serial, any DecodeWorkers and any ShardWorkers count give the same bits, provided Finalize is a pure function of its input.

Finalize MUST be pure and re-entrant: grouped requests call it once per bucket, and the same instance may be finalized more than once. It must not retain the input or anything reachable from it past the call. An error refuses the matrix; a *errors.CodedError (for example PULSE_MATRIX_SINGULAR) is returned to the caller as is.

type MatrixOutput ¶

type MatrixOutput struct {
	// Primary is the operator's matrix. Required.
	Primary *types.MatrixValues
	// Auxiliary holds named companion matrices.
	Auxiliary map[string]*types.MatrixValues
	// Vectors holds per-member or per-pair summaries.
	Vectors map[string]any
	// Scalars holds whole-matrix figures; NaN is null on the wire.
	Scalars map[string]float64
	// Warnings are this matrix's data-quality diagnostics.
	Warnings []*types.ResponseWarning
	// Components are the operator's own Response.Components.Matrices
	// keys, merged after the engine's matrix floor ({n, n_null,
	// n_listwise_dropped}, the pairwise and the weighted floor keys,
	// which the operator never emits). Nil adds none.
	Components map[string]any
}

MatrixOutput is what MatrixOperator.Finalize returns: the parts of one types.MatrixResult. Every key the operator fills must be declared in its registration's OutputKeys (Auxiliary, Vectors and Scalars by map key) and every Components key in its ComponentSchema; pulse.New checks this against a synthetic co-moment. Nil maps are absent parts.

type MatrixRows ¶

type MatrixRows interface {
	// Len returns the number of rows.
	Len() int
	// Row returns row i's member values in Members order, 0 <= i <
	// Len(). A NaN is a missing member (possible only in pairwise mode,
	// where a row with at least one present member is admitted). Do not
	// mutate or retain the slice.
	Row(i int) []float64
	// Weight returns row i's valid weight, or 1 on an unweighted slot.
	Weight(i int) float64
}

MatrixRows is the read-only buffered row set of a Streamable: false matrix operator: the rows the co-moment admitted, in record order. It is valid only for the duration of Finalize; do not retain it or a slice it returns. Consumer-only: Pulse implements it and may add methods in a minor release.

type MergeableAggregator ¶

type MergeableAggregator interface {
	OnlineAggregator
	// Merge folds other's running state into the receiver.
	Merge(other OnlineAggregator) error
}

MergeableAggregator is the optional sibling of OnlineAggregator for aggregators whose running state folds across input partitions. When an aggregator implements it and its registration declares both Streamable: true and Mergeable: true, the parallel reducers (pulse.Options.ShardWorkers over a shard archive, pulse.Options.DecodeWorkers over a large single-file cohort) fold each partition through UpdateRow on a fresh instance, then combine the partials with Merge before calling Finalize once on the receiver. The same declaration admits the operator to ProcessChain stages. Without it the request runs serially and a chain refuses it.

Merge receives another instance built by the SAME factory from the SAME spec — the embedder's own value, never an engine wrapper — so a type assertion to the concrete type succeeds. It absorbs other's state into the receiver; other is not reused afterwards. Merge must be associative: partials are combined in a deterministic order, but how the cohort is partitioned depends on the worker count, so a floating-point fold may differ from the serial result in the last ULP. After Merge, the receiver's Finalize, Rich and ComponentsFunc output must describe the merged state. A registration declaring Mergeable whose value lacks this interface, or whose ComponentSchema classifies its keys as non-mergeable, is refused at pulse.New with PULSE_EXTENSION_MERGEABLE_MISMATCH.

type MergeableGrouper ¶

type MergeableGrouper interface {
	Grouper
	// MergeState folds other's components state into the receiver.
	MergeState(other Grouper) error
}

MergeableGrouper is the optional merge sibling for groupers whose components state folds across input partitions. When a registration declares both Streamable: true and Mergeable: true, the parallel reducers (pulse.Options.ShardWorkers over a shard archive, pulse.Options.DecodeWorkers over a large single-file cohort) build one grouper per partition, key that partition's rows through KeyForRow / KeysForRow, then fold the partials into the first with MergeState before reading the ComponentsFunc (or Components()) output once, off the merged receiver. The same declaration admits the operator to ProcessChain stages. Without it the request runs serially and a chain refuses it.

The bucket rows themselves are merged by the engine (per-key aggregator state), so MergeState folds ONLY what the grouper keeps for its components figures — per-bucket counts, an observed range. A Mergeable grouper that emits no components has nothing to fold and need not implement this interface; one that does emit must, or pulse.New refuses the registration with PULSE_EXTENSION_MERGEABLE_MISMATCH.

MergeState receives another instance built by the SAME factory from the SAME spec — the embedder's own value, never an engine wrapper — so a type assertion to the concrete type succeeds. It absorbs other's state into the receiver; other is not reused afterwards. Partials arrive in a deterministic order (shard order, or segment order for a single file), but how the cohort is partitioned depends on the worker count, so the fold must be associative.

type MultiKeyStreamingGrouper ¶

type MultiKeyStreamingGrouper interface {
	KeysForRow(rec Record, field string) (keys []string, ok bool, err error)
}

MultiKeyStreamingGrouper is the optional fan-out sibling: one row lands in every returned bucket. A registration whose factory returns it MUST declare FansOut: true, and one declaring FansOut: true MUST return it; pulse.New refuses either mismatch with PULSE_EXTENSION_FANOUT_MISMATCH. ok=false, an empty key slice, or an ErrGrouperKeyNull error skips the row.

type OnlineAggregator ¶

type OnlineAggregator interface {
	// UpdateRow folds one row into the running state.
	UpdateRow(rec Record, field string) error
	// Finalize returns the aggregated value.
	Finalize() (float64, error)
}

OnlineAggregator is the optional streaming sibling of Aggregator. When an aggregator implements it and its registration declares Streamable: true, the engine folds rows one at a time instead of buffering them: UpdateRow once per row that passed the filters, then Finalize once. Finalize MUST be safe to call with no prior UpdateRow (the empty-input case). A Streamable registration whose factory does not return an OnlineAggregator is refused at pulse.New with PULSE_EXTENSION_STREAMABLE_MISMATCH. The declaration holds for every field type, decimal128 included: UpdateRow sees decimal fields via Record.DecimalValue.

type PostTest ¶

type PostTest interface {
	Run(rows []map[string]any) (*types.TestResult, error)
}

PostTest is a tier-2 statistical test run once over the materialized result rows (after the window stage). Each row is the column-name to value map the response would carry. Tier-2 tests are always buffered.

type PostTestFactory ¶

type PostTestFactory func(spec *types.Test, schema *encoding.Schema) (PostTest, error)

PostTestFactory builds a fresh PostTest for one test slot of one request.

type Record ¶

type Record interface {
	// Schema returns the cohort schema the row was decoded against.
	// Callers MUST NOT mutate it.
	Schema() *encoding.Schema

	// IsNull reports whether the named field is null on this row. A
	// field that is missing or projected out also reports null. It
	// agrees with NumericValue on a native vector element member
	// "f[k]": null exactly when that read reports false.
	IsNull(field string) bool

	// NumericValue returns the field's value as float64 and true, or
	// (0, false) when the field is null, missing or projected out.
	// It is ALWAYS false for set-typed fields — read them through
	// SetMaskValue — and for vector fields (VectorValue). A
	// categorical field yields its dictionary index; date yields epoch
	// days and datetime epoch seconds; u64 values above 2^53 lose
	// precision through the float64 echo; decimal128 yields only a
	// rounded echo (use DecimalValue).
	//
	// An element member "f[k]" of a native vector field f — the
	// literal member names a vector-input slot hands its factory in
	// spec.Fields — reads element k (1-based, canonical base-10, 1..dim)
	// in place, without allocating: a NaN element and a null vector
	// are a null member (false), as in every built-in vector consumer,
	// so a native vector reads exactly like its scalar-column twin. The
	// plain name is looked up first, so a real column spelled "x[2]"
	// and every name whose base is not a vector field (or whose k is
	// out of range or non-canonical) keep the ordinary lookup.
	NumericValue(field string) (float64, bool)

	// StringValue returns the dictionary-resolved label of a
	// categorical field. It is (“”, false) for a null field and for
	// every non-categorical field.
	StringValue(field string) (string, bool)

	// SetMaskValue returns the membership bitmask of a set-typed field
	// at any rung (set_u8 .. set_u256): bit i is dictionary entry i. An
	// empty mask with true is a valid "nothing selected", distinct from
	// null, which reports false. The mask is a value and safe to keep.
	SetMaskValue(field string) (encoding.SetMask, bool)

	// DecimalValue returns the exact value of a decimal128 field. It is
	// (zero, false) for a null, missing or non-decimal field. The value
	// is safe to keep.
	DecimalValue(field string) (encoding.Decimal128, bool)

	// VectorValue returns the elements of a vector field (vec_f32 /
	// vec_f64) as float64 — vec_f32 elements widen exactly — and true.
	// It is (nil, false) for a null, missing, projected-out or
	// non-vector field; NumericValue is always false for a vector. A
	// NaN or ±Inf element is a stored value, returned verbatim. The
	// slice is a fresh copy and safe to keep.
	VectorValue(field string) ([]float64, bool)

	// Weight returns the row weight the engine resolved for the slot
	// this operator runs in (slot weight → request weight →
	// pulse.Options.DefaultWeight) and true, or (0, false) when no
	// weight is in force on the slot. It reports a weight only to an
	// operator whose registration declares WeightAware, and only a
	// VALID one: finite and non-negative (zero included), and an
	// integer under kind frequency. A row whose weight is invalid never
	// reaches a weight-aware aggregator or row test (it is excluded and
	// counted by the engine); a weight-aware attribute, which owes a
	// value for every row, still receives it with Weight reporting
	// false. Every other operator — and every operator on an unweighted
	// slot — always sees (0, false). The orchestrator emits the
	// weighted floor keys (sum_weights, n_eff, n_weight_invalid); the
	// operator never does.
	Weight() (float64, bool)
}

Record is the read-only view of one decoded row handed to extension operators. It is valid only for the duration of the call that received it (see the package doc's reuse contract). Consumer-only: Pulse implements it and may add methods in a minor release.

type RichAggregator ¶

type RichAggregator interface {
	Rich() (any, error)
}

RichAggregator is the optional sibling for aggregators whose natural output is not a scalar (a label list, a per-label count map, …). The engine calls Aggregate or Finalize first, then Rich exactly once; a non-nil result replaces the float64 in the response. Returning (nil, nil) keeps the float64.

type RowLocalAttribute ¶

type RowLocalAttribute interface {
	Row(rec Record, field string) (float64, error)
}

RowLocalAttribute is the optional streaming sibling for an attribute whose value depends only on the current row. A registration declaring Mode "row_local" MUST return a value implementing it; the engine then calls Row once per row inline instead of buffering.

type RowTest ¶

type RowTest interface {
	UpdateRow(rec Record) error
	Finalize() (*types.TestResult, error)
}

RowTest is a tier-1 statistical test folded row by row during the aggregation pass: UpdateRow once per row that passed the filters, then Finalize once. Finalize MUST be safe to call with no prior UpdateRow (the empty-input case).

type RowTestFactory ¶

type RowTestFactory func(spec *types.Test, schema *encoding.Schema) (RowTest, error)

RowTestFactory builds a fresh RowTest for one test slot of one request.

type Rows ¶

type Rows interface {
	// Len returns the number of rows in the view.
	Len() int
	// At returns row i, 0 <= i < Len().
	At(i int) Record
}

Rows is a read-only, zero-copy view over the rows of one buffered call. It is valid only for the duration of that call; do not retain it or any Record obtained from it. Consumer-only: Pulse implements it and may add methods in a minor release.

type StreamingFeatureComputer ¶

type StreamingFeatureComputer interface {
	PrePass(rec Record, field string) error
	Finalize() error
	EmitRow(rec Record, field string) (map[string]FeatureOutput, error)
}

StreamingFeatureComputer is the optional streaming sibling of FeatureComputer. The engine type-asserts it on the value the factory returns: present, a stream-eligible request keeps streaming; absent, the request falls back to the buffered path.

Lifecycle: PrePass once per row on the first pass, Finalize once, then EmitRow once per row on the second pass. EmitRow returns single-row outputs (len(Values) == 1; Nulls nil or length 1). Stateless per-row operators implement PrePass and Finalize as no-ops.

type StreamingGrouper ¶

type StreamingGrouper interface {
	KeyForRow(rec Record, field string) (key string, ok bool, err error)
}

StreamingGrouper is the optional streaming sibling of Grouper: it derives one bucket key from a single row. When a registration declares Streamable: true the factory MUST return a value implementing it (or MultiKeyStreamingGrouper for a fan-out grouper); the engine then keys rows one at a time instead of buffering them. ok=false (or an ErrGrouperKeyNull error) skips the row.

Implementing it also makes the grouper eligible as a fused crosstab axis, whatever Streamable declares: the engine keys each record with field set to the grouper's own types.Group.Field.

type TwoPassAttribute ¶

type TwoPassAttribute interface {
	RowLocalAttribute
	// PrePass folds one row into the running population state.
	PrePass(rec Record, field string) error
	// Finalize closes the PrePass phase; Row may be called after it.
	Finalize() error
}

TwoPassAttribute is the optional streaming sibling for an attribute that needs population statistics (a mean, a min/max) before it can emit a row's value. A registration declaring Mode "two_pass" MUST return a value implementing it. Lifecycle: PrePass once per filter-passing row, then Finalize exactly once (it MUST tolerate no prior PrePass), then Row once per row. State is per-instance; the factory builds a fresh instance per request.

type WindowComputer ¶

type WindowComputer interface {
	Compute(rows []map[string]any, partitions [][]int, label string) error
}

WindowComputer adds one window column to the materialized result rows, mutating them in place. partitions holds each partition's row indices, already sorted by the window's OrderBy keys; label is the output column name to write. Window operators always run buffered.

type WindowFactory ¶

type WindowFactory func(spec *types.Window, opts WindowOptions) (WindowComputer, error)

WindowFactory builds a fresh WindowComputer for one window slot of one request.

type WindowOptions ¶

type WindowOptions struct{}

WindowOptions is reserved for future schema-aware window operators. It is empty today; accept it and ignore it.