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(