Fix: Stop leaking temporary tables in the Snowflake adapter (#2865)

Author: erindru

sqlmesh/core/engine_adapter/snowflake.py

@@ -230,7 +230,10 @@ def query_factory() -> Query:
                             df[column] = pd.to_datetime(df[column]).dt.strftime(
                                 "%Y-%m-%d %H:%M:%S.%f"
                             )  # type: ignore
-                self.create_table(temp_table, columns_to_types)
+
+                # create the table first using our usual method ensure the column datatypes match what we parsed with sqlglot
+                # otherwise we would be trusting `write_pandas()` from the snowflake lib to do this correctly
+                self.create_table(temp_table, columns_to_types, table_kind="TEMPORARY TABLE")
 
                 write_pandas(
                     self._connection_pool.get(),
@@ -240,7 +243,7 @@ def query_factory() -> Query:
                     database=temp_table.catalog or None,
                     chunk_size=self.DEFAULT_BATCH_SIZE,
                     overwrite=True,
-                    table_type="temp",
+                    table_type="temp",  # if you dont have this, it will convert the table we created above into a normal table and it wont get dropped when the session ends
                 )
             else:
                 raise SQLMeshError(
@@ -292,6 +295,7 @@ def _get_data_objects(
                 )
                 .when(exp.column("TABLE_TYPE").eq("BASE TABLE"), exp.Literal.string("TABLE"))
                 .when(exp.column("TABLE_TYPE").eq("TEMPORARY TABLE"), exp.Literal.string("TABLE"))
+                .when(exp.column("TABLE_TYPE").eq("LOCAL TEMPORARY"), exp.Literal.string("TABLE"))
                 .when(exp.column("TABLE_TYPE").eq("EXTERNAL TABLE"), exp.Literal.string("TABLE"))
                 .when(exp.column("TABLE_TYPE").eq("EVENT TABLE"), exp.Literal.string("TABLE"))
                 .when(exp.column("TABLE_TYPE").eq("VIEW"), exp.Literal.string("VIEW"))

tests/core/engine_adapter/test_integration.py

@@ -781,13 +781,22 @@ def test_temp_table(ctx: TestContext):
         ]
     )
     table = ctx.table("example")
+
+    # The snowflake adapter persists the DataFrame to an intermediate table because we use the `write_pandas()` function from the Snowflake python library
+    # Other adapters just use SQLGlot to convert the dataframe directly into a SELECT query
+    expected_tables = 2 if ctx.dialect == "snowflake" and ctx.test_type == "df" else 1
     with ctx.engine_adapter.temp_table(ctx.input_data(input_data), table.sql()) as table_name:
         results = ctx.get_metadata_results()
         assert len(results.views) == 0
-        assert len(results.tables) == 1
+        assert len(results.tables) == expected_tables
         assert len(results.non_temp_tables) == 0
         assert len(results.materialized_views) == 0
         ctx.compare_with_current(table_name, input_data)
+
+    if ctx.dialect == "snowflake":
+        # force the next query to create a new connection to prove temp tables have been dropped
+        ctx.engine_adapter._connection_pool.close()
+
     results = ctx.get_metadata_results()
     assert len(results.views) == len(results.tables) == len(results.non_temp_tables) == 0