Adds ByteSequence constructors to Key (apache#5327)

Key has methods to get data as ByteSequences but no methods to create
keys using ByteSequence.  This leads to either using Text or
ByteSequence plus byte[].  Text always does an intermediate copy and
ByteSequence plus byte[] is cumbersome.  These new method make it easier
to efficiently create a new Key from an existing Key w/ less code and
less copying of data.

Co-authored-by: Christopher Tubbs <ctubbsii@apache.org>

Author: keith-turner

core/src/main/java/org/apache/accumulo/core/data/Key.java

@@ -552,6 +552,63 @@ public Key(CharSequence row, CharSequence cf, CharSequence cq, ColumnVisibility
         new Text(cv.getExpression()), ts);
   }
 
+  /**
+   * Creates a key with the specified row, the specified column family, the specified column
+   * qualifier, the specified column visibility, the specified timestamp, and delete marker false.
+   * This constructor creates a copy of the fields.
+   * <p>
+   * To avoid copying, use
+   * {@link Key#Key(byte[] row, byte[] cf, byte[] cq, byte[] cv, long ts, boolean deleted, boolean copy)}
+   * instead.
+   *
+   * @see #builder()
+   * @since 3.1.0
+   */
+  public Key(ByteSequence row, ByteSequence cf, ByteSequence cq, ByteSequence cv, long ts) {
+    byte[] rowBytes, cfBytes, cqBytes, cvBytes;
+    int rowOffset, cfOffset, cqOffset, cvOffset;
+    int rowLen, cfLen, cqLen, cvLen;
+
+    if (row.isBackedByArray()) {
+      rowBytes = row.getBackingArray();
+      rowOffset = row.offset();
+    } else {
+      rowBytes = row.toArray();
+      rowOffset = 0;
+    }
+    rowLen = row.length();
+
+    if (cf.isBackedByArray()) {
+      cfBytes = cf.getBackingArray();
+      cfOffset = cf.offset();
+    } else {
+      cfBytes = cf.toArray();
+      cfOffset = 0;
+    }
+    cfLen = cf.length();
+
+    if (cq.isBackedByArray()) {
+      cqBytes = cq.getBackingArray();
+      cqOffset = cq.offset();
+    } else {
+      cqBytes = cq.toArray();
+      cqOffset = 0;
+    }
+    cqLen = cq.length();
+
+    if (cv.isBackedByArray()) {
+      cvBytes = cv.getBackingArray();
+      cvOffset = cv.offset();
+    } else {
+      cvBytes = cv.toArray();
+      cvOffset = 0;
+    }
+    cvLen = cv.length();
+
+    init(rowBytes, rowOffset, rowLen, cfBytes, cfOffset, cfLen, cqBytes, cqOffset, cqLen, cvBytes,
+        cvOffset, cvLen, ts, false, true);
+  }
+
   private byte[] followingArray(byte[] ba) {
     byte[] fba = new byte[ba.length + 1];
     System.arraycopy(ba, 0, fba, 0, ba.length);

core/src/main/java/org/apache/accumulo/core/data/KeyBuilder.java

@@ -100,6 +100,15 @@ public interface RowStep extends Build {
      */
     ColumnFamilyStep row(final byte[] row);
 
+    /**
+     * Set the row of the {@link Key} that this builder will build to the parameter.
+     *
+     * @param row the row to use for the key
+     * @return this builder
+     * @since 3.1.0
+     */
+    ColumnFamilyStep row(final ByteSequence row);
+
     /**
      * Set the row of the {@link Key} that this builder will build to the parameter.
      *
@@ -136,6 +145,15 @@ public interface ColumnFamilyStep extends ColumnVisibilityStep {
      */
     ColumnQualifierStep family(final byte[] columnFamily);
 
+    /**
+     * Set the column family of the {@link Key} that this builder will build to the parameter.
+     *
+     * @param columnFamily the column family to use for the {@link Key}
+     * @return this builder
+     * @since 3.1.0
+     */
+    ColumnQualifierStep family(final ByteSequence columnFamily);
+
     /**
      * Set the column family of the {@link Key} that this builder will build to the parameter.
      *
@@ -180,6 +198,15 @@ public interface ColumnQualifierStep extends ColumnVisibilityStep {
      */
     ColumnVisibilityStep qualifier(final byte[] columnQualifier);
 
+    /**
+     * Set the column qualifier of the {@link Key} that this builder will build to the parameter.
+     *
+     * @param columnQualifier the column qualifier to use for the {@link Key}
+     * @return this builder
+     * @since 3.1.0
+     */
+    ColumnVisibilityStep qualifier(final ByteSequence columnQualifier);
+
     /**
      * Set the column qualifier of the {@link Key} that this builder will build to the parameter.
      *
@@ -225,6 +252,15 @@ public interface ColumnVisibilityStep extends Build {
      */
     Build visibility(final byte[] columnVisibility);
 
+    /**
+     * Set the column qualifier of the {@link Key} that this builder will build to the parameter.
+     *
+     * @param columnVisibility the column visibility to use for the {@link Key}
+     * @return this builder
+     * @since 3.1.0
+     */
+    Build visibility(ByteSequence columnVisibility);
+
     /**
      * Set the column qualifier of the {@link Key} that this builder will build to the parameter.
      *
@@ -312,6 +348,19 @@ public ColumnFamilyStep row(final byte[] row) {
       return row(row, 0, row.length);
     }
 
+    @Override
+    public ColumnFamilyStep row(ByteSequence row) {
+      if (row.isBackedByArray()) {
+        this.row = row.getBackingArray();
+        this.rowOffset = row.offset();
+      } else {
+        this.row = row.toArray();
+        this.rowOffset = 0;
+      }
+      this.rowLength = row.length();
+      return this;
+    }
+
     @Override
     public ColumnFamilyStep row(final Text row) {
       return row(row.getBytes(), 0, row.getLength());
@@ -335,6 +384,19 @@ public ColumnQualifierStep family(final byte[] family) {
       return family(family, 0, family.length);
     }
 
+    @Override
+    public ColumnQualifierStep family(ByteSequence columnFamily) {
+      if (columnFamily.isBackedByArray()) {
+        this.family = columnFamily.getBackingArray();
+        this.familyOffset = columnFamily.offset();
+      } else {
+        this.family = columnFamily.toArray();
+        this.familyOffset = 0;
+      }
+      this.familyLength = columnFamily.length();
+      return this;
+    }
+
     @Override
     public ColumnQualifierStep family(Text family) {
       return family(family.getBytes(), 0, family.getLength());
@@ -358,6 +420,19 @@ public ColumnVisibilityStep qualifier(final byte[] qualifier) {
       return qualifier(qualifier, 0, qualifier.length);
     }
 
+    @Override
+    public ColumnVisibilityStep qualifier(ByteSequence columnQualifier) {
+      if (columnQualifier.isBackedByArray()) {
+        this.qualifier = columnQualifier.getBackingArray();
+        this.qualifierOffset = columnQualifier.offset();
+      } else {
+        this.qualifier = columnQualifier.toArray();
+        this.qualifierOffset = 0;
+      }
+      this.qualifierLength = columnQualifier.length();
+      return this;
+    }
+
     @Override
     public ColumnVisibilityStep qualifier(Text qualifier) {
       return qualifier(qualifier.getBytes(), 0, qualifier.getLength());
@@ -381,6 +456,19 @@ public Build visibility(final byte[] visibility) {
       return visibility(visibility, 0, visibility.length);
     }
 
+    @Override
+    public Build visibility(ByteSequence columnVisibility) {
+      if (columnVisibility.isBackedByArray()) {
+        this.visibility = columnVisibility.getBackingArray();
+        this.visibilityOffset = columnVisibility.offset();
+      } else {
+        this.visibility = columnVisibility.toArray();
+        this.visibilityOffset = 0;
+      }
+      this.visibilityLength = columnVisibility.length();
+      return this;
+    }
+
     @Override
     public Build visibility(Text visibility) {
       return visibility(visibility.getBytes(), 0, visibility.getLength());

core/src/test/java/org/apache/accumulo/core/data/KeyTest.java

@@ -328,4 +328,54 @@ public void testBytesText() {
         new Key(new Text(row), new Text(colFamily), new Text(colQualifier), colVisibility2, ts);
     assertEquals(bytesColVisibilityKey2, textColVisibilityKey2);
   }
+
+  private static class TestByteSequence extends ArrayByteSequence {
+
+    private static final long serialVersionUID = 1234L;
+
+    public TestByteSequence(String s) {
+      super(s);
+    }
+
+    @Override
+    public boolean isBackedByArray() {
+      return false;
+    }
+  }
+
+  @Test
+  public void testByteSequenceConstructor() {
+    var row1 = new ArrayByteSequence("Row");
+    var row2 = new ArrayByteSequence("TheRowData").subSequence(3, 6);
+    var row3 = new TestByteSequence("Row");
+
+    var fam1 = new ArrayByteSequence("Family");
+    var fam2 = new ArrayByteSequence("SomeFamilyData").subSequence(4, 10);
+    var fam3 = new TestByteSequence("Family");
+
+    var qual1 = new ArrayByteSequence("Qual");
+    var qual2 = new ArrayByteSequence("TheQualData").subSequence(3, 7);
+    var qual3 = new TestByteSequence("Qual");
+
+    var vis1 = new ArrayByteSequence("Vis");
+    var vis2 = new ArrayByteSequence("AVisData").subSequence(1, 4);
+    var vis3 = new TestByteSequence("Vis");
+
+    var expectedKey = new Key("Row", "Family", "Qual", "Vis", 4);
+
+    for (var r : List.of(row1, row2, row3)) {
+      for (var f : List.of(fam1, fam2, fam3)) {
+        for (var q : List.of(qual1, qual2, qual3)) {
+          for (var v : List.of(vis1, vis2, vis3)) {
+            var actualKey = new Key(r, f, q, v, 4);
+            assertEquals(expectedKey, actualKey);
+            var actualKey2 =
+                Key.builder().row(r).family(f).qualifier(q).visibility(v).timestamp(4).build();
+            assertEquals(expectedKey, actualKey2);
+          }
+        }
+      }
+    }
+
+  }
 }