diff --git a/src/integrations/prefect-snowflake/prefect_snowflake/database.py b/src/integrations/prefect-snowflake/prefect_snowflake/database.py
index 0ca83b5733b6..9e82afcdf9ad 100644
--- a/src/integrations/prefect-snowflake/prefect_snowflake/database.py
+++ b/src/integrations/prefect-snowflake/prefect_snowflake/database.py
@@ -548,9 +548,7 @@ def fetch_all(
         )
         new, cursor = self._get_cursor(inputs, cursor_type)
         if new:
-            self.execute(
-                operation, parameters, cursor_type=cursor_type, **execute_kwargs
-            )
+            cursor.execute(operation, params=parameters, **execute_kwargs)
         self.logger.debug("Preparing to fetch all rows.")
         return cursor.fetchall()
 
diff --git a/src/integrations/prefect-snowflake/tests/test_database.py b/src/integrations/prefect-snowflake/tests/test_database.py
index 1a5497e38284..bb23673140d7 100644
--- a/src/integrations/prefect-snowflake/tests/test_database.py
+++ b/src/integrations/prefect-snowflake/tests/test_database.py
@@ -280,6 +280,21 @@ def test_fetch_all_cursor_default(self, snowflake_connector: SnowflakeConnector)
 
         assert args[0] == OriginalSnowflakeCursorClass
 
+    def test_fetch_all_executes_directly_on_cursor(
+        self, snowflake_connector: SnowflakeConnector
+    ):
+        """Test that fetch_all executes directly on the cursor instead of using self.execute."""
+        snowflake_connector._start_connection()
+        mock_cursor = MagicMock()
+        snowflake_connector._connection.cursor.return_value = mock_cursor
+        mock_cursor.fetchall.return_value = [(1,)]
+
+        result = snowflake_connector.fetch_all("SELECT 1")
+
+        mock_cursor.execute.assert_called_once_with("SELECT 1", params=None)
+        mock_cursor.fetchall.assert_called_once()
+        assert result == [(1,)]
+
     @pytest.mark.parametrize("fetch_function_name", ["fetch_many", "fetch_many_async"])
     async def test_fetch_many(self, fetch_function_name, snowflake_connector):
         fetch_function = getattr(snowflake_connector, fetch_function_name)