SNOW-1961756: Enable AST capture from Session.read.dbapi (#3134)

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

   [SNOW-1961756](https://snowflakecomputing.atlassian.net/browse/SNOW-1961756)

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.

   Propagate `_emit_ast=False` across all internal calls to public APIs for new `dbapi` functionality.

   Importantly, we will capture uses of `session.read.dbapi` as calls to `session.table` since we only support this functionality on the client side. Attempting to capture the call would mean creating an AST entity for `session.read.dbapi` specifically just to have it show up in query history, and this does not seem valuable from an AST or server-side execution perspective. 


[SNOW-1961756]: https://snowflakecomputing.atlassian.net/browse/SNOW-1961756?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ

Author: sfc-gh-heshah

src/snowflake/snowpark/dataframe_reader.py

@@ -458,7 +458,7 @@ def table(self, name: Union[str, Iterable[str]], _emit_ast: bool = True) -> Tabl
             ast.reader.CopyFrom(self._ast)
             build_table_name(ast.name, name)
 
-        table = self._session.table(name)
+        table = self._session.table(name, _emit_ast=False)
 
         if _emit_ast:
             table._ast_id = stmt.var_id.bitfield1
@@ -1087,6 +1087,7 @@ def _read_semi_structured_file(self, path: str, format: str) -> DataFrame:
         return df
 
     @private_preview(version="1.29.0")
+    @publicapi
     def dbapi(
         self,
         create_connection: Callable[[], "Connection"],
@@ -1103,6 +1104,7 @@ def dbapi(
         custom_schema: Optional[Union[str, StructType]] = None,
         predicates: Optional[List[str]] = None,
         session_init_statement: Optional[str] = None,
+        _emit_ast: bool = True,
     ) -> DataFrame:
         """Reads data from a database table using a DBAPI connection with optional partitioning, parallel processing, and query customization.
         By default, the function reads the entire table at a time without a query timeout.
@@ -1222,17 +1224,17 @@ def dbapi(
             )
             params = (snowflake_table_name,)
             logger.debug(f"Creating temporary Snowflake table: {snowflake_table_name}")
-            self._session.sql(create_table_sql, params=params).collect(
-                statement_params=statements_params_for_telemetry
+            self._session.sql(create_table_sql, params=params, _emit_ast=False).collect(
+                statement_params=statements_params_for_telemetry, _emit_ast=False
             )
             # create temp stage
             snowflake_stage_name = random_name_for_temp_object(TempObjectType.STAGE)
             sql_create_temp_stage = (
                 f"create {get_temp_type_for_object(self._session._use_scoped_temp_objects, True)} stage"
                 f" if not exists {snowflake_stage_name} {DATA_SOURCE_SQL_COMMENT}"
             )
-            self._session.sql(sql_create_temp_stage).collect(
-                statement_params=statements_params_for_telemetry
+            self._session.sql(sql_create_temp_stage, _emit_ast=False).collect(
+                statement_params=statements_params_for_telemetry, _emit_ast=False
             )
 
             try:
@@ -1327,7 +1329,11 @@ def ingestion_thread_cleanup_callback(parquet_file_path, _):
             self._session._conn._telemetry_client.send_data_source_perf_telemetry(
                 DATA_SOURCE_DBAPI_SIGNATURE, time.perf_counter() - start_time
             )
-            res_df = self.table(snowflake_table_name)
+            # Knowingly generating AST for `session.read.dbapi` calls as simply `session.read.table` calls
+            # with the new name for the temporary table into which the external db data was ingressed.
+            # Leaving this functionality as client-side only means capturing an AST specifically for
+            # this API in a new entity is not valuable from a server-side execution or AST perspective.
+            res_df = self.table(snowflake_table_name, _emit_ast=_emit_ast)
             set_api_call_source(res_df, DATA_SOURCE_DBAPI_SIGNATURE)
             return res_df
 
@@ -1476,9 +1482,11 @@ def _upload_and_copy_into_table(
         ON_ERROR={on_error}
         {DATA_SOURCE_SQL_COMMENT}
         """
-        self._session.sql(put_query).collect(statement_params=statements_params)
-        self._session.sql(copy_into_table_query).collect(
-            statement_params=statements_params
+        self._session.sql(put_query, _emit_ast=False).collect(
+            statement_params=statements_params, _emit_ast=False
+        )
+        self._session.sql(copy_into_table_query, _emit_ast=False).collect(
+            statement_params=statements_params, _emit_ast=False
         )
 
     def _upload_and_copy_into_table_with_retry(

src/snowflake/snowpark/dataframe_writer.py

@@ -337,6 +337,9 @@ def save_as_table(
         """
 
         kwargs = {}
+        statement_params = self._track_data_source_statement_params(
+            self._dataframe, statement_params or self._dataframe._statement_params
+        )
         if _emit_ast:
             # Add an Assign node that applies WriteTable() to the input, followed by its Eval.
             repr = self._dataframe._session._ast_batch.assign()
@@ -466,9 +469,6 @@ def save_as_table(
             else:
                 table_exists = None
 
-            statement_params = self._track_data_source_statement_params(
-                self._dataframe, statement_params or self._dataframe._statement_params
-            )
             create_table_logic_plan = SnowflakeCreateTable(
                 table_name,
                 column_names,
@@ -590,6 +590,9 @@ def copy_into_location(
         """
 
         kwargs = {}
+        statement_params = self._track_data_source_statement_params(
+            self._dataframe, statement_params or self._dataframe._statement_params
+        )
         if _emit_ast:
             # Add an Assign node that applies WriteCopyIntoLocation() to the input, followed by its Eval.
             repr = self._dataframe._session._ast_batch.assign()
@@ -649,10 +652,6 @@ def copy_into_location(
 
             cur_format_type_options.update(format_type_aliased_options)
 
-        statement_params = self._track_data_source_statement_params(
-            self._dataframe, statement_params or self._dataframe._statement_params
-        )
-
         df = self._dataframe._with_plan(
             CopyIntoLocationNode(
                 self._dataframe._plan,