Repository: GoogleCloudPlatform/cloud-bigtable-client
Pull request #1940 (2018-09-13 16:26:10)

Ensuring that rows are sequential in RowMerger


Patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
index 0f84e2cc7..d77e69068 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManager.java
@@ -91,6 +91,15 @@ class ReadRowsRequestManager {
     }
     RowSet.Builder rowSetBuilder = RowSet.newBuilder();
 
+    if (originalRows.getRowKeysCount() == 0 && originalRows.getRowRangesCount() == 0) {
+      rowSetBuilder.addRowRanges(
+          RowRange
+              .newBuilder()
+              .setStartKeyOpen(lastFoundKey)
+              .setEndKeyOpen(ByteString.EMPTY)
+              .build());
+    }
+
     for (ByteString key : originalRows.getRowKeysList()) {
       if (!startKeyIsAlreadyRead(key)) {
         rowSetBuilder.addRowKeys(key);
@@ -107,16 +116,16 @@ class ReadRowsRequestManager {
         continue;
       }
 
-      RowRange newRange = rowRange;
+      RowRange newRowRange = rowRange;
       StartKeyCase startKeyCase = rowRange.getStartKeyCase();
       if ((startKeyCase == StartKeyCase.START_KEY_CLOSED
           && startKeyIsAlreadyRead(rowRange.getStartKeyClosed()))
           || (startKeyCase == StartKeyCase.START_KEY_OPEN
           && startKeyIsAlreadyRead(rowRange.getStartKeyOpen()))
           || startKeyCase == StartKeyCase.STARTKEY_NOT_SET) {
-        newRange = rowRange.toBuilder().setStartKeyOpen(lastFoundKey).build();
+        newRowRange = newRowRange.toBuilder().setStartKeyOpen(lastFoundKey).build();
       }
-      rowSetBuilder.addRowRanges(newRange);
+      rowSetBuilder.addRowRanges(newRowRange);
     }
 
     return rowSetBuilder.build();
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
index adcb6e0ec..afff1c392 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/grpc/scanner/RowMerger.java
@@ -22,6 +22,7 @@ import java.util.Map;
 import java.util.TreeMap;
 
 import com.google.cloud.bigtable.config.Logger;
+import com.google.cloud.bigtable.util.ByteStringComparator;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Objects;
 import com.google.bigtable.v2.ReadRowsRequest;
@@ -121,14 +122,16 @@ protected static final Logger LOG = new Logger(RowMerger.class);
         final ByteString rowKey = newChunk.getRowKey();
         Preconditions.checkArgument(!rowKey.isEmpty(), "A row key must be set: %s",
           newChunk);
-        Preconditions.checkState(previousKey == null || !rowKey.equals(previousKey),
-          "A commit happened but the same key followed: %s", newChunk);
+        if (previousKey != null &&
+            ByteStringComparator.INSTANCE.compare(previousKey, rowKey) >= 0) {
+          throw new IllegalArgumentException(String
+              .format("Found key '%s' after key '%s'", rowKey.toStringUtf8(),
+                  previousKey.toStringUtf8()));
+        }
         Preconditions.checkArgument(newChunk.hasQualifier(), "A column qualifier must be set: %s",
           newChunk);
-        if (newChunk.getValueSize() > 0) {
-          Preconditions.checkArgument(!newChunk.getCommitRow(),
-            "A row cannot be have a value size and be a commit row: %s", newChunk);
-        }
+        Preconditions.checkArgument(!newChunk.getCommitRow() || newChunk.getValueSize() == 0,
+          "A row cannot be have a value size and be a commit row: %s", newChunk);
       }
 
       @Override


Test patch:
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java
index 65d875084..0984baa20 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/main/java/com/google/cloud/bigtable/util/ByteStringComparator.java
@@ -28,9 +28,21 @@ public final class ByteStringComparator implements Comparator<ByteString>{
 
   @Override
   public int compare(ByteString key1, ByteString key2) {
-    int size1 = key1.size();
-    int size2 = key2.size();
-    int size = Math.min(size1, size2);
+    if (key1 == null) {
+      if (key2 == null) {
+        return 0;
+      } else {
+        return 1;
+      }
+    } else if (key2 == null) {
+      return -1;
+    }
+
+    if (key1 == key2) {
+      return 0;
+    }
+
+    int size = Math.min(key1.size(), key2.size());
 
     for (int i = 0; i < size; i++) {
       // compare bytes as unsigned
@@ -42,6 +54,6 @@ public final class ByteStringComparator implements Comparator<ByteString>{
         return comparison;
       }
     }
-    return Integer.compare(size1, size2);
+    return Integer.compare(key1.size(), key2.size());
   }
 }
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java
index b52faddd1..c28506069 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/ReadRowsRequestManagerTest.java
@@ -36,12 +36,6 @@ public class ReadRowsRequestManagerTest {
 
   static final ByteString BLANK = ByteString.EMPTY;
 
-  static FlatRow buildRow(String rowKey) {
-    return FlatRow.newBuilder()
-        .withRowKey(ByteString.copyFromUtf8(rowKey))
-        .build();
-  }
-
   private static ReadRowsRequest createRequest(RowRange range) {
     return ReadRowsRequest.newBuilder().setRows(RowSet.newBuilder().addRowRanges(range)).build();
   }
@@ -62,14 +56,13 @@ public class ReadRowsRequestManagerTest {
     return RowSet.newBuilder().addAllRowKeys(keys).build();
   }
 
-
   /**
-   * Test a single, full table scan scenario for {@Link ResumingStreamingResultScanner#filterRows()}
+   * Test a single, full table scan scenario for {@link ReadRowsRequestManager#buildUpdatedRequest()}
    * .
    * @throws IOException
    */
   @Test
-  public void test_filterRows_testAllRange() throws IOException{
+  public void test_filterRows_testAllRange() {
     ByteString key1 = ByteString.copyFrom("row1".getBytes());
 
     ReadRowsRequest originalRequest =
@@ -83,11 +76,33 @@ public class ReadRowsRequestManagerTest {
   }
 
   /**
-   * Test rowKeys scenario for {@Link ReadRowsRequestManager#getUpdatedRequest()}.
+   * Test a single, full table scan scenario for {@link ReadRowsRequestManager#buildUpdatedRequest()}}
+   * .
+   */
+   @Test
+  public void test_filterRows_empty() {
+     ByteString key1 = ByteString.copyFrom("row1".getBytes());
+
+     ReadRowsRequest originalRequest = createRequest(RowRange.getDefaultInstance());
+
+     ReadRowsRequestManager underTest = new ReadRowsRequestManager(originalRequest);
+
+     underTest.updateLastFoundKey(key1);
+     RowSet actual = underTest.buildUpdatedRequest().getRows();
+     Assert.assertEquals(0, actual.getRowKeysCount());
+     Assert.assertEquals(1, actual.getRowRangesCount());
+     RowRange actualRange = actual.getRowRanges(0);
+     Assert.assertEquals(key1, actualRange.getStartKeyOpen());
+     RowRange.EndKeyCase endKeyCase = actualRange.getEndKeyCase();
+     Assert.assertTrue(endKeyCase == RowRange.EndKeyCase.ENDKEY_NOT_SET);
+   }
+
+  /**
+   * Test rowKeys scenario for {@link ReadRowsRequestManager#buildUpdatedRequest()}}.
    * @throws IOException
    */
   @Test
-  public void test_filterRows_rowKeys() throws IOException{
+  public void test_filterRows_rowKeys() {
     ByteString key1 = ByteString.copyFrom("row1".getBytes());
     ByteString key2 = ByteString.copyFrom("row2".getBytes());
     ByteString key3 = ByteString.copyFrom("row3".getBytes());
@@ -104,11 +119,11 @@ public class ReadRowsRequestManagerTest {
   }
 
   /**
-   * Test multiple rowset filter scenarios for {@Link ReadRowsRequestManager#getUpdatedRequest()}.
+   * Test multiple rowset filter scenarios for {@link ReadRowsRequestManager#buildUpdatedRequest()}}.
    * @throws IOException
    */
   @Test
-  public void test_filterRows_multiRowSetFilters() throws IOException{
+  public void test_filterRows_multiRowSetFilters() {
     ByteString key1 = ByteString.copyFrom("row1".getBytes());
     ByteString key2 = ByteString.copyFrom("row2".getBytes());
     ByteString key3 = ByteString.copyFrom("row3".getBytes());
diff --git a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java
index 8bee5ffe6..9be8bc649 100644
--- a/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java
+++ b/bigtable-client-core-parent/bigtable-client-core/src/test/java/com/google/cloud/bigtable/grpc/scanner/RowMergerTest.java
@@ -16,6 +16,7 @@
 package com.google.cloud.bigtable.grpc.scanner;
 
 import static org.mockito.Matchers.eq;
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
@@ -23,7 +24,6 @@ import java.util.Arrays;
 
 import com.google.bigtable.v2.ReadRowsResponse;
 import com.google.bigtable.v2.ReadRowsResponse.CellChunk;
-import com.google.cloud.bigtable.grpc.scanner.RowMerger;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.BytesValue;
 import com.google.protobuf.StringValue;
@@ -49,7 +49,7 @@ public class RowMergerTest {
   }
 
   @Test
-  public void testOneCellRow() {
+  public void testOneCellRow_success() {
     CellChunk cellChunk1 = createCell("row_key1", "family", "qualifier", "value", 1, true);
     CellChunk cellChunk2 = createCell("row_key2", "family", "qualifier", "value", 1, true);
     RowMerger underTest = new RowMerger(observer);
@@ -57,6 +57,17 @@ public class RowMergerTest {
     underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk2).build());
     verify(observer, times(1)).onNext(eq(toRow(cellChunk1)));
     verify(observer, times(1)).onNext(eq(toRow(cellChunk2)));
+    verify(observer, times(0)).onError(any(Exception.class));
+  }
+
+  @Test
+  public void testOneCellRow_reverse_key() {
+    CellChunk cellChunk1 = createCell("row_key2", "family", "qualifier", "value", 1, true);
+    CellChunk cellChunk2 = createCell("row_key1", "family", "qualifier", "value", 1, true);
+    RowMerger underTest = new RowMerger(observer);
+    underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk1).build());
+    underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk2).build());
+    verify(observer, times(1)).onError(any(IllegalArgumentException.class));
   }
 
   @Test
diff --git a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java
index fdaf71041..99eb69bed 100644
--- a/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java
+++ b/bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/RowMergerPerf.java
@@ -56,7 +56,8 @@ public class RowMergerPerf {
     final int qualifiersPerFamily = 15;
     Preconditions.checkArgument(size > 0, "size has to be > 0.");
     final ByteString rowKey = ByteString.copyFrom(Bytes.toBytes("rowkey-0"));
-    for (int i = 0; i < cellCount; i++) {
+    int quals = Math.max(1, cellCount - 1);
+    for (int i = 0; i < quals; i++) {
       CellChunk.Builder contentChunk =
           CellChunk.newBuilder()
               .setRowKey(i == 0 ? rowKey : ByteString.EMPTY)
@@ -64,13 +65,23 @@ public class RowMergerPerf {
                   .setValue(ByteString.copyFromUtf8("Qualifier" + (i % qualifiersPerFamily))))
               .setValue(ByteString.copyFrom(RandomStringUtils.randomAlphanumeric(size).getBytes()))
               .setTimestampMicros(330020L)
-              .setCommitRow(i == cellCount - 1);
+              .setCommitRow(i == quals - 1);
       if (i % qualifiersPerFamily == 0) {
         contentChunk.setFamilyName(StringValue.newBuilder().setValue("Family" + (i / qualifiersPerFamily)));
       }
 
       readRowsResponse.addChunks(contentChunk);
     }
+
+    // This adds a new row to test performance of comparing two different row keys.
+    readRowsResponse.addChunks(CellChunk.newBuilder()
+        .setRowKey(ByteString.copyFromUtf8("rowkey-1"))
+        .setQualifier(BytesValue.newBuilder()
+            .setValue(ByteString.copyFromUtf8("Qualifier0")))
+        .setFamilyName(StringValue.newBuilder().setValue("Family1").build())
+        .setValue(ByteString.copyFrom(RandomStringUtils.randomAlphanumeric(size).getBytes()))
+        .setTimestampMicros(330020L)
+        .setCommitRow(true));
     return readRowsResponse.build();
   }