refactor: export operations use sesssion.execute path (#418)

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a [bug/issue](https://togithub.com/googleapis/python-bigquery-dataframes/issues/new/choose) before writing your code!  That way we can discuss the change, evaluate designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

Author: TrevorBergeron

bigframes/dataframe.py

@@ -2932,8 +2932,9 @@ def map_columns_on_occurrence(columns):
 
         return clustering_columns_for_index + clustering_columns_for_df
 
-    def _create_io_query(self, index: bool, ordering_id: Optional[str]) -> str:
-        """Create query text representing this dataframe for I/O."""
+    def _prepare_export(
+        self, index: bool, ordering_id: Optional[str]
+    ) -> Tuple[bigframes.core.ArrayValue, Dict[str, str]]:
         array_value = self._block.expr
 
         new_col_labels, new_idx_labels = utils.get_standardized_ids(
@@ -2961,10 +2962,7 @@ def _create_io_query(self, index: bool, ordering_id: Optional[str]) -> str:
 
         if ordering_id is not None:
             array_value = array_value.promote_offsets(ordering_id)
-        return self._block.session._to_sql(
-            array_value=array_value,
-            col_id_overrides=id_overrides,
-        )
+        return array_value, id_overrides
 
     def _run_io_query(
         self,
@@ -2974,11 +2972,16 @@ def _run_io_query(
     ) -> bigquery.TableReference:
         """Executes a query job presenting this dataframe and returns the destination
         table."""
-        expr = self._block.expr
-        session = expr.session
-        sql = self._create_io_query(index=index, ordering_id=ordering_id)
-        _, query_job = session._start_query(
-            sql=sql, job_config=job_config  # type: ignore
+        session = self._block.expr.session
+        export_array, id_overrides = self._prepare_export(
+            index=index, ordering_id=ordering_id
+        )
+
+        _, query_job = session._execute(
+            export_array,
+            job_config=job_config,
+            sorted=False,
+            col_id_overrides=id_overrides,
         )
         self._set_internal_query_job(query_job)
 

bigframes/session/__init__.py

@@ -1626,9 +1626,15 @@ def _execute(
         *,
         sorted: bool = True,
         dry_run=False,
+        col_id_overrides: Mapping[str, str] = {},
     ) -> tuple[bigquery.table.RowIterator, bigquery.QueryJob]:
-        sql = self._to_sql(array_value, sorted=sorted)  # type:ignore
-        job_config = bigquery.QueryJobConfig(dry_run=dry_run)
+        sql = self._to_sql(
+            array_value, sorted=sorted, col_id_overrides=col_id_overrides
+        )  # type:ignore
+        if job_config is None:
+            job_config = bigquery.QueryJobConfig(dry_run=dry_run)
+        else:
+            job_config.dry_run = dry_run
         return self._start_query(
             sql=sql,
             job_config=job_config,