ARROW-6355: [Java] Make range equal visitor reusable

java/algorithm/src/main/java/org/apache/arrow/algorithm/deduplicate/DeduplicationUtils.java

@@ -22,6 +22,7 @@
 import org.apache.arrow.vector.BitVectorHelper;
 import org.apache.arrow.vector.IntVector;
 import org.apache.arrow.vector.ValueVector;
+import org.apache.arrow.vector.compare.Range;
 import org.apache.arrow.vector.compare.RangeEqualsVisitor;
 
 import io.netty.buffer.ArrowBuf;
@@ -43,11 +44,11 @@ public static <V extends ValueVector> void populateRunStartIndicators(V vector,
     runStarts.setZero(0, bufSize);
 
     BitVectorHelper.setValidityBitToOne(runStarts, 0);
-
+    RangeEqualsVisitor visitor = new RangeEqualsVisitor(vector, vector, false);
+    Range range = new Range(0, 0, 1);
     for (int i = 1; i < vector.getValueCount(); i++) {
-      RangeEqualsVisitor visitor = new RangeEqualsVisitor(
-              vector, i - 1, i, /* length */1, /* need check type*/false);
-      if (!visitor.equals(vector)) {
+      range.setLeftStart(i).setRightStart(i - 1);
+      if (!visitor.rangeEquals(range)) {
         BitVectorHelper.setValidityBitToOne(runStarts, i);
       }
     }

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

@@ -17,18 +17,10 @@
 
 package org.apache.arrow.vector.compare;
 
-import java.util.List;
-
 import org.apache.arrow.vector.BaseFixedWidthVector;
-import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.Float4Vector;
 import org.apache.arrow.vector.Float8Vector;
 import org.apache.arrow.vector.ValueVector;
-import org.apache.arrow.vector.complex.BaseRepeatedValueVector;
-import org.apache.arrow.vector.complex.FixedSizeListVector;
-import org.apache.arrow.vector.complex.ListVector;
-import org.apache.arrow.vector.complex.NonNullableStructVector;
-import org.apache.arrow.vector.complex.UnionVector;
 
 /**
  * Visitor to compare floating point.
@@ -49,24 +41,16 @@ public class ApproxEqualsVisitor extends RangeEqualsVisitor {
   private DiffFunction<Double> doubleDiffFunction =
       (Double value1, Double value2) -> Math.abs(value1 - value2);
 
-  public ApproxEqualsVisitor(ValueVector right, float epsilon) {
-    this (right, epsilon, epsilon, true);
-  }
-
-  public ApproxEqualsVisitor(ValueVector right, float floatEpsilon, double doubleEpsilon) {
-    this (right, floatEpsilon, doubleEpsilon, true);
-  }
-
-  public ApproxEqualsVisitor(ValueVector right, float floatEpsilon, double doubleEpsilon, boolean typeCheckNeeded) {
-    this (right, floatEpsilon, doubleEpsilon, typeCheckNeeded, 0, 0, right.getValueCount());
-  }
-
   /**
-   * Construct an instance.
+   * Constructs a new instance.
+   *
+   * @param left left vector
+   * @param right right vector
+   * @param floatEpsilon difference for float values
+   * @param doubleEpsilon difference for double values
    */
-  public ApproxEqualsVisitor(ValueVector right, float floatEpsilon, double doubleEpsilon, boolean typeCheckNeeded,
-      int leftStart, int rightStart, int length) {
-    super(right, rightStart, leftStart, length, typeCheckNeeded);
+  public ApproxEqualsVisitor(ValueVector left, ValueVector right, float floatEpsilon, double doubleEpsilon) {
+    super(left, right, true);
     this.floatEpsilon = floatEpsilon;
     this.doubleEpsilon = doubleEpsilon;
   }
@@ -80,151 +64,38 @@ public void setDoubleDiffFunction(DiffFunction<Double> doubleDiffFunction) {
   }
 
   @Override
-  public Boolean visit(BaseFixedWidthVector left, Void value) {
+  public Boolean visit(BaseFixedWidthVector left, Range range) {
     if (left instanceof Float4Vector) {
-      return validate(left) && float4ApproxEquals((Float4Vector) left);
+      return float4ApproxEquals(range);
     } else if (left instanceof Float8Vector) {
-      return validate(left) && float8ApproxEquals((Float8Vector) left);
+      return float8ApproxEquals(range);
     } else {
-      return super.visit(left, value);
+      return super.visit(left, range);
     }
   }
 
   @Override
-  protected boolean compareUnionVectors(UnionVector left) {
-
-    UnionVector rightVector = (UnionVector) right;
-
-    List<FieldVector> leftChildren = left.getChildrenFromFields();
-    List<FieldVector> rightChildren = rightVector.getChildrenFromFields();
-
-    if (leftChildren.size() != rightChildren.size()) {
-      return false;
-    }
-
-    for (int k = 0; k < leftChildren.size(); k++) {
-      ApproxEqualsVisitor visitor = new ApproxEqualsVisitor(rightChildren.get(k),
-          floatEpsilon, doubleEpsilon);
-      if (!leftChildren.get(k).accept(visitor, null)) {
-        return false;
-      }
-    }
-    return true;
+  protected ApproxEqualsVisitor createInnerVisitor(ValueVector left, ValueVector right) {
+    return new ApproxEqualsVisitor(left, right, floatEpsilon, doubleEpsilon);
   }
 
-  @Override
-  protected boolean compareStructVectors(NonNullableStructVector left) {
+  private boolean float4ApproxEquals(Range range) {
+    Float4Vector leftVector  = (Float4Vector) getLeft();
+    Float4Vector rightVector  = (Float4Vector) getRight();
 
-    NonNullableStructVector rightVector = (NonNullableStructVector) right;
+    for (int i = 0; i < range.getLength(); i++) {
+      int leftIndex = range.getLeftStart() + i;
+      int rightIndex = range.getRightStart() + i;
 
-    if (!left.getChildFieldNames().equals(rightVector.getChildFieldNames())) {
-      return false;
-    }
+      boolean isNull = leftVector.isNull(leftIndex);
 
-    for (String name : left.getChildFieldNames()) {
-      ApproxEqualsVisitor visitor = new ApproxEqualsVisitor(rightVector.getChild(name),
-          floatEpsilon, doubleEpsilon);
-      if (!left.getChild(name).accept(visitor, null)) {
+      if (isNull != rightVector.isNull(rightIndex)) {
         return false;
       }
-    }
-
-    return true;
-  }
-
-  @Override
-  protected boolean compareListVectors(ListVector left) {
-
-    for (int i = 0; i < length; i++) {
-      int leftIndex = leftStart + i;
-      int rightIndex = rightStart + i;
-
-      boolean isNull = left.isNull(leftIndex);
-      if (isNull != right.isNull(rightIndex)) {
-        return false;
-      }
-
-      int offsetWidth = BaseRepeatedValueVector.OFFSET_WIDTH;
 
       if (!isNull) {
-        final int startIndexLeft = left.getOffsetBuffer().getInt(leftIndex * offsetWidth);
-        final int endIndexLeft = left.getOffsetBuffer().getInt((leftIndex + 1) * offsetWidth);
-
-        final int startIndexRight = right.getOffsetBuffer().getInt(rightIndex * offsetWidth);
-        final int endIndexRight = right.getOffsetBuffer().getInt((rightIndex + 1) * offsetWidth);
-
-        if ((endIndexLeft - startIndexLeft) != (endIndexRight - startIndexRight)) {
-          return false;
-        }
-
-        ValueVector leftDataVector = left.getDataVector();
-        ValueVector rightDataVector = ((ListVector)right).getDataVector();
-
-        if (!leftDataVector.accept(new ApproxEqualsVisitor(rightDataVector, floatEpsilon, doubleEpsilon,
-            typeCheckNeeded, startIndexLeft, startIndexRight, (endIndexLeft - startIndexLeft)), null)) {
-          return false;
-        }
-      }
-    }
-    return true;
-  }
-
-  protected boolean compareFixedSizeListVectors(FixedSizeListVector left) {
-
-    if (left.getListSize() != ((FixedSizeListVector)right).getListSize()) {
-      return false;
-    }
-
-    for (int i = 0; i < length; i++) {
-      int leftIndex = leftStart + i;
-      int rightIndex = rightStart + i;
-
-      boolean isNull = left.isNull(leftIndex);
-      if (isNull != right.isNull(rightIndex)) {
-        return false;
-      }
-
-      int listSize = left.getListSize();
-
-      if (!isNull) {
-        final int startIndexLeft = leftIndex * listSize;
-        final int endIndexLeft = (leftIndex + 1) * listSize;
-
-        final int startIndexRight = rightIndex * listSize;
-        final int endIndexRight = (rightIndex + 1) * listSize;
-
-        if ((endIndexLeft - startIndexLeft) != (endIndexRight - startIndexRight)) {
-          return false;
-        }
-
-        ValueVector leftDataVector = left.getDataVector();
-        ValueVector rightDataVector = ((FixedSizeListVector)right).getDataVector();
-
-        if (!leftDataVector.accept(new ApproxEqualsVisitor(rightDataVector, floatEpsilon, doubleEpsilon,
-            typeCheckNeeded, startIndexLeft, startIndexRight, (endIndexLeft - startIndexLeft)), null)) {
-          return false;
-        }
-      }
-    }
-    return true;
-  }
-
-  private boolean float4ApproxEquals(Float4Vector left) {
-
-    for (int i = 0; i < length; i++) {
-      int leftIndex = leftStart + i;
-      int rightIndex = rightStart + i;
-
-      boolean isNull = left.isNull(leftIndex);
-
-      if (isNull != right.isNull(rightIndex)) {
-        return false;
-      }
-
-      if (!isNull) {
-
-        Float leftValue = left.get(leftIndex);
-        Float rightValue = ((Float4Vector)right).get(rightIndex);
+        Float leftValue = leftVector.get(leftIndex);
+        Float rightValue = rightVector.get(rightIndex);
         if (leftValue.isNaN()) {
           return rightValue.isNaN();
         }
@@ -239,21 +110,24 @@ private boolean float4ApproxEquals(Float4Vector left) {
     return true;
   }
 
-  private boolean float8ApproxEquals(Float8Vector left) {
-    for (int i = 0; i < length; i++) {
-      int leftIndex = leftStart + i;
-      int rightIndex = rightStart + i;
+  private boolean float8ApproxEquals(Range range) {
+    Float8Vector leftVector  = (Float8Vector) getLeft();
+    Float8Vector rightVector  = (Float8Vector) getRight();
+
+    for (int i = 0; i < range.getLength(); i++) {
+      int leftIndex = range.getLeftStart() + i;
+      int rightIndex = range.getRightStart() + i;
 
-      boolean isNull = left.isNull(leftIndex);
+      boolean isNull = leftVector.isNull(leftIndex);
 
-      if (isNull != right.isNull(rightIndex)) {
+      if (isNull != rightVector.isNull(rightIndex)) {
         return false;
       }
 
       if (!isNull) {
 
-        Double leftValue = left.get(leftIndex);
-        Double rightValue = ((Float8Vector)right).get(rightIndex);
+        Double leftValue = leftVector.get(leftIndex);
+        Double rightValue = rightVector.get(rightIndex);
         if (leftValue.isNaN()) {
           return rightValue.isNaN();
         }

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

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.vector.compare;
+
+/**
+ * Wrapper for the parameters of comparing a range of values in two vectors.
+ */
+public class Range {
+
+  /**
+   * Start position in the left vector.
+   */
+  private int leftStart = -1;
+
+  /**
+   * Start position in the right vector.
+   */
+  private int rightStart = -1;
+
+  /**
+   * Length of the range.
+   */
+  private int length = -1;
+
+
+  /**
+   * Constructs a new instance.
+   */
+  public Range() {}
+
+  /**
+   * Constructs a new instance.
+   *
+   * @param leftStart start index in left vector
+   * @param rightStart start index in right vector
+   * @param length length of range
+   */
+  public Range(int leftStart, int rightStart, int length) {
+    this.leftStart = leftStart;
+    this.rightStart = rightStart;
+    this.length = length;
+  }
+
+  public int getLeftStart() {
+    return leftStart;
+  }
+
+  public int getRightStart() {
+    return rightStart;
+  }
+
+  public int getLength() {
+    return length;
+  }
+
+  public Range setLeftStart(int leftStart) {
+    this.leftStart = leftStart;
+    return this;
+  }
+
+  public Range setRightStart(int rightStart) {
+    this.rightStart = rightStart;
+    return this;
+  }
+
+  public Range setLength(int length) {
+    this.length = length;
+    return this;
+  }
+}