fix(snowflake): allow backend to choose how to prefix table names dur…

…ing compilation

ibis/backends/base/sql/alchemy/__init__.py

@@ -930,9 +930,13 @@ def _get_sqla_table(
 
         with self._use_schema(ident, current_db, current_schema):
             result = self._table_from_schema(name, schema=ibis_schema)
-        result.schema = schema
+        result.schema = self._get_schema_for_table(qualname=ident, schema=schema)
         return result
 
+    @abc.abstractmethod
+    def _get_schema_for_table(self, *, qualname: str, schema: str) -> str:
+        """Choose whether to prefix a table with its fully qualified path or schema."""
+
     def drop_table(
         self, name: str, database: str | None = None, force: bool = False
     ) -> None:
@@ -943,17 +947,3 @@ def drop_table(
         drop_stmt = "DROP TABLE" + (" IF EXISTS" * force) + f" {name}"
         with self.begin() as con:
             con.exec_driver_sql(drop_stmt)
-
-
-@compiles(sa.Table, "snowflake")
-def compile_table(element, compiler, **kw):
-    """Override compilation of leaf tables.
-
-    The override is necessary because the dialect does not handle database
-    hierarchies and/or quoting properly.
-    """
-    schema = element.schema
-    name = compiler.preparer.quote_identifier(element.name)
-    if schema is not None:
-        return f"{schema}.{name}"
-    return name

ibis/backends/snowflake/__init__.py

@@ -805,6 +805,23 @@ def read_json(
 
         return self.table(table)
 
+    def _get_schema_for_table(self, *, qualname: str, schema: str) -> str:
+        return qualname
+
+
+@compiles(sa.Table, "snowflake")
+def compile_table(element, compiler, **kw):
+    """Override compilation of leaf tables.
+
+    The override is necessary because snowflake-sqlalchemy does not handle
+    quoting databases and schemas correctly.
+    """
+    schema = element.schema
+    name = compiler.preparer.quote_identifier(element.name)
+    if schema is not None:
+        return f"{schema}.{name}"
+    return name
+
 
 @compiles(sa.sql.Join, "snowflake")
 def compile_join(element, compiler, **kw):

ibis/backends/trino/__init__.py

@@ -363,3 +363,11 @@ def _table_from_schema(
             trino_catalog=database or self.current_database,
             **kwargs,
         )
+
+    def _get_schema_for_table(self, *, qualname: str, schema: str) -> str:
+        """Trino compiles the `trino_catalog` argument into `sa.Table`.
+
+        This means we only need the schema and not the fully qualified
+        $catalog.$schema identifier.
+        """
+        return schema