GH-39982: [Java] Add RunEndEncodedVector

java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/binder/ColumnBinderArrowTypeVisitor.java

@@ -96,6 +96,11 @@ public ColumnBinder visit(ArrowType.Union type) {
     throw new UnsupportedOperationException("No column binder implemented for type " + type);
   }
 
+  @Override
+  public ColumnBinder visit(ArrowType.RunEndEncoded type) {
+    throw new UnsupportedOperationException("No column binder implemented for type " + type);
+  }
+
   @Override
   public ColumnBinder visit(ArrowType.Map type) {
     return new MapBinder((MapVector) vector);

java/c/src/main/java/org/apache/arrow/c/BufferImportTypeVisitor.java

@@ -185,6 +185,11 @@ public List<ArrowBuf> visit(ArrowType.Union type) {
     }
   }
 
+  @Override
+  public List<ArrowBuf> visit(ArrowType.RunEndEncoded type) {
+    throw new UnsupportedOperationException("Importing buffers for type: " + type);
+  }
+
   @Override
   public List<ArrowBuf> visit(ArrowType.Map type) {
     return Arrays.asList(maybeImportBitmap(type), importOffsets(type, MapVector.OFFSET_WIDTH));

java/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/AvaticaParameterBinder.java

@@ -281,5 +281,11 @@ public Boolean visit(ArrowType.ListView type) {
     public Boolean visit(ArrowType.LargeListView type) {
       throw new UnsupportedOperationException("Binding is not yet supported for type " + type);
     }
+
+    @Override
+    public Boolean visit(ArrowType.RunEndEncoded type) {
+      throw new UnsupportedOperationException(
+          "No Avatica parameter binder implemented for type " + type);
+    }
   }
 }

java/flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/utils/ConvertUtils.java

@@ -284,5 +284,11 @@ public AvaticaParameter visit(ArrowType.LargeListView type) {
       throw new UnsupportedOperationException(
           "AvaticaParameter not yet supported for type " + type);
     }
+
+    @Override
+    public AvaticaParameter visit(ArrowType.RunEndEncoded type) {
+      throw new UnsupportedOperationException(
+          "No Avatica parameter binder implemented for type " + type);
+    }
   }
 }

java/vector/src/main/codegen/data/ArrowTypes.tdd

@@ -139,6 +139,11 @@
        name: "LargeListView",
        fields: [],
        complex: true
+    },
+    {
+       name: "RunEndEncoded",
+       fields: [],
+       complex: true
     }
   ]
 }

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

@@ -39,7 +39,7 @@
 @SuppressWarnings("unused")
 public class UnionReader extends AbstractFieldReader {
 
-  private static final int NUM_SUPPORTED_TYPES = 50;
+  private static final int NUM_SUPPORTED_TYPES = 51;
 
   private BaseReader[] readers = new BaseReader[NUM_SUPPORTED_TYPES];
   public UnionVector data;

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

@@ -40,6 +40,7 @@
 import org.apache.arrow.vector.types.pojo.ArrowType.LargeUtf8;
 import org.apache.arrow.vector.types.pojo.ArrowType.Map;
 import org.apache.arrow.vector.types.pojo.ArrowType.Null;
+import org.apache.arrow.vector.types.pojo.ArrowType.RunEndEncoded;
 import org.apache.arrow.vector.types.pojo.ArrowType.Struct;
 import org.apache.arrow.vector.types.pojo.ArrowType.Time;
 import org.apache.arrow.vector.types.pojo.ArrowType.Timestamp;
@@ -280,6 +281,11 @@ public TypeLayout visit(Interval type) {
               public TypeLayout visit(Duration type) {
                 return newFixedWidthTypeLayout(BufferLayout.dataBuffer(64));
               }
+
+              @Override
+              public TypeLayout visit(RunEndEncoded type) {
+                return new TypeLayout(Collections.<BufferLayout>emptyList());
+              }
             });
     return layout;
   }
@@ -444,6 +450,11 @@ public Integer visit(Interval type) {
           public Integer visit(Duration type) {
             return FIXED_WIDTH_BUFFER_COUNT;
           }
+
+          @Override
+          public Integer visit(RunEndEncoded type) {
+            return 0;
+          }
         });
   }
 

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

@@ -29,6 +29,7 @@
 import org.apache.arrow.vector.BaseVariableWidthViewVector;
 import org.apache.arrow.vector.BitVector;
 import org.apache.arrow.vector.ExtensionTypeVector;
+import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.NullVector;
 import org.apache.arrow.vector.ValueVector;
 import org.apache.arrow.vector.complex.BaseLargeRepeatedValueViewVector;
@@ -41,11 +42,13 @@
 import org.apache.arrow.vector.complex.ListVector;
 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.StructVector;
 import org.apache.arrow.vector.complex.UnionVector;
 
 /** Visitor to compare a range of values for vectors. */
 public class RangeEqualsVisitor implements VectorVisitor<Boolean, Range> {
+
   private ValueVector left;
   private ValueVector right;
 
@@ -226,6 +229,14 @@ public Boolean visit(NullVector left, Range range) {
     return true;
   }
 
+  @Override
+  public Boolean visit(RunEndEncodedVector left, Range range) {
+    if (!validate(left)) {
+      return false;
+    }
+    return compareRunEndEncodedVectors(range);
+  }
+
   @Override
   public Boolean visit(ExtensionTypeVector<?> left, Range range) {
     if (!(right instanceof ExtensionTypeVector<?>) || !validate(left)) {
@@ -255,6 +266,48 @@ public Boolean visit(LargeListViewVector left, Range range) {
     return compareLargeListViewVectors(range);
   }
 
+  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();
+
+    FieldVector leftValuesVector = leftVector.getValuesVector();
+    FieldVector rightValuesVector = rightVector.getValuesVector();
+
+    RangeEqualsVisitor innerVisitor = createInnerVisitor(leftValuesVector, rightValuesVector, null);
+
+    int leftLogicalIndex = range.getLeftStart();
+    int rightLogicalIndex = range.getRightStart();
+
+    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) {
+          return false;
+        } else {
+          leftLogicalIndex = leftRunEnd;
+          rightLogicalIndex = rightRunEnd;
+        }
+      } else {
+        return false;
+      }
+    }
+
+    return true;
+  }
+
   protected RangeEqualsVisitor createInnerVisitor(
       ValueVector leftInner,
       ValueVector rightInner,

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

@@ -32,6 +32,7 @@
 import org.apache.arrow.vector.complex.ListVector;
 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.UnionVector;
 import org.apache.arrow.vector.types.pojo.Field;
 
@@ -136,6 +137,11 @@ public Boolean visit(LargeListViewVector left, Void value) {
     return compareField(left.getField(), right.getField());
   }
 
+  @Override
+  public Boolean visit(RunEndEncodedVector left, Void value) {
+    return compareField(left.getField(), right.getField());
+  }
+
   private boolean compareField(Field leftField, Field rightField) {
 
     if (leftField == rightField) {

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

@@ -29,6 +29,7 @@
 import org.apache.arrow.vector.complex.ListVector;
 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.UnionVector;
 
 /**
@@ -71,4 +72,9 @@ default OUT visit(LargeListViewVector left, IN value) {
     throw new UnsupportedOperationException(
         "VectorVisitor for LargeListViewVector is not supported.");
   }
+
+  default OUT visit(RunEndEncodedVector left, IN value) {
+    throw new UnsupportedOperationException(
+        "VectorVisitor for LargeListViewVector is not supported.");
+  };
 }