feat: [SNOW-2326969] implement dcm refresh

Author: sfc-gh-jwilkowski

src/snowflake/cli/_plugins/dcm/commands.py

@@ -16,7 +16,10 @@
 
 import typer
 from snowflake.cli._plugins.dcm.manager import DCMProjectManager
-from snowflake.cli._plugins.dcm.utils import format_test_failures
+from snowflake.cli._plugins.dcm.utils import (
+    format_refresh_results,
+    format_test_failures,
+)
 from snowflake.cli._plugins.object.command_aliases import add_object_command_aliases
 from snowflake.cli._plugins.object.commands import scope_option
 from snowflake.cli._plugins.object.manager import ObjectManager
@@ -278,6 +281,33 @@ def test(
     return MessageResult(f"All {len(expectations)} expectation(s) passed successfully.")
 
 
+@app.command(requires_connection=True)
+def refresh(
+    identifier: FQN = dcm_identifier,
+    **options,
+):
+    """
+    Refreshes dynamic tables defined in DCM project.
+    """
+    with cli_console.spinner() as spinner:
+        spinner.add_task(description=f"Refreshing dcm project {identifier}", total=None)
+        result = DCMProjectManager().refresh(project_identifier=identifier)
+
+    row = result.fetchone()
+    if not row:
+        return MessageResult("No data.")
+
+    result_data = row[0]
+    result_json = (
+        json.loads(result_data) if isinstance(result_data, str) else result_data
+    )
+
+    refreshed_tables = result_json.get("refreshed_tables", [])
+    message = format_refresh_results(refreshed_tables)
+
+    return MessageResult(message)
+
+
 def _get_effective_stage(identifier: FQN, from_location: Optional[str]):
     manager = DCMProjectManager()
     if not from_location:

src/snowflake/cli/_plugins/dcm/manager.py

@@ -140,6 +140,10 @@ def test(self, project_identifier: FQN):
         query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} TEST ALL"
         return self.execute_query(query=query)
 
+    def refresh(self, project_identifier: FQN):
+        query = f"EXECUTE DCM PROJECT {project_identifier.sql_identifier} REFRESH ALL"
+        return self.execute_query(query=query)
+
     @staticmethod
     def sync_local_files(
         project_identifier: FQN, source_directory: str | None = None

src/snowflake/cli/_plugins/dcm/utils.py

@@ -41,3 +41,17 @@ def format_test_failures(
     )
 
     return "\n".join(lines)
+
+
+def format_refresh_results(refreshed_tables: list) -> str:
+    """Format refresh results into a concise user-friendly message."""
+    if not refreshed_tables:
+        return "No dynamic tables found in the project."
+
+    total_tables = len(refreshed_tables)
+    refreshed_count = sum(
+        1 for table in refreshed_tables if table.get("refreshed_dt_count", 0) > 0
+    )
+    up_to_date_count = total_tables - refreshed_count
+
+    return f"{refreshed_count} dynamic table(s) refreshed. {up_to_date_count} dynamic table(s) up-to-date."

src/snowflake/cli/api/output/types.py

@@ -146,9 +146,13 @@ def __init__(self, cursor: SnowflakeCursor):
     def _prepare_payload(self, cursor):
         results = list(QueryResult(cursor).result)
         if results:
-            # Return value of the first tuple
-            return json.loads(list(results[0].items())[0][1])
-        return None
+            # Parse the JSON value from the first tuple
+            parsed_value = json.loads(list(results[0].items())[0][1])
+            # If it's a list, yield each element; if it's a dict, yield the single dict
+            if isinstance(parsed_value, list):
+                yield from parsed_value
+            else:
+                yield parsed_value
 
 
 class MessageResult(CommandResult):

tests/__snapshots__/test_help_messages.ambr

@@ -0,0 +1,22 @@
+# serializer version: 1
+# name: TestDCMRefresh.test_refresh_with_fresh_tables
+  '''
+  
+  0 dynamic table(s) refreshed. 1 dynamic table(s) up-to-date.
+  
+  '''
+# ---
+# name: TestDCMRefresh.test_refresh_with_no_dynamic_tables
+  '''
+  
+  No dynamic tables found in the project.
+  
+  '''
+# ---
+# name: TestDCMRefresh.test_refresh_with_outdated_tables
+  '''
+  
+  1 dynamic table(s) refreshed. 0 dynamic table(s) up-to-date.
+  
+  '''
+# ---

tests/dcm/__snapshots__/test_commands.ambr

@@ -799,3 +799,70 @@ def test_test_with_multiple_failed_expectations(self, mock_pm, runner, mock_curs
         mock_pm().test.assert_called_once_with(
             project_identifier=FQN.from_string("my_project")
         )
+
+
+class TestDCMRefresh:
+    @mock.patch(DCMProjectManager)
+    def test_refresh_with_outdated_tables(self, mock_pm, runner, mock_cursor, snapshot):
+        refresh_result = {
+            "refreshed_tables": [
+                {
+                    "dt_name": "JW_DCM_TESTALL.ANALYTICS.DYNAMIC_EMPLOYEES",
+                    "refreshed_dt_count": 1,
+                    "data_timestamp": "1760357032.175",
+                    "statistics": '{"insertedRows":5,"copiedRows":0,"deletedRows":5}',
+                }
+            ]
+        }
+        mock_pm().refresh.return_value = mock_cursor(
+            rows=[(json.dumps(refresh_result),)], columns=("result",)
+        )
+
+        result = runner.invoke(["dcm", "refresh", "my_project"])
+
+        assert result.exit_code == 0, result.output
+        assert result.output == snapshot
+        mock_pm().refresh.assert_called_once_with(
+            project_identifier=FQN.from_string("my_project")
+        )
+
+    @mock.patch(DCMProjectManager)
+    def test_refresh_with_fresh_tables(self, mock_pm, runner, mock_cursor, snapshot):
+        refresh_result = {
+            "refreshed_tables": [
+                {
+                    "dt_name": "JW_DCM_TESTALL.ANALYTICS.DYNAMIC_EMPLOYEES",
+                    "refreshed_dt_count": 0,
+                    "data_timestamp": "1760356974.543",
+                    "statistics": "No new data",
+                }
+            ]
+        }
+        mock_pm().refresh.return_value = mock_cursor(
+            rows=[(json.dumps(refresh_result),)], columns=("result",)
+        )
+
+        result = runner.invoke(["dcm", "refresh", "my_project"])
+
+        assert result.exit_code == 0, result.output
+        assert result.output == snapshot
+        mock_pm().refresh.assert_called_once_with(
+            project_identifier=FQN.from_string("my_project")
+        )
+
+    @mock.patch(DCMProjectManager)
+    def test_refresh_with_no_dynamic_tables(
+        self, mock_pm, runner, mock_cursor, snapshot
+    ):
+        refresh_result = {"refreshed_tables": []}
+        mock_pm().refresh.return_value = mock_cursor(
+            rows=[(json.dumps(refresh_result),)], columns=("result",)
+        )
+
+        result = runner.invoke(["dcm", "refresh", "my_project"])
+
+        assert result.exit_code == 0, result.output
+        assert result.output == snapshot
+        mock_pm().refresh.assert_called_once_with(
+            project_identifier=FQN.from_string("my_project")
+        )

tests/dcm/test_commands.py

@@ -169,6 +169,16 @@ def test_test_project(mock_execute_query):
     )
 
 
+@mock.patch(execute_queries)
+def test_refresh_project(mock_execute_query):
+    mgr = DCMProjectManager()
+    mgr.refresh(project_identifier=TEST_PROJECT)
+
+    mock_execute_query.assert_called_once_with(
+        query="EXECUTE DCM PROJECT IDENTIFIER('my_project') REFRESH ALL"
+    )
+
+
 @mock.patch(execute_queries)
 def test_plan_project_with_output_path__stage(mock_execute_query, project_directory):
     mgr = DCMProjectManager()

tests/dcm/test_manager.py

@@ -642,3 +642,94 @@ def test_dcm_test_command(
         result = runner.invoke_with_connection(["dcm", "test", project_name])
         assert result.exit_code == 0, result.output
         assert "expectation(s) passed successfully" in result.output
+
+
+@pytest.mark.qa_only
+@pytest.mark.integration
+def test_dcm_refresh_command(
+    runner,
+    test_database,
+    project_directory,
+    object_name_provider,
+    sql_test_helper,
+):
+    project_name = object_name_provider.create_and_get_next_object_name()
+    base_table_name = f"{test_database}.PUBLIC.RefreshBaseTable"
+    dynamic_table_name = f"{test_database}.PUBLIC.RefreshDynamicTable"
+
+    with project_directory("dcm_project") as project_root:
+        result = runner.invoke_with_connection(["dcm", "create", project_name])
+        assert result.exit_code == 0, result.output
+
+        # Deploy the project.
+        result = runner.invoke_with_connection_json(
+            [
+                "dcm",
+                "deploy",
+                project_name,
+                "-D",
+                f"table_name='{test_database}.PUBLIC.OutputTestTable'",
+            ]
+        )
+        assert result.exit_code == 0, result.output
+
+        # 1) Without any dynamic tables, run refresh command - should report no dynamic tables.
+        result = runner.invoke_with_connection(["dcm", "refresh", project_name])
+        assert result.exit_code == 0, result.output
+        assert "No dynamic tables found in the project." in result.output
+
+        # 2) Define base table and dynamic table with long refresh time.
+        table_definitions = f"""
+define table identifier('{base_table_name}') (
+  id int, name varchar, email varchar
+);
+
+define dynamic table identifier('{dynamic_table_name}')
+target_lag = '1000 minutes'
+WAREHOUSE = xs
+as select * from {base_table_name};
+"""
+        file_a_path = project_root / "file_a.sql"
+        original_content = file_a_path.read_text()
+        file_a_path.write_text(original_content + table_definitions)
+
+        # Deploy the project.
+        result = runner.invoke_with_connection_json(
+            [
+                "dcm",
+                "deploy",
+                project_name,
+                "-D",
+                f"table_name='{test_database}.PUBLIC.OutputTestTable'",
+            ]
+        )
+        assert result.exit_code == 0, result.output
+
+        # 3) Insert data into the base table.
+        insert_data_sql = f"""
+INSERT INTO {base_table_name} (id, name, email) VALUES
+    (1, 'Alice Johnson', '[email protected]'),
+    (2, 'Bob Williams', '[email protected]'),
+    (3, 'Charlie Brown', '[email protected]');
+"""
+        sql_test_helper.execute_single_sql(insert_data_sql)
+
+        # 4) Verify that data is NOT yet in the dynamic table (due to long refresh time).
+        check_dt_sql = f"SELECT COUNT(*) as cnt FROM {dynamic_table_name}"
+        result = sql_test_helper.execute_single_sql(check_dt_sql)
+        count_before = result[0]["CNT"]
+        assert count_before == 0, "Dynamic table should be empty before refresh."
+
+        # 5) Run dcm refresh command.
+        result = runner.invoke_with_connection(["dcm", "refresh", project_name])
+        assert result.exit_code == 0, result.output
+        # Should show at least 1 table was refreshed
+        assert (
+            "1 dynamic table(s) refreshed" in result.output
+            or "dynamic table(s) refreshed" in result.output
+        )
+
+        # 6) Verify that data is NOW in the dynamic table.
+        result = sql_test_helper.execute_single_sql(check_dt_sql)
+        count_after = result[0]["CNT"]
+        assert count_after == 3, "Dynamic table should have 3 rows after refresh."