GH-964: Fix IndexOutOfBoundsException in Array.getResultSet() for JDBC clients (#965)

## What's Changed

- Fixed JDBC specification in ArrowFlightJdbcArray.getResultSet() that
caused IndexOutOfBoundsException in JDBC clients like DBeaver when
reading array columns
- The method returned a single-column ResultSet containing only array
values, but JDBC spec requires a 2-column format
- Not it returns two columns:
  - Column 1 (INDEX): 1-based element indices per JDBC specification
  - Column 2: The actual array element values


Closes #964.

Author: xborder

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

@@ -26,6 +26,7 @@
 import org.apache.arrow.driver.jdbc.utils.SqlTypes;
 import org.apache.arrow.memory.util.LargeMemoryUtil;
 import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.IntVector;
 import org.apache.arrow.vector.ValueVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.types.pojo.ArrowType;
@@ -135,12 +136,22 @@ public ResultSet getResultSet(long index, int count) throws SQLException {
 
   private static ResultSet getResultSetNoBoundariesCheck(
       ValueVector dataVector, long start, long count) throws SQLException {
+    int intStart = LargeMemoryUtil.checkedCastToInt(start);
+    int intCount = LargeMemoryUtil.checkedCastToInt(count);
+
+    // Create an index vector with 1-based indices (per JDBC spec) to return with value vector
+    IntVector indexVector = new IntVector("INDEX", dataVector.getAllocator());
+    indexVector.allocateNew(intCount);
+    for (int i = 0; i < intCount; i++) {
+      indexVector.set(i, i + 1);
+    }
+    indexVector.setValueCount(intCount);
+
     TransferPair transferPair = dataVector.getTransferPair(dataVector.getAllocator());
-    transferPair.splitAndTransfer(
-        LargeMemoryUtil.checkedCastToInt(start), LargeMemoryUtil.checkedCastToInt(count));
-    FieldVector vectorSlice = (FieldVector) transferPair.getTo();
+    transferPair.splitAndTransfer(intStart, intCount);
+    FieldVector valueVector = (FieldVector) transferPair.getTo();
 
-    VectorSchemaRoot vectorSchemaRoot = VectorSchemaRoot.of(vectorSlice);
+    VectorSchemaRoot vectorSchemaRoot = VectorSchemaRoot.of(indexVector, valueVector);
     return ArrowFlightJdbcVectorSchemaRootResultSet.fromVectorSchemaRoot(vectorSchemaRoot);
   }
 

flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/ArrowFlightJdbcArrayTest.java

@@ -129,7 +129,7 @@ public void testShouldGetResultSetReturnValidResultSet() throws SQLException {
     try (ResultSet resultSet = arrowFlightJdbcArray.getResultSet()) {
       int count = 0;
       while (resultSet.next()) {
-        assertEquals((Object) resultSet.getInt(1), dataVector.getObject(count));
+        assertEquals((Object) resultSet.getInt(2), dataVector.getObject(count));
         count++;
       }
     }
@@ -142,7 +142,7 @@ public void testShouldGetResultSetReturnValidResultSetWithOffsets() throws SQLEx
     try (ResultSet resultSet = arrowFlightJdbcArray.getResultSet(3, 5)) {
       int count = 0;
       while (resultSet.next()) {
-        assertEquals((Object) resultSet.getInt(1), dataVector.getObject(count + 3));
+        assertEquals((Object) resultSet.getInt(2), dataVector.getObject(count + 3));
         count++;
       }
       assertEquals(5, count);

flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/AbstractArrowFlightJdbcListAccessorTest.java

@@ -191,7 +191,12 @@ public void testShouldGetArrayGetResultSetReturnValidResultSet(
           try (ResultSet rs = array.getResultSet()) {
             int count = 0;
             while (rs.next()) {
-              final int value = rs.getInt(1);
+              // Column 1: 1-based index (per JDBC spec)
+              final int index = rs.getInt(1);
+              assertThat(index, equalTo(count + 1));
+
+              // Column 2: actual value (per JDBC spec)
+              final int value = rs.getInt(2);
               assertThat(value, equalTo(currentRow * count));
               count++;
             }

flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/accessor/impl/complex/ArrowFlightJdbcMapVectorAccessorTest.java

@@ -153,15 +153,15 @@ public void testShouldGetArrayReturnValidArray() throws SQLException {
 
     try (ResultSet resultSet = array.getResultSet()) {
       assertTrue(resultSet.next());
-      Map<?, ?> entry = resultSet.getObject(1, Map.class);
+      Map<?, ?> entry = resultSet.getObject(2, Map.class);
       assertEquals(1, entry.get("key"));
       assertEquals(11, entry.get("value"));
       assertTrue(resultSet.next());
-      entry = resultSet.getObject(1, Map.class);
+      entry = resultSet.getObject(2, Map.class);
       assertEquals(2, entry.get("key"));
       assertEquals(22, entry.get("value"));
       assertTrue(resultSet.next());
-      entry = resultSet.getObject(1, Map.class);
+      entry = resultSet.getObject(2, Map.class);
       assertEquals(3, entry.get("key"));
       assertEquals(33, entry.get("value"));
       assertFalse(resultSet.next());
@@ -173,7 +173,7 @@ public void testShouldGetArrayReturnValidArray() throws SQLException {
     assertFalse(accessor.wasNull());
     try (ResultSet resultSet = array.getResultSet()) {
       assertTrue(resultSet.next());
-      Map<?, ?> entry = resultSet.getObject(1, Map.class);
+      Map<?, ?> entry = resultSet.getObject(2, Map.class);
       assertEquals(2, entry.get("key"));
       assertNull(entry.get("value"));
       assertFalse(resultSet.next());
@@ -185,19 +185,19 @@ public void testShouldGetArrayReturnValidArray() throws SQLException {
     assertFalse(accessor.wasNull());
     try (ResultSet resultSet = array.getResultSet()) {
       assertTrue(resultSet.next());
-      Map<?, ?> entry = resultSet.getObject(1, Map.class);
+      Map<?, ?> entry = resultSet.getObject(2, Map.class);
       assertEquals(0, entry.get("key"));
       assertEquals(2000, entry.get("value"));
       assertTrue(resultSet.next());
-      entry = resultSet.getObject(1, Map.class);
+      entry = resultSet.getObject(2, Map.class);
       assertEquals(1, entry.get("key"));
       assertEquals(2001, entry.get("value"));
       assertTrue(resultSet.next());
-      entry = resultSet.getObject(1, Map.class);
+      entry = resultSet.getObject(2, Map.class);
       assertEquals(2, entry.get("key"));
       assertEquals(2002, entry.get("value"));
       assertTrue(resultSet.next());
-      entry = resultSet.getObject(1, Map.class);
+      entry = resultSet.getObject(2, Map.class);
       assertEquals(3, entry.get("key"));
       assertEquals(2003, entry.get("value"));
       assertFalse(resultSet.next());