fix(mssql): fix temporary table creation and implement cache

MSSQL has two kinds of temporary tables.

1. Local temporary tables, created in `tempdb.dbo`, prefixed with a
single `#` and suffixed with ~~112 underscores and a number.
2. Global temporary tables, created in `tempdb.dbo`, and prefixed with a
double `#`.

The suffixing on local temporary tables is to avoid name collisions
between separate user sessions trying to create the same temporary table
name.
Otherwise, the two types of temporary tables are functionally
equivalent. Users need specific access to look at temporary tables that
they didn't create, and all temp tables are cleaned up after the session
closes and all stored procedures referencing those tables have run.

Because of the slightly wacky semantics of local temporary tables, this
PR makes all temp tables created by the MSSQL backend "global". This
makes the names much more predictable.

A user creating a temporary table with the name "hithere", will find
that there is no table called "hithere" in the output of
`con.list_tables()`. Instead, they will find a table called "##hithere"
if they run `con.list_tables(database=("tempdb", "dbo"))`.

The returned table reference from the `create_table` call works the same
as any other table reference, and you can create persistent tables from
temporary tables.  The only major hiccough is the perpending of the `##`
on the table name, but that is the behavior of MSSQL.

I've also added in support for passing in explicit `catalog` and
`database` to `create_table`, which was missing (I believe this was an
oversight, since MSSQL definitely supports this).

Author: gforsyth

ibis/backends/mssql/__init__.py

@@ -133,6 +133,9 @@ def do_connect(
     def get_schema(
         self, name: str, *, catalog: str | None = None, database: str | None = None
     ) -> sch.Schema:
+        if name.startswith("ibis_cache_"):
+            catalog, database = ("tempdb", "dbo")
+            name = "##" + name
         conditions = [sg.column("table_name").eq(sge.convert(name))]
 
         if database is not None:
@@ -481,20 +484,56 @@ def create_table(
         temp: bool = False,
         overwrite: bool = False,
     ) -> ir.Table:
+        """Create a new table.
+
+        Parameters
+        ----------
+        name
+            Name of the new table.
+        obj
+            An Ibis table expression or pandas table that will be used to
+            extract the schema and the data of the new table. If not provided,
+            `schema` must be given.
+        schema
+            The schema for the new table. Only one of `schema` or `obj` can be
+            provided.
+        database
+            Name of the database where the table will be created, if not the
+            default.
+
+            To specify a location in a separate catalog, you can pass in the
+            catalog and database as a string `"catalog.database"`, or as a tuple of
+            strings `("catalog", "database")`.
+        temp
+            Whether a table is temporary or not.
+            All created temp tables are "Global Temporary Tables". They will be
+            created in "tempdb.dbo" and will be prefixed with "##".
+        overwrite
+            Whether to clobber existing data.
+            `overwrite` and `temp` cannot be used together with MSSQL.
+
+        Returns
+        -------
+        Table
+            The table that was created.
+
+        """
         if obj is None and schema is None:
             raise ValueError("Either `obj` or `schema` must be specified")
 
-        if database is not None and database != self.current_database:
-            raise com.UnsupportedOperationError(
-                "Creating tables in other databases is not supported by Postgres"
+        if temp and overwrite:
+            raise ValueError(
+                "MSSQL doesn't support overwriting temp tables, create a new temp table instead."
             )
-        else:
-            database = None
+
+        table_loc = self._to_sqlglot_table(database)
+        catalog, db = self._to_catalog_db_tuple(table_loc)
 
         properties = []
 
         if temp:
             properties.append(sge.TemporaryProperty())
+            catalog, db = None, None
 
         temp_memtable_view = None
         if obj is not None:
@@ -528,8 +567,10 @@ def create_table(
         else:
             temp_name = name
 
-        table = sg.table(temp_name, catalog=database, quoted=self.compiler.quoted)
-        raw_table = sg.table(temp_name, catalog=database, quoted=False)
+        table = sg.table(
+            "#" * temp + temp_name, catalog=catalog, db=db, quoted=self.compiler.quoted
+        )
+        raw_table = sg.table(temp_name, catalog=catalog, db=db, quoted=False)
         target = sge.Schema(this=table, expressions=column_defs)
 
         create_stmt = sge.Create(
@@ -538,11 +579,22 @@ def create_table(
             properties=sge.Properties(expressions=properties),
         )
 
-        this = sg.table(name, catalog=database, quoted=self.compiler.quoted)
-        raw_this = sg.table(name, catalog=database, quoted=False)
+        this = sg.table(name, catalog=catalog, db=db, quoted=self.compiler.quoted)
+        raw_this = sg.table(name, catalog=catalog, db=db, quoted=False)
         with self._safe_raw_sql(create_stmt) as cur:
             if query is not None:
-                insert_stmt = sge.Insert(this=table, expression=query).sql(self.dialect)
+                # You can specify that a table is temporary for the sqlglot `Create` but not
+                # for the subsequent `Insert`, so we need to shove a `#` in
+                # front of the table identifier.
+                _table = sg.table(
+                    "##" * temp + temp_name,
+                    catalog=catalog,
+                    db=db,
+                    quoted=self.compiler.quoted,
+                )
+                insert_stmt = sge.Insert(this=_table, expression=query).sql(
+                    self.dialect
+                )
                 cur.execute(insert_stmt)
 
             if overwrite:
@@ -558,11 +610,17 @@ def create_table(
             # for in-memory reads
             if temp_memtable_view is not None:
                 self.drop_table(temp_memtable_view)
-            return self.table(name, database=database)
+            return self.table(
+                "##" * temp + name,
+                database=("tempdb" * temp or catalog, "dbo" * temp or db),
+            )
 
         # preserve the input schema if it was provided
         return ops.DatabaseTable(
-            name, schema=schema, source=self, namespace=ops.Namespace(database=database)
+            name,
+            schema=schema,
+            source=self,
+            namespace=ops.Namespace(catalog=catalog, database=db),
         ).to_expr()
 
     def _register_in_memory_table(self, op: ops.InMemoryTable) -> None:

ibis/backends/mssql/tests/test_client.py

@@ -134,3 +134,21 @@ def test_list_tables_schema_warning_refactor(con):
 
     assert con.list_tables(database="msdb.dbo", like="restore") == restore_tables
     assert con.list_tables(database=("msdb", "dbo"), like="restore") == restore_tables
+
+
+def test_create_temp_table_from_obj(con):
+    obj = {"team": ["john", "joe"]}
+
+    t = con.create_table("team", obj, temp=True)
+
+    t2 = con.table("##team", database="tempdb.dbo")
+
+    assert t.to_pyarrow().equals(t2.to_pyarrow())
+
+    persisted_from_temp = con.create_table("fuhreal", t2)
+
+    assert "fuhreal" in con.list_tables()
+
+    assert persisted_from_temp.to_pyarrow().equals(t2.to_pyarrow())
+
+    con.drop_table("fuhreal")

ibis/backends/tests/test_client.py

@@ -129,7 +129,11 @@ def test_create_table(backend, con, temp_table, func, sch):
                     ["pyspark", "trino", "exasol", "risingwave"],
                     reason="No support for temp tables",
                 ),
-                pytest.mark.broken(["mssql"], reason="Incorrect temp table syntax"),
+                pytest.mark.notyet(
+                    ["mssql"],
+                    reason="Can't rename temp tables",
+                    raises=ValueError,
+                ),
                 pytest.mark.broken(
                     ["bigquery"],
                     reason="tables created with temp=True cause a 404 on retrieval",
@@ -1722,7 +1726,6 @@ def test_json_to_pyarrow(con):
     assert result == expected
 
 
-@pytest.mark.notyet(["mssql"], raises=PyODBCProgrammingError)
 @pytest.mark.notyet(
     ["risingwave", "exasol"],
     raises=com.UnsupportedOperationError,