SNOW-1830603 UDTFs for the decoder (#2930)

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

  Fixes SNOW-1830603

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.

- Got UDTFs to work
- changed the UDTF test (will prob go through another change in the
future and will make a change to this test in main then).
- got RelationalGroupedDataFrame to work
- added a way to reset the recorded callables and their IDs

Author: sfc-gh-vbudati

src/snowflake/snowpark/_internal/ast/batch.py

@@ -72,6 +72,10 @@ def reset_id_gen(self) -> None:
         """Resets the ID generator."""
         self._id_gen = itertools.count(start=1)
 
+    def reset_callables(self) -> None:
+        """Resets the callables."""
+        self._callables = {}
+
     def assign(self, symbol: Optional[str] = None) -> proto.Assign:
         """
         Creates a new assignment statement.

src/snowflake/snowpark/relational_grouped_dataframe.py

@@ -423,7 +423,7 @@ def end_partition(self, pdf: pandas.DataFrame) -> pandas.DataFrame:
         _apply_in_pandas_udtf = self._dataframe._session.udtf.register(
             _ApplyInPandas,
             output_schema=output_schema,
-            _emit_ast=_emit_ast,
+            _emit_ast=False,
             **kwargs,
         )
         partition_by = [Column(expr, _emit_ast=False) for expr in self._grouping_exprs]

src/snowflake/snowpark/types.py

@@ -784,7 +784,7 @@ def json_value(self) -> Dict[str, Any]:
     def _fill_ast(self, ast: proto.SpDataType) -> None:
         ast.sp_struct_type.structured = self.structured
         for field in self.fields:
-            field._fill_ast(ast.sp_struct_type.fields.add())
+            field._fill_ast(ast.sp_struct_type.fields.list.add())
 
 
 class VariantType(DataType):