From f8d3b11818b2fa3cb3ce049c71882592727ab5c3 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Thu, 15 Oct 2026 20:03:37 +0000
Subject: [PATCH 01/34] fix(bigtable): resolve O(n^2) bottleneck in
 ReadRowsQuery.add_range

---
 .../cloud/bigtable/data/read_rows_query.py    | 22 ++++++++++++++++--
 .../tests/unit/data/test_read_rows_query.py   | 23 +++++++++++++++++++
 2 files changed, 43 insertions(+), 2 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 9fd42e2d871a..1038ce0bafe8 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -87,6 +87,17 @@ def __init__(
                 init_dict["end_key_open"] = end_key
         self._pb = RowRangePB(**init_dict)
 
+    def _key(self) -> tuple[bytes | None, bool, bytes | None, bool]:
+        return (
+            self.start_key,
+            self.start_is_inclusive,
+            self.end_key,
+            self.end_is_inclusive,
+        )
+
+    def __hash__(self) -> int:
+        return hash(self._key())
+
     @property
     def start_key(self) -> bytes | None:
         """
@@ -186,7 +197,7 @@ def __bool__(self) -> bool:
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, RowRange):
             return NotImplemented
-        return self._pb == other._pb
+        return self._key() == other._key()
 
     def __str__(self) -> str:
         """
@@ -250,6 +261,11 @@ def __init__(
         self._row_set = RowSetPB(
             row_keys=row_keys, row_ranges=[r._pb for r in row_ranges]
         )
+        # Pre-initialize our signature set using hashable RowRange wrappers
+        self._range_signatures = set(
+            RowRange._from_pb(r) for r in self._row_set.row_ranges
+        )
+
         self.limit = limit or None
         self.filter = row_filter
 
@@ -349,7 +365,9 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
-        if row_range not in self.row_ranges:
+        # Utilize pure O(1) set-membership checking
+        if row_range not in self._range_signatures:
+            self._range_signatures.add(row_range)
             self._row_set.row_ranges.append(row_range._pb)
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index e88ca4a9bb71..0758f17a55f2 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+
 import pytest
 
 TEST_ROWS = [
@@ -586,3 +588,24 @@ def test_empty_row_set(self):
         """Empty strings should be treated as keys inputs"""
         query = self._make_one(row_keys="")
         assert query.row_keys == [b""]
+
+    def test_add_range_performance_regression(self):
+        """Assert O(1) range insertion scaling and prevent regression back to O(N^2)."""
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        query = self._make_one()
+        ranges_count = 1000
+
+        start_time = time.perf_counter()
+        for i in range(ranges_count):
+            # Using 4-digit zero-padding ensures 'key_0009' is lexicographically smaller than 'key_0010'
+            r = RowRange(start_key=f"key_{i:04d}", end_key=f"key_{i + 1:04d}")
+            query.add_range(r)
+        end_time = time.perf_counter()
+
+        elapsed = end_time - start_time
+        # Under O(N^2) on test runners, 1k iterations takes seconds.
+        # Under O(1) with tuple hashing, it takes <0.02s.
+        assert elapsed < 0.5, (
+            f"O(N^2) Performance regression detected! Inserting {ranges_count} ranges took {elapsed:.2f}s"
+        )

From 679ebedf210f283bdd754aedbc3e7455d177ea54 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Thu, 15 Oct 2026 20:09:00 +0000
Subject: [PATCH 02/34] address offline feedback

---
 .../google/cloud/bigtable/data/read_rows_query.py             | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 1038ce0bafe8..84de74e31923 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -151,14 +151,14 @@ def _to_pb(self) -> RowRangePB:
     @classmethod
     def _from_pb(cls, data: RowRangePB) -> RowRange:
         """
-        Creates a RowRange from a protobuf
+        Creates a RowRange from a protobuf without allocating temporary unused objects.
 
         Args:
             data (RowRangePB): The protobuf to convert
         Returns:
             RowRange: The converted RowRange
         """
-        instance = cls()
+        instance = cls.__new__(cls)
         instance._pb = data
         return instance
 

From 285b2c1f067c8c9d1a12058a93a977cd6fff99eb Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Thu, 15 Oct 2026 20:21:15 +0000
Subject: [PATCH 03/34] update comment

---
 .../tests/unit/data/test_read_rows_query.py                  | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 0758f17a55f2..a9b0dd9850bb 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -605,7 +605,8 @@ def test_add_range_performance_regression(self):
 
         elapsed = end_time - start_time
         # Under O(N^2) on test runners, 1k iterations takes seconds.
-        # Under O(1) with tuple hashing, it takes <0.02s.
-        assert elapsed < 0.5, (
+        # Under O(1) with tuple hashing, it takes <0.10s.
+        # Double the value to avoid flaky tests
+        assert elapsed < 0.20, (
             f"O(N^2) Performance regression detected! Inserting {ranges_count} ranges took {elapsed:.2f}s"
         )

From 01e759ad9fdcb2626b699bf41b5903b7ae3ed67d Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Thu, 15 Oct 2026 21:06:05 +0000
Subject: [PATCH 04/34] fix build; add test

---
 .../cloud/bigtable/data/read_rows_query.py    |  2 +-
 .../tests/unit/data/test_read_rows_query.py   | 23 +++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 84de74e31923..a15b51db7cb0 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -197,7 +197,7 @@ def __bool__(self) -> bool:
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, RowRange):
             return NotImplemented
-        return self._key() == other._key()
+        return self._pb == other._pb
 
     def __str__(self) -> str:
         """
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index a9b0dd9850bb..84d2aa577df8 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -610,3 +610,26 @@ def test_add_range_performance_regression(self):
         assert elapsed < 0.20, (
             f"O(N^2) Performance regression detected! Inserting {ranges_count} ranges took {elapsed:.2f}s"
         )
+
+    def test_add_range_deduplication_behavior(self):
+        """Assert that identical ranges are deduplicated, but distinct ranges are kept."""
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        query = self._make_one()
+
+        # Create two identical ranges
+        r1 = RowRange(start_key="a", end_key="b")
+        r2 = RowRange(start_key="a", end_key="b")
+
+        # Create a distinct range
+        r3 = RowRange(start_key="c", end_key="d")
+
+        # Add all three ranges
+        query.add_range(r1)
+        query.add_range(r2)  # Duplicate, should be ignored
+        query.add_range(r3)  # Distinct, should be kept
+
+        # Assert that only r1 and r3 are in the final query
+        assert len(query.row_ranges) == 2
+        assert query.row_ranges[0] == r1
+        assert query.row_ranges[1] == r3

From d08cc34367f1031492b1c1afd1fd50c585308d03 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Thu, 15 Oct 2026 21:18:34 +0000
Subject: [PATCH 05/34] address review feedback

---
 .../tests/unit/data/test_read_rows_query.py   | 47 ++++++++++++-------
 1 file changed, 29 insertions(+), 18 deletions(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 84d2aa577df8..66eaa26663ad 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -589,27 +589,38 @@ def test_empty_row_set(self):
         query = self._make_one(row_keys="")
         assert query.row_keys == [b""]
 
-    def test_add_range_performance_regression(self):
-        """Assert O(1) range insertion scaling and prevent regression back to O(N^2)."""
+    def test_add_range_does_not_materialize_row_ranges(self):
+        """
+        Ensures add_range scales in O(1) time and does not trigger O(N) scans.
+
+        The legacy implementation called the `row_ranges` property helper during
+        membership checks, which dynamically re-allocated and materialized brand
+        new Python RowRange wrapper objects for every element in the array on
+        each insertion. This led to an accidental O(N^2) memory and CPU leak.
+
+        This test mocks out the `row_ranges` property on the class and asserts
+        that it is never accessed during batch additions, proving we bypass the
+        quadratic evaluation path entirely.
+        """
+        from unittest import mock
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
         query = self._make_one()
-        ranges_count = 1000
-
-        start_time = time.perf_counter()
-        for i in range(ranges_count):
-            # Using 4-digit zero-padding ensures 'key_0009' is lexicographically smaller than 'key_0010'
-            r = RowRange(start_key=f"key_{i:04d}", end_key=f"key_{i + 1:04d}")
-            query.add_range(r)
-        end_time = time.perf_counter()
-
-        elapsed = end_time - start_time
-        # Under O(N^2) on test runners, 1k iterations takes seconds.
-        # Under O(1) with tuple hashing, it takes <0.10s.
-        # Double the value to avoid flaky tests
-        assert elapsed < 0.20, (
-            f"O(N^2) Performance regression detected! Inserting {ranges_count} ranges took {elapsed:.2f}s"
-        )
+
+        # Intercept the `row_ranges` class property with a mock to track accesses
+        with mock.patch.object(
+            type(query), "row_ranges", new_callable=mock.PropertyMock
+        ) as row_ranges:
+            for i in range(100):
+                # Add unique sequential ranges. Utilizing our fast O(1) set signature lookup
+                # should never touch the slow wrapper-materialization properties.
+                query.add_range(RowRange(f"k{i:03d}", f"k{i + 1:03d}"))
+
+            # ASSERT: The slow, O(N) wrapper-building property was never accessed
+            row_ranges.assert_not_called()
+
+        # Verify that all ranges were successfully committed to the underlying protobuf
+        assert len(query._row_set.row_ranges) == 100
 
     def test_add_range_deduplication_behavior(self):
         """Assert that identical ranges are deduplicated, but distinct ranges are kept."""

From b864707aa5f0885f0b0b59cb4f5dc4642bc9a164 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Thu, 15 Oct 2026 21:20:11 +0000
Subject: [PATCH 06/34] move import

---
 .../tests/unit/data/test_read_rows_query.py                    | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 66eaa26663ad..703f9118646d 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import time
+from unittest import mock
 
 import pytest
 
@@ -602,7 +602,6 @@ def test_add_range_does_not_materialize_row_ranges(self):
         that it is never accessed during batch additions, proving we bypass the
         quadratic evaluation path entirely.
         """
-        from unittest import mock
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
         query = self._make_one()

From ef09fca9cffe8ce31c343810cf50ee977379ebb7 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Sat, 17 Oct 2026 20:16:35 +0000
Subject: [PATCH 07/34] fix(bigtable): align RowRange.__hash__ with __eq__
 across base and legacy classes

---
 .../cloud/bigtable/data/read_rows_query.py    | 13 +++++-----
 .../google/cloud/bigtable/row_set.py          |  3 ---
 .../tests/unit/data/test_read_rows_query.py   | 24 +++++++++++++++++++
 3 files changed, 31 insertions(+), 9 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index a15b51db7cb0..4fcc14995737 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -87,16 +87,17 @@ def __init__(
                 init_dict["end_key_open"] = end_key
         self._pb = RowRangePB(**init_dict)
 
-    def _key(self) -> tuple[bytes | None, bool, bytes | None, bool]:
+    def _signature(self) -> tuple[bytes, bytes, bytes, bytes]:
+        pb = self._pb
         return (
-            self.start_key,
-            self.start_is_inclusive,
-            self.end_key,
-            self.end_is_inclusive,
+            pb.start_key_closed,
+            pb.start_key_open,
+            pb.end_key_closed,
+            pb.end_key_open,
         )
 
     def __hash__(self) -> int:
-        return hash(self._key())
+        return hash(self._signature())
 
     @property
     def start_key(self) -> bytes | None:
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
index 4e4bacfa7cda..0732b3e2d312 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
@@ -165,9 +165,6 @@ def _key(self):
             self.end_is_inclusive,
         )
 
-    def __hash__(self):
-        return hash(self._key())
-
     def get_range_kwargs(self):
         """Convert row range object to dict which can be passed to
         google.bigtable.v2.RowRange add method.
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 703f9118646d..2a8f51dc49ce 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -643,3 +643,27 @@ def test_add_range_deduplication_behavior(self):
         assert len(query.row_ranges) == 2
         assert query.row_ranges[0] == r1
         assert query.row_ranges[1] == r3
+
+    def test_row_range_hash_matches_legacy(self):
+        """
+        Verify that base RowRange and legacy row_set.RowRange instances representing
+        the same range compare equal, produce identical hashes, and deduplicate
+        properly when combined in a ReadRowsQuery.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+        from google.cloud.bigtable.row_set import RowRange as LegacyRowRange
+
+        # Instantiate equivalent ranges using both the data API and legacy classes.
+        base_range = RowRange(b"a", b"b")
+        legacy_range = LegacyRowRange(b"a", b"b")
+
+        # Equal RowRange objects must produce identical hashes so set membership
+        # operates consistently across base and legacy types.
+        assert base_range == legacy_range
+        assert hash(base_range) == hash(legacy_range)
+
+        # Adding an equivalent legacy range to a query initialized with a base range
+        # should be recognized as a duplicate and ignored.
+        query = self._make_one(row_ranges=[base_range])
+        query.add_range(legacy_range)
+        assert len(query.row_ranges) == 1

From 183df93244810a1412001255c41e52174965a1e3 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Sat, 17 Oct 2026 20:22:27 +0000
Subject: [PATCH 08/34] address review feedback

---
 .../google/cloud/bigtable/data/read_rows_query.py              | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 4fcc14995737..36b4a8e6091b 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -152,13 +152,14 @@ def _to_pb(self) -> RowRangePB:
     @classmethod
     def _from_pb(cls, data: RowRangePB) -> RowRange:
         """
-        Creates a RowRange from a protobuf without allocating temporary unused objects.
+        Creates a RowRange from a protobuf
 
         Args:
             data (RowRangePB): The protobuf to convert
         Returns:
             RowRange: The converted RowRange
         """
+        # Skip __init__ to avoid allocating an unused RowRangePB
         instance = cls.__new__(cls)
         instance._pb = data
         return instance

From 39ddcf045961968bd6da474a57192b66b985fa96 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Sat, 17 Oct 2026 20:23:40 +0000
Subject: [PATCH 09/34] address offline feedback

---
 .../google/cloud/bigtable/data/read_rows_query.py              | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 36b4a8e6091b..f9f82b38d733 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -177,7 +177,8 @@ def _from_dict(cls, data: dict[str, bytes | str]) -> RowRange:
         formatted_data = {
             k: v.encode() if isinstance(v, str) else v for k, v in data.items()
         }
-        instance = cls()
+        # Skip __init__ to avoid allocating an unused RowRangePB
+        instance = cls.__new__(cls)
         instance._pb = RowRangePB(**formatted_data)
         return instance
 

From f9b35b11f7b150345714a50b7846a8000ffc60eb Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 18:18:30 +0000
Subject: [PATCH 10/34] fix(bigtable): store immutable tuple signatures in
 _range_signatures and deduplicate in __init__

---
 .../cloud/bigtable/data/read_rows_query.py    | 18 +++++++---------
 .../tests/unit/data/test_read_rows_query.py   | 21 +++++++++++++++++++
 2 files changed, 28 insertions(+), 11 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index f9f82b38d733..5c4c233e874d 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -261,14 +261,10 @@ def __init__(
         if not isinstance(row_keys, list):
             row_keys = [row_keys]
         row_keys = [key.encode() if isinstance(key, str) else key for key in row_keys]
-        self._row_set = RowSetPB(
-            row_keys=row_keys, row_ranges=[r._pb for r in row_ranges]
-        )
-        # Pre-initialize our signature set using hashable RowRange wrappers
-        self._range_signatures = set(
-            RowRange._from_pb(r) for r in self._row_set.row_ranges
-        )
-
+        self._row_set = RowSetPB(row_keys=row_keys)
+        self._range_signatures: set[tuple[bytes, bytes, bytes, bytes]] = set()
+        for r in row_ranges:
+            self.add_range(r)
         self.limit = limit or None
         self.filter = row_filter
 
@@ -368,9 +364,9 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
-        # Utilize pure O(1) set-membership checking
-        if row_range not in self._range_signatures:
-            self._range_signatures.add(row_range)
+        sig = row_range._signature()
+        if sig not in self._range_signatures:
+            self._range_signatures.add(sig)
             self._row_set.row_ranges.append(row_range._pb)
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 2a8f51dc49ce..bf37ee929fc7 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -667,3 +667,24 @@ def test_row_range_hash_matches_legacy(self):
         query = self._make_one(row_ranges=[base_range])
         query.add_range(legacy_range)
         assert len(query.row_ranges) == 1
+
+    def test_add_range_immune_to_caller_mutating_row_range_after_insert(self):
+        """
+        Verify that mutating a RowRange's underlying protobuf after adding it to
+        a query does not corrupt the query's internal deduplication set.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        query = self._make_one()
+        r1 = RowRange(start_key=b"a", end_key=b"b")
+        query.add_range(r1)
+
+        # Mutate the caller's RowRange wrapper reference after insertion.
+        r1._pb = RowRange(start_key=b"x", end_key=b"y")._pb
+
+        # Adding a new RowRange with the original ("a", "b") bounds must still be
+        # detected as a duplicate of the previously added range.
+        query.add_range(RowRange(start_key=b"a", end_key=b"b"))
+
+        assert len(query.row_ranges) == 1
+        assert query.row_ranges[0] == RowRange(start_key=b"a", end_key=b"b")

From 213fa5b3c6bf00516a4521073969e179435f8a4b Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 18:37:07 +0000
Subject: [PATCH 11/34] perf(bigtable): resolve O(n^2) in ReadRowsQuery.add_key
 and __eq__, enforce slots

---
 .../cloud/bigtable/data/read_rows_query.py    |  45 ++++--
 .../tests/unit/data/test_read_rows_query.py   | 133 ++++++++++++++++++
 2 files changed, 164 insertions(+), 14 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 5c4c233e874d..6cf5c6ce8148 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -14,6 +14,7 @@
 #
 from __future__ import annotations
 
+import copy
 from bisect import bisect_left, bisect_right
 from collections import defaultdict
 from typing import TYPE_CHECKING, Any
@@ -243,7 +244,13 @@ class ReadRowsQuery:
         row_filter: a RowFilter to apply to the query
     """
 
-    slots = ("_limit", "_filter", "_row_set")
+    __slots__ = (
+        "_limit",
+        "_filter",
+        "_row_set",
+        "_key_set",
+        "_range_signatures",
+    )
 
     def __init__(
         self,
@@ -260,9 +267,11 @@ def __init__(
             row_ranges = [row_ranges]
         if not isinstance(row_keys, list):
             row_keys = [row_keys]
-        row_keys = [key.encode() if isinstance(key, str) else key for key in row_keys]
-        self._row_set = RowSetPB(row_keys=row_keys)
+        self._row_set = RowSetPB()
+        self._key_set: set[bytes] = set()
         self._range_signatures: set[tuple[bytes, bytes, bytes, bytes]] = set()
+        for k in row_keys:
+            self.add_key(k)
         for r in row_ranges:
             self.add_range(r)
         self.limit = limit or None
@@ -351,7 +360,8 @@ def add_key(self, row_key: str | bytes):
             row_key = row_key.encode()
         elif not isinstance(row_key, bytes):
             raise ValueError("row_key must be string or bytes")
-        if row_key not in self._row_set.row_keys:
+        if row_key not in self._key_set:
+            self._key_set.add(row_key)
             self._row_set.row_keys.append(row_key)
 
     def add_range(
@@ -512,6 +522,13 @@ def _to_pb(self, table) -> ReadRowsRequestPB:
             **table._request_path,
         )
 
+    def __copy__(self) -> ReadRowsQuery:
+        new_query = ReadRowsQuery(limit=self.limit, row_filter=self.filter)
+        new_query._row_set = copy.deepcopy(self._row_set)
+        new_query._key_set = self._key_set.copy()
+        new_query._range_signatures = self._range_signatures.copy()
+        return new_query
+
     def __eq__(self, other):
         """
         RowRanges are equal if they have the same row keys, row ranges,
@@ -526,24 +543,24 @@ def __eq__(self, other):
         if not isinstance(other, ReadRowsQuery):
             return False
         # empty queries are equal
-        if len(self.row_keys) == 0 and len(other.row_keys) == 0:
-            this_range_empty = len(self.row_ranges) == 0 or all(
-                [bool(r) is False for r in self.row_ranges]
+        if len(self._key_set) == 0 and len(other._key_set) == 0:
+            empty_sig = (b"", b"", b"", b"")
+            this_range_empty = len(self._range_signatures) == 0 or (
+                self._range_signatures == {empty_sig}
             )
-            other_range_empty = len(other.row_ranges) == 0 or all(
-                [bool(r) is False for r in other.row_ranges]
+            other_range_empty = len(other._range_signatures) == 0 or (
+                other._range_signatures == {empty_sig}
             )
             if this_range_empty and other_range_empty:
                 return self.filter == other.filter and self.limit == other.limit
         # otherwise, sets should have same sizes
-        if len(self.row_keys) != len(other.row_keys):
+        if len(self._key_set) != len(other._key_set):
             return False
-        if len(self.row_ranges) != len(other.row_ranges):
+        if len(self._range_signatures) != len(other._range_signatures):
             return False
-        ranges_match = all([row in other.row_ranges for row in self.row_ranges])
         return (
-            self.row_keys == other.row_keys
-            and ranges_match
+            self._key_set == other._key_set
+            and self._range_signatures == other._range_signatures
             and self.filter == other.filter
             and self.limit == other.limit
         )
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index bf37ee929fc7..b2de08dced30 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -621,6 +621,57 @@ def test_add_range_does_not_materialize_row_ranges(self):
         # Verify that all ranges were successfully committed to the underlying protobuf
         assert len(query._row_set.row_ranges) == 100
 
+    def test_add_key_deduplicates_via_set_without_scanning_row_set(self):
+        """
+        Verify that ReadRowsQuery deduplicates row_keys in __init__ and add_key
+        using its internal O(1) key set without linearly scanning _row_set.row_keys.
+        """
+        # Duplicate keys passed to __init__ (even in mixed str/bytes forms)
+        # must be deduplicated on construction.
+        query = self._make_one(row_keys=["k0", b"k0"])
+        assert query.row_keys == [b"k0"]
+
+        # When adding keys that are already present in the query, add_key must
+        # short-circuit via _key_set without accessing _row_set.row_keys.
+        with mock.patch.object(query, "_row_set", wraps=query._row_set) as mock_row_set:
+            query.add_key(b"k0")
+            query.add_key("k0")
+            assert not mock_row_set.row_keys.called
+
+        assert query.row_keys == [b"k0"]
+
+    def test_query_equality_does_not_materialize_row_keys_or_ranges(self):
+        """
+        Verify that ReadRowsQuery.__eq__ compares keys and ranges in order-independent
+        O(N) time using internal sets without materializing row_keys or row_ranges lists.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        # Create two equivalent queries with identical keys and ranges inserted
+        # in opposite orders.
+        q1 = self._make_one(
+            row_keys=[f"k{i}" for i in range(50)],
+            row_ranges=[RowRange(f"a{i}", f"b{i}") for i in range(50)],
+        )
+        q2 = self._make_one(
+            row_keys=[f"k{i}" for i in reversed(range(50))],
+            row_ranges=[RowRange(f"a{i}", f"b{i}") for i in reversed(range(50))],
+        )
+
+        # Intercept the row_keys and row_ranges properties to verify that equality
+        # comparisons operate on internal sets without constructing wrapper lists.
+        with (
+            mock.patch.object(
+                type(q1), "row_keys", new_callable=mock.PropertyMock
+            ) as mock_keys,
+            mock.patch.object(
+                type(q1), "row_ranges", new_callable=mock.PropertyMock
+            ) as mock_ranges,
+        ):
+            assert q1 == q2
+            mock_keys.assert_not_called()
+            mock_ranges.assert_not_called()
+
     def test_add_range_deduplication_behavior(self):
         """Assert that identical ranges are deduplicated, but distinct ranges are kept."""
         from google.cloud.bigtable.data.read_rows_query import RowRange
@@ -688,3 +739,85 @@ def test_add_range_immune_to_caller_mutating_row_range_after_insert(self):
 
         assert len(query.row_ranges) == 1
         assert query.row_ranges[0] == RowRange(start_key=b"a", end_key=b"b")
+
+    def test_row_range_from_dict_does_not_call_init(self):
+        """
+        Verify that RowRange._from_dict constructs the wrapper instance directly
+        without invoking __init__ and allocating an unused intermediate protobuf.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        # Spy on RowRange.__init__ to ensure dictionary deserialization bypasses
+        # default constructor initialization and dummy protobuf creation.
+        with mock.patch.object(
+            RowRange, "__init__", wraps=RowRange.__init__
+        ) as mock_init:
+            row_range = RowRange._from_dict(
+                {"start_key_closed": b"a", "end_key_open": b"b"}
+            )
+            mock_init.assert_not_called()
+
+        # Verify the resulting RowRange exposes the expected boundaries and inclusivity.
+        assert row_range.start_key == b"a"
+        assert row_range.start_is_inclusive is True
+        assert row_range.end_key == b"b"
+        assert row_range.end_is_inclusive is False
+
+    def test_read_rows_query_enforces_slots(self):
+        """
+        Verify that ReadRowsQuery enforces __slots__ so instances do not allocate
+        a dynamic __dict__ or permit undeclared attributes.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        # Initialize a query with both keys and ranges so all internal attributes
+        # are populated on the instance.
+        query = self._make_one(
+            row_keys=[b"k1"],
+            row_ranges=[RowRange(start_key=b"a", end_key=b"b")],
+        )
+
+        # Enforcing __slots__ prevents creation of a per-instance __dict__.
+        assert not hasattr(query, "__dict__")
+
+        # Assigning an attribute not declared in __slots__ must raise AttributeError.
+        with pytest.raises(AttributeError):
+            query._undeclared_attribute = "invalid"
+
+    def test_copy_query_preserves_deduplication_isolation(self):
+        """
+        Verify that shallow-copying a ReadRowsQuery produces independent
+        deduplication tracking sets so mutations on the copy do not affect the
+        original query, and existing items remain deduplicated on the copy.
+        """
+        import copy
+
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        r1 = RowRange(start_key="a", end_key="b")
+        r2 = RowRange(start_key="c", end_key="d")
+
+        original = self._make_one(row_keys=[b"k1"], row_ranges=[r1])
+        copied = copy.copy(original)
+
+        # Re-adding existing items to the copy must be ignored as duplicates.
+        copied.add_key(b"k1")
+        copied.add_range(r1)
+        assert copied.row_keys == [b"k1"]
+        assert copied.row_ranges == [r1]
+
+        # Mutate only the copied query by adding a new key and range.
+        copied.add_key(b"k2")
+        copied.add_range(r2)
+
+        # Confirm the original query was not modified by operations on the copy.
+        assert original.row_keys == [b"k1"]
+        assert original.row_ranges == [r1]
+
+        # Adding the same new key and range to the original query must succeed
+        # because its internal deduplication sets are independent from the copy.
+        original.add_key(b"k2")
+        original.add_range(r2)
+
+        assert original.row_keys == [b"k1", b"k2"]
+        assert original.row_ranges == [r1, r2]

From bb519805efde6deb4ff2401894768a128e3834e6 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 18:56:20 +0000
Subject: [PATCH 12/34] fix build

---
 .../google/cloud/bigtable/data/read_rows_query.py             | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 6cf5c6ce8148..4ab06b5714f3 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -366,7 +366,7 @@ def add_key(self, row_key: str | bytes):
 
     def add_range(
         self,
-        row_range: RowRange,
+        row_range: RowRange | RowRangePB,
     ):
         """
         Add a range of row keys to this query.
@@ -374,6 +374,8 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
+        if isinstance(row_range, RowRangePB):
+            row_range = RowRange._from_pb(row_range)
         sig = row_range._signature()
         if sig not in self._range_signatures:
             self._range_signatures.add(sig)

From d2b8192c1e445e32d5408ad7a996679f5a6ba8e8 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 19:18:22 +0000
Subject: [PATCH 13/34] mypy

---
 .../google/cloud/bigtable/data/_async/_read_rows.py         | 6 +++++-
 .../google/cloud/bigtable/data/_sync_autogen/_read_rows.py  | 6 +++++-
 2 files changed, 10 insertions(+), 2 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py
index 3e6c2b533b1e..4180d4a960d5 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py
@@ -88,7 +88,7 @@ class _ReadRowsOperationAsync:
 
     def __init__(
         self,
-        query: ReadRowsQuery,
+        query: ReadRowsQuery | dict[str, object],
         target: TargetType,
         operation_timeout: float,
         attempt_timeout: float,
@@ -99,6 +99,10 @@ def __init__(
             attempt_timeout, operation_timeout
         )
         self.operation_timeout = operation_timeout
+        # ReadRowsQuery defines __slots__, which gives it a C-level memory layout
+        # disjoint from dict. Consequently, `query` must be explicitly typed as
+        # `ReadRowsQuery | dict[str, object]` in the signature above; otherwise,
+        # mypy flags this runtime `isinstance(query, dict)` branch as unreachable.
         if isinstance(query, dict):
             self.request = ReadRowsRequestPB(
                 **query,
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py
index ce29226cac53..ddb73ad175bb 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py
@@ -82,7 +82,7 @@ class _ReadRowsOperation:
 
     def __init__(
         self,
-        query: ReadRowsQuery,
+        query: ReadRowsQuery | dict[str, object],
         target: TargetType,
         operation_timeout: float,
         attempt_timeout: float,
@@ -93,6 +93,10 @@ def __init__(
             attempt_timeout, operation_timeout
         )
         self.operation_timeout = operation_timeout
+        # ReadRowsQuery defines __slots__, which gives it a C-level memory layout
+        # disjoint from dict. Consequently, `query` must be explicitly typed as
+        # `ReadRowsQuery | dict[str, object]` in the signature above; otherwise,
+        # mypy flags this runtime `isinstance(query, dict)` branch as unreachable.
         if isinstance(query, dict):
             self.request = ReadRowsRequestPB(
                 **query, **target._request_path, app_profile_id=target.app_profile_id

From bff7b0b9becebc58e385f4750c318032841c4e46 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 19:45:52 +0000
Subject: [PATCH 14/34] address review feedback

---
 .../tests/unit/data/test_read_rows_query.py              | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index b2de08dced30..760ed3d53489 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -676,8 +676,6 @@ def test_add_range_deduplication_behavior(self):
         """Assert that identical ranges are deduplicated, but distinct ranges are kept."""
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
-        query = self._make_one()
-
         # Create two identical ranges
         r1 = RowRange(start_key="a", end_key="b")
         r2 = RowRange(start_key="a", end_key="b")
@@ -685,10 +683,11 @@ def test_add_range_deduplication_behavior(self):
         # Create a distinct range
         r3 = RowRange(start_key="c", end_key="d")
 
-        # Add all three ranges
-        query.add_range(r1)
-        query.add_range(r2)  # Duplicate, should be ignored
+        # Seed r1 via __init__, then add r2 (duplicate) and r3 (distinct)
+        query = self._make_one(row_ranges=[r1])
+        query.add_range(r2)  # Duplicate of constructor-seeded r1, should be ignored
         query.add_range(r3)  # Distinct, should be kept
+        query.add_range(RowRange(start_key="c", end_key="d"))  # Duplicate of r3, should be ignored
 
         # Assert that only r1 and r3 are in the final query
         assert len(query.row_ranges) == 2

From 09110b34f90aed07b8d60f060604041350daa610 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 19:57:42 +0000
Subject: [PATCH 15/34] lint

---
 .../tests/unit/data/test_read_rows_query.py                   | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 760ed3d53489..4af7abaea6bc 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -687,7 +687,9 @@ def test_add_range_deduplication_behavior(self):
         query = self._make_one(row_ranges=[r1])
         query.add_range(r2)  # Duplicate of constructor-seeded r1, should be ignored
         query.add_range(r3)  # Distinct, should be kept
-        query.add_range(RowRange(start_key="c", end_key="d"))  # Duplicate of r3, should be ignored
+        query.add_range(
+            RowRange(start_key="c", end_key="d")
+        )  # Duplicate of r3, should be ignored
 
         # Assert that only r1 and r3 are in the final query
         assert len(query.row_ranges) == 2

From 73f1e1cd1aebfc8d67a5b1625e4e6bbfef972e1a Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 20:18:24 +0000
Subject: [PATCH 16/34] fix(bigtable): distinguish unset and empty-byte bounds
 in RowRange._signature

---
 .../cloud/bigtable/data/read_rows_query.py    | 63 ++++++++++++++++---
 .../tests/unit/data/test_read_rows_query.py   | 34 ++++++++++
 2 files changed, 87 insertions(+), 10 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 4ab06b5714f3..5cba50438b09 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -88,14 +88,55 @@ def __init__(
                 init_dict["end_key_open"] = end_key
         self._pb = RowRangePB(**init_dict)
 
-    def _signature(self) -> tuple[bytes, bytes, bytes, bytes]:
-        pb = self._pb
-        return (
-            pb.start_key_closed,
-            pb.start_key_open,
-            pb.end_key_closed,
-            pb.end_key_open,
-        )
+    @staticmethod
+    def _signature_from_pb(
+        pb: RowRangePB,
+    ) -> tuple[bytes | None, bool, bytes | None, bool]:
+        raw_pb = pb._pb
+
+        # Fast path: non-empty keys bypass HasField and proto-plus overhead
+        start_closed = raw_pb.start_key_closed
+        if start_closed:
+            start_key: bytes | None = start_closed
+            start_is_inclusive = True
+        else:
+            start_open = raw_pb.start_key_open
+            if start_open:
+                start_key = start_open
+                start_is_inclusive = False
+            elif raw_pb.HasField("start_key_closed"):
+                start_key = b""
+                start_is_inclusive = True
+            elif raw_pb.HasField("start_key_open"):
+                start_key = b""
+                start_is_inclusive = False
+            else:
+                start_key = None
+                start_is_inclusive = True
+
+        end_open = raw_pb.end_key_open
+        if end_open:
+            end_key: bytes | None = end_open
+            end_is_inclusive = False
+        else:
+            end_closed = raw_pb.end_key_closed
+            if end_closed:
+                end_key = end_closed
+                end_is_inclusive = True
+            elif raw_pb.HasField("end_key_open"):
+                end_key = b""
+                end_is_inclusive = False
+            elif raw_pb.HasField("end_key_closed"):
+                end_key = b""
+                end_is_inclusive = True
+            else:
+                end_key = None
+                end_is_inclusive = False
+
+        return (start_key, start_is_inclusive, end_key, end_is_inclusive)
+
+    def _signature(self) -> tuple[bytes | None, bool, bytes | None, bool]:
+        return self._signature_from_pb(self._pb)
 
     def __hash__(self) -> int:
         return hash(self._signature())
@@ -269,7 +310,9 @@ def __init__(
             row_keys = [row_keys]
         self._row_set = RowSetPB()
         self._key_set: set[bytes] = set()
-        self._range_signatures: set[tuple[bytes, bytes, bytes, bytes]] = set()
+        self._range_signatures: set[tuple[bytes | None, bool, bytes | None, bool]] = (
+            set()
+        )
         for k in row_keys:
             self.add_key(k)
         for r in row_ranges:
@@ -546,7 +589,7 @@ def __eq__(self, other):
             return False
         # empty queries are equal
         if len(self._key_set) == 0 and len(other._key_set) == 0:
-            empty_sig = (b"", b"", b"", b"")
+            empty_sig = (None, True, None, False)
             this_range_empty = len(self._range_signatures) == 0 or (
                 self._range_signatures == {empty_sig}
             )
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 4af7abaea6bc..af5d1ac91bf8 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -822,3 +822,37 @@ def test_copy_query_preserves_deduplication_isolation(self):
 
         assert original.row_keys == [b"k1", b"k2"]
         assert original.row_ranges == [r1, r2]
+
+    def test_row_range_signature_distinguishes_empty_bytes_and_inclusivity(self):
+        """
+        Verify that unset bounds (None), inclusive empty-byte bounds (b""),
+        and exclusive empty-byte bounds (b"") produce distinct signatures
+        and are not conflated during deduplication or query equality.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+
+        unbounded_start = RowRange(start_key=None, end_key=b"z")
+        closed_empty_start = RowRange._from_dict(
+            {"start_key_closed": b"", "end_key_open": b"z"}
+        )
+        open_empty_start = RowRange._from_dict(
+            {"start_key_open": b"", "end_key_open": b"z"}
+        )
+        # All three ranges are semantically distinct and must not collide
+        assert unbounded_start != closed_empty_start
+        assert closed_empty_start != open_empty_start
+        assert unbounded_start._signature() != closed_empty_start._signature()
+        assert closed_empty_start._signature() != open_empty_start._signature()
+
+        query = self._make_one(
+            row_ranges=[unbounded_start, closed_empty_start, open_empty_start]
+        )
+        assert len(query.row_ranges) == 3
+
+        # A range targeting the single row b"" must NOT compare equal to an empty query
+        single_empty_key_range = RowRange._from_dict(
+            {"start_key_closed": b"", "end_key_closed": b""}
+        )
+        non_empty_query = self._make_one(row_ranges=[single_empty_key_range])
+        empty_query = self._make_one()
+        assert non_empty_query != empty_query

From 6d94d4ffb5a9fe52cc70efdbdde309fae2072070 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 20:22:38 +0000
Subject: [PATCH 17/34] test(bigtable): tighten mock and mutation assertions in
 ReadRowsQuery tests

---
 .../tests/unit/data/test_read_rows_query.py          | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index af5d1ac91bf8..7d1fda4b874f 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -633,10 +633,13 @@ def test_add_key_deduplicates_via_set_without_scanning_row_set(self):
 
         # When adding keys that are already present in the query, add_key must
         # short-circuit via _key_set without accessing _row_set.row_keys.
-        with mock.patch.object(query, "_row_set", wraps=query._row_set) as mock_row_set:
+        mock_row_set = mock.MagicMock()
+        row_keys_prop = mock.PropertyMock()
+        type(mock_row_set).row_keys = row_keys_prop
+        with mock.patch.object(query, "_row_set", mock_row_set):
             query.add_key(b"k0")
             query.add_key("k0")
-            assert not mock_row_set.row_keys.called
+            row_keys_prop.assert_not_called()
 
         assert query.row_keys == [b"k0"]
 
@@ -731,8 +734,9 @@ def test_add_range_immune_to_caller_mutating_row_range_after_insert(self):
         r1 = RowRange(start_key=b"a", end_key=b"b")
         query.add_range(r1)
 
-        # Mutate the caller's RowRange wrapper reference after insertion.
-        r1._pb = RowRange(start_key=b"x", end_key=b"y")._pb
+        # Mutate the caller's underlying RowRangePB in place after insertion.
+        r1._pb.start_key_closed = b"x"
+        r1._pb.end_key_open = b"y"
 
         # Adding a new RowRange with the original ("a", "b") bounds must still be
         # detected as a duplicate of the previously added range.

From 8e32f9c6a847cbf58d82e813a9cd666ac997d8bd Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 20:41:30 +0000
Subject: [PATCH 18/34] perf(bigtable): enforce __slots__ on legacy
 row_set.RowRange

---
 .../google/cloud/bigtable/helpers.py                |  2 ++
 .../google/cloud/bigtable/row_set.py                |  2 ++
 .../tests/unit/data/test_read_rows_query.py         | 13 +++++++++++++
 3 files changed, 17 insertions(+)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py
index 71af2f539f2c..933dd4614c4e 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py
@@ -43,6 +43,8 @@ class and define a map _attribute_map from legacy client attributes to data clie
     Attributes are remapped and redefined in __init__ as well as getattr/setattr.
     """
 
+    __slots__ = ()
+
     def __init__(self, *args, **kwargs):
         new_kwargs = {self._attribute_map.get(k, k): v for (k, v) in kwargs.items()}
         super(_MappableAttributesMixin, self).__init__(*args, **new_kwargs)
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
index 0732b3e2d312..d9a6bb08ab86 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
@@ -152,6 +152,8 @@ class RowRange(_MappableAttributesMixin, BaseRowRange):
                   considered inclusive. The default is False (exclusive).
     """
 
+    __slots__ = ()
+
     _attribute_map = {
         "start_inclusive": "start_is_inclusive",
         "end_inclusive": "end_is_inclusive",
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 7d1fda4b874f..28d7a8117567 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -789,6 +789,19 @@ def test_read_rows_query_enforces_slots(self):
         with pytest.raises(AttributeError):
             query._undeclared_attribute = "invalid"
 
+    def test_legacy_row_range_enforces_slots(self):
+        """
+        Verify that legacy row_set.RowRange also defines __slots__ = () so it
+        inherits BaseRowRange's slotted layout without allocating a __dict__.
+        """
+        from google.cloud.bigtable.row_set import RowRange as LegacyRowRange
+
+        legacy_range = LegacyRowRange(start_key=b"a", end_key=b"b")
+        assert not hasattr(legacy_range, "__dict__")
+
+        with pytest.raises(AttributeError):
+            legacy_range._undeclared_attribute = "invalid"
+
     def test_copy_query_preserves_deduplication_isolation(self):
         """
         Verify that shallow-copying a ReadRowsQuery produces independent

From 3491e36c6143e510599f06e9d12fe6913f95aa9c Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 20:46:05 +0000
Subject: [PATCH 19/34] perf(bigtable): avoid RowRange wrapper allocation in
 add_range(RowRangePB)

---
 .../cloud/bigtable/data/read_rows_query.py    |  9 +++++---
 .../tests/unit/data/test_read_rows_query.py   | 22 +++++++++++++++++++
 2 files changed, 28 insertions(+), 3 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 5cba50438b09..ad6836bb8f38 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -418,11 +418,14 @@ def add_range(
             row_range: a range of row keys to add to this query
         """
         if isinstance(row_range, RowRangePB):
-            row_range = RowRange._from_pb(row_range)
-        sig = row_range._signature()
+            pb = row_range
+            sig = RowRange._signature_from_pb(pb)
+        else:
+            pb = row_range._pb
+            sig = row_range._signature()
         if sig not in self._range_signatures:
             self._range_signatures.add(sig)
-            self._row_set.row_ranges.append(row_range._pb)
+            self._row_set.row_ranges.append(pb)
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
         """
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 28d7a8117567..ffa58ccd7954 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -873,3 +873,25 @@ def test_row_range_signature_distinguishes_empty_bytes_and_inclusivity(self):
         non_empty_query = self._make_one(row_ranges=[single_empty_key_range])
         empty_query = self._make_one()
         assert non_empty_query != empty_query
+
+    def test_add_range_pb_does_not_allocate_wrapper(self):
+        """
+        Verify that passing a RowRangePB directly to add_range computes the
+        deduplication signature via _signature_from_pb without allocating a
+        temporary RowRange wrapper via _from_pb.
+        """
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+        from google.cloud.bigtable_v2.types import RowRange as RowRangePB
+
+        query = self._make_one()
+        pb = RowRangePB(start_key_closed=b"a", end_key_open=b"b")
+
+        with mock.patch.object(
+            RowRange, "_from_pb", wraps=RowRange._from_pb
+        ) as mock_from_pb:
+            query.add_range(pb)
+            query.add_range(pb)
+            mock_from_pb.assert_not_called()
+
+        assert len(query.row_ranges) == 1
+        assert query.row_ranges[0] == RowRange(start_key=b"a", end_key=b"b")

From d88fecbebe1a6f78dacae65dbcaad1e0d84a29aa Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 20:52:23 +0000
Subject: [PATCH 20/34] run nox -s generate_sync

---
 .../google/cloud/bigtable/data/_sync_autogen/_read_rows.py    | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py
index ddb73ad175bb..f8c2d28b2903 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_sync_autogen/_read_rows.py
@@ -93,10 +93,6 @@ def __init__(
             attempt_timeout, operation_timeout
         )
         self.operation_timeout = operation_timeout
-        # ReadRowsQuery defines __slots__, which gives it a C-level memory layout
-        # disjoint from dict. Consequently, `query` must be explicitly typed as
-        # `ReadRowsQuery | dict[str, object]` in the signature above; otherwise,
-        # mypy flags this runtime `isinstance(query, dict)` branch as unreachable.
         if isinstance(query, dict):
             self.request = ReadRowsRequestPB(
                 **query, **target._request_path, app_profile_id=target.app_profile_id

From 0f285aad3e25ba1115184dcde9518c6c00b9c6e9 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 20:58:29 +0000
Subject: [PATCH 21/34] address review feedback

---
 .../cloud/bigtable/data/read_rows_query.py    |  4 ++-
 .../tests/unit/data/test_read_rows_query.py   | 27 +++++++++++++++++++
 2 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index ad6836bb8f38..536fff8a5d0b 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -572,7 +572,9 @@ def _to_pb(self, table) -> ReadRowsRequestPB:
 
     def __copy__(self) -> ReadRowsQuery:
         new_query = ReadRowsQuery(limit=self.limit, row_filter=self.filter)
-        new_query._row_set = copy.deepcopy(self._row_set)
+        type(self._row_set).pb(new_query._row_set).CopyFrom(
+            type(self._row_set).pb(self._row_set)
+        )
         new_query._key_set = self._key_set.copy()
         new_query._range_signatures = self._range_signatures.copy()
         return new_query
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index ffa58ccd7954..da8cc3c3261e 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -895,3 +895,30 @@ def test_add_range_pb_does_not_allocate_wrapper(self):
 
         assert len(query.row_ranges) == 1
         assert query.row_ranges[0] == RowRange(start_key=b"a", end_key=b"b")
+
+    def test_copy_query_uses_protobuf_copy_from_without_deepcopy(self):
+        """
+        Verify that ReadRowsQuery.__copy__ clones _row_set using native
+        protobuf CopyFrom via type(_row_set).pb() rather than copy.deepcopy().
+        """
+        import copy
+
+        from google.cloud.bigtable.data.read_rows_query import RowRange
+        from google.cloud.bigtable_v2.types import RowSet as RowSetPB
+
+        r1 = RowRange(start_key=b"a", end_key=b"b")
+        original = self._make_one(row_keys=[b"k1"], row_ranges=[r1], limit=10)
+
+        with (
+            mock.patch("copy.deepcopy", wraps=copy.deepcopy) as mock_deepcopy,
+            mock.patch.object(RowSetPB, "pb", wraps=RowSetPB.pb) as mock_pb,
+        ):
+            copied = copy.copy(original)
+            mock_deepcopy.assert_not_called()
+            mock_pb.assert_any_call(original._row_set)
+            mock_pb.assert_any_call(copied._row_set)
+
+        assert copied == original
+        assert copied._row_set is not original._row_set
+        assert copied.row_keys == [b"k1"]
+        assert copied.row_ranges == [r1]

From 64125aa00147db1c2868538ac2df17d025a9d4a1 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Mon, 19 Oct 2026 21:14:41 +0000
Subject: [PATCH 22/34] lint

---
 .../google/cloud/bigtable/data/read_rows_query.py                | 1 -
 1 file changed, 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 536fff8a5d0b..e8731e2147de 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -14,7 +14,6 @@
 #
 from __future__ import annotations
 
-import copy
 from bisect import bisect_left, bisect_right
 from collections import defaultdict
 from typing import TYPE_CHECKING, Any

From 1024c83fbe45d19312412aad19a0fbcc26edbd3c Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 13:47:30 +0000
Subject: [PATCH 23/34] refactor(bigtable): remove HasField fallback in
 RowRange._signature_from_pb

---
 .../cloud/bigtable/data/read_rows_query.py    | 14 +------
 .../tests/unit/data/test_read_rows_query.py   | 40 +++++++++++--------
 2 files changed, 25 insertions(+), 29 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index e8731e2147de..1ffe6bbf4bb4 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -93,7 +93,7 @@ def _signature_from_pb(
     ) -> tuple[bytes | None, bool, bytes | None, bool]:
         raw_pb = pb._pb
 
-        # Fast path: non-empty keys bypass HasField and proto-plus overhead
+        # Bigtable treats falsy/empty keys as unbounded (None).
         start_closed = raw_pb.start_key_closed
         if start_closed:
             start_key: bytes | None = start_closed
@@ -103,12 +103,6 @@ def _signature_from_pb(
             if start_open:
                 start_key = start_open
                 start_is_inclusive = False
-            elif raw_pb.HasField("start_key_closed"):
-                start_key = b""
-                start_is_inclusive = True
-            elif raw_pb.HasField("start_key_open"):
-                start_key = b""
-                start_is_inclusive = False
             else:
                 start_key = None
                 start_is_inclusive = True
@@ -122,12 +116,6 @@ def _signature_from_pb(
             if end_closed:
                 end_key = end_closed
                 end_is_inclusive = True
-            elif raw_pb.HasField("end_key_open"):
-                end_key = b""
-                end_is_inclusive = False
-            elif raw_pb.HasField("end_key_closed"):
-                end_key = b""
-                end_is_inclusive = True
             else:
                 end_key = None
                 end_is_inclusive = False
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index da8cc3c3261e..fd91d64b569e 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -840,39 +840,47 @@ def test_copy_query_preserves_deduplication_isolation(self):
         assert original.row_keys == [b"k1", b"k2"]
         assert original.row_ranges == [r1, r2]
 
-    def test_row_range_signature_distinguishes_empty_bytes_and_inclusivity(self):
+    def test_row_range_signature_collapses_falsy_keys_to_none(self):
         """
-        Verify that unset bounds (None), inclusive empty-byte bounds (b""),
-        and exclusive empty-byte bounds (b"") produce distinct signatures
-        and are not conflated during deduplication or query equality.
+        Verify that falsy empty-byte bounds (b"") in a RowRange protobuf are
+        collapsed into unbounded (None) bounds in _signature_from_pb, matching
+        RowRange.__init__, RowRange properties, and bool(RowRange).
         """
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
         unbounded_start = RowRange(start_key=None, end_key=b"z")
+        empty_init_start = RowRange(start_key=b"", end_key=b"z")
         closed_empty_start = RowRange._from_dict(
             {"start_key_closed": b"", "end_key_open": b"z"}
         )
         open_empty_start = RowRange._from_dict(
             {"start_key_open": b"", "end_key_open": b"z"}
         )
-        # All three ranges are semantically distinct and must not collide
-        assert unbounded_start != closed_empty_start
-        assert closed_empty_start != open_empty_start
-        assert unbounded_start._signature() != closed_empty_start._signature()
-        assert closed_empty_start._signature() != open_empty_start._signature()
+
+        # All four represent the same unbounded-start range [None, b"z")
+        expected_sig = (None, True, b"z", False)
+        assert unbounded_start._signature() == expected_sig
+        assert empty_init_start._signature() == expected_sig
+        assert closed_empty_start._signature() == expected_sig
+        assert open_empty_start._signature() == expected_sig
 
         query = self._make_one(
-            row_ranges=[unbounded_start, closed_empty_start, open_empty_start]
+            row_ranges=[
+                unbounded_start,
+                empty_init_start,
+                closed_empty_start,
+                open_empty_start,
+            ]
         )
-        assert len(query.row_ranges) == 3
+        assert len(query.row_ranges) == 1
 
-        # A range targeting the single row b"" must NOT compare equal to an empty query
-        single_empty_key_range = RowRange._from_dict(
+        # A range with only empty-byte bounds has bool(r) is False and must
+        # compare equal to an empty query.
+        empty_key_range = RowRange._from_dict(
             {"start_key_closed": b"", "end_key_closed": b""}
         )
-        non_empty_query = self._make_one(row_ranges=[single_empty_key_range])
-        empty_query = self._make_one()
-        assert non_empty_query != empty_query
+        assert bool(empty_key_range) is False
+        assert self._make_one(row_ranges=[empty_key_range]) == self._make_one()
 
     def test_add_range_pb_does_not_allocate_wrapper(self):
         """

From a4046e646c50bc4f19bdc5ce52d678ddc1a4726c Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:07:04 +0000
Subject: [PATCH 24/34] refactor(bigtable): move _key() to base RowRange and
 hide RowRangePB in add_range

---
 .../cloud/bigtable/data/read_rows_query.py    | 67 +++++--------------
 .../google/cloud/bigtable/row_set.py          |  8 ---
 .../tests/unit/data/test_read_rows_query.py   | 41 +++---------
 3 files changed, 27 insertions(+), 89 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 1ffe6bbf4bb4..3a658d57618c 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -87,46 +87,18 @@ def __init__(
                 init_dict["end_key_open"] = end_key
         self._pb = RowRangePB(**init_dict)
 
-    @staticmethod
-    def _signature_from_pb(
-        pb: RowRangePB,
-    ) -> tuple[bytes | None, bool, bytes | None, bool]:
-        raw_pb = pb._pb
-
-        # Bigtable treats falsy/empty keys as unbounded (None).
-        start_closed = raw_pb.start_key_closed
-        if start_closed:
-            start_key: bytes | None = start_closed
-            start_is_inclusive = True
-        else:
-            start_open = raw_pb.start_key_open
-            if start_open:
-                start_key = start_open
-                start_is_inclusive = False
-            else:
-                start_key = None
-                start_is_inclusive = True
-
-        end_open = raw_pb.end_key_open
-        if end_open:
-            end_key: bytes | None = end_open
-            end_is_inclusive = False
-        else:
-            end_closed = raw_pb.end_key_closed
-            if end_closed:
-                end_key = end_closed
-                end_is_inclusive = True
-            else:
-                end_key = None
-                end_is_inclusive = False
-
-        return (start_key, start_is_inclusive, end_key, end_is_inclusive)
-
-    def _signature(self) -> tuple[bytes | None, bool, bytes | None, bool]:
-        return self._signature_from_pb(self._pb)
+    def _key(
+        self,
+    ) -> tuple[bytes | None, bytes | None, bool | None, bool | None]:
+        return (
+            self.start_key,
+            self.end_key,
+            self.start_is_inclusive,
+            self.end_is_inclusive,
+        )
 
     def __hash__(self) -> int:
-        return hash(self._signature())
+        return hash(self._key())
 
     @property
     def start_key(self) -> bytes | None:
@@ -297,9 +269,9 @@ def __init__(
             row_keys = [row_keys]
         self._row_set = RowSetPB()
         self._key_set: set[bytes] = set()
-        self._range_signatures: set[tuple[bytes | None, bool, bytes | None, bool]] = (
-            set()
-        )
+        self._range_signatures: set[
+            tuple[bytes | None, bytes | None, bool | None, bool | None]
+        ] = set()
         for k in row_keys:
             self.add_key(k)
         for r in row_ranges:
@@ -396,7 +368,7 @@ def add_key(self, row_key: str | bytes):
 
     def add_range(
         self,
-        row_range: RowRange | RowRangePB,
+        row_range: RowRange,
     ):
         """
         Add a range of row keys to this query.
@@ -404,15 +376,10 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
-        if isinstance(row_range, RowRangePB):
-            pb = row_range
-            sig = RowRange._signature_from_pb(pb)
-        else:
-            pb = row_range._pb
-            sig = row_range._signature()
+        sig = row_range._key()
         if sig not in self._range_signatures:
             self._range_signatures.add(sig)
-            self._row_set.row_ranges.append(pb)
+            self._row_set.row_ranges.append(row_range._pb)
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
         """
@@ -581,7 +548,7 @@ def __eq__(self, other):
             return False
         # empty queries are equal
         if len(self._key_set) == 0 and len(other._key_set) == 0:
-            empty_sig = (None, True, None, False)
+            empty_sig = (None, None, True, True)
             this_range_empty = len(self._range_signatures) == 0 or (
                 self._range_signatures == {empty_sig}
             )
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
index d9a6bb08ab86..edb1bd58d755 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
@@ -159,14 +159,6 @@ class RowRange(_MappableAttributesMixin, BaseRowRange):
         "end_inclusive": "end_is_inclusive",
     }
 
-    def _key(self):
-        return (
-            self.start_key,
-            self.end_key,
-            self.start_is_inclusive,
-            self.end_is_inclusive,
-        )
-
     def get_range_kwargs(self):
         """Convert row range object to dict which can be passed to
         google.bigtable.v2.RowRange add method.
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index fd91d64b569e..3b2f18bcd5b1 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -715,6 +715,8 @@ def test_row_range_hash_matches_legacy(self):
         # Equal RowRange objects must produce identical hashes so set membership
         # operates consistently across base and legacy types.
         assert base_range == legacy_range
+        assert LegacyRowRange._key is RowRange._key
+        assert LegacyRowRange.__hash__ is RowRange.__hash__
         assert hash(base_range) == hash(legacy_range)
 
         # Adding an equivalent legacy range to a query initialized with a base range
@@ -840,11 +842,11 @@ def test_copy_query_preserves_deduplication_isolation(self):
         assert original.row_keys == [b"k1", b"k2"]
         assert original.row_ranges == [r1, r2]
 
-    def test_row_range_signature_collapses_falsy_keys_to_none(self):
+    def test_row_range_key_collapses_falsy_keys_to_none(self):
         """
         Verify that falsy empty-byte bounds (b"") in a RowRange protobuf are
-        collapsed into unbounded (None) bounds in _signature_from_pb, matching
-        RowRange.__init__, RowRange properties, and bool(RowRange).
+        collapsed into unbounded (None) bounds in _key(), matching RowRange.__init__,
+        RowRange properties, and bool(RowRange).
         """
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
@@ -858,12 +860,11 @@ def test_row_range_signature_collapses_falsy_keys_to_none(self):
         )
 
         # All four represent the same unbounded-start range [None, b"z")
-        expected_sig = (None, True, b"z", False)
-        assert unbounded_start._signature() == expected_sig
-        assert empty_init_start._signature() == expected_sig
-        assert closed_empty_start._signature() == expected_sig
-        assert open_empty_start._signature() == expected_sig
-
+        expected_key = (None, b"z", True, False)
+        assert unbounded_start._key() == expected_key
+        assert empty_init_start._key() == expected_key
+        assert closed_empty_start._key() == expected_key
+        assert open_empty_start._key() == expected_key
         query = self._make_one(
             row_ranges=[
                 unbounded_start,
@@ -882,28 +883,6 @@ def test_row_range_signature_collapses_falsy_keys_to_none(self):
         assert bool(empty_key_range) is False
         assert self._make_one(row_ranges=[empty_key_range]) == self._make_one()
 
-    def test_add_range_pb_does_not_allocate_wrapper(self):
-        """
-        Verify that passing a RowRangePB directly to add_range computes the
-        deduplication signature via _signature_from_pb without allocating a
-        temporary RowRange wrapper via _from_pb.
-        """
-        from google.cloud.bigtable.data.read_rows_query import RowRange
-        from google.cloud.bigtable_v2.types import RowRange as RowRangePB
-
-        query = self._make_one()
-        pb = RowRangePB(start_key_closed=b"a", end_key_open=b"b")
-
-        with mock.patch.object(
-            RowRange, "_from_pb", wraps=RowRange._from_pb
-        ) as mock_from_pb:
-            query.add_range(pb)
-            query.add_range(pb)
-            mock_from_pb.assert_not_called()
-
-        assert len(query.row_ranges) == 1
-        assert query.row_ranges[0] == RowRange(start_key=b"a", end_key=b"b")
-
     def test_copy_query_uses_protobuf_copy_from_without_deepcopy(self):
         """
         Verify that ReadRowsQuery.__copy__ clones _row_set using native

From 9f8cbd471e8b361eb2776c45025607585bb33129 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:19:16 +0000
Subject: [PATCH 25/34] remove comment

---
 .../google/cloud/bigtable/data/_async/_read_rows.py           | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py
index 4180d4a960d5..d6fcfc73520b 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/_read_rows.py
@@ -99,10 +99,6 @@ def __init__(
             attempt_timeout, operation_timeout
         )
         self.operation_timeout = operation_timeout
-        # ReadRowsQuery defines __slots__, which gives it a C-level memory layout
-        # disjoint from dict. Consequently, `query` must be explicitly typed as
-        # `ReadRowsQuery | dict[str, object]` in the signature above; otherwise,
-        # mypy flags this runtime `isinstance(query, dict)` branch as unreachable.
         if isinstance(query, dict):
             self.request = ReadRowsRequestPB(
                 **query,

From 371d796981f9b8d0460a2f3d5d8313dffdeee15e Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:27:17 +0000
Subject: [PATCH 26/34] fix build

---
 .../tests/unit/v2_client/test_table.py           | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

diff --git a/packages/google-cloud-bigtable/tests/unit/v2_client/test_table.py b/packages/google-cloud-bigtable/tests/unit/v2_client/test_table.py
index 433444e0b8a9..af1ed4004928 100644
--- a/packages/google-cloud-bigtable/tests/unit/v2_client/test_table.py
+++ b/packages/google-cloud-bigtable/tests/unit/v2_client/test_table.py
@@ -1299,41 +1299,41 @@ def test__create_row_request_row_range_row_set_conflict():
 
 
 def test__create_row_request_row_range_start_key():
+    from google.cloud.bigtable.data.read_rows_query import RowRange
     from google.cloud.bigtable.table import _create_row_request
-    from google.cloud.bigtable_v2.types import RowRange
 
     start_key = b"begin_key"
     result = _create_row_request(start_key=start_key)
-    expected_result = ReadRowsQuery(row_ranges=RowRange(start_key_closed=start_key))
+    expected_result = ReadRowsQuery(row_ranges=RowRange(start_key=start_key))
     assert result == expected_result
 
 
 def test__create_row_request_row_range_end_key():
+    from google.cloud.bigtable.data.read_rows_query import RowRange
     from google.cloud.bigtable.table import _create_row_request
-    from google.cloud.bigtable_v2.types import RowRange
 
     end_key = b"begin_key"
     result = _create_row_request(end_key=end_key)
-    expected_result = ReadRowsQuery(row_ranges=RowRange(end_key_open=end_key))
+    expected_result = ReadRowsQuery(row_ranges=RowRange(end_key=end_key))
     assert result == expected_result
 
 
 def test__create_row_request_row_range_both_keys():
+    from google.cloud.bigtable.data.read_rows_query import RowRange
     from google.cloud.bigtable.table import _create_row_request
-    from google.cloud.bigtable_v2.types import RowRange
 
     start_key = b"begin_key"
     end_key = b"end_key"
     result = _create_row_request(start_key=start_key, end_key=end_key)
     expected_result = ReadRowsQuery(
-        row_ranges=RowRange(start_key_closed=start_key, end_key_open=end_key)
+        row_ranges=RowRange(start_key=start_key, end_key=end_key)
     )
     assert result == expected_result
 
 
 def test__create_row_request_row_range_both_keys_inclusive():
+    from google.cloud.bigtable.data.read_rows_query import RowRange
     from google.cloud.bigtable.table import _create_row_request
-    from google.cloud.bigtable_v2.types import RowRange
 
     start_key = b"begin_key"
     end_key = b"end_key"
@@ -1341,7 +1341,7 @@ def test__create_row_request_row_range_both_keys_inclusive():
         start_key=start_key, end_key=end_key, end_inclusive=True
     )
     expected_result = ReadRowsQuery(
-        row_ranges=RowRange(start_key_closed=start_key, end_key_closed=end_key)
+        row_ranges=RowRange(start_key=start_key, end_key=end_key, end_is_inclusive=True)
     )
     assert result == expected_result
 

From b31232732eeaf8038d409abc7c7d69b03090994c Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:37:57 +0000
Subject: [PATCH 27/34] refactor(bigtable): remove duplicate _row_set state
 from ReadRowsQuery

---
 .../cloud/bigtable/data/read_rows_query.py    | 54 ++++++----------
 .../tests/unit/data/test_read_rows_query.py   | 64 ++-----------------
 2 files changed, 24 insertions(+), 94 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 3a658d57618c..86644b44a233 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -21,7 +21,6 @@
 from google.cloud.bigtable.data.row_filters import RowFilter
 from google.cloud.bigtable_v2.types import ReadRowsRequest as ReadRowsRequestPB
 from google.cloud.bigtable_v2.types import RowRange as RowRangePB
-from google.cloud.bigtable_v2.types import RowSet as RowSetPB
 
 if TYPE_CHECKING:
     from google.cloud.bigtable.data import RowKeySamples, ShardedQuery
@@ -247,9 +246,8 @@ class ReadRowsQuery:
     __slots__ = (
         "_limit",
         "_filter",
-        "_row_set",
-        "_key_set",
-        "_range_signatures",
+        "_row_keys",
+        "_row_ranges",
     )
 
     def __init__(
@@ -267,11 +265,8 @@ def __init__(
             row_ranges = [row_ranges]
         if not isinstance(row_keys, list):
             row_keys = [row_keys]
-        self._row_set = RowSetPB()
-        self._key_set: set[bytes] = set()
-        self._range_signatures: set[
-            tuple[bytes | None, bytes | None, bool | None, bool | None]
-        ] = set()
+        self._row_keys: dict[bytes, None] = {}
+        self._row_ranges: dict[RowRange, None] = {}
         for k in row_keys:
             self.add_key(k)
         for r in row_ranges:
@@ -287,7 +282,7 @@ def row_keys(self) -> list[bytes]:
         Returns:
             list[bytes]: the row keys in this query
         """
-        return list(self._row_set.row_keys)
+        return list(self._row_keys.keys())
 
     @property
     def row_ranges(self) -> list[RowRange]:
@@ -297,7 +292,7 @@ def row_ranges(self) -> list[RowRange]:
         Returns:
             list[RowRange]: the row ranges in this query
         """
-        return [RowRange._from_pb(r) for r in self._row_set.row_ranges]
+        return list(self._row_ranges.keys())
 
     @property
     def limit(self) -> int | None:
@@ -362,9 +357,7 @@ def add_key(self, row_key: str | bytes):
             row_key = row_key.encode()
         elif not isinstance(row_key, bytes):
             raise ValueError("row_key must be string or bytes")
-        if row_key not in self._key_set:
-            self._key_set.add(row_key)
-            self._row_set.row_keys.append(row_key)
+        self._row_keys[row_key] = None
 
     def add_range(
         self,
@@ -376,10 +369,7 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
-        sig = row_range._key()
-        if sig not in self._range_signatures:
-            self._range_signatures.add(sig)
-            self._row_set.row_ranges.append(row_range._pb)
+        self._row_ranges[row_range] = None
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
         """
@@ -526,11 +516,8 @@ def _to_pb(self, table) -> ReadRowsRequestPB:
 
     def __copy__(self) -> ReadRowsQuery:
         new_query = ReadRowsQuery(limit=self.limit, row_filter=self.filter)
-        type(self._row_set).pb(new_query._row_set).CopyFrom(
-            type(self._row_set).pb(self._row_set)
-        )
-        new_query._key_set = self._key_set.copy()
-        new_query._range_signatures = self._range_signatures.copy()
+        new_query._row_keys = self._row_keys.copy()
+        new_query._row_ranges = self._row_ranges.copy()
         return new_query
 
     def __eq__(self, other):
@@ -547,24 +534,19 @@ def __eq__(self, other):
         if not isinstance(other, ReadRowsQuery):
             return False
         # empty queries are equal
-        if len(self._key_set) == 0 and len(other._key_set) == 0:
-            empty_sig = (None, None, True, True)
-            this_range_empty = len(self._range_signatures) == 0 or (
-                self._range_signatures == {empty_sig}
+        if len(self._row_keys) == 0 and len(other._row_keys) == 0:
+            empty_range = RowRange()
+            this_range_empty = len(self._row_ranges) == 0 or (
+                self._row_ranges.keys() == {empty_range}
             )
-            other_range_empty = len(other._range_signatures) == 0 or (
-                other._range_signatures == {empty_sig}
+            other_range_empty = len(other._row_ranges) == 0 or (
+                other._row_ranges.keys() == {empty_range}
             )
             if this_range_empty and other_range_empty:
                 return self.filter == other.filter and self.limit == other.limit
-        # otherwise, sets should have same sizes
-        if len(self._key_set) != len(other._key_set):
-            return False
-        if len(self._range_signatures) != len(other._range_signatures):
-            return False
         return (
-            self._key_set == other._key_set
-            and self._range_signatures == other._range_signatures
+            self._row_keys.keys() == other._row_keys.keys()
+            and self._row_ranges.keys() == other._row_ranges.keys()
             and self.filter == other.filter
             and self.limit == other.limit
         )
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 3b2f18bcd5b1..9135deb77fea 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -618,25 +618,22 @@ def test_add_range_does_not_materialize_row_ranges(self):
             # ASSERT: The slow, O(N) wrapper-building property was never accessed
             row_ranges.assert_not_called()
 
-        # Verify that all ranges were successfully committed to the underlying protobuf
-        assert len(query._row_set.row_ranges) == 100
+        # Verify that all ranges were successfully recorded in the query
+        assert len(query.row_ranges) == 100
 
     def test_add_key_deduplicates_via_set_without_scanning_row_set(self):
         """
         Verify that ReadRowsQuery deduplicates row_keys in __init__ and add_key
-        using its internal O(1) key set without linearly scanning _row_set.row_keys.
+        in O(1) time without materializing the row_keys property list.
         """
         # Duplicate keys passed to __init__ (even in mixed str/bytes forms)
         # must be deduplicated on construction.
         query = self._make_one(row_keys=["k0", b"k0"])
         assert query.row_keys == [b"k0"]
 
-        # When adding keys that are already present in the query, add_key must
-        # short-circuit via _key_set without accessing _row_set.row_keys.
-        mock_row_set = mock.MagicMock()
-        row_keys_prop = mock.PropertyMock()
-        type(mock_row_set).row_keys = row_keys_prop
-        with mock.patch.object(query, "_row_set", mock_row_set):
+        with mock.patch.object(
+            type(query), "row_keys", new_callable=mock.PropertyMock
+        ) as row_keys_prop:
             query.add_key(b"k0")
             query.add_key("k0")
             row_keys_prop.assert_not_called()
@@ -725,28 +722,6 @@ def test_row_range_hash_matches_legacy(self):
         query.add_range(legacy_range)
         assert len(query.row_ranges) == 1
 
-    def test_add_range_immune_to_caller_mutating_row_range_after_insert(self):
-        """
-        Verify that mutating a RowRange's underlying protobuf after adding it to
-        a query does not corrupt the query's internal deduplication set.
-        """
-        from google.cloud.bigtable.data.read_rows_query import RowRange
-
-        query = self._make_one()
-        r1 = RowRange(start_key=b"a", end_key=b"b")
-        query.add_range(r1)
-
-        # Mutate the caller's underlying RowRangePB in place after insertion.
-        r1._pb.start_key_closed = b"x"
-        r1._pb.end_key_open = b"y"
-
-        # Adding a new RowRange with the original ("a", "b") bounds must still be
-        # detected as a duplicate of the previously added range.
-        query.add_range(RowRange(start_key=b"a", end_key=b"b"))
-
-        assert len(query.row_ranges) == 1
-        assert query.row_ranges[0] == RowRange(start_key=b"a", end_key=b"b")
-
     def test_row_range_from_dict_does_not_call_init(self):
         """
         Verify that RowRange._from_dict constructs the wrapper instance directly
@@ -882,30 +857,3 @@ def test_row_range_key_collapses_falsy_keys_to_none(self):
         )
         assert bool(empty_key_range) is False
         assert self._make_one(row_ranges=[empty_key_range]) == self._make_one()
-
-    def test_copy_query_uses_protobuf_copy_from_without_deepcopy(self):
-        """
-        Verify that ReadRowsQuery.__copy__ clones _row_set using native
-        protobuf CopyFrom via type(_row_set).pb() rather than copy.deepcopy().
-        """
-        import copy
-
-        from google.cloud.bigtable.data.read_rows_query import RowRange
-        from google.cloud.bigtable_v2.types import RowSet as RowSetPB
-
-        r1 = RowRange(start_key=b"a", end_key=b"b")
-        original = self._make_one(row_keys=[b"k1"], row_ranges=[r1], limit=10)
-
-        with (
-            mock.patch("copy.deepcopy", wraps=copy.deepcopy) as mock_deepcopy,
-            mock.patch.object(RowSetPB, "pb", wraps=RowSetPB.pb) as mock_pb,
-        ):
-            copied = copy.copy(original)
-            mock_deepcopy.assert_not_called()
-            mock_pb.assert_any_call(original._row_set)
-            mock_pb.assert_any_call(copied._row_set)
-
-        assert copied == original
-        assert copied._row_set is not original._row_set
-        assert copied.row_keys == [b"k1"]
-        assert copied.row_ranges == [r1]

From aba57f9e8c206d639731c1592139fe129063260a Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:42:20 +0000
Subject: [PATCH 28/34] fix(bigtable): use _key() in RowRange.__eq__ for
 consistency with __hash__

---
 .../google/cloud/bigtable/data/read_rows_query.py               | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 86644b44a233..08c5a709a062 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -200,7 +200,7 @@ def __bool__(self) -> bool:
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, RowRange):
             return NotImplemented
-        return self._pb == other._pb
+        return self._key() == other._key()
 
     def __str__(self) -> str:
         """

From d6c7441fe07cdd3ba27e6004c671ad4ed216867e Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:48:49 +0000
Subject: [PATCH 29/34] chore(bigtable): revert unneeded optimizations and code
 clean up

---
 .../cloud/bigtable/data/read_rows_query.py    |  7 +--
 .../google/cloud/bigtable/helpers.py          |  2 -
 .../google/cloud/bigtable/row_set.py          |  2 -
 .../tests/unit/data/test_read_rows_query.py   | 57 +------------------
 4 files changed, 6 insertions(+), 62 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 08c5a709a062..9c7b64a2d4ac 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -159,8 +159,7 @@ def _from_pb(cls, data: RowRangePB) -> RowRange:
         Returns:
             RowRange: The converted RowRange
         """
-        # Skip __init__ to avoid allocating an unused RowRangePB
-        instance = cls.__new__(cls)
+        instance = cls()
         instance._pb = data
         return instance
 
@@ -177,8 +176,7 @@ def _from_dict(cls, data: dict[str, bytes | str]) -> RowRange:
         formatted_data = {
             k: v.encode() if isinstance(v, str) else v for k, v in data.items()
         }
-        # Skip __init__ to avoid allocating an unused RowRangePB
-        instance = cls.__new__(cls)
+        instance = cls()
         instance._pb = RowRangePB(**formatted_data)
         return instance
 
@@ -265,6 +263,7 @@ def __init__(
             row_ranges = [row_ranges]
         if not isinstance(row_keys, list):
             row_keys = [row_keys]
+        # Use dicts as ordered sets to preserve insertion order with O(1) deduplication
         self._row_keys: dict[bytes, None] = {}
         self._row_ranges: dict[RowRange, None] = {}
         for k in row_keys:
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py
index 933dd4614c4e..71af2f539f2c 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/helpers.py
@@ -43,8 +43,6 @@ class and define a map _attribute_map from legacy client attributes to data clie
     Attributes are remapped and redefined in __init__ as well as getattr/setattr.
     """
 
-    __slots__ = ()
-
     def __init__(self, *args, **kwargs):
         new_kwargs = {self._attribute_map.get(k, k): v for (k, v) in kwargs.items()}
         super(_MappableAttributesMixin, self).__init__(*args, **new_kwargs)
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
index edb1bd58d755..6e041e3ed7b8 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/row_set.py
@@ -152,8 +152,6 @@ class RowRange(_MappableAttributesMixin, BaseRowRange):
                   considered inclusive. The default is False (exclusive).
     """
 
-    __slots__ = ()
-
     _attribute_map = {
         "start_inclusive": "start_is_inclusive",
         "end_inclusive": "end_is_inclusive",
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index 9135deb77fea..d8128ee599a5 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -591,37 +591,22 @@ def test_empty_row_set(self):
 
     def test_add_range_does_not_materialize_row_ranges(self):
         """
-        Ensures add_range scales in O(1) time and does not trigger O(N) scans.
-
-        The legacy implementation called the `row_ranges` property helper during
-        membership checks, which dynamically re-allocated and materialized brand
-        new Python RowRange wrapper objects for every element in the array on
-        each insertion. This led to an accidental O(N^2) memory and CPU leak.
-
-        This test mocks out the `row_ranges` property on the class and asserts
-        that it is never accessed during batch additions, proving we bypass the
-        quadratic evaluation path entirely.
+        Verify that add_range deduplicates in O(1) time without accessing the
+        row_ranges property list.
         """
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
         query = self._make_one()
 
-        # Intercept the `row_ranges` class property with a mock to track accesses
         with mock.patch.object(
             type(query), "row_ranges", new_callable=mock.PropertyMock
         ) as row_ranges:
             for i in range(100):
-                # Add unique sequential ranges. Utilizing our fast O(1) set signature lookup
-                # should never touch the slow wrapper-materialization properties.
                 query.add_range(RowRange(f"k{i:03d}", f"k{i + 1:03d}"))
-
-            # ASSERT: The slow, O(N) wrapper-building property was never accessed
             row_ranges.assert_not_called()
-
-        # Verify that all ranges were successfully recorded in the query
         assert len(query.row_ranges) == 100
 
-    def test_add_key_deduplicates_via_set_without_scanning_row_set(self):
+    def test_add_key_does_not_materialize_row_keys(self):
         """
         Verify that ReadRowsQuery deduplicates row_keys in __init__ and add_key
         in O(1) time without materializing the row_keys property list.
@@ -722,29 +707,6 @@ def test_row_range_hash_matches_legacy(self):
         query.add_range(legacy_range)
         assert len(query.row_ranges) == 1
 
-    def test_row_range_from_dict_does_not_call_init(self):
-        """
-        Verify that RowRange._from_dict constructs the wrapper instance directly
-        without invoking __init__ and allocating an unused intermediate protobuf.
-        """
-        from google.cloud.bigtable.data.read_rows_query import RowRange
-
-        # Spy on RowRange.__init__ to ensure dictionary deserialization bypasses
-        # default constructor initialization and dummy protobuf creation.
-        with mock.patch.object(
-            RowRange, "__init__", wraps=RowRange.__init__
-        ) as mock_init:
-            row_range = RowRange._from_dict(
-                {"start_key_closed": b"a", "end_key_open": b"b"}
-            )
-            mock_init.assert_not_called()
-
-        # Verify the resulting RowRange exposes the expected boundaries and inclusivity.
-        assert row_range.start_key == b"a"
-        assert row_range.start_is_inclusive is True
-        assert row_range.end_key == b"b"
-        assert row_range.end_is_inclusive is False
-
     def test_read_rows_query_enforces_slots(self):
         """
         Verify that ReadRowsQuery enforces __slots__ so instances do not allocate
@@ -766,19 +728,6 @@ def test_read_rows_query_enforces_slots(self):
         with pytest.raises(AttributeError):
             query._undeclared_attribute = "invalid"
 
-    def test_legacy_row_range_enforces_slots(self):
-        """
-        Verify that legacy row_set.RowRange also defines __slots__ = () so it
-        inherits BaseRowRange's slotted layout without allocating a __dict__.
-        """
-        from google.cloud.bigtable.row_set import RowRange as LegacyRowRange
-
-        legacy_range = LegacyRowRange(start_key=b"a", end_key=b"b")
-        assert not hasattr(legacy_range, "__dict__")
-
-        with pytest.raises(AttributeError):
-            legacy_range._undeclared_attribute = "invalid"
-
     def test_copy_query_preserves_deduplication_isolation(self):
         """
         Verify that shallow-copying a ReadRowsQuery produces independent

From c3e903ae407a4a1d13f71935343bc4bb069b9801 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 14:56:57 +0000
Subject: [PATCH 30/34] refactor(bigtable): build RowSetPB in
 ReadRowsQuery._to_pb

---
 .../google/cloud/bigtable/data/read_rows_query.py           | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 9c7b64a2d4ac..c2aacc53530d 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -21,6 +21,7 @@
 from google.cloud.bigtable.data.row_filters import RowFilter
 from google.cloud.bigtable_v2.types import ReadRowsRequest as ReadRowsRequestPB
 from google.cloud.bigtable_v2.types import RowRange as RowRangePB
+from google.cloud.bigtable_v2.types import RowSet as RowSetPB
 
 if TYPE_CHECKING:
     from google.cloud.bigtable.data import RowKeySamples, ShardedQuery
@@ -509,7 +510,10 @@ def _to_pb(self, table) -> ReadRowsRequestPB:
             app_profile_id=table.app_profile_id,
             filter=self.filter._to_pb() if self.filter else None,
             rows_limit=self.limit or 0,
-            rows=self._row_set,
+            rows=RowSetPB(
+                row_keys=list(self._row_keys.keys()),
+                row_ranges=[r._pb for r in self._row_ranges.keys()],
+            ),
             **table._request_path,
         )
 

From 4f6ea9795032c007ca644e96743eadabe3ea1c7c Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 15:01:22 +0000
Subject: [PATCH 31/34] add comment

---
 .../google/cloud/bigtable/data/read_rows_query.py                | 1 +
 1 file changed, 1 insertion(+)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index c2aacc53530d..cc29f11a9580 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -369,6 +369,7 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
+        # Insert as dict key to deduplicate in O(1) while preserving insertion order
         self._row_ranges[row_range] = None
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:

From 47960f85ec1390f9306a7045cb5a3133ef8c24bf Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 15:04:46 +0000
Subject: [PATCH 32/34] minor feedback from offline review

---
 .../google/cloud/bigtable/data/read_rows_query.py        | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index cc29f11a9580..2bae61a47c77 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -512,8 +512,8 @@ def _to_pb(self, table) -> ReadRowsRequestPB:
             filter=self.filter._to_pb() if self.filter else None,
             rows_limit=self.limit or 0,
             rows=RowSetPB(
-                row_keys=list(self._row_keys.keys()),
-                row_ranges=[r._pb for r in self._row_ranges.keys()],
+                row_keys=self.row_keys,
+                row_ranges=[r._pb for r in self._row_ranges],
             ),
             **table._request_path,
         )
@@ -539,12 +539,11 @@ def __eq__(self, other):
             return False
         # empty queries are equal
         if len(self._row_keys) == 0 and len(other._row_keys) == 0:
-            empty_range = RowRange()
             this_range_empty = len(self._row_ranges) == 0 or (
-                self._row_ranges.keys() == {empty_range}
+                [bool(r) is False for r in self._row_ranges]
             )
             other_range_empty = len(other._row_ranges) == 0 or (
-                other._row_ranges.keys() == {empty_range}
+                [bool(r) is False for r in other._row_ranges]
             )
             if this_range_empty and other_range_empty:
                 return self.filter == other.filter and self.limit == other.limit

From de1b465fe40d712b2b4420b388858df48388d746 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 15:09:41 +0000
Subject: [PATCH 33/34] fix(bigtable): restore all() in ReadRowsQuery.__eq__
 and remove redundant test

---
 .../cloud/bigtable/data/read_rows_query.py    |  5 +--
 .../tests/unit/data/test_read_rows_query.py   | 41 -------------------
 2 files changed, 2 insertions(+), 44 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index 2bae61a47c77..c43315a383e5 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -369,7 +369,6 @@ def add_range(
         Args:
             row_range: a range of row keys to add to this query
         """
-        # Insert as dict key to deduplicate in O(1) while preserving insertion order
         self._row_ranges[row_range] = None
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
@@ -539,10 +538,10 @@ def __eq__(self, other):
             return False
         # empty queries are equal
         if len(self._row_keys) == 0 and len(other._row_keys) == 0:
-            this_range_empty = len(self._row_ranges) == 0 or (
+            this_range_empty = len(self._row_ranges) == 0 or all(
                 [bool(r) is False for r in self._row_ranges]
             )
-            other_range_empty = len(other._row_ranges) == 0 or (
+            other_range_empty = len(other._row_ranges) == 0 or all(
                 [bool(r) is False for r in other._row_ranges]
             )
             if this_range_empty and other_range_empty:
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index d8128ee599a5..ace6a8798074 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -765,44 +765,3 @@ def test_copy_query_preserves_deduplication_isolation(self):
 
         assert original.row_keys == [b"k1", b"k2"]
         assert original.row_ranges == [r1, r2]
-
-    def test_row_range_key_collapses_falsy_keys_to_none(self):
-        """
-        Verify that falsy empty-byte bounds (b"") in a RowRange protobuf are
-        collapsed into unbounded (None) bounds in _key(), matching RowRange.__init__,
-        RowRange properties, and bool(RowRange).
-        """
-        from google.cloud.bigtable.data.read_rows_query import RowRange
-
-        unbounded_start = RowRange(start_key=None, end_key=b"z")
-        empty_init_start = RowRange(start_key=b"", end_key=b"z")
-        closed_empty_start = RowRange._from_dict(
-            {"start_key_closed": b"", "end_key_open": b"z"}
-        )
-        open_empty_start = RowRange._from_dict(
-            {"start_key_open": b"", "end_key_open": b"z"}
-        )
-
-        # All four represent the same unbounded-start range [None, b"z")
-        expected_key = (None, b"z", True, False)
-        assert unbounded_start._key() == expected_key
-        assert empty_init_start._key() == expected_key
-        assert closed_empty_start._key() == expected_key
-        assert open_empty_start._key() == expected_key
-        query = self._make_one(
-            row_ranges=[
-                unbounded_start,
-                empty_init_start,
-                closed_empty_start,
-                open_empty_start,
-            ]
-        )
-        assert len(query.row_ranges) == 1
-
-        # A range with only empty-byte bounds has bool(r) is False and must
-        # compare equal to an empty query.
-        empty_key_range = RowRange._from_dict(
-            {"start_key_closed": b"", "end_key_closed": b""}
-        )
-        assert bool(empty_key_range) is False
-        assert self._make_one(row_ranges=[empty_key_range]) == self._make_one()

From c3ac92d05f630da54a9d165ff0a8f18c62578894 Mon Sep 17 00:00:00 2001
From: Anthonios Partheniou <partheniou@google.com>
Date: Tue, 20 Oct 2026 15:17:52 +0000
Subject: [PATCH 34/34] chore(bigtable): revert RowRange.__eq__ change and trim
 verbose test comments)

---
 .../cloud/bigtable/data/read_rows_query.py    |  6 +--
 .../tests/unit/data/test_read_rows_query.py   | 38 ++-----------------
 2 files changed, 6 insertions(+), 38 deletions(-)

diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
index c43315a383e5..938303ea5a4c 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/read_rows_query.py
@@ -199,7 +199,7 @@ def __bool__(self) -> bool:
     def __eq__(self, other: Any) -> bool:
         if not isinstance(other, RowRange):
             return NotImplemented
-        return self._key() == other._key()
+        return self._pb == other._pb
 
     def __str__(self) -> str:
         """
@@ -282,7 +282,7 @@ def row_keys(self) -> list[bytes]:
         Returns:
             list[bytes]: the row keys in this query
         """
-        return list(self._row_keys.keys())
+        return list(self._row_keys)
 
     @property
     def row_ranges(self) -> list[RowRange]:
@@ -292,7 +292,7 @@ def row_ranges(self) -> list[RowRange]:
         Returns:
             list[RowRange]: the row ranges in this query
         """
-        return list(self._row_ranges.keys())
+        return list(self._row_ranges)
 
     @property
     def limit(self) -> int | None:
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
index ace6a8798074..cfef2da4b793 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test_read_rows_query.py
@@ -611,8 +611,6 @@ def test_add_key_does_not_materialize_row_keys(self):
         Verify that ReadRowsQuery deduplicates row_keys in __init__ and add_key
         in O(1) time without materializing the row_keys property list.
         """
-        # Duplicate keys passed to __init__ (even in mixed str/bytes forms)
-        # must be deduplicated on construction.
         query = self._make_one(row_keys=["k0", b"k0"])
         assert query.row_keys == [b"k0"]
 
@@ -632,8 +630,6 @@ def test_query_equality_does_not_materialize_row_keys_or_ranges(self):
         """
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
-        # Create two equivalent queries with identical keys and ranges inserted
-        # in opposite orders.
         q1 = self._make_one(
             row_keys=[f"k{i}" for i in range(50)],
             row_ranges=[RowRange(f"a{i}", f"b{i}") for i in range(50)],
@@ -642,9 +638,6 @@ def test_query_equality_does_not_materialize_row_keys_or_ranges(self):
             row_keys=[f"k{i}" for i in reversed(range(50))],
             row_ranges=[RowRange(f"a{i}", f"b{i}") for i in reversed(range(50))],
         )
-
-        # Intercept the row_keys and row_ranges properties to verify that equality
-        # comparisons operate on internal sets without constructing wrapper lists.
         with (
             mock.patch.object(
                 type(q1), "row_keys", new_callable=mock.PropertyMock
@@ -661,22 +654,16 @@ def test_add_range_deduplication_behavior(self):
         """Assert that identical ranges are deduplicated, but distinct ranges are kept."""
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
-        # Create two identical ranges
         r1 = RowRange(start_key="a", end_key="b")
         r2 = RowRange(start_key="a", end_key="b")
 
-        # Create a distinct range
         r3 = RowRange(start_key="c", end_key="d")
 
-        # Seed r1 via __init__, then add r2 (duplicate) and r3 (distinct)
         query = self._make_one(row_ranges=[r1])
-        query.add_range(r2)  # Duplicate of constructor-seeded r1, should be ignored
-        query.add_range(r3)  # Distinct, should be kept
-        query.add_range(
-            RowRange(start_key="c", end_key="d")
-        )  # Duplicate of r3, should be ignored
+        query.add_range(r2)
+        query.add_range(r3)
+        query.add_range(RowRange(start_key="c", end_key="d"))
 
-        # Assert that only r1 and r3 are in the final query
         assert len(query.row_ranges) == 2
         assert query.row_ranges[0] == r1
         assert query.row_ranges[1] == r3
@@ -690,19 +677,12 @@ def test_row_range_hash_matches_legacy(self):
         from google.cloud.bigtable.data.read_rows_query import RowRange
         from google.cloud.bigtable.row_set import RowRange as LegacyRowRange
 
-        # Instantiate equivalent ranges using both the data API and legacy classes.
         base_range = RowRange(b"a", b"b")
         legacy_range = LegacyRowRange(b"a", b"b")
-
-        # Equal RowRange objects must produce identical hashes so set membership
-        # operates consistently across base and legacy types.
         assert base_range == legacy_range
         assert LegacyRowRange._key is RowRange._key
         assert LegacyRowRange.__hash__ is RowRange.__hash__
         assert hash(base_range) == hash(legacy_range)
-
-        # Adding an equivalent legacy range to a query initialized with a base range
-        # should be recognized as a duplicate and ignored.
         query = self._make_one(row_ranges=[base_range])
         query.add_range(legacy_range)
         assert len(query.row_ranges) == 1
@@ -714,17 +694,11 @@ def test_read_rows_query_enforces_slots(self):
         """
         from google.cloud.bigtable.data.read_rows_query import RowRange
 
-        # Initialize a query with both keys and ranges so all internal attributes
-        # are populated on the instance.
         query = self._make_one(
             row_keys=[b"k1"],
             row_ranges=[RowRange(start_key=b"a", end_key=b"b")],
         )
-
-        # Enforcing __slots__ prevents creation of a per-instance __dict__.
         assert not hasattr(query, "__dict__")
-
-        # Assigning an attribute not declared in __slots__ must raise AttributeError.
         with pytest.raises(AttributeError):
             query._undeclared_attribute = "invalid"
 
@@ -744,22 +718,16 @@ def test_copy_query_preserves_deduplication_isolation(self):
         original = self._make_one(row_keys=[b"k1"], row_ranges=[r1])
         copied = copy.copy(original)
 
-        # Re-adding existing items to the copy must be ignored as duplicates.
         copied.add_key(b"k1")
         copied.add_range(r1)
         assert copied.row_keys == [b"k1"]
         assert copied.row_ranges == [r1]
 
-        # Mutate only the copied query by adding a new key and range.
         copied.add_key(b"k2")
         copied.add_range(r2)
-
-        # Confirm the original query was not modified by operations on the copy.
         assert original.row_keys == [b"k1"]
         assert original.row_ranges == [r1]
 
-        # Adding the same new key and range to the original query must succeed
-        # because its internal deduplication sets are independent from the copy.
         original.add_key(b"k2")
         original.add_range(r2)