chore: add experimental bpd.from_glob_path method (#1230)

* chore: add experimental bpd.from_glob_path method

* fix mypy

* fix mypy

Author: GarrettWu

bigframes/pandas/__init__.py

@@ -36,6 +36,7 @@
 import bigframes.enums
 import bigframes.functions._utils as functions_utils
 from bigframes.pandas.io.api import (
+    from_glob_path,
     read_csv,
     read_gbq,
     read_gbq_function,
@@ -311,6 +312,7 @@ def reset_session():
     "read_pickle",
     "remote_function",
     "to_datetime",
+    "from_glob_path",
     # pandas dtype attributes
     "NA",
     "BooleanDtype",

bigframes/pandas/io/api.py

@@ -307,6 +307,20 @@ def read_gbq_function(function_name: str, is_row_processor: bool = False):
 read_gbq_function.__doc__ = inspect.getdoc(bigframes.session.Session.read_gbq_function)
 
 
+def from_glob_path(
+    path: str, *, connection: Optional[str] = None, name: Optional[str] = None
+) -> bigframes.dataframe.DataFrame:
+    return global_session.with_default_session(
+        bigframes.session.Session.from_glob_path,
+        path=path,
+        connection=connection,
+        name=name,
+    )
+
+
+from_glob_path.__doc__ = inspect.getdoc(bigframes.session.Session.from_glob_path)
+
+
 def _set_default_session_location_if_possible(query):
     # Set the location as per the query if this is the first query the user is
     # running and:

bigframes/session/__init__.py

@@ -1475,6 +1475,63 @@ def _start_query_ml_ddl(
             self.bqclient, sql, job_config, metrics=self._metrics
         )
 
+    def _create_object_table(self, path: str, connection: str) -> str:
+        """Create a random id Object Table from the input path and connection."""
+        table = str(self._loader._storage_manager._random_table())
+
+        import textwrap
+
+        sql = textwrap.dedent(
+            f"""
+            CREATE EXTERNAL TABLE `{table}`
+            WITH CONNECTION `{connection}`
+            OPTIONS(
+                object_metadata = 'SIMPLE',
+                uris = ['{path}']);
+            """
+        )
+        bf_io_bigquery.start_query_with_client(
+            self.bqclient,
+            sql,
+            job_config=bigquery.QueryJobConfig(),
+            metrics=self._metrics,
+        )
+
+        return table
+
+    def from_glob_path(
+        self, path: str, *, connection: Optional[str] = None, name: Optional[str] = None
+    ) -> dataframe.DataFrame:
+        r"""Create a BigFrames DataFrame that contains a BigFrames Blob column from a global wildcard path.
+
+        Args:
+            path (str):
+                The wildcard global path, such as "gs://<bucket>/<folder>/\*".
+            connection (str or None, default None):
+                Connection to connect with remote service. str of the format <PROJECT_NUMBER/PROJECT_ID>.<LOCATION>.<CONNECTION_ID>.
+                If None, use default connection in session context. BigQuery DataFrame will try to create the connection and attach
+                permission if the connection isn't fully set up.
+            name (str):
+                The column name of the Blob column.
+        Returns:
+            bigframes.pandas.DataFrame:
+                Result BigFrames DataFrame.
+        """
+        if not bigframes.options.experiments.blob:
+            raise NotImplementedError()
+
+        connection = connection or self._bq_connection
+        connection = bigframes.clients.resolve_full_bq_connection_name(
+            connection,
+            default_project=self._project,
+            default_location=self._location,
+        )
+
+        table = self._create_object_table(path, connection)
+
+        s = self.read_gbq(table)["uri"].str.to_blob(connection)
+        return s.rename(name).to_frame()
+
 
 def connect(context: Optional[bigquery_options.BigQueryOptions] = None) -> Session:
     return Session(context)