parthea · Oct 15, 2026 · Oct 15, 2026 · Oct 15, 2026 · Oct 15, 2026 · Oct 15, 2026
@@ -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,

@@ -82,7 +82,7 @@ class _ReadRowsOperation:
 
     def __init__(
         self,
-        query: ReadRowsQuery,
+        query: ReadRowsQuery | dict[str, object],
         target: TargetType,
         operation_timeout: float,
         attempt_timeout: float,

@@ -87,6 +87,59 @@ 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
+
+        # 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())
+
     @property
     def start_key(self) -> bytes | None:
         """
@@ -147,7 +200,8 @@ def _from_pb(cls, data: RowRangePB) -> RowRange:
         Returns:
             RowRange: The converted RowRange
         """
-        instance = cls()
+        # Skip __init__ to avoid allocating an unused RowRangePB
+        instance = cls.__new__(cls)
         instance._pb = data
         return instance
 
@@ -164,7 +218,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
 
@@ -229,7 +284,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,
@@ -246,10 +307,15 @@ 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, row_ranges=[r._pb for r in row_ranges]
+        self._row_set = RowSetPB()
+        self._key_set: set[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:
+            self.add_range(r)
         self.limit = limit or None
         self.filter = row_filter
 
@@ -336,21 +402,29 @@ 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(
         self,
-        row_range: RowRange,
+        row_range: RowRange | RowRangePB,
     ):
         """
         Add a range of row keys to this query.
 
         Args:
             row_range: a range of row keys to add to this query
         """
-        if row_range not in self.row_ranges:
-            self._row_set.row_ranges.append(row_range._pb)
+        if isinstance(row_range, RowRangePB):
+            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(pb)
 
     def shard(self, shard_keys: RowKeySamples) -> ShardedQuery:
         """
@@ -495,6 +569,15 @@ def _to_pb(self, table) -> ReadRowsRequestPB:
             **table._request_path,
         )
 
+    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()
+        return new_query
+
     def __eq__(self, other):
         """
         RowRanges are equal if they have the same row keys, row ranges,
@@ -509,24 +592,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 = (None, True, None, False)
+            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
         )

@@ -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)

@@ -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",
@@ -165,9 +167,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.