From 7de4cdd622995167086581ea0adaad3b501f33e7 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 3 Jun 2026 21:27:55 +0000
Subject: [PATCH 01/18] fix(bigframes): avoid exceptions for unnamed JSON
 columns in SQL Cell outputs

---
 packages/bigframes/bigframes/core/indexers.py | 123 +++++++++-
 packages/bigframes/bigframes/dataframe.py     |  22 +-
 packages/bigframes/bigframes/series.py        |   1 -
 .../generative_ai/ai_functions.ipynb          |  44 ++--
 .../bigframes/tests/unit/test_iloc_setitem.py | 212 ++++++++++++++++++
 5 files changed, 374 insertions(+), 28 deletions(-)
 create mode 100644 packages/bigframes/tests/unit/test_iloc_setitem.py

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index c7cfc4f52ade..b3f8e4ce5639 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -28,6 +28,7 @@
 import bigframes.core.guid as guid
 import bigframes.core.indexes as indexes
 import bigframes.core.scalar
+import bigframes.core.validations as validations
 import bigframes.core.window_spec as windows
 import bigframes.dataframe
 import bigframes.dtypes
@@ -102,6 +103,18 @@ def __getitem__(
 
         Other key types are not yet supported.
         """
+        requires_ordering = True
+        if (
+            isinstance(key, slice)
+            and (key.start is None or key.start == 0)
+            and (key.step is None or key.step == 1)
+            and key.stop is None
+        ):
+            requires_ordering = False
+
+        if requires_ordering:
+            validations.enforce_ordered(self._series, "iloc")
+
         return _iloc_getitem_series_or_dataframe(self._series, key)
 
 
@@ -244,8 +257,113 @@ def __getitem__(self, key) -> Union[bigframes.dataframe.DataFrame, pd.Series]:
 
         Other key types are not yet supported.
         """
+        requires_ordering = True
+        if isinstance(key, tuple):
+            if len(key) > 0:
+                row_indexer = key[0]
+                if (
+                    isinstance(row_indexer, slice)
+                    and (row_indexer.start is None or row_indexer.start == 0)
+                    and (row_indexer.step is None or row_indexer.step == 1)
+                    and row_indexer.stop is None
+                ):
+                    requires_ordering = False
+        else:
+            if (
+                isinstance(key, slice)
+                and (key.start is None or key.start == 0)
+                and (key.step is None or key.step == 1)
+                and key.stop is None
+            ):
+                requires_ordering = False
+
+        if requires_ordering:
+            validations.enforce_ordered(self._dataframe, "iloc")
+
         return _iloc_getitem_series_or_dataframe(self._dataframe, key)
 
+    def __setitem__(
+        self,
+        key: Tuple[
+            slice, Union[int, typing.Sequence[int], slice, typing.Sequence[bool]]
+        ],
+        value: Union[
+            bigframes.dataframe.SingleItemValue, bigframes.dataframe.DataFrame
+        ],
+    ):
+        if not (
+            isinstance(key, tuple)
+            and len(key) == 2
+            and isinstance(key[0], slice)
+            and (key[0].start is None or key[0].start == 0)
+            and (key[0].step is None or key[0].step == 1)
+            and key[0].stop is None
+        ):
+            raise NotImplementedError(
+                "Only DataFrame.iloc[:, col_indexer] = value is supported."
+            )
+
+        col_indexer = key[1]
+        n_cols = len(self._dataframe.columns)
+
+        if isinstance(col_indexer, bool):
+            raise TypeError(
+                "pos must be integer or slice or list-like of integers/booleans"
+            )
+
+        if isinstance(col_indexer, int):
+            col_offset = col_indexer
+            if col_offset < 0:
+                col_offset += n_cols
+            if col_offset < 0 or col_offset >= n_cols:
+                raise IndexError("single positional indexer is out-of-bounds")
+
+            col_label = self._dataframe.columns[col_offset]
+            df = self._dataframe.assign(**{col_label: value})
+            self._dataframe._set_block(df._get_block())
+
+        elif isinstance(col_indexer, slice):
+            col_offsets = list(range(*col_indexer.indices(n_cols)))
+            col_labels = [self._dataframe.columns[idx] for idx in col_offsets]
+            if not col_labels:
+                return
+            df = self._dataframe._assign_multi_items(col_labels, value)
+            self._dataframe._set_block(df._get_block())
+
+        elif pd.api.types.is_list_like(col_indexer):
+            col_indexer_list = list(col_indexer)
+
+            if len(col_indexer_list) > 0 and all(
+                isinstance(x, bool) for x in col_indexer_list
+            ):
+                if len(col_indexer_list) != n_cols:
+                    raise ValueError(
+                        f"Boolean index has wrong length: {len(col_indexer_list)} instead of {n_cols}"
+                    )
+                col_offsets = [i for i, val in enumerate(col_indexer_list) if val]
+            else:
+                col_offsets = []
+                for idx in col_indexer_list:
+                    if isinstance(idx, bool):
+                        raise TypeError("pos list must contain only integers")
+                    if not isinstance(idx, int):
+                        raise TypeError("pos list must contain only integers")
+                    if idx < 0:
+                        idx += n_cols
+                    if idx < 0 or idx >= n_cols:
+                        raise IndexError("positional indexer is out-of-bounds")
+                    col_offsets.append(idx)
+
+            col_labels = [self._dataframe.columns[idx] for idx in col_offsets]
+            if not col_labels:
+                return
+            df = self._dataframe._assign_multi_items(col_labels, value)
+            self._dataframe._set_block(df._get_block())
+        else:
+            raise TypeError(
+                "pos must be integer or slice or list-like of integers/booleans"
+            )
+
 
 class IatDataFrameIndexer:
     def __init__(self, dataframe: bigframes.dataframe.DataFrame):
@@ -470,8 +588,11 @@ def _iloc_getitem_series_or_dataframe(
 
         # len(key) == 2
         df = typing.cast(bigframes.dataframe.DataFrame, series_or_dataframe)
-        if isinstance(key[1], int):
+        if isinstance(key[0], int) and isinstance(key[1], int):
             return df.iat[key]
+        elif isinstance(key[1], int):
+            col_label = df.columns[key[1]]
+            return df[col_label].iloc[key[0]]
         elif isinstance(key[1], list):
             columns = df.columns[key[1]]
             return _iloc_getitem_series_or_dataframe(df[columns], key[0])
diff --git a/packages/bigframes/bigframes/dataframe.py b/packages/bigframes/bigframes/dataframe.py
index 6b7922fe9753..14e064b4c942 100644
--- a/packages/bigframes/bigframes/dataframe.py
+++ b/packages/bigframes/bigframes/dataframe.py
@@ -317,7 +317,6 @@ def loc(self) -> indexers.LocDataFrameIndexer:
         return indexers.LocDataFrameIndexer(self)
 
     @property
-    @validations.requires_ordering()
     def iloc(self) -> indexers.ILocDataFrameIndexer:
         return indexers.ILocDataFrameIndexer(self)
 
@@ -821,22 +820,25 @@ def __repr__(self) -> str:
 
     def _get_display_df(self) -> DataFrame:
         """Process ObjectRef and JSON/nested JSON columns for display."""
+        import bigframes.bigquery as bbq
+
         df = self
         # Arrow/Pandas to_pandas_batches does not support raw JSON/nested JSON
         # columns. Pre-serialize them to string format to bypass this limit.
         # Using TO_JSON_STRING via SqlScalarOp handles complex nested STRUCT
-        # types correctly.
-        json_cols = [
-            col
-            for col in df.columns
+        # types correctly. Use the offset so that we can handle duplicate and
+        # non-string column names.
+        json_col_indexes = [
+            col_index
+            for col_index, col in enumerate(df.columns)
             if bigframes.dtypes.contains_db_dtypes_json_dtype(df[col].dtype)
         ]
-        if json_cols:
-            op = ops.SqlScalarOp(
-                _output_type=bigframes.dtypes.STRING_DTYPE,
-                sql_template="TO_JSON_STRING({0})",
+        if json_col_indexes:
+            df._block.apply_analytic
+            df.iloc[:, json_col_indexes] = cast(
+                DataFrame,
+                df.iloc[:, json_col_indexes].apply(bbq.to_json_string),  # type: ignore
             )
-            df = df.assign(**{col: df[col]._apply_unary_op(op) for col in json_cols})
         return df
 
     def _repr_mimebundle_(self, include=None, exclude=None):
diff --git a/packages/bigframes/bigframes/series.py b/packages/bigframes/bigframes/series.py
index 60acad0c301f..f4985010b925 100644
--- a/packages/bigframes/bigframes/series.py
+++ b/packages/bigframes/bigframes/series.py
@@ -243,7 +243,6 @@ def loc(self) -> bigframes.core.indexers.LocSeriesIndexer:
         return bigframes.core.indexers.LocSeriesIndexer(self)
 
     @property
-    @validations.requires_ordering()
     def iloc(self) -> bigframes.core.indexers.IlocSeriesIndexer:
         return bigframes.core.indexers.IlocSeriesIndexer(self)
 
diff --git a/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb b/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
index 13234414df5e..dbb044d3943f 100644
--- a/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
+++ b/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
@@ -71,14 +71,14 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 2,
    "id": "c9f924aa",
    "metadata": {},
    "outputs": [],
    "source": [
-    "import bigframes.pandas as bpd \n",
+    "import bigframes.pandas as bpd\n",
     "\n",
-    "PROJECT_ID = \"\" # @param {type:\"string\"}\n",
+    "PROJECT_ID = \"bigframes-dev\" # @param {type:\"string\"}\n",
     "\n",
     "bpd.options.bigquery.project = PROJECT_ID\n",
     "bpd.options.bigquery.ordering_mode = \"partial\"\n",
@@ -105,16 +105,24 @@
      "name": "stderr",
      "output_type": "stream",
      "text": [
-      "/usr/local/google/home/sycai/src/python-bigquery-dataframes/bigframes/core/global_session.py:103: DefaultLocationWarning: No explicit location is set, so using location US for the session.\n",
-      "  _global_session = bigframes.session.connect(\n"
+      "/usr/local/google/home/swast/src/github.com/googleapis/google-cloud-python/packages/bigframes/bigframes/core/global_session.py:113: DefaultLocationWarning: No explicit location is set, so using location US for the session.\n",
+      "  _global_session = bigframes.session.connect(\n",
+      "/usr/local/google/home/swast/src/github.com/googleapis/google-cloud-python/packages/bigframes/bigframes/dtypes.py:1044: JSONDtypeWarning: JSON columns will be represented as pandas.ArrowDtype(pyarrow.json_())\n",
+      "instead of using `db_dtypes` in the future when available in pandas\n",
+      "(https://github.com/pandas-dev/pandas/issues/60958) and pyarrow.\n",
+      "  warnings.warn(msg, bigframes.exceptions.JSONDtypeWarning)\n"
      ]
     },
     {
      "data": {
+      "text/html": [
+       "<pre>0    {\"result\":\"Salad\",\"full_response\":{\"candidates...\n",
+       "1    {\"result\":\"Hotdog\",\"full_response\":{\"candidate...</pre>"
+      ],
       "text/plain": [
-       "0    {'result': 'Salad\\n', 'full_response': '{\"cand...\n",
-       "1    {'result': 'Sausageroll\\n', 'full_response': '...\n",
-       "dtype: struct<result: string, full_response: extension<dbjson<JSONArrowType>>, status: string>[pyarrow]"
+       "0    {\"result\":\"Salad\",\"full_response\":{\"candidates...\n",
+       "1    {\"result\":\"Hotdog\",\"full_response\":{\"candidate...\n",
+       "Name: 0, dtype: string"
       ]
      },
      "execution_count": 3,
@@ -156,9 +164,13 @@
    "outputs": [
     {
      "data": {
+      "text/html": [
+       "<pre>0    <NA>\n",
+       "1    <NA></pre>"
+      ],
       "text/plain": [
-       "0     Lettuce\n",
-       "1    The food\n",
+       "0    <NA>\n",
+       "1    <NA>\n",
        "Name: result, dtype: string"
       ]
      },
@@ -327,7 +339,7 @@
        "    <tr>\n",
        "      <th>0</th>\n",
        "      <td>tiger</td>\n",
-       "      <td>8.0</td>\n",
+       "      <td>7.0</td>\n",
        "    </tr>\n",
        "    <tr>\n",
        "      <th>2</th>\n",
@@ -342,7 +354,7 @@
       "text/plain": [
        "      animals  relative_weight\n",
        "1      spider              1.0\n",
-       "0       tiger              8.0\n",
+       "0       tiger              7.0\n",
        "2  blue whale             10.0\n",
        "\n",
        "[3 rows x 2 columns]"
@@ -465,7 +477,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 9,
+   "execution_count": 8,
    "id": "2e66110a",
    "metadata": {},
    "outputs": [
@@ -518,7 +530,7 @@
        "[2 rows x 2 columns]"
       ]
      },
-     "execution_count": 9,
+     "execution_count": 8,
      "metadata": {},
      "output_type": "execute_result"
     }
@@ -533,7 +545,7 @@
  ],
  "metadata": {
   "kernelspec": {
-   "display_name": "venv (3.10.17)",
+   "display_name": "venv",
    "language": "python",
    "name": "python3"
   },
@@ -547,7 +559,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.10.17"
+   "version": "3.14.3"
   }
  },
  "nbformat": 4,
diff --git a/packages/bigframes/tests/unit/test_iloc_setitem.py b/packages/bigframes/tests/unit/test_iloc_setitem.py
new file mode 100644
index 000000000000..66b9bf9e3086
--- /dev/null
+++ b/packages/bigframes/tests/unit/test_iloc_setitem.py
@@ -0,0 +1,212 @@
+# Copyright 2026 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.
+
+from typing import Generator
+
+import numpy as np
+import pandas as pd
+import pytest
+
+import bigframes
+import bigframes.pandas as bpd
+from bigframes.testing.utils import assert_frame_equal, assert_series_equal
+
+pytest.importorskip("polars")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def session() -> Generator[bigframes.Session, None, None]:
+    import bigframes.core.global_session
+    from bigframes.testing import polars_session
+
+    session = polars_session.TestSession()
+    with bigframes.core.global_session._GlobalSessionContext(session):
+        yield session
+
+
+@pytest.fixture
+def sample_df() -> bpd.DataFrame:
+    pd_df = pd.DataFrame(
+        {
+            "A": [1, 2, 3],
+            "B": [4, 5, 6],
+            "C": [7, 8, 9],
+        }
+    )
+    return bpd.read_pandas(pd_df)
+
+
+def test_iloc_setitem_single_integer(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    bf_df.iloc[:, 1] = 99
+    pd_df.iloc[:, 1] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_single_integer_negative(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    bf_df.iloc[:, -1] = 99
+    pd_df.iloc[:, -1] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_list_integer(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    bf_df.iloc[:, [0, 2]] = [99, 88]
+    pd_df.iloc[:, [0, 2]] = [99, 88]
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_slice(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    bf_df.iloc[:, 0:2] = 99
+    pd_df.iloc[:, 0:2] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_boolean_mask(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    mask = [True, False, True]
+    bf_df.iloc[:, mask] = 99
+    pd_df.iloc[:, np.array(mask)] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_dataframe(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    value_df = bpd.DataFrame({"B": [99, 88, 77], "C": [66, 55, 44]})
+    bf_df.iloc[:, 1:3] = value_df
+    pd_df.iloc[:, 1:3] = value_df.to_pandas()
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_getitem_single_integer(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, 1].to_pandas()
+    pd_result = pd_df.iloc[:, 1]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_unordered(sample_df):
+    session = sample_df._session
+    original_strictly_ordered = session._strictly_ordered
+    original_allow_ambiguity = session._allow_ambiguity
+
+    try:
+        session._strictly_ordered = False
+        session._allow_ambiguity = True
+
+        import unittest.mock as mock
+
+        with (
+            mock.patch.object(
+                type(sample_df._block.expr),
+                "order_ambiguous",
+                new_callable=mock.PropertyMock,
+            ) as mock_ambiguous,
+            mock.patch.object(
+                type(sample_df._block),
+                "explicitly_ordered",
+                new_callable=mock.PropertyMock,
+            ) as mock_explicit,
+        ):
+            mock_ambiguous.return_value = True
+            mock_explicit.return_value = False
+
+            # 1. Column indexing only - should NOT raise
+            try:
+                sample_df.iloc[:, 1]
+            except bigframes.exceptions.OrderRequiredError:
+                pytest.fail("iloc[:, col] raised OrderRequiredError unexpectedly!")
+
+            # 1b. Column indexing with slice(0, None) (NOT exactly `:` but fine) - should NOT raise
+            try:
+                sample_df.iloc[slice(0, None), 1]
+            except bigframes.exceptions.OrderRequiredError:
+                pytest.fail("iloc[0:, col] raised OrderRequiredError unexpectedly!")
+
+            # 1c. Column indexing with slice(None, None, 1) (NOT exactly `:` but fine) - should NOT raise
+            try:
+                sample_df.iloc[slice(None, None, 1), 1]
+            except bigframes.exceptions.OrderRequiredError:
+                pytest.fail("iloc[::1, col] raised OrderRequiredError unexpectedly!")
+
+            # 1d. Column indexing with slice(1, None) (row subset) - should RAISE
+            with pytest.raises(bigframes.exceptions.OrderRequiredError):
+                sample_df.iloc[slice(1, None), 1]
+
+            # 1e. Column indexing with slice(None, 2) (row subset) - should RAISE
+            with pytest.raises(bigframes.exceptions.OrderRequiredError):
+                sample_df.iloc[slice(None, 2), 1]
+
+            # 2. Column setitem only - should NOT raise
+            try:
+                bf_df = sample_df.copy()
+                bf_df.iloc[:, 1] = 99
+            except bigframes.exceptions.OrderRequiredError:
+                pytest.fail(
+                    "iloc[:, col] = val raised OrderRequiredError unexpectedly!"
+                )
+
+            # 3. Row indexing - should RAISE
+            with pytest.raises(bigframes.exceptions.OrderRequiredError):
+                sample_df.iloc[1, :]
+
+            # 4. Single indexer (row indexing) - should RAISE
+            with pytest.raises(bigframes.exceptions.OrderRequiredError):
+                sample_df.iloc[1]
+
+    finally:
+        session._strictly_ordered = original_strictly_ordered
+        session._allow_ambiguity = original_allow_ambiguity
+
+
+def test_iloc_setitem_errors(sample_df):
+    bf_df = sample_df.copy()
+
+    # Out of bounds
+    with pytest.raises(IndexError):
+        bf_df.iloc[:, 3] = 99
+
+    with pytest.raises(IndexError):
+        bf_df.iloc[:, -4] = 99
+
+    # Invalid key type (not slice(None) for rows)
+    with pytest.raises(NotImplementedError):
+        bf_df.iloc[0, 1] = 99
+
+    # Invalid col indexer type
+    with pytest.raises(TypeError):
+        bf_df.iloc[:, "B"] = 99

From c8548bbdbfa5cadc59630b3cbb89ca2487966c2b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a=20=28Swast=29?= <swast@google.com>
Date: Wed, 8 Jul 2026 10:31:49 -0500
Subject: [PATCH 02/18] Apply suggestions from code review

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
---
 packages/bigframes/bigframes/core/indexers.py                 | 2 +-
 packages/bigframes/notebooks/generative_ai/ai_functions.ipynb | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index b3f8e4ce5639..6b4fb0c0abf6 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -319,7 +319,7 @@ def __setitem__(
                 raise IndexError("single positional indexer is out-of-bounds")
 
             col_label = self._dataframe.columns[col_offset]
-            df = self._dataframe.assign(**{col_label: value})
+            df = self._dataframe._assign_multi_items([col_label], value)
             self._dataframe._set_block(df._get_block())
 
         elif isinstance(col_indexer, slice):
diff --git a/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb b/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
index dbb044d3943f..beaf12a64e39 100644
--- a/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
+++ b/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
@@ -78,7 +78,7 @@
    "source": [
     "import bigframes.pandas as bpd\n",
     "\n",
-    "PROJECT_ID = \"bigframes-dev\" # @param {type:\"string\"}\n",
+    "PROJECT_ID = \"\" # @param {type:\"string\"}\n",
     "\n",
     "bpd.options.bigquery.project = PROJECT_ID\n",
     "bpd.options.bigquery.ordering_mode = \"partial\"\n",

From 30577dc9661fd572e0d6848a1ef01a595d1f8ee4 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 8 Jul 2026 15:33:33 +0000
Subject: [PATCH 03/18] remove useless expression

---
 packages/bigframes/bigframes/dataframe.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/packages/bigframes/bigframes/dataframe.py b/packages/bigframes/bigframes/dataframe.py
index b2b96e4790fd..d7f9187dd2eb 100644
--- a/packages/bigframes/bigframes/dataframe.py
+++ b/packages/bigframes/bigframes/dataframe.py
@@ -858,7 +858,6 @@ def _prepare_display_df(self) -> DataFrame:
             if bigframes.dtypes.contains_db_dtypes_json_dtype(df[col].dtype)
         ]
         if json_col_indexes:
-            df._block.apply_analytic
             df.iloc[:, json_col_indexes] = cast(
                 DataFrame,
                 df.iloc[:, json_col_indexes].apply(bbq.to_json_string),  # type: ignore

From a193555f448db2048fceaed26c0acf72bb1b11ba Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 8 Jul 2026 15:44:18 +0000
Subject: [PATCH 04/18] parametrize tests

---
 .../bigframes/tests/unit/test_iloc_setitem.py | 135 +++++++++---------
 1 file changed, 68 insertions(+), 67 deletions(-)

diff --git a/packages/bigframes/tests/unit/test_iloc_setitem.py b/packages/bigframes/tests/unit/test_iloc_setitem.py
index 66b9bf9e3086..f84c0c06918d 100644
--- a/packages/bigframes/tests/unit/test_iloc_setitem.py
+++ b/packages/bigframes/tests/unit/test_iloc_setitem.py
@@ -47,7 +47,7 @@ def sample_df() -> bpd.DataFrame:
     return bpd.read_pandas(pd_df)
 
 
-def test_iloc_setitem_single_integer(sample_df):
+def test_iloc_setitem_column_single_integer(sample_df):
     bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
@@ -57,7 +57,7 @@ def test_iloc_setitem_single_integer(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_setitem_single_integer_negative(sample_df):
+def test_iloc_setitem_column_single_integer_negative(sample_df):
     bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
@@ -67,7 +67,7 @@ def test_iloc_setitem_single_integer_negative(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_setitem_list_integer(sample_df):
+def test_iloc_setitem_columns_list_integer(sample_df):
     bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
@@ -77,7 +77,7 @@ def test_iloc_setitem_list_integer(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_setitem_slice(sample_df):
+def test_iloc_setitem_columns_slice(sample_df):
     bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
@@ -87,7 +87,7 @@ def test_iloc_setitem_slice(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_setitem_boolean_mask(sample_df):
+def test_iloc_setitem_columns_boolean_mask(sample_df):
     bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
@@ -98,7 +98,7 @@ def test_iloc_setitem_boolean_mask(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_setitem_dataframe(sample_df):
+def test_iloc_setitem_columns_dataframe(sample_df):
     bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
@@ -109,7 +109,7 @@ def test_iloc_setitem_dataframe(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_getitem_single_integer(sample_df):
+def test_iloc_getitem_column_single_integer(sample_df):
     bf_df = sample_df
     pd_df = sample_df.to_pandas()
 
@@ -119,7 +119,8 @@ def test_iloc_getitem_single_integer(sample_df):
     assert_series_equal(bf_result, pd_result)
 
 
-def test_iloc_getitem_unordered(sample_df):
+@pytest.fixture
+def unordered_sample_df(sample_df: bpd.DataFrame) -> Generator[bpd.DataFrame, None, None]:
     session = sample_df._session
     original_strictly_ordered = session._strictly_ordered
     original_allow_ambiguity = session._allow_ambiguity
@@ -144,69 +145,69 @@ def test_iloc_getitem_unordered(sample_df):
         ):
             mock_ambiguous.return_value = True
             mock_explicit.return_value = False
-
-            # 1. Column indexing only - should NOT raise
-            try:
-                sample_df.iloc[:, 1]
-            except bigframes.exceptions.OrderRequiredError:
-                pytest.fail("iloc[:, col] raised OrderRequiredError unexpectedly!")
-
-            # 1b. Column indexing with slice(0, None) (NOT exactly `:` but fine) - should NOT raise
-            try:
-                sample_df.iloc[slice(0, None), 1]
-            except bigframes.exceptions.OrderRequiredError:
-                pytest.fail("iloc[0:, col] raised OrderRequiredError unexpectedly!")
-
-            # 1c. Column indexing with slice(None, None, 1) (NOT exactly `:` but fine) - should NOT raise
-            try:
-                sample_df.iloc[slice(None, None, 1), 1]
-            except bigframes.exceptions.OrderRequiredError:
-                pytest.fail("iloc[::1, col] raised OrderRequiredError unexpectedly!")
-
-            # 1d. Column indexing with slice(1, None) (row subset) - should RAISE
-            with pytest.raises(bigframes.exceptions.OrderRequiredError):
-                sample_df.iloc[slice(1, None), 1]
-
-            # 1e. Column indexing with slice(None, 2) (row subset) - should RAISE
-            with pytest.raises(bigframes.exceptions.OrderRequiredError):
-                sample_df.iloc[slice(None, 2), 1]
-
-            # 2. Column setitem only - should NOT raise
-            try:
-                bf_df = sample_df.copy()
-                bf_df.iloc[:, 1] = 99
-            except bigframes.exceptions.OrderRequiredError:
-                pytest.fail(
-                    "iloc[:, col] = val raised OrderRequiredError unexpectedly!"
-                )
-
-            # 3. Row indexing - should RAISE
-            with pytest.raises(bigframes.exceptions.OrderRequiredError):
-                sample_df.iloc[1, :]
-
-            # 4. Single indexer (row indexing) - should RAISE
-            with pytest.raises(bigframes.exceptions.OrderRequiredError):
-                sample_df.iloc[1]
-
+            yield sample_df
     finally:
         session._strictly_ordered = original_strictly_ordered
         session._allow_ambiguity = original_allow_ambiguity
 
 
-def test_iloc_setitem_errors(sample_df):
+@pytest.mark.parametrize(
+    ["key", "value", "expected_error"],
+    [
+        pytest.param((slice(None), 1), None, None, id="col_index"),
+        pytest.param((slice(0, None), 1), None, None, id="col_index_slice_0_none"),
+        pytest.param(
+            (slice(None, None, 1), 1), None, None, id="col_index_slice_none_none_1"
+        ),
+        pytest.param(
+            (slice(1, None), 1),
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="col_index_slice_1_none",
+        ),
+        pytest.param(
+            (slice(None, 2), 1),
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="col_index_slice_none_2",
+        ),
+        pytest.param((slice(None), 1), 99, None, id="col_setitem"),
+        pytest.param(
+            (1, slice(None)),
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="row_index_slice",
+        ),
+        pytest.param(
+            1,
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="single_row_index",
+        ),
+    ],
+)
+def test_iloc_getitem_unordered(unordered_sample_df, key, value, expected_error):
+    if value is not None:
+        bf_df = unordered_sample_df.copy()
+        bf_df.iloc[key] = value
+    elif expected_error is not None:
+        with pytest.raises(expected_error):
+            unordered_sample_df.iloc[key]
+    else:
+        unordered_sample_df.iloc[key]
+
+
+@pytest.mark.parametrize(
+    ["key", "expected_error"],
+    [
+        pytest.param((slice(None), 3), IndexError, id="out_of_bounds_positive"),
+        pytest.param((slice(None), -4), IndexError, id="out_of_bounds_negative"),
+        pytest.param((0, 1), NotImplementedError, id="invalid_row_indexer"),
+        pytest.param((slice(None), "B"), TypeError, id="invalid_col_indexer_type"),
+    ],
+)
+def test_iloc_setitem_column_errors(sample_df, key, expected_error):
     bf_df = sample_df.copy()
 
-    # Out of bounds
-    with pytest.raises(IndexError):
-        bf_df.iloc[:, 3] = 99
-
-    with pytest.raises(IndexError):
-        bf_df.iloc[:, -4] = 99
-
-    # Invalid key type (not slice(None) for rows)
-    with pytest.raises(NotImplementedError):
-        bf_df.iloc[0, 1] = 99
-
-    # Invalid col indexer type
-    with pytest.raises(TypeError):
-        bf_df.iloc[:, "B"] = 99
+    with pytest.raises(expected_error):
+        bf_df.iloc[key] = 99

From 4c8acf3190ee39751040ffe3b2419162d3ad3bbd Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 8 Jul 2026 16:11:56 +0000
Subject: [PATCH 05/18] refactor to use offsets

---
 packages/bigframes/bigframes/core/indexers.py |  13 +-
 packages/bigframes/bigframes/dataframe.py     | 158 +++++++++++++++---
 .../bigframes/tests/unit/test_iloc_setitem.py |  43 ++++-
 3 files changed, 180 insertions(+), 34 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index 6b4fb0c0abf6..f0cae0542b23 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -318,16 +318,14 @@ def __setitem__(
             if col_offset < 0 or col_offset >= n_cols:
                 raise IndexError("single positional indexer is out-of-bounds")
 
-            col_label = self._dataframe.columns[col_offset]
-            df = self._dataframe._assign_multi_items([col_label], value)
+            df = self._dataframe._assign_multi_items_by_offsets([col_offset], value)
             self._dataframe._set_block(df._get_block())
 
         elif isinstance(col_indexer, slice):
             col_offsets = list(range(*col_indexer.indices(n_cols)))
-            col_labels = [self._dataframe.columns[idx] for idx in col_offsets]
-            if not col_labels:
+            if not col_offsets:
                 return
-            df = self._dataframe._assign_multi_items(col_labels, value)
+            df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
             self._dataframe._set_block(df._get_block())
 
         elif pd.api.types.is_list_like(col_indexer):
@@ -354,10 +352,9 @@ def __setitem__(
                         raise IndexError("positional indexer is out-of-bounds")
                     col_offsets.append(idx)
 
-            col_labels = [self._dataframe.columns[idx] for idx in col_offsets]
-            if not col_labels:
+            if not col_offsets:
                 return
-            df = self._dataframe._assign_multi_items(col_labels, value)
+            df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
             self._dataframe._set_block(df._get_block())
         else:
             raise TypeError(
diff --git a/packages/bigframes/bigframes/dataframe.py b/packages/bigframes/bigframes/dataframe.py
index d7f9187dd2eb..786f5cd75f6e 100644
--- a/packages/bigframes/bigframes/dataframe.py
+++ b/packages/bigframes/bigframes/dataframe.py
@@ -52,7 +52,6 @@
 import google.cloud.bigquery.table
 import numpy
 import pandas
-import pandas.io.formats.format
 import pyarrow
 import tabulate
 from pandas.api import extensions as pd_ext
@@ -2244,10 +2243,47 @@ def _assign_single_item(
         else:
             return self._assign_scalar(k, v)  # type: ignore
 
-    def _assign_multi_items(
+    def _assign_single_item_by_offset(
         self,
-        k: list[str] | pandas.Index,
+        offset: int,
+        value: SingleItemValue | MultiItemValue,
+    ) -> DataFrame:
+        if isinstance(value, bigframes.series.Series):
+            return self._assign_series_join_on_index_by_offset(offset, value)
+        elif isinstance(value, bigframes.core.col.Expression):
+            label_to_col_ref = {
+                label: ex.deref(id) for id, label in self._block.col_id_to_label.items()
+            }
+            resolved_expr = value._value.bind_variables(label_to_col_ref)
+            block, (new_col_id,) = self._block.project_expr(resolved_expr)
+            target_col_id = self._block.value_columns[offset]
+            block = block.copy_values(new_col_id, target_col_id).drop_columns(
+                [new_col_id]
+            )
+            return DataFrame(block)
+        elif isinstance(value, DataFrame):
+            v_df_col_count = len(value._block.value_columns)
+            if v_df_col_count != 1:
+                raise ValueError(
+                    f"Cannot set a DataFrame with {v_df_col_count} columns to the single column at offset {offset}"
+                )
+            return self._assign_series_join_on_index_by_offset(
+                offset, cast(bigframes.series.Series, value[value.columns[0]])
+            )
+        elif callable(value):
+            raise NotImplementedError(
+                "Callable assignment is not supported by column offset."
+            )
+        elif utils.is_list_like(value):
+            return self._assign_single_item_listlike_by_offset(offset, value)
+        else:
+            return self._assign_scalar_by_offset(offset, value)  # type: ignore
+
+    def _assign_multi_items_helper(
+        self,
+        k: Sequence[Any],
         v: SingleItemValue | MultiItemValue,
+        assign_single_fn: Callable[[DataFrame, Any, Any], DataFrame],
     ) -> DataFrame:
         value_sources: Sequence[Any] = []
         if isinstance(v, DataFrame):
@@ -2265,13 +2301,35 @@ def _assign_multi_items(
             raise ValueError("Columns must be same length as key")
 
         # Repeatedly assign columns in order.
-        result = self._assign_single_item(k[0], value_sources[0])
+        result = assign_single_fn(self, k[0], value_sources[0])
         for target, source in zip(k[1:], value_sources[1:]):
-            result = result._assign_single_item(target, source)
+            result = assign_single_fn(result, target, source)
         return result
 
-    def _assign_single_item_listlike(self, k: str, v: Sequence) -> DataFrame:
-        given_rows = len(v)
+    def _assign_multi_items(
+        self,
+        k: list[str] | pandas.Index,
+        v: SingleItemValue | MultiItemValue,
+    ) -> DataFrame:
+        return self._assign_multi_items_helper(k, v, DataFrame._assign_single_item)
+
+    def _assign_multi_items_by_offsets(
+        self,
+        k: Sequence[int],
+        v: SingleItemValue | MultiItemValue,
+    ) -> DataFrame:
+        return self._assign_multi_items_helper(
+            k, v, DataFrame._assign_single_item_by_offset
+        )
+
+    _assign_multi_items_by_offset = _assign_multi_items_by_offsets
+    _assign_multi_items_by_label = _assign_multi_items
+    _assign_multi_items_by_labels = _assign_multi_items
+
+    def _assign_single_item_listlike_to_col_ids(
+        self, col_ids: Sequence[str], label: Optional[str], value: Sequence
+    ) -> DataFrame:
+        given_rows = len(value)
         actual_rows = len(self)
         assigning_to_empty_df = len(self.columns) == 0 and actual_rows == 0
         if not assigning_to_empty_df and given_rows != actual_rows:
@@ -2279,7 +2337,14 @@ def _assign_single_item_listlike(self, k: str, v: Sequence) -> DataFrame:
                 f"Length of values ({given_rows}) does not match length of index ({actual_rows})"
             )
 
-        local_df = DataFrame({k: v}, session=self._get_block().expr.session)
+        temp_col_name = (
+            label
+            if label is not None
+            else bigframes.core.guid.generate_guid("listlike_col_")
+        )
+        local_df = DataFrame(
+            {temp_col_name: value}, session=self._get_block().expr.session
+        )
         # local_df is likely (but not guaranteed) to be cached locally
         # since the original list came from memory and so is probably < MAX_INLINE_DF_SIZE
 
@@ -2287,13 +2352,17 @@ def _assign_single_item_listlike(self, k: str, v: Sequence) -> DataFrame:
         original_index_column_ids = self._block.index_columns
         self_block = self._block.reset_index(drop=False)
         if assigning_to_empty_df:
+            if label is None:
+                raise ValueError(
+                    "Label required when assigning listlike to empty DataFrame."
+                )
             if len(self._block.index_columns) > 1:
                 # match error raised by pandas here
                 raise ValueError(
                     "Assigning listlike to a first column under multiindex is not supported."
                 )
             result_block = new_column_block.with_index_labels(self._block.index.names)
-            result_block = result_block.with_column_labels([k])
+            result_block = result_block.with_column_labels([label])
         else:
             (
                 result_block,
@@ -2308,7 +2377,6 @@ def _assign_single_item_listlike(self, k: str, v: Sequence) -> DataFrame:
             )
             src_col = get_column_right[new_column_block.value_columns[0]]
             # Check to see if key exists, and modify in place
-            col_ids = self._block.cols_matching_label(k)
             for col_id in col_ids:
                 result_block = result_block.copy_values(
                     src_col, get_column_left[col_id]
@@ -2317,9 +2385,22 @@ def _assign_single_item_listlike(self, k: str, v: Sequence) -> DataFrame:
                 result_block = result_block.drop_columns([src_col])
         return DataFrame(result_block)
 
-    def _assign_scalar(self, label: str, value: Union[int, float, str]) -> DataFrame:
-        col_ids = self._block.cols_matching_label(label)
+    def _assign_single_item_listlike(self, k: str, v: Sequence) -> DataFrame:
+        col_ids = self._block.cols_matching_label(k)
+        return self._assign_single_item_listlike_to_col_ids(col_ids, k, v)
+
+    def _assign_single_item_listlike_by_offset(
+        self, offset: int, value: Sequence
+    ) -> DataFrame:
+        col_ids = [self._block.value_columns[offset]]
+        return self._assign_single_item_listlike_to_col_ids(col_ids, None, value)
 
+    def _assign_scalar_to_col_ids(
+        self,
+        col_ids: Sequence[str],
+        label: Optional[str],
+        value: Union[int, float, str],
+    ) -> DataFrame:
         block, constant_col_id = self._block.create_constant(value, label)
         for col_id in col_ids:
             block = block.copy_values(constant_col_id, col_id)
@@ -2329,25 +2410,40 @@ def _assign_scalar(self, label: str, value: Union[int, float, str]) -> DataFrame
 
         return DataFrame(block)
 
-    def _assign_series_join_on_index(
-        self, label: str, series: bigframes.series.Series
+    def _assign_scalar(self, label: str, value: Union[int, float, str]) -> DataFrame:
+        col_ids = self._block.cols_matching_label(label)
+        return self._assign_scalar_to_col_ids(col_ids, label, value)
+
+    def _assign_scalar_by_offset(
+        self, offset: int, value: Union[int, float, str]
+    ) -> DataFrame:
+        col_ids = [self._block.value_columns[offset]]
+        return self._assign_scalar_to_col_ids(col_ids, None, value)
+
+    def _assign_series_join_on_index_to_col_ids(
+        self,
+        column_ids: Sequence[str],
+        label: Optional[str],
+        series: bigframes.series.Series,
     ) -> DataFrame:
         block, (get_column_left, get_column_right) = self._block.join(
             series._block, how="left"
         )
 
-        column_ids = [
-            get_column_left[col_id] for col_id in self._block.cols_matching_label(label)
-        ]
+        mapped_column_ids = [get_column_left[col_id] for col_id in column_ids]
         source_column = get_column_right[series._value_column]
 
-        # Replace each column matching the label
-        for column_id in column_ids:
-            block = block.copy_values(source_column, column_id).assign_label(
-                column_id, label
-            )
+        # Replace each column matching the ids
+        for column_id in mapped_column_ids:
+            block = block.copy_values(source_column, column_id)
+            if label is not None:
+                block = block.assign_label(column_id, label)
 
-        if not column_ids:
+        if not mapped_column_ids:
+            if label is None:
+                raise ValueError(
+                    "Label required when appending a new column from Series."
+                )
             # Append case, so new column needs appropriate label
             block = block.assign_label(source_column, label)
         else:
@@ -2356,6 +2452,18 @@ def _assign_series_join_on_index(
 
         return DataFrame(block.with_index_labels(self._block.index.names))
 
+    def _assign_series_join_on_index(
+        self, label: str, series: bigframes.series.Series
+    ) -> DataFrame:
+        column_ids = self._block.cols_matching_label(label)
+        return self._assign_series_join_on_index_to_col_ids(column_ids, label, series)
+
+    def _assign_series_join_on_index_by_offset(
+        self, offset: int, series: bigframes.series.Series
+    ) -> DataFrame:
+        column_ids = [self._block.value_columns[offset]]
+        return self._assign_series_join_on_index_to_col_ids(column_ids, None, series)
+
     @overload  # type: ignore[override]
     def reset_index(
         self,
@@ -2618,9 +2726,9 @@ def take(
         if not utils.is_list_like(indices):
             raise ValueError("indices should be a list-like object.")
         if axis == 0 or axis == "index":
-            return self.iloc[indices]
+            return typing.cast(DataFrame, self.iloc[indices])
         elif axis == 1 or axis == "columns":
-            return self.iloc[:, indices]
+            return typing.cast(DataFrame, self.iloc[:, indices])
         else:
             raise ValueError(f"No axis named {axis} for object type DataFrame")
 
diff --git a/packages/bigframes/tests/unit/test_iloc_setitem.py b/packages/bigframes/tests/unit/test_iloc_setitem.py
index f84c0c06918d..9661026763f3 100644
--- a/packages/bigframes/tests/unit/test_iloc_setitem.py
+++ b/packages/bigframes/tests/unit/test_iloc_setitem.py
@@ -120,7 +120,9 @@ def test_iloc_getitem_column_single_integer(sample_df):
 
 
 @pytest.fixture
-def unordered_sample_df(sample_df: bpd.DataFrame) -> Generator[bpd.DataFrame, None, None]:
+def unordered_sample_df(
+    sample_df: bpd.DataFrame,
+) -> Generator[bpd.DataFrame, None, None]:
     session = sample_df._session
     original_strictly_ordered = session._strictly_ordered
     original_allow_ambiguity = session._allow_ambiguity
@@ -211,3 +213,42 @@ def test_iloc_setitem_column_errors(sample_df, key, expected_error):
 
     with pytest.raises(expected_error):
         bf_df.iloc[key] = 99
+
+
+@pytest.fixture
+def duplicate_columns_df() -> bpd.DataFrame:
+    pd_df = pd.DataFrame(
+        [[1, 2, 3], [4, 5, 6], [7, 8, 9]],
+        columns=["A", "B", "A"],
+    )
+    return bpd.read_pandas(pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_single_integer(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, 2] = 99
+    pd_df.iloc[:, 2] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_list_integer(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, [0, 2]] = [99, 88]
+    pd_df.iloc[:, [0, 2]] = [99, 88]
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_slice(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, 1:3] = 99
+    pd_df.iloc[:, 1:3] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)

From 53054acf280e515b82a3e97b8be734670f64dde8 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 8 Jul 2026 16:35:39 +0000
Subject: [PATCH 06/18] use autospec

---
 .../tests/unit/display/test_anywidget.py      | 51 +++++++++++--------
 1 file changed, 31 insertions(+), 20 deletions(-)

diff --git a/packages/bigframes/tests/unit/display/test_anywidget.py b/packages/bigframes/tests/unit/display/test_anywidget.py
index 2ad4c1d3f4b5..b184a8c459e2 100644
--- a/packages/bigframes/tests/unit/display/test_anywidget.py
+++ b/packages/bigframes/tests/unit/display/test_anywidget.py
@@ -24,11 +24,12 @@
 pytest.importorskip("anywidget")
 pytest.importorskip("traitlets")
 
+import bigframes.dataframe
+import bigframes.operations
+import bigframes.series
 from bigframes.core.blocks import Block
-from bigframes.dataframe import DataFrame
 from bigframes.display.anywidget import TableWidget
 from bigframes.dtypes import JSON_DTYPE, STRING_DTYPE, struct_type
-from bigframes.operations import SqlScalarOp
 
 
 def test_navigation_to_invalid_page_resets_to_valid_page_without_deadlock():
@@ -198,25 +199,30 @@ def test_json_column_converted_to_string_for_display():
     mock_block.column_labels = pd.Index(["col_json"])
     mock_block.value_columns = ["col_json"]
 
-    df = DataFrame(mock_block)
+    df = bigframes.dataframe.DataFrame(mock_block)
     df._block = mock_block
 
-    mock_series = mock.Mock()
+    mock_series = mock.create_autospec(bigframes.series.Series, instance=True)
     mock_series.dtype = JSON_DTYPE
+    mock_series._slice.return_value.apply.side_effect = (
+        lambda func, *args, **kwargs: func(mock_series)
+    )
 
-    with mock.patch.object(DataFrame, "__getitem__", return_value=mock_series):
-        with mock.patch.object(DataFrame, "assign") as mock_assign:
+    with mock.patch.object(
+        bigframes.dataframe.DataFrame, "__getitem__", return_value=mock_series
+    ):
+        with mock.patch.object(
+            bigframes.dataframe.DataFrame, "_assign_multi_items_by_offsets"
+        ) as mock_assign:
             df._prepare_display_df()
 
             mock_assign.assert_called_once()
-            _, kwargs = mock_assign.call_args
-            assert "col_json" in kwargs
+            args, _ = mock_assign.call_args
+            assert 0 in args[0]
 
             mock_series._apply_unary_op.assert_called_once()
             call_arg = mock_series._apply_unary_op.call_args[0][0]
-            assert isinstance(call_arg, SqlScalarOp)
-            assert call_arg._output_type == STRING_DTYPE
-            assert call_arg.sql_template == "TO_JSON_STRING({0})"
+            assert isinstance(call_arg, bigframes.operations.ToJSONString)
 
 
 def test_struct_column_with_nested_json_converted_to_string_for_display():
@@ -228,25 +234,30 @@ def test_struct_column_with_nested_json_converted_to_string_for_display():
     mock_block.column_labels = pd.Index(["col_struct"])
     mock_block.value_columns = ["col_struct"]
 
-    df = DataFrame(mock_block)
+    df = bigframes.dataframe.DataFrame(mock_block)
     df._block = mock_block
 
-    mock_series = mock.Mock()
+    mock_series = mock.create_autospec(bigframes.series.Series, instance=True)
     mock_series.dtype = nested_struct_dtype
+    mock_series._slice.return_value.apply.side_effect = (
+        lambda func, *args, **kwargs: func(mock_series)
+    )
 
-    with mock.patch.object(DataFrame, "__getitem__", return_value=mock_series):
-        with mock.patch.object(DataFrame, "assign") as mock_assign:
+    with mock.patch.object(
+        bigframes.dataframe.DataFrame, "__getitem__", return_value=mock_series
+    ):
+        with mock.patch.object(
+            bigframes.dataframe.DataFrame, "_assign_multi_items_by_offsets"
+        ) as mock_assign:
             df._prepare_display_df()
 
             mock_assign.assert_called_once()
-            _, kwargs = mock_assign.call_args
-            assert "col_struct" in kwargs
+            args, _ = mock_assign.call_args
+            assert 0 in args[0]
 
             mock_series._apply_unary_op.assert_called_once()
             call_arg = mock_series._apply_unary_op.call_args[0][0]
-            assert isinstance(call_arg, SqlScalarOp)
-            assert call_arg._output_type == STRING_DTYPE
-            assert call_arg.sql_template == "TO_JSON_STRING({0})"
+            assert isinstance(call_arg, bigframes.operations.ToJSONString)
 
 
 @pytest.fixture

From 2400e1cf78d18b902b5c8b8fcb72558841932d26 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Wed, 8 Jul 2026 18:43:19 +0000
Subject: [PATCH 07/18] resolve mypy failures

---
 packages/bigframes/bigframes/dataframe.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/bigframes/bigframes/dataframe.py b/packages/bigframes/bigframes/dataframe.py
index 786f5cd75f6e..07ba6aba1c0e 100644
--- a/packages/bigframes/bigframes/dataframe.py
+++ b/packages/bigframes/bigframes/dataframe.py
@@ -2255,7 +2255,7 @@ def _assign_single_item_by_offset(
                 label: ex.deref(id) for id, label in self._block.col_id_to_label.items()
             }
             resolved_expr = value._value.bind_variables(label_to_col_ref)
-            block, (new_col_id,) = self._block.project_expr(resolved_expr)
+            block, new_col_id = self._block.project_expr(resolved_expr)
             target_col_id = self._block.value_columns[offset]
             block = block.copy_values(new_col_id, target_col_id).drop_columns(
                 [new_col_id]
@@ -2281,7 +2281,7 @@ def _assign_single_item_by_offset(
 
     def _assign_multi_items_helper(
         self,
-        k: Sequence[Any],
+        k: Sequence[Any] | pandas.Index,
         v: SingleItemValue | MultiItemValue,
         assign_single_fn: Callable[[DataFrame, Any, Any], DataFrame],
     ) -> DataFrame:

From 5378ba97b16f716c3a4420786627e63db492ba93 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 9 Jul 2026 17:14:23 +0000
Subject: [PATCH 08/18] refactor out noop slice helper

---
 packages/bigframes/bigframes/core/indexers.py | 56 +++++++------------
 1 file changed, 21 insertions(+), 35 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index f0cae0542b23..e5cb734d575b 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -16,7 +16,7 @@
 
 import typing
 import warnings
-from typing import Tuple, Union
+from typing import Any, Tuple, Union
 
 import bigframes_vendored.constants as constants
 import bigframes_vendored.ibis.common.exceptions as ibis_exceptions
@@ -103,16 +103,7 @@ def __getitem__(
 
         Other key types are not yet supported.
         """
-        requires_ordering = True
-        if (
-            isinstance(key, slice)
-            and (key.start is None or key.start == 0)
-            and (key.step is None or key.step == 1)
-            and key.stop is None
-        ):
-            requires_ordering = False
-
-        if requires_ordering:
+        if not _is_noop_slice(key):
             validations.enforce_ordered(self._series, "iloc")
 
         return _iloc_getitem_series_or_dataframe(self._series, key)
@@ -202,10 +193,7 @@ def __setitem__(
         if (
             isinstance(key, tuple)
             and len(key) == 2
-            and isinstance(key[0], slice)
-            and (key[0].start is None or key[0].start == 0)
-            and (key[0].step is None or key[0].step == 1)
-            and key[0].stop is None
+            and _is_noop_slice(key[0])
         ):
             # TODO(swast): Support setting multiple columns with key[1] as a list
             # of labels and value as a DataFrame.
@@ -261,21 +249,10 @@ def __getitem__(self, key) -> Union[bigframes.dataframe.DataFrame, pd.Series]:
         if isinstance(key, tuple):
             if len(key) > 0:
                 row_indexer = key[0]
-                if (
-                    isinstance(row_indexer, slice)
-                    and (row_indexer.start is None or row_indexer.start == 0)
-                    and (row_indexer.step is None or row_indexer.step == 1)
-                    and row_indexer.stop is None
-                ):
+                if _is_noop_slice(row_indexer):
                     requires_ordering = False
-        else:
-            if (
-                isinstance(key, slice)
-                and (key.start is None or key.start == 0)
-                and (key.step is None or key.step == 1)
-                and key.stop is None
-            ):
-                requires_ordering = False
+        elif _is_noop_slice(key):
+            requires_ordering = False
 
         if requires_ordering:
             validations.enforce_ordered(self._dataframe, "iloc")
@@ -294,10 +271,7 @@ def __setitem__(
         if not (
             isinstance(key, tuple)
             and len(key) == 2
-            and isinstance(key[0], slice)
-            and (key[0].start is None or key[0].start == 0)
-            and (key[0].step is None or key[0].step == 1)
-            and key[0].stop is None
+            and _is_noop_slice(key[0])
         ):
             raise NotImplementedError(
                 "Only DataFrame.iloc[:, col_indexer] = value is supported."
@@ -410,6 +384,16 @@ def _loc_getitem_series_or_dataframe(
 ) -> Union[bigframes.dataframe.DataFrame, pd.Series]: ...
 
 
+def _is_noop_slice(key: Any) -> bool:
+    """Return True if key is a slice selecting all elements in the original order."""
+    return (
+        isinstance(key, slice)
+        and (key.start is None or key.start == 0)
+        and (key.step is None or key.step == 1)
+        and key.stop is None
+    )
+
+
 def _loc_getitem_series_or_dataframe(
     series_or_dataframe: Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
     key: LocSingleKey,
@@ -419,12 +403,14 @@ def _loc_getitem_series_or_dataframe(
     pd.Series,
     bigframes.core.scalar.Scalar,
 ]:
+    if _is_noop_slice(key):
+        return series_or_dataframe.copy()
+
     if isinstance(key, slice):
-        if (key.start is None) and (key.stop is None) and (key.step is None):
-            return series_or_dataframe.copy()
         raise NotImplementedError(
             f"loc does not yet support indexing with a slice. {constants.FEEDBACK_LINK}"
         )
+
     if isinstance(key, bigframes.core.col.Expression):
         label_to_col_ref = {
             label: ex.deref(id)

From 2d696c160a2d3fdf5116424806b9b530f9812735 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 9 Jul 2026 18:23:53 +0000
Subject: [PATCH 09/18] refactor to use column offsets

---
 packages/bigframes/bigframes/core/indexers.py | 168 ++++++++++--------
 1 file changed, 90 insertions(+), 78 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index e5cb734d575b..e4c9c3d32a83 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -14,13 +14,17 @@
 
 from __future__ import annotations
 
+import numbers
 import typing
 import warnings
-from typing import Any, Tuple, Union
+from typing import Any, Sequence, Tuple, Union
 
 import bigframes_vendored.constants as constants
 import bigframes_vendored.ibis.common.exceptions as ibis_exceptions
+import numpy as np
 import pandas as pd
+import pyarrow as pa
+import pyarrow.types
 
 import bigframes.core.blocks
 import bigframes.core.col
@@ -46,6 +50,9 @@
     ]
 
 
+_DATAFRAME_ILOC_ERROR = "Only DataFrame.iloc[:, col_indexer] = value is supported."
+
+
 class LocSeriesIndexer:
     def __init__(self, series: bigframes.series.Series):
         self._series = series
@@ -190,11 +197,7 @@ def __setitem__(
         key: Tuple[slice, str],
         value: bigframes.dataframe.SingleItemValue,
     ):
-        if (
-            isinstance(key, tuple)
-            and len(key) == 2
-            and _is_noop_slice(key[0])
-        ):
+        if isinstance(key, tuple) and len(key) == 2 and _is_noop_slice(key[0]):
             # TODO(swast): Support setting multiple columns with key[1] as a list
             # of labels and value as a DataFrame.
             df = self._dataframe.assign(**{key[1]: value})
@@ -268,72 +271,17 @@ def __setitem__(
             bigframes.dataframe.SingleItemValue, bigframes.dataframe.DataFrame
         ],
     ):
-        if not (
-            isinstance(key, tuple)
-            and len(key) == 2
-            and _is_noop_slice(key[0])
-        ):
-            raise NotImplementedError(
-                "Only DataFrame.iloc[:, col_indexer] = value is supported."
-            )
-
-        col_indexer = key[1]
-        n_cols = len(self._dataframe.columns)
-
-        if isinstance(col_indexer, bool):
-            raise TypeError(
-                "pos must be integer or slice or list-like of integers/booleans"
-            )
-
-        if isinstance(col_indexer, int):
-            col_offset = col_indexer
-            if col_offset < 0:
-                col_offset += n_cols
-            if col_offset < 0 or col_offset >= n_cols:
-                raise IndexError("single positional indexer is out-of-bounds")
+        if not (isinstance(key, tuple) and len(key) == 2):
+            raise NotImplementedError(_DATAFRAME_ILOC_ERROR)
 
-            df = self._dataframe._assign_multi_items_by_offsets([col_offset], value)
-            self._dataframe._set_block(df._get_block())
+        row_indexer, col_indexer = key
 
-        elif isinstance(col_indexer, slice):
-            col_offsets = list(range(*col_indexer.indices(n_cols)))
-            if not col_offsets:
-                return
-            df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
-            self._dataframe._set_block(df._get_block())
+        if not _is_noop_slice(row_indexer):
+            raise NotImplementedError(_DATAFRAME_ILOC_ERROR)
 
-        elif pd.api.types.is_list_like(col_indexer):
-            col_indexer_list = list(col_indexer)
-
-            if len(col_indexer_list) > 0 and all(
-                isinstance(x, bool) for x in col_indexer_list
-            ):
-                if len(col_indexer_list) != n_cols:
-                    raise ValueError(
-                        f"Boolean index has wrong length: {len(col_indexer_list)} instead of {n_cols}"
-                    )
-                col_offsets = [i for i, val in enumerate(col_indexer_list) if val]
-            else:
-                col_offsets = []
-                for idx in col_indexer_list:
-                    if isinstance(idx, bool):
-                        raise TypeError("pos list must contain only integers")
-                    if not isinstance(idx, int):
-                        raise TypeError("pos list must contain only integers")
-                    if idx < 0:
-                        idx += n_cols
-                    if idx < 0 or idx >= n_cols:
-                        raise IndexError("positional indexer is out-of-bounds")
-                    col_offsets.append(idx)
-
-            if not col_offsets:
-                return
-            df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
-            self._dataframe._set_block(df._get_block())
-        else:
-            raise TypeError(
-                "pos must be integer or slice or list-like of integers/booleans"
-            )
+        col_offsets = _iloc_col_indexer_to_offsets(self._dataframe, col_indexer)
+        df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
+        self._dataframe._set_block(df._get_block())
 
 
 class IatDataFrameIndexer:
@@ -539,6 +487,74 @@ def _iloc_getitem_series_or_dataframe(
 ) -> Union[bigframes.dataframe.DataFrame, pd.Series, bigframes.core.scalar.Scalar]: ...
 
 
+def _iloc_clip_to_offset(index: int, length: int, name: str) -> int:
+    """Support negative values for offsets."""
+    offset = index
+    if offset < 0:
+        offset += length
+
+    if offset < 0 or offset >= length:
+        raise IndexError(f"{name} {index} is out-of-bounds")
+
+    return index
+
+
+def _iloc_col_indexer_to_offsets(
+    df: bigframes.dataframe.DataFrame, col_indexer: Any
+) -> Sequence[int]:
+    """Convert col_indexer from one of the many pandas-compatible formats to a list of offsets."""
+    n_cols = len(df.columns)
+
+    if (
+        isinstance(col_indexer, bool)
+        or isinstance(col_indexer, np.bool_)
+        or (
+            isinstance(col_indexer, pa.Scalar) and pyarrow.types.is_boolean(col_indexer)
+        )
+    ):
+        raise TypeError("scalar boolean is not supported as an iloc column indexer")
+
+    if isinstance(col_indexer, numbers.Integral) or (
+        isinstance(col_indexer, pa.Scalar) and pyarrow.types.is_integer(col_indexer)
+    ):
+        col_offset = int(col_indexer)
+        return [
+            _iloc_clip_to_offset(
+                col_offset, n_cols, "single positional iloc column indexer"
+            )
+        ]
+
+    elif isinstance(col_indexer, slice):
+        return list(range(*col_indexer.indices(n_cols)))
+
+    elif pd.api.types.is_list_like(col_indexer):
+        col_indexer_list = list(col_indexer)
+
+        if len(col_indexer_list) > 0 and all(
+            isinstance(x, bool) for x in col_indexer_list
+        ):
+            if len(col_indexer_list) != n_cols:
+                raise ValueError(
+                    f"Boolean iloc column indexer has wrong length: {len(col_indexer_list)} instead of {n_cols}"
+                )
+            return [i for i, val in enumerate(col_indexer_list) if val]
+        else:
+            return [
+                _iloc_clip_to_offset(idx, n_cols, "iloc column indexer")
+                for idx in col_indexer_list
+            ]
+
+    raise TypeError(f"got unexpected {type(col_indexer)} for iloc column indexer")
+
+
+def _iloc_df_from_column_offsets(
+    df: bigframes.dataframe.DataFrame, key: Sequence[int]
+) -> bigframes.dataframe.DataFrame:
+    block = df._block
+    selected_ids = tuple(block.value_columns[offset] for offset in key)
+    return bigframes.dataframe.DataFrame(block.select_columns(selected_ids))
+
+
 def _iloc_getitem_series_or_dataframe(
     series_or_dataframe: Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
     key,
@@ -573,15 +589,11 @@ def _iloc_getitem_series_or_dataframe(
         df = typing.cast(bigframes.dataframe.DataFrame, series_or_dataframe)
         if isinstance(key[0], int) and isinstance(key[1], int):
             return df.iat[key]
-        elif isinstance(key[1], int):
-            col_label = df.columns[key[1]]
-            return df[col_label].iloc[key[0]]
-        elif isinstance(key[1], list):
-            columns = df.columns[key[1]]
-            return _iloc_getitem_series_or_dataframe(df[columns], key[0])
-        raise NotImplementedError(
-            f"iloc does not yet support indexing with {key}. {constants.FEEDBACK_LINK}"
-        )
+
+        row_indexer, column_indexer = key
+        column_offsets = _iloc_col_indexer_to_offsets(df, column_indexer)
+        df_subset = _iloc_df_from_column_offsets(df, column_offsets)
+        return _iloc_getitem_series_or_dataframe(df_subset, row_indexer)
     elif pd.api.types.is_list_like(key):
         if len(key) == 0:
             return typing.cast(

From 89822aa689547c3d59aadd57d1315feac495c15c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 9 Jul 2026 18:43:54 +0000
Subject: [PATCH 10/18] fix getitem with single column

---
 packages/bigframes/bigframes/core/indexers.py | 26 ++++---
 .../tests/unit/display/test_anywidget.py      | 72 +++++--------------
 2 files changed, 34 insertions(+), 64 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index e4c9c3d32a83..0287953fa2ae 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -17,7 +17,7 @@
 import numbers
 import typing
 import warnings
-from typing import Any, Sequence, Tuple, Union
+from typing import Any, Sequence, Tuple, Union, cast
 
 import bigframes_vendored.constants as constants
 import bigframes_vendored.ibis.common.exceptions as ibis_exceptions
@@ -279,7 +279,7 @@ def __setitem__(
         if not _is_noop_slice(row_indexer):
             raise NotImplementedError(_DATAFRAME_ILOC_ERROR)
 
-        col_offsets = _iloc_col_indexer_to_offsets(self._dataframe, col_indexer)
+        col_offsets, _ = _iloc_col_indexer_to_offsets(self._dataframe, col_indexer)
         df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
         self._dataframe._set_block(df._get_block())
 
@@ -501,7 +501,7 @@ def _iloc_clip_to_offset(index: int, length: int, name: str) -> int:
 
 def _iloc_col_indexer_to_offsets(
     df: bigframes.dataframe.DataFrame, col_indexer: Any
-) -> Sequence[int]:
+) -> Tuple[Sequence[int], str]:
     """Convert col_indexer from one of the many pandas-compatible formats to a list of offsets."""
     n_cols = len(df.columns)
 
@@ -522,10 +522,10 @@ def _iloc_col_indexer_to_offsets(
             _iloc_clip_to_offset(
                 col_offset, n_cols, "single positional iloc column indexer"
             )
-        ]
+        ], "Series"
 
     elif isinstance(col_indexer, slice):
-        return list(range(*col_indexer.indices(n_cols)))
+        return list(range(*col_indexer.indices(n_cols))), "DataFrame"
 
     elif pd.api.types.is_list_like(col_indexer):
         col_indexer_list = list(col_indexer)
@@ -537,12 +537,12 @@ def _iloc_col_indexer_to_offsets(
                 raise ValueError(
                     f"Boolean iloc column indexer has wrong length: {len(col_indexer_list)} instead of {n_cols}"
                 )
-            return [i for i, val in enumerate(col_indexer_list) if val]
+            return [i for i, val in enumerate(col_indexer_list) if val], "DataFrame"
         else:
             return [
                 _iloc_clip_to_offset(idx, n_cols, "iloc column indexer")
                 for idx in col_indexer_list
-            ]
+            ], "DataFrame"
 
     raise TypeError(f"got unexpected {type(col_indexer)} for iloc column indexer")
 
@@ -591,9 +591,17 @@ def _iloc_getitem_series_or_dataframe(
             return df.iat[key]
 
         row_indexer, column_indexer = key
-        column_offsets = _iloc_col_indexer_to_offsets(df, column_indexer)
+        column_offsets, output_type = _iloc_col_indexer_to_offsets(df, column_indexer)
         df_subset = _iloc_df_from_column_offsets(df, column_offsets)
-        return _iloc_getitem_series_or_dataframe(df_subset, row_indexer)
+        if output_type == "Series":
+            selected_columns = cast(
+                Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
+                df_subset[df_subset.columns[0]],
+            )
+        else:
+            selected_columns = df_subset
+
+        return _iloc_getitem_series_or_dataframe(selected_columns, row_indexer)
     elif pd.api.types.is_list_like(key):
         if len(key) == 0:
             return typing.cast(
diff --git a/packages/bigframes/tests/unit/display/test_anywidget.py b/packages/bigframes/tests/unit/display/test_anywidget.py
index b184a8c459e2..2e2800064124 100644
--- a/packages/bigframes/tests/unit/display/test_anywidget.py
+++ b/packages/bigframes/tests/unit/display/test_anywidget.py
@@ -27,7 +27,6 @@
 import bigframes.dataframe
 import bigframes.operations
 import bigframes.series
-from bigframes.core.blocks import Block
 from bigframes.display.anywidget import TableWidget
 from bigframes.dtypes import JSON_DTYPE, STRING_DTYPE, struct_type
 
@@ -194,70 +193,33 @@ def test_cell_execution_count_propagation(mock_df):
     )
 
 
-def test_json_column_converted_to_string_for_display():
-    mock_block = mock.Mock(spec=Block)
-    mock_block.column_labels = pd.Index(["col_json"])
-    mock_block.value_columns = ["col_json"]
-
-    df = bigframes.dataframe.DataFrame(mock_block)
-    df._block = mock_block
-
-    mock_series = mock.create_autospec(bigframes.series.Series, instance=True)
-    mock_series.dtype = JSON_DTYPE
-    mock_series._slice.return_value.apply.side_effect = (
-        lambda func, *args, **kwargs: func(mock_series)
+def test_json_column_converted_to_string_for_display(polars_session):
+    series = bigframes.series.Series(
+        ['{"a": 1}', '{"b": 2}'], dtype=JSON_DTYPE, session=polars_session
     )
+    df = series.to_frame("col_json")
 
-    with mock.patch.object(
-        bigframes.dataframe.DataFrame, "__getitem__", return_value=mock_series
-    ):
-        with mock.patch.object(
-            bigframes.dataframe.DataFrame, "_assign_multi_items_by_offsets"
-        ) as mock_assign:
-            df._prepare_display_df()
-
-            mock_assign.assert_called_once()
-            args, _ = mock_assign.call_args
-            assert 0 in args[0]
+    result = df._prepare_display_df()
 
-            mock_series._apply_unary_op.assert_called_once()
-            call_arg = mock_series._apply_unary_op.call_args[0][0]
-            assert isinstance(call_arg, bigframes.operations.ToJSONString)
+    assert result["col_json"].dtype == STRING_DTYPE
 
 
-def test_struct_column_with_nested_json_converted_to_string_for_display():
+def test_struct_column_with_nested_json_converted_to_string_for_display(
+    polars_session,
+):
     nested_struct_dtype = struct_type(
         [("field1", STRING_DTYPE), ("field2", JSON_DTYPE)]
     )
-
-    mock_block = mock.Mock(spec=Block)
-    mock_block.column_labels = pd.Index(["col_struct"])
-    mock_block.value_columns = ["col_struct"]
-
-    df = bigframes.dataframe.DataFrame(mock_block)
-    df._block = mock_block
-
-    mock_series = mock.create_autospec(bigframes.series.Series, instance=True)
-    mock_series.dtype = nested_struct_dtype
-    mock_series._slice.return_value.apply.side_effect = (
-        lambda func, *args, **kwargs: func(mock_series)
+    series = bigframes.series.Series(
+        [{"field1": "hello", "field2": '{"a": 1}'}],
+        dtype=nested_struct_dtype,
+        session=polars_session,
     )
+    df = series.to_frame("col_struct")
 
-    with mock.patch.object(
-        bigframes.dataframe.DataFrame, "__getitem__", return_value=mock_series
-    ):
-        with mock.patch.object(
-            bigframes.dataframe.DataFrame, "_assign_multi_items_by_offsets"
-        ) as mock_assign:
-            df._prepare_display_df()
-
-            mock_assign.assert_called_once()
-            args, _ = mock_assign.call_args
-            assert 0 in args[0]
-
-            mock_series._apply_unary_op.assert_called_once()
-            call_arg = mock_series._apply_unary_op.call_args[0][0]
-            assert isinstance(call_arg, bigframes.operations.ToJSONString)
+    result = df._prepare_display_df()
+
+    assert result["col_struct"].dtype == STRING_DTYPE
 
 
 @pytest.fixture

From ff4c432e37fb6b0fa2dc5d0cc47f6a9230921665 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 9 Jul 2026 21:56:48 +0000
Subject: [PATCH 11/18] workaround for json in polars engine

---
 packages/bigframes/bigframes/core/indexers.py | 44 ++++++------
 .../bigframes/bigframes/core/local_data.py    | 68 +++++++++++++++++++
 packages/bigframes/bigframes/dataframe.py     | 38 +++++++++--
 packages/bigframes/bigframes/series.py        | 38 +++++++++--
 4 files changed, 154 insertions(+), 34 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index 0287953fa2ae..042fa88170a3 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -320,6 +320,16 @@ def __getitem__(
         return self._dataframe.loc[key]
 
 
+def _is_noop_slice(key: Any) -> bool:
+    """Return True if key is a slice selecting all elements in the original order."""
+    return (
+        isinstance(key, slice)
+        and (key.start is None or key.start == 0)
+        and (key.step is None or key.step == 1)
+        and key.stop is None
+    )
+
+
 @typing.overload
 def _loc_getitem_series_or_dataframe(
     series_or_dataframe: bigframes.series.Series, key
@@ -332,16 +342,6 @@ def _loc_getitem_series_or_dataframe(
 ) -> Union[bigframes.dataframe.DataFrame, pd.Series]: ...
 
 
-def _is_noop_slice(key: Any) -> bool:
-    """Return True if key is a slice selecting all elements in the original order."""
-    return (
-        isinstance(key, slice)
-        and (key.start is None or key.start == 0)
-        and (key.step is None or key.step == 1)
-        and key.stop is None
-    )
-
-
 def _loc_getitem_series_or_dataframe(
     series_or_dataframe: Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
     key: LocSingleKey,
@@ -475,18 +475,6 @@ def _struct_accessor_check_and_warn(
         warnings.warn(msg, stacklevel=7, category=bfe.BadIndexerKeyWarning)
 
 
-@typing.overload
-def _iloc_getitem_series_or_dataframe(
-    series_or_dataframe: bigframes.series.Series, key
-) -> Union[bigframes.series.Series, bigframes.core.scalar.Scalar]: ...
-
-
-@typing.overload
-def _iloc_getitem_series_or_dataframe(
-    series_or_dataframe: bigframes.dataframe.DataFrame, key
-) -> Union[bigframes.dataframe.DataFrame, pd.Series, bigframes.core.scalar.Scalar]: ...
-
-
 def _iloc_clip_to_offset(index: int, length: int, name: str) -> int:
     """Support negative values for offsets."""
     offset = index
@@ -555,6 +543,18 @@ def _iloc_df_from_column_offsets(
     return bigframes.dataframe.DataFrame(block.select_columns(selected_ids))
 
 
+@typing.overload
+def _iloc_getitem_series_or_dataframe(
+    series_or_dataframe: bigframes.series.Series, key
+) -> Union[bigframes.series.Series, bigframes.core.scalar.Scalar]: ...
+
+
+@typing.overload
+def _iloc_getitem_series_or_dataframe(
+    series_or_dataframe: bigframes.dataframe.DataFrame, key
+) -> Union[bigframes.dataframe.DataFrame, pd.Series, bigframes.core.scalar.Scalar]: ...
+
+
 def _iloc_getitem_series_or_dataframe(
     series_or_dataframe: Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
     key,
diff --git a/packages/bigframes/bigframes/core/local_data.py b/packages/bigframes/bigframes/core/local_data.py
index c05bda7a7fbc..227a6e3023a4 100644
--- a/packages/bigframes/bigframes/core/local_data.py
+++ b/packages/bigframes/bigframes/core/local_data.py
@@ -521,3 +521,71 @@ def _schema_durations_to_ints(schema: pa.Schema) -> pa.Schema:
     return pa.schema(
         pa.field(field.name, _durations_to_ints(field.type)) for field in schema
     )
+
+
+def _has_extension_type(pa_type: pa.DataType) -> bool:
+    """Check if the given pyarrow DataType contains an ExtensionType at any nesting level."""
+    if isinstance(pa_type, pa.ExtensionType):
+        return True
+    if pa.types.is_struct(pa_type):
+        struct_type = cast(pa.StructType, pa_type)
+        return any(
+            _has_extension_type(struct_type.field(i).type)
+            for i in range(struct_type.num_fields)
+        )
+    if pa.types.is_list(pa_type) or pa.types.is_large_list(pa_type):
+        list_type = cast(pa.ListType, pa_type)
+        return _has_extension_type(list_type.value_type)
+    return False
+
+
+def pyarrow_array_from_sequence(data: Any, pa_type: pa.DataType) -> pa.Array:
+    """Construct a pyarrow Array from a Sequence, DataFrame, or Series,
+    recursively handling ExtensionTypes inside structs and lists.
+
+    This is a workaround for issue b/456774399 where ArrowNotImplementedError is
+    thrown when constructing arrays of structs containing extension dtypes like
+    JSON.
+    """
+    if isinstance(data, (pa.Array, pa.ChunkedArray)):
+        return data.cast(pa_type) if data.type != pa_type else data
+    if isinstance(pa_type, pa.ExtensionType):
+        storage_arr = pyarrow_array_from_sequence(data, pa_type.storage_type)
+        return pa.ExtensionArray.from_storage(pa_type, storage_arr)
+    if pa.types.is_struct(pa_type):
+        struct_type = cast(pa.StructType, pa_type)
+        if isinstance(data, (pd.Series, pd.Index)):
+            df = pd.DataFrame(data.tolist())
+        elif isinstance(data, pd.DataFrame):
+            df = data
+        else:
+            df = pd.DataFrame(
+                list(data)
+                if not isinstance(data, (list, tuple, np.ndarray, pd.Series, pd.Index))
+                else data
+            )
+        arrays = []
+        fields = []
+        for i in range(struct_type.num_fields):
+            field = struct_type.field(i)
+            col_data = df[field.name] if field.name in df.columns else [None] * len(df)
+            arrays.append(pyarrow_array_from_sequence(col_data, field.type))
+            fields.append(field)
+        return pa.StructArray.from_arrays(arrays, fields=fields)
+    if pa.types.is_list(pa_type) or pa.types.is_large_list(pa_type):
+        list_type = cast(pa.ListType, pa_type)
+        if _has_extension_type(list_type.value_type):
+            flat_values = []
+            offsets = [0]
+            iterable = (
+                data.tolist() if isinstance(data, (pd.Series, pd.Index)) else data
+            )
+            for item in iterable:
+                if item is None or pd.isna(item):
+                    offsets.append(offsets[-1])
+                else:
+                    flat_values.extend(item)
+                    offsets.append(offsets[-1] + len(item))
+            values_arr = pyarrow_array_from_sequence(flat_values, list_type.value_type)
+            return pa.ListArray.from_arrays(pa.array(offsets), values_arr)
+    return pa.array(data, type=pa_type)
diff --git a/packages/bigframes/bigframes/dataframe.py b/packages/bigframes/bigframes/dataframe.py
index 07ba6aba1c0e..7668aeb17feb 100644
--- a/packages/bigframes/bigframes/dataframe.py
+++ b/packages/bigframes/bigframes/dataframe.py
@@ -221,14 +221,40 @@ def __init__(
             if isinstance(dtype, str) and dtype.lower() == "json":
                 dtype = bigframes.dtypes.JSON_DTYPE
 
+            import bigframes.core.local_data
             import bigframes.pandas
 
-            pd_dataframe = pandas.DataFrame(
-                data=data,
-                index=index,  # type:ignore
-                columns=columns,  # type:ignore
-                dtype=dtype,  # type:ignore
-            )
+            if (
+                hasattr(pandas.arrays, "ArrowExtensionArray")
+                and isinstance(dtype, pandas.ArrowDtype)
+                and not isinstance(
+                    data,
+                    (
+                        pandas.DataFrame,
+                        pandas.Series,
+                        pandas.Index,
+                        pyarrow.Array,
+                        pyarrow.ChunkedArray,
+                        pyarrow.Table,
+                        pandas.arrays.ArrowExtensionArray,
+                    ),
+                )
+            ):
+                pa_array = bigframes.core.local_data.pyarrow_array_from_sequence(
+                    data, dtype.pyarrow_dtype
+                )
+                pd_dataframe = pandas.DataFrame(
+                    data=pandas.arrays.ArrowExtensionArray(pa_array),
+                    index=index,  # type:ignore
+                    columns=columns,  # type:ignore
+                )
+            else:
+                pd_dataframe = pandas.DataFrame(
+                    data=data,
+                    index=index,  # type:ignore
+                    columns=columns,  # type:ignore
+                    dtype=dtype,  # type:ignore
+                )
             if session:
                 block = session.read_pandas(pd_dataframe)._get_block()
             else:
diff --git a/packages/bigframes/bigframes/series.py b/packages/bigframes/bigframes/series.py
index 68dc66625b90..56baf7e682a1 100644
--- a/packages/bigframes/bigframes/series.py
+++ b/packages/bigframes/bigframes/series.py
@@ -195,12 +195,38 @@ def __init__(
         else:
             if isinstance(dtype, str) and dtype.lower() == "json":
                 dtype = bigframes.dtypes.JSON_DTYPE
-            pd_series = pandas.Series(
-                data=data,
-                index=index,  # type:ignore
-                dtype=dtype,  # type:ignore
-                name=name,
-            )
+
+            import bigframes.core.local_data
+
+            if (
+                hasattr(pandas.arrays, "ArrowExtensionArray")
+                and isinstance(dtype, pandas.ArrowDtype)
+                and not isinstance(
+                    data,
+                    (
+                        pandas.Series,
+                        pandas.Index,
+                        pa.Array,
+                        pa.ChunkedArray,
+                        pandas.arrays.ArrowExtensionArray,
+                    ),
+                )
+            ):
+                pa_array = bigframes.core.local_data.pyarrow_array_from_sequence(
+                    data, dtype.pyarrow_dtype
+                )
+                pd_series = pandas.Series(
+                    data=pandas.arrays.ArrowExtensionArray(pa_array),
+                    index=index,  # type:ignore
+                    name=name,
+                )
+            else:
+                pd_series = pandas.Series(
+                    data=data,
+                    index=index,  # type:ignore
+                    dtype=dtype,  # type:ignore
+                    name=name,
+                )
             block = read_pandas_func(pd_series)._get_block()  # type:ignore
 
         assert block is not None

From 70f35876b46398bdeab9c246d7eb830883c41466 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 9 Jul 2026 22:08:18 +0000
Subject: [PATCH 12/18] mypy fixes

---
 .../bigframes/bigframes/core/local_data.py    | 26 +++++++++++++++----
 1 file changed, 21 insertions(+), 5 deletions(-)

diff --git a/packages/bigframes/bigframes/core/local_data.py b/packages/bigframes/bigframes/core/local_data.py
index 227a6e3023a4..87f1ca0e6ba3 100644
--- a/packages/bigframes/bigframes/core/local_data.py
+++ b/packages/bigframes/bigframes/core/local_data.py
@@ -22,7 +22,17 @@
 import itertools
 import json
 import uuid
-from typing import Any, Callable, Generator, Iterable, Literal, Optional, Union, cast
+from typing import (
+    Any,
+    Callable,
+    Generator,
+    Iterable,
+    Literal,
+    Optional,
+    Sequence,
+    Union,
+    cast,
+)
 
 import geopandas  # type: ignore
 import numpy
@@ -575,17 +585,23 @@ def pyarrow_array_from_sequence(data: Any, pa_type: pa.DataType) -> pa.Array:
     if pa.types.is_list(pa_type) or pa.types.is_large_list(pa_type):
         list_type = cast(pa.ListType, pa_type)
         if _has_extension_type(list_type.value_type):
-            flat_values = []
+            flat_values: list[Any] = []
             offsets = [0]
             iterable = (
                 data.tolist() if isinstance(data, (pd.Series, pd.Index)) else data
             )
             for item in iterable:
-                if item is None or pd.isna(item):
+                if item is None or (
+                    not isinstance(
+                        item, (list, tuple, np.ndarray, pd.Series, pd.Index, pa.Array)
+                    )
+                    and pd.isna(item)
+                ):
                     offsets.append(offsets[-1])
                 else:
-                    flat_values.extend(item)
-                    offsets.append(offsets[-1] + len(item))
+                    item_seq = cast(Sequence[Any], item)
+                    flat_values.extend(item_seq)
+                    offsets.append(offsets[-1] + len(item_seq))
             values_arr = pyarrow_array_from_sequence(flat_values, list_type.value_type)
             return pa.ListArray.from_arrays(pa.array(offsets), values_arr)
     return pa.array(data, type=pa_type)

From a02310d97b773a90684e54c2ec2ce5043ff0ee00 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Thu, 9 Jul 2026 22:25:17 +0000
Subject: [PATCH 13/18] offset converter should just do offset conversion

---
 packages/bigframes/bigframes/core/indexers.py | 41 ++++++++++---------
 1 file changed, 21 insertions(+), 20 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index 042fa88170a3..8ade090ff14f 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -279,7 +279,7 @@ def __setitem__(
         if not _is_noop_slice(row_indexer):
             raise NotImplementedError(_DATAFRAME_ILOC_ERROR)
 
-        col_offsets, _ = _iloc_col_indexer_to_offsets(self._dataframe, col_indexer)
+        col_offsets = _iloc_col_indexer_to_offsets(self._dataframe, col_indexer)
         df = self._dataframe._assign_multi_items_by_offsets(col_offsets, value)
         self._dataframe._set_block(df._get_block())
 
@@ -487,33 +487,33 @@ def _iloc_clip_to_offset(index: int, length: int, name: str) -> int:
     return index
 
 
+def _is_integer_scalar(value: Any) -> bool:
+    return not (
+        isinstance(value, bool)
+        or isinstance(value, np.bool_)
+        or (isinstance(value, pa.Scalar) and pyarrow.types.is_boolean(value))
+    ) and (
+        isinstance(value, numbers.Integral)
+        or (isinstance(value, pa.Scalar) and pyarrow.types.is_integer(value))
+    )
+
+
 def _iloc_col_indexer_to_offsets(
     df: bigframes.dataframe.DataFrame, col_indexer: Any
-) -> Tuple[Sequence[int], str]:
+) -> Sequence[int]:
     """Convert col_indexer from one of the many pandas-compatible formats to a list of offsets."""
     n_cols = len(df.columns)
 
-    if (
-        isinstance(col_indexer, bool)
-        or isinstance(col_indexer, np.bool_)
-        or (
-            isinstance(col_indexer, pa.Scalar) and pyarrow.types.is_boolean(col_indexer)
-        )
-    ):
-        raise TypeError("scalar boolean is not supported as an iloc column indexer")
-
-    if isinstance(col_indexer, numbers.Integral) or (
-        isinstance(col_indexer, pa.Scalar) and pyarrow.types.is_integer(col_indexer)
-    ):
+    if _is_integer_scalar(col_indexer):
         col_offset = int(col_indexer)
         return [
             _iloc_clip_to_offset(
                 col_offset, n_cols, "single positional iloc column indexer"
             )
-        ], "Series"
+        ]
 
     elif isinstance(col_indexer, slice):
-        return list(range(*col_indexer.indices(n_cols))), "DataFrame"
+        return list(range(*col_indexer.indices(n_cols)))
 
     elif pd.api.types.is_list_like(col_indexer):
         col_indexer_list = list(col_indexer)
@@ -525,12 +525,12 @@ def _iloc_col_indexer_to_offsets(
                 raise ValueError(
                     f"Boolean iloc column indexer has wrong length: {len(col_indexer_list)} instead of {n_cols}"
                 )
-            return [i for i, val in enumerate(col_indexer_list) if val], "DataFrame"
+            return [i for i, val in enumerate(col_indexer_list) if val]
         else:
             return [
                 _iloc_clip_to_offset(idx, n_cols, "iloc column indexer")
                 for idx in col_indexer_list
-            ], "DataFrame"
+            ]
 
     raise TypeError(f"got unexpected {type(col_indexer)} for iloc column indexer")
 
@@ -591,9 +591,10 @@ def _iloc_getitem_series_or_dataframe(
             return df.iat[key]
 
         row_indexer, column_indexer = key
-        column_offsets, output_type = _iloc_col_indexer_to_offsets(df, column_indexer)
+        column_offsets = _iloc_col_indexer_to_offsets(df, column_indexer)
         df_subset = _iloc_df_from_column_offsets(df, column_offsets)
-        if output_type == "Series":
+
+        if _is_integer_scalar(column_indexer):
             selected_columns = cast(
                 Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
                 df_subset[df_subset.columns[0]],

From d75dd98ede999732d63c36448ac990e584b15db6 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Fri, 10 Jul 2026 16:04:46 +0000
Subject: [PATCH 14/18] revert Series and DataFrame constructor changes and use
 pyarrow directly in unit test

---
 .../bigframes/bigframes/core/local_data.py    | 86 +------------------
 packages/bigframes/bigframes/dataframe.py     | 38 ++------
 packages/bigframes/bigframes/series.py        | 37 ++------
 .../tests/unit/display/test_anywidget.py      | 35 +++++---
 4 files changed, 35 insertions(+), 161 deletions(-)

diff --git a/packages/bigframes/bigframes/core/local_data.py b/packages/bigframes/bigframes/core/local_data.py
index 87f1ca0e6ba3..c05bda7a7fbc 100644
--- a/packages/bigframes/bigframes/core/local_data.py
+++ b/packages/bigframes/bigframes/core/local_data.py
@@ -22,17 +22,7 @@
 import itertools
 import json
 import uuid
-from typing import (
-    Any,
-    Callable,
-    Generator,
-    Iterable,
-    Literal,
-    Optional,
-    Sequence,
-    Union,
-    cast,
-)
+from typing import Any, Callable, Generator, Iterable, Literal, Optional, Union, cast
 
 import geopandas  # type: ignore
 import numpy
@@ -531,77 +521,3 @@ def _schema_durations_to_ints(schema: pa.Schema) -> pa.Schema:
     return pa.schema(
         pa.field(field.name, _durations_to_ints(field.type)) for field in schema
     )
-
-
-def _has_extension_type(pa_type: pa.DataType) -> bool:
-    """Check if the given pyarrow DataType contains an ExtensionType at any nesting level."""
-    if isinstance(pa_type, pa.ExtensionType):
-        return True
-    if pa.types.is_struct(pa_type):
-        struct_type = cast(pa.StructType, pa_type)
-        return any(
-            _has_extension_type(struct_type.field(i).type)
-            for i in range(struct_type.num_fields)
-        )
-    if pa.types.is_list(pa_type) or pa.types.is_large_list(pa_type):
-        list_type = cast(pa.ListType, pa_type)
-        return _has_extension_type(list_type.value_type)
-    return False
-
-
-def pyarrow_array_from_sequence(data: Any, pa_type: pa.DataType) -> pa.Array:
-    """Construct a pyarrow Array from a Sequence, DataFrame, or Series,
-    recursively handling ExtensionTypes inside structs and lists.
-
-    This is a workaround for issue b/456774399 where ArrowNotImplementedError is
-    thrown when constructing arrays of structs containing extension dtypes like
-    JSON.
-    """
-    if isinstance(data, (pa.Array, pa.ChunkedArray)):
-        return data.cast(pa_type) if data.type != pa_type else data
-    if isinstance(pa_type, pa.ExtensionType):
-        storage_arr = pyarrow_array_from_sequence(data, pa_type.storage_type)
-        return pa.ExtensionArray.from_storage(pa_type, storage_arr)
-    if pa.types.is_struct(pa_type):
-        struct_type = cast(pa.StructType, pa_type)
-        if isinstance(data, (pd.Series, pd.Index)):
-            df = pd.DataFrame(data.tolist())
-        elif isinstance(data, pd.DataFrame):
-            df = data
-        else:
-            df = pd.DataFrame(
-                list(data)
-                if not isinstance(data, (list, tuple, np.ndarray, pd.Series, pd.Index))
-                else data
-            )
-        arrays = []
-        fields = []
-        for i in range(struct_type.num_fields):
-            field = struct_type.field(i)
-            col_data = df[field.name] if field.name in df.columns else [None] * len(df)
-            arrays.append(pyarrow_array_from_sequence(col_data, field.type))
-            fields.append(field)
-        return pa.StructArray.from_arrays(arrays, fields=fields)
-    if pa.types.is_list(pa_type) or pa.types.is_large_list(pa_type):
-        list_type = cast(pa.ListType, pa_type)
-        if _has_extension_type(list_type.value_type):
-            flat_values: list[Any] = []
-            offsets = [0]
-            iterable = (
-                data.tolist() if isinstance(data, (pd.Series, pd.Index)) else data
-            )
-            for item in iterable:
-                if item is None or (
-                    not isinstance(
-                        item, (list, tuple, np.ndarray, pd.Series, pd.Index, pa.Array)
-                    )
-                    and pd.isna(item)
-                ):
-                    offsets.append(offsets[-1])
-                else:
-                    item_seq = cast(Sequence[Any], item)
-                    flat_values.extend(item_seq)
-                    offsets.append(offsets[-1] + len(item_seq))
-            values_arr = pyarrow_array_from_sequence(flat_values, list_type.value_type)
-            return pa.ListArray.from_arrays(pa.array(offsets), values_arr)
-    return pa.array(data, type=pa_type)
diff --git a/packages/bigframes/bigframes/dataframe.py b/packages/bigframes/bigframes/dataframe.py
index 7668aeb17feb..07ba6aba1c0e 100644
--- a/packages/bigframes/bigframes/dataframe.py
+++ b/packages/bigframes/bigframes/dataframe.py
@@ -221,40 +221,14 @@ def __init__(
             if isinstance(dtype, str) and dtype.lower() == "json":
                 dtype = bigframes.dtypes.JSON_DTYPE
 
-            import bigframes.core.local_data
             import bigframes.pandas
 
-            if (
-                hasattr(pandas.arrays, "ArrowExtensionArray")
-                and isinstance(dtype, pandas.ArrowDtype)
-                and not isinstance(
-                    data,
-                    (
-                        pandas.DataFrame,
-                        pandas.Series,
-                        pandas.Index,
-                        pyarrow.Array,
-                        pyarrow.ChunkedArray,
-                        pyarrow.Table,
-                        pandas.arrays.ArrowExtensionArray,
-                    ),
-                )
-            ):
-                pa_array = bigframes.core.local_data.pyarrow_array_from_sequence(
-                    data, dtype.pyarrow_dtype
-                )
-                pd_dataframe = pandas.DataFrame(
-                    data=pandas.arrays.ArrowExtensionArray(pa_array),
-                    index=index,  # type:ignore
-                    columns=columns,  # type:ignore
-                )
-            else:
-                pd_dataframe = pandas.DataFrame(
-                    data=data,
-                    index=index,  # type:ignore
-                    columns=columns,  # type:ignore
-                    dtype=dtype,  # type:ignore
-                )
+            pd_dataframe = pandas.DataFrame(
+                data=data,
+                index=index,  # type:ignore
+                columns=columns,  # type:ignore
+                dtype=dtype,  # type:ignore
+            )
             if session:
                 block = session.read_pandas(pd_dataframe)._get_block()
             else:
diff --git a/packages/bigframes/bigframes/series.py b/packages/bigframes/bigframes/series.py
index 56baf7e682a1..1a56c76b0237 100644
--- a/packages/bigframes/bigframes/series.py
+++ b/packages/bigframes/bigframes/series.py
@@ -196,37 +196,12 @@ def __init__(
             if isinstance(dtype, str) and dtype.lower() == "json":
                 dtype = bigframes.dtypes.JSON_DTYPE
 
-            import bigframes.core.local_data
-
-            if (
-                hasattr(pandas.arrays, "ArrowExtensionArray")
-                and isinstance(dtype, pandas.ArrowDtype)
-                and not isinstance(
-                    data,
-                    (
-                        pandas.Series,
-                        pandas.Index,
-                        pa.Array,
-                        pa.ChunkedArray,
-                        pandas.arrays.ArrowExtensionArray,
-                    ),
-                )
-            ):
-                pa_array = bigframes.core.local_data.pyarrow_array_from_sequence(
-                    data, dtype.pyarrow_dtype
-                )
-                pd_series = pandas.Series(
-                    data=pandas.arrays.ArrowExtensionArray(pa_array),
-                    index=index,  # type:ignore
-                    name=name,
-                )
-            else:
-                pd_series = pandas.Series(
-                    data=data,
-                    index=index,  # type:ignore
-                    dtype=dtype,  # type:ignore
-                    name=name,
-                )
+            pd_series = pandas.Series(
+                data=data,
+                index=index,  # type:ignore
+                dtype=dtype,  # type:ignore
+                name=name,
+            )
             block = read_pandas_func(pd_series)._get_block()  # type:ignore
 
         assert block is not None
diff --git a/packages/bigframes/tests/unit/display/test_anywidget.py b/packages/bigframes/tests/unit/display/test_anywidget.py
index 2e2800064124..b50fcf380d42 100644
--- a/packages/bigframes/tests/unit/display/test_anywidget.py
+++ b/packages/bigframes/tests/unit/display/test_anywidget.py
@@ -16,6 +16,8 @@
 import unittest.mock as mock
 
 import pandas as pd
+import pyarrow as pa
+import pyarrow.compute
 import pytest
 
 import bigframes
@@ -25,10 +27,9 @@
 pytest.importorskip("traitlets")
 
 import bigframes.dataframe
-import bigframes.operations
+import bigframes.dtypes
 import bigframes.series
 from bigframes.display.anywidget import TableWidget
-from bigframes.dtypes import JSON_DTYPE, STRING_DTYPE, struct_type
 
 
 def test_navigation_to_invalid_page_resets_to_valid_page_without_deadlock():
@@ -195,31 +196,39 @@ def test_cell_execution_count_propagation(mock_df):
 
 def test_json_column_converted_to_string_for_display(polars_session):
     series = bigframes.series.Series(
-        ['{"a": 1}', '{"b": 2}'], dtype=JSON_DTYPE, session=polars_session
+        ['{"a": 1}', '{"b": 2}'], dtype=bigframes.dtypes.JSON_DTYPE, session=polars_session
     )
     df = series.to_frame("col_json")
 
     result = df._prepare_display_df()
 
-    assert result["col_json"].dtype == STRING_DTYPE
+    assert result["col_json"].dtype == bigframes.dtypes.STRING_DTYPE
 
 
 def test_struct_column_with_nested_json_converted_to_string_for_display(
     polars_session,
 ):
-    nested_struct_dtype = struct_type(
-        [("field1", STRING_DTYPE), ("field2", JSON_DTYPE)]
+    if not hasattr(pa, "json_"):
+        pytest.skip(reason=f"pyarrow=={pa.__version__} does not support json_")
+
+    # Arrange
+    json_type = pa.json_(storage_type=pa.utf8())
+    json_data = pa.array(['{"a": 1}'], type=json_type)
+    string_data = pa.array(["hello"], type=pa.string())
+    struct_data = pa.StructArray.from_arrays(
+        [string_data, json_data], names=["field1", "field2"]
     )
-    series = bigframes.series.Series(
-        [{"field1": "hello", "field2": '{"a": 1}'}],
-        dtype=nested_struct_dtype,
-        session=polars_session,
-    )
-    df = series.to_frame("col_struct")
+    nested_data = pa.table([struct_data], names=["nested"])
+    df = polars_session.read_arrow(nested_data)
+    exploded = df["nested"].struct.explode()
+    # Ensure that we are actually using the JSON dtype in this test.
+    assert exploded["field2"].dtype == bigframes.dtypes.JSON_DTYPE
 
+    # Act
     result = df._prepare_display_df()
 
-    assert result["col_struct"].dtype == STRING_DTYPE
+    # Assert
+    assert result["nested"].dtype == bigframes.dtypes.STRING_DTYPE
 
 
 @pytest.fixture

From 13dd27a8c4ffbf897cd52657cef4e2c8a7217877 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Fri, 10 Jul 2026 16:05:13 +0000
Subject: [PATCH 15/18] nox format

---
 packages/bigframes/tests/unit/display/test_anywidget.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/packages/bigframes/tests/unit/display/test_anywidget.py b/packages/bigframes/tests/unit/display/test_anywidget.py
index b50fcf380d42..f9aeeeac18f3 100644
--- a/packages/bigframes/tests/unit/display/test_anywidget.py
+++ b/packages/bigframes/tests/unit/display/test_anywidget.py
@@ -17,7 +17,6 @@
 
 import pandas as pd
 import pyarrow as pa
-import pyarrow.compute
 import pytest
 
 import bigframes
@@ -196,7 +195,9 @@ def test_cell_execution_count_propagation(mock_df):
 
 def test_json_column_converted_to_string_for_display(polars_session):
     series = bigframes.series.Series(
-        ['{"a": 1}', '{"b": 2}'], dtype=bigframes.dtypes.JSON_DTYPE, session=polars_session
+        ['{"a": 1}', '{"b": 2}'],
+        dtype=bigframes.dtypes.JSON_DTYPE,
+        session=polars_session,
     )
     df = series.to_frame("col_json")
 

From db5f9f0eeab6bfd1be15edfafac55006d05027fa Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Fri, 10 Jul 2026 17:03:35 +0000
Subject: [PATCH 16/18] test numpy and pyarrow scalars more thoroughly

---
 packages/bigframes/bigframes/core/indexers.py | 124 ++++++--
 .../generative_ai/ai_functions.ipynb          |  16 +-
 .../bigframes/tests/unit/test_iloc_getitem.py | 297 ++++++++++++++++++
 .../bigframes/tests/unit/test_iloc_setitem.py | 153 +++++----
 4 files changed, 471 insertions(+), 119 deletions(-)
 create mode 100644 packages/bigframes/tests/unit/test_iloc_getitem.py

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index 8ade090ff14f..7748f3536509 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -121,9 +121,9 @@ def __init__(self, series: bigframes.series.Series):
         self._series = series
 
     def __getitem__(self, key: int) -> bigframes.core.scalar.Scalar:
-        if not isinstance(key, int):
+        if not _is_integer_scalar(key):
             raise ValueError("Series iAt based indexing can only have integer indexers")
-        return self._series.iloc[key]
+        return self._series.iloc[_to_python_int(key)]
 
 
 class AtSeriesIndexer:
@@ -291,19 +291,21 @@ def __init__(self, dataframe: bigframes.dataframe.DataFrame):
     def __getitem__(self, key: tuple) -> bigframes.core.scalar.Scalar:
         error_message = "DataFrame.iat should be indexed by a tuple of exactly 2 ints"
         # we raise TypeError or ValueError under the same conditions that pandas does
-        if isinstance(key, int):
+        if _is_integer_scalar(key):
             raise TypeError(error_message)
         if not isinstance(key, tuple):
             raise ValueError(error_message)
-        key_values_are_ints = [isinstance(key_value, int) for key_value in key]
+        key_values_are_ints = [_is_integer_scalar(key_value) for key_value in key]
         if not all(key_values_are_ints):
             raise ValueError(error_message)
         if len(key) != 2:
             raise TypeError(error_message)
+        row_idx = _to_python_int(key[0])
+        col_idx = _to_python_int(key[1])
         block: bigframes.core.blocks.Block = self._dataframe._block
-        column_block = block.select_columns([block.value_columns[key[1]]])
+        column_block = block.select_columns([block.value_columns[col_idx]])
         column = bigframes.series.Series(column_block)
-        return column.iloc[key[0]]
+        return column.iloc[row_idx]
 
 
 class AtDataFrameIndexer:
@@ -475,29 +477,72 @@ def _struct_accessor_check_and_warn(
         warnings.warn(msg, stacklevel=7, category=bfe.BadIndexerKeyWarning)
 
 
-def _iloc_clip_to_offset(index: int, length: int, name: str) -> int:
+def _to_python_int(value: Any) -> int:
+    if isinstance(value, pa.Scalar):
+        return int(value.as_py())
+    return int(value)
+
+
+def _iloc_clip_to_offset(index: Any, length: int, name: str) -> int:
     """Support negative values for offsets."""
-    offset = index
+    if not _is_integer_scalar(index):
+        raise TypeError(f"got unexpected {type(index)} for {name}")
+    offset = _to_python_int(index)
     if offset < 0:
         offset += length
 
     if offset < 0 or offset >= length:
         raise IndexError(f"{name} {index} is out-of-bounds")
 
-    return index
+    return offset
 
 
 def _is_integer_scalar(value: Any) -> bool:
     return not (
         isinstance(value, bool)
         or isinstance(value, np.bool_)
-        or (isinstance(value, pa.Scalar) and pyarrow.types.is_boolean(value))
+        or (isinstance(value, pa.Scalar) and pyarrow.types.is_boolean(value.type))
     ) and (
         isinstance(value, numbers.Integral)
-        or (isinstance(value, pa.Scalar) and pyarrow.types.is_integer(value))
+        or (isinstance(value, pa.Scalar) and pyarrow.types.is_integer(value.type))
     )
 
 
+def _is_boolean_scalar(value: Any) -> bool:
+    return (
+        isinstance(value, bool)
+        or isinstance(value, np.bool_)
+        or (isinstance(value, pa.Scalar) and pyarrow.types.is_boolean(value.type))
+    )
+
+
+def _truth_val(value: Any) -> bool:
+    if value is None or value is pd.NA or pd.isna(value):
+        return False
+    if isinstance(value, pa.Scalar):
+        return bool(value.as_py()) if value.is_valid else False
+    return bool(value)
+
+
+def _is_boolean_indexer(indexer: Any) -> bool:
+    if hasattr(indexer, "dtype") and pd.api.types.is_bool_dtype(indexer.dtype):
+        return True
+    if hasattr(indexer, "type") and isinstance(indexer.type, pa.DataType) and pyarrow.types.is_boolean(indexer.type):
+        return True
+    if pd.api.types.is_list_like(indexer):
+        lst = (
+            list(indexer)
+            if not isinstance(indexer, (bigframes.series.Series, indexes.Index))
+            else list(indexer.to_pandas())
+        )
+        if len(lst) > 0 and all(
+            _is_boolean_scalar(x) or (x is None) or (x is pd.NA) or pd.isna(x)
+            for x in lst
+        ):
+            return any(_is_boolean_scalar(x) for x in lst)
+    return False
+
+
 def _iloc_col_indexer_to_offsets(
     df: bigframes.dataframe.DataFrame, col_indexer: Any
 ) -> Sequence[int]:
@@ -505,7 +550,7 @@ def _iloc_col_indexer_to_offsets(
     n_cols = len(df.columns)
 
     if _is_integer_scalar(col_indexer):
-        col_offset = int(col_indexer)
+        col_offset = _to_python_int(col_indexer)
         return [
             _iloc_clip_to_offset(
                 col_offset, n_cols, "single positional iloc column indexer"
@@ -515,22 +560,20 @@ def _iloc_col_indexer_to_offsets(
     elif isinstance(col_indexer, slice):
         return list(range(*col_indexer.indices(n_cols)))
 
-    elif pd.api.types.is_list_like(col_indexer):
+    elif _is_boolean_indexer(col_indexer):
         col_indexer_list = list(col_indexer)
+        if len(col_indexer_list) != n_cols:
+            raise ValueError(
+                f"Boolean iloc column indexer has wrong length: {len(col_indexer_list)} instead of {n_cols}"
+            )
+        return [i for i, val in enumerate(col_indexer_list) if _truth_val(val)]
 
-        if len(col_indexer_list) > 0 and all(
-            isinstance(x, bool) for x in col_indexer_list
-        ):
-            if len(col_indexer_list) != n_cols:
-                raise ValueError(
-                    f"Boolean iloc column indexer has wrong length: {len(col_indexer_list)} instead of {n_cols}"
-                )
-            return [i for i, val in enumerate(col_indexer_list) if val]
-        else:
-            return [
-                _iloc_clip_to_offset(idx, n_cols, "iloc column indexer")
-                for idx in col_indexer_list
-            ]
+    elif pd.api.types.is_list_like(col_indexer):
+        col_indexer_list = list(col_indexer)
+        return [
+            _iloc_clip_to_offset(idx, n_cols, "iloc column indexer")
+            for idx in col_indexer_list
+        ]
 
     raise TypeError(f"got unexpected {type(col_indexer)} for iloc column indexer")
 
@@ -564,9 +607,10 @@ def _iloc_getitem_series_or_dataframe(
     bigframes.core.scalar.Scalar,
     pd.Series,
 ]:
-    if isinstance(key, int):
-        stop_key = key + 1 if key != -1 else None
-        internal_slice_result = series_or_dataframe._slice(key, stop_key, 1)
+    if _is_integer_scalar(key):
+        key_int = _to_python_int(key)
+        stop_key = key_int + 1 if key_int != -1 else None
+        internal_slice_result = series_or_dataframe._slice(key_int, stop_key, 1)
         result_pd_df = internal_slice_result.to_pandas()
         if result_pd_df.empty:
             raise IndexError("single positional indexer is out-of-bounds")
@@ -587,7 +631,7 @@ def _iloc_getitem_series_or_dataframe(
 
         # len(key) == 2
         df = typing.cast(bigframes.dataframe.DataFrame, series_or_dataframe)
-        if isinstance(key[0], int) and isinstance(key[1], int):
+        if _is_integer_scalar(key[0]) and _is_integer_scalar(key[1]):
             return df.iat[key]
 
         row_indexer, column_indexer = key
@@ -610,6 +654,26 @@ def _iloc_getitem_series_or_dataframe(
                 series_or_dataframe.iloc[0:0],
             )
 
+        if _is_boolean_indexer(key):
+            key_list = (
+                list(key)
+                if not isinstance(key, (bigframes.series.Series, indexes.Index))
+                else list(key.to_pandas())
+            )
+            n_rows = len(series_or_dataframe)
+            if len(key_list) != n_rows:
+                raise IndexError(
+                    f"Boolean index has wrong length: {len(key_list)} instead of {n_rows}"
+                )
+            key = [i for i, val in enumerate(key_list) if _truth_val(val)]
+            if len(key) == 0:
+                return typing.cast(
+                    Union[bigframes.dataframe.DataFrame, bigframes.series.Series],
+                    series_or_dataframe.iloc[0:0],
+                )
+        else:
+            key = [_to_python_int(k) for k in list(key)]
+
         # Check if both positive index and negative index are necessary
         if isinstance(key, (bigframes.series.Series, indexes.Index)):
             # Avoid data download
diff --git a/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb b/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
index beaf12a64e39..0831ea0412b0 100644
--- a/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
+++ b/packages/bigframes/notebooks/generative_ai/ai_functions.ipynb
@@ -2,7 +2,7 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": null,
    "id": "acd53f9d",
    "metadata": {},
    "outputs": [],
@@ -71,7 +71,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": null,
    "id": "c9f924aa",
    "metadata": {},
    "outputs": [],
@@ -97,7 +97,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 3,
+   "execution_count": null,
    "id": "471a47fe",
    "metadata": {},
    "outputs": [
@@ -158,7 +158,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 4,
+   "execution_count": null,
    "id": "4a3229a8",
    "metadata": {},
    "outputs": [
@@ -213,7 +213,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 5,
+   "execution_count": null,
    "id": "718c6622",
    "metadata": {},
    "outputs": [
@@ -301,7 +301,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 6,
+   "execution_count": null,
    "id": "6875fe36",
    "metadata": {},
    "outputs": [
@@ -390,7 +390,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 7,
+   "execution_count": null,
    "id": "8cfb844b",
    "metadata": {},
    "outputs": [
@@ -477,7 +477,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 8,
+   "execution_count": null,
    "id": "2e66110a",
    "metadata": {},
    "outputs": [
diff --git a/packages/bigframes/tests/unit/test_iloc_getitem.py b/packages/bigframes/tests/unit/test_iloc_getitem.py
new file mode 100644
index 000000000000..7f030a16c92b
--- /dev/null
+++ b/packages/bigframes/tests/unit/test_iloc_getitem.py
@@ -0,0 +1,297 @@
+# Copyright 2026 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.
+
+from typing import Generator
+
+import numpy as np
+import pandas as pd
+import pyarrow as pa
+import pytest
+
+import bigframes
+import bigframes.pandas as bpd
+from bigframes.testing.utils import assert_frame_equal, assert_series_equal
+
+pytest.importorskip("polars")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def session() -> Generator[bigframes.Session, None, None]:
+    import bigframes.core.global_session
+    from bigframes.testing import polars_session
+
+    session = polars_session.TestSession()
+    with bigframes.core.global_session._GlobalSessionContext(session):
+        yield session
+
+
+@pytest.fixture
+def sample_df() -> bpd.DataFrame:
+    pd_df = pd.DataFrame(
+        {
+            "A": [1, 2, 3],
+            "B": [4, 5, 6],
+            "C": [7, 8, 9],
+        }
+    )
+    return bpd.read_pandas(pd_df)
+
+
+@pytest.fixture
+def unordered_sample_df(
+    sample_df: bpd.DataFrame,
+) -> Generator[bpd.DataFrame, None, None]:
+    session = sample_df._session
+    original_strictly_ordered = session._strictly_ordered
+    original_allow_ambiguity = session._allow_ambiguity
+
+    try:
+        session._strictly_ordered = False
+        session._allow_ambiguity = True
+
+        import unittest.mock as mock
+
+        with (
+            mock.patch.object(
+                type(sample_df._block.expr),
+                "order_ambiguous",
+                new_callable=mock.PropertyMock,
+            ) as mock_ambiguous,
+            mock.patch.object(
+                type(sample_df._block),
+                "explicitly_ordered",
+                new_callable=mock.PropertyMock,
+            ) as mock_explicit,
+        ):
+            mock_ambiguous.return_value = True
+            mock_explicit.return_value = False
+            yield sample_df
+    finally:
+        session._strictly_ordered = original_strictly_ordered
+        session._allow_ambiguity = original_allow_ambiguity
+
+
+@pytest.fixture
+def duplicate_columns_df() -> bpd.DataFrame:
+    pd_df = pd.DataFrame(
+        [[1, 2, 3], [4, 5, 6], [7, 8, 9]],
+        columns=["A", "B", "A"],
+    )
+    return bpd.read_pandas(pd_df)
+
+
+def test_iloc_getitem_column_single_integer(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, 1].to_pandas()
+    pd_result = pd_df.iloc[:, 1]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_column_numpy_scalar(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, np.int64(1)].to_pandas()
+    pd_result = pd_df.iloc[:, np.int64(1)]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_columns_numpy_array(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, np.array([0, 2], dtype=np.int64)].to_pandas()
+    pd_result = pd_df.iloc[:, np.array([0, 2], dtype=np.int64)]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_column_pyarrow_scalar(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, pa.scalar(1, type=pa.int64())].to_pandas()
+    pd_result = pd_df.iloc[:, 1]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_columns_pyarrow_array(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, pa.array([0, 2], type=pa.int64())].to_pandas()
+    pd_result = pd_df.iloc[:, pa.array([0, 2], type=pa.int64())]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_row_numpy_scalar(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[np.int64(1)]
+    pd_result = pd_df.iloc[np.int64(1)]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_rows_numpy_array(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[np.array([0, 2], dtype=np.int64)].to_pandas()
+    pd_result = pd_df.iloc[np.array([0, 2], dtype=np.int64)]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_row_pyarrow_scalar(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[pa.scalar(1, type=pa.int64())]
+    pd_result = pd_df.iloc[1]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_rows_pyarrow_array(sample_df):
+    bf_df = sample_df
+    pd_df = sample_df.to_pandas()
+
+    bf_result = bf_df.iloc[pa.array([0, 2], type=pa.int64())].to_pandas()
+    pd_result = pd_df.iloc[pa.array([0, 2], type=pa.int64())]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+@pytest.mark.parametrize(
+    ["key", "value", "expected_error"],
+    [
+        pytest.param((slice(None), 1), None, None, id="col_index"),
+        pytest.param((slice(0, None), 1), None, None, id="col_index_slice_0_none"),
+        pytest.param(
+            (slice(None, None, 1), 1), None, None, id="col_index_slice_none_none_1"
+        ),
+        pytest.param(
+            (slice(1, None), 1),
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="col_index_slice_1_none",
+        ),
+        pytest.param(
+            (slice(None, 2), 1),
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="col_index_slice_none_2",
+        ),
+        pytest.param((slice(None), 1), 99, None, id="col_setitem"),
+        pytest.param(
+            (1, slice(None)),
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="row_index_slice",
+        ),
+        pytest.param(
+            1,
+            None,
+            bigframes.exceptions.OrderRequiredError,
+            id="single_row_index",
+        ),
+    ],
+)
+def test_iloc_getitem_unordered(unordered_sample_df, key, value, expected_error):
+    if value is not None:
+        bf_df = unordered_sample_df.copy()
+        bf_df.iloc[key] = value
+    elif expected_error is not None:
+        with pytest.raises(expected_error):
+            unordered_sample_df.iloc[key]
+    else:
+        unordered_sample_df.iloc[key]
+
+
+def test_iloc_getitem_duplicate_columns_single_integer(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, 2].to_pandas()
+    pd_result = pd_df.iloc[:, 2]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_duplicate_columns_list_integer(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, [0, 2]].to_pandas()
+    pd_result = pd_df.iloc[:, [0, 2]]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_duplicate_columns_slice(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, 1:3].to_pandas()
+    pd_result = pd_df.iloc[:, 1:3]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_duplicate_columns_numpy_scalar(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, np.int64(2)].to_pandas()
+    pd_result = pd_df.iloc[:, np.int64(2)]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_duplicate_columns_numpy_array(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, np.array([0, 2], dtype=np.int64)].to_pandas()
+    pd_result = pd_df.iloc[:, np.array([0, 2], dtype=np.int64)]
+
+    assert_frame_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_duplicate_columns_pyarrow_scalar(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, pa.scalar(2, type=pa.int64())].to_pandas()
+    pd_result = pd_df.iloc[:, 2]
+
+    assert_series_equal(bf_result, pd_result)
+
+
+def test_iloc_getitem_duplicate_columns_pyarrow_array(duplicate_columns_df):
+    bf_df = duplicate_columns_df
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_result = bf_df.iloc[:, pa.array([0, 2], type=pa.int64())].to_pandas()
+    pd_result = pd_df.iloc[:, pa.array([0, 2], type=pa.int64())]
+
+    assert_frame_equal(bf_result, pd_result)
diff --git a/packages/bigframes/tests/unit/test_iloc_setitem.py b/packages/bigframes/tests/unit/test_iloc_setitem.py
index 9661026763f3..9e59b39a8b65 100644
--- a/packages/bigframes/tests/unit/test_iloc_setitem.py
+++ b/packages/bigframes/tests/unit/test_iloc_setitem.py
@@ -16,6 +16,7 @@
 
 import numpy as np
 import pandas as pd
+import pyarrow as pa
 import pytest
 
 import bigframes
@@ -109,94 +110,44 @@ def test_iloc_setitem_columns_dataframe(sample_df):
     assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-def test_iloc_getitem_column_single_integer(sample_df):
-    bf_df = sample_df
+def test_iloc_setitem_column_numpy_scalar(sample_df):
+    bf_df = sample_df.copy()
     pd_df = sample_df.to_pandas()
 
-    bf_result = bf_df.iloc[:, 1].to_pandas()
-    pd_result = pd_df.iloc[:, 1]
+    bf_df.iloc[:, np.int64(1)] = 99
+    pd_df.iloc[:, np.int64(1)] = 99
 
-    assert_series_equal(bf_result, pd_result)
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
-@pytest.fixture
-def unordered_sample_df(
-    sample_df: bpd.DataFrame,
-) -> Generator[bpd.DataFrame, None, None]:
-    session = sample_df._session
-    original_strictly_ordered = session._strictly_ordered
-    original_allow_ambiguity = session._allow_ambiguity
-
-    try:
-        session._strictly_ordered = False
-        session._allow_ambiguity = True
-
-        import unittest.mock as mock
-
-        with (
-            mock.patch.object(
-                type(sample_df._block.expr),
-                "order_ambiguous",
-                new_callable=mock.PropertyMock,
-            ) as mock_ambiguous,
-            mock.patch.object(
-                type(sample_df._block),
-                "explicitly_ordered",
-                new_callable=mock.PropertyMock,
-            ) as mock_explicit,
-        ):
-            mock_ambiguous.return_value = True
-            mock_explicit.return_value = False
-            yield sample_df
-    finally:
-        session._strictly_ordered = original_strictly_ordered
-        session._allow_ambiguity = original_allow_ambiguity
+def test_iloc_setitem_columns_numpy_array(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
 
+    bf_df.iloc[:, np.array([0, 2], dtype=np.int64)] = [99, 88]
+    pd_df.iloc[:, np.array([0, 2], dtype=np.int64)] = [99, 88]
 
-@pytest.mark.parametrize(
-    ["key", "value", "expected_error"],
-    [
-        pytest.param((slice(None), 1), None, None, id="col_index"),
-        pytest.param((slice(0, None), 1), None, None, id="col_index_slice_0_none"),
-        pytest.param(
-            (slice(None, None, 1), 1), None, None, id="col_index_slice_none_none_1"
-        ),
-        pytest.param(
-            (slice(1, None), 1),
-            None,
-            bigframes.exceptions.OrderRequiredError,
-            id="col_index_slice_1_none",
-        ),
-        pytest.param(
-            (slice(None, 2), 1),
-            None,
-            bigframes.exceptions.OrderRequiredError,
-            id="col_index_slice_none_2",
-        ),
-        pytest.param((slice(None), 1), 99, None, id="col_setitem"),
-        pytest.param(
-            (1, slice(None)),
-            None,
-            bigframes.exceptions.OrderRequiredError,
-            id="row_index_slice",
-        ),
-        pytest.param(
-            1,
-            None,
-            bigframes.exceptions.OrderRequiredError,
-            id="single_row_index",
-        ),
-    ],
-)
-def test_iloc_getitem_unordered(unordered_sample_df, key, value, expected_error):
-    if value is not None:
-        bf_df = unordered_sample_df.copy()
-        bf_df.iloc[key] = value
-    elif expected_error is not None:
-        with pytest.raises(expected_error):
-            unordered_sample_df.iloc[key]
-    else:
-        unordered_sample_df.iloc[key]
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_column_pyarrow_scalar(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    bf_df.iloc[:, pa.scalar(1, type=pa.int64())] = 99
+    pd_df.iloc[:, 1] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_columns_pyarrow_array(sample_df):
+    bf_df = sample_df.copy()
+    pd_df = sample_df.to_pandas()
+
+    bf_df.iloc[:, pa.array([0, 2], type=pa.int64())] = [99, 88]
+    pd_df.iloc[:, pa.array([0, 2], type=pa.int64())] = [99, 88]
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
 
 
 @pytest.mark.parametrize(
@@ -252,3 +203,43 @@ def test_iloc_setitem_duplicate_columns_slice(duplicate_columns_df):
     pd_df.iloc[:, 1:3] = 99
 
     assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_numpy_scalar(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, np.int64(2)] = 99
+    pd_df.iloc[:, np.int64(2)] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_numpy_array(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, np.array([0, 2], dtype=np.int64)] = [99, 88]
+    pd_df.iloc[:, np.array([0, 2], dtype=np.int64)] = [99, 88]
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_pyarrow_scalar(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, pa.scalar(2, type=pa.int64())] = 99
+    pd_df.iloc[:, 2] = 99
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)
+
+
+def test_iloc_setitem_duplicate_columns_pyarrow_array(duplicate_columns_df):
+    bf_df = duplicate_columns_df.copy()
+    pd_df = duplicate_columns_df.to_pandas()
+
+    bf_df.iloc[:, pa.array([0, 2], type=pa.int64())] = [99, 88]
+    pd_df.iloc[:, pa.array([0, 2], type=pa.int64())] = [99, 88]
+
+    assert_frame_equal(bf_df.to_pandas(), pd_df)

From 4ec9c20a14ea86870a03171b98dfda3f161b5206 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Fri, 10 Jul 2026 17:06:37 +0000
Subject: [PATCH 17/18] nox format

---
 packages/bigframes/bigframes/core/indexers.py      | 6 +++++-
 packages/bigframes/tests/unit/test_iloc_setitem.py | 2 +-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index 7748f3536509..f23eb767017a 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -527,7 +527,11 @@ def _truth_val(value: Any) -> bool:
 def _is_boolean_indexer(indexer: Any) -> bool:
     if hasattr(indexer, "dtype") and pd.api.types.is_bool_dtype(indexer.dtype):
         return True
-    if hasattr(indexer, "type") and isinstance(indexer.type, pa.DataType) and pyarrow.types.is_boolean(indexer.type):
+    if (
+        hasattr(indexer, "type")
+        and isinstance(indexer.type, pa.DataType)
+        and pyarrow.types.is_boolean(indexer.type)
+    ):
         return True
     if pd.api.types.is_list_like(indexer):
         lst = (
diff --git a/packages/bigframes/tests/unit/test_iloc_setitem.py b/packages/bigframes/tests/unit/test_iloc_setitem.py
index 9e59b39a8b65..98c515fd4eeb 100644
--- a/packages/bigframes/tests/unit/test_iloc_setitem.py
+++ b/packages/bigframes/tests/unit/test_iloc_setitem.py
@@ -21,7 +21,7 @@
 
 import bigframes
 import bigframes.pandas as bpd
-from bigframes.testing.utils import assert_frame_equal, assert_series_equal
+from bigframes.testing.utils import assert_frame_equal
 
 pytest.importorskip("polars")
 

From 4f86564e2e7081ba25be918f90d241a9179d6ad3 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Tim=20Swe=C3=B1a?= <swast@google.com>
Date: Fri, 10 Jul 2026 17:15:30 +0000
Subject: [PATCH 18/18] mypy

---
 packages/bigframes/bigframes/core/indexers.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/bigframes/bigframes/core/indexers.py b/packages/bigframes/bigframes/core/indexers.py
index f23eb767017a..faf76f7d4f08 100644
--- a/packages/bigframes/bigframes/core/indexers.py
+++ b/packages/bigframes/bigframes/core/indexers.py
@@ -24,7 +24,7 @@
 import numpy as np
 import pandas as pd
 import pyarrow as pa
-import pyarrow.types
+import pyarrow.types  # type: ignore
 
 import bigframes.core.blocks
 import bigframes.core.col