GH-39355: [Java] Improve JdbcConsumer exceptions (#39356)

### Rationale for this change

This helps debug Arrow conversion errors from JDBC by exposing the JdbcFieldInfo for the related column and the ArrowType for the corresponding vector.

### What changes are included in this PR?

A new JdbcConsumerException which is thrown by the CompositeJdbcConsumer while consuming data for a specific vector.

### Are these changes tested?

N/A

### Are there any user-facing changes?

Users can now catch `JdbcConsumerException`s and extract the related ArrowType and JdbcFieldInfo from it for debugging.

* Closes: #39355

Authored-by: Diego Fernandez <aiguo.fernandez@gmail.com>
Signed-off-by: David Li <li.davidm96@gmail.com>

java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/ArrowVectorIterator.java

@@ -26,6 +26,7 @@
 
 import org.apache.arrow.adapter.jdbc.consumer.CompositeJdbcConsumer;
 import org.apache.arrow.adapter.jdbc.consumer.JdbcConsumer;
+import org.apache.arrow.adapter.jdbc.consumer.exceptions.JdbcConsumerException;
 import org.apache.arrow.util.AutoCloseables;
 import org.apache.arrow.util.Preconditions;
 import org.apache.arrow.vector.FieldVector;
@@ -114,7 +115,11 @@ private void consumeData(VectorSchemaRoot root) {
       root.setRowCount(readRowCount);
     } catch (Throwable e) {
       compositeConsumer.close();
-      throw new RuntimeException("Error occurred while consuming data.", e);
+      if (e instanceof JdbcConsumerException) {
+        throw (JdbcConsumerException) e;
+      } else {
+        throw new RuntimeException("Error occurred while consuming data.", e);
+      }
     }
   }
 
@@ -168,6 +173,7 @@ public boolean hasNext() {
    * Gets the next vector.
    * If {@link JdbcToArrowConfig#isReuseVectorSchemaRoot()} is false,
    * the client is responsible for freeing its resources.
+   * @throws JdbcConsumerException on error from VectorConsumer
    */
   @Override
   public VectorSchemaRoot next() {
@@ -178,7 +184,11 @@ public VectorSchemaRoot next() {
       return ret;
     } catch (Exception e) {
       close();
-      throw new RuntimeException("Error occurred while getting next schema root.", e);
+      if (e instanceof JdbcConsumerException) {
+        throw (JdbcConsumerException) e;
+      } else {
+        throw new RuntimeException("Error occurred while getting next schema root.", e);
+      }
     }
   }
 

java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/JdbcToArrowUtils.java

@@ -58,6 +58,7 @@
 import org.apache.arrow.adapter.jdbc.consumer.TimestampTZConsumer;
 import org.apache.arrow.adapter.jdbc.consumer.TinyIntConsumer;
 import org.apache.arrow.adapter.jdbc.consumer.VarCharConsumer;
+import org.apache.arrow.adapter.jdbc.consumer.exceptions.JdbcConsumerException;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.arrow.util.Preconditions;
 import org.apache.arrow.vector.BigIntVector;
@@ -386,6 +387,7 @@ static boolean isColumnNullable(ResultSetMetaData resultSetMetadata, int index,
    * @param root   Arrow {@link VectorSchemaRoot} object to populate
    * @param config The configuration to use when reading the data.
    * @throws SQLException on error
+   * @throws JdbcConsumerException on error from VectorConsumer
    */
   public static void jdbcToArrowVectors(ResultSet rs, VectorSchemaRoot root, JdbcToArrowConfig config)
       throws SQLException, IOException {

java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/CompositeJdbcConsumer.java

@@ -21,9 +21,12 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 
+import org.apache.arrow.adapter.jdbc.JdbcFieldInfo;
+import org.apache.arrow.adapter.jdbc.consumer.exceptions.JdbcConsumerException;
 import org.apache.arrow.util.AutoCloseables;
 import org.apache.arrow.vector.ValueVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.ArrowType;
 
 /**
  * Composite consumer which hold all consumers.
@@ -43,7 +46,18 @@ public CompositeJdbcConsumer(JdbcConsumer[] consumers) {
   @Override
   public void consume(ResultSet rs) throws SQLException, IOException {
     for (int i = 0; i < consumers.length; i++) {
-      consumers[i].consume(rs);
+      try {
+        consumers[i].consume(rs);
+      } catch (Exception e) {
+        if (consumers[i] instanceof BaseConsumer) {
+          BaseConsumer consumer = (BaseConsumer) consumers[i];
+          JdbcFieldInfo fieldInfo = new JdbcFieldInfo(rs.getMetaData(), consumer.columnIndexInResultSet);
+          ArrowType arrowType = consumer.vector.getMinorType().getType();
+          throw new JdbcConsumerException("Exception while consuming JDBC value", e, fieldInfo, arrowType);
+        } else {
+          throw e;
+        }
+      }
     }
   }
 
@@ -74,3 +88,4 @@ public void resetVectorSchemaRoot(VectorSchemaRoot root) {
     }
   }
 }
+

java/adapter/jdbc/src/main/java/org/apache/arrow/adapter/jdbc/consumer/exceptions/JdbcConsumerException.java

@@ -0,0 +1,52 @@
+/*
+ * 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.adapter.jdbc.consumer.exceptions;
+
+import org.apache.arrow.adapter.jdbc.JdbcFieldInfo;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+
+/**
+ * Exception while consuming JDBC data. This exception stores the JdbcFieldInfo for the column and the
+ * ArrowType for the corresponding vector for easier debugging.
+ */
+public class JdbcConsumerException extends RuntimeException {
+  final JdbcFieldInfo fieldInfo;
+  final ArrowType arrowType;
+
+  /**
+   * Construct JdbcConsumerException with all fields.
+   *
+   * @param message   error message
+   * @param cause     original exception
+   * @param fieldInfo JdbcFieldInfo for the column
+   * @param arrowType ArrowType for the corresponding vector
+   */
+  public JdbcConsumerException(String message, Throwable cause, JdbcFieldInfo fieldInfo, ArrowType arrowType) {
+    super(message, cause);
+    this.fieldInfo = fieldInfo;
+    this.arrowType = arrowType;
+  }
+
+  public ArrowType getArrowType() {
+    return this.arrowType;
+  }
+
+  public JdbcFieldInfo getFieldInfo() {
+    return this.fieldInfo;
+  }
+}