GH-62: Optimize Flight SQL JDBC Statement execution paths

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

@@ -19,6 +19,8 @@
 import static org.apache.arrow.driver.jdbc.utils.ArrowFlightConnectionConfigImpl.ArrowFlightConnectionProperty.replaceSemiColons;
 
 import io.netty.util.concurrent.DefaultThreadFactory;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -257,4 +259,35 @@ BufferAllocator getBufferAllocator() {
   public ArrowFlightMetaImpl getMeta() {
     return (ArrowFlightMetaImpl) this.meta;
   }
+
+  @Override
+  public PreparedStatement prepareStatement(final String sql) throws SQLException {
+    checkOpen();
+    return prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
+  }
+
+  @Override
+  public PreparedStatement prepareStatement(
+      final String sql, final int resultSetType, final int resultSetConcurrency)
+      throws SQLException {
+    checkOpen();
+    return prepareStatement(sql, resultSetType, resultSetConcurrency, getHoldability());
+  }
+
+  @Override
+  public PreparedStatement prepareStatement(
+      final String sql,
+      final int resultSetType,
+      final int resultSetConcurrency,
+      final int resultSetHoldability)
+      throws SQLException {
+    checkOpen();
+    return ArrowFlightPreparedStatement.builder(this)
+        .withQuery(sql)
+        .withGeneratedHandle()
+        .withResultSetType(resultSetType)
+        .withResultSetConcurrency(resultSetConcurrency)
+        .withResultSetHoldability(resultSetHoldability)
+        .build();
+  }
 }

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

@@ -20,7 +20,6 @@
 import java.sql.SQLException;
 import java.util.Properties;
 import java.util.TimeZone;
-import org.apache.arrow.driver.jdbc.client.ArrowFlightSqlClientHandler;
 import org.apache.arrow.memory.RootAllocator;
 import org.apache.calcite.avatica.AvaticaConnection;
 import org.apache.calcite.avatica.AvaticaFactory;
@@ -79,20 +78,20 @@ public ArrowFlightPreparedStatement newPreparedStatement(
       final Meta.Signature signature,
       final int resultType,
       final int resultSetConcurrency,
-      final int resultSetHoldability)
-      throws SQLException {
+      final int resultSetHoldability) {
     final ArrowFlightConnection flightConnection = (ArrowFlightConnection) connection;
-    ArrowFlightSqlClientHandler.PreparedStatement preparedStatement =
-        flightConnection.getMeta().getPreparedStatement(statementHandle);
+    final AvaticaStatement existingStatement =
+        flightConnection.statementMap.get(statementHandle.id);
+    if (existingStatement instanceof ArrowFlightPreparedStatement) {
+      return (ArrowFlightPreparedStatement) existingStatement;
+    }
+    if (existingStatement != null) {
+      throw new IllegalStateException(
+          "Unexpected statement type found for prepared statement handle: " + statementHandle);
+    }
 
-    return ArrowFlightPreparedStatement.newPreparedStatement(
-        flightConnection,
-        preparedStatement,
-        statementHandle,
-        signature,
-        resultType,
-        resultSetConcurrency,
-        resultSetHoldability);
+    throw new IllegalStateException(
+        "PreparedStatement was not pre-created for handle: " + statementHandle);
   }
 
   @Override

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

@@ -67,7 +67,7 @@ public final class ArrowFlightJdbcFlightStreamResultSet
       throws SQLException {
     super(statement, state, signature, resultSetMetaData, timeZone, firstFrame);
     this.connection = (ArrowFlightConnection) statement.connection;
-    this.flightInfo = ((ArrowFlightInfoStatement) statement).executeFlightInfoQuery();
+    this.flightInfo = ((ArrowFlightMetaStatement) statement).executeFlightInfoQuery();
   }
 
   /** Private constructor for fromFlightInfo. */
@@ -106,7 +106,7 @@ static ArrowFlightJdbcFlightStreamResultSet fromFlightInfo(
     final TimeZone timeZone = TimeZone.getDefault();
     final QueryState state = new QueryState();
 
-    final Meta.Signature signature = ArrowFlightMetaImpl.newSignature(null, null, null);
+    final Meta.Signature signature = ArrowFlightMetaImpl.buildDefaultSignature();
 
     final AvaticaResultSetMetaData resultSetMetaData =
         new AvaticaResultSetMetaData(null, null, signature);

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

@@ -73,7 +73,7 @@ public static ArrowFlightJdbcVectorSchemaRootResultSet fromVectorSchemaRoot(
     final TimeZone timeZone = TimeZone.getDefault();
     final QueryState state = new QueryState();
 
-    final Meta.Signature signature = ArrowFlightMetaImpl.newSignature(null, null, null);
+    final Meta.Signature signature = ArrowFlightMetaImpl.buildDefaultSignature();
 
     final AvaticaResultSetMetaData resultSetMetaData =
         new AvaticaResultSetMetaData(null, null, signature);