fix: use skipped_results from query results to adjust offset (#399)

Fixes a bug where we blithely assumed that if we sent
Datastore/Firestore a query with an offset, that the first batch returned
would skip the entire offset. In practice, for high offsets, it's possible
for Datastore/Firestore to return a results batch that is empty and
which has `skipped_results` set to some number less than the value of
`offset` that we sent it. In this case, we still need to send a value
for `offset` when retreiving the next batch. This patch uses
`skipped_results` to compute a new `offset` for the follow up batch.

Fixes #392

Author: Chris Rossi

packages/google-cloud-ndb/google/cloud/ndb/_datastore_query.py

@@ -316,8 +316,15 @@ def _next_batch(self):
             limit = self._query.limit
             if limit is not None:
                 limit -= len(self._batch)
+
+            offset = self._query.offset
+            if offset:
+                offset -= response.batch.skipped_results
+
             self._query = self._query.copy(
-                start_cursor=Cursor(batch.end_cursor), offset=None, limit=limit
+                start_cursor=Cursor(batch.end_cursor),
+                offset=offset,
+                limit=limit,
             )
 
     def next(self):

packages/google-cloud-ndb/tests/system/test_query.py

@@ -1752,3 +1752,26 @@ class SomeKind(ndb.Model):
     assert results[0].foo == ndb.key.Key(
         "test_key", 3, project=project, namespace=namespace
     )
+
+
+@pytest.mark.usefixtures("client_context")
+def test_high_offset(dispose_of):
+    """Regression test for Issue #392
+
+    https://github.com/googleapis/python-ndb/issues/392
+    """
+    n_entities = 1100
+
+    class SomeKind(ndb.Model):
+        foo = ndb.IntegerProperty()
+
+    entities = [SomeKind(id=i + 1, foo=i) for i in range(n_entities)]
+    keys = ndb.put_multi(entities)
+    for key in keys:
+        dispose_of(key._key)
+
+    eventually(SomeKind.query().fetch, _length_equals(n_entities))
+    query = SomeKind.query(order_by=[SomeKind.foo])
+    index = n_entities - 5
+    result = query.fetch(offset=index, limit=1)[0]
+    assert result.foo == index

packages/google-cloud-ndb/tests/unit/test__datastore_query.py

@@ -393,6 +393,7 @@ def test__next_batch_has_more_w_offset_and_limit(_datastore_run_query):
                     entity_result_type=query_pb2.EntityResult.FULL,
                     entity_results=entity_results,
                     end_cursor=b"abc",
+                    skipped_results=5,
                     more_results=query_pb2.QueryResultBatch.NOT_FINISHED,
                 )
             )
@@ -408,7 +409,7 @@ def test__next_batch_has_more_w_offset_and_limit(_datastore_run_query):
         assert iterator._batch[0].order_by is None
         assert iterator._has_next_batch
         assert iterator._query.start_cursor.cursor == b"abc"
-        assert iterator._query.offset is None
+        assert iterator._query.offset == 0
         assert iterator._query.limit == 2
 
     @staticmethod