Add BytesReader to replace BytesIO in decode_message()

Introduce a lightweight BytesReader class that provides the same
read() interface as io.BytesIO but operates directly on the input
bytes/memoryview without internal buffering overhead.

Changes:
- Add BytesReader class with __slots__ for memory efficiency
- Replace io.BytesIO(body) with BytesReader(body) in decode_message()
- BytesReader.read() returns slices directly, converting memoryview
  to bytes only when necessary for compatibility

Benefits:
- Eliminates BytesIO's internal buffer allocation and management
- Reduces memory overhead for protocol message decoding
- Works seamlessly with both bytes and memoryview inputs
- Maintains full API compatibility with existing read_* functions

The BytesReader is a minimal implementation focused on the read()
method needed by the protocol decoder. It avoids the overhead of
io.BytesIO's full file-like interface.

Signed-off-by: Yaniv Kaul <ykaul@scylladb.com>

Author: mykaul

cassandra/protocol.py

@@ -53,6 +53,34 @@ class NotSupportedError(Exception):
 class InternalError(Exception):
     pass
 
+
+class BytesReader:
+    """
+    Lightweight reader for bytes data without BytesIO overhead.
+    Provides the same read() interface but operates directly on a
+    bytes or memoryview object, avoiding internal buffer copies.
+    """
+    __slots__ = ('_data', '_pos', '_size')
+
+    def __init__(self, data):
+        self._data = data
+        self._pos = 0
+        self._size = len(data)
+
+    def read(self, n=-1):
+        if n < 0:
+            result = self._data[self._pos:]
+            self._pos = self._size
+        else:
+            end = self._pos + n
+            if end > self._size:
+                raise EOFError("Cannot read past the end of the buffer")
+            result = self._data[self._pos:end]
+            self._pos = end
+        # Return bytes to maintain compatibility with unpack functions
+        return bytes(result) if isinstance(result, memoryview) else result
+
+
 ColumnMetadata = namedtuple("ColumnMetadata", ['keyspace_name', 'table_name', 'name', 'type'])
 
 HEADER_DIRECTION_TO_CLIENT = 0x80
@@ -1142,7 +1170,8 @@ def decode_message(cls, protocol_version, protocol_features, user_type_map, stre
             body = decompressor(body)
             flags ^= COMPRESSED_FLAG
 
-        body = io.BytesIO(body)
+        # Use lightweight BytesReader instead of io.BytesIO to avoid buffer copy
+        body = BytesReader(body)
         if flags & TRACING_FLAG:
             trace_id = UUID(bytes=body.read(16))
             flags ^= TRACING_FLAG