GH-15187: [Java] Made reader initialization lazy and added new `get…

…TransferPair()` function that takes in a `Field` type (#34424)

### Rationale for this change
This PR closes #15187. `FieldReader` is being allocated directly in the constructor today, and this PR changes it such that the initialization becomes lazy. Additionally, a new function `getTransferPair(Field, Allocator)` is introduced so that a new `Field` method is not constructed each time `getTransferPair` is called on the Vector.

### What changes are included in this PR?

1. Introduce a new `getTransferPair` method.
2. Make initializing `FieldReader` lazy.

### Are these changes tested?

Yes, some tests have been added to verify these changes.

### Are there any user-facing changes?

I am not 100% sure if there are any user facing changes.

There should not be any breaking changes.
* Closes: #15187

Authored-by: Ramasai <ramasai.tadepalli+3108@gmail.com>
Signed-off-by: David Li <li.davidm96@gmail.com>

java/vector/src/main/java/org/apache/arrow/vector/BaseFixedWidthVector.java

@@ -580,6 +580,14 @@ public TransferPair getTransferPair(BufferAllocator allocator) {
    */
   public abstract TransferPair getTransferPair(String ref, BufferAllocator allocator);
 
+  /**
+   * Construct a transfer pair of this vector and another vector of same type.
+   * @param field Field object used by the target vector
+   * @param allocator allocator for the target vector
+   * @return TransferPair
+   */
+  public abstract TransferPair getTransferPair(Field field, BufferAllocator allocator);
+
   /**
    * Transfer this vector'data to another vector. The memory associated
    * with this vector is transferred to the allocator of target vector

java/vector/src/main/java/org/apache/arrow/vector/BaseValueVector.java

@@ -24,6 +24,7 @@
 import org.apache.arrow.memory.BufferAllocator;
 import org.apache.arrow.memory.ReferenceManager;
 import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.complex.reader.FieldReader;
 import org.apache.arrow.vector.util.DataSizeRoundingUtil;
 import org.apache.arrow.vector.util.TransferPair;
 import org.apache.arrow.vector.util.ValueVectorUtility;
@@ -50,6 +51,8 @@ public abstract class BaseValueVector implements ValueVector {
 
   protected final BufferAllocator allocator;
 
+  protected volatile FieldReader fieldReader;
+
   protected BaseValueVector(BufferAllocator allocator) {
     this.allocator = Preconditions.checkNotNull(allocator, "allocator cannot be null");
   }
@@ -143,6 +146,35 @@ long computeCombinedBufferSize(int valueCount, int typeWidth) {
     return allocator.getRoundingPolicy().getRoundedSize(bufferSize);
   }
 
+  /**
+   * Each vector has a different reader that implements the FieldReader interface. Overridden methods must make
+   * sure to return the correct concrete reader implementation.
+   *
+   * @return Returns a lambda that initializes a reader when called.
+   */
+  protected abstract FieldReader getReaderImpl();
+
+  /**
+   * Default implementation to create a reader for the vector. Depends on the individual vector
+   * class' implementation of {@link #getReaderImpl} to initialize the reader appropriately.
+   *
+   * @return Concrete instance of FieldReader by using double-checked locking.
+   */
+  public FieldReader getReader() {
+    FieldReader reader = fieldReader;
+
+    if (reader != null) {
+      return reader;
+    }
+    synchronized (this) {
+      if (fieldReader == null) {
+        fieldReader = getReaderImpl();
+      }
+
+      return fieldReader;
+    }
+  }
+
   /**
    * Container for primitive vectors (1 for the validity bit-mask and one to hold the values).
    */

java/vector/src/main/java/org/apache/arrow/vector/BigIntVector.java

@@ -37,7 +37,6 @@
  */
 public final class BigIntVector extends BaseFixedWidthVector implements BaseIntVector {
   public static final byte TYPE_WIDTH = 8;
-  private final FieldReader reader;
 
   /**
    * Instantiate a BigIntVector. This doesn't allocate any memory for
@@ -71,17 +70,11 @@ public BigIntVector(String name, FieldType fieldType, BufferAllocator allocator)
    */
   public BigIntVector(Field field, BufferAllocator allocator) {
     super(field, allocator, TYPE_WIDTH);
-    reader = new BigIntReaderImpl(BigIntVector.this);
   }
 
-  /**
-   * Get a reader that supports reading values from this vector.
-   *
-   * @return Field Reader for this vector
-   */
   @Override
-  public FieldReader getReader() {
-    return reader;
+  protected FieldReader getReaderImpl() {
+    return new BigIntReaderImpl(BigIntVector.this);
   }
 
   /**
@@ -286,7 +279,7 @@ public static long get(final ArrowBuf buffer, final int index) {
 
 
   /**
-   * Construct a TransferPair comprising of this and a target vector of
+   * Construct a TransferPair comprising this and a target vector of
    * the same type.
    *
    * @param ref name of the target vector
@@ -298,6 +291,19 @@ public TransferPair getTransferPair(String ref, BufferAllocator allocator) {
     return new TransferImpl(ref, allocator);
   }
 
+  /**
+   * Construct a TransferPair comprising of this and a target vector of
+   * the same type.
+   *
+   * @param field Field object used by the target vector
+   * @param allocator allocator for the target vector
+   * @return {@link TransferPair}
+   */
+  @Override
+  public TransferPair getTransferPair(Field field, BufferAllocator allocator) {
+    return new TransferImpl(field, allocator);
+  }
+
   /**
    * Construct a TransferPair with a desired target vector of the same type.
    *
@@ -331,6 +337,10 @@ public TransferImpl(String ref, BufferAllocator allocator) {
       to = new BigIntVector(ref, field.getFieldType(), allocator);
     }
 
+    public TransferImpl(Field field, BufferAllocator allocator) {
+      to = new BigIntVector(field, allocator);
+    }
+
     public TransferImpl(BigIntVector to) {
       this.to = to;
     }

java/vector/src/main/java/org/apache/arrow/vector/BitVector.java

@@ -46,8 +46,6 @@ public final class BitVector extends BaseFixedWidthVector {
 
   private static final int HASH_CODE_FOR_ONE = 19;
 
-  private final FieldReader reader;
-
   /**
    * Instantiate a BitVector. This doesn't allocate any memory for
    * the data in vector.
@@ -80,17 +78,11 @@ public BitVector(String name, FieldType fieldType, BufferAllocator allocator) {
    */
   public BitVector(Field field, BufferAllocator allocator) {
     super(field, allocator, 0);
-    reader = new BitReaderImpl(BitVector.this);
   }
 
-  /**
-   * Get a reader that supports reading values from this vector.
-   *
-   * @return Field Reader for this vector
-   */
   @Override
-  public FieldReader getReader() {
-    return reader;
+  protected FieldReader getReaderImpl() {
+    return new BitReaderImpl(BitVector.this);
   }
 
   /**
@@ -542,7 +534,7 @@ public void setRangeToOne(int firstBitIndex, int count) {
 
 
   /**
-   * Construct a TransferPair comprising of this and a target vector of
+   * Construct a TransferPair comprising this and a target vector of
    * the same type.
    *
    * @param ref       name of the target vector
@@ -554,6 +546,19 @@ public TransferPair getTransferPair(String ref, BufferAllocator allocator) {
     return new TransferImpl(ref, allocator);
   }
 
+  /**
+   * Construct a TransferPair comprising this and a target vector of
+   * the same type.
+   *
+   * @param field Field object used by the target vector
+   * @param allocator allocator for the target vector
+   * @return {@link TransferPair}
+   */
+  @Override
+  public TransferPair getTransferPair(Field field, BufferAllocator allocator) {
+    return new TransferImpl(field, allocator);
+  }
+
   /**
    * Construct a TransferPair with a desired target vector of the same type.
    *
@@ -572,6 +577,10 @@ public TransferImpl(String ref, BufferAllocator allocator) {
       to = new BitVector(ref, field.getFieldType(), allocator);
     }
 
+    public TransferImpl(Field field, BufferAllocator allocator) {
+      to = new BitVector(field, allocator);
+    }
+
     public TransferImpl(BitVector to) {
       this.to = to;
     }

java/vector/src/main/java/org/apache/arrow/vector/DateDayVector.java

@@ -38,7 +38,6 @@
 public final class DateDayVector extends BaseFixedWidthVector {
 
   public static final byte TYPE_WIDTH = 4;
-  private final FieldReader reader;
 
   /**
    * Instantiate a DateDayVector. This doesn't allocate any memory for
@@ -72,17 +71,11 @@ public DateDayVector(String name, FieldType fieldType, BufferAllocator allocator
    */
   public DateDayVector(Field field, BufferAllocator allocator) {
     super(field, allocator, TYPE_WIDTH);
-    reader = new DateDayReaderImpl(DateDayVector.this);
   }
 
-  /**
-   * Get a reader that supports reading values from this vector.
-   *
-   * @return Field Reader for this vector
-   */
   @Override
-  public FieldReader getReader() {
-    return reader;
+  protected FieldReader getReaderImpl() {
+    return new DateDayReaderImpl(DateDayVector.this);
   }
 
   /**
@@ -290,7 +283,7 @@ public static int get(final ArrowBuf buffer, final int index) {
 
 
   /**
-   * Construct a TransferPair comprising of this and a target vector of
+   * Construct a TransferPair comprising this and a target vector of
    * the same type.
    *
    * @param ref name of the target vector
@@ -302,6 +295,19 @@ public TransferPair getTransferPair(String ref, BufferAllocator allocator) {
     return new TransferImpl(ref, allocator);
   }
 
+  /**
+   * Construct a TransferPair comprising this and a target vector of
+   * the same type.
+   *
+   * @param field Field object used by the target vector
+   * @param allocator allocator for the target vector
+   * @return {@link TransferPair}
+   */
+  @Override
+  public TransferPair getTransferPair(Field field, BufferAllocator allocator) {
+    return new TransferImpl(field, allocator);
+  }
+
   /**
    * Construct a TransferPair with a desired target vector of the same type.
    *
@@ -320,6 +326,10 @@ public TransferImpl(String ref, BufferAllocator allocator) {
       to = new DateDayVector(ref, field.getFieldType(), allocator);
     }
 
+    public TransferImpl(Field field, BufferAllocator allocator) {
+      to = new DateDayVector(field, allocator);
+    }
+
     public TransferImpl(DateDayVector to) {
       this.to = to;
     }

java/vector/src/main/java/org/apache/arrow/vector/DateMilliVector.java

@@ -40,7 +40,6 @@
  */
 public final class DateMilliVector extends BaseFixedWidthVector {
   public static final byte TYPE_WIDTH = 8;
-  private final FieldReader reader;
 
   /**
    * Instantiate a DateMilliVector. This doesn't allocate any memory for
@@ -74,17 +73,11 @@ public DateMilliVector(String name, FieldType fieldType, BufferAllocator allocat
    */
   public DateMilliVector(Field field, BufferAllocator allocator) {
     super(field, allocator, TYPE_WIDTH);
-    reader = new DateMilliReaderImpl(DateMilliVector.this);
   }
 
-  /**
-   * Get a reader that supports reading values from this vector.
-   *
-   * @return Field Reader for this vector
-   */
   @Override
-  public FieldReader getReader() {
-    return reader;
+  protected FieldReader getReaderImpl() {
+    return new DateMilliReaderImpl(DateMilliVector.this);
   }
 
   /**
@@ -293,7 +286,7 @@ public static long get(final ArrowBuf buffer, final int index) {
 
 
   /**
-   * Construct a TransferPair comprising of this and a target vector of
+   * Construct a TransferPair comprising this and a target vector of
    * the same type.
    *
    * @param ref name of the target vector
@@ -305,6 +298,19 @@ public TransferPair getTransferPair(String ref, BufferAllocator allocator) {
     return new TransferImpl(ref, allocator);
   }
 
+  /**
+   * Construct a TransferPair comprising this and a target vector of
+   * the same type.
+   *
+   * @param field Field object used by the target vector
+   * @param allocator allocator for the target vector
+   * @return {@link TransferPair}
+   */
+  @Override
+  public TransferPair getTransferPair(Field field, BufferAllocator allocator) {
+    return new TransferImpl(field, allocator);
+  }
+
   /**
    * Construct a TransferPair with a desired target vector of the same type.
    *
@@ -323,6 +329,10 @@ public TransferImpl(String ref, BufferAllocator allocator) {
       to = new DateMilliVector(ref, field.getFieldType(), allocator);
     }
 
+    public TransferImpl(Field field, BufferAllocator allocator) {
+      to = new DateMilliVector(field, allocator);
+    }
+
     public TransferImpl(DateMilliVector to) {
       this.to = to;
     }