add test + use single pass

Author: sfc-gh-joshi

src/snowflake/snowpark/_internal/compiler/plan_compiler.py

@@ -206,7 +206,8 @@ def replace_temp_obj_placeholders(
         placeholders = {}
         # Final execution queries
         execution_queries = {}
-        for query_list in queries.values():
+        for query_type, query_list in queries.items():
+            execution_queries[query_type] = []
             for query in query_list:
                 # If the query contains a temp object name placeholder, we generate a random
                 # name for the temp object and add it to the placeholders dictionary.
@@ -215,13 +216,10 @@ def replace_temp_obj_placeholders(
                         placeholder_name,
                         temp_obj_type,
                     ) = query.temp_obj_name_placeholder
-                    placeholders[placeholder_name] = random_name_for_temp_object(
-                        temp_obj_type
-                    )
-        # This loop must be done in a separate pass to ensure a CREATE/DROP pair actually refer to the same object.
-        for query_type, query_list in queries.items():
-            execution_queries[query_type] = []
-            for query in query_list:
+                    if placeholder_name not in placeholders:
+                        placeholders[placeholder_name] = random_name_for_temp_object(
+                            temp_obj_type
+                        )
                 copied_query = copy.copy(query)
                 for placeholder_name, target_temp_name in placeholders.items():
                     # Copy the original query and replace all the placeholder names with the

tests/integ/test_multithreading.py

@@ -848,6 +848,7 @@ def process_data(df_, thread_id):
                 len(unique_create_table_queries) == expected_num_queries
             ), queries_sent
             assert len(unique_drop_table_queries) == expected_num_queries, queries_sent
+            assert unique_create_table_queries == unique_drop_table_queries
 
         finally:
             analyzer.ARRAY_BIND_THRESHOLD = original_value
@@ -913,6 +914,7 @@ def process_data(df_, thread_id):
 
     assert len(unique_create_file_format_queries) == 10
     assert len(unique_drop_file_format_queries) == 10
+    assert unique_create_file_format_queries == unique_drop_file_format_queries
 
 
 @pytest.mark.xfail(