SNOW-1830556 Add decoder logic for Session.write_pandas (#2904)

1. Which Jira issue is this PR addressing? Make sure that there is an
accompanying issue to your PR.

   Fixes SNOW-1830556

2. Fill out the following pre-review checklist:

- [ ] I am adding a new automated test(s) to verify correctness of my
new code
- [ ] If this test skips Local Testing mode, I'm requesting review from
@snowflakedb/local-testing
   - [ ] I am adding new logging messages
   - [ ] I am adding a new telemetry message
   - [ ] I am adding new credentials
   - [ ] I am adding a new dependency
- [ ] If this is a new feature/behavior, I'm adding the Local Testing
parity changes.
- [x] I acknowledge that I have ensured my changes to be thread-safe.
Follow the link for more information: [Thread-safe Developer
Guidelines](https://github.com/snowflakedb/snowpark-python/blob/main/CONTRIBUTING.md#thread-safe-development)

3. Please describe how your code solves the related issue.

Added decoder logic for Session.write_pandas, fixed bug in decoder for
sp_table, and verified shadowed_local_name.test works.

Author: sfc-gh-vbudati

tests/ast/decoder.py

@@ -8,8 +8,9 @@
 from datetime import date, datetime, time, timedelta, timezone
 from decimal import Decimal
 
-from snowflake.snowpark.window import WindowSpec, Window, WindowRelativePosition
+from pandas import DataFrame as PandasDataFrame
 
+from snowflake.snowpark.window import WindowSpec, Window, WindowRelativePosition
 import snowflake.snowpark._internal.proto.generated.ast_pb2 as proto
 
 from google.protobuf.json_format import MessageToDict
@@ -168,7 +169,7 @@ def convert_name_to_list(self, name: any) -> List:
             return [name]
         return [qualified_name for qualified_name in name]
 
-    def decode_name_expr(self, table_name: proto.SpName) -> str:
+    def decode_name_expr(self, table_name: proto.SpName) -> Union[str, List]:
         """
         Decode a table name expression to get the table name.
 
@@ -185,7 +186,7 @@ def decode_name_expr(self, table_name: proto.SpName) -> str:
         if table_name.name.HasField("sp_name_flat"):
             return table_name.name.sp_name_flat.name
         elif table_name.name.HasField("sp_name_structured"):
-            return table_name.name.sp_name_structured.name
+            return [name for name in table_name.name.sp_name_structured.name]
         else:
             raise ValueError("Table name not found in proto.SpTableName")
 
@@ -236,7 +237,9 @@ def decode_fn_ref_expr(self, fn_ref_expr: proto.FnRefExpr) -> str:
                     % fn_ref_expr.WhichOneof("variant")
                 )
 
-    def decode_dataframe_data_expr(self, df_data_expr: proto.SpDataframeData) -> List:
+    def decode_dataframe_data_expr(
+        self, df_data_expr: proto.SpDataframeData
+    ) -> Union[List, PandasDataFrame]:
         """
         Decode a dataframe data expression to get the underlying data.
 
@@ -247,7 +250,7 @@ def decode_dataframe_data_expr(self, df_data_expr: proto.SpDataframeData) -> Lis
 
         Returns
         -------
-        List
+        List or pandas.DataFrame
             The decoded data.
         """
         match df_data_expr.WhichOneof("sealed_value"):
@@ -265,14 +268,15 @@ def decode_dataframe_data_expr(self, df_data_expr: proto.SpDataframeData) -> Lis
                         ]
                 else:
                     return []
-            # case "sp_dataframe_data__pandas":
-            #     pass
+            case "sp_dataframe_data__pandas":
+                # We don't know what pandas DataFrame was passed in, return an empty one.
+                return PandasDataFrame()
             # case "sp_dataframe_data__tuple":
             #     pass
             case _:
                 raise ValueError(
                     "Unknown dataframe data type: %s"
-                    % df_data_expr.WhichOneof("variant")
+                    % df_data_expr.WhichOneof("sealed_value")
                 )
 
     def decode_dataframe_schema_expr(
@@ -1750,7 +1754,12 @@ def decode_expr(self, expr: proto.Expr, **kwargs) -> Any:
             case "sp_table":
                 assert expr.sp_table.HasField("name")
                 table_name = self.decode_name_expr(expr.sp_table.name)
-                return self.session.table(table_name)
+                return self.session.table(
+                    table_name,
+                    is_temp_table_for_cleanup=expr.sp_table.is_temp_table_for_cleanup
+                    if hasattr(expr.sp_table, "is_temp_table_for_cleanup")
+                    else False,
+                )
 
             case "sp_to_snowpark_pandas":
                 df = self.decode_expr(expr.sp_to_snowpark_pandas.df)
@@ -2118,6 +2127,48 @@ def decode_expr(self, expr: proto.Expr, **kwargs) -> Any:
                         columns, rowcount=row_count, timelimit=time_limit_seconds
                     )
 
+            case "sp_write_pandas":
+                df = self.decode_dataframe_data_expr(expr.sp_write_pandas.df)
+                table_name = self.decode_name_expr(expr.sp_write_pandas.table_name)
+                if isinstance(table_name, str):
+                    database, schema = None, None
+                else:
+                    database, schema, table_name = (
+                        table_name[0],
+                        table_name[1],
+                        table_name[2],
+                    )
+                chunk_size = (
+                    expr.sp_write_pandas.chunk_size.value
+                    if expr.sp_write_pandas.HasField("chunk_size")
+                    else None
+                )
+                compression = expr.sp_write_pandas.compression
+                on_error = expr.sp_write_pandas.on_error
+                parallel = expr.sp_write_pandas.parallel
+                quote_identifiers = expr.sp_write_pandas.quote_identifiers
+                auto_create_table = expr.sp_write_pandas.auto_create_table
+                create_temp_table = expr.sp_write_pandas.create_temp_table
+                overwrite = expr.sp_write_pandas.overwrite
+                table_type = expr.sp_write_pandas.table_type
+                kwargs = self.decode_dsl_map_expr(expr.sp_write_pandas.kwargs)
+                return self.session.write_pandas(
+                    df,
+                    table_name,
+                    database=database,
+                    schema=schema,
+                    chunk_size=chunk_size,
+                    compression=compression,
+                    on_error=on_error,
+                    parallel=parallel,
+                    quote_identifiers=quote_identifiers,
+                    auto_create_table=auto_create_table,
+                    create_temp_table=create_temp_table,
+                    overwrite=overwrite,
+                    table_type=table_type,
+                    **kwargs,
+                )
+
             case "sp_row":
                 names = [name for name in expr.sp_row.names.list]
                 values = [self.decode_expr(value) for value in expr.sp_row.vs]