feat: semi-additive measures

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: ck0001results
+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 = "ck0001results"
+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/api/cubes.py

@@ -14,6 +14,10 @@
     _reorder_partition_column_last,
     build_combiner_sql_from_preaggs,
 )
+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
@@ -201,6 +205,38 @@ def _build_metrics_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
+
+    if not await _metric_graph_has_reaggregate(
+        session,
+        cube.current.cube_node_metrics,
+    ):
+        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 +324,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 +557,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/dataloaders.py

@@ -377,6 +377,7 @@ async def batch_load_extracted_measures(
                             DBNodeRevision.id,
                             DBNodeRevision.name,
                             DBNodeRevision.query,
+                            DBNodeRevision.reaggregate,
                         ),
                     ),
                 )

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

@@ -15,9 +15,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,6 +38,17 @@ class Unit:
     abbreviation: str | None
 
 
+@strawberry.experimental.pydantic.type(
+    model=DimensionReaggregateRule_,
+    all_fields=True,
+)
+class DimensionReaggregateRule: ...
+
+
+@strawberry.experimental.pydantic.type(model=ReaggregateSpec_, all_fields=True)
+class ReaggregateSpec: ...
+
+
 @strawberry.experimental.pydantic.type(model=AggregationRule_, all_fields=True)
 class AggregationRule: ...
 

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,26 @@ 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(
+            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!
@@ -397,6 +403,7 @@ type NodeRevision {
   dimensionLinks: [DimensionLink!]!
   availability: AvailabilityState
   materializations: [MaterializationConfig!]
+  reaggregate: ReaggregateSpec
   primaryKey: [String!]!
   metricMetadata: MetricMetadata
   isDerivedMetric: Boolean!
@@ -716,6 +723,21 @@ type Query {
   listNamespaces: [Namespace!]!
 }
 
+type ReaggregateSpec {
+  rules: [DimensionReaggregateRule!]!
+}
+
+enum ReaggregationFunction {
+  AUTO
+  NONE
+  SUM
+  AVG
+  LAST_VALUE
+  FIRST_VALUE
+  MIN
+  MAX
+}
+
 type SemanticEntity {
   name: String!
 

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

@@ -21,6 +21,7 @@
 )
 from datajunction_server.construction.build_v3.decomposition import (
     decompose_and_group_metrics,
+    missing_reaggregate_dimensions,
 )
 from datajunction_server.construction.build_v3.dimensions import parse_dimension_ref
 from datajunction_server.construction.build_v3.filters import (
@@ -318,6 +319,14 @@ async def setup_build_context(
 
     # Add dimensions referenced in metric expressions (e.g., LAG ORDER BY)
     add_dimensions_from_metric_expressions(ctx, ctx.decomposed_metrics)
+    output_dimensions_after_expression_scan = list(ctx.dimensions)
+    internal_reaggregate_dimensions = missing_reaggregate_dimensions(
+        ctx.decomposed_metrics.values(),
+        output_dimensions_after_expression_scan,
+    )
+    for dimension in internal_reaggregate_dimensions:
+        if dimension not in ctx.dimensions:
+            ctx.dimensions.append(dimension)
 
     # A second load_nodes pass is needed when either:
     # 1. metric expressions introduced dimension nodes not yet in ctx.nodes, OR
@@ -331,8 +340,15 @@ async def setup_build_context(
     }
     missing_dim_nodes = dim_roots_after - ctx.nodes.keys()
     internally_added_roots = dim_roots_after - dim_roots_before_load
-    if missing_dim_nodes or internally_added_roots:
-        await load_nodes(ctx)
+    try:
+        if (
+            missing_dim_nodes
+            or internally_added_roots
+            or internal_reaggregate_dimensions
+        ):
+            await load_nodes(ctx)
+    finally:
+        ctx.dimensions = output_dimensions_after_expression_scan
 
     # Classify filters into dimension filters (WHERE) and metric filters (HAVING)
     # This MUST happen AFTER all nodes are loaded so we can correctly identify