Fix has_table() to exclude other sessions' local temp tables

gordthompson · gordthompson · commit 207ec35c2e17 · 2021-08-23T11:01:10.000-06:00
Fixes: sqlalchemy#6910 Change-Id: I9986566e1195d42ad7e9a01f0f84ef2074576257
diff --git a/doc/build/changelog/unreleased_14/6910.rst b/doc/build/changelog/unreleased_14/6910.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, mssql, reflection
+    :tickets: 6910
+
+    Fixed an issue where :meth:`_reflection.has_table` returned
+    ``True`` for local temporary tables that actually belonged to a
+    different SQL Server session (connection). An extra check is now
+    performed to ensure that the temp table detected is in fact owned
+    by the current session.
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
@@ -758,6 +758,7 @@ class MyTable(Base):
 from ... import schema as sa_schema
 from ... import Sequence
 from ... import sql
+from ... import text
 from ... import types as sqltypes
 from ... import util
 from ...engine import cursor as _cursor
@@ -2823,8 +2824,16 @@ def has_table(self, connection, tablename, dbname, owner, schema):
                 )
             )
 
-            result = connection.execute(s.limit(1))
-            return result.scalar() is not None
+            table_name = connection.execute(s.limit(1)).scalar()
+            if table_name:
+                # #6910: verify it's not a temp table from another session
+                obj_id = connection.execute(
+                    text("SELECT object_id(:table_name)"),
+                    {"table_name": "tempdb.dbo.[{}]".format(table_name)},
+                ).scalar()
+                return bool(obj_id)
+            else:
+                return False
         else:
             tables = ischema.tables
 
@@ -3009,7 +3018,12 @@ def get_view_definition(
             return view_def
 
     def _temp_table_name_like_pattern(self, tablename):
-        return tablename + (("___%") if not tablename.startswith("##") else "")
+        # LIKE uses '%' to match zero or more characters and '_' to match any
+        # single character. We want to match literal underscores, so T-SQL
+        # requires that we enclose them in square brackets.
+        return tablename + (
+            ("[_][_][_]%") if not tablename.startswith("##") else ""
+        )
 
     def _get_internal_temp_table_name(self, connection, tablename):
         # it's likely that schema is always "dbo", but since we can
