Python connector changes for server-side Snowpark (#1964)

* Initial checkin for Phase 0 changes to the connector (#1963)

* Appended the required DESCRIPTION.md change describing this diff

* Removed the snippet from DESCRIPTION.md since our change isn't public

Author: sfc-gh-azwiegincew

src/snowflake/connector/connection.py

@@ -1287,6 +1287,7 @@ def cmd_query(
         _update_current_object: bool = True,
         _no_retry: bool = False,
         timeout: int | None = None,
+        dataframe_ast: str | None = None,
     ) -> dict[str, Any]:
         """Executes a query with a sequence counter."""
         logger.debug("_cmd_query")
@@ -1296,6 +1297,8 @@ def cmd_query(
             "sequenceId": sequence_counter,
             "querySubmissionTime": get_time_millis(),
         }
+        if dataframe_ast is not None:
+            data["dataframeAst"] = dataframe_ast
         if statement_params is not None:
             data["parameters"] = statement_params
         if is_internal:

src/snowflake/connector/cursor.py

@@ -601,6 +601,7 @@ def _execute_helper(
         _no_results: bool = False,
         _is_put_get=None,
         _no_retry: bool = False,
+        dataframe_ast: str | None = None,
     ) -> dict[str, Any]:
         del self.messages[:]
 
@@ -704,6 +705,7 @@ def interrupt_handler(*_):  # pragma: no cover
                 _no_results=_no_results,
                 _no_retry=_no_retry,
                 timeout=real_timeout,
+                dataframe_ast=dataframe_ast,
             )
         finally:
             try:
@@ -809,6 +811,7 @@ def execute(
         _skip_upload_on_content_match: bool = False,
         file_stream: IO[bytes] | None = None,
         num_statements: int | None = None,
+        _dataframe_ast: str | None = None,
     ) -> Self | None: ...
 
     @overload
@@ -838,6 +841,7 @@ def execute(
         _skip_upload_on_content_match: bool = False,
         file_stream: IO[bytes] | None = None,
         num_statements: int | None = None,
+        _dataframe_ast: str | None = None,
     ) -> dict[str, Any] | None: ...
 
     def execute(
@@ -866,6 +870,7 @@ def execute(
         _skip_upload_on_content_match: bool = False,
         file_stream: IO[bytes] | None = None,
         num_statements: int | None = None,
+        _dataframe_ast: str | None = None,
     ) -> Self | dict[str, Any] | None:
         """Executes a command/query.
 
@@ -900,6 +905,7 @@ def execute(
             file_stream: File-like object to be uploaded with PUT
             num_statements: Query level parameter submitted in _statement_params constraining exact number of
             statements being submitted (or 0 if submitting an uncounted number) when using a multi-statement query.
+            _dataframe_ast: Base64-encoded dataframe request abstract syntax tree.
 
         Returns:
             The cursor itself, or None if some error happened, or the response returned
@@ -941,6 +947,7 @@ def execute(
             "_no_results": _no_results,
             "_is_put_get": _is_put_get,
             "_no_retry": _no_retry,
+            "dataframe_ast": _dataframe_ast,
         }
 
         if self._connection.is_pyformat: