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_08_24_0000-rg0001reaggregate_add_reaggregate_to_noderevision.py

@@ -0,0 +1,27 @@
+"""
+Add reaggregate column to noderevision
+
+Revision ID: rg0001reaggregate
+Revises: cm0003dropowner
+Create Date: 2026-08-24 00:00:00.000000+00:00
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "rg0001reaggregate"
+down_revision = "cm0003dropowner"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "noderevision",
+        sa.Column("reaggregate", sa.JSON(), nullable=True),
+    )
+
+
+def downgrade():
+    op.drop_column("noderevision", "reaggregate")

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: rg0001reaggregate
+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 = "rg0001reaggregate"
+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

@@ -15,6 +15,9 @@
     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 (
+    validate_cube_reaggregate_materialization,
+)
 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
@@ -30,7 +33,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,
@@ -40,6 +42,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,
@@ -59,11 +62,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
@@ -175,32 +177,56 @@ 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
 
 
+async def _validate_cube_reaggregate_materialization(
+    session: AsyncSession,
+    cube: Node,
+) -> None:
+    """
+    Validate materialization safety using full metric decomposition.
+    """
+    if not cube.current:  # pragma: no cover
+        return
+
+    from datajunction_server.construction.build_v3.builder import setup_build_context
+
+    ctx = await setup_build_context(
+        session=session,
+        metrics=cube.current.cube_node_metrics,
+        dimensions=cube.current.cube_node_dimensions,
+        filters=cube.current.cube_filters or None,
+        dialect=Dialect.SPARK,
+        use_materialized=False,
+    )
+    validate_cube_reaggregate_materialization(
+        cube.current,
+        decomposed_metrics=ctx.decomposed_metrics,
+    )
+
+
 @router.get("/cubes", name="Get all Cubes")
 async def get_all_cubes(
     *,
@@ -288,6 +314,7 @@ async def cube_materialization_info(
             message=f"Cube node `{name}` does not exist.",
             http_status_code=404,
         )
+    await _validate_cube_reaggregate_materialization(session, node)
     temporal_partitions = node.current.temporal_partition_columns()  # type: ignore
     if len(temporal_partitions) != 1:
         raise DJInvalidInputException(
@@ -520,6 +547,7 @@ async def materialize_cube(
             message=f"Cube '{name}' has no current revision",
             http_status_code=HTTPStatus.NOT_FOUND,
         )
+    await _validate_cube_reaggregate_materialization(session, node)
 
     cube_tps = cube_revision.temporal_partition_columns()
 

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_,
@@ -15,9 +16,15 @@
     MetricComponent as MetricComponent_,
 )
 from datajunction_server.models.node import MetricDirection as MetricDirection_
+from datajunction_server.models.reaggregate import (
+    DimensionReaggregateRule as DimensionReaggregateRule_,
+    ReaggregateSpec as ReaggregateSpec_,
+    ReaggregationFunction as ReaggregationFunction_,
+)
 
 MetricDirection = strawberry.enum(MetricDirection_)
 Aggregability = strawberry.enum(Aggregability_)
+ReaggregationFunction = strawberry.enum(ReaggregationFunction_)
 
 
 @strawberry.type
@@ -32,12 +39,48 @@ class Unit:
     abbreviation: str | None
 
 
+@strawberry.experimental.pydantic.type(
+    model=DimensionReaggregateRule_,
+    all_fields=True,
+)
+class DimensionReaggregateRule: ...
+
+
+@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.
+    """
+
+    fn: strawberry.auto
+    weight: strawberry.auto
+    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

@@ -31,7 +31,9 @@
 )
 from datajunction_server.api.graphql.scalars.metricmetadata import (
     DecomposedMetric,
+    DimensionReaggregateRule,
     MetricMetadata,
+    ReaggregateSpec,
 )
 from datajunction_server.api.graphql.scalars.user import User
 from datajunction_server.api.graphql.utils import extract_fields
@@ -45,6 +47,7 @@
 from datajunction_server.models.node import NodeMode as NodeMode_
 from datajunction_server.models.node import NodeStatus as NodeStatus_
 from datajunction_server.models.node import NodeType as NodeType_
+from datajunction_server.models.reaggregate import parse_reaggregate_spec
 from datajunction_server.sql.parsing.backends.antlr4 import ast, parse
 
 NodeType = strawberry.enum(NodeType_)
@@ -410,6 +413,29 @@ def materializations(
     # Only metrics will have these fields
     required_dimensions: list[Column] | None = None
 
+    @strawberry.field
+    def reaggregate(self, root: DBNodeRevision) -> ReaggregateSpec | None:
+        """
+        Metric reaggregation declaration.
+        """
+        if root.type != NodeType.METRIC:
+            return None
+        spec = parse_reaggregate_spec(root.reaggregate)
+        if not spec:
+            return None
+        return ReaggregateSpec(
+            fn=spec.fn,  # type: ignore
+            weight=spec.weight,
+            params=spec.params,
+            rules=[
+                DimensionReaggregateRule(
+                    dimension=rule.dimension,
+                    fn=rule.fn,  # type: ignore
+                )
+                for rule in spec.rules
+            ],
+        )
+
     @strawberry.field
     def primary_key(self, root: DBNodeRevision) -> list[str]:
         """

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

@@ -7,6 +7,7 @@ enum Aggregability {
 type AggregationRule {
   type: Aggregability!
   level: [String!]
+  reaggregate: DimensionReaggregateRule
 }
 
 type Attribute {
@@ -149,6 +150,11 @@ type DimensionLink {
   defaultValue: String
 }
 
+type DimensionReaggregateRule {
+  dimension: String!
+  fn: ReaggregationFunction!
+}
+
 type Engine {
   name: String!
   version: String!
@@ -296,6 +302,7 @@ type MetricComponent {
   merge: String
   rule: AggregationRule!
   grainAlias: String
+  params: JSON
 }
 
 enum MetricDirection {
@@ -397,6 +404,7 @@ type NodeRevision {
   dimensionLinks: [DimensionLink!]!
   availability: AvailabilityState
   materializations: [MaterializationConfig!]
+  reaggregate: ReaggregateSpec
   primaryKey: [String!]!
   metricMetadata: MetricMetadata
   isDerivedMetric: Boolean!
@@ -716,6 +724,25 @@ type Query {
   listNamespaces: [Namespace!]!
 }
 
+type ReaggregateSpec {
+  fn: ReaggregationFunction
+  weight: String
+  rules: [DimensionReaggregateRule!]!
+  params: JSON
+}
+
+enum ReaggregationFunction {
+  AUTO
+  NONE
+  SUM
+  AVG
+  WEIGHTED_AVG
+  LAST_VALUE
+  FIRST_VALUE
+  MIN
+  MAX
+}
+
 type SemanticEntity {
   name: String!