Add a refresh_cube_materialization end point and adjust the UI messaging.

Author: agorajek

datajunction-query/pyproject.toml

@@ -26,7 +26,7 @@ dependencies = [
     "snowflake-connector-python>=3.3.1",
     "pyyaml>=6.0.1",
     "trino>=0.324.0",
-    "psycopg[async,pool]>=3.2.1",
+    "psycopg[pool]>=3.2.1",
     "sqlalchemy>=2.0.34",
     "pytest-asyncio>=0.24.0",
     "pytest-integration>=0.2.3",
@@ -64,7 +64,7 @@ test = [
     "pydruid>=0.6.4",
     "typing-extensions>=4.3.0",
     "httpx>=0.24.1",
-    "psycopg[async,pool]>=3.2.1",
+    "psycopg[pool]>=3.2.1",
     "testcontainers[postgres]>=4.8.1",
     "types-toml",
 ]

datajunction-server/datajunction_server/internal/nodes.py

@@ -1279,7 +1279,7 @@ async def update_cube_node(
     """
     node = await Node.get_cube_by_name(session, node_revision.name)
     node_revision = node.current  # type: ignore
-    minor_changes = has_minor_changes(node_revision, data) or refresh_materialization
+    minor_changes = has_minor_changes(node_revision, data)
     old_metrics = [m.name for m in node_revision.cube_metrics()]
     old_dimensions = node_revision.cube_dimensions()
     major_changes = (data.metrics and data.metrics != old_metrics) or (
@@ -1297,6 +1297,52 @@ async def update_cube_node(
         limit=data.limit or None,
     )
     if not major_changes and not minor_changes:
+        # If refresh_materialization requested but no other changes, refresh and return
+        if refresh_materialization and query_service_client:
+            _logger.info(
+                "Refreshing materializations for cube=%s without version change",
+                node_revision.name,
+            )
+            # Re-activate deactivated materializations that match the current version
+            current_version = node_revision.version
+            for mat in node_revision.materializations:
+                mat_config = mat.config if isinstance(mat.config, dict) else {}
+                mat_version = mat_config.get("cube", {}).get("version", "")
+                if mat.deactivated_at and mat_version == current_version:
+                    _logger.info(
+                        "Re-activating materialization %s for cube=%s version=%s",
+                        mat.name,
+                        node_revision.name,
+                        current_version,
+                    )
+                    mat.deactivated_at = None
+            await session.commit()
+
+            # Serialize active materializations as complete, ready-to-use dicts
+            # so dj-query can parse them directly into model inputs without merging.
+            # This avoids a callback from query service back to DJ.
+            active_mats = []
+            for mat in node_revision.materializations:
+                if mat.deactivated_at:
+                    continue
+                mat_dict = {
+                    **(mat.config if isinstance(mat.config, dict) else {}),
+                    "name": mat.name,
+                    "job": mat.job,
+                    "strategy": mat.strategy.value if mat.strategy else None,
+                    "schedule": mat.schedule,
+                    "cube": {
+                        "name": node_revision.name,
+                        "version": node_revision.version,
+                    },
+                }
+                active_mats.append(mat_dict)
+            query_service_client.refresh_cube_materialization(
+                cube_name=node_revision.name,
+                cube_version=node_revision.version,
+                materializations=active_mats,
+                request_headers=request_headers,
+            )
         return None
 
     # Disable autoflush to prevent partial state from being persisted if an error

datajunction-server/datajunction_server/query_clients/http.py

@@ -206,6 +206,21 @@ def deactivate_materialization(
             request_headers=request_headers,
         )
 
+    def refresh_cube_materialization(
+        self,
+        cube_name: str,
+        cube_version: Optional[str] = None,
+        materializations: Optional[List[Dict[str, Any]]] = None,
+        request_headers: Optional[Dict[str, str]] = None,
+    ) -> MaterializationInfo:
+        """Refresh/rebuild materialization workflows for a cube via HTTP query service."""
+        return self._client.refresh_cube_materialization(
+            cube_name=cube_name,
+            cube_version=cube_version,
+            materializations=materializations,
+            request_headers=request_headers,
+        )
+
     def get_materialization_info(
         self,
         node_name: str,

datajunction-server/datajunction_server/service_clients.py

@@ -541,6 +541,74 @@ def deactivate_materialization(
         )
         return MaterializationInfo(**result)
 
+    def refresh_cube_materialization(
+        self,
+        cube_name: str,
+        cube_version: Optional[str] = None,
+        materializations: Optional[List[Dict]] = None,
+        request_headers: Optional[Dict[str, str]] = None,
+    ) -> MaterializationInfo:
+        """
+        Refresh/rebuild materialization workflows for a cube without creating a new version.
+
+        This calls the query service to recreate the cube's materialization workflows
+        with the same names, effectively overwriting them in the scheduler.
+
+        Args:
+            cube_name: Name of the cube node
+            cube_version: Optional cube version (defaults to latest if not specified)
+            materializations: List of active materialization dicts to rebuild
+            request_headers: Optional HTTP headers
+
+        Returns:
+            MaterializationInfo with URLs of recreated workflows
+        """
+        refresh_endpoint = f"/cubes/{cube_name}/refresh-materialization"
+        params = {}
+        if cube_version:
+            params["cube_version"] = cube_version
+
+        _logger.info(
+            "[DJQS] Refreshing materializations for cube=%s version=%s",
+            cube_name,
+            cube_version or "latest",
+        )
+
+        # Merge request headers with session headers
+        headers = dict(self.requests_session.headers)
+        if request_headers:
+            headers.update(request_headers)
+
+        # Build request body with materialization data
+        body = {}
+        if materializations is not None:
+            body["materializations"] = materializations
+
+        response = self.requests_session.post(
+            refresh_endpoint,
+            headers=headers,
+            params=params,
+            json=body,
+        )
+
+        if response.status_code not in (200, 201):  # pragma: no cover
+            _logger.exception(
+                "[DJQS] Failed to refresh materializations for cube=%s with `POST %s`: %s",
+                cube_name,
+                refresh_endpoint,
+                response.text,
+                exc_info=True,
+            )
+            return MaterializationInfo(urls=[], output_tables=[])
+
+        result = response.json()
+        _logger.info(
+            "[DJQS] Successfully refreshed materializations for cube=%s with `POST %s`",
+            cube_name,
+            refresh_endpoint,
+        )
+        return MaterializationInfo(**result)
+
     def get_materialization_info(
         self,
         node_name: str,

datajunction-server/tests/api/cubes_test.py

@@ -17,7 +17,7 @@
 from datajunction_server.models.query import ColumnMetadata, V3ColumnMetadata
 from datajunction_server.service_clients import QueryServiceClient
 from datajunction_server.sql.parsing.backends.antlr4 import parse
-from datajunction_server.utils import get_query_service_client
+from datajunction_server.utils import get_query_service_client, get_session
 from tests.sql.utils import compare_query_strings
 from tests.construction.build_v3 import assert_sql_equal
 
@@ -4283,3 +4283,93 @@ async def test_backfill_cube_query_service_failure(
         )
         assert response.status_code == 500
         assert "failed to run cube backfill" in response.json()["message"].lower()
+
+
+class TestCubeRefreshMaterialization:
+    """Tests for refreshing cube materializations without version bump."""
+
+    @pytest.mark.asyncio
+    async def test_refresh_materialization_no_version_bump(
+        self,
+        client_with_repairs_cube: AsyncClient,
+        mocker,
+    ):
+        """
+        Test that PATCH with refresh_materialization=true and no other changes
+        calls refresh_cube_materialization without bumping the version.
+        """
+        cube_name = "default.test_refresh_cube"
+        await make_a_test_cube(
+            client_with_repairs_cube,
+            cube_name,
+            with_materialization=True,
+        )
+
+        # Get the initial version
+        response = await client_with_repairs_cube.get(f"/nodes/{cube_name}/")
+        assert response.status_code == 200
+        initial_version = response.json()["version"]
+
+        # Add a deactivated materialization to the node revision so the
+        # re-activation logic gets exercised
+        from datetime import datetime, timezone
+
+        from sqlalchemy import select
+
+        from datajunction_server.database.materialization import Materialization
+        from datajunction_server.database.node import NodeRevision
+
+        session_factory = client_with_repairs_cube.app.dependency_overrides[get_session]
+        session = session_factory()
+        result = await session.execute(
+            select(NodeRevision).where(NodeRevision.name == cube_name),
+        )
+        node_rev = result.scalars().first()
+        deactivated_mat = Materialization(
+            node_revision_id=node_rev.id,
+            name="deactivated_mat",
+            strategy=None,
+            schedule="",
+            config={"cube": {"version": initial_version}},
+            job="DruidCubeMaterializationJob",
+            deactivated_at=datetime.now(timezone.utc),
+        )
+        session.add(deactivated_mat)
+        await session.commit()
+
+        # Mock the refresh method on the query service client
+        qs_client = client_with_repairs_cube.app.dependency_overrides[
+            get_query_service_client
+        ]()
+        mock_refresh = mocker.patch.object(
+            qs_client,
+            "refresh_cube_materialization",
+            return_value=mocker.MagicMock(urls=["http://workflow/refreshed"]),
+        )
+
+        # PATCH with refresh_materialization=true but no actual changes
+        response = await client_with_repairs_cube.patch(
+            f"/nodes/{cube_name}/?refresh_materialization=true",
+            json={},
+        )
+        # Should return 200 with no version change (returns None -> no content)
+        assert response.status_code == 200
+
+        # Verify refresh was called with correct args
+        mock_refresh.assert_called_once()
+        call_kwargs = mock_refresh.call_args[1]
+        assert call_kwargs["cube_name"] == cube_name
+        assert call_kwargs["cube_version"] == initial_version
+        assert isinstance(call_kwargs["materializations"], list)
+        assert len(call_kwargs["materializations"]) > 0
+        # Deactivated materialization should be re-activated and included
+        mat_names = [m["name"] for m in call_kwargs["materializations"]]
+        assert "deactivated_mat" in mat_names
+
+        # Verify the materialization was re-activated in the DB
+        await session.refresh(deactivated_mat)
+        assert deactivated_mat.deactivated_at is None
+
+        # Verify version didn't change
+        response = await client_with_repairs_cube.get(f"/nodes/{cube_name}/")
+        assert response.json()["version"] == initial_version

datajunction-server/tests/query_clients/http_query_client_test.py

@@ -102,3 +102,27 @@ def test_run_cube_backfill(self, mock_client):
             backfill_input=backfill_input,
             request_headers={"X-Test": "1"},
         )
+
+    def test_refresh_cube_materialization(self, mock_client):
+        """Test refresh_cube_materialization delegates to underlying client."""
+        client, mock_inner = mock_client
+        mock_inner.refresh_cube_materialization.return_value = MagicMock(
+            urls=["http://wf1"],
+        )
+
+        materializations = [
+            {"name": "mat1", "job": "DruidCubeMaterializationJob"},
+        ]
+        client.refresh_cube_materialization(
+            cube_name="test.cube",
+            cube_version="v1.0",
+            materializations=materializations,
+            request_headers={"X-Test": "1"},
+        )
+
+        mock_inner.refresh_cube_materialization.assert_called_once_with(
+            cube_name="test.cube",
+            cube_version="v1.0",
+            materializations=materializations,
+            request_headers={"X-Test": "1"},
+        )

datajunction-server/tests/service_clients_test.py

@@ -1181,3 +1181,82 @@ def test_run_cube_backfill_failure(self, mocker: MockerFixture) -> None:
         with pytest.raises(Exception) as exc_info:
             query_service_client.run_cube_backfill(backfill_input)
         assert "Query service error" in str(exc_info.value)
+
+    def test_refresh_cube_materialization_success(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """
+        Test successful refresh of cube materializations.
+        """
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "urls": ["http://workflow1"],
+            "output_tables": [],
+        }
+
+        mock_request = mocker.patch(
+            "datajunction_server.service_clients.RequestsSessionWithEndpoint.post",
+            return_value=mock_response,
+        )
+
+        materializations = [
+            {
+                "name": "druid_cube__incremental_time__date",
+                "job": "DruidCubeMaterializationJob",
+                "strategy": "incremental_time",
+                "schedule": "@daily",
+                "cube": {"name": "test.cube", "version": "v1.0"},
+            },
+        ]
+
+        query_service_client = QueryServiceClient(uri=self.endpoint)
+        result = query_service_client.refresh_cube_materialization(
+            cube_name="test.cube",
+            cube_version="v1.0",
+            materializations=materializations,
+            request_headers={"Cookie": "session=abc123"},
+        )
+
+        mock_request.assert_called_once_with(
+            "/cubes/test.cube/refresh-materialization",
+            headers=ANY,
+            params={"cube_version": "v1.0"},
+            json={"materializations": materializations},
+        )
+        assert result == MaterializationInfo(
+            urls=["http://workflow1"],
+            output_tables=[],
+        )
+
+    def test_refresh_cube_materialization_no_version(
+        self,
+        mocker: MockerFixture,
+    ) -> None:
+        """
+        Test refresh without specifying a version sends no params.
+        """
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "urls": [],
+            "output_tables": [],
+        }
+
+        mock_request = mocker.patch(
+            "datajunction_server.service_clients.RequestsSessionWithEndpoint.post",
+            return_value=mock_response,
+        )
+
+        query_service_client = QueryServiceClient(uri=self.endpoint)
+        query_service_client.refresh_cube_materialization(
+            cube_name="test.cube",
+        )
+
+        mock_request.assert_called_once_with(
+            "/cubes/test.cube/refresh-materialization",
+            headers=ANY,
+            params={},
+            json={},
+        )