fix: Clean up RowResultAdapter (#3267)

* fix: Clean up RowResultAdapter

* use rawCells() and update javadoc

* update getKey

Author: mutianf

bigtable-client-core-parent/bigtable-hbase/src/main/java/com/google/cloud/bigtable/hbase/wrappers/veneer/RowResultAdapter.java

@@ -22,7 +22,6 @@
 import com.google.cloud.bigtable.hbase.adapters.read.RowCell;
 import com.google.cloud.bigtable.hbase.util.ByteStringer;
 import com.google.cloud.bigtable.hbase.util.TimestampConverter;
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
@@ -47,59 +46,44 @@
  * on duplicate labelled cells for its implementation. So this adapter will not deduplicate labelled
  * cells.
  *
+ * <p>This adapter will also return and check for scan marker rows, which will be an empty row with
+ * the scan marker row label.
+ *
  * <p>For internal use only - public for technical reasons.
  */
 @InternalApi("For internal usage only")
 public class RowResultAdapter implements RowAdapter<Result> {
 
   private static final byte[] EMPTY_VALUE = new byte[0];
+  private static final String SCAN_MARKER_ROW_LABEL = "bigtable-scan-marker-row";
 
   @Override
   public RowBuilder<Result> createRowBuilder() {
     return new RowResultBuilder();
   }
 
+  /**
+   * Checks if the result is a scan marker row. Returns true if the row's family, qualifier, and
+   * value are empty, and only has a scan marker row label.
+   */
   @Override
   public boolean isScanMarkerRow(Result result) {
-    Preconditions.checkState(result instanceof RowResult, "Should be an instance of RowResult");
-    return ((RowResult) result).isMarkerRow();
+    if (result.rawCells().length != 1 || !(result.rawCells()[0] instanceof RowCell)) {
+      return false;
+    }
+
+    RowCell cell = (RowCell) result.rawCells()[0];
+    return cell.getLabels().size() == 1
+        && cell.getLabels().get(0).equals(SCAN_MARKER_ROW_LABEL)
+        && cell.getValueArray().length == 0
+        && cell.getFamilyArray().length == 0
+        && cell.getQualifierArray().length == 0;
   }
 
   @Override
   public ByteString getKey(Result result) {
-    Preconditions.checkState(result instanceof RowResult, "Should be an instance of RowResult");
-    return ((RowResult) result).getKey();
-  }
-
-  @VisibleForTesting
-  static class RowResult extends Result {
-    private final ByteString rowKey;
-    private final boolean isMarkerRow;
-
-    static RowResult create(ByteString rowKey, List<Cell> cells) {
-      return new RowResult(rowKey, false, cells);
-    }
-
-    static RowResult createMarker(ByteString rowKey) {
-      return new RowResult(rowKey, true, ImmutableList.<Cell>of());
-    }
-
-    private RowResult(ByteString rowKey, boolean isMarkerRow, List<Cell> cells) {
-      this.rowKey = rowKey;
-      this.isMarkerRow = isMarkerRow;
-
-      // Only default ctor of Result is public, so instantiating cells through copyFrom() because
-      // value(), size(), isEmpty() rawCells() etc. directly usages Result's cells field.
-      this.copyFrom(Result.create(cells));
-    }
-
-    ByteString getKey() {
-      return rowKey;
-    }
-
-    boolean isMarkerRow() {
-      return isMarkerRow;
-    }
+    Cell cell = result.rawCells()[0];
+    return ByteStringer.wrap(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
   }
 
   static class RowResultBuilder implements RowBuilder<Result> {
@@ -226,7 +210,7 @@ public Result finishRow() {
         combined.addAll(familyCellList);
       }
 
-      return RowResult.create(currentKey, combined.build());
+      return Result.create(combined.build());
     }
 
     @Override
@@ -243,9 +227,21 @@ public void reset() {
       this.previousNoLabelCell = null;
     }
 
+    /**
+     * Creates a marker row with rowKey with a scan marker row label and empty family, qualifier and
+     * value.
+     */
     @Override
     public Result createScanMarkerRow(ByteString rowKey) {
-      return RowResult.createMarker(rowKey);
+      return Result.create(
+          ImmutableList.<Cell>of(
+              new RowCell(
+                  ByteStringer.extract(rowKey),
+                  EMPTY_VALUE,
+                  EMPTY_VALUE,
+                  0l,
+                  EMPTY_VALUE,
+                  ImmutableList.<String>of(SCAN_MARKER_ROW_LABEL))));
     }
   }
 }

bigtable-client-core-parent/bigtable-hbase/src/test/java/com/google/cloud/bigtable/hbase/wrappers/veneer/TestRowResultAdapter.java

@@ -18,12 +18,10 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
 import com.google.cloud.bigtable.data.v2.models.RowAdapter;
 import com.google.cloud.bigtable.hbase.adapters.read.RowCell;
-import com.google.cloud.bigtable.hbase.wrappers.veneer.RowResultAdapter.RowResult;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ByteString;
 import java.util.ArrayList;
@@ -63,9 +61,8 @@ public void testWithSingleCellRow() {
     rowBuilder.cellValue(VALUE);
     rowBuilder.finishCell();
 
-    RowResult expected =
-        RowResult.create(
-            ROW_KEY,
+    Result expected =
+        Result.create(
             ImmutableList.<Cell>of(
                 new RowCell(
                     ROW_KEY.toByteArray(),
@@ -186,9 +183,8 @@ public void testDeduplicationLogic() {
     Result actual = rowBuilder.finishRow();
     assertEquals(3, actual.size());
 
-    RowResult expected =
-        RowResult.create(
-            ROW_KEY,
+    Result expected =
+        Result.create(
             ImmutableList.<Cell>of(
                 new RowCell(
                     ROW_KEY.toByteArray(),
@@ -251,13 +247,12 @@ public void testFamilyOrdering() {
 
   @Test
   public void testWithMarkerRow() {
-    RowResult markerRow = RowResult.createMarker(ROW_KEY);
+    Result markerRow = new RowResultAdapter.RowResultBuilder().createScanMarkerRow(ROW_KEY);
     assertTrue(underTest.isScanMarkerRow(markerRow));
-    assertSame(ROW_KEY, underTest.getKey(markerRow));
+    assertEquals(ROW_KEY, underTest.getKey(markerRow));
 
-    RowResult resultWithOneCell =
-        RowResult.create(
-            ROW_KEY,
+    Result resultWithOneCell =
+        Result.create(
             ImmutableList.<Cell>of(
                 new RowCell(
                     ROW_KEY.toByteArray(),