We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Spanner: IndexError: pop from empty list when merge_array
```
return merger(lhs, rhs, type_)
  File ".../google/cloud/spanner_v1/streamed.py", line 259, in _merge_array
    first = rhs.pop(0)
IndexError: pop from empty list
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `spanner/google/cloud/spanner_v1/streamed.py`
Content:
```
1 # Copyright 2016 Google Inc. All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 """Wrapper for streaming results."""
16 
17 from google.protobuf.struct_pb2 import ListValue
18 from google.protobuf.struct_pb2 import Value
19 from google.cloud import exceptions
20 from google.cloud.spanner_v1.proto import type_pb2
21 import six
22 
23 # pylint: disable=ungrouped-imports
24 from google.cloud.spanner_v1._helpers import _parse_value_pb
25 # pylint: enable=ungrouped-imports
26 
27 
28 class StreamedResultSet(object):
29     """Process a sequence of partial result sets into a single set of row data.
30 
31     :type response_iterator:
32     :param response_iterator:
33         Iterator yielding
34         :class:`google.cloud.spanner_v1.proto.result_set_pb2.PartialResultSet`
35         instances.
36 
37     :type source: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`
38     :param source: Snapshot from which the result set was fetched.
39     """
40     def __init__(self, response_iterator, source=None):
41         self._response_iterator = response_iterator
42         self._rows = []             # Fully-processed rows
43         self._counter = 0           # Counter for processed responses
44         self._metadata = None       # Until set from first PRS
45         self._stats = None          # Until set from last PRS
46         self._current_row = []      # Accumulated values for incomplete row
47         self._pending_chunk = None  # Incomplete value
48         self._source = source       # Source snapshot
49 
50     @property
51     def rows(self):
52         """Fully-processed rows.
53 
54         :rtype: list of row-data lists.
55         :returns: list of completed row data, from proceesd PRS responses.
56         """
57         return self._rows
58 
59     @property
60     def fields(self):
61         """Field descriptors for result set columns.
62 
63         :rtype: list of :class:`~google.cloud.spanner_v1.proto.type_pb2.Field`
64         :returns: list of fields describing column names / types.
65         """
66         return self._metadata.row_type.fields
67 
68     @property
69     def metadata(self):
70         """Result set metadata
71 
72         :rtype: :class:`~.result_set_pb2.ResultSetMetadata`
73         :returns: structure describing the results
74         """
75         return self._metadata
76 
77     @property
78     def stats(self):
79         """Result set statistics
80 
81         :rtype:
82            :class:`~google.cloud.spanner_v1.proto.result_set_pb2.ResultSetStats`
83         :returns: structure describing status about the response
84         """
85         return self._stats
86 
87     def _merge_chunk(self, value):
88         """Merge pending chunk with next value.
89 
90         :type value: :class:`~google.protobuf.struct_pb2.Value`
91         :param value: continuation of chunked value from previous
92                       partial result set.
93 
94         :rtype: :class:`~google.protobuf.struct_pb2.Value`
95         :returns: the merged value
96         """
97         current_column = len(self._current_row)
98         field = self.fields[current_column]
99         merged = _merge_by_type(self._pending_chunk, value, field.type)
100         self._pending_chunk = None
101         return merged
102 
103     def _merge_values(self, values):
104         """Merge values into rows.
105 
106         :type values: list of :class:`~google.protobuf.struct_pb2.Value`
107         :param values: non-chunked values from partial result set.
108         """
109         width = len(self.fields)
110         for value in values:
111             index = len(self._current_row)
112             field = self.fields[index]
113             self._current_row.append(_parse_value_pb(value, field.type))
114             if len(self._current_row) == width:
115                 self._rows.append(self._current_row)
116                 self._current_row = []
117 
118     def consume_next(self):
119         """Consume the next partial result set from the stream.
120 
121         Parse the result set into new/existing rows in :attr:`_rows`
122         """
123         response = six.next(self._response_iterator)
124         self._counter += 1
125 
126         if self._metadata is None:  # first response
127             metadata = self._metadata = response.metadata
128 
129             source = self._source
130             if source is not None and source._transaction_id is None:
131                 source._transaction_id = metadata.transaction.id
132 
133         if response.HasField('stats'):  # last response
134             self._stats = response.stats
135 
136         values = list(response.values)
137         if self._pending_chunk is not None:
138             values[0] = self._merge_chunk(values[0])
139 
140         if response.chunked_value:
141             self._pending_chunk = values.pop()
142 
143         self._merge_values(values)
144 
145     def consume_all(self):
146         """Consume the streamed responses until there are no more."""
147         while True:
148             try:
149                 self.consume_next()
150             except StopIteration:
151                 break
152 
153     def __iter__(self):
154         iter_rows, self._rows[:] = self._rows[:], ()
155         while True:
156             if not iter_rows:
157                 self.consume_next()  # raises StopIteration
158                 iter_rows, self._rows[:] = self._rows[:], ()
159             while iter_rows:
160                 yield iter_rows.pop(0)
161 
162     def one(self):
163         """Return exactly one result, or raise an exception.
164 
165         :raises: :exc:`NotFound`: If there are no results.
166         :raises: :exc:`ValueError`: If there are multiple results.
167         :raises: :exc:`RuntimeError`: If consumption has already occurred,
168             in whole or in part.
169         """
170         answer = self.one_or_none()
171         if answer is None:
172             raise exceptions.NotFound('No rows matched the given query.')
173         return answer
174 
175     def one_or_none(self):
176         """Return exactly one result, or None if there are no results.
177 
178         :raises: :exc:`ValueError`: If there are multiple results.
179         :raises: :exc:`RuntimeError`: If consumption has already occurred,
180             in whole or in part.
181         """
182         # Sanity check: Has consumption of this query already started?
183         # If it has, then this is an exception.
184         if self._metadata is not None:
185             raise RuntimeError('Can not call `.one` or `.one_or_none` after '
186                                'stream consumption has already started.')
187 
188         # Consume the first result of the stream.
189         # If there is no first result, then return None.
190         iterator = iter(self)
191         try:
192             answer = next(iterator)
193         except StopIteration:
194             return None
195 
196         # Attempt to consume more. This should no-op; if we get additional
197         # rows, then this is an error case.
198         try:
199             next(iterator)
200             raise ValueError('Expected one result; got more.')
201         except StopIteration:
202             return answer
203 
204 
205 class Unmergeable(ValueError):
206     """Unable to merge two values.
207 
208     :type lhs: :class:`google.protobuf.struct_pb2.Value`
209     :param lhs: pending value to be merged
210 
211     :type rhs: :class:`google.protobuf.struct_pb2.Value`
212     :param rhs: remaining value to be merged
213 
214     :type type_: :class:`google.cloud.spanner_v1.proto.type_pb2.Type`
215     :param type_: field type of values being merged
216     """
217     def __init__(self, lhs, rhs, type_):
218         message = "Cannot merge %s values: %s %s" % (
219             type_pb2.TypeCode.Name(type_.code), lhs, rhs)
220         super(Unmergeable, self).__init__(message)
221 
222 
223 def _unmergeable(lhs, rhs, type_):
224     """Helper for '_merge_by_type'."""
225     raise Unmergeable(lhs, rhs, type_)
226 
227 
228 def _merge_float64(lhs, rhs, type_):  # pylint: disable=unused-argument
229     """Helper for '_merge_by_type'."""
230     lhs_kind = lhs.WhichOneof('kind')
231     if lhs_kind == 'string_value':
232         return Value(string_value=lhs.string_value + rhs.string_value)
233     rhs_kind = rhs.WhichOneof('kind')
234     array_continuation = (
235         lhs_kind == 'number_value' and
236         rhs_kind == 'string_value' and
237         rhs.string_value == '')
238     if array_continuation:
239         return lhs
240     raise Unmergeable(lhs, rhs, type_)
241 
242 
243 def _merge_string(lhs, rhs, type_):  # pylint: disable=unused-argument
244     """Helper for '_merge_by_type'."""
245     return Value(string_value=lhs.string_value + rhs.string_value)
246 
247 
248 _UNMERGEABLE_TYPES = (type_pb2.BOOL,)
249 
250 
251 def _merge_array(lhs, rhs, type_):
252     """Helper for '_merge_by_type'."""
253     element_type = type_.array_element_type
254     if element_type.code in _UNMERGEABLE_TYPES:
255         # Individual values cannot be merged, just concatenate
256         lhs.list_value.values.extend(rhs.list_value.values)
257         return lhs
258     lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)
259     first = rhs.pop(0)
260     if first.HasField('null_value'):  # can't merge
261         lhs.append(first)
262     else:
263         last = lhs.pop()
264         try:
265             merged = _merge_by_type(last, first, element_type)
266         except Unmergeable:
267             lhs.append(last)
268             lhs.append(first)
269         else:
270             lhs.append(merged)
271     return Value(list_value=ListValue(values=(lhs + rhs)))
272 
273 
274 def _merge_struct(lhs, rhs, type_):
275     """Helper for '_merge_by_type'."""
276     fields = type_.struct_type.fields
277     lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)
278     candidate_type = fields[len(lhs) - 1].type
279     first = rhs.pop(0)
280     if (first.HasField('null_value') or
281             candidate_type.code in _UNMERGEABLE_TYPES):
282         lhs.append(first)
283     else:
284         last = lhs.pop()
285         lhs.append(_merge_by_type(last, first, candidate_type))
286     return Value(list_value=ListValue(values=lhs + rhs))
287 
288 
289 _MERGE_BY_TYPE = {
290     type_pb2.ARRAY: _merge_array,
291     type_pb2.BOOL: _unmergeable,
292     type_pb2.BYTES: _merge_string,
293     type_pb2.DATE: _merge_string,
294     type_pb2.FLOAT64: _merge_float64,
295     type_pb2.INT64: _merge_string,
296     type_pb2.STRING: _merge_string,
297     type_pb2.STRUCT: _merge_struct,
298     type_pb2.TIMESTAMP: _merge_string,
299 }
300 
301 
302 def _merge_by_type(lhs, rhs, type_):
303     """Helper for '_merge_chunk'."""
304     merger = _MERGE_BY_TYPE[type_.code]
305     return merger(lhs, rhs, type_)
306
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/spanner/google/cloud/spanner_v1/streamed.py b/spanner/google/cloud/spanner_v1/streamed.py
--- a/spanner/google/cloud/spanner_v1/streamed.py
+++ b/spanner/google/cloud/spanner_v1/streamed.py
@@ -256,6 +256,12 @@
         lhs.list_value.values.extend(rhs.list_value.values)
         return lhs
     lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)
+
+    # Sanity check: If either list is empty, short-circuit.
+    # This is effectively a no-op.
+    if not len(lhs) or not len(rhs):
+        return Value(list_value=ListValue(values=(lhs + rhs)))
+
     first = rhs.pop(0)
     if first.HasField('null_value'):  # can't merge
         lhs.append(first)

{"golden_diff": "diff --git a/spanner/google/cloud/spanner_v1/streamed.py b/spanner/google/cloud/spanner_v1/streamed.py\n--- a/spanner/google/cloud/spanner_v1/streamed.py\n+++ b/spanner/google/cloud/spanner_v1/streamed.py\n@@ -256,6 +256,12 @@\n         lhs.list_value.values.extend(rhs.list_value.values)\n         return lhs\n     lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)\n+\n+    # Sanity check: If either list is empty, short-circuit.\n+    # This is effectively a no-op.\n+    if not len(lhs) or not len(rhs):\n+        return Value(list_value=ListValue(values=(lhs + rhs)))\n+\n     first = rhs.pop(0)\n     if first.HasField('null_value'):  # can't merge\n         lhs.append(first)\n", "issue": "Spanner: IndexError: pop from empty list when merge_array\n```\r\nreturn merger(lhs, rhs, type_)\r\n  File \".../google/cloud/spanner_v1/streamed.py\", line 259, in _merge_array\r\n    first = rhs.pop(0)\r\nIndexError: pop from empty list\r\n```\n", "before_files": [{"content": "# Copyright 2016 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"Wrapper for streaming results.\"\"\"\n\nfrom google.protobuf.struct_pb2 import ListValue\nfrom google.protobuf.struct_pb2 import Value\nfrom google.cloud import exceptions\nfrom google.cloud.spanner_v1.proto import type_pb2\nimport six\n\n# pylint: disable=ungrouped-imports\nfrom google.cloud.spanner_v1._helpers import _parse_value_pb\n# pylint: enable=ungrouped-imports\n\n\nclass StreamedResultSet(object):\n    \"\"\"Process a sequence of partial result sets into a single set of row data.\n\n    :type response_iterator:\n    :param response_iterator:\n        Iterator yielding\n        :class:`google.cloud.spanner_v1.proto.result_set_pb2.PartialResultSet`\n        instances.\n\n    :type source: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`\n    :param source: Snapshot from which the result set was fetched.\n    \"\"\"\n    def __init__(self, response_iterator, source=None):\n        self._response_iterator = response_iterator\n        self._rows = []             # Fully-processed rows\n        self._counter = 0           # Counter for processed responses\n        self._metadata = None       # Until set from first PRS\n        self._stats = None          # Until set from last PRS\n        self._current_row = []      # Accumulated values for incomplete row\n        self._pending_chunk = None  # Incomplete value\n        self._source = source       # Source snapshot\n\n    @property\n    def rows(self):\n        \"\"\"Fully-processed rows.\n\n        :rtype: list of row-data lists.\n        :returns: list of completed row data, from proceesd PRS responses.\n        \"\"\"\n        return self._rows\n\n    @property\n    def fields(self):\n        \"\"\"Field descriptors for result set columns.\n\n        :rtype: list of :class:`~google.cloud.spanner_v1.proto.type_pb2.Field`\n        :returns: list of fields describing column names / types.\n        \"\"\"\n        return self._metadata.row_type.fields\n\n    @property\n    def metadata(self):\n        \"\"\"Result set metadata\n\n        :rtype: :class:`~.result_set_pb2.ResultSetMetadata`\n        :returns: structure describing the results\n        \"\"\"\n        return self._metadata\n\n    @property\n    def stats(self):\n        \"\"\"Result set statistics\n\n        :rtype:\n           :class:`~google.cloud.spanner_v1.proto.result_set_pb2.ResultSetStats`\n        :returns: structure describing status about the response\n        \"\"\"\n        return self._stats\n\n    def _merge_chunk(self, value):\n        \"\"\"Merge pending chunk with next value.\n\n        :type value: :class:`~google.protobuf.struct_pb2.Value`\n        :param value: continuation of chunked value from previous\n                      partial result set.\n\n        :rtype: :class:`~google.protobuf.struct_pb2.Value`\n        :returns: the merged value\n        \"\"\"\n        current_column = len(self._current_row)\n        field = self.fields[current_column]\n        merged = _merge_by_type(self._pending_chunk, value, field.type)\n        self._pending_chunk = None\n        return merged\n\n    def _merge_values(self, values):\n        \"\"\"Merge values into rows.\n\n        :type values: list of :class:`~google.protobuf.struct_pb2.Value`\n        :param values: non-chunked values from partial result set.\n        \"\"\"\n        width = len(self.fields)\n        for value in values:\n            index = len(self._current_row)\n            field = self.fields[index]\n            self._current_row.append(_parse_value_pb(value, field.type))\n            if len(self._current_row) == width:\n                self._rows.append(self._current_row)\n                self._current_row = []\n\n    def consume_next(self):\n        \"\"\"Consume the next partial result set from the stream.\n\n        Parse the result set into new/existing rows in :attr:`_rows`\n        \"\"\"\n        response = six.next(self._response_iterator)\n        self._counter += 1\n\n        if self._metadata is None:  # first response\n            metadata = self._metadata = response.metadata\n\n            source = self._source\n            if source is not None and source._transaction_id is None:\n                source._transaction_id = metadata.transaction.id\n\n        if response.HasField('stats'):  # last response\n            self._stats = response.stats\n\n        values = list(response.values)\n        if self._pending_chunk is not None:\n            values[0] = self._merge_chunk(values[0])\n\n        if response.chunked_value:\n            self._pending_chunk = values.pop()\n\n        self._merge_values(values)\n\n    def consume_all(self):\n        \"\"\"Consume the streamed responses until there are no more.\"\"\"\n        while True:\n            try:\n                self.consume_next()\n            except StopIteration:\n                break\n\n    def __iter__(self):\n        iter_rows, self._rows[:] = self._rows[:], ()\n        while True:\n            if not iter_rows:\n                self.consume_next()  # raises StopIteration\n                iter_rows, self._rows[:] = self._rows[:], ()\n            while iter_rows:\n                yield iter_rows.pop(0)\n\n    def one(self):\n        \"\"\"Return exactly one result, or raise an exception.\n\n        :raises: :exc:`NotFound`: If there are no results.\n        :raises: :exc:`ValueError`: If there are multiple results.\n        :raises: :exc:`RuntimeError`: If consumption has already occurred,\n            in whole or in part.\n        \"\"\"\n        answer = self.one_or_none()\n        if answer is None:\n            raise exceptions.NotFound('No rows matched the given query.')\n        return answer\n\n    def one_or_none(self):\n        \"\"\"Return exactly one result, or None if there are no results.\n\n        :raises: :exc:`ValueError`: If there are multiple results.\n        :raises: :exc:`RuntimeError`: If consumption has already occurred,\n            in whole or in part.\n        \"\"\"\n        # Sanity check: Has consumption of this query already started?\n        # If it has, then this is an exception.\n        if self._metadata is not None:\n            raise RuntimeError('Can not call `.one` or `.one_or_none` after '\n                               'stream consumption has already started.')\n\n        # Consume the first result of the stream.\n        # If there is no first result, then return None.\n        iterator = iter(self)\n        try:\n            answer = next(iterator)\n        except StopIteration:\n            return None\n\n        # Attempt to consume more. This should no-op; if we get additional\n        # rows, then this is an error case.\n        try:\n            next(iterator)\n            raise ValueError('Expected one result; got more.')\n        except StopIteration:\n            return answer\n\n\nclass Unmergeable(ValueError):\n    \"\"\"Unable to merge two values.\n\n    :type lhs: :class:`google.protobuf.struct_pb2.Value`\n    :param lhs: pending value to be merged\n\n    :type rhs: :class:`google.protobuf.struct_pb2.Value`\n    :param rhs: remaining value to be merged\n\n    :type type_: :class:`google.cloud.spanner_v1.proto.type_pb2.Type`\n    :param type_: field type of values being merged\n    \"\"\"\n    def __init__(self, lhs, rhs, type_):\n        message = \"Cannot merge %s values: %s %s\" % (\n            type_pb2.TypeCode.Name(type_.code), lhs, rhs)\n        super(Unmergeable, self).__init__(message)\n\n\ndef _unmergeable(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    raise Unmergeable(lhs, rhs, type_)\n\n\ndef _merge_float64(lhs, rhs, type_):  # pylint: disable=unused-argument\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    lhs_kind = lhs.WhichOneof('kind')\n    if lhs_kind == 'string_value':\n        return Value(string_value=lhs.string_value + rhs.string_value)\n    rhs_kind = rhs.WhichOneof('kind')\n    array_continuation = (\n        lhs_kind == 'number_value' and\n        rhs_kind == 'string_value' and\n        rhs.string_value == '')\n    if array_continuation:\n        return lhs\n    raise Unmergeable(lhs, rhs, type_)\n\n\ndef _merge_string(lhs, rhs, type_):  # pylint: disable=unused-argument\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    return Value(string_value=lhs.string_value + rhs.string_value)\n\n\n_UNMERGEABLE_TYPES = (type_pb2.BOOL,)\n\n\ndef _merge_array(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    element_type = type_.array_element_type\n    if element_type.code in _UNMERGEABLE_TYPES:\n        # Individual values cannot be merged, just concatenate\n        lhs.list_value.values.extend(rhs.list_value.values)\n        return lhs\n    lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)\n    first = rhs.pop(0)\n    if first.HasField('null_value'):  # can't merge\n        lhs.append(first)\n    else:\n        last = lhs.pop()\n        try:\n            merged = _merge_by_type(last, first, element_type)\n        except Unmergeable:\n            lhs.append(last)\n            lhs.append(first)\n        else:\n            lhs.append(merged)\n    return Value(list_value=ListValue(values=(lhs + rhs)))\n\n\ndef _merge_struct(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    fields = type_.struct_type.fields\n    lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)\n    candidate_type = fields[len(lhs) - 1].type\n    first = rhs.pop(0)\n    if (first.HasField('null_value') or\n            candidate_type.code in _UNMERGEABLE_TYPES):\n        lhs.append(first)\n    else:\n        last = lhs.pop()\n        lhs.append(_merge_by_type(last, first, candidate_type))\n    return Value(list_value=ListValue(values=lhs + rhs))\n\n\n_MERGE_BY_TYPE = {\n    type_pb2.ARRAY: _merge_array,\n    type_pb2.BOOL: _unmergeable,\n    type_pb2.BYTES: _merge_string,\n    type_pb2.DATE: _merge_string,\n    type_pb2.FLOAT64: _merge_float64,\n    type_pb2.INT64: _merge_string,\n    type_pb2.STRING: _merge_string,\n    type_pb2.STRUCT: _merge_struct,\n    type_pb2.TIMESTAMP: _merge_string,\n}\n\n\ndef _merge_by_type(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_chunk'.\"\"\"\n    merger = _MERGE_BY_TYPE[type_.code]\n    return merger(lhs, rhs, type_)\n", "path": "spanner/google/cloud/spanner_v1/streamed.py"}], "after_files": [{"content": "# Copyright 2016 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\n\"\"\"Wrapper for streaming results.\"\"\"\n\nfrom google.protobuf.struct_pb2 import ListValue\nfrom google.protobuf.struct_pb2 import Value\nfrom google.cloud import exceptions\nfrom google.cloud.spanner_v1.proto import type_pb2\nimport six\n\n# pylint: disable=ungrouped-imports\nfrom google.cloud.spanner_v1._helpers import _parse_value_pb\n# pylint: enable=ungrouped-imports\n\n\nclass StreamedResultSet(object):\n    \"\"\"Process a sequence of partial result sets into a single set of row data.\n\n    :type response_iterator:\n    :param response_iterator:\n        Iterator yielding\n        :class:`google.cloud.spanner_v1.proto.result_set_pb2.PartialResultSet`\n        instances.\n\n    :type source: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`\n    :param source: Snapshot from which the result set was fetched.\n    \"\"\"\n    def __init__(self, response_iterator, source=None):\n        self._response_iterator = response_iterator\n        self._rows = []             # Fully-processed rows\n        self._counter = 0           # Counter for processed responses\n        self._metadata = None       # Until set from first PRS\n        self._stats = None          # Until set from last PRS\n        self._current_row = []      # Accumulated values for incomplete row\n        self._pending_chunk = None  # Incomplete value\n        self._source = source       # Source snapshot\n\n    @property\n    def rows(self):\n        \"\"\"Fully-processed rows.\n\n        :rtype: list of row-data lists.\n        :returns: list of completed row data, from proceesd PRS responses.\n        \"\"\"\n        return self._rows\n\n    @property\n    def fields(self):\n        \"\"\"Field descriptors for result set columns.\n\n        :rtype: list of :class:`~google.cloud.spanner_v1.proto.type_pb2.Field`\n        :returns: list of fields describing column names / types.\n        \"\"\"\n        return self._metadata.row_type.fields\n\n    @property\n    def metadata(self):\n        \"\"\"Result set metadata\n\n        :rtype: :class:`~.result_set_pb2.ResultSetMetadata`\n        :returns: structure describing the results\n        \"\"\"\n        return self._metadata\n\n    @property\n    def stats(self):\n        \"\"\"Result set statistics\n\n        :rtype:\n           :class:`~google.cloud.spanner_v1.proto.result_set_pb2.ResultSetStats`\n        :returns: structure describing status about the response\n        \"\"\"\n        return self._stats\n\n    def _merge_chunk(self, value):\n        \"\"\"Merge pending chunk with next value.\n\n        :type value: :class:`~google.protobuf.struct_pb2.Value`\n        :param value: continuation of chunked value from previous\n                      partial result set.\n\n        :rtype: :class:`~google.protobuf.struct_pb2.Value`\n        :returns: the merged value\n        \"\"\"\n        current_column = len(self._current_row)\n        field = self.fields[current_column]\n        merged = _merge_by_type(self._pending_chunk, value, field.type)\n        self._pending_chunk = None\n        return merged\n\n    def _merge_values(self, values):\n        \"\"\"Merge values into rows.\n\n        :type values: list of :class:`~google.protobuf.struct_pb2.Value`\n        :param values: non-chunked values from partial result set.\n        \"\"\"\n        width = len(self.fields)\n        for value in values:\n            index = len(self._current_row)\n            field = self.fields[index]\n            self._current_row.append(_parse_value_pb(value, field.type))\n            if len(self._current_row) == width:\n                self._rows.append(self._current_row)\n                self._current_row = []\n\n    def consume_next(self):\n        \"\"\"Consume the next partial result set from the stream.\n\n        Parse the result set into new/existing rows in :attr:`_rows`\n        \"\"\"\n        response = six.next(self._response_iterator)\n        self._counter += 1\n\n        if self._metadata is None:  # first response\n            metadata = self._metadata = response.metadata\n\n            source = self._source\n            if source is not None and source._transaction_id is None:\n                source._transaction_id = metadata.transaction.id\n\n        if response.HasField('stats'):  # last response\n            self._stats = response.stats\n\n        values = list(response.values)\n        if self._pending_chunk is not None:\n            values[0] = self._merge_chunk(values[0])\n\n        if response.chunked_value:\n            self._pending_chunk = values.pop()\n\n        self._merge_values(values)\n\n    def consume_all(self):\n        \"\"\"Consume the streamed responses until there are no more.\"\"\"\n        while True:\n            try:\n                self.consume_next()\n            except StopIteration:\n                break\n\n    def __iter__(self):\n        iter_rows, self._rows[:] = self._rows[:], ()\n        while True:\n            if not iter_rows:\n                self.consume_next()  # raises StopIteration\n                iter_rows, self._rows[:] = self._rows[:], ()\n            while iter_rows:\n                yield iter_rows.pop(0)\n\n    def one(self):\n        \"\"\"Return exactly one result, or raise an exception.\n\n        :raises: :exc:`NotFound`: If there are no results.\n        :raises: :exc:`ValueError`: If there are multiple results.\n        :raises: :exc:`RuntimeError`: If consumption has already occurred,\n            in whole or in part.\n        \"\"\"\n        answer = self.one_or_none()\n        if answer is None:\n            raise exceptions.NotFound('No rows matched the given query.')\n        return answer\n\n    def one_or_none(self):\n        \"\"\"Return exactly one result, or None if there are no results.\n\n        :raises: :exc:`ValueError`: If there are multiple results.\n        :raises: :exc:`RuntimeError`: If consumption has already occurred,\n            in whole or in part.\n        \"\"\"\n        # Sanity check: Has consumption of this query already started?\n        # If it has, then this is an exception.\n        if self._metadata is not None:\n            raise RuntimeError('Can not call `.one` or `.one_or_none` after '\n                               'stream consumption has already started.')\n\n        # Consume the first result of the stream.\n        # If there is no first result, then return None.\n        iterator = iter(self)\n        try:\n            answer = next(iterator)\n        except StopIteration:\n            return None\n\n        # Attempt to consume more. This should no-op; if we get additional\n        # rows, then this is an error case.\n        try:\n            next(iterator)\n            raise ValueError('Expected one result; got more.')\n        except StopIteration:\n            return answer\n\n\nclass Unmergeable(ValueError):\n    \"\"\"Unable to merge two values.\n\n    :type lhs: :class:`google.protobuf.struct_pb2.Value`\n    :param lhs: pending value to be merged\n\n    :type rhs: :class:`google.protobuf.struct_pb2.Value`\n    :param rhs: remaining value to be merged\n\n    :type type_: :class:`google.cloud.spanner_v1.proto.type_pb2.Type`\n    :param type_: field type of values being merged\n    \"\"\"\n    def __init__(self, lhs, rhs, type_):\n        message = \"Cannot merge %s values: %s %s\" % (\n            type_pb2.TypeCode.Name(type_.code), lhs, rhs)\n        super(Unmergeable, self).__init__(message)\n\n\ndef _unmergeable(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    raise Unmergeable(lhs, rhs, type_)\n\n\ndef _merge_float64(lhs, rhs, type_):  # pylint: disable=unused-argument\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    lhs_kind = lhs.WhichOneof('kind')\n    if lhs_kind == 'string_value':\n        return Value(string_value=lhs.string_value + rhs.string_value)\n    rhs_kind = rhs.WhichOneof('kind')\n    array_continuation = (\n        lhs_kind == 'number_value' and\n        rhs_kind == 'string_value' and\n        rhs.string_value == '')\n    if array_continuation:\n        return lhs\n    raise Unmergeable(lhs, rhs, type_)\n\n\ndef _merge_string(lhs, rhs, type_):  # pylint: disable=unused-argument\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    return Value(string_value=lhs.string_value + rhs.string_value)\n\n\n_UNMERGEABLE_TYPES = (type_pb2.BOOL,)\n\n\ndef _merge_array(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    element_type = type_.array_element_type\n    if element_type.code in _UNMERGEABLE_TYPES:\n        # Individual values cannot be merged, just concatenate\n        lhs.list_value.values.extend(rhs.list_value.values)\n        return lhs\n    lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)\n\n    # Sanity check: If either list is empty, short-circuit.\n    # This is effectively a no-op.\n    if not len(lhs) or not len(rhs):\n        return Value(list_value=ListValue(values=(lhs + rhs)))\n\n    first = rhs.pop(0)\n    if first.HasField('null_value'):  # can't merge\n        lhs.append(first)\n    else:\n        last = lhs.pop()\n        try:\n            merged = _merge_by_type(last, first, element_type)\n        except Unmergeable:\n            lhs.append(last)\n            lhs.append(first)\n        else:\n            lhs.append(merged)\n    return Value(list_value=ListValue(values=(lhs + rhs)))\n\n\ndef _merge_struct(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_by_type'.\"\"\"\n    fields = type_.struct_type.fields\n    lhs, rhs = list(lhs.list_value.values), list(rhs.list_value.values)\n    candidate_type = fields[len(lhs) - 1].type\n    first = rhs.pop(0)\n    if (first.HasField('null_value') or\n            candidate_type.code in _UNMERGEABLE_TYPES):\n        lhs.append(first)\n    else:\n        last = lhs.pop()\n        lhs.append(_merge_by_type(last, first, candidate_type))\n    return Value(list_value=ListValue(values=lhs + rhs))\n\n\n_MERGE_BY_TYPE = {\n    type_pb2.ARRAY: _merge_array,\n    type_pb2.BOOL: _unmergeable,\n    type_pb2.BYTES: _merge_string,\n    type_pb2.DATE: _merge_string,\n    type_pb2.FLOAT64: _merge_float64,\n    type_pb2.INT64: _merge_string,\n    type_pb2.STRING: _merge_string,\n    type_pb2.STRUCT: _merge_struct,\n    type_pb2.TIMESTAMP: _merge_string,\n}\n\n\ndef _merge_by_type(lhs, rhs, type_):\n    \"\"\"Helper for '_merge_chunk'.\"\"\"\n    merger = _MERGE_BY_TYPE[type_.code]\n    return merger(lhs, rhs, type_)\n", "path": "spanner/google/cloud/spanner_v1/streamed.py"}]}