@gestaltbi/stream

Build a registry pre-populated with the built-in ops under the canonical names referenced in processing.json.

Every column acting as a dimension, in {@link DIMENSION_TAGS} order, deduplicated.

Turn an accumulator into the value that lands in the output record.

Empty accumulator for a kind.

Find the column carrying time, in order of confidence: 1. an explicit column name from the op's options 2. any of {@link TIME_TAGS} in the column directory 3. a column literally named uatu:date — the canonical code that mapping.json produces, present even when no structure is loaded 4. the first column whose first value is a Date

Returns undefined when nothing matches, so callers can decide whether that is fatal or merely means "leave the order alone".

Run one check and return its verdict.

Never throws: a missing column, an empty frame or a unit mismatch comes back as skip with a reason, because a check that cannot run has not failed and must not be reported as though it had.

Fold one value into an accumulator.

Non-numeric input is skipped rather than folded in: parseFloat('') is NaN, and one blank cell used to poison a whole column's sum.

The predicate shared by the filter ops.

A filter is an object keyed by column: an array means "one of these", and an object means a comparison — between, gt, lt and friends. Absent or empty entries match everything, so a partly-filled filter narrows rather than excludes.

Base class for every op.

Holds the options and the context, and passes the frame straight through until a subclass overrides run. Constructed without a context it falls back to one that knows no columns, so an op built outside a process graph runs rather than throwing on first access.

Rolls the frame up along options.groupby.

How each measure folds comes from the structure, not from here: a column declares its own aggregation specs, so one op produces :sum, :avg, :last and ratio columns together, and a rate is accumulated as numerator and denominator and divided once.

Runs validation checks over the upstream frame.

Unlike every other op, assert is normally terminal: it returns an array of {@link Verdict} records rather than data rows, so the host can render them as pass/fail cards. Set passthrough: true to return the rows untouched and read the verdicts off {@link Assert.getVerdicts} instead — useful when you want a check in the middle of a graph.

``json { "op": "assert", "require": ["margin"], "options": { "checks": [ { "id": "gross-margin-positive", "type": "sign", "measure": "margin_accrual", "expect": ">0" } ] } } ``

Drops rows with no identity.

A CSV with a trailing newline, or a sheet exported with padding, arrives with rows whose id is empty. They are not observations, and counting them shifts every total.

Cohort axis: bucket rows by the period a subject first appeared, and stamp how many periods have elapsed since.

Retention and conversion curves need "periods since first event" as a real dimension; aggregate can only group by columns that already exist, so this op creates them.

``json { "op": "cohort", "options": { "subject": "user_id", "date": "event_date", "period": "month", "into": { "cohort": "cohort_month", "since": "periods_since" }, "window": { "days": 30, "asOf": "2013-11-06", "mask": ["retained"] } } } ``

window is the guard that stops a half-observed cohort reading as a perfect one: any cohort younger than days as of asOf has the listed measures set to null and is flagged incomplete. Without it a cohort formed yesterday shows 100% 30-day retention, which is an artefact, not a result.

How strongly the columns of a frame move together.

The rest of the package reads a frame along time — deltas, recognition, cohorts. This reads it across dimensions instead, and answers the question a pivot table raises but cannot settle: of everything collected, which pairs actually travel together, and how strongly.

Three relationships, because "correlation" means a different statistic depending on what is being related:

- measure x measure — Pearson's r, or Spearman's rho on ranks when the relationship is monotonic but not straight. Signed: -1 to 1. - dimension x measure — the correlation ratio eta: how much of the measure's variance is explained by which group a row falls in. 0 to 1. - dimension x dimension — Cramer's V over the contingency table. 0 to 1.

``json { "op": "correlate", "require": ["clean"], "options": { "method": "spearman", "minCoefficient": 0.3, "limit": 20 } } ``

Terminal, like assert: it returns {@link Association} records rather than rows, ordered strongest first.

A coefficient is not a cause. Two measures derived from one another — a total and its own component — will correlate near 1 and mean nothing.

Joins two streams on aggField, optionally pads, then applies an Enhance pass to compute derived fields. Used by the diff/change visualizations.

Period-over-period deltas, and the stock/flow bridge.

For each configured measure emits the absolute change, the relative change, and (for a stock) the implied flow — the first difference of a level. Rows are ordered by the date column; the first period gets null, not 0, so a missing baseline never reads as "no change".

``json { "op": "diffcalc", "options": { "date": "month", "measures": [ { "column": "users", "kind": "stock", "flowInto": "users_added" }, { "column": "revenue", "kind": "flow", "lag": 12, "suffix": "_yoy" } ] } } ``

With no measures, every column tagged uatu:measure:stock or uatu:measure:flow is picked up automatically.

Adds derived columns.

Each entry names the column to write and how to compute it: expr for a Polish-notation expression over other columns, or func for a windowed function such as cumsum accumulated along cumulateOn. With nullSafe a missing input yields an absent result rather than a spurious zero.

Parses the frame into the types the rest of the graph assumes.

A CSV arrives as strings. Dates are parsed with options.dateFormat for every column carrying options.dateTag, numbers for every column carrying options.numberTag — by tag, so the same op serves any structure.

Attaches coordinates by looking rows up in external geocoding files.

The files are named in options.geocoding and fetched through the context's ExternalFetcher, so this package never depends on an HTTP client of its own.

Like Deviation but reshapes both upstream GeoJSON FeatureCollections into row-oriented data first, joins, then re-wraps the result as GeoJSON for map rendering.

Turns rows into a GeoJSON FeatureCollection.

Coordinates come from the columns tagged gcx:lat and gcx:lon, and every other field rides along as a feature property. Also records each numeric property's range, which is what a choropleth needs to pick a scale.

Applies the filter every stream shares.

Use it above the fork point, for the constraints that belong to the whole page — the period under study, the company being looked at.

Reserved: registered as heatmap, but not implemented.

It inherits the pass-through run, so a graph naming it gets its input back unchanged rather than an error. Named here so a config written against it keeps loading.

Bring the columns of one frame onto the rows of another, matched on a key.

A left join by default, because the usual reason to reach for this is enrichment — targets onto actuals, a lookup table onto transactions — and losing the rows that did not match would quietly change every total computed downstream. inner is there when the match itself is the question.

A right-hand key with several rows is a fan-out waiting to happen, so only the first match is taken: a join that silently multiplies row counts breaks every sum below it, and is the classic way a dashboard starts overstating.

Applies the filter belonging to one stream.

Keyed by options.identifier, so two views reading the same graph can narrow their own copy without disturbing each other.

Maps op-name strings (as referenced from processing.json) to op classes. Pre-populated with the eleven built-in ops by Processor's default constructor; callers can register additional custom ops at runtime.

Cross-tabulation: dimensions down the side, dimensions across the top, one aggregated measure in the cells.

aggregate rolls a frame up along a single axis. This puts a second axis across the top, which is what it takes to see how two dimensions interact rather than each one's totals separately.

``json { "op": "pivot", "require": ["clean"], "options": { "rows": ["product_family"], "columns": ["region"], "measure": "revenue", "type": "sum", "totals": true } } ``

Emits one plain record per row key — the row dimensions, then one field per column bucket — so a grid or a chart consumes it without unpacking a nested shape.

Column buckets are capped: pivoting across something high-cardinality would otherwise emit thousands of fields. The tail is collected under otherLabel and counted in {@link Pivot.getOmitted}, never dropped without saying so.

Orchestrates streaming data through a graph of named ops.

Held data: - start: the immutable input dataframe (after workOn) - work: the current "live" dataframe (mutated in place by ops) - one OLAP Cube materialized from the dimension hierarchy

Streams are keyed by an identifier string. Multiple consumers can subscribe to independent processed streams concurrently.

Revenue recognition — the matching principle, mechanized.

Spreads an amount booked in one period across the periods it actually serves, so a twelve-month prepayment stops looking like twelve months of margin on the day it lands. Also emits the deferred balance: the obligation still owed.

``json { "op": "recognize", "options": { "amount": "cash_yearly", "date": "month", "term": 12, "into": "recognized_yearly", "deferredInto": "deferred_yearly" } } ``

term is either a constant number of periods or, via termColumn, a column holding a per-row term. Rows are matched on their position in the date order, so the frame must carry one row per period (no gaps) — {@link Cohort} or a prior aggregate normally guarantees that.

Amounts scheduled past the last row are not lost: they stay in the final deferred balance and are counted in {@link Recognize.getSpill}.

Reserved: registered as regionify, but not implemented.

Like {@link Heatmap}, it passes the frame through untouched.

Reference {@link ColumnDirectory} backed by a structure.json document.

The Angular client wraps its own service; this exists so the package is usable — and testable — on its own, with no host framework.

Stack several frames into one.

The simplest thing two branches can do once the graph allows them: last year and this year, two regions prepared differently, a forecast beside an actual. Rows keep their own columns — a column missing from one input is simply absent on those rows rather than filled with a zero nobody measured.

sourceInto writes which branch each row came from, which is usually what makes the union worth doing: it turns the origin into a dimension you can then group by.

ratio reads its own two columns off the fact rather than a single value.

One scored pair: which two columns, how strongly they move together, and by what method.

What a check is run against, beyond the rows themselves.

Column metadata directory consumed by ops that need to look up which raw columns are tagged a certain way (e.g. all date columns, all geocodable address columns).

Implement this with whatever backing store fits (a JSON document, a dbt manifest, a config file). The Angular adapter wraps a service that loads it from assets/structure.json.

Which pairs to score, and how strong a result has to be to be worth reporting.

Asserts one measure is at least as large as another, every period.

Unit-aware: comparing a column tagged in terabytes against one in dollars skips rather than answering a question nobody asked.

Asserts two measures stay within a relative gap of each other.

For the pairs that should track — a cash and an accrual basis of the same quantity — where the interesting event is them coming apart.

Which key to match on, which rows to keep, and what to carry across.

Asserts a measure never falls (or never rises) across the ordered periods.

The contract an op implements: run over a frame, and say what it needs loaded first.

Context handed to every op at construction. Held on AbstractOp so subclasses can access the column directory, fetch external resources, and ask the host pipeline for filter state.

How to lay out a cross-tab, and how to fold the measure in its cells.

The shape of processing.json: a map of process names to specs.

Everything a {@link Processor} needs to run a graph.

A single named transformation step in a process graph.

Asserts a ratio stays inside an expected band.

A conversion rate above 1 or below 0 is a broken denominator, not a result.

Asserts a measure keeps its sign.

A column entry in a structure.json document.

The parsed structure.json: what every column is, and how it aggregates.

How to stack the frames, and whether to record where each row came from.

What a check concluded.

Carries enough to render without re-running anything: a one-line summary, how many periods were evaluated, how many satisfied it, and the rows that broke it.

Asserts a period is settled enough to be compared with the others.

A 30-day-active metric for a cohort that started a week ago is not low, it is unfinished. This is what stops the newest bar being read as a decline.

Aggregation kinds, shared by every op that rolls values up.

aggregate and pivot accumulate the same way on purpose: a sum in a pivot cell has to mean what a sum in a rolled-up row means, or the two views of the same figure disagree.

Any check, discriminated by type.

The four answers a check can give. skip is not a failure: it means the check could not run.

External resource fetcher. Ops that need to load files (e.g. geocoding GeoJSON) request them through this. The Angular adapter wraps HttpClient; in Node you'd pass a function backed by fetch or axios.

Constructor signature expected by the registry.

What kind of pair a coefficient describes.

True when no verdict failed. Warnings do not fail a suite.

Ascending comparator over a date-ish column. Unparseable values compare equal.

Every tag that marks a column as a dimension.

uatu:dimension:time, :geo and :cohort are refinements of uatu:dimension, so a structure carrying only the refinement still has a dimension. Reading the base tag alone makes a dataset whose only axis is time look like it has no axes at all.

Coerce to a finite number, or null. Blank cells are absent, not zero.

Run a list of checks. Order preserved.

Tags that mark the time dimension, newest vocabulary last.

Two vocabularies are in circulation. Config repos authored for gestaltbi-core (see GestaltBI/sample-config) rename the date column to the canonical code uatu:date via mapping.json and tag it uatu:timedimension. @gestaltbi/infer emits the newer uatu:dimension / uatu:dimension:time pair instead. Both are accepted everywhere so an op works against either.

Tag vocabulary.

Tags are the contract between structure.json and the ops: an op never hard-codes a column name, it asks the ColumnDirectory for the columns carrying a tag. Everything here is a plain string so a structure document authored by hand (or by @gestaltbi/infer) stays readable.