diff --git a/airflow/providers/snowflake/CHANGELOG.rst b/airflow/providers/snowflake/CHANGELOG.rst
index f11a45e48a158..e237e822da787 100644
--- a/airflow/providers/snowflake/CHANGELOG.rst
+++ b/airflow/providers/snowflake/CHANGELOG.rst
@@ -27,6 +27,11 @@ Changelog
 4.0.1
 .....
 
+.. warning::
+
+    This version is yanked, as it contained problems when interacting with common.sql provider. Please install
+    a version released afterwards.
+
 Bug Fixes
 ~~~~~~~~~
 
@@ -41,17 +46,45 @@ Bug Fixes
 4.0.0
 .....
 
+.. warning::
+
+    This version is yanked, as it contained problems when interacting with common.sql provider. Please install
+    a version released afterwards.
+
 This release of provider is only available for Airflow 2.3+ as explained in the
 `Apache Airflow providers support policy <https://github.com/apache/airflow/blob/main/README.md#support-for-providers>`_.
 
 Breaking changes
 ~~~~~~~~~~~~~~~~
 
+The ``SnowflakeHook`` is now conforming to the same semantics as all the other ``DBApiHook``
+implementations and returns the same kind of response in its ``run`` method. Previously (pre 4.* versions
+of the provider, the Hook returned Dictionary of ``{ "column": "value" ... }`` which was not compatible
+with other DBApiHooks that return just sequence of sequences. After this change (and dependency
+on common.sql >= 1.3.1),the ``SnowflakeHook`` returns now python DbApi-compatible "results" by default.
+
+The ``description`` (i.e. among others names and types of columns returned) can be retrieved
+via ``descriptions`` and ``last_description`` fields of the hook after ``run`` method completes.
+
+That makes the ``DatabricksSqlHook`` suitable for generic SQL operator and detailed lineage analysis.
+
+If you had custom hooks or used the Hook in your TaskFlow code or custom operators that relied on this
+behaviour, you need to adapt your DAGs or you can switch back the ``SnowflakeHook`` to return dictionaries
+by passing ``return_dictionaries=True`` to the run method of the hook.
+
+The ``SnowflakeOperator`` is also more standard and derives from common
+``SQLExecuteQueryOperator`` and uses more consistent approach to process output when SQL queries are run.
+However in this case the result returned by ``execute`` method is unchanged (it still returns Dictionaries
+rather than sequences and those dictionaries are pushed to XCom, so your DAGs relying on this behaviour
+should continue working without any change.
+
 In SnowflakeHook, if both ``extra__snowflake__foo`` and ``foo`` existed in connection extra
 dict, the prefixed version would be used; now, the non-prefixed version will be preferred.
 
+
 * ``Update snowflake hook to not use extra prefix (#26764)``
 
+
 Misc
 ~~~~
 
diff --git a/airflow/providers/snowflake/hooks/snowflake.py b/airflow/providers/snowflake/hooks/snowflake.py
index 79bb6893180a9..a4f6190fec5e3 100644
--- a/airflow/providers/snowflake/hooks/snowflake.py
+++ b/airflow/providers/snowflake/hooks/snowflake.py
@@ -18,7 +18,7 @@
 from __future__ import annotations
 
 import os
-from contextlib import closing
+from contextlib import closing, contextmanager
 from functools import wraps
 from io import StringIO
 from pathlib import Path
@@ -329,6 +329,7 @@ def run(
         handler: Callable | None = None,
         split_statements: bool = True,
         return_last: bool = True,
+        return_dictionaries: bool = False,
     ) -> Any | list[Any] | None:
         """
         Runs a command or a list of commands. Pass a list of sql
@@ -346,6 +347,9 @@ def run(
         :param handler: The result handler which is called with the result of each statement.
         :param split_statements: Whether to split a single SQL string into statements and run separately
         :param return_last: Whether to return result for only last statement or for all after split
+        :param return_dictionaries: Whether to return dictionaries rather than regular DBApi sequences
+            as rows in the result. The dictionaries are of form:
+            ``{ 'column1_name': value1, 'column2_name': value2 ... }``.
         :return: return only result of the LAST SQL expression if handler was provided.
         """
         self.query_ids = []
@@ -369,15 +373,19 @@ def run(
         with closing(self.get_conn()) as conn:
             self.set_autocommit(conn, autocommit)
 
-            # SnowflakeCursor does not extend ContextManager, so we have to ignore mypy error here
-            with closing(conn.cursor(DictCursor)) as cur:  # type: ignore[type-var]
+            with self._get_cursor(conn, return_dictionaries) as cur:
                 results = []
                 for sql_statement in sql_list:
                     self._run_command(cur, sql_statement, parameters)
 
                     if handler is not None:
                         result = handler(cur)
-                        results.append(result)
+                        if return_single_query_results(sql, return_last, split_statements):
+                            _last_result = result
+                            _last_description = cur.description
+                        else:
+                            results.append(result)
+                            self.descriptions.append(cur.description)
 
                     query_id = cur.sfqid
                     self.log.info("Rows affected: %s", cur.rowcount)
@@ -390,7 +398,21 @@ def run(
 
         if handler is None:
             return None
-        elif return_single_query_results(sql, return_last, split_statements):
-            return results[-1]
+        if return_single_query_results(sql, return_last, split_statements):
+            self.descriptions = [_last_description]
+            return _last_result
         else:
             return results
+
+    @contextmanager
+    def _get_cursor(self, conn: Any, return_dictionaries: bool):
+        cursor = None
+        try:
+            if return_dictionaries:
+                cursor = conn.cursor(DictCursor)
+            else:
+                cursor = conn.cursor()
+            yield cursor
+        finally:
+            if cursor is not None:
+                cursor.close()
diff --git a/airflow/providers/snowflake/operators/snowflake.py b/airflow/providers/snowflake/operators/snowflake.py
index cf7835ef65518..5fddf73adc9c0 100644
--- a/airflow/providers/snowflake/operators/snowflake.py
+++ b/airflow/providers/snowflake/operators/snowflake.py
@@ -61,6 +61,7 @@ class SnowflakeOperator(SQLExecuteQueryOperator):
         through native Okta.
     :param session_parameters: You can set session-level parameters at
         the time you connect to Snowflake
+    :return Returns list of dictionaries in { 'column': 'value', 'column2': 'value2' } form.
     """
 
     template_fields: Sequence[str] = ("sql",)
@@ -91,7 +92,6 @@ def __init__(
                 "session_parameters": session_parameters,
                 **hook_params,
             }
-
         super().__init__(conn_id=snowflake_conn_id, **kwargs)
         warnings.warn(
             """This class is deprecated.
@@ -103,6 +103,26 @@ def __init__(
             stacklevel=2,
         )
 
+    def _process_output(self, results: list[Any], descriptions: list[Sequence[Sequence] | None]) -> list[Any]:
+        validated_descriptions: list[Sequence[Sequence]] = []
+        for idx, description in enumerate(descriptions):
+            if not description:
+                raise RuntimeError(
+                    f"The query did not return descriptions of the cursor for query number {idx}. "
+                    "Cannot return values in a form of dictionary for that query."
+                )
+            validated_descriptions.append(description)
+        returned_results = []
+        for result_id, result_list in enumerate(results):
+            current_processed_result = []
+            for row in result_list:
+                dict_result: dict[Any, Any] = {}
+                for idx, description in enumerate(validated_descriptions[result_id]):
+                    dict_result[description[0]] = row[idx]
+                current_processed_result.append(dict_result)
+            returned_results.append(current_processed_result)
+        return returned_results
+
 
 class SnowflakeCheckOperator(SQLCheckOperator):
     """
diff --git a/tests/providers/snowflake/hooks/test_sql.py b/tests/providers/snowflake/hooks/test_sql.py
new file mode 100644
index 0000000000000..629cbadfdce85
--- /dev/null
+++ b/tests/providers/snowflake/hooks/test_sql.py
@@ -0,0 +1,249 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+#
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from snowflake.connector import DictCursor
+
+from airflow.providers.common.sql.hooks.sql import fetch_all_handler
+from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
+
+TASK_ID = "sql-operator"
+HOST = "host"
+DEFAULT_CONN_ID = "snowflake_default"
+PASSWORD = "password"
+
+
+class SnowflakeHookForTests(SnowflakeHook):
+    conn_name_attr = "snowflake_conn_id"
+    get_conn = MagicMock(name="conn")
+
+
+def get_cursor_descriptions(fields: list[str]) -> list[tuple[str]]:
+    return [(field,) for field in fields]
+
+
+@pytest.mark.parametrize(
+    "return_last, split_statements, sql, cursor_calls,"
+    "cursor_descriptions, cursor_results, hook_descriptions, hook_results, return_dictionaries",
+    [
+        pytest.param(
+            True,
+            False,
+            "select * from test.test",
+            ["select * from test.test"],
+            [["id", "value"]],
+            ([[1, 2], [11, 12]],),
+            [[("id",), ("value",)]],
+            [[1, 2], [11, 12]],
+            False,
+            id="The return_last set and no split statements set on single query in string",
+        ),
+        pytest.param(
+            False,
+            False,
+            "select * from test.test;",
+            ["select * from test.test;"],
+            [["id", "value"]],
+            ([[1, 2], [11, 12]],),
+            [[("id",), ("value",)]],
+            [[1, 2], [11, 12]],
+            False,
+            id="The return_last not set and no split statements set on single query in string",
+        ),
+        pytest.param(
+            True,
+            True,
+            "select * from test.test;",
+            ["select * from test.test;"],
+            [["id", "value"]],
+            ([[1, 2], [11, 12]],),
+            [[("id",), ("value",)]],
+            [[1, 2], [11, 12]],
+            False,
+            id="The return_last set and split statements set on single query in string",
+        ),
+        pytest.param(
+            False,
+            True,
+            "select * from test.test;",
+            ["select * from test.test;"],
+            [["id", "value"]],
+            ([[1, 2], [11, 12]],),
+            [[("id",), ("value",)]],
+            [[[1, 2], [11, 12]]],
+            False,
+            id="The return_last not set and split statements set on single query in string",
+        ),
+        pytest.param(
+            True,
+            True,
+            "select * from test.test;select * from test.test2;",
+            ["select * from test.test;", "select * from test.test2;"],
+            [["id", "value"], ["id2", "value2"]],
+            ([[1, 2], [11, 12]], [[3, 4], [13, 14]]),
+            [[("id2",), ("value2",)]],
+            [[3, 4], [13, 14]],
+            False,
+            id="The return_last set and split statements set on multiple queries in string",
+        ),  # Failing
+        pytest.param(
+            False,
+            True,
+            "select * from test.test;select * from test.test2;",
+            ["select * from test.test;", "select * from test.test2;"],
+            [["id", "value"], ["id2", "value2"]],
+            ([[1, 2], [11, 12]], [[3, 4], [13, 14]]),
+            [[("id",), ("value",)], [("id2",), ("value2",)]],
+            [[[1, 2], [11, 12]], [[3, 4], [13, 14]]],
+            False,
+            id="The return_last not set and split statements set on multiple queries in string",
+        ),
+        pytest.param(
+            True,
+            True,
+            ["select * from test.test;"],
+            ["select * from test.test"],
+            [["id", "value"]],
+            ([[1, 2], [11, 12]],),
+            [[("id",), ("value",)]],
+            [[[1, 2], [11, 12]]],
+            False,
+            id="The return_last set on single query in list",
+        ),
+        pytest.param(
+            False,
+            True,
+            ["select * from test.test;"],
+            ["select * from test.test"],
+            [["id", "value"]],
+            ([[1, 2], [11, 12]],),
+            [[("id",), ("value",)]],
+            [[[1, 2], [11, 12]]],
+            False,
+            id="The return_last not set on single query in list",
+        ),
+        pytest.param(
+            True,
+            True,
+            "select * from test.test;select * from test.test2;",
+            ["select * from test.test", "select * from test.test2"],
+            [["id", "value"], ["id2", "value2"]],
+            ([[1, 2], [11, 12]], [[3, 4], [13, 14]]),
+            [[("id2",), ("value2",)]],
+            [[3, 4], [13, 14]],
+            False,
+            id="The return_last set set on multiple queries in list",
+        ),
+        pytest.param(
+            False,
+            True,
+            "select * from test.test;select * from test.test2;",
+            ["select * from test.test", "select * from test.test2"],
+            [["id", "value"], ["id2", "value2"]],
+            ([[1, 2], [11, 12]], [[3, 4], [13, 14]]),
+            [[("id",), ("value",)], [("id2",), ("value2",)]],
+            [[[1, 2], [11, 12]], [[3, 4], [13, 14]]],
+            False,
+            id="The return_last not set on multiple queries not set",
+        ),
+        pytest.param(
+            False,
+            True,
+            "select * from test.test;select * from test.test2;",
+            ["select * from test.test", "select * from test.test2"],
+            [["id", "value"], ["id2", "value2"]],
+            (
+                [{"id": 1, "value": 2}, {"id": 11, "value": 12}],
+                [{"id2": 3, "value2": 4}, {"id2": 13, "value2": 14}],
+            ),
+            [[("id",), ("value",)], [("id2",), ("value2",)]],
+            [
+                [{"id": 1, "value": 2}, {"id": 11, "value": 12}],
+                [{"id2": 3, "value2": 4}, {"id2": 13, "value2": 14}],
+            ],
+            True,
+            id="DictCursor: The return_last not set on multiple queries not set",
+        ),
+    ],
+)
+def test_query(
+    return_last,
+    split_statements,
+    sql,
+    cursor_calls,
+    cursor_descriptions,
+    cursor_results,
+    hook_descriptions,
+    hook_results,
+    return_dictionaries,
+):
+    modified_descriptions = [
+        get_cursor_descriptions(cursor_description) for cursor_description in cursor_descriptions
+    ]
+    dbapi_hook = SnowflakeHookForTests()
+    dbapi_hook.get_conn.return_value.cursor.return_value.rowcount = 2
+    dbapi_hook.get_conn.return_value.cursor.return_value._description_index = 0
+
+    def mock_execute(*args, **kwargs):
+        # the run method accesses description property directly, and we need to modify it after
+        # every execute, to make sure that different descriptions are returned. I could not find easier
+        # method with mocking
+        dbapi_hook.get_conn.return_value.cursor.return_value.description = modified_descriptions[
+            dbapi_hook.get_conn.return_value.cursor.return_value._description_index
+        ]
+        dbapi_hook.get_conn.return_value.cursor.return_value._description_index += 1
+
+    dbapi_hook.get_conn.return_value.cursor.return_value.execute = mock_execute
+    dbapi_hook.get_conn.return_value.cursor.return_value.fetchall.side_effect = cursor_results
+    results = dbapi_hook.run(
+        sql=sql,
+        handler=fetch_all_handler,
+        return_last=return_last,
+        split_statements=split_statements,
+        return_dictionaries=return_dictionaries,
+    )
+
+    assert dbapi_hook.descriptions == hook_descriptions
+    assert dbapi_hook.last_description == hook_descriptions[-1]
+    assert results == hook_results
+
+    if return_dictionaries:
+        dbapi_hook.get_conn.return_value.cursor.assert_called_with(DictCursor)
+    else:
+        dbapi_hook.get_conn.return_value.cursor.assert_called_with()
+    dbapi_hook.get_conn.return_value.cursor.return_value.close.assert_called()
+
+
+@pytest.mark.parametrize(
+    "empty_statement",
+    [
+        pytest.param([], id="Empty list"),
+        pytest.param("", id="Empty string"),
+        pytest.param("\n", id="Only EOL"),
+    ],
+)
+def test_no_query(empty_statement):
+    dbapi_hook = SnowflakeHookForTests()
+    dbapi_hook.get_conn.return_value.cursor.rowcount = 0
+    with pytest.raises(ValueError) as err:
+        dbapi_hook.run(sql=empty_statement)
+    assert err.value.args[0] == "List of SQL statements is empty"
diff --git a/tests/providers/snowflake/operators/test_snowflake_sql.py b/tests/providers/snowflake/operators/test_snowflake_sql.py
new file mode 100644
index 0000000000000..3794bfc550bf4
--- /dev/null
+++ b/tests/providers/snowflake/operators/test_snowflake_sql.py
@@ -0,0 +1,140 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from databricks.sql.types import Row
+
+from airflow.providers.common.sql.hooks.sql import fetch_all_handler
+from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
+
+DATE = "2017-04-20"
+TASK_ID = "databricks-sql-operator"
+DEFAULT_CONN_ID = "snowflake_default"
+
+
+@pytest.mark.parametrize(
+    "sql, return_last, split_statement, hook_results, hook_descriptions, expected_results",
+    [
+        pytest.param(
+            "select * from dummy",
+            True,
+            True,
+            [Row(id=1, value="value1"), Row(id=2, value="value2")],
+            [[("id",), ("value",)]],
+            ([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]),
+            id="Scalar: Single SQL statement, return_last, split statement",
+        ),
+        pytest.param(
+            "select * from dummy;select * from dummy2",
+            True,
+            True,
+            [Row(id=1, value="value1"), Row(id=2, value="value2")],
+            [[("id",), ("value",)]],
+            ([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]),
+            id="Scalar: Multiple SQL statements, return_last, split statement",
+        ),
+        pytest.param(
+            "select * from dummy",
+            False,
+            False,
+            [Row(id=1, value="value1"), Row(id=2, value="value2")],
+            [[("id",), ("value",)]],
+            ([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]),
+            id="Scalar: Single SQL statements, no return_last (doesn't matter), no split statement",
+        ),
+        pytest.param(
+            "select * from dummy",
+            True,
+            False,
+            [Row(id=1, value="value1"), Row(id=2, value="value2")],
+            [[("id",), ("value",)]],
+            ([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]),
+            id="Scalar: Single SQL statements, return_last (doesn't matter), no split statement",
+        ),
+        pytest.param(
+            ["select * from dummy"],
+            False,
+            False,
+            [[Row(id=1, value="value1"), Row(id=2, value="value2")]],
+            [[("id",), ("value",)]],
+            [([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}])],
+            id="Non-Scalar: Single SQL statements in list, no return_last, no split statement",
+        ),
+        pytest.param(
+            ["select * from dummy", "select * from dummy2"],
+            False,
+            False,
+            [
+                [Row(id=1, value="value1"), Row(id=2, value="value2")],
+                [Row(id2=1, value2="value1"), Row(id2=2, value2="value2")],
+            ],
+            [[("id",), ("value",)], [("id2",), ("value2",)]],
+            [
+                ([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]),
+                ([{"id2": 1, "value2": "value1"}, {"id2": 2, "value2": "value2"}]),
+            ],
+            id="Non-Scalar: Multiple SQL statements in list, no return_last (no matter), no split statement",
+        ),
+        pytest.param(
+            ["select * from dummy", "select * from dummy2"],
+            True,
+            False,
+            [
+                [Row(id=1, value="value1"), Row(id=2, value="value2")],
+                [Row(id2=1, value2="value1"), Row(id2=2, value2="value2")],
+            ],
+            [[("id",), ("value",)], [("id2",), ("value2",)]],
+            [
+                ([{"id": 1, "value": "value1"}, {"id": 2, "value": "value2"}]),
+                ([{"id2": 1, "value2": "value1"}, {"id2": 2, "value2": "value2"}]),
+            ],
+            id="Non-Scalar: Multiple SQL statements in list, return_last (no matter), no split statement",
+        ),
+    ],
+)
+def test_exec_success(sql, return_last, split_statement, hook_results, hook_descriptions, expected_results):
+    """
+    Test the execute function in case where SQL query was successful.
+    """
+    with patch("airflow.providers.common.sql.operators.sql.BaseSQLOperator.get_db_hook") as get_db_hook_mock:
+        op = SnowflakeOperator(
+            task_id=TASK_ID,
+            sql=sql,
+            do_xcom_push=True,
+            return_last=return_last,
+            split_statements=split_statement,
+        )
+        dbapi_hook = MagicMock()
+        get_db_hook_mock.return_value = dbapi_hook
+        dbapi_hook.run.return_value = hook_results
+        dbapi_hook.descriptions = hook_descriptions
+
+        execute_results = op.execute(None)
+
+        assert execute_results == expected_results
+        dbapi_hook.run.assert_called_once_with(
+            sql=sql,
+            parameters=None,
+            handler=fetch_all_handler,
+            autocommit=False,
+            return_last=return_last,
+            split_statements=split_statement,
+        )
