dialect aware sketch decomposition

.gitignore

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

datajunction-clients/python/datajunction/client.py

@@ -207,31 +207,7 @@ def plan(
         include_temporal_filters: bool = False,
         lookback_window: str | None = None,
     ):
-        """
-        Returns a query execution plan for the given metrics and dimensions.
-
-        The plan shows:
-        - grain_groups: How metrics are grouped and their intermediate SQL
-        - metric_formulas: How each metric combines its components
-        - requested_dimensions: The dimensions being queried
-
-        This is useful for understanding how DJ decomposes metrics into
-        atomic aggregations and how multiple fact tables are joined together.
-
-        Args:
-            metrics: List of metric names to include
-            dimensions: List of dimensions to group by
-            filters: List of filter expressions
-            cube: Optional cube node name. When provided, the cube's stored
-                filters are automatically prepended to the query filters.
-            dialect: SQL dialect (e.g., 'spark', 'trino'). Defaults to engine dialect.
-            use_materialized: Whether to use materialized tables when available
-            include_temporal_filters: Whether to include temporal partition filters.
-                Only applies if the metrics and dimensions resolve to a cube with
-                temporal partitions.
-            lookback_window: Lookback window for temporal filters (e.g., '3 DAY',
-                '1 WEEK'). Only applicable when include_temporal_filters is True.
-        """
+        """Returns a query execution plan for the given metrics and dimensions."""
         params: dict = {
             "metrics": metrics,
             "dimensions": dimensions or [],

datajunction-clients/python/datajunction/deployment.py

@@ -342,30 +342,7 @@ def build_codeowners(
         default_owner: str | None = None,
         exclude_dirs: list[str] | None = None,
     ) -> int:
-        """
-        Generate a CODEOWNERS file from the owners fields in DJ node YAML files.
-
-        Walks base_dir recursively, reads every *.yaml file (skipping dj.yaml),
-        and maps each file path to its owners list.  Files with no owners are
-        omitted.  Paths in the output are relative to base_dir and prefixed with
-        / so GitHub resolves them from the repo root.
-
-        If github_api_url is provided (and GITHUB_TOKEN / github_token_env is set),
-        email addresses in owners fields are resolved to GitHub usernames via the
-        search API.  Unresolvable emails are emitted as-is with a warning comment.
-
-        ``default_owner``, when set, emits a leading ``* <default_owner>`` rule so
-        unmatched files (and any excluded directories) fall through to it. Because
-        CODEOWNERS is last-match-wins, the per-file rules below it still take
-        precedence for the files they name.
-
-        ``exclude_dirs`` lists directories (relative to base_dir) whose nodes are
-        NOT given per-file owners — they fall through to ``default_owner`` instead.
-        This is for machine-generated trees (e.g. ``nodes/generated``) that have no
-        individual human owner and should be team-owned as a block.
-
-        Returns the number of per-file entries written (excludes the default rule).
-        """
+        """Generate a CODEOWNERS file from the owners fields in DJ node YAML files."""
         base = Path(base_dir).resolve()
         excluded_dirs = [(base / d).resolve() for d in (exclude_dirs or [])]
 

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]:
         """