perf(spanner): optimize single-chunk results and single-row lookups (#18602)

Optimize the streamed result set decoding pipeline for point queries and
small result sets, and fix async support in `to_dict_list()`.

Key changes:
- Zero-copy protobuf values: In `_consume_next()`, avoid copying
`response_pb.values` into a Python list on unchunked responses.
Downstream decoders read or slice directly from the protobuf container.
- Fast paths in `_merge_values()`:
- Point lookups (`total_values == width`): Directly decode and append
the single row via `_append_single_row()`, bypassing row count
arithmetic, stride looping, and index allocations.
- Complete chunks (`total_values % width == 0`): Skip partial-row
checking and directly batch-decode complete rows.
- Single-indexing in eager decoding: Cache cell lookups using the walrus
operator `(cell := values[...])` instead of indexing twice.
- Direct consumption in `one()` / `one_or_none()`: Replace the generator
state machine and iterator setup with a direct `_consume_next()` loop,
reducing latency on point queries and avoiding Python 3.14 async
generator cleanup warnings when queries short-circuit.
- Faster, async-safe `to_dict_list()`:
- Cache the column name list once on row 0 instead of re-inspecting
protobuf metadata on every row (~2x speedup).
- Add `@CrossSync.convert` with `async for` in `_async/streamed.py`,
fixing a latent bug where calling `to_dict_list()` on an async streamed
result set raised a TypeError.

Author: olavloite

packages/google-cloud-spanner/google/cloud/spanner_v1/_async/streamed.py

@@ -165,8 +165,8 @@ def _decode_eager_rows(self, values, values_offset, batch_end, width):
             decoder = self._decoders[0]
             self._rows.extend(
                 [
-                    [None if value.HasField("null_value") else decoder(value)]
-                    for value in values[values_offset:batch_end]
+                    [None if cell.HasField("null_value") else decoder(cell)]
+                    for cell in values[values_offset:batch_end]
                 ]
             )
         else:
@@ -177,16 +177,51 @@ def _decode_eager_rows(self, values, values_offset, batch_end, width):
                 rows_append(
                     [
                         None
-                        if values[row_start + column_index].HasField("null_value")
-                        else decoders[column_index](values[row_start + column_index])
+                        if (cell := values[row_start + column_index]).HasField(
+                            "null_value"
+                        )
+                        else decoders[column_index](cell)
                         for column_index in column_indices
                     ]
                 )
 
+    def _append_single_row(self, values, width):
+        """Decode and append a single complete row on the fast path."""
+        if self._lazy_decode:
+            self._rows.append(values[:width])
+        elif width == 1:
+            cell = values[0]
+            self._rows.append(
+                [None if cell.HasField("null_value") else self._decoders[0](cell)]
+            )
+        else:
+            decoders = self._decoders
+            self._rows.append(
+                [
+                    None
+                    if (cell := values[i]).HasField("null_value")
+                    else decoders[i](cell)
+                    for i in range(width)
+                ]
+            )
+
+    def _complete_partial_row(self, values, width, total_values):
+        """Complete a pending partial row from a previous chunk.
+
+        :returns: The number of values consumed to complete or fill the partial row.
+        """
+        needed = width - len(self._current_row)
+        fill_count = min(needed, total_values)
+        self._append_to_current_row(values[:fill_count])
+        if len(self._current_row) == width:
+            self._rows.append(self._current_row)
+            self._current_row = []
+        return fill_count
+
     def _merge_values(self, values):
         """Merge values into rows.
 
-        :type values: list of :class:`~google.protobuf.struct_pb2.Value`
+        :type values: list or RepeatedCompositeContainer of :class:`~google.protobuf.struct_pb2.Value`
         :param values: non-chunked values from partial result set.
         """
         if not values:
@@ -196,36 +231,44 @@ def _merge_values(self, values):
         if width == 0:
             return
 
-        values_offset = 0
         total_values = len(values)
 
-        # 1. Complete pending partial row from previous chunk (if any)
+        if not self._current_row:
+            # 1. Fast path for a single-row result (point lookup)
+            if total_values == width:
+                self._append_single_row(values, width)
+                return
+
+            # 2. Fast path when all rows in this chunk are complete (e.g. single PartialResultSet)
+            if total_values % width == 0:
+                if self._lazy_decode:
+                    self._decode_lazy_rows(values, 0, total_values, width)
+                else:
+                    self._decode_eager_rows(values, 0, total_values, width)
+                return
+
+        # 3. Complete pending partial row from previous chunk (if any)
+        values_offset = 0
         if self._current_row:
-            needed = width - len(self._current_row)
-            fill_count = min(needed, total_values)
-            self._append_to_current_row(values[:fill_count])
-            values_offset = fill_count
-            if len(self._current_row) == width:
-                self._rows.append(self._current_row)
-                self._current_row = []
-            else:
+            values_offset = self._complete_partial_row(values, width, total_values)
+            if self._current_row:
                 return
 
         remaining_values = total_values - values_offset
         if remaining_values == 0:
             return
 
+        # 4. Batch-decode complete rows
         row_count = remaining_values // width
         full_values_count = row_count * width
         batch_end = values_offset + full_values_count
 
-        # 2. Batch-decode complete rows
         if self._lazy_decode:
             self._decode_lazy_rows(values, values_offset, batch_end, width)
         else:
             self._decode_eager_rows(values, values_offset, batch_end, width)
 
-        # 3. Buffer trailing partial row remainder for the next chunk (if any)
+        # 5. Buffer trailing partial row remainder for the next chunk (if any)
         if remaining_values > full_values_count:
             self._append_to_current_row(values[batch_end:])
 
@@ -244,12 +287,14 @@ async def _consume_next(self):
         if response_pb.HasField("stats"):  # last response
             self._stats = response.stats
 
-        values = list(response_pb.values)
-        if self._pending_chunk is not None:
-            values[0] = self._merge_chunk(values[0])
-
-        if response_pb.chunked_value:
-            self._pending_chunk = values.pop()
+        if self._pending_chunk is not None or response_pb.chunked_value:
+            values = list(response_pb.values)
+            if self._pending_chunk is not None:
+                values[0] = self._merge_chunk(values[0])
+            if response_pb.chunked_value:
+                self._pending_chunk = values.pop()
+        else:
+            values = response_pb.values
 
         self._merge_values(values)
 
@@ -259,15 +304,18 @@ async def _consume_next(self):
     @CrossSync.convert(sync_name="__iter__")
     async def __aiter__(self):
         while True:
+            if not self._rows:
+                if self._done:
+                    return
+                try:
+                    await self._consume_next()
+                except StopAsyncIteration:
+                    return
             iter_rows, self._rows = self._rows, []
             for row in iter_rows:
                 yield row
             if self._done:
                 return
-            try:
-                await self._consume_next()
-            except StopAsyncIteration:
-                return
 
     def decode_row(self, row: []) -> []:
         """Decodes a row from protobuf values to Python objects. This function
@@ -327,23 +375,20 @@ async def one_or_none(self):
                 "stream consumption has already started."
             )
 
-        # Consume the first result of the stream.
-        # If there is no first result, then return None.
-        iterator = self.__aiter__()
-        try:
-            answer = await iterator.__anext__()
-        except StopAsyncIteration:
-            return None
+        while not self._done and len(self._rows) < 2:
+            try:
+                await self._consume_next()
+            except StopAsyncIteration:
+                break
 
-        # Attempt to consume more. This should no-op; if we get additional
-        # rows, then this is an error case.
-        try:
-            await iterator.__anext__()
-            raise ValueError("Expected one result; got more.")
-        except StopAsyncIteration:
-            return answer
+        if len(self._rows) == 1:
+            return self._rows.pop()
+        if not self._rows:
+            return None
+        raise ValueError("Expected one result; got more.")
 
-    def to_dict_list(self):
+    @CrossSync.convert
+    async def to_dict_list(self):
         """Return the result of a query as a list of dictionaries.
         In each dictionary the key is the column name and the value is the
         value of the that column in a given row.
@@ -353,15 +398,13 @@ def to_dict_list(self):
         :returns: result rows as a list of dictionaries
         """
         rows = []
-        for row in self:
-            rows.append(
-                {
-                    column: value
-                    for column, value in zip(
-                        [column.name for column in self._metadata.row_type.fields], row
-                    )
-                }
-            )
+        column_names = None
+        async for row in self:
+            if column_names is None:
+                column_names = [
+                    column.name for column in self._metadata.row_type.fields
+                ]
+            rows.append(dict(zip(column_names, row)))
         return rows