feat: Add read_arrow methods to Session and pandas

Adds `read_arrow` methods to `bigframes.session.Session` and
`bigframes.pandas.read_arrow` for creating BigQuery DataFrames
DataFrames from PyArrow Tables.

The implementation refactors existing logic from
`bigframes.session._io.bigquery.read_gbq_query` for converting
Arrow data into BigFrames DataFrames.

Includes:
- New file `bigframes/session/_io/arrow.py` with the core conversion logic.
- `read_arrow(pa.Table) -> bpd.DataFrame` in `Session` class.
- `read_arrow(pa.Table) -> bpd.DataFrame` in `pandas` module.
- Unit and system tests for the new functionality.
- Docstrings for new methods/functions.

Note: Unit tests for direct DataFrame operations (shape, to_pandas) on
the result of read_arrow are currently failing due to the complexity of
mocking the session and executor for LocalDataNode interactions.
System tests are recommended for full end-to-end validation.

Author: google-labs-jules[bot]

bigframes/pandas/__init__.py

@@ -25,6 +25,7 @@
 
 import bigframes_vendored.pandas.core.tools.datetimes as vendored_pandas_datetimes
 import pandas
+import pyarrow as pa
 
 import bigframes._config as config
 from bigframes.core import log_adapter
@@ -54,6 +55,21 @@
 )
 import bigframes.series
 import bigframes.session
+
+
+def read_arrow(pa_table: pa.Table) -> bigframes.dataframe.DataFrame:
+    """Load a PyArrow Table to a BigQuery DataFrames DataFrame.
+
+    Args:
+        pa_table (pyarrow.Table):
+            PyArrow table to load data from.
+
+    Returns:
+        bigframes.dataframe.DataFrame:
+            A new DataFrame representing the data from the PyArrow table.
+    """
+    session = global_session.get_global_session()
+    return session.read_arrow(pa_table=pa_table)
 import bigframes.session._io.bigquery
 import bigframes.session.clients
 import bigframes.version

bigframes/session/_io/arrow.py

@@ -0,0 +1,80 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Private helpers for reading pyarrow objects."""
+
+from __future__ import annotations
+
+import pyarrow as pa
+
+from bigframes import dataframe
+import bigframes.core as core
+import bigframes.dtypes
+from bigframes.core import local_data, pyarrow_utils
+import bigframes.core.blocks as blocks
+import bigframes.core.guid
+import bigframes.core.schema as schemata
+import bigframes.session
+
+
+def create_dataframe_from_arrow_table(
+    pa_table: pa.Table, *, session: bigframes.session.Session
+) -> dataframe.DataFrame:
+    """Convert a PyArrow Table into a BigQuery DataFrames DataFrame.
+
+    This DataFrame will wrap a LocalNode, meaning the data is processed locally.
+
+    Args:
+        pa_table (pyarrow.Table):
+            The PyArrow Table to convert.
+        session (bigframes.session.Session):
+            The BigQuery DataFrames session to associate with the new DataFrame.
+
+    Returns:
+        bigframes.dataframe.DataFrame:
+            A new DataFrame representing the data from the PyArrow table.
+    """
+    # TODO(tswast): Use array_value.promote_offsets() instead once that node is
+    # supported by the local engine.
+    offsets_col = bigframes.core.guid.generate_guid()
+    # TODO(https://github.com/googleapis/python-bigquery-dataframes/issues/859):
+    # Allow users to specify the "total ordering" column(s) or allow multiple
+    # such columns.
+    pa_table = pyarrow_utils.append_offsets(pa_table, offsets_col=offsets_col)
+
+    # We use the ManagedArrowTable constructor directly, because the
+    # results of to_arrow() should be the source of truth with regards
+    # to canonical formats since it comes from either the BQ Storage
+    # Read API or has been transformed by google-cloud-bigquery to look
+    # like the output of the BQ Storage Read API.
+    schema_items = []
+    for field in pa_table.schema:
+        bf_dtype = bigframes.dtypes.arrow_dtype_to_bigframes_dtype(field.type, allow_lossless_cast=True)
+        schema_items.append(schemata.SchemaItem(field.name, bf_dtype))
+    bf_schema = schemata.ArraySchema(tuple(schema_items))
+
+    mat = local_data.ManagedArrowTable(
+        pa_table,
+        bf_schema,
+    )
+    mat.validate()
+
+    array_value = core.ArrayValue.from_managed(mat, session)
+    block = blocks.Block(
+        array_value,
+        (offsets_col,),
+        [field.name for field in pa_table.schema if field.name != offsets_col],
+        (None,),
+    )
+    return dataframe.DataFrame(block)

bigframes/session/_io/bigquery/read_gbq_query.py

@@ -23,12 +23,8 @@
 import pandas
 
 from bigframes import dataframe
-from bigframes.core import local_data, pyarrow_utils
-import bigframes.core as core
-import bigframes.core.blocks as blocks
-import bigframes.core.guid
-import bigframes.core.schema as schemata
 import bigframes.session
+from bigframes.session._io.arrow import create_dataframe_from_arrow_table
 
 
 def create_dataframe_from_query_job_stats(
@@ -61,30 +57,4 @@ def create_dataframe_from_row_iterator(
     'jobless' case where there's no destination table.
     """
     pa_table = rows.to_arrow()
-
-    # TODO(tswast): Use array_value.promote_offsets() instead once that node is
-    # supported by the local engine.
-    offsets_col = bigframes.core.guid.generate_guid()
-    pa_table = pyarrow_utils.append_offsets(pa_table, offsets_col=offsets_col)
-
-    # We use the ManagedArrowTable constructor directly, because the
-    # results of to_arrow() should be the source of truth with regards
-    # to canonical formats since it comes from either the BQ Storage
-    # Read API or has been transformed by google-cloud-bigquery to look
-    # like the output of the BQ Storage Read API.
-    mat = local_data.ManagedArrowTable(
-        pa_table,
-        schemata.ArraySchema.from_bq_schema(
-            list(rows.schema) + [bigquery.SchemaField(offsets_col, "INTEGER")]
-        ),
-    )
-    mat.validate()
-
-    array_value = core.ArrayValue.from_managed(mat, session)
-    block = blocks.Block(
-        array_value,
-        (offsets_col,),
-        [field.name for field in rows.schema],
-        (None,),
-    )
-    return dataframe.DataFrame(block)
+    return create_dataframe_from_arrow_table(pa_table, session=session)

bigframes/session/bigquery_session.py

@@ -21,9 +21,12 @@
 # TODO: Non-ibis implementation
 import bigframes_vendored.ibis.backends.bigquery.datatypes as ibis_bq
 import google.cloud.bigquery as bigquery
+import pyarrow as pa
 
+import bigframes.dataframe
 from bigframes.core.compile import googlesql
 from bigframes.session import temporary_storage
+from bigframes.session._io.arrow import create_dataframe_from_arrow_table
 
 KEEPALIVE_QUERY_TIMEOUT_SECONDS = 5.0
 
@@ -142,6 +145,19 @@ def _keep_session_alive(self):
             except Exception as e:
                 logging.warning("BigQuery session keep-alive query errored : %s", e)
 
+    def read_arrow(self, pa_table: pa.Table) -> bigframes.dataframe.DataFrame:
+        """Load a PyArrow Table to a BigQuery DataFrames DataFrame.
+
+        Args:
+            pa_table (pyarrow.Table):
+                PyArrow table to load data from.
+
+        Returns:
+            bigframes.dataframe.DataFrame:
+                A new DataFrame representing the data from the PyArrow table.
+        """
+        return create_dataframe_from_arrow_table(pa_table, session=self)
+
 
 class RecurringTaskDaemon:
     def __init__(self, task: Callable[[], None], frequency: datetime.timedelta):