fix: Add paging to hbase client

[ron-gal]

Thank you for opening a Pull Request! Before submitting your PR, there are a few things you can do to make sure it goes smoothly:

• Make sure to open an issue as a bug/issue before writing your code! That way we can discuss the change, evaluate designs, and agree on the general idea
• Ensure the tests and linter pass
• Code coverage does not decrease (if any source code was changed)
• Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> ☕️

If you write sample code, please follow the samples format.

[igorbernstein2]

Why do you defer the creation of the ScanCallContext? The effect of re-creating it is that you are extending the operation timeout across all of the pages. I think the operation deadline should stay consistent from the callers perspective regardless if its broken up into multiple segments

[ron-gal]

done

[igorbernstein2]

please revert irrelevant changes to minimize the noise

[ron-gal]

done

[igorbernstein2]

should this be in the if statement?

[ron-gal]

Done

[igorbernstein2]

why static?

[ron-gal]

copy pasted from beam, removed

[mutianf]

Maybe we can pass in the default to begin with?

[ron-gal]

It's private, and I wouldn't want to expose it. This value is just for tests, so I don't think it's critical.

[mutianf]

We could move the default value in this class?

[ron-gal]

Done

[mutianf]

I'm a bit confused at this test. Why would it be called twice when the page size is 100 and you're only returning 2 results?

[ron-gal]

I am calling readRows twice. Once with results, and once without

[mutianf]

Ok, I see it now. However I don't think this actually tested anything. I think we can set a smaller page size (for example, 2), return more rows (say 3), and verify that readRowsCallable are called 3 times.

[ron-gal]

Done

[mutianf]

why do we return an empty result first?

[ron-gal]

That's similar to the test for the non paginated scanner

[igorbernstein2]

This test doesnt really make sense. You want to be testing what happens between pages. Something like this:

private static Result createRow(String key) {
    return Result.create(
        ImmutableList.<Cell>of(
            new com.google.cloud.bigtable.hbase.adapters.read.RowCell(
                Bytes.toBytes(key),
                Bytes.toBytes("cf"),
                Bytes.toBytes("q"),
                10L,
                Bytes.toBytes("value"),
                ImmutableList.of("label"))));
  }
  @Test
  public void testReadPaginatedRows() throws IOException {
    Query query = Query.create(TABLE_ID).range("a", "z");
    when(mockDataClient.readRowsCallable(Mockito.<RowResultAdapter>any()))
        .thenReturn(mockStreamingCallable);

    // First Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onResponse(createRow("a"));
      observer.onResponse(createRow("b"));
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(eq(Query.create(TABLE_ID).range("a", "z").limit(2)), any(), any());

    // 2nd Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onResponse(createRow("c"));
      observer.onResponse(createRow("d"));
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(
            eq(
              Query.create(TABLE_ID)
                .range(ByteStringRange.unbounded().startOpen("b").endOpen("z")).limit(2)),
            any(), any());

    // 3rd Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onResponse(createRow("e"));
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(
        eq(
          Query.create(TABLE_ID)
            .range(ByteStringRange.unbounded().startOpen("d").endOpen("z")).limit(2)), any(), any());

    // 3rd Page
    doAnswer((args) -> {
      ResponseObserver<Result> observer = args.getArgument(1);
      observer.onComplete();
      return null;
    })
        .when(mockStreamingCallable)
        .call(
            eq(
                Query.create(TABLE_ID)
                    .range(ByteStringRange.unbounded().startOpen("e").endOpen("z")).limit(2)), any(), any());

    ResultScanner resultScanner = dataClientWrapper.readRows(query.createPaginator(2), 1000);

    assertThat(resultScanner)
        .comparingElementsUsing(Correspondence.transforming((Result r) -> new String(r.getRow()), "row key"))
        .containsExactly("a", "b", "c", "d", "e");
  }

[ron-gal]

Done

[mutianf]

same question here, why do we return an empty result here? I don't think that's gonna use any memory? Maybe you want to return another expected result and assert readRowsCallable is called twice.

[ron-gal]

Again, just copying the old test. No real reason, but I don't think it's not valid

[mutianf]

Similar to above, I don't think this actually tested the memory buffer function. We should at least return another result, and verify that readRowsCallable are called twice (first time the request got cancelled becuase we filled up the buffer, and the second time we finish the read).

[mutianf]

what's the difference between line 353 - 372 and line 331 - 351? Seem to be duplicates?

[ron-gal]

Fixed

[mutianf]

Was this a new test that's added to test the paginator?

[ron-gal]

Yes. Adjusted to new paginator impl

[igorbernstein2]

This test isnt really testing anything, I would just drop it

[ron-gal]

Done

[mutianf]

Sorry I should be more clear on my previous comment. I think we want to test both getScanner without pagination and getScanner with pagiation. So maybe we can add a parameter:

testManyResultsInScanner(int rowsToWrite, boolean with pagination);

And add one more test: testManyResultsInScanner(100, false);

[ron-gal]

Done

[mutianf]

We could move the default value in this class?

[mutianf]

Ok, I see it now. However I don't think this actually tested anything. I think we can set a smaller page size (for example, 2), return more rows (say 3), and verify that readRowsCallable are called 3 times.

[mutianf]

Similar to above, I don't think this actually tested the memory buffer function. We should at least return another result, and verify that readRowsCallable are called twice (first time the request got cancelled becuase we filled up the buffer, and the second time we finish the read).

[igorbernstein2]

this doesnt make sense. readRowsCallable() will never throw an exception. You want to test what happens when the server returns an error. I think something like this:

  public void testReadRows_Errors() throws IOException {
    Query query = Query.create(TABLE_ID).rowKey(ROW_KEY);
    when(mockDataClient.readRowsCallable(any(RowResultAdapter.class)))
        .thenReturn(mockStreamingCallable);
    when(mockStreamingCallable.call(any(Query.class), any(GrpcCallContext.class)))
        .thenReturn(serverStream);
    when(serverStream.iterator())
        .thenReturn(new Iterator<Result>() {
          @Override
          public boolean hasNext() {
            return true;
          }
          @Override
          public Result next() {
            throw new InternalException("fake error", null, GrpcStatusCode.of(Code.INTERNAL), false);
          }
        })
        .thenReturn(ImmutableList.<Result>of().iterator());

    assertThrows(Exception.class, () -> dataClientWrapper.readRows(query).next());

    ResultScanner noRowsResultScanner = dataClientWrapper.readRows(query);
    assertNull(noRowsResultScanner.next());
    noRowsResultScanner.close();

    verify(mockDataClient, times(2)).readRowsCallable(Mockito.<RowResultAdapter>any());
    verify(serverStream, times(2)).iterator();
    verify(mockStreamingCallable, times(2))
        .call(any(Query.class), any(GrpcCallContext.class));
  }

[ron-gal]

Done

[igorbernstein2]

Similar to above, you want to be testing the rpc between pages. so something like this:

 public void testReadRowsLowMemory() throws IOException {
    Query query = Query.create(TABLE_ID);
    when(mockDataClient.readRowsCallable(any(RowResultAdapter.class)))
        .thenReturn(mockStreamingCallable);

    StreamController mockController = Mockito.mock(StreamController.class);
    doAnswer(invocation -> {
          cancelled.set(true);
          return null;
        })
        .when(mockController)
        .cancel();


    // Generate
    doAnswer(
        (Answer<Void>) invocation -> {
          ResponseObserver<Result> observer = invocation.getArgument(1);
          observer.onStart(mockController);

          for(int i=0; i < 1000 && !cancelled.get(); i++) {
            observer.onResponse(createRow(String.format("row%010d", i)));
            Thread.sleep(10);
          }
          observer.onComplete();
          return null;
        })
        .doAnswer(
        (Answer<Void>) invocation -> {
          ResponseObserver<Result> observer = invocation.getArgument(1);
          observer.onComplete();
          return null;
        })
        .when(mockStreamingCallable)
        .call(any(), any(), any());

    ResultScanner resultScanner = dataClientWrapper.readRows(query.createPaginator(100), 3);
    // Consume the stream
    Lists.newArrayList(resultScanner);

    verify(mockStreamingCallable, times(2))
        .call(
            any(Query.class),
            any(ResponseObserver.class),
            any(GrpcCallContext.class));
    assertTrue(cancelled.get());
  }

[ron-gal]

Done