Validate long dictionary indices before narrowing

Signed-off-by: Efe Gökdemir <gokdemirefe1903@gmail.com>

Author: efegokdemir

vector/src/main/java/org/apache/arrow/vector/dictionary/DictionaryEncoder.java

@@ -164,12 +164,12 @@ static void retrieveIndexVector(
       BaseIntVector indices, TransferPair transfer, int dictionaryCount, int start, int end) {
     for (int i = start; i < end; i++) {
       if (!indices.isNull(i)) {
-        int indexAsInt = (int) indices.getValueAsLong(i);
-        if (indexAsInt < 0 || indexAsInt >= dictionaryCount) {
+        long index = indices.getValueAsLong(i);
+        if (index < 0 || index >= dictionaryCount) {
           throw new IllegalArgumentException(
-              "Provided dictionary does not contain value for index " + indexAsInt);
+              "Provided dictionary does not contain value for index " + index);
         }
-        transfer.copyValueSafe(indexAsInt, i);
+        transfer.copyValueSafe((int) index, i);
       }
     }
   }

vector/src/test/java/org/apache/arrow/vector/TestDictionaryVector.java

@@ -972,6 +972,32 @@ public void testDecodeRejectsDictionaryIndicesOutsideBounds() {
     assertEquals(0, allocator.getAllocatedMemory(), "decode memory leak");
   }
 
+  @Test
+  public void testDecodeRejectsBigIntDictionaryIndexOutsideBounds() {
+    try (final BigIntVector indices = new BigIntVector("indices", allocator);
+        final VarCharVector dictionaryVector = newVarCharVector("dict", allocator)) {
+      setVector(dictionaryVector, zero, one);
+      Dictionary dictionary =
+          new Dictionary(dictionaryVector, new DictionaryEncoding(1L, false, null));
+
+      setVector(indices, 1L);
+      try (ValueVector decoded = DictionaryEncoder.decode(indices, dictionary, allocator)) {
+        assertEquals(new Text("bar"), decoded.getObject(0));
+      }
+
+      long largeIndex = 1L << 32;
+      setVector(indices, largeIndex);
+      IllegalArgumentException exception =
+          assertThrows(
+              IllegalArgumentException.class,
+              () -> DictionaryEncoder.decode(indices, dictionary, allocator));
+      assertEquals(
+          "Provided dictionary does not contain value for index " + largeIndex,
+          exception.getMessage());
+    }
+    assertEquals(0, allocator.getAllocatedMemory(), "decode memory leak");
+  }
+
   @Test
   public void testListNoMemoryLeak() {
     // Create a new value vector