ReactiveDag.Node.Rows (reactive_dag v0.17.0-rc.40)

Copy Markdown View Source

Reads a cell's own rows, keyed the way the DAG keys them.

A node's rows live in its resource, but the DAG addresses them by cell key — a "|"-joined identity for a composite-PK node, a single column's value otherwise. Anything that wants to ask a question about a cell rather than about a table (ReactiveDag.Insights, ReactiveDag.Verdict, a union reading its inputs) needs the rows under those keys, not under the resource's own primary key.

This is the read side of what ReactiveDag.Node.Payload writes, and it deliberately mirrors that module's key derivation: identity_fields when the node is identity-keyed, payload_key (defaulting to :key) otherwise.

Why it is not the coordination tuple

These reads used to go to reactive_dag_tuple, which carried a status per (cell_id, key). That made the tuple a second home for derived results — one with a fixed two-column schema, updated by a writer the host had to configure, and queryable only through this library. Reading the resource instead means a status is an ordinary column with ordinary Ash semantics: policies apply, loads work, and a host can add a second column without asking us.

Missing columns are not errors

A node need not have a :status column — most don't; a rollup has sums. Rows from such a node come back with status: nil, and callers that count statuses simply find nothing to count. A node with no attributes at all keeps its rows somewhere else entirely and reads as empty. Asking is always safe.

Summary

Types

one of a cell's rows, addressed by cell key

Where a node's rows live and how they are keyed — a cell's meta, or the same three fields lifted out of it (which is what a union carries for each of its inputs, since it reads rows it does not own).

Functions

Every row the cell currently holds, as %{key:, status:, record:}.

Clear the stored fingerprint on keys, so the next recompute treats those rows as needing work.

How many units the cell holds, counted by the datastore.

The keys whose status is in statuses, at most limit of them (sorted, so a sample is stable between calls rather than reshuffling on every render).

The keys whose rows match filter — the selection behind "reprocess just this year".

Turn a selected slice into the options a POLL is asked with.

Reconcile a leaf's rows against the key set a scan found — the algorithm every leaf driver otherwise hand-rolls.

The dimensions this node declared a human may select it by, with their options resolved.

%{status => count} over the cell's rows — the histogram Insights shows and Verdict folds into one answer.

Types

row()

@type row() :: %{key: String.t(), status: String.t() | nil, record: struct()}

one of a cell's rows, addressed by cell key

source()

@type source() :: %{
  optional(:resource) => module() | nil,
  optional(:payload_key) => atom() | nil,
  optional(:identity_fields) => [atom()] | nil
}

Where a node's rows live and how they are keyed — a cell's meta, or the same three fields lifted out of it (which is what a union carries for each of its inputs, since it reads rows it does not own).

Functions

all(source)

@spec all(ReactiveDag.Cell.t() | source()) :: [row()]

Every row the cell currently holds, as %{key:, status:, record:}.

Returns [] for a node that keeps no rows here — no resource at all, or a resource with no attributes (the shape a compute/custom-upsert: node has, where the real writes land somewhere this library never sees). That is different from "holds nothing", so a caller that must tell the two apart should check meta[:resource] itself.

Raises whatever the underlying Ash.read!/1 raises. Callers on a display path (Insights) already run these behind their own rescue; a caller on a compute path wants the failure.

invalidate(source, keys)

@spec invalidate(ReactiveDag.Cell.t() | source(), [String.t()] | :all) :: [String.t()]

Clear the stored fingerprint on keys, so the next recompute treats those rows as needing work.

A fingerprint answers "did the input move?". After a prompt change or a fixed fold the input has NOT moved and the answer is still valid — it is simply the wrong question, because what changed was the function. A per_key node would therefore skip exactly the rows you asked it to redo.

Clearing the stored value makes the comparison fail honestly rather than bypassing it: a null fingerprint means "no valid prior result", which is precisely true once the code that produced it has changed. Nothing needs a force flag threaded through the recompute, and the next run stores a fresh fingerprint as it always would.

A node with no fingerprint column has nothing to clear and recomputes regardless, so this is a no-op there. Returns the keys it actually cleared.

key_count(source)

@spec key_count(ReactiveDag.Cell.t() | source()) :: non_neg_integer()

How many units the cell holds, counted by the datastore.

Ash.count! rather than reading the rows: an overview wants the number, and loading a table to reduce it to one integer decodes every payload column on the way — which for a node whose rows carry a JSON blob is the whole cost of the read for none of the value.

Returns 0 for a node that keeps no rows here.

keys_by_status(cell_or_source, statuses, opts \\ [])

@spec keys_by_status(ReactiveDag.Cell.t() | source(), [String.t() | nil], keyword()) ::
  [String.t()]

The keys whose status is in statuses, at most limit of them (sorted, so a sample is stable between calls rather than reshuffling on every render).

keys_where(source, filter)

@spec keys_where(
  ReactiveDag.Cell.t() | source(),
  keyword()
) :: [String.t()]

The keys whose rows match filter — the selection behind "reprocess just this year".

Rows.keys_where(cell, fiscal_year: "FY25")
#=> ["FY25|gf", "FY25|water"]

The filter goes to the datastore; keys are built here, because an identity-keyed node's key is a "|"-join no datastore knows. So this decodes what matched rather than the table.

A node declares which columns are meant for this with slice, and slices/1 reports them — but nothing stops a caller filtering on any column it knows about. The declaration is what makes a UI possible, not what makes the filter legal.

poll_opts(cell_or_source, selection)

@spec poll_opts(ReactiveDag.Cell.t() | source(), map() | keyword()) :: keyword()

Turn a selected slice into the options a POLL is asked with.

Rows.poll_opts(cell, %{"fiscal_year" => "FY25/26"})
#=> [fiscal: "FY25/26"]

A slice narrows two different things. keys_where/2 filters rows this node already HOLDS — "re-derive FY25 from documents I have". This narrows the FETCH: a source whose upstream is addressable by the same dimension can be asked for just that part, and a crawler that takes fiscal: walks twelve months instead of the whole corpus.

The selection is keyed by COLUMN, because that is what a UI has — it rendered a button per value under the slice's own name. The result is keyed by poll_as, because that is the scanner's vocabulary. Translating here is the point: neither side has to learn the other's spelling.

Accepts string or atom column names, since a selection usually arrives from a form or a job argument. A column this node never declared as a slice is IGNORED rather than passed through — an unrecognised option would otherwise reach poll/1 as if the node had offered it, and a scanner that pattern matches its arguments would crash on a typo the DSL could not vouch for.

reconcile(cell, want_keys, opts)

@spec reconcile(ReactiveDag.Cell.t(), [String.t()] | MapSet.t(), keyword()) ::
  {:ok, [String.t()],
   %{
     created: [String.t()],
     updated: [String.t()],
     revived: [String.t()],
     retired: [String.t()]
   }}

Reconcile a leaf's rows against the key set a scan found — the algorithm every leaf driver otherwise hand-rolls.

Returns {:ok, changed, detail}changed is the flat list that propagates, and detail is %{created:, updated:, revived:, retired:} saying WHY each key is in it.

The detail is free: reconcile/3 computes those four sets to build changed and used to flatten them away, leaving a host to rebuild the classification a scan report shows. It cannot be reconstructed afterwards — by the time you look, the rows are already written.

current  = the cell's current keys (read from its resource)
want     = `want_keys`, what the scan found
upsert   each want key    the host writes the row, returns true iff CHANGED
vanished = current  want  retired
 changed_upserts ++ vanished    (the keys to propagate)

:upsert is called once per want-key, in either of two forms:

  • (key -> row | nil) — the common case. Return the row you observed and the library writes it, deciding changed? against the node's declared fingerprint (or every attribute, if it declares none). Return nil for a key you could not observe: nothing is written and the key is not reported, which is how a partial outage stays honest.

  • (key -> boolean | :created | :changed | :unchanged) — full control. Write the row yourself and say what happened. Use this when the write is not an upsert into the node's own resource.

    true/false mean :changed/:unchanged. Prefer the atoms when you know which: only you can tell an insert from an update, since the library never saw the row, and true reports a brand-new key as updated — which it cannot be, having had no prior row.

  • :observed — was this scan COMPLETE? :all (the default) means absence from want_keys is evidence a key is gone, so the vanish diff runs. :partial means the scan looked at only part of the upstream — a scoped only: poll, a windowed recent: one, a crawl whose index page failed — so absence means "not looked at" and nothing can vanish. See below.

  • :retire — how vanished keys leave. See the table below.

  • :current — the baseline vanished is computed against. Defaults to the cell's current keys. Pass it when your live set is narrower than all your rows, or when the scan asked a NARROWER QUESTION than "everything" — a date-scoped scan with a whole-table baseline retires everything outside its window.

When a key stops being returned

One decision, three answers:

you wantyou writethe rowpropagates?
destroy itnothing — the defaultdestroyedyes
keep itretain_if_vanished true on the nodeuntouchedno
mark itretain_if_vanished mark: &tombstone/1yours to writeyes

Keep and mark are the same operation with one question between them — do we write something to say it is gone? — and propagation follows from the answer rather than being a separate switch:

  • destroying removes a unit downstream was counting, so it is a change;
  • keeping changes nothing — the row is still there, unmodified — so reporting it would be a lie, and would report it again on every poll forever, since nothing marks it as handled;
  • marking is a change you made, so downstream hears about it. It also means you own :current, and inherit one case a fingerprint cannot see: a marked-retired row returning with unmoved bytes. changed? compares fingerprints, so the revival is invisible — the library warns when it sees that shape but cannot fix it (u2i/reactive_dag#82). Report it yourself with the boolean :upsert form.

A leaf declaring fingerprint needs no :upsert at all for the row-returning form to be worth using — that is the point: poll/1 becomes fetch, build rows, call reconcile.

Partial observations

Retiring a key is an inference: the upstream no longer lists it, so it is gone. That inference is only valid from a complete observation. A scoped poll, a windowed one, or a crawl whose index page failed all produce a want-set that is real but incomplete, where absence means "not looked at".

Rows.reconcile(cell, observed_keys, observed: :partial, upsert: &fetch/1)

Nothing vanishes, nothing is retired, and the keys you did see are written and reported exactly as usual.

This was previously spelled current: [] — "measure vanishing against nothing" — which works, and says nothing about why. The distinction is worth a name because the failure is silent and severe in one direction only: getting :partial wrong under-retires, leaving rows that should have gone. Getting :all wrong tombstones everything the scan did not happen to look at, which for an archival consumer is a mass-deletion wave from one upstream 500.

The honest gap

Call this only with a want-set you actually observed. An upstream you could not reach must write NOTHING — handing an empty want_keys to a scan that failed retires every key the cell has, and a downstream rollup over an empty set typically reads as vacuously fine. A scan that couldn't look must never render as a scan that found nothing.

A total outage and a partial one differ: an outage writes nothing at all, so it marks nothing dirty and the drain correctly does no downstream work. A partial observation writes what it saw — it just must not conclude anything from what it did not.

slices(source)

@spec slices(ReactiveDag.Cell.t() | source()) :: [map()]

The dimensions this node declared a human may select it by, with their options resolved.

Rows.slices(cell)
#=> [%{column: :fiscal_year, label: "fiscal_year",
#      values: ["FY24", "FY25"], poll_as: :fiscal}]

values is nil when the node named no options — a UI then takes free text, or offers nothing.

poll_as is what to call this dimension when asking the SCANNER for it, which defaults to the column. See poll_opts/2.

status_histogram(source)

@spec status_histogram(ReactiveDag.Cell.t() | source()) :: %{
  required(String.t() | nil) => non_neg_integer()
}

%{status => count} over the cell's rows — the histogram Insights shows and Verdict folds into one answer.

Rows with no status are counted under nil, so the counts always sum to the cell's key count and a node without a :status column reports %{nil => n} rather than lying with %{}.