ARROW-18014: [Java] Implement copy functions for vectors and Table

java/vector/src/main/java/org/apache/arrow/vector/table/BaseTable.java

@@ -90,6 +90,10 @@ public BaseTable(List<FieldVector> fieldVectors, int rowCount, DictionaryProvide
     this.schema = new Schema(fields);
   }
 
+  BaseTable() {
+    this.fieldVectors = new ArrayList<>();
+  }
+
   /**
    * Returns a FieldReader for the vector with the given name.
    *
@@ -266,6 +270,48 @@ FieldVector getVector(int columnIndex) {
     return fieldVectors.get(columnIndex);
   }
 
+
+  /**
+   * Returns a copy of the vector with the given name, or throws IllegalArgumentException if the name is not found.
+   * Names are case-sensitive.
+   *
+   * @param columnName The name of the vector to copy
+   * @return A copy of the Vector with the given name
+   * @throws IllegalArgumentException if the name is not the name of a vector in the table.
+   */
+  public FieldVector getVectorCopy(String columnName) {
+    FieldVector source;
+    for (Map.Entry<Field, FieldVector> entry : fieldVectorsMap.entrySet()) {
+      if (entry.getKey().getName().equals(columnName)) {
+        source = entry.getValue();
+        FieldVector copy = source.getField().createVector(source.getAllocator());
+        copy.allocateNew();
+        for (int i = 0; i < source.getValueCount(); i++) {
+          copy.copyFromSafe(i, i, source);
+        }
+        copy.setValueCount(source.getValueCount());
+        return copy;
+      }
+    }
+    throw new IllegalStateException(String.format("No vector named '%s' is present in the table", columnName));
+  }
+
+  /**
+   * Returns a copy of the vector at the given position.
+   *
+   * @param columnIndex The 0-based position of the vector to be copied
+   */
+  public FieldVector getVectorCopy(int columnIndex) {
+    FieldVector source = fieldVectors.get(columnIndex);
+    FieldVector copy = source.getField().createVector(source.getAllocator());
+    copy.allocateNew();
+    for (int i = 0; i < source.getValueCount(); i++) {
+      copy.copyFromSafe(i, i, source);
+    }
+    copy.setValueCount(source.getValueCount());
+    return copy;
+  }
+
   /**
    * Returns an immutable Row object holding a reference to this table. The default character
    * encoding used by the cursor to decode Strings will be StandardCharsets.UTF_8 as this is the only charset

java/vector/src/main/java/org/apache/arrow/vector/table/Table.java

@@ -17,16 +17,20 @@
 
 package org.apache.arrow.vector.table;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
 import org.apache.arrow.util.Preconditions;
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.dictionary.Dictionary;
 import org.apache.arrow.vector.dictionary.DictionaryProvider;
+import org.apache.arrow.vector.types.pojo.DictionaryEncoding;
 import org.apache.arrow.vector.util.TransferPair;
 
 /**
@@ -95,6 +99,36 @@ public Table(VectorSchemaRoot vsr) {
     vsr.clear();
   }
 
+  /**
+   * Returns a deep copy of this table.
+   */
+  public Table copy() {
+    List<FieldVector> vectorCopies = new ArrayList<>();
+    for (int i = 0; i < getVectorCount(); i++) {
+      vectorCopies.add(getVectorCopy(i));
+    }
+    DictionaryProvider providerCopy = null;
+    if (dictionaryProvider != null) {
+      Set<Long> ids = dictionaryProvider.getDictionaryIds();
+      Dictionary[] dictionaryCopies = new Dictionary[ids.size()];
+      int i = 0;
+      for (Long id : ids) {
+        Dictionary src = dictionaryProvider.lookup(id);
+        FieldVector srcVector = src.getVector();
+        FieldVector destVector = srcVector.getField().createVector(srcVector.getAllocator());
+        destVector.copyFromSafe(0, srcVector.getValueCount(), srcVector); // TODO: Remove safe copy for perf
+        DictionaryEncoding srcEncoding = src.getEncoding();
+        Dictionary dest = new Dictionary(destVector,
+            new DictionaryEncoding(srcEncoding.getId(), srcEncoding.isOrdered(), srcEncoding.getIndexType()));
+        dictionaryCopies[i] = dest;
+        i++;
+      }
+      providerCopy = new DictionaryProvider.MapDictionaryProvider(dictionaryCopies);
+    }
+    return new Table(vectorCopies, (int) getRowCount(), providerCopy);
+  }
+
+
   /**
    * Returns a new Table created by adding the given vector to the vectors in this Table.
    *

java/vector/src/test/java/org/apache/arrow/vector/table/BaseTableTest.java

@@ -180,6 +180,47 @@ void testGetVector() {
     }
   }
 
+  @Test
+  void getVectorCopyByIndex() {
+    List<FieldVector> vectorList = twoIntColumns(allocator);
+    List<FieldVector> vectorList2 = twoIntColumns(allocator);
+    try (Table t = new Table(vectorList)) {
+      // compare value by value
+      for (int vIdx = 0; vIdx < vectorList.size(); vIdx++) {
+        IntVector original = (IntVector) vectorList2.get(vIdx);
+        IntVector copy = (IntVector) t.getVectorCopy(vIdx);
+        assertNotNull(copy);
+        assertEquals(2, copy.getValueCount());
+        assertEquals(0, copy.getNullCount());
+        for (int i = 0; i < t.getRowCount(); i++) {
+          assertEquals(original.getObject(i), copy.getObject(i));
+        }
+      }
+      assertThrows(IllegalStateException.class,
+          () -> t.getVector("wrong name"));
+    }
+  }
+
+  @Test
+  void getVectorCopyByName() {
+    List<FieldVector> vectorList = twoIntColumns(allocator);
+    List<FieldVector> vectorList2 = twoIntColumns(allocator);
+    try (Table t = new Table(vectorList)) {
+      assertNotNull(t.getVectorCopy(INT_VECTOR_NAME_1));
+      for (int vIdx = 0; vIdx < vectorList.size(); vIdx++) {
+        IntVector original = (IntVector) vectorList2.get(vIdx);
+        IntVector copy = (IntVector) t.getVectorCopy(original.getName());
+        assertEquals(2, copy.getValueCount());
+        assertEquals(0, copy.getNullCount());
+        for (int i = 0; i < t.getRowCount(); i++) {
+          assertEquals(original.getObject(i), copy.getObject(i));
+        }
+      }
+      assertThrows(IllegalStateException.class,
+          () -> t.getVector("wrong name"));
+    }
+  }
+
   @Test
   void immutableCursor() {
     List<FieldVector> vectorList = twoIntColumns(allocator);

java/vector/src/test/java/org/apache/arrow/vector/table/TableTest.java

@@ -127,6 +127,26 @@ void constructor2() {
     }
   }
 
+  @Test
+  void copy() {
+    List<FieldVector> vectorList = twoIntColumns(allocator);
+    try (Table t = new Table(vectorList)) {
+      assertEquals(2, t.getVectorCount());
+      try (Table copy = t.copy()) {
+        for (FieldVector v: t.fieldVectors) {
+          FieldVector vCopy = copy.getVector(v.getName());
+          assertNotNull(vCopy);
+          assertEquals(v.getValueCount(), vCopy.getValueCount());
+          for (int i = 0; i < v.getValueCount(); i++) {
+            Integer vValue = ((IntVector) v).getObject(i);
+            Integer vCopyValue = ((IntVector) vCopy).getObject(i);
+            assertEquals(vValue, vCopyValue);
+          }
+        }
+      }
+    }
+  }
+
   @Test
   void addVector() {
     List<FieldVector> vectorList = twoIntColumns(allocator);