Adding new Mutation constructor with ByteSequence parameter to remove 1 TODO

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

@@ -60,8 +60,7 @@ public ConditionalMutation(CharSequence row, Condition... conditions) {
   }
 
   public ConditionalMutation(ByteSequence row, Condition... conditions) {
-    // TODO add ByteSequence methods to mutations
-    super(row.toArray());
+    super(row);
     init(conditions);
   }
 

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

@@ -134,6 +134,24 @@ private ByteBuffer serializedSnapshot() {
     }
   }
 
+  static byte[] copyIfNeeded(byte[] ba, int off, int len, boolean copyData) {
+    if (len == 0) {
+      return EMPTY_BYTES;
+    }
+
+    if (!copyData && ba.length == len && off == 0) {
+      return ba;
+    }
+
+    byte[] copy = new byte[len];
+    System.arraycopy(ba, off, copy, 0, len);
+    return copy;
+  }
+
+  private final void init(byte[] r, int rOff, int rLen, boolean copy) {
+    row = copyIfNeeded(r, rOff, rLen, copy);
+  }
+
   /**
    * Creates a new mutation. A defensive copy is made.
    *
@@ -230,6 +248,26 @@ public Mutation(CharSequence row, int initialBufferSize) {
    */
   public Mutation() {}
 
+  /**
+   * Creates a mutation with the specified row, This constructor creates a copy of the fields.
+   */
+  public Mutation(ByteSequence row) {
+    byte[] rowBytes;
+    int rowOffset;
+    int rowLen;
+
+    if (row.isBackedByArray()) {
+      rowBytes = row.getBackingArray();
+      rowOffset = row.offset();
+    } else {
+      rowBytes = row.toArray();
+      rowOffset = 0;
+    }
+    rowLen = row.length();
+
+    init(rowBytes, rowOffset, rowLen, true);
+  }
+
   /**
    * Creates a new mutation from a Thrift mutation.
    *
@@ -271,6 +309,16 @@ public byte[] getRow() {
     return row;
   }
 
+  /**
+   * Returns the row ID as a byte sequence. This method returns a pointer to the key's internal data
+   * and does not copy it.
+   *
+   * @return ByteSequence that points to the internal key row ID data
+   */
+  public ByteSequence getRowData() {
+    return new ArrayByteSequence(row);
+  }
+
   private void fill(byte[] b) {
     fill(b, b.length);
   }

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

@@ -1055,4 +1055,31 @@ public void testKeyColumnsAddingNullValue() {
     assertThrows(NullPointerException.class, () -> m.at().keyColumns(k).put(v));
   }
 
+  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 expectedMutation = new Mutation(new ArrayByteSequence("Row".getBytes(UTF_8)));
+
+    for (var r : List.of(row1, row2, row3)) {
+      var actualMutation = new Mutation(new ArrayByteSequence(r));
+      assertEquals(expectedMutation.getRowData(), actualMutation.getRowData());
+    }
+  }
 }