refactor(snowflake): use snowflake-connector-python directly, drop ibis dependency

core/wren/justfile

@@ -59,6 +59,9 @@ test-postgres:
 test-mysql:
     uv run pytest tests/connectors/test_mysql.py -v -m mysql
 
+test-snowflake:
+    uv run pytest tests/connectors/test_snowflake.py -v -m snowflake
+
 test-connector marker:
     uv run pytest tests/connectors/ -v -m {{ marker }}
 

core/wren/pyproject.toml

@@ -46,7 +46,7 @@ dependencies = [
 postgres = ["psycopg>=3", "ibis-framework[postgres]"]
 mysql = ["mysqlclient>=2.2", "ibis-framework[mysql]"]
 bigquery = ["ibis-framework[bigquery]", "google-auth"]
-snowflake = ["ibis-framework[snowflake]"]
+snowflake = ["snowflake-connector-python[pandas]>=3.10"]
 clickhouse = ["ibis-framework[clickhouse]"]
 trino = ["ibis-framework[trino]", "trino>=0.321"]
 mssql = ["ibis-framework[mssql]"]

core/wren/src/wren/connector/factory.py

@@ -22,7 +22,7 @@
     DataSource.trino: "wren.connector.ibis",
     DataSource.clickhouse: "wren.connector.ibis",
     DataSource.oracle: "wren.connector.oracle",
-    DataSource.snowflake: "wren.connector.ibis",
+    DataSource.snowflake: "wren.connector.snowflake",
     DataSource.athena: "wren.connector.ibis",
 }
 
@@ -41,7 +41,6 @@
     DataSource.doris,
     DataSource.trino,
     DataSource.clickhouse,
-    DataSource.snowflake,
     DataSource.athena,
 }
 

core/wren/src/wren/connector/snowflake.py

@@ -0,0 +1,93 @@
+"""Native snowflake-connector-python connector — bypasses ibis snowflake backend."""
+
+from __future__ import annotations
+
+import pyarrow as pa
+import snowflake.connector
+
+from wren.connector.base import ConnectorABC
+from wren.model.error import DIALECT_SQL, ErrorCode, ErrorPhase, WrenError
+
+
+def _build_connection_params(connection_info) -> dict:
+    params: dict = {
+        "user": connection_info.user,
+        "account": connection_info.account,
+        "database": connection_info.database,
+        "schema": connection_info.sf_schema,
+    }
+    if (
+        hasattr(connection_info, "private_key")
+        and connection_info.private_key is not None
+    ):
+        params["private_key"] = connection_info.private_key.get_secret_value()
+    elif connection_info.password is not None:
+        params["password"] = connection_info.password.get_secret_value()
+
+    if hasattr(connection_info, "warehouse") and connection_info.warehouse:
+        params["warehouse"] = connection_info.warehouse
+
+    if connection_info.kwargs:
+        extra_kwargs = dict(connection_info.kwargs)
+        statement_timeout = extra_kwargs.pop("statement_timeout", None)
+        params.update(extra_kwargs)
+        if statement_timeout is not None:
+            session_parameters = params.setdefault("session_parameters", {})
+            session_parameters.setdefault(
+                "STATEMENT_TIMEOUT_IN_SECONDS", int(statement_timeout)
+            )
+    return params
+
+
+def make_snowflake_connection(connection_info):
+    return snowflake.connector.connect(**_build_connection_params(connection_info))
+
+
+class SnowflakeConnector(ConnectorABC):
+    def __init__(self, connection_info):
+        self.connection = make_snowflake_connection(connection_info)
+
+    def query(self, sql: str, limit: int | None = None) -> pa.Table:
+        try:
+            with self.connection.cursor() as cursor:
+                cursor.execute(sql)
+                arrow_table = cursor.fetch_arrow_all()
+        except snowflake.connector.errors.ProgrammingError as e:
+            raise WrenError(
+                ErrorCode.INVALID_SQL,
+                str(e),
+                phase=ErrorPhase.SQL_EXECUTION,
+                metadata={DIALECT_SQL: sql},
+            ) from e
+
+        if arrow_table is None:
+            return pa.table({})
+        if limit is not None:
+            return arrow_table.slice(0, limit)
+        return arrow_table
+
+    def dry_run(self, sql: str) -> None:
+        try:
+            with self.connection.cursor() as cursor:
+                cursor.describe(sql)
+        except snowflake.connector.errors.ProgrammingError as e:
+            raise WrenError(
+                ErrorCode.INVALID_SQL,
+                str(e),
+                phase=ErrorPhase.SQL_DRY_RUN,
+                metadata={DIALECT_SQL: sql},
+            ) from e
+
+    def close(self) -> None:
+        if self.connection is None:
+            return
+        try:
+            self.connection.close()
+        except Exception:
+            pass
+        finally:
+            self.connection = None
+
+
+def create_connector(connection_info) -> SnowflakeConnector:
+    return SnowflakeConnector(connection_info)

core/wren/src/wren/model/data_source.py

@@ -400,28 +400,10 @@ def get_oracle_connection(info: OracleConnectionInfo) -> BaseBackend:
         )
 
     @staticmethod
-    def get_snowflake_connection(info: SnowflakeConnectionInfo) -> BaseBackend:
-        if hasattr(info, "private_key") and info.private_key:
-            params = {
-                "user": info.user,
-                "private_key": info.private_key.get_secret_value(),
-                "account": info.account,
-                "database": info.database,
-                "schema": info.sf_schema,
-            }
-        else:
-            params = {
-                "user": info.user,
-                "password": info.password.get_secret_value(),
-                "account": info.account,
-                "database": info.database,
-                "schema": info.sf_schema,
-            }
-        if hasattr(info, "warehouse") and info.warehouse:
-            params["warehouse"] = info.warehouse
-        if info.kwargs:
-            params.update(info.kwargs)
-        return ibis.snowflake.connect(**params)
+    def get_snowflake_connection(info: SnowflakeConnectionInfo):
+        from wren.connector.snowflake import make_snowflake_connection  # noqa: PLC0415
+
+        return make_snowflake_connection(info)
 
     @staticmethod
     def get_trino_connection(info: TrinoConnectionInfo) -> BaseBackend:

core/wren/tests/conftest.py

@@ -16,3 +16,6 @@ def pytest_configure(config: pytest.Config) -> None:
         "markers", "postgres: PostgreSQL connector tests — requires Docker"
     )
     config.addinivalue_line("markers", "mysql: MySQL connector tests — requires Docker")
+    config.addinivalue_line(
+        "markers", "snowflake: Snowflake connector tests — mocked, no Docker required"
+    )