feat: prepare sketch-backed reaggregation: params, measure identity, Druid hooks

.gitignore

@@ -134,3 +134,6 @@ Untitled*
 postgres_metadata
 postgres_superset
 node_modules
+
+# Local Claude Code workspace state (large; never committed)
+.claude/

datajunction-server/datajunction_server/alembic/versions/2026_09_15_0000-fm0001params_add_params_to_frozen_measures.py

@@ -0,0 +1,29 @@
+"""
+Add params column to frozen_measures
+
+Adds a JSON column to persist tuning parameters (e.g., accuracy) for sketch-backed measures.
+
+Revision ID: fm0001params
+Revises: dv0001typed
+Create Date: 2026-09-15 00:00:00.000000+00:00
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "fm0001params"
+down_revision = "dv0001typed"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "frozen_measures",
+        sa.Column("params", sa.JSON(), nullable=True),
+    )
+
+
+def downgrade():
+    op.drop_column("frozen_measures", "params")

datajunction-server/datajunction_server/api/cubes.py

@@ -14,11 +14,11 @@
     _reorder_partition_column_last,
     build_combiner_sql_from_preaggs,
 )
+from datajunction_server.construction.build_v3.cte import strip_role_suffix
 from datajunction_server.construction.build_v3.cube_matcher import (
     _metric_graph_has_reaggregate,
     validate_cube_reaggregate_materialization,
 )
-from datajunction_server.construction.build_v3.cte import strip_role_suffix
 from datajunction_server.construction.dimensions import build_dimensions_from_cube_query
 from datajunction_server.database.materialization import Materialization
 from datajunction_server.database.node import Node
@@ -34,7 +34,6 @@
     AccessDenialMode,
     get_access_checker,
 )
-from datajunction_server.models.access import ResourceAction
 from datajunction_server.internal.materializations import (
     build_cube_materialization,
     stop_cube_materialization_workflows,
@@ -44,6 +43,7 @@
     get_single_cube_revision_metadata,
 )
 from datajunction_server.internal.views import CubeViewNames, _build_view_body
+from datajunction_server.models.access import ResourceAction
 from datajunction_server.models.cube import (
     CubeRevisionMetadata,
     DimensionValue,
@@ -63,11 +63,10 @@
 )
 from datajunction_server.models.dialect import Dialect
 from datajunction_server.models.materialization import (
-    DRUID_AGG_MAPPING,
-    DRUID_SKETCH_TYPES,
     Granularity,
     MaterializationJobTypeEnum,
     MaterializationStrategy,
+    get_druid_aggregator_spec,
 )
 from datajunction_server.models.metric import TranslatedSQL
 from datajunction_server.models.node_type import NodeNameVersion
@@ -179,27 +178,25 @@ def _build_metrics_spec(
             if internal_name:
                 component = component_by_name.get(internal_name)
 
-        druid_type = "longSum"  # Default fallback
-
-        if component:
-            # Use merge function for pre-aggregated data, fall back to aggregation
-            agg_func = component.merge or component.aggregation
-            if agg_func:
-                key = (col.type, agg_func.lower())
-                if key in DRUID_AGG_MAPPING:
-                    druid_type = DRUID_AGG_MAPPING[key]
-
-        metric_spec = {
+        metric_spec = (
+            get_druid_aggregator_spec(
+                column_name=col.name,
+                column_type=col.type,
+                aggregation=component.aggregation,
+                merge=component.merge,
+                params=component.params,
+            )
+            if component
+            else None
+        )
+        # Unmappable measures fall back to longSum here, since we're loading
+        # pre-aggregated data; the materialization config omits them instead.
+        metric_spec = metric_spec or {
             "fieldName": col.name,
             "name": col.name,
-            "type": druid_type,
+            "type": "longSum",
         }
 
-        # HLL sketches need additional configuration
-        if druid_type in DRUID_SKETCH_TYPES:
-            metric_spec["lgK"] = 12  # Log2 of K, controls precision (4-21)
-            metric_spec["tgtHllType"] = "HLL_4"  # HLL_4, HLL_6, or HLL_8
-
         metrics.append(metric_spec)
 
     return metrics

datajunction-server/datajunction_server/api/graphql/scalars/metricmetadata.py

@@ -1,6 +1,7 @@
 """Metric metadata scalars"""
 
 import strawberry
+from strawberry.scalars import JSON
 
 from datajunction_server.models.cube_materialization import (
     Aggregability as Aggregability_,
@@ -45,16 +46,39 @@ class Unit:
 class DimensionReaggregateRule: ...
 
 
-@strawberry.experimental.pydantic.type(model=ReaggregateSpec_, all_fields=True)
-class ReaggregateSpec: ...
+@strawberry.experimental.pydantic.type(model=ReaggregateSpec_)
+class ReaggregateSpec:
+    """
+    Metric reaggregation declaration.
+
+    Fields are listed explicitly rather than via `all_fields` because `params`
+    is an open dict, which has no automatic GraphQL mapping.
+    """
+
+    rules: strawberry.auto
+    params: JSON | None = None
 
 
 @strawberry.experimental.pydantic.type(model=AggregationRule_, all_fields=True)
 class AggregationRule: ...
 
 
-@strawberry.experimental.pydantic.type(model=MetricComponent_, all_fields=True)
-class MetricComponent: ...
+@strawberry.experimental.pydantic.type(model=MetricComponent_)
+class MetricComponent:
+    """
+    A single measure with accumulate/merge phases.
+
+    Fields are listed explicitly rather than via `all_fields` because `params`
+    is an open dict, which has no automatic GraphQL mapping.
+    """
+
+    name: strawberry.auto
+    expression: strawberry.auto
+    aggregation: strawberry.auto
+    merge: strawberry.auto
+    rule: strawberry.auto
+    grain_alias: strawberry.auto
+    params: JSON | None = None
 
 
 @strawberry.experimental.pydantic.type(model=DecomposedMetric_, all_fields=True)

datajunction-server/datajunction_server/api/graphql/scalars/node.py

@@ -433,6 +433,7 @@ def reaggregate(self, root: DBNodeRevision) -> ReaggregateSpec | None:
         if not spec:
             return None
         return ReaggregateSpec(
+            params=spec.params,
             rules=[
                 DimensionReaggregateRule(
                     dimension=rule.dimension,

datajunction-server/datajunction_server/api/graphql/schema.graphql

@@ -303,6 +303,7 @@ type MetricComponent {
   merge: String
   rule: AggregationRule!
   grainAlias: String
+  params: JSON
 }
 
 enum MetricDirection {
@@ -727,6 +728,7 @@ type Query {
 
 type ReaggregateSpec {
   rules: [DimensionReaggregateRule!]!
+  params: JSON
 }
 
 enum ReaggregationFunction {

datajunction-server/datajunction_server/construction/build_v3/combiners.py

@@ -600,6 +600,7 @@ async def build_combiner_sql_from_preaggs(
             measure_identity_token(
                 compute_expression_hash(m.expression),
                 m.aggregation,
+                m.params,
             )
             for m in gg.components
             if m.expression

datajunction-server/datajunction_server/construction/build_v3/preagg_matcher.py

@@ -19,6 +19,8 @@
 from datajunction_server.database.preaggregation import (
     PreAggregation,
     compute_expression_hash,
+    get_measure_identities,
+    measure_identity_token,
 )
 from datajunction_server.errors import DJInvalidInputException
 from datajunction_server.models.decompose import Aggregability, MetricComponent
@@ -42,10 +44,9 @@
 
 def get_required_measure_identities(
     grain_group: GrainGroup,
-) -> set[tuple[str, str]]:
+) -> set[str]:
     """
-    Identity of each measure a grain group needs: (expression hash, Phase-1
-    aggregation).
+    Identity token of each measure a grain group needs.
 
     Matching is name-independent, so ``expr_hash`` stands in for identity -- but
     it covers the expression alone, making ``SUM(x)`` and ``MAX(x)`` hash alike
@@ -55,11 +56,16 @@ def get_required_measure_identities(
 
     Phase-1 ``aggregation`` rather than ``merge``, because merge is not unique --
     COUNT accumulates with COUNT but merges with SUM.
+
+    Built through ``measure_identity_token`` rather than assembled here, so the
+    definition of identity lives in one place -- it also folds in sketch tuning
+    parameters, which a locally-built tuple would silently omit.
     """
     return {
-        (
+        measure_identity_token(
             compute_expression_hash(component.expression),
-            component.normalized_aggregation,
+            component.aggregation,
+            component.params,
         )
         for _, component in grain_group.components
     }
@@ -348,13 +354,9 @@ def find_matching_preagg(
                 continue
             join_back = coverage
 
-        # Coverage check on (expression hash, Phase-1 aggregation) -- see
-        # get_required_measure_identities for why the aggregation is part of it.
-        preagg_measures = {
-            (measure.expr_hash, measure.normalized_aggregation)
-            for measure in preagg.measures
-            if measure.expr_hash
-        }
+        # Coverage check on measure identity -- see
+        # get_required_measure_identities for what that comprises.
+        preagg_measures = get_measure_identities(preagg.measures)
         if not required_measures.issubset(preagg_measures):
             logger.debug(
                 f"[BuildV3] Pre-agg {preagg.id} measures {preagg_measures} "
@@ -406,13 +408,21 @@ def get_preagg_measure_column(
     Returns:
         Column name in the pre-agg, or None if not found
     """
-    target = (
+    target = measure_identity_token(
         compute_expression_hash(component.expression),
-        component.normalized_aggregation,
+        component.aggregation,
+        component.params,
     )
 
     for measure in preagg.measures:
-        if (measure.expr_hash, measure.normalized_aggregation) == target:
+        if (
+            measure_identity_token(
+                measure.expr_hash,
+                measure.aggregation,
+                measure.params,
+            )
+            == target
+        ):
             # Externally-registered pre-aggs bind the measure to a physical
             # column name that may differ from the DJ component name.
             return measure.source_column or measure.name

datajunction-server/datajunction_server/database/measure.py

@@ -119,6 +119,12 @@ class FrozenMeasure(Base):
     # How to aggregate the resolved expression (e.g., SUM, COUNT, AVG)
     aggregation: Mapped[str]
 
+    # Tuning parameters for `aggregation`, for sketch-backed measures whose
+    # function name alone doesn't pin down the accumulate (a t-digest still needs
+    # a compression). Part of the measure's identity: a sketch frozen at one
+    # accuracy must not be reused for a metric asking for another.
+    params: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
     # Additivity or rollup rule - tells the planner if this measure can be summed,
     # needs special handling, or is non-additive.
     rule: Mapped[MeasureAggregationRule] = mapped_column(MeasureAggregationRuleType)

datajunction-server/datajunction_server/database/preaggregation.py

@@ -6,6 +6,7 @@
 import json
 from datetime import UTC, datetime
 from functools import partial
+from typing import Any
 
 import sqlalchemy as sa
 from sqlalchemy import (
@@ -80,16 +81,64 @@ def compute_grain_group_hash(
     return hashlib.md5(content.encode()).hexdigest()
 
 
-def measure_identity_token(expr_hash: str, aggregation: str | None) -> str:
+def canonical_params(params: dict[str, Any] | None) -> str:
+    """
+    Stable string form of a component's tuning parameters, for identity.
+
+    Keys are sorted and numeric values normalized, so ``{"compression": 200}``
+    and ``{"compression": 200.0}`` cannot yield two identities for one sketch.
+    """
+    if not params:
+        return ""
+
+    def norm(value: Any) -> Any:
+        # 200 and 200.0 are the same compression; don't let the literal's
+        # spelling fork the identity. (bools are not floats, so they pass
+        # through untouched.)
+        if isinstance(value, float) and value.is_integer():
+            return int(value)
+        return value
+
+    # `default=str` so an exotic value can't raise from inside pre-agg matching.
+    # Params arrive as parsed YAML/JSON, so this should be unreachable; an opaque
+    # TypeError deep in the matcher is a bad way to find out otherwise.
+    return json.dumps(
+        {key: norm(params[key]) for key in sorted(params)},
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+
+
+def measure_identity_token(
+    expr_hash: str,
+    aggregation: str | None,
+    params: dict[str, Any] | None = None,
+) -> str:
     """
     Canonical token identifying one measure: expression hash plus Phase-1
-    aggregation.
+    aggregation, plus tuning parameters when the aggregation takes them.
 
     The hash alone is not an identity -- ``SUM(x)`` and ``MAX(x)`` hash alike --
     and a partial is reusable only by a metric that accumulates the same way.
     Everything comparing or hashing measures goes through this.
+
+    Parameters extend that reasoning to sketches: a t-digest accumulated at
+    ``compression=100`` is not interchangeable with one at ``compression=1000``,
+    though both are ``acme_tdigest_agg`` over the same expression. The segment is
+    appended **only** when params are present, so tokens for the overwhelming
+    majority of components -- plain ``SUM``, ``COUNT``, ``MAX`` -- are byte-identical
+    to what this returned before params existed. Stored ``preagg_hash`` values are
+    built from these tokens, so that stability is a compatibility requirement, not
+    a nicety.
+
+    Note the quantile *fractions* a metric asks for are deliberately not here. One
+    sketch column serves p50, p95 and p99; folding fractions in would fragment it
+    into three identical columns.
     """
-    return f"{expr_hash}:{(aggregation or '').strip().upper()}"
+    base = f"{expr_hash}:{(aggregation or '').strip().upper()}"
+    suffix = canonical_params(params)
+    return f"{base}:{suffix}" if suffix else base
 
 
 def get_measure_identities(measures: list[PreAggMeasure]) -> set[str]:
@@ -103,7 +152,7 @@ def get_measure_identities(measures: list[PreAggMeasure]) -> set[str]:
         Set of identity tokens
     """
     return {
-        measure_identity_token(m.expr_hash, m.aggregation)
+        measure_identity_token(m.expr_hash, m.aggregation, m.params)
         for m in measures
         if m.expr_hash
     }