fix(spanner): be permissive about merging an empty struct

spanner/google/cloud/spanner_v1/streamed.py

@@ -276,6 +276,12 @@ def _merge_struct(lhs, rhs, type_):
     """Helper for '_merge_by_type'."""
     fields = type_.struct_type.fields
     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)))
+
     candidate_type = fields[len(lhs) - 1].type
     first = rhs.pop(0)
     if first.HasField("null_value") or candidate_type.code in _UNMERGEABLE_TYPES:

spanner/tests/unit/test_streamed.py

@@ -411,6 +411,23 @@ def test__merge_chunk_array_of_struct(self):
         self.assertEqual(merged, expected)
         self.assertIsNone(streamed._pending_chunk)
 
+    def test__merge_chunk_array_of_struct_with_empty(self):
+        iterator = _MockCancellableIterator()
+        streamed = self._make_one(iterator)
+        struct_type = self._make_struct_type([("name", "STRING"), ("age", "INT64")])
+        FIELDS = [self._make_array_field("test", element_type=struct_type)]
+        streamed._metadata = self._make_result_set_metadata(FIELDS)
+        partial = self._make_list_value([u"Phred "])
+        streamed._pending_chunk = self._make_list_value(value_pbs=[partial])
+        rest = self._make_list_value([])
+        chunk = self._make_list_value(value_pbs=[rest])
+
+        merged = streamed._merge_chunk(chunk)
+
+        expected = self._make_list_value(value_pbs=[partial])
+        self.assertEqual(merged, expected)
+        self.assertIsNone(streamed._pending_chunk)
+
     def test__merge_chunk_array_of_struct_unmergeable(self):
         iterator = _MockCancellableIterator()
         streamed = self._make_one(iterator)