feat: read_gbq creates order deterministically without table copy (#191)

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

@@ -2719,7 +2719,8 @@ def _get_block(self) -> blocks.Block:
         return self._block
 
     def _cached(self) -> DataFrame:
-        return DataFrame(self._block.cached())
+        self._set_block(self._block.cached())
+        return self
 
     _DataFrameOrSeries = typing.TypeVar("_DataFrameOrSeries")
 

bigframes/ml/core.py

@@ -126,7 +126,7 @@ def generate_text_embedding(
 
     def forecast(self) -> bpd.DataFrame:
         sql = self._model_manipulation_sql_generator.ml_forecast()
-        return self._session.read_gbq(sql)
+        return self._session.read_gbq(sql, index_col="forecast_timestamp").reset_index()
 
     def evaluate(self, input_data: Optional[bpd.DataFrame] = None):
         # TODO: validate input data schema
@@ -139,14 +139,18 @@ def centroids(self) -> bpd.DataFrame:
 
         sql = self._model_manipulation_sql_generator.ml_centroids()
 
-        return self._session.read_gbq(sql)
+        return self._session.read_gbq(
+            sql, index_col=["centroid_id", "feature"]
+        ).reset_index()
 
     def principal_components(self) -> bpd.DataFrame:
         assert self._model.model_type == "PCA"
 
         sql = self._model_manipulation_sql_generator.ml_principal_components()
 
-        return self._session.read_gbq(sql)
+        return self._session.read_gbq(
+            sql, index_col=["principal_component_id", "feature"]
+        ).reset_index()
 
     def principal_component_info(self) -> bpd.DataFrame:
         assert self._model.model_type == "PCA"
@@ -228,10 +232,12 @@ def create_model(
         Returns: a BqmlModel, wrapping a trained model in BigQuery
         """
         options = dict(options)
+        # Cache dataframes to make sure base table is not a snapshot
+        # cached dataframe creates a full copy, never uses snapshot
         if y_train is None:
-            input_data = X_train
+            input_data = X_train._cached()
         else:
-            input_data = X_train.join(y_train, how="outer")
+            input_data = X_train._cached().join(y_train._cached(), how="outer")
             options.update({"INPUT_LABEL_COLS": y_train.columns.tolist()})
 
         session = X_train._session
@@ -259,7 +265,9 @@ def create_time_series_model(
         ), "Time stamp data input must only contain 1 column."
 
         options = dict(options)
-        input_data = X_train.join(y_train, how="outer")
+        # Cache dataframes to make sure base table is not a snapshot
+        # cached dataframe creates a full copy, never uses snapshot
+        input_data = X_train._cached().join(y_train._cached(), how="outer")
         options.update({"TIME_SERIES_TIMESTAMP_COL": X_train.columns.tolist()[0]})
         options.update({"TIME_SERIES_DATA_COL": y_train.columns.tolist()[0]})
 

bigframes/series.py

@@ -1503,7 +1503,8 @@ def _slice(
         )
 
     def _cached(self) -> Series:
-        return Series(self._block.cached())
+        self._set_block(self._block.cached())
+        return self
 
 
 def _is_list_like(obj: typing.Any) -> typing_extensions.TypeGuard[typing.Sequence]: