protect the change

Author: sfc-gh-aling

src/snowflake/snowpark/_internal/analyzer/select_statement.py

@@ -586,9 +586,15 @@ def __deepcopy__(self, memodict={}) -> "SelectableEntity":  # noqa: B006
             deepcopy(self.entity, memodict), analyzer=self.analyzer
         )
         _deepcopy_selectable_fields(from_selectable=self, to_selectable=copied)
-        copied._attributes = (
-            deepcopy(self._attributes) if self._attributes is not None else None
-        )
+        if (
+            self._session.reduce_describe_query_enabled
+            and self._session.cte_optimization_enabled
+        ):
+            copied._attributes = (
+                deepcopy(self._attributes, memodict)
+                if self._attributes is not None
+                else None
+            )
         return copied
 
     @property
@@ -929,9 +935,6 @@ def __copy__(self):
         )
         # The following values will change if they're None in the newly copied one so reset their values here
         # to avoid problems.
-        new._attributes = (
-            self._attributes.copy() if self._attributes is not None else None
-        )
         new._projection_in_str = None
         new._schema_query = None
         new._column_states = None
@@ -945,6 +948,13 @@ def __copy__(self):
             self._merge_projection_complexity_with_subquery
         )
         new.df_ast_ids = self.df_ast_ids.copy() if self.df_ast_ids is not None else None
+        if (
+            self._session.reduce_describe_query_enabled
+            and self._session.cte_optimization_enabled
+        ):
+            new._attributes = (
+                self._attributes.copy() if self._attributes is not None else None
+            )
         return new
 
     def __deepcopy__(self, memodict={}) -> "SelectStatement":  # noqa: B006
@@ -964,9 +974,15 @@ def __deepcopy__(self, memodict={}) -> "SelectStatement":  # noqa: B006
         )
 
         _deepcopy_selectable_fields(from_selectable=self, to_selectable=copied)
-        copied._attributes = (
-            deepcopy(self._attributes) if self._attributes is not None else None
-        )
+        if (
+            self._session.reduce_describe_query_enabled
+            and self._session.cte_optimization_enabled
+        ):
+            copied._attributes = (
+                deepcopy(self._attributes, memodict)
+                if self._attributes is not None
+                else None
+            )
         copied._projection_in_str = self._projection_in_str
         copied._query_params = deepcopy(self._query_params)
         copied._merge_projection_complexity_with_subquery = (
@@ -1412,7 +1428,11 @@ def select(self, cols: List[Expression]) -> "SelectStatement":
         if can_be_flattened:
             new = copy(self)
             final_projection = []
-            new._attributes = None  # reset attributes since projection changed
+            if (
+                self._session.reduce_describe_query_enabled
+                and self._session.cte_optimization_enabled
+            ):
+                new._attributes = None  # reset attributes since projection changed
             assert new_column_states is not None
             for col, state in new_column_states.items():
                 if state.change_state in (

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

@@ -20,6 +20,7 @@
 from snowflake.snowpark._internal.utils import (
     TEMP_OBJECT_NAME_PREFIX,
     TempObjectType,
+    random_name_for_temp_object,
 )
 
 
@@ -164,11 +165,14 @@ def _update_parents(
                         node.encoded_node_id_with_query
                     ]
                 else:
-                    # create a WithQueryBlock node with deterministic name
-                    # Use first 16 chars of encoded_node_id_with_query (SHA256 hash)
-                    # This ensures the same node always gets the same CTE name
-                    cte_name = f"{TEMP_OBJECT_NAME_PREFIX}{TempObjectType.CTE.value}_{node.encoded_node_id_with_query[:16].upper()}"
-                    with_block = WithQueryBlock(name=cte_name, child=node)
+                    if self._query_generator.session.reduce_describe_query_enabled:
+                        # create a deterministic name using the first 16 chars of encoded_node_id_with_query (SHA256 hash)
+                        # This ensures the same node always gets the same CTE name.
+                        #  it helps when DataFrame.queries is called multiple times, they will get the same CTE name.
+                        cte_name = f"{TEMP_OBJECT_NAME_PREFIX}{TempObjectType.CTE.value}_{node.encoded_node_id_with_query[:16].upper()}"
+                    else:
+                        cte_name = random_name_for_temp_object(TempObjectType.CTE)
+                    with_block = WithQueryBlock(name=cte_name, child=node)  # type: ignore
                     with_block._is_valid_for_replacement = True
 
                     resolved_with_block = self._query_generator.resolve(with_block)