From 8fbc68f30c8845c4b6847a032b28655b4b901291 Mon Sep 17 00:00:00 2001
From: Shahar Epstein <60007259+shahar1@users.noreply.github.com>
Date: Fri, 24 Jul 2026 17:58:15 +0300
Subject: [PATCH] Revert Dataform workflow invocation dict normalization
 (#69161)

The reverted change was meant to fix #53843 (invocation_config believed
dropped when workflow_invocation is passed as a dict), but it is a
wire-identical no-op: proto-plus already converts nested dicts
recursively, and the pre- and post-change requests serialize
byte-identically, so the dict path never dropped the config. The
reported symptom came from dataform-core 3.0.10-3.0.16 defaulting
incremental tables to protected=true (silently disabling full refresh
server-side; fixed upstream in core 3.0.17 via
dataform-co/dataform#1942). #53843 is closed as not planned. The
accompanying test only pinned third-party proto-plus conversion
behavior, which our testing standards exclude.
---
 .../providers/google/cloud/hooks/dataform.py  |  2 --
 .../unit/google/cloud/hooks/test_dataform.py  | 23 +------------------
 2 files changed, 1 insertion(+), 24 deletions(-)

diff --git a/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py b/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
index 9167689daa842..48162f1fd8cb9 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/dataform.py
@@ -187,8 +187,6 @@ def create_workflow_invocation(
         """
         client = self.get_dataform_client()
         parent = f"projects/{project_id}/locations/{region}/repositories/{repository_id}"
-        if isinstance(workflow_invocation, dict):
-            workflow_invocation = WorkflowInvocation(workflow_invocation)
         return client.create_workflow_invocation(
             request={"parent": parent, "workflow_invocation": workflow_invocation},
             retry=retry,
diff --git a/providers/google/tests/unit/google/cloud/hooks/test_dataform.py b/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
index c70492a7784e7..2dda5ffe431ae 100644
--- a/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
+++ b/providers/google/tests/unit/google/cloud/hooks/test_dataform.py
@@ -50,12 +50,6 @@
         f"{REPOSITORY_ID}/compilationResults/{COMPILATION_RESULT_ID}"
     ),
 }
-FULL_REFRESH_WORKFLOW_INVOCATION = {
-    **WORKFLOW_INVOCATION,
-    "invocation_config": {
-        "fully_refresh_incremental_tables_enabled": True,
-    },
-}
 WORKFLOW_INVOCATION_ID = "test_workflow_invocation_id"
 PATH_TO_FOLDER = "path/to/folder"
 FILEPATH = "path/to/file.txt"
@@ -119,27 +113,12 @@ def test_create_workflow_invocation(self, mock_client):
         )
         parent = f"projects/{PROJECT_ID}/locations/{REGION}/repositories/{REPOSITORY_ID}"
         mock_client.return_value.create_workflow_invocation.assert_called_once_with(
-            request=dict(parent=parent, workflow_invocation=WorkflowInvocation(WORKFLOW_INVOCATION)),
+            request=dict(parent=parent, workflow_invocation=WORKFLOW_INVOCATION),
             retry=DEFAULT,
             timeout=None,
             metadata=(),
         )
 
-    @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
-    def test_create_workflow_invocation_preserves_invocation_config_from_dict(self, mock_client):
-        self.hook.create_workflow_invocation(
-            project_id=PROJECT_ID,
-            region=REGION,
-            repository_id=REPOSITORY_ID,
-            workflow_invocation=FULL_REFRESH_WORKFLOW_INVOCATION,
-        )
-
-        request = mock_client.return_value.create_workflow_invocation.call_args.kwargs["request"]
-        workflow_invocation = request["workflow_invocation"]
-        assert isinstance(workflow_invocation, WorkflowInvocation)
-        assert workflow_invocation.compilation_result == WORKFLOW_INVOCATION["compilation_result"]
-        assert workflow_invocation.invocation_config.fully_refresh_incremental_tables_enabled is True
-
     @mock.patch(DATAFORM_STRING.format("DataformHook.get_dataform_client"))
     def test_get_workflow_invocation(self, mock_client):
         self.hook.get_workflow_invocation(