ARROW-6210: [Java] remove equals API from ValueVector

Related to [ARROW-6210](https://issues.apache.org/jira/browse/ARROW-6210).
This is a follow-up from #4933
The callers should be fixed to use the RangeEquals API instead.

Closes #5065 from tianchen92/ARROW-6210 and squashes the following commits:

cae440a <tianchen> resolve comments
ad2277b <tianchen> resolve comments and port tests
fb1510e <tianchen> add ValueMatcher
14b3e67 <tianchen> ARROW-6210:  remove equals API from ValueVector

Authored-by: tianchen <niki.lj@alibaba-inc.com>
Signed-off-by: Bryan Cutler <cutlerb@gmail.com>

Author: tianchen92

java/vector/src/main/codegen/templates/UnionVector.java

@@ -677,20 +677,6 @@ public int hashCode(int index) {
       return getVector(index).hashCode(index);
     }
 
-    @Override
-    public boolean equals(int index, ValueVector to, int toIndex) {
-      if (to == null) {
-        return false;
-      }
-      Preconditions.checkArgument(index >= 0 && index < valueCount,
-          "index %s out of range[0, %s]:", index, valueCount - 1);
-      Preconditions.checkArgument(toIndex >= 0 && toIndex < to.getValueCount(),
-          "index %s out of range[0, %s]:", index, to.getValueCount() - 1);
-
-      RangeEqualsVisitor visitor = new RangeEqualsVisitor(to, index, toIndex, 1);
-      return this.accept(visitor);
-    }
-
     @Override
     public boolean accept(RangeEqualsVisitor visitor) {
       return visitor.visit(this);

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

@@ -884,21 +884,6 @@ public int hashCode(int index) {
     return ByteFunctionHelpers.hash(this.getDataBuffer(), start, end);
   }
 
-  @Override
-  public boolean equals(int index, ValueVector to, int toIndex) {
-    if (to == null) {
-      return false;
-    }
-
-    Preconditions.checkArgument(index >= 0 && index < valueCount,
-        "index %s out of range[0, %s]:", index, valueCount - 1);
-    Preconditions.checkArgument(toIndex >= 0 && toIndex < to.getValueCount(),
-        "index %s out of range[0, %s]:", index, to.getValueCount() - 1);
-
-    RangeEqualsVisitor visitor = new RangeEqualsVisitor(to, index, toIndex, 1);
-    return this.accept(visitor);
-  }
-
   @Override
   public boolean accept(RangeEqualsVisitor visitor) {
     return visitor.visit(this);

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

@@ -1367,20 +1367,6 @@ public int hashCode(int index) {
     return ByteFunctionHelpers.hash(this.getDataBuffer(), start, end);
   }
 
-  @Override
-  public boolean equals(int index, ValueVector to, int toIndex) {
-    if (to == null) {
-      return false;
-    }
-    Preconditions.checkArgument(index >= 0 && index < valueCount,
-        "index %s out of range[0, %s]:", index, valueCount - 1);
-    Preconditions.checkArgument(toIndex >= 0 && toIndex < to.getValueCount(),
-        "index %s out of range[0, %s]:", index, to.getValueCount() - 1);
-
-    RangeEqualsVisitor visitor = new RangeEqualsVisitor(to, index, toIndex, 1);
-    return this.accept(visitor);
-  }
-
   @Override
   public boolean accept(RangeEqualsVisitor visitor) {
     return visitor.visit(this);

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

@@ -244,15 +244,6 @@ public interface ValueVector extends Closeable, Iterable<ValueVector> {
    */
   int hashCode(int index);
 
-  /**
-   * Check whether the element in index equals to the element in targetIndex from the target vector.
-   * @param index index to compare in this vector
-   * @param target target vector
-   * @param targetIndex index to compare in target vector
-   * @return true if equals, otherwise false.
-   */
-  boolean equals(int index, ValueVector target, int targetIndex);
-
   /**
    * Copy a cell value from a particular index in source vector to a particular
    * position in this vector.

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

@@ -251,11 +251,6 @@ public int hashCode(int index) {
     return 0;
   }
 
-  @Override
-  public boolean equals(int index, ValueVector to, int toIndex) {
-    return false;
-  }
-
   @Override
   public void copyFrom(int fromIndex, int thisIndex, ValueVector from) {
     throw new UnsupportedOperationException();

java/vector/src/main/java/org/apache/arrow/vector/compare/RangeEqualsVisitor.java

@@ -20,6 +20,7 @@
 import java.util.List;
 
 import org.apache.arrow.memory.util.ByteFunctionHelpers;
+import org.apache.arrow.util.Preconditions;
 import org.apache.arrow.vector.BaseFixedWidthVector;
 import org.apache.arrow.vector.BaseVariableWidthVector;
 import org.apache.arrow.vector.FieldVector;
@@ -37,62 +38,93 @@
 public class RangeEqualsVisitor {
 
   protected final ValueVector right;
-  protected final int leftStart;
-  protected final int rightStart;
-  protected final int length;
+  protected int leftStart;
+  protected int rightStart;
+  protected int length;
+
+  protected boolean typeCheckNeeded = true;
 
   /**
    * Constructs a new instance.
    */
-  public RangeEqualsVisitor(ValueVector right, int leftStart, int rightStart, int length) {
+  public RangeEqualsVisitor(ValueVector right, int rightStart,  int leftStart, int length, boolean typeCheckNeeded) {
     this.leftStart = leftStart;
     this.rightStart = rightStart;
     this.right = right;
     this.length = length;
+    this.typeCheckNeeded = typeCheckNeeded;
+    Preconditions.checkArgument(length >= 0, "length must be non negative");
+  }
+
+  /**
+   * Constructs a new instance.
+   */
+  public RangeEqualsVisitor(ValueVector right, int leftStart, int rightStart, int length) {
+    this(right, rightStart, leftStart, length, true);
+  }
+
+  /**
+   * Do some validation work, like type check and indices check.
+   */
+  private boolean validate(ValueVector left) {
+
+    if (!compareValueVector(left, right)) {
+      return false;
+    }
+
+    Preconditions.checkArgument(leftStart >= 0,
+        "leftStart %s must be non negative.", leftStart);
+    Preconditions.checkArgument((leftStart + length) <= left.getValueCount(),
+        "(leftStart + length) %s out of range[0, %s].", 0, left.getValueCount());
+    Preconditions.checkArgument(rightStart >= 0,
+        "rightStart %s must be non negative.", rightStart);
+    Preconditions.checkArgument((rightStart + length) <= right.getValueCount(),
+        "(rightStart + length) %s out of range[0, %s].", 0, right.getValueCount());
+
+    return true;
   }
 
   public boolean visit(BaseFixedWidthVector left) {
-    return compareBaseFixedWidthVectors(left);
+    return validate(left) && compareBaseFixedWidthVectors(left);
   }
 
   public boolean visit(BaseVariableWidthVector left) {
-    return compareBaseVariableWidthVectors(left);
+    return validate(left) && compareBaseVariableWidthVectors(left);
   }
 
   public boolean visit(ListVector left) {
-    return compareListVectors(left);
+    return validate(left) && compareListVectors(left);
   }
 
   public boolean visit(FixedSizeListVector left) {
-    return compareFixedSizeListVectors(left);
+    return validate(left) && compareFixedSizeListVectors(left);
   }
 
   public boolean visit(NonNullableStructVector left) {
-    return compareStructVectors(left);
+    return validate(left) && compareStructVectors(left);
   }
 
   public boolean visit(UnionVector left) {
-    return compareUnionVectors(left);
+    return validate(left) && compareUnionVectors(left);
   }
 
   public boolean visit(ZeroVector left) {
-    return compareValueVector(left, right);
+    return validate(left);
   }
 
   public boolean visit(ValueVector left) {
     throw new UnsupportedOperationException();
   }
 
   protected boolean compareValueVector(ValueVector left, ValueVector right) {
+    if (!typeCheckNeeded) {
+      return true;
+    }
     return left.getField().getType().equals(right.getField().getType());
   }
 
   protected boolean compareUnionVectors(UnionVector left) {
 
-    if (!compareValueVector(left, right)) {
-      return false;
-    }
-
     UnionVector rightVector = (UnionVector) right;
 
     List<FieldVector> leftChildren = left.getChildrenFromFields();
@@ -113,9 +145,6 @@ protected boolean compareUnionVectors(UnionVector left) {
   }
 
   protected boolean compareStructVectors(NonNullableStructVector left) {
-    if (!compareValueVector(left, right)) {
-      return false;
-    }
 
     NonNullableStructVector rightVector = (NonNullableStructVector) right;
 
@@ -136,10 +165,6 @@ protected boolean compareStructVectors(NonNullableStructVector left) {
 
   protected boolean compareBaseFixedWidthVectors(BaseFixedWidthVector left) {
 
-    if (!compareValueVector(left, right)) {
-      return false;
-    }
-
     for (int i = 0; i < length; i++) {
       int leftIndex = leftStart + i;
       int rightIndex = rightStart + i;
@@ -152,14 +177,14 @@ protected boolean compareBaseFixedWidthVectors(BaseFixedWidthVector left) {
 
       int typeWidth = left.getTypeWidth();
       if (!isNull) {
-        int startByteLeft = typeWidth * leftIndex;
-        int endByteLeft = typeWidth * (leftIndex + 1);
+        int startIndexLeft = typeWidth * leftIndex;
+        int endIndexLeft = typeWidth * (leftIndex + 1);
 
-        int startByteRight = typeWidth * rightIndex;
-        int endByteRight = typeWidth * (rightIndex + 1);
+        int startIndexRight = typeWidth * rightIndex;
+        int endIndexRight = typeWidth * (rightIndex + 1);
 
-        int ret = ByteFunctionHelpers.equal(left.getDataBuffer(), startByteLeft, endByteLeft,
-            right.getDataBuffer(), startByteRight, endByteRight);
+        int ret = ByteFunctionHelpers.equal(left.getDataBuffer(), startIndexLeft, endIndexLeft,
+            right.getDataBuffer(), startIndexRight, endIndexRight);
 
         if (ret == 0) {
           return false;
@@ -170,9 +195,6 @@ protected boolean compareBaseFixedWidthVectors(BaseFixedWidthVector left) {
   }
 
   protected boolean compareBaseVariableWidthVectors(BaseVariableWidthVector left) {
-    if (!compareValueVector(left, right)) {
-      return false;
-    }
 
     for (int i = 0; i < length; i++) {
       int leftIndex = leftStart + i;
@@ -186,14 +208,14 @@ protected boolean compareBaseVariableWidthVectors(BaseVariableWidthVector left)
       int offsetWidth = BaseVariableWidthVector.OFFSET_WIDTH;
 
       if (!isNull) {
-        final int startByteLeft = left.getOffsetBuffer().getInt(leftIndex * offsetWidth);
-        final int endByteLeft = left.getOffsetBuffer().getInt((leftIndex + 1) * offsetWidth);
+        final int startIndexLeft = left.getOffsetBuffer().getInt(leftIndex * offsetWidth);
+        final int endIndexLeft = left.getOffsetBuffer().getInt((leftIndex + 1) * offsetWidth);
 
-        final int startByteRight = right.getOffsetBuffer().getInt(rightIndex * offsetWidth);
-        final int endByteRight = right.getOffsetBuffer().getInt((rightIndex + 1) * offsetWidth);
+        final int startIndexRight = right.getOffsetBuffer().getInt(rightIndex * offsetWidth);
+        final int endIndexRight = right.getOffsetBuffer().getInt((rightIndex + 1) * offsetWidth);
 
-        int ret = ByteFunctionHelpers.equal(left.getDataBuffer(), startByteLeft, endByteLeft,
-            right.getDataBuffer(), startByteRight, endByteRight);
+        int ret = ByteFunctionHelpers.equal(left.getDataBuffer(), startIndexLeft, endIndexLeft,
+            right.getDataBuffer(), startIndexRight, endIndexRight);
 
         if (ret == 0) {
           return false;
@@ -204,9 +226,6 @@ protected boolean compareBaseVariableWidthVectors(BaseVariableWidthVector left)
   }
 
   protected boolean compareListVectors(ListVector left) {
-    if (!compareValueVector(left, right)) {
-      return false;
-    }
 
     for (int i = 0; i < length; i++) {
       int leftIndex = leftStart + i;
@@ -220,21 +239,21 @@ protected boolean compareListVectors(ListVector left) {
       int offsetWidth = BaseRepeatedValueVector.OFFSET_WIDTH;
 
       if (!isNull) {
-        final int startByteLeft = left.getOffsetBuffer().getInt(leftIndex * offsetWidth);
-        final int endByteLeft = left.getOffsetBuffer().getInt((leftIndex + 1) * offsetWidth);
+        final int startIndexLeft = left.getOffsetBuffer().getInt(leftIndex * offsetWidth);
+        final int endIndexLeft = left.getOffsetBuffer().getInt((leftIndex + 1) * offsetWidth);
 
-        final int startByteRight = right.getOffsetBuffer().getInt(rightIndex * offsetWidth);
-        final int endByteRight = right.getOffsetBuffer().getInt((rightIndex + 1) * offsetWidth);
+        final int startIndexRight = right.getOffsetBuffer().getInt(rightIndex * offsetWidth);
+        final int endIndexRight = right.getOffsetBuffer().getInt((rightIndex + 1) * offsetWidth);
 
-        if ((endByteLeft - startByteLeft) != (endByteRight - startByteRight)) {
+        if ((endIndexLeft - startIndexLeft) != (endIndexRight - startIndexRight)) {
           return false;
         }
 
         ValueVector leftDataVector = left.getDataVector();
         ValueVector rightDataVector = ((ListVector)right).getDataVector();
 
-        if (!leftDataVector.accept(new RangeEqualsVisitor(rightDataVector, startByteLeft,
-            startByteRight, (endByteLeft - startByteLeft)))) {
+        if (!leftDataVector.accept(new RangeEqualsVisitor(rightDataVector, startIndexLeft,
+            startIndexRight, (endIndexLeft - startIndexLeft)))) {
           return false;
         }
       }
@@ -243,9 +262,6 @@ protected boolean compareListVectors(ListVector left) {
   }
 
   protected boolean compareFixedSizeListVectors(FixedSizeListVector left) {
-    if (!compareValueVector(left, right)) {
-      return false;
-    }
 
     if (left.getListSize() != ((FixedSizeListVector)right).getListSize()) {
       return false;
@@ -263,26 +279,25 @@ protected boolean compareFixedSizeListVectors(FixedSizeListVector left) {
       int listSize = left.getListSize();
 
       if (!isNull) {
-        final int startByteLeft = leftIndex * listSize;
-        final int endByteLeft = (leftIndex + 1) * listSize;
+        final int startIndexLeft = leftIndex * listSize;
+        final int endIndexLeft = (leftIndex + 1) * listSize;
 
-        final int startByteRight = rightIndex * listSize;
-        final int endByteRight = (rightIndex + 1) * listSize;
+        final int startIndexRight = rightIndex * listSize;
+        final int endIndexRight = (rightIndex + 1) * listSize;
 
-        if ((endByteLeft - startByteLeft) != (endByteRight - startByteRight)) {
+        if ((endIndexLeft - startIndexLeft) != (endIndexRight - startIndexRight)) {
           return false;
         }
 
         ValueVector leftDataVector = left.getDataVector();
         ValueVector rightDataVector = ((FixedSizeListVector)right).getDataVector();
 
-        if (!leftDataVector.accept(new RangeEqualsVisitor(rightDataVector, startByteLeft, startByteRight,
-            (endByteLeft - startByteLeft)))) {
+        if (!leftDataVector.accept(new RangeEqualsVisitor(rightDataVector, startIndexLeft, startIndexRight,
+            (endIndexLeft - startIndexLeft)))) {
           return false;
         }
       }
     }
     return true;
   }
-
 }

java/vector/src/main/java/org/apache/arrow/vector/compare/VectorEqualsVisitor.java

@@ -26,7 +26,11 @@
 public class VectorEqualsVisitor extends RangeEqualsVisitor {
 
   public VectorEqualsVisitor(ValueVector right) {
-    super(Preconditions.checkNotNull(right), 0, 0, right.getValueCount());
+    this(right, true);
+  }
+
+  public VectorEqualsVisitor(ValueVector right, boolean typeCheckNeeded) {
+    super(Preconditions.checkNotNull(right), 0, 0, right.getValueCount(), typeCheckNeeded);
   }
 
   @Override

java/vector/src/main/java/org/apache/arrow/vector/complex/FixedSizeListVector.java

@@ -538,20 +538,6 @@ public int hashCode(int index) {
     return hash;
   }
 
-  @Override
-  public boolean equals(int index, ValueVector to, int toIndex) {
-    if (to == null) {
-      return false;
-    }
-    Preconditions.checkArgument(index >= 0 && index < valueCount,
-        "index %s out of range[0, %s]:", index, valueCount - 1);
-    Preconditions.checkArgument(toIndex >= 0 && toIndex < to.getValueCount(),
-        "index %s out of range[0, %s]:", index, to.getValueCount() - 1);
-
-    RangeEqualsVisitor visitor = new RangeEqualsVisitor(to, index, toIndex, 1);
-    return this.accept(visitor);
-  }
-
   @Override
   public boolean accept(RangeEqualsVisitor visitor) {
     return visitor.visit(this);