GH-52: Make RangeEqualsVisitor of RunEndEncodedVector more efficient

Author: ViggoC

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

@@ -43,6 +43,7 @@
 import org.apache.arrow.vector.complex.ListViewVector;
 import org.apache.arrow.vector.complex.NonNullableStructVector;
 import org.apache.arrow.vector.complex.RunEndEncodedVector;
+import org.apache.arrow.vector.complex.RunEndEncodedVector.RangeIterator;
 import org.apache.arrow.vector.complex.StructVector;
 import org.apache.arrow.vector.complex.UnionVector;
 
@@ -270,35 +271,24 @@ protected boolean compareRunEndEncodedVectors(Range range) {
     RunEndEncodedVector leftVector = (RunEndEncodedVector) left;
     RunEndEncodedVector rightVector = (RunEndEncodedVector) right;
 
-    final int leftRangeEnd = range.getLeftStart() + range.getLength();
-    final int rightRangeEnd = range.getRightStart() + range.getLength();
+    final RunEndEncodedVector.RangeIterator leftIterator =
+        new RangeIterator(leftVector, range.getLeftStart(), range.getLength());
+    final RunEndEncodedVector.RangeIterator rightIterator =
+        new RangeIterator(rightVector, range.getRightStart(), range.getLength());
 
     FieldVector leftValuesVector = leftVector.getValuesVector();
     FieldVector rightValuesVector = rightVector.getValuesVector();
 
     RangeEqualsVisitor innerVisitor = createInnerVisitor(leftValuesVector, rightValuesVector, null);
 
-    int leftLogicalIndex = range.getLeftStart();
-    int rightLogicalIndex = range.getRightStart();
+    while (leftIterator.nextRun() | rightIterator.nextRun()) {
+      int leftPhysicalIndex = leftIterator.getRunIndex();
+      int rightPhysicalIndex = rightIterator.getRunIndex();
 
-    while (leftLogicalIndex < leftRangeEnd) {
-      // TODO: implement it more efficient
-      // https://github.com/apache/arrow/issues/44157
-      int leftPhysicalIndex = leftVector.getPhysicalIndex(leftLogicalIndex);
-      int rightPhysicalIndex = rightVector.getPhysicalIndex(rightLogicalIndex);
       if (leftValuesVector.accept(
           innerVisitor, new Range(leftPhysicalIndex, rightPhysicalIndex, 1))) {
-        int leftRunEnd = leftVector.getRunEnd(leftLogicalIndex);
-        int rightRunEnd = rightVector.getRunEnd(rightLogicalIndex);
-
-        int leftRunLength = Math.min(leftRunEnd, leftRangeEnd) - leftLogicalIndex;
-        int rightRunLength = Math.min(rightRunEnd, rightRangeEnd) - rightLogicalIndex;
-
-        if (leftRunLength != rightRunLength) {
+        if (leftIterator.getRunLength() != rightIterator.getRunLength()) {
           return false;
-        } else {
-          leftLogicalIndex = leftRunEnd;
-          rightLogicalIndex = rightRunEnd;
         }
       } else {
         return false;

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

@@ -820,4 +820,54 @@ static int getPhysicalIndex(FieldVector runEndVector, int logicalIndex) {
 
     return result;
   }
+
+  public static class RangeIterator {
+
+    private final RunEndEncodedVector runEndEncodedVector;
+    private final int rangeEnd;
+    private int runIndex;
+    private int runEnd;
+    private int logicalPos;
+
+    public RangeIterator(RunEndEncodedVector runEndEncodedVector, int startIndex, int length) {
+      this.runEndEncodedVector = runEndEncodedVector;
+      this.rangeEnd = startIndex + length;
+      this.runIndex = runEndEncodedVector.getPhysicalIndex(startIndex) - 1;
+      this.runEnd = startIndex;
+      this.logicalPos = -1;
+    }
+
+    public boolean nextRun() {
+      logicalPos = runEnd;
+      if (logicalPos >= rangeEnd) {
+        return false;
+      }
+      updateRun();
+      return true;
+    }
+
+    private void updateRun() {
+      runIndex++;
+      runEnd = (int) ((BaseIntVector) runEndEncodedVector.runEndsVector).getValueAsLong(runIndex);
+    }
+
+    public boolean nextValue() {
+      logicalPos++;
+      if (logicalPos >= rangeEnd) {
+        return false;
+      }
+      if (logicalPos == runEnd) {
+        updateRun();
+      }
+      return true;
+    }
+
+    public int getRunIndex() {
+      return runIndex;
+    }
+
+    public int getRunLength() {
+      return Math.min(runEnd, rangeEnd) - logicalPos;
+    }
+  }
 }