Add SQLGlot Plugin to transpile from any SQL dialect into DuckDb

dbt/adapters/duckdb/environments/__init__.py

@@ -225,6 +225,7 @@ def initialize_cursor(
         if plugins:
             for plugin in plugins.values():
                 plugin.configure_cursor(cursor)
+                cursor = plugin.modify_cursor(cursor)
 
         for df_name, df in registered_df.items():
             cursor.register(df_name, df)

dbt/adapters/duckdb/plugins/__init__.py

@@ -147,5 +147,15 @@ def configure_cursor(self, cursor):
         """
         pass
 
+    def modify_cursor(self, cursor: DuckDBPyConnection) -> DuckDBPyConnection:
+        """
+        Modify each copy of the DuckDB cursor.
+        This method should be overridden by subclasses to extend the behaviour of cursor
+
+        :param cursor: A DuckDBPyConnection instance to be modified.
+        :return: A DuckDBPyConnection instance.
+        """
+        return cursor
+
     def default_materialization(self):
         return "table"

dbt/adapters/duckdb/plugins/sqlglot.py

@@ -0,0 +1,28 @@
+from typing import Any
+from typing import Dict
+
+import sqlglot
+
+from . import BasePlugin
+from ..environments.local import DuckDBCursorWrapper
+
+
+class SqlglotWrapper(DuckDBCursorWrapper):
+    def __init__(self, cursor, sql_from: str):
+        self.sql_from = sql_from
+        self._cursor = cursor
+
+    def execute(self, sql, bindings=None):
+        sql = sqlglot.transpile(sql, read=self.sql_from, write="duckdb").pop()
+        if bindings is None:
+            return self._cursor.execute(sql)
+        else:
+            return self._cursor.execute(sql, bindings)
+
+
+class Plugin(BasePlugin):
+    def initialize(self, plugin_config: Dict[str, Any]):
+        self.sql_from = plugin_config.get("sql_from", "duckdb")
+
+    def modify_cursor(self, cursor):
+        return SqlglotWrapper(cursor, self.sql_from)

tests/functional/plugins/test_sqlglot.py

@@ -0,0 +1,41 @@
+import pytest
+
+from dbt.tests.fixtures.project import TestProjInfo
+
+
+def _create_config(dbt_profile_target, sql_from: str) -> dict:
+    return {
+        "test": {
+            "outputs": {
+                "dev": {
+                    "type": "duckdb",
+                    "path": dbt_profile_target.get("path", ":memory:"),
+                    "plugins": [
+                        {"module": "sqlglot", "config": {"sql_from": sql_from}}
+                    ],
+                }
+            },
+            "target": "dev",
+        }
+    }
+
+
+class TestSqlglotPluginPostgres:
+    @pytest.fixture(scope="class")
+    def profiles_config_update(self, dbt_profile_target):
+        return _create_config(dbt_profile_target, "postgres")
+
+    def test_sqlglot_plugin(self, project: TestProjInfo):
+        res = project.run_sql("select to_date('2024-12-31', 'YYYY-MM-DD')", fetch="one")
+
+
+class TestSqlglotPluginMySql:
+    @pytest.fixture(scope="class")
+    def profiles_config_update(self, dbt_profile_target):
+        return _create_config(dbt_profile_target, "mysql")
+
+    def test_sqlglot_plugin(self, project: TestProjInfo):
+        project.test_config
+        res = project.run_sql(
+            "select str_to_date('2024-12-31', '%Y-%m-%d')", fetch="one"
+        )