SNOW-1830529 Add decoder logic for to_local_iterator, to_pandas, to_snowpark_pandas, and to_pandas_batch (#2866)

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

   Fixes SNOW-1830529

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 `to_local_iterator`, `to_pandas`,
`to_snowpark_pandas`, and `to_pandas_batch`.

Author: sfc-gh-vbudati

src/snowflake/snowpark/mock/_nop_connection.py

@@ -21,7 +21,10 @@
 from snowflake.connector.cursor import ResultMetadata
 from snowflake.snowpark._internal.analyzer.analyzer_utils import unquote_if_quoted
 from snowflake.snowpark._internal.analyzer.expression import Attribute
-from snowflake.snowpark._internal.analyzer.snowflake_plan import SnowflakePlan
+from snowflake.snowpark._internal.analyzer.snowflake_plan import (
+    SnowflakePlan,
+    PlanQueryType,
+)
 from snowflake.snowpark._internal.analyzer.snowflake_plan_node import (
     LogicalPlan,
     SaveMode,
@@ -120,6 +123,28 @@ def execute(
     ]:
         source_plan = plan.source_plan
 
+        if hasattr(source_plan, "execution_queries"):
+            # If temp read-only table, explicitly create it.
+            # This occurs when code such as to_snowpark_pandas is run where the Snowpark version of the table is
+            # cloned and then read.
+            from snowflake.snowpark.mock import TableEmulator
+
+            for plan_query_type, query in source_plan.execution_queries.items():
+                if query:
+                    query_sql = query[0].sql
+                    if (
+                        plan_query_type == PlanQueryType.QUERIES
+                        and "TEMPORARY READ ONLY TABLE" in query_sql
+                    ):
+                        temp_table_name = query_sql.split("TEMPORARY READ ONLY TABLE ")[
+                            1
+                        ].split(" ")[0]
+                        self.entity_registry.write_table(
+                            temp_table_name,
+                            TableEmulator({"A": [1], "B": [1], "C": [1]}),
+                            SaveMode.IGNORE,
+                        )
+
         if isinstance(source_plan, SnowflakeCreateTable):
             result = self.entity_registry.write_table(
                 source_plan.table_name,

src/snowflake/snowpark/mock/_nop_plan.py

@@ -6,6 +6,7 @@
 from typing import Any, Dict, List, Optional
 
 import snowflake.snowpark
+from snowflake.snowpark.mock import TableEmulator
 from snowflake.snowpark._internal.analyzer.analyzer_utils import unquote_if_quoted
 from snowflake.snowpark._internal.analyzer.binary_plan_node import Join
 from snowflake.snowpark._internal.analyzer.expression import (
@@ -117,6 +118,9 @@ def resolve_attributes(
         pivot_attrs.extend(pivot_result_cols)
         attributes = pivot_attrs
 
+    elif isinstance(plan, TableEmulator):
+        attributes = [Attribute(name, _NumericType(), False) for name in plan.columns]
+
     elif isinstance(plan, TableUpdate):
         attributes = [
             Attribute(name, _NumericType(), False)

tests/ast/data/Dataframe.to_snowpark_pandas.test

@@ -30,8 +30,10 @@ body {
     expr {
       sp_table {
         name {
-          sp_table_name_flat {
-            name: "table1"
+          name {
+            sp_name_flat {
+              name: "table1"
+            }
           }
         }
         src {

tests/ast/decoder.py

@@ -1437,6 +1437,41 @@ def decode_expr(self, expr: proto.Expr) -> Any:
                 else:
                     return df.to_df(col_names)
 
+            case "sp_dataframe_to_local_iterator":
+                df = self.symbol_table[
+                    expr.sp_dataframe_to_local_iterator.id.bitfield1
+                ][1]
+                statement_params = self.get_statement_params(
+                    MessageToDict(expr.sp_dataframe_to_local_iterator)
+                )
+                block = expr.sp_dataframe_to_local_iterator.block
+                case_sensitive = expr.sp_dataframe_to_local_iterator.case_sensitive
+                return df.to_local_iterator(
+                    statement_params=statement_params,
+                    block=block,
+                    case_sensitive=case_sensitive,
+                )
+
+            case "sp_dataframe_to_pandas":
+                df = self.symbol_table[expr.sp_dataframe_to_pandas.id.bitfield1][1]
+                statement_params = self.get_statement_params(
+                    MessageToDict(expr.sp_dataframe_to_pandas)
+                )
+                block = expr.sp_dataframe_to_pandas.block
+                return df.to_pandas(statement_params=statement_params, block=block)
+
+            case "sp_dataframe_to_pandas_batches":
+                df = self.symbol_table[
+                    expr.sp_dataframe_to_pandas_batches.id.bitfield1
+                ][1]
+                statement_params = self.get_statement_params(
+                    MessageToDict(expr.sp_dataframe_to_pandas_batches)
+                )
+                block = expr.sp_dataframe_to_pandas_batches.block
+                return df.to_pandas_batches(
+                    statement_params=statement_params, block=block
+                )
+
             case "sp_dataframe_unpivot":
                 df = self.decode_expr(expr.sp_dataframe_unpivot.df)
                 column_list = [
@@ -1520,6 +1555,24 @@ def decode_expr(self, expr: proto.Expr) -> Any:
                 table_name = self.decode_name_expr(expr.sp_table.name)
                 return self.session.table(table_name)
 
+            case "sp_to_snowpark_pandas":
+                df = self.decode_expr(expr.sp_to_snowpark_pandas.df)
+                d = MessageToDict(expr.sp_to_snowpark_pandas)
+                index_col, columns = None, None
+                if "indexCol" in d:
+                    index_col = [
+                        col for col in expr.sp_to_snowpark_pandas.index_col.list
+                    ]
+                if "columns" in d:
+                    columns = [col for col in expr.sp_to_snowpark_pandas.columns.list]
+                # Returning the result of to_snowpark_pandas causes recursion issues when local_testing_mode is enabled.
+                # When disabled, to_snowpark_pandas will raise an error since df will be an empty Dataframe
+                # (passing non-None values of index_col or columns will make the snowpark_to_pandas_helper complain
+                # about columns that do not exist).
+                # Therefore, silently execute to_snowpark_pandas to record the AST and return None.
+                df.to_snowpark_pandas(index_col, columns)
+                return None
+
             case "udf":
                 return_type = self.decode_data_type_expr(expr.udf.return_type)
                 input_types = [