Teaching Hedgehogs to Hibernate: Suspendable Query DAGs at Permutive

Andi Miller

7th October 2026

Some Background

Who are Permutive?

Privacy-first advertising platform for publishers, media companies and advertisers
Audience targeting without third-party cookies
Events and identity data at massive scale

What are Insights at Permutive?

The shape of the problem

Volume

Events	138k/s at daily peak
Datapoints	7.5m/s at daily peak
Page views	40bn over 30 days
Devices	4.3bn over 30 days

Dimensionality

Dimensions	~20
Cardinality	min 4, median 24, max 10k
Lookback	30 days to 1 year

Count Distinct

Nearly every metric is a Count Distinct, and at that scale we have to approximate them: Apache DataSketches

Sketch	Union	Full Set Operations	Notes
HyperLogLog	✓	✗	smallest in memory
CPC	✓	✗	smallest on disk
Theta	✓	✓	most featureful sketch
Tuple	✓	✓	carries a value per key

So how do we cope with that scale? Well, nearly every number in those screenshots boils down to a count distinct — distinct users, distinct views — and counting distinct over four billion devices exactly just isn’t feasible. So we approximate, with Apache DataSketches: fixed-size, mergeable summaries of sets. There’s a family of them with different tradeoffs. HyperLogLog and CPC are the cheapest if all you ever do is union. But remember those boolean cohort combinations — intersections — that’s what Theta gives you. And Tuple is a Theta that carries a value alongside each key, so you can do things like views per user. You’ll see these flowing through our query graphs later. The names there are links, by the way — I’ve got little scala.js simulators on my site that show how each of these actually work, if you want to play with them after.

Architecture?

Medallion Architecture

Layer	Store	Data
Bronze	BigQuery	raw data
Silver	Iceberg	cleaned and organised
Gold	Apache Pinot	fully indexed

The Problem

We serve all insights from Pinot today, but it’s not feasible to generate full indexes for every query, and the unindexed queries can harm cluster performance.

New Architecture

How does this perform?

Pinot: p99 of 30 seconds

BigQuery: p99 of… 30 minutes?

How can we run a query Dag?

Real World Example

Imperative visitor

val results = mutable.Map.empty[String, Set[UserId]]

for id <- graph.topologicalSort do
  results(id) = run(graph.nodeMap(id), inputs(results, id))

Simple, and you have full control
Sequential: fast nodes queue behind slow ones
Make it concurrent and the state escapes into threads and in-flight work

Catamorphism

val algebra: Algebra[QueryF, IO[Set[UserId]]] =
  Algebra:
    case QueryF.Query(sql)  => pinot.query(sql)
    case QueryF.Union(a, b) => (a, b).mapN(_ ++ _)

scheme.cata(algebra).apply(queryTree)

Principled: the fold is the program
Executes in structural order, not readiness order
Can’t serialise half a fold

What do we actually want?

Concurrent: start every node as soon as its inputs are ready
Run state somewhere we can see it, and serialise it
Plain cats-effect, no workflow platform

Hedgehogs

What does it do?

DataGraph[Id, NodeData, EdgeData] — directed edges, data on nodes and edges

enum Node:
  case Query(sql: String)
  case Union

val graph: DataGraph[String, Node, Unit] =
  DataGraph
    .empty[String, Node, Unit]
    .addNode("cohort-a", Node.Query("select ..."))
    .addNode("cohort-b", Node.Query("select ..."))
    .addNode("union", Node.Union)
    .addEdge("cohort-a", "union", ())
    .addEdge("cohort-b", "union", ())

What does that look like?

Mermaid.flowchart(graph)(
  extractNodeId = identity,
  extractNodeName = node => Some(node.toString)
)(
  direction = MermaidDirection.LeftRight,
  defaultNode = MermaidShape.Stadium
)

Functional Visitor

type UserId = String

object QueryVisitor
    extends SimpleDagVisitor[IO, String, Node, Set[UserId], Unit]:
  def run(id: String, node: Node,
          inputs: Map[String, Set[UserId]]): IO[Set[UserId]] =
    node match
      case Node.Query(sql) => pinot.query(sql)
      case Node.Union      => IO.pure(inputs.values.toSet.flatten)

val results: IO[DataGraph[String, Set[UserId], Unit]] =
  DagVisitor.runConcurrent(QueryVisitor)(graph)

Starts every node as soon as its inputs are done

The Changes to support Suspension

Abstraction

While a run is in progress, every node is in one of two states:

// Left: not run yet, Right: finished
type NodeState[In, Out] = Either[In, Out]

The run’s entire state is the graph of these
A node can run when all its inputs are Right

Interface

trait SimpleDagVisitor[F[_], Id, In, Out, Edge]:
  def run(id: Id, node: In, inputs: Map[Id, Out]): F[Out]

run must always return an F[Out]

Old Logic

Abstraction changes

Our Either can’t say “waiting”, so it grows a third case:

enum NodeState[In, Out, Wait]:
  case Pending(input: In)
  case Waiting(input: In, handle: Wait)
  case Done(output: Out)

Waiting holds a handle to the external work
Deliberately no Running: a fibre doesn’t survive a restart

New Interface

enum StepResult[Out, Wait]:
  case Complete(output: Out)
  case Suspend(handle: Wait)

trait SimpleSuspendableDagVisitor[F[_], Id, In, Out, Edge, Wait, Payload]:
  def run(id: Id, node: In,
          inputs: Map[Id, Out]): F[StepResult[Out, Wait]]
  def resume(id: Id, node: In, handle: Wait, payload: Payload,
             inputs: Map[Id, Out]): F[StepResult[Out, Wait]]

run can now answer “not yet”: Suspend with a handle
resume is called when the external work delivers a payload
Both can complete, or suspend again

New Logic

Our visitor, suspendable

import StepResult.*

type JobId = String

object QueryVisitor extends SimpleSuspendableDagVisitor[
    IO, String, Node, Set[UserId], Unit, JobId, Unit]:

  def run(id: String, node: Node, inputs: Map[String, Set[UserId]]) =
    node match
      case Node.Pinot(sql)    => pinot.query(sql).map(Complete(_))
      case Node.BigQuery(sql) => bq.submitJob(sql).map(Suspend(_))
      case Node.Union         => IO.pure(Complete(inputs.values.toSet.flatten))

  def resume(id: String, node: Node, handle: JobId, payload: Unit,
             inputs: Map[String, Set[UserId]]) =
    bq.readJobResult(handle).map(Complete(_))

Snapshot Example

{
  "version": 1,
  "direction": "flow",
  "nodes": [
    { "id": "cohort-a",
      "state": { "type": "done", "output": ["device-1", "device-2"] } },
    { "id": "cohort-b",
      "state": { "type": "waiting",
                 "input": { "type": "bigquery", "sql": "select ..." },
                 "handle": "bq-job-4ca1" } },
    { "id": "union",
      "state": { "type": "pending", "input": { "type": "union" } } }
  ],
  "edges": [
    { "from": "cohort-a", "to": "union", "data": {} },
    { "from": "cohort-b", "to": "union", "data": {} }
  ]
}

Wiring it into a service

def save(runId: RunId, run: RunResult[...]): IO[Unit] = run match
  case Finished(results) => db.storeResults(runId, results)
  case Suspended(snap)   => db.storeSnapshot(runId, snap)

HttpRoutes.of[IO]:
  case req @ POST -> Root / "query" =>
    for
      graph <- req.as[UserQuery].map(unfold)
      runId <- RunId.random
      run   <- SuspendableDagVisitor.start(QueryVisitor)(graph)
      _     <- save(runId, run)
      resp  <- Accepted(runId)
    yield resp

  case GET -> Root / "poll" / runId =>
    db.readResults(runId).flatMap:
      case Some(results) => Ok(results)
      case None          => Accepted("still running")

  case POST -> Root / "bq_callback" / jobId =>
    for
      (runId, nodeId, snap) <- db.readSnapshotFor(jobId)
      run  <- SuspendableDagVisitor.resume(QueryVisitor)(snap, Map(nodeId -> ()))
      _    <- save(runId, run)
      resp <- Ok()
    yield resp

Demo

andimiller.net/hedgehogs-demo

Final Comparison

	Imperative Visitor	Catamorphism	Suspendable Visitor
Scheduling	sequential	structural order	concurrent, by readiness
Run state	a map you could checkpoint	half-applied functions	a serialisable graph
Snapshots	✓ between nodes	✗	✓
Dependencies	none	recursion schemes	cats-effect