[GitHub] [arrow] lidavidm commented on a change in pull request #10906: ARROW-12922: [Java] Add flight-sql to the flight package.

[GitBox <gi...@apache.org>]

lidavidm commented on a change in pull request #10906:
URL: https://github.com/apache/arrow/pull/10906#discussion_r686805454



##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.
+   */
+  repeated uint32 info = 1;
+}
+
+/*
+ * Represents a request to retrieve the list of catalogs on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name.
+ */
+message CommandGetCatalogs {
+}
+
+/*
+ * Represents a request to retrieve the list of schemas on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, then schema_name.
+ */
+message CommandGetSchemas {
+  /*
+   * Specifies the Catalog to search for schemas.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, the pattern will not be used to narrow the search.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+}
+
+/*
+ * Represents a request to retrieve the list of tables, and optionally their schemas, on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  table_type: utf8,
+ *  table_schema: bytes

Review comment:
       We should clarify the encoding of the schema (an IPC-encapsulated Flatbuffer, presumably)

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.
+   */
+  repeated uint32 info = 1;
+}
+
+/*
+ * Represents a request to retrieve the list of catalogs on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name.
+ */
+message CommandGetCatalogs {
+}
+
+/*
+ * Represents a request to retrieve the list of schemas on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, then schema_name.
+ */
+message CommandGetSchemas {
+  /*
+   * Specifies the Catalog to search for schemas.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, the pattern will not be used to narrow the search.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+}
+
+/*
+ * Represents a request to retrieve the list of tables, and optionally their schemas, on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  table_type: utf8,
+ *  table_schema: bytes
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type.
+ */
+message CommandGetTables {
+  /*
+   * Specifies the Catalog to search for the tables.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, all schemas matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+
+  /*
+   * Specifies a filter pattern for tables to search for.
+   * When no table_name_filter_pattern is provided, all tables matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue table_name_filter_pattern = 3;
+
+  // Specifies a filter of table types which must match.
+  repeated string table_types = 4;
+
+  // Specifies if the schema should be returned for found tables.
+  bool include_schema = 5;
+}
+
+/*
+ * Represents a request to retrieve the list of table types on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  table_type: utf8
+ * >
+ * The returned data should be ordered by table_type.
+ */
+message CommandGetTableTypes {
+}
+
+/*
+ * Represents a request to retrieve the primary keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  column_name: utf8,
+ *  key_sequence: int,
+ *  key_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetPrimaryKeys {
+  // Specifies the catalog to search for the table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the table to get the primary keys for.
+  google.protobuf.StringValue table = 3;
+}
+
+/*
+ * Represents a request to retrieve a description of the foreign key columns that reference the given table's
+ * primary key columns (the foreign keys exported by a table) of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetExportedKeys {
+  // Specifies the catalog to search for the foreign key table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the foreign key table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the foreign key table to get the foreign keys for.
+  string table = 3;
+}
+
+/*
+ * Represents a request to retrieve the foreign keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetImportedKeys {
+  // Specifies the catalog to search for the primary key table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the primary key table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the primary key table to get the foreign keys for.
+  string table = 3;
+}
+
+// SQL Execution Action Messages
+
+/*
+ * Request message for the "GetPreparedStatement" action on a Flight SQL enabled backend.
+ */
+message ActionCreatePreparedStatementRequest {
+  // The valid SQL string to create a prepared statement for.
+  string query = 1;
+}
+
+/*
+ * Wrap the result of a "GetPreparedStatement" action.
+ */
+message ActionCreatePreparedStatementResult {
+  // Opaque handle for the prepared statement on the server.
+  bytes prepared_statement_handle = 1;
+
+  // If a result set generating query was provided, dataset_schema contains the 
+  // schema of the dataset as described in Schema.fbs::Schema, it is serialized as an IPC message.
+  bytes dataset_schema = 2;
+
+  // If the query provided contained parameters, parameter_schema contains the 
+  // Schema of the expected parameters as described in Schema.fbs::Schema.

Review comment:
       We should also be consistent in how we describe how schemas are encoded.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;

Review comment:
       Note that this was never standardized. I don't think we should use it, here or in the other files.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java
##########
@@ -0,0 +1,704 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.ActionType;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightProducer;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.UnionMode;
+import org.apache.arrow.vector.types.pojo.ArrowType.Union;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import io.grpc.Status;
+
+/**
+ * API to Implement an Arrow Flight SQL producer.
+ */
+public interface FlightSqlProducer extends FlightProducer, AutoCloseable {
+  /**
+   * Depending on the provided command, method either:
+   * 1. Return information about a SQL query, or
+   * 2. Return information about a prepared statement. In this case, parameters binding is allowed.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return information about the given SQL query, or the given prepared statement.
+   */
+  @Override
+  default FlightInfo getFlightInfo(CallContext context, FlightDescriptor descriptor) {
+    final Any command = FlightSqlUtils.parseOrThrow(descriptor.getCommand());
+
+    if (command.is(CommandStatementQuery.class)) {
+      return getFlightInfoStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      return getFlightInfoPreparedStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      return getFlightInfoCatalogs(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetCatalogs.class), context, descriptor);
+    } else if (command.is(CommandGetSchemas.class)) {
+      return getFlightInfoSchemas(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetSchemas.class), context, descriptor);
+    } else if (command.is(CommandGetTables.class)) {
+      return getFlightInfoTables(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetTables.class), context, descriptor);
+    } else if (command.is(CommandGetTableTypes.class)) {
+      return getFlightInfoTableTypes(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetTableTypes.class), context, descriptor);
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      return getFlightInfoSqlInfo(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetSqlInfo.class), context, descriptor);
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      return getFlightInfoPrimaryKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetPrimaryKeys.class), context, descriptor);
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      return getFlightInfoExportedKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetExportedKeys.class), context, descriptor);
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      return getFlightInfoImportedKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetImportedKeys.class), context, descriptor);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();

Review comment:
       We should include an error message.
   
   Also, let's use Flight's CallStatus, not the gRPC Status.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);

Review comment:
       If we're taking `CallOptions` here maybe add that to every call?

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);

Review comment:
       Or, maybe provide methods that just construct FlightDescriptor for you.

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/sql/StatementContext.java
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.flight.sql;
+
+import java.io.Serializable;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Objects;
+import java.util.Optional;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.util.Preconditions;
+
+/**
+ * Context for {@link T} to be persisted in memory in between {@link FlightSqlProducer} calls.
+ *
+ * @param <T> the {@link Statement} to be persisted.
+ */
+public final class StatementContext<T extends Statement> implements AutoCloseable, Serializable {

Review comment:
       (Actually, why implement serializable if this is for in-memory usage?)

##########
File path: java/pom.xml
##########
@@ -549,6 +549,27 @@
         <version>2.8.2</version>
         <scope>provided</scope>
       </dependency>
+      <dependency>
+        <groupId>org.apache.calcite.avatica</groupId>
+        <artifactId>avatica</artifactId>
+        <version>1.18.0</version>
+      </dependency>
+      <dependency>
+        <groupId>org.bouncycastle</groupId>
+        <artifactId>bcpkix-jdk15on</artifactId>
+        <version>1.61</version>
+      </dependency>
+      <dependency>
+        <groupId>com.google.code.findbugs</groupId>
+        <artifactId>annotations</artifactId>
+        <version>3.0.1</version>
+      </dependency>
+      <dependency>
+        <groupId>org.hamcrest</groupId>
+        <artifactId>hamcrest</artifactId>
+        <version>2.2</version>
+        <scope>test</scope>
+      </dependency>

Review comment:
       Where do we actually use these dependencies?

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.
+   */
+  repeated uint32 info = 1;
+}
+
+/*
+ * Represents a request to retrieve the list of catalogs on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name.
+ */
+message CommandGetCatalogs {
+}
+
+/*
+ * Represents a request to retrieve the list of schemas on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, then schema_name.
+ */
+message CommandGetSchemas {
+  /*
+   * Specifies the Catalog to search for schemas.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, the pattern will not be used to narrow the search.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+}
+
+/*
+ * Represents a request to retrieve the list of tables, and optionally their schemas, on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  table_type: utf8,
+ *  table_schema: bytes
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type.
+ */
+message CommandGetTables {
+  /*
+   * Specifies the Catalog to search for the tables.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, all schemas matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+
+  /*
+   * Specifies a filter pattern for tables to search for.
+   * When no table_name_filter_pattern is provided, all tables matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue table_name_filter_pattern = 3;
+
+  // Specifies a filter of table types which must match.
+  repeated string table_types = 4;
+
+  // Specifies if the schema should be returned for found tables.
+  bool include_schema = 5;
+}
+
+/*
+ * Represents a request to retrieve the list of table types on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  table_type: utf8
+ * >
+ * The returned data should be ordered by table_type.
+ */
+message CommandGetTableTypes {
+}
+
+/*
+ * Represents a request to retrieve the primary keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  column_name: utf8,
+ *  key_sequence: int,
+ *  key_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetPrimaryKeys {
+  // Specifies the catalog to search for the table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the table to get the primary keys for.
+  google.protobuf.StringValue table = 3;
+}
+
+/*
+ * Represents a request to retrieve a description of the foreign key columns that reference the given table's
+ * primary key columns (the foreign keys exported by a table) of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int

Review comment:
       What are the values of update_rule and delete_rule?

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.
+   */
+  repeated uint32 info = 1;
+}
+
+/*
+ * Represents a request to retrieve the list of catalogs on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name.
+ */
+message CommandGetCatalogs {
+}
+
+/*
+ * Represents a request to retrieve the list of schemas on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, then schema_name.
+ */
+message CommandGetSchemas {
+  /*
+   * Specifies the Catalog to search for schemas.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, the pattern will not be used to narrow the search.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+}
+
+/*
+ * Represents a request to retrieve the list of tables, and optionally their schemas, on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  table_type: utf8,
+ *  table_schema: bytes
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type.
+ */
+message CommandGetTables {
+  /*
+   * Specifies the Catalog to search for the tables.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, all schemas matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+
+  /*
+   * Specifies a filter pattern for tables to search for.
+   * When no table_name_filter_pattern is provided, all tables matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue table_name_filter_pattern = 3;
+
+  // Specifies a filter of table types which must match.
+  repeated string table_types = 4;
+
+  // Specifies if the schema should be returned for found tables.
+  bool include_schema = 5;
+}
+
+/*
+ * Represents a request to retrieve the list of table types on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  table_type: utf8
+ * >
+ * The returned data should be ordered by table_type.
+ */
+message CommandGetTableTypes {
+}
+
+/*
+ * Represents a request to retrieve the primary keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  column_name: utf8,
+ *  key_sequence: int,
+ *  key_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetPrimaryKeys {
+  // Specifies the catalog to search for the table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the table to get the primary keys for.
+  google.protobuf.StringValue table = 3;
+}
+
+/*
+ * Represents a request to retrieve a description of the foreign key columns that reference the given table's
+ * primary key columns (the foreign keys exported by a table) of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetExportedKeys {
+  // Specifies the catalog to search for the foreign key table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the foreign key table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the foreign key table to get the foreign keys for.
+  string table = 3;
+}
+
+/*
+ * Represents a request to retrieve the foreign keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetImportedKeys {
+  // Specifies the catalog to search for the primary key table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the primary key table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the primary key table to get the foreign keys for.
+  string table = 3;
+}
+
+// SQL Execution Action Messages
+
+/*
+ * Request message for the "GetPreparedStatement" action on a Flight SQL enabled backend.
+ */
+message ActionCreatePreparedStatementRequest {

Review comment:
       Is it now "GetPreparedStatement" or "CreatePreparedStatement"?

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.

Review comment:
       I'm curious, why the switch to integers? Strings would naturally provide namespacing. Are there enough values that strings would be a prohibitive cost?

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();

Review comment:
       Objects.requireNonNull, NullPointerException, or IllegalArgumentException with a descriptive message might be more natural here (I would consider StatusRuntimeException as something that you could actually react to vs. a programming bug here)

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.

Review comment:
       Nit: this is a little confusingly worded. This is saying catalog and schema are nullable, but table is NonNull? It's a little unfortunate those annotations didn't get standardized, but maybe this info can be put in the `@param` blocks instead.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);

Review comment:
       nit: no need for `this.`

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetExportedKeys.Builder builder = CommandGetExportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which imports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The primary key table catalog.
+   * @param schema  The primary key table schema.
+   * @param table   The primary key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getImportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetImportedKeys.Builder builder = CommandGetImportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of table types.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTableTypes() {
+    final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Create a prepared statement on the server.
+   *
+   * @param query The query to prepare.
+   * @return The representation of the prepared statement which exists on the server.
+   */
+  public PreparedStatement prepare(String query) {
+    return new PreparedStatement(client, query);
+  }
+
+  /**
+   * Helper class to encapsulate Flight SQL prepared statement logic.
+   */
+  public static class PreparedStatement implements Closeable {

Review comment:
       Use AutoCloseable?

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetExportedKeys.Builder builder = CommandGetExportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which imports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The primary key table catalog.
+   * @param schema  The primary key table schema.
+   * @param table   The primary key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getImportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetImportedKeys.Builder builder = CommandGetImportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of table types.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTableTypes() {
+    final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Create a prepared statement on the server.
+   *
+   * @param query The query to prepare.
+   * @return The representation of the prepared statement which exists on the server.
+   */
+  public PreparedStatement prepare(String query) {
+    return new PreparedStatement(client, query);
+  }
+
+  /**
+   * Helper class to encapsulate Flight SQL prepared statement logic.
+   */
+  public static class PreparedStatement implements Closeable {
+    private final FlightClient client;
+    private final ActionCreatePreparedStatementResult preparedStatementResult;
+    private AtomicLong invocationCount;
+    private boolean isClosed;
+    private Schema resultSetSchema = null;
+    private Schema parameterSchema = null;
+    private VectorSchemaRoot parameterBindingRoot;
+
+    /**
+     * Constructor.
+     *
+     * @param client The client. FlightSqlPreparedStatement does not maintain this resource.
+     * @param sql    The query.
+     */
+    public PreparedStatement(FlightClient client, String sql) {
+      this.client = client;
+
+      final Iterator<Result> preparedStatementResults = client.doAction(new Action(
+          FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType(),
+          Any.pack(ActionCreatePreparedStatementRequest
+              .newBuilder()
+              .setQuery(sql)
+              .build())
+              .toByteArray()));
+
+      preparedStatementResult = FlightSqlUtils.unpackAndParseOrThrow(
+          preparedStatementResults.next().getBody(),
+          ActionCreatePreparedStatementResult.class);
+
+      invocationCount = new AtomicLong(0);
+      isClosed = false;
+    }
+
+    /**
+     * Set the {@link VectorSchemaRoot} containing the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     * @param parameterBindingRoot  a {@link VectorSchemaRoot} object contain the values to be used in the
+     *                              PreparedStatement setters.
+     */
+    public void setParameters(VectorSchemaRoot parameterBindingRoot) {
+      this.parameterBindingRoot = parameterBindingRoot;
+    }
+
+    /**
+     * Empty the {@link VectorSchemaRoot} that contains the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     */
+    public void clearParameters() {
+      this.parameterBindingRoot = null;
+    }
+
+    /**
+     * Returns the Schema of the resultset.
+     *
+     * @return the Schema of the resultset.
+     */
+    public Schema getResultSetSchema() {
+      if (resultSetSchema == null && preparedStatementResult.getDatasetSchema() != null) {
+        resultSetSchema = Schema.deserialize(preparedStatementResult.getDatasetSchema().asReadOnlyByteBuffer());
+      }
+      return resultSetSchema;
+    }
+
+    /**
+     * Returns the Schema of the parameters.
+     *
+     * @return the Schema of the parameters.
+     */
+    public Schema getParameterSchema() {
+      if (parameterSchema == null && preparedStatementResult.getParameterSchema() != null) {
+        parameterSchema = Schema.deserialize(preparedStatementResult.getParameterSchema().asReadOnlyByteBuffer());
+      }
+      return parameterSchema;
+    }
+
+    /**
+     * Executes the prepared statement query on the server.
+     *
+     * @return a FlightInfo object representing the stream(s) to fetch.
+     * @throws IOException if the PreparedStatement is closed.
+     */
+    public FlightInfo execute() throws IOException {
+      if (isClosed) {
+        throw new IllegalStateException("Prepared statement has already been closed on the server.");
+      }
+
+      final FlightDescriptor descriptor = FlightDescriptor
+          .command(Any.pack(CommandPreparedStatementQuery.newBuilder()
+              .setClientExecutionHandle(
+                  ByteString.copyFrom(ByteBuffer.allocate(Long.BYTES).putLong(invocationCount.getAndIncrement())))
+              .setPreparedStatementHandle(preparedStatementResult.getPreparedStatementHandle())
+              .build())
+              .toByteArray());
+
+      if (parameterBindingRoot != null) {
+        final SyncPutListener putListener = new SyncPutListener();
+
+        FlightClient.ClientStreamListener listener =
+            client.startPut(descriptor, this.parameterBindingRoot, putListener);
+
+        listener.putNext();
+        listener.completed();
+      }

Review comment:
       Hmm, so creating a prepared statement and executing it with lots of different parameters will require many RPCs/round-trips. I guess that works, but I wonder if we couldn't improve that. (Then again, this at least makes it easy to delineate where each result set starts/ends). Doesn't need to be solved now though.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetExportedKeys.Builder builder = CommandGetExportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which imports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The primary key table catalog.
+   * @param schema  The primary key table schema.
+   * @param table   The primary key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getImportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetImportedKeys.Builder builder = CommandGetImportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of table types.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTableTypes() {
+    final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Create a prepared statement on the server.
+   *
+   * @param query The query to prepare.
+   * @return The representation of the prepared statement which exists on the server.
+   */
+  public PreparedStatement prepare(String query) {
+    return new PreparedStatement(client, query);
+  }
+
+  /**
+   * Helper class to encapsulate Flight SQL prepared statement logic.
+   */
+  public static class PreparedStatement implements Closeable {
+    private final FlightClient client;
+    private final ActionCreatePreparedStatementResult preparedStatementResult;
+    private AtomicLong invocationCount;
+    private boolean isClosed;
+    private Schema resultSetSchema = null;
+    private Schema parameterSchema = null;
+    private VectorSchemaRoot parameterBindingRoot;
+
+    /**
+     * Constructor.
+     *
+     * @param client The client. FlightSqlPreparedStatement does not maintain this resource.
+     * @param sql    The query.
+     */
+    public PreparedStatement(FlightClient client, String sql) {
+      this.client = client;
+
+      final Iterator<Result> preparedStatementResults = client.doAction(new Action(
+          FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType(),
+          Any.pack(ActionCreatePreparedStatementRequest
+              .newBuilder()
+              .setQuery(sql)
+              .build())
+              .toByteArray()));
+
+      preparedStatementResult = FlightSqlUtils.unpackAndParseOrThrow(
+          preparedStatementResults.next().getBody(),
+          ActionCreatePreparedStatementResult.class);
+
+      invocationCount = new AtomicLong(0);
+      isClosed = false;
+    }
+
+    /**
+     * Set the {@link VectorSchemaRoot} containing the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     * @param parameterBindingRoot  a {@link VectorSchemaRoot} object contain the values to be used in the
+     *                              PreparedStatement setters.
+     */
+    public void setParameters(VectorSchemaRoot parameterBindingRoot) {
+      this.parameterBindingRoot = parameterBindingRoot;
+    }
+
+    /**
+     * Empty the {@link VectorSchemaRoot} that contains the parameter binding from a preparedStatemnt
+     * operation.
+     *

Review comment:
       Maybe clarify (either here or above) that this does not clear or otherwise free resources associated with the root.

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.

Review comment:
       This is inconsistent with the values above.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {

Review comment:
       For InterruptedException: `Thread.currentThread().interrupt()`?

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.
+   */
+  repeated uint32 info = 1;
+}
+
+/*
+ * Represents a request to retrieve the list of catalogs on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name.
+ */
+message CommandGetCatalogs {
+}
+
+/*
+ * Represents a request to retrieve the list of schemas on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, then schema_name.
+ */
+message CommandGetSchemas {
+  /*
+   * Specifies the Catalog to search for schemas.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, the pattern will not be used to narrow the search.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+}
+
+/*
+ * Represents a request to retrieve the list of tables, and optionally their schemas, on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  table_type: utf8,
+ *  table_schema: bytes
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type.
+ */
+message CommandGetTables {
+  /*
+   * Specifies the Catalog to search for the tables.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, all schemas matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+
+  /*
+   * Specifies a filter pattern for tables to search for.
+   * When no table_name_filter_pattern is provided, all tables matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue table_name_filter_pattern = 3;
+
+  // Specifies a filter of table types which must match.
+  repeated string table_types = 4;
+
+  // Specifies if the schema should be returned for found tables.
+  bool include_schema = 5;
+}
+
+/*
+ * Represents a request to retrieve the list of table types on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  table_type: utf8
+ * >
+ * The returned data should be ordered by table_type.
+ */
+message CommandGetTableTypes {
+}
+
+/*
+ * Represents a request to retrieve the primary keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  column_name: utf8,
+ *  key_sequence: int,
+ *  key_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.

Review comment:
       Let's keep the ordering consistent with the ordering of the fields in the schema.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetExportedKeys.Builder builder = CommandGetExportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which imports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The primary key table catalog.
+   * @param schema  The primary key table schema.
+   * @param table   The primary key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getImportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetImportedKeys.Builder builder = CommandGetImportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of table types.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTableTypes() {
+    final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Create a prepared statement on the server.
+   *
+   * @param query The query to prepare.
+   * @return The representation of the prepared statement which exists on the server.
+   */
+  public PreparedStatement prepare(String query) {
+    return new PreparedStatement(client, query);
+  }
+
+  /**
+   * Helper class to encapsulate Flight SQL prepared statement logic.
+   */
+  public static class PreparedStatement implements Closeable {
+    private final FlightClient client;
+    private final ActionCreatePreparedStatementResult preparedStatementResult;
+    private AtomicLong invocationCount;
+    private boolean isClosed;
+    private Schema resultSetSchema = null;
+    private Schema parameterSchema = null;
+    private VectorSchemaRoot parameterBindingRoot;
+
+    /**
+     * Constructor.
+     *
+     * @param client The client. FlightSqlPreparedStatement does not maintain this resource.
+     * @param sql    The query.
+     */
+    public PreparedStatement(FlightClient client, String sql) {
+      this.client = client;
+
+      final Iterator<Result> preparedStatementResults = client.doAction(new Action(
+          FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType(),
+          Any.pack(ActionCreatePreparedStatementRequest
+              .newBuilder()
+              .setQuery(sql)
+              .build())
+              .toByteArray()));
+
+      preparedStatementResult = FlightSqlUtils.unpackAndParseOrThrow(
+          preparedStatementResults.next().getBody(),
+          ActionCreatePreparedStatementResult.class);
+
+      invocationCount = new AtomicLong(0);
+      isClosed = false;
+    }
+
+    /**
+     * Set the {@link VectorSchemaRoot} containing the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     * @param parameterBindingRoot  a {@link VectorSchemaRoot} object contain the values to be used in the
+     *                              PreparedStatement setters.
+     */
+    public void setParameters(VectorSchemaRoot parameterBindingRoot) {
+      this.parameterBindingRoot = parameterBindingRoot;
+    }
+
+    /**
+     * Empty the {@link VectorSchemaRoot} that contains the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     */
+    public void clearParameters() {
+      this.parameterBindingRoot = null;
+    }
+
+    /**
+     * Returns the Schema of the resultset.
+     *
+     * @return the Schema of the resultset.
+     */
+    public Schema getResultSetSchema() {
+      if (resultSetSchema == null && preparedStatementResult.getDatasetSchema() != null) {
+        resultSetSchema = Schema.deserialize(preparedStatementResult.getDatasetSchema().asReadOnlyByteBuffer());
+      }
+      return resultSetSchema;
+    }
+
+    /**
+     * Returns the Schema of the parameters.
+     *
+     * @return the Schema of the parameters.
+     */
+    public Schema getParameterSchema() {
+      if (parameterSchema == null && preparedStatementResult.getParameterSchema() != null) {
+        parameterSchema = Schema.deserialize(preparedStatementResult.getParameterSchema().asReadOnlyByteBuffer());
+      }
+      return parameterSchema;
+    }
+
+    /**
+     * Executes the prepared statement query on the server.
+     *
+     * @return a FlightInfo object representing the stream(s) to fetch.
+     * @throws IOException if the PreparedStatement is closed.
+     */
+    public FlightInfo execute() throws IOException {
+      if (isClosed) {
+        throw new IllegalStateException("Prepared statement has already been closed on the server.");

Review comment:
       IIRC, the original proposal had comments about standardizing some errors. This might be a candidate for that, so that the server and client can raise a consistent exception for attempting to use an already-closed statement.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetExportedKeys.Builder builder = CommandGetExportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which imports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The primary key table catalog.
+   * @param schema  The primary key table schema.
+   * @param table   The primary key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getImportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetImportedKeys.Builder builder = CommandGetImportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of table types.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTableTypes() {
+    final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Create a prepared statement on the server.
+   *
+   * @param query The query to prepare.
+   * @return The representation of the prepared statement which exists on the server.
+   */
+  public PreparedStatement prepare(String query) {
+    return new PreparedStatement(client, query);
+  }
+
+  /**
+   * Helper class to encapsulate Flight SQL prepared statement logic.
+   */
+  public static class PreparedStatement implements Closeable {
+    private final FlightClient client;
+    private final ActionCreatePreparedStatementResult preparedStatementResult;
+    private AtomicLong invocationCount;
+    private boolean isClosed;
+    private Schema resultSetSchema = null;
+    private Schema parameterSchema = null;
+    private VectorSchemaRoot parameterBindingRoot;
+
+    /**
+     * Constructor.
+     *
+     * @param client The client. FlightSqlPreparedStatement does not maintain this resource.

Review comment:
       ```suggestion
        * @param client The client. PreparedStatement does not maintain this resource.
   ```

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.
+   *
+   * 1. Server Information [0-500): Provides basic information about the Flight SQL Server.
+   *
+   * The name of the Flight SQL Server.
+   * 0 = FLIGHT_SQL_SERVER_NAME
+   *
+   * The native version of the Flight SQL Server.
+   * 1 = FLIGHT_SQL_SERVER_VERSION
+   *
+   * The Arrow format version of the Flight SQL Server.
+   * 2 = FLIGHT_SQL_SERVER_ARROW_VERSION
+   *
+   * Indicates whether the Flight SQL Server is read only.
+   * 3 = FLIGHT_SQL_SERVER_READ_ONLY
+   *
+   * 2. SQL Syntax Information [500-1000): provides information about SQL syntax supported by the Flight SQL Server.
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of catalogs.
+   * In a SQL environment, a catalog is a collection of schemas.
+   * 500 = SQL_DDL_CATALOG
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of schemas.
+   * In a SQL environment, a catalog is a collection of tables, views, indexes etc.
+   * 501 = SQL_DDL_SCHEMA
+   *
+   * Indicates whether the Flight SQL Server supports CREATE and DROP of tables.
+   * In a SQL environment, a table is a collection of rows of information. Each row of information
+   * may have one or more columns of data.
+   * 502 = SQL_DDL_TABLE
+   *
+   * Indicates the case sensitivity of catalog, table and schema names.
+   * 503 = SQL_IDENTIFIER_CASE
+   *
+   * Indicates the supported character(s) used to surround a delimited identifier.
+   * 504 = SQL_IDENTIFIER_QUOTE_CHAR
+   *
+   * Indicates case sensitivity of quoted identifiers.
+   * 505 = SQL_QUOTED_IDENTIFIER_CASE
+   *
+   * If omitted, then all metadata will be retrieved.
+   * Flight SQL Servers may choose to include additional metadata above and beyond the specified set, however they must
+   * at least return the specified set. IDs ranging from 0 to 10,000 (exclusive) are reserved.
+   * If additional metadata is included, the metadata IDs should start from 10,000.
+   */
+  repeated uint32 info = 1;
+}
+
+/*
+ * Represents a request to retrieve the list of catalogs on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name.
+ */
+message CommandGetCatalogs {
+}
+
+/*
+ * Represents a request to retrieve the list of schemas on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, then schema_name.
+ */
+message CommandGetSchemas {
+  /*
+   * Specifies the Catalog to search for schemas.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, the pattern will not be used to narrow the search.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+}
+
+/*
+ * Represents a request to retrieve the list of tables, and optionally their schemas, on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  table_type: utf8,
+ *  table_schema: bytes
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type.
+ */
+message CommandGetTables {
+  /*
+   * Specifies the Catalog to search for the tables.
+   * If omitted, then all catalogs are searched.
+   */
+  google.protobuf.StringValue catalog = 1;
+
+  /*
+   * Specifies a filter pattern for schemas to search for.
+   * When no schema_filter_pattern is provided, all schemas matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue schema_filter_pattern = 2;
+
+  /*
+   * Specifies a filter pattern for tables to search for.
+   * When no table_name_filter_pattern is provided, all tables matching other filters are searched.
+   * In the pattern string, two special characters can be used to denote matching rules:
+   *    - "%" means to match any substring with 0 or more characters.
+   *    - "_" means to match any one character.
+   */
+  google.protobuf.StringValue table_name_filter_pattern = 3;
+
+  // Specifies a filter of table types which must match.
+  repeated string table_types = 4;
+
+  // Specifies if the schema should be returned for found tables.
+  bool include_schema = 5;
+}
+
+/*
+ * Represents a request to retrieve the list of table types on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  table_type: utf8
+ * >
+ * The returned data should be ordered by table_type.
+ */
+message CommandGetTableTypes {
+}
+
+/*
+ * Represents a request to retrieve the primary keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  catalog_name: utf8,
+ *  schema_name: utf8,
+ *  table_name: utf8,
+ *  column_name: utf8,
+ *  key_sequence: int,
+ *  key_name: utf8
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetPrimaryKeys {
+  // Specifies the catalog to search for the table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the table to get the primary keys for.
+  google.protobuf.StringValue table = 3;
+}
+
+/*
+ * Represents a request to retrieve a description of the foreign key columns that reference the given table's
+ * primary key columns (the foreign keys exported by a table) of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetExportedKeys {
+  // Specifies the catalog to search for the foreign key table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the foreign key table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the foreign key table to get the foreign keys for.
+  string table = 3;
+}
+
+/*
+ * Represents a request to retrieve the foreign keys of a table on a Flight SQL enabled backend.
+ * Used in the command member of FlightDescriptor for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the catalog metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  pk_catalog_name: utf8,
+ *  pk_schema_name: utf8,
+ *  pk_table_name: utf8,
+ *  pk_column_name: utf8,
+ *  fk_catalog_name: utf8,
+ *  fk_schema_name: utf8,
+ *  fk_table_name: utf8,
+ *  fk_column_name: utf8,
+ *  key_sequence: int,
+ *  fk_key_name: utf8,
+ *  pk_key_name: utf8,
+ *  update_rule: int,
+ *  delete_rule: int
+ * >
+ * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence.
+ */
+message CommandGetImportedKeys {
+  // Specifies the catalog to search for the primary key table.
+  google.protobuf.StringValue catalog = 1;
+
+  // Specifies the schema to search for the primary key table.
+  google.protobuf.StringValue schema = 2;
+
+  // Specifies the primary key table to get the foreign keys for.
+  string table = 3;
+}
+
+// SQL Execution Action Messages
+
+/*
+ * Request message for the "GetPreparedStatement" action on a Flight SQL enabled backend.
+ */
+message ActionCreatePreparedStatementRequest {
+  // The valid SQL string to create a prepared statement for.
+  string query = 1;
+}
+
+/*
+ * Wrap the result of a "GetPreparedStatement" action.
+ */
+message ActionCreatePreparedStatementResult {
+  // Opaque handle for the prepared statement on the server.
+  bytes prepared_statement_handle = 1;
+
+  // If a result set generating query was provided, dataset_schema contains the 
+  // schema of the dataset as described in Schema.fbs::Schema, it is serialized as an IPC message.
+  bytes dataset_schema = 2;
+
+  // If the query provided contained parameters, parameter_schema contains the 
+  // Schema of the expected parameters as described in Schema.fbs::Schema.
+  bytes parameter_schema = 3;
+}
+
+/*
+ * Request message for the "ClosePreparedStatement" action on a Flight SQL enabled backend.
+ * Closes server resources associated with the prepared statement handle.
+ */
+message ActionClosePreparedStatementRequest {
+  // Opaque handle for the prepared statement on the server.
+  bytes prepared_statement_handle = 1;
+}
+
+
+// SQL Execution Messages.
+
+/*
+ * Represents a SQL query. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the query.
+ */
+message CommandStatementQuery {
+  // The SQL syntax.
+  string query = 1;
+
+  // Unique identifier for the instance of the prepared statement to execute.
+  bytes client_execution_handle = 2;
+}
+

Review comment:
       There used to be a way to execute a query without a prepared statement, was that dropped?

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlUtils.java
##########
@@ -0,0 +1,89 @@
+/*
+ * 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.flight.sql;
+
+import java.util.List;
+
+import org.apache.arrow.flight.ActionType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+
+/**
+ * Utilities to work with Flight SQL semantics.
+ */
+public final class FlightSqlUtils {
+  public static final ActionType FLIGHT_SQL_CREATEPREPAREDSTATEMENT = new ActionType("CreatePreparedStatement",
+      "Creates a reusable prepared statement resource on the server. \n" +
+          "Request Message: ActionCreatePreparedStatementRequest\n" +
+          "Response Message: ActionCreatePreparedStatementResult");
+
+  public static final ActionType FLIGHT_SQL_CLOSEPREPAREDSTATEMENT = new ActionType("ClosePreparedStatement",
+      "Closes a reusable prepared statement resource on the server. \n" +
+          "Request Message: ActionClosePreparedStatementRequest\n" +
+          "Response Message: N/A");
+
+  public static final List<ActionType> FLIGHT_SQL_ACTIONS = ImmutableList.of(
+      FLIGHT_SQL_CREATEPREPAREDSTATEMENT,
+      FLIGHT_SQL_CLOSEPREPAREDSTATEMENT
+  );
+
+  /**
+   * Helper to parse {@link com.google.protobuf.Any} objects to the specific protobuf object.
+   *
+   * @param source the raw bytes source value.
+   * @return the materialized protobuf object.
+   */
+  public static Any parseOrThrow(byte[] source) {
+    try {
+      return Any.parseFrom(source);
+    } catch (InvalidProtocolBufferException e) {
+      throw new AssertionError(e.getMessage());

Review comment:
       I'm not sure AssertionError is the way to go here. Maybe StatusRuntimeException, with a message indicating that the remote sent an invalid protobuf?

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/sql/FlightSqlExample.java
##########
@@ -0,0 +1,1768 @@
+/*
+ * 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.flight.sql;
+
+import static com.google.common.base.Preconditions.checkNotNull;

Review comment:
       Why use the Guava one over Objects.requireNonNull?

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/sql/StatementContext.java
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.flight.sql;
+
+import java.io.Serializable;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Objects;
+import java.util.Optional;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.util.Preconditions;
+
+/**
+ * Context for {@link T} to be persisted in memory in between {@link FlightSqlProducer} calls.
+ *
+ * @param <T> the {@link Statement} to be persisted.
+ */
+public final class StatementContext<T extends Statement> implements AutoCloseable, Serializable {

Review comment:
       I think we should put this and the example in their own namespace to make it clear they're examples.

##########
File path: format/FlightSql.proto
##########
@@ -0,0 +1,402 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.
+ */
+
+syntax = "proto3";
+import "google/protobuf/wrappers.proto";
+
+option java_package = "org.apache.arrow.flight.sql.impl";
+package arrow.flight.protocol.sql;
+
+/*
+ * Represents a metadata request. Used in the command member of FlightDescriptor
+ * for the following RPC calls:
+ *  - GetSchema: return the schema of the query.
+ *  - GetFlightInfo: execute the metadata request.
+ *
+ * The returned schema will be:
+ * <
+ *  info_name: int32,
+ *  value: dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+ * >
+ * where there is one row per requested piece of metadata information.
+ */
+message CommandGetSqlInfo {
+  /*
+   * Values are modelled after ODBC's SQLGetInfo() function. This information is intended to provide
+   * Flight SQL clients with basic, SQL syntax and SQL functions related information.
+   * More information types can be added in future releases.
+   * E.g. more SQL syntax support types, scalar functions support, type conversion support etc.
+   *
+   * // TODO: Flesh out the available set of metadata below.
+   *
+   * Initially, Flight SQL will support the following information types:
+   * - Server Information - Range [0-500)
+   * - Syntax Information - Ragne [500-1000)
+   * Range [0-100000) is reserved for defaults. Custom options should start at 100000.

Review comment:
       (In which case, I'd wonder if we could use integers with a oneof to provide namespacing instead.)

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/sql/FlightSqlExample.java
##########
@@ -0,0 +1,1768 @@
+/*
+ * 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.flight.sql;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Strings.emptyToNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static com.google.protobuf.Any.pack;
+import static com.google.protobuf.ByteString.copyFrom;
+import static java.lang.String.format;
+import static java.util.Collections.singletonList;
+import static java.util.Objects.isNull;
+import static java.util.Optional.empty;
+import static java.util.UUID.randomUUID;
+import static java.util.stream.StreamSupport.stream;
+import static org.apache.arrow.adapter.jdbc.JdbcToArrow.sqlToArrowVectorIterator;
+import static org.apache.arrow.adapter.jdbc.JdbcToArrowUtils.jdbcToArrowSchema;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.Date;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.adapter.jdbc.ArrowVectorIterator;
+import org.apache.arrow.adapter.jdbc.JdbcFieldInfo;
+import org.apache.arrow.adapter.jdbc.JdbcToArrowConfig;
+import org.apache.arrow.adapter.jdbc.JdbcToArrowUtils;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.Criteria;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightEndpoint;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightRuntimeException;
+import org.apache.arrow.flight.FlightStatusCode;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.Location;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DateMilliVector;
+import org.apache.arrow.vector.Decimal256Vector;
+import org.apache.arrow.vector.DecimalVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.LargeVarCharVector;
+import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeMicroVector;
+import org.apache.arrow.vector.TimeMilliVector;
+import org.apache.arrow.vector.TimeNanoVector;
+import org.apache.arrow.vector.TimeSecVector;
+import org.apache.arrow.vector.TimeStampMicroTZVector;
+import org.apache.arrow.vector.TimeStampMilliTZVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampSecTZVector;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.UInt1Vector;
+import org.apache.arrow.vector.UInt2Vector;
+import org.apache.arrow.vector.UInt4Vector;
+import org.apache.arrow.vector.UInt8Vector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorLoader;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.VectorUnloader;
+import org.apache.arrow.vector.complex.DenseUnionVector;
+import org.apache.arrow.vector.holders.NullableIntHolder;
+import org.apache.arrow.vector.holders.NullableVarCharHolder;
+import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.Text;
+import org.apache.commons.dbcp2.ConnectionFactory;
+import org.apache.commons.dbcp2.DriverManagerConnectionFactory;
+import org.apache.commons.dbcp2.PoolableConnection;
+import org.apache.commons.dbcp2.PoolableConnectionFactory;
+import org.apache.commons.dbcp2.PoolingDataSource;
+import org.apache.commons.pool2.ObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.slf4j.Logger;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
+import com.google.protobuf.ProtocolStringList;
+
+import io.grpc.Status;
+
+/**
+ * Proof of concept {@link FlightSqlProducer} implementation showing an Apache Derby backed Flight SQL server capable
+ * of the following workflows:
+ * <!--
+ * TODO Revise summary: is it still matching?
+ * -->
+ * - returning a list of tables from the action `GetTables`.
+ * - creation of a prepared statement from the action `CreatePreparedStatement`.
+ * - execution of a prepared statement by using a {@link CommandPreparedStatementQuery}
+ * with {@link #getFlightInfo} and {@link #getStream}.
+ */
+public class FlightSqlExample implements FlightSqlProducer, AutoCloseable {
+  private static final String DATABASE_URI = "jdbc:derby:target/derbyDB";
+  private static final Logger LOGGER = getLogger(FlightSqlExample.class);
+  private static final Calendar DEFAULT_CALENDAR = JdbcToArrowUtils.getUtcCalendar();
+  private final Location location;
+  private final PoolingDataSource<PoolableConnection> dataSource;
+  private final LoadingCache<ByteString, ResultSet> commandExecutePreparedStatementLoadingCache;
+  private final BufferAllocator rootAllocator = new RootAllocator();
+  private final Cache<ByteString, StatementContext<PreparedStatement>> preparedStatementLoadingCache;
+  private final Cache<ByteString, StatementContext<Statement>> statementLoadingCache;
+  private final LoadingCache<ByteString, ResultSet> commandExecuteStatementLoadingCache;
+
+  public FlightSqlExample(final Location location) {
+    // TODO Constructor should not be doing work.
+    Preconditions.checkState(
+        removeDerbyDatabaseIfExists() && populateDerbyDatabase(),
+        "Failed to reset Derby database!");
+    final ConnectionFactory connectionFactory =
+        new DriverManagerConnectionFactory(DATABASE_URI, new Properties());
+    final PoolableConnectionFactory poolableConnectionFactory =
+        new PoolableConnectionFactory(connectionFactory, null);
+    final ObjectPool<PoolableConnection> connectionPool = new GenericObjectPool<>(poolableConnectionFactory);
+
+    poolableConnectionFactory.setPool(connectionPool);
+    // PoolingDataSource takes ownership of `connectionPool`
+    dataSource = new PoolingDataSource<>(connectionPool);
+
+    preparedStatementLoadingCache =
+        CacheBuilder.newBuilder()
+            .maximumSize(100)
+            .expireAfterWrite(10, TimeUnit.MINUTES)
+            .removalListener(new StatementRemovalListener<PreparedStatement>())
+            .build();
+
+    commandExecutePreparedStatementLoadingCache =
+        CacheBuilder.newBuilder()
+            .maximumSize(100)
+            .expireAfterWrite(10, TimeUnit.MINUTES)
+            .removalListener(new CommandExecuteStatementRemovalListener())
+            .build(new CommandExecutePreparedStatementCacheLoader(preparedStatementLoadingCache));
+
+    statementLoadingCache =
+        CacheBuilder.newBuilder()
+            .maximumSize(100)
+            .expireAfterWrite(10, TimeUnit.MINUTES)
+            .removalListener(new StatementRemovalListener<>())
+            .build();
+
+    commandExecuteStatementLoadingCache =
+        CacheBuilder.newBuilder()
+            .maximumSize(100)
+            .expireAfterWrite(10, TimeUnit.MINUTES)
+            .removalListener(new CommandExecuteStatementRemovalListener())
+            .build(new CommandExecuteStatementCacheLoader(statementLoadingCache));
+
+    this.location = location;
+  }
+
+  private static boolean removeDerbyDatabaseIfExists() {
+    boolean wasSuccess;
+    final Path path = Paths.get("target" + File.separator + "derbyDB");
+
+    try (final Stream<Path> walk = Files.walk(path)) {
+      /*
+       * Iterate over all paths to delete, mapping each path to the outcome of its own
+       * deletion as a boolean representing whether or not each individual operation was
+       * successful; then reduce all booleans into a single answer, and store that into
+       * `wasSuccess`, which will later be returned by this method.
+       * If for whatever reason the resulting `Stream<Boolean>` is empty, throw an `IOException`;
+       * this not expected.
+       */
+      wasSuccess = walk.sorted(Comparator.reverseOrder()).map(Path::toFile).map(File::delete)
+          .reduce(Boolean::logicalAnd).orElseThrow(IOException::new);
+    } catch (IOException e) {
+      /*
+       * The only acceptable scenario for an `IOException` to be thrown here is if
+       * an attempt to delete an non-existing file takes place -- which should be
+       * alright, since they would be deleted anyway.
+       */
+      if (!(wasSuccess = e instanceof NoSuchFileException)) {
+        LOGGER.error(format("Failed attempt to clear DerbyDB: <%s>", e.getMessage()), e);
+      }
+    }
+
+    return wasSuccess;
+  }
+
+  private static boolean populateDerbyDatabase() {
+    Optional<SQLException> exception = empty();
+    try (final Connection connection = DriverManager.getConnection("jdbc:derby:target/derbyDB;create=true");
+         Statement statement = connection.createStatement()) {
+      statement.execute("CREATE TABLE foreignTable (" +
+          "id INT not null primary key GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1), " +
+          "foreignName varchar(100), " +
+          "value int)");
+      statement.execute("CREATE TABLE intTable (" +
+          "id INT not null primary key GENERATED ALWAYS AS IDENTITY (START WITH 1, INCREMENT BY 1), " +
+          "keyName varchar(100), " +
+          "value int, " +
+          "foreignId int references foreignTable(id))");
+      statement.execute("INSERT INTO foreignTable (foreignName, value) VALUES ('keyOne', 1)");
+      statement.execute("INSERT INTO foreignTable (foreignName, value) VALUES ('keyTwo', 0)");
+      statement.execute("INSERT INTO foreignTable (foreignName, value) VALUES ('keyThree', -1)");
+      statement.execute("INSERT INTO intTable (keyName, value, foreignId) VALUES ('one', 1, 1)");
+      statement.execute("INSERT INTO intTable (keyName, value, foreignId) VALUES ('zero', 0, 1)");
+      statement.execute("INSERT INTO intTable (keyName, value, foreignId) VALUES ('negative one', -1, 1)");
+    } catch (SQLException e) {
+      LOGGER.error(
+          format("Failed attempt to populate DerbyDB: <%s>", e.getMessage()),
+          (exception = Optional.of(e)).get());
+    }
+
+    return !exception.isPresent();

Review comment:
       Why the convolutions with the optional?

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/TestFlightSql.java
##########
@@ -0,0 +1,653 @@
+/*
+ * 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.flight;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
+import static java.util.Collections.singletonList;
+import static java.util.Objects.isNull;
+import static org.apache.arrow.util.AutoCloseables.close;
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
+
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.io.Reader;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.stream.IntStream;
+
+import org.apache.arrow.flight.sql.FlightSqlClient;
+import org.apache.arrow.flight.sql.FlightSqlClient.PreparedStatement;
+import org.apache.arrow.flight.sql.FlightSqlExample;
+import org.apache.arrow.flight.sql.FlightSqlProducer;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.complex.DenseUnionVector;
+import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.Text;
+import org.hamcrest.Matcher;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ErrorCollector;
+
+import com.google.common.collect.ImmutableList;
+
+/**
+ * Test direct usage of Flight SQL workflows.
+ */
+public class TestFlightSql {

Review comment:
       I wonder if it might make more sense to test against a 'mock' server that isn't actually backed by a database. There are a lot of details here that are tied to Apache Derby.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlUtils.java
##########
@@ -0,0 +1,89 @@
+/*
+ * 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.flight.sql;
+
+import java.util.List;
+
+import org.apache.arrow.flight.ActionType;
+
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.Message;
+
+/**
+ * Utilities to work with Flight SQL semantics.
+ */
+public final class FlightSqlUtils {
+  public static final ActionType FLIGHT_SQL_CREATEPREPAREDSTATEMENT = new ActionType("CreatePreparedStatement",
+      "Creates a reusable prepared statement resource on the server. \n" +
+          "Request Message: ActionCreatePreparedStatementRequest\n" +
+          "Response Message: ActionCreatePreparedStatementResult");
+
+  public static final ActionType FLIGHT_SQL_CLOSEPREPAREDSTATEMENT = new ActionType("ClosePreparedStatement",
+      "Closes a reusable prepared statement resource on the server. \n" +
+          "Request Message: ActionClosePreparedStatementRequest\n" +
+          "Response Message: N/A");
+
+  public static final List<ActionType> FLIGHT_SQL_ACTIONS = ImmutableList.of(
+      FLIGHT_SQL_CREATEPREPAREDSTATEMENT,
+      FLIGHT_SQL_CLOSEPREPAREDSTATEMENT
+  );
+
+  /**
+   * Helper to parse {@link com.google.protobuf.Any} objects to the specific protobuf object.
+   *
+   * @param source the raw bytes source value.
+   * @return the materialized protobuf object.
+   */
+  public static Any parseOrThrow(byte[] source) {
+    try {
+      return Any.parseFrom(source);
+    } catch (InvalidProtocolBufferException e) {
+      throw new AssertionError(e.getMessage());

Review comment:
       And ideally, unpackAndParseOrThrow would also include the expected message in the error message (and the actual message if we got past the parsing step).

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java
##########
@@ -0,0 +1,704 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.ActionType;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightProducer;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.UnionMode;
+import org.apache.arrow.vector.types.pojo.ArrowType.Union;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import io.grpc.Status;
+
+/**
+ * API to Implement an Arrow Flight SQL producer.
+ */
+public interface FlightSqlProducer extends FlightProducer, AutoCloseable {
+  /**
+   * Depending on the provided command, method either:
+   * 1. Return information about a SQL query, or
+   * 2. Return information about a prepared statement. In this case, parameters binding is allowed.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return information about the given SQL query, or the given prepared statement.
+   */
+  @Override
+  default FlightInfo getFlightInfo(CallContext context, FlightDescriptor descriptor) {
+    final Any command = FlightSqlUtils.parseOrThrow(descriptor.getCommand());
+
+    if (command.is(CommandStatementQuery.class)) {
+      return getFlightInfoStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      return getFlightInfoPreparedStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      return getFlightInfoCatalogs(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetCatalogs.class), context, descriptor);
+    } else if (command.is(CommandGetSchemas.class)) {
+      return getFlightInfoSchemas(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetSchemas.class), context, descriptor);
+    } else if (command.is(CommandGetTables.class)) {
+      return getFlightInfoTables(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetTables.class), context, descriptor);
+    } else if (command.is(CommandGetTableTypes.class)) {
+      return getFlightInfoTableTypes(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetTableTypes.class), context, descriptor);
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      return getFlightInfoSqlInfo(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetSqlInfo.class), context, descriptor);
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      return getFlightInfoPrimaryKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetPrimaryKeys.class), context, descriptor);
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      return getFlightInfoExportedKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetExportedKeys.class), context, descriptor);
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      return getFlightInfoImportedKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetImportedKeys.class), context, descriptor);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Returns the schema of the result produced by the SQL query.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return the result set schema.
+   */
+  @Override
+  default SchemaResult getSchema(CallContext context, FlightDescriptor descriptor) {
+    final Any command = FlightSqlUtils.parseOrThrow(descriptor.getCommand());
+
+    if (command.is(CommandStatementQuery.class)) {
+      return getSchemaStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      return getSchemaCatalogs();
+    } else if (command.is(CommandGetSchemas.class)) {
+      return getSchemaSchemas();
+    } else if (command.is(CommandGetTables.class)) {
+      return getSchemaTables();
+    } else if (command.is(CommandGetTableTypes.class)) {
+      return getSchemaTableTypes();
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      return getSchemaSqlInfo();
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      return getSchemaPrimaryKeys();
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      return getSchemaForImportedAndExportedKeys();
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      return getSchemaForImportedAndExportedKeys();
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Depending on the provided command, method either:
+   * 1. Return data for a stream produced by executing the provided SQL query, or
+   * 2. Return data for a prepared statement. In this case, parameters binding is allowed.
+   *
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  @Override
+  default void getStream(CallContext context, Ticket ticket, ServerStreamListener listener) {
+    final Any command;
+
+    try {
+      command = Any.parseFrom(ticket.getBytes());
+    } catch (InvalidProtocolBufferException e) {
+      listener.error(e);
+      return;
+    }
+
+    if (command.is(CommandStatementQuery.class)) {
+      getStreamStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, ticket, listener);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      getStreamPreparedStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class), context, ticket, listener);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      getStreamCatalogs(context, ticket, listener);
+    } else if (command.is(CommandGetSchemas.class)) {
+      getStreamSchemas(FlightSqlUtils.unpackOrThrow(command, CommandGetSchemas.class), context, ticket, listener);
+    } else if (command.is(CommandGetTables.class)) {
+      getStreamTables(FlightSqlUtils.unpackOrThrow(command, CommandGetTables.class), context, ticket, listener);
+    } else if (command.is(CommandGetTableTypes.class)) {
+      getStreamTableTypes(context, ticket, listener);
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      getStreamSqlInfo(FlightSqlUtils.unpackOrThrow(command, CommandGetSqlInfo.class), context, ticket, listener);
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      getStreamPrimaryKeys(FlightSqlUtils.unpackOrThrow(command, CommandGetPrimaryKeys.class),
+          context, ticket, listener);
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      getStreamExportedKeys(FlightSqlUtils.unpackOrThrow(command, CommandGetExportedKeys.class),
+          context, ticket, listener);
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      getStreamImportedKeys(FlightSqlUtils.unpackOrThrow(command, CommandGetImportedKeys.class),
+          context, ticket, listener);
+    } else {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+  }
+
+  /**
+   * Depending on the provided command, method either:
+   * 1. Execute provided SQL query as an update statement, or
+   * 2. Execute provided update SQL query prepared statement. In this case, parameters binding
+   * is allowed, or
+   * 3. Binds parameters to the provided prepared statement.
+   *
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The data stream listener for update result acknowledgement.
+   * @return a Runnable to process the stream.
+   */
+  @Override
+  default Runnable acceptPut(CallContext context, FlightStream flightStream, StreamListener<PutResult> ackStream) {
+    final Any command = FlightSqlUtils.parseOrThrow(flightStream.getDescriptor().getCommand());
+
+    if (command.is(CommandStatementUpdate.class)) {
+      return acceptPutStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementUpdate.class),
+          context, flightStream, ackStream);
+    } else if (command.is(CommandPreparedStatementUpdate.class)) {
+      return acceptPutPreparedStatementUpdate(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementUpdate.class),
+          context, flightStream, ackStream);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      return acceptPutPreparedStatementQuery(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class),
+          context, flightStream, ackStream);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Lists all available Flight SQL actions.
+   *
+   * @param context  Per-call context.
+   * @param listener An interface for sending data back to the client.
+   */
+  @Override
+  default void listActions(CallContext context, StreamListener<ActionType> listener) {
+    FlightSqlUtils.FLIGHT_SQL_ACTIONS.forEach(listener::onNext);
+    listener.onCompleted();
+  }
+
+  /**
+   * Performs the requested Flight SQL action.
+   *
+   * @param context  Per-call context.
+   * @param action   Client-supplied parameters.
+   * @param listener A stream of responses.
+   */
+  @Override
+  default void doAction(CallContext context, Action action, StreamListener<Result> listener) {
+    final String actionType = action.getType();
+    if (actionType.equals(FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType())) {
+      final ActionCreatePreparedStatementRequest request = FlightSqlUtils.unpackAndParseOrThrow(action.getBody(),
+          ActionCreatePreparedStatementRequest.class);
+      createPreparedStatement(request, context, listener);
+    } else if (actionType.equals(FlightSqlUtils.FLIGHT_SQL_CLOSEPREPAREDSTATEMENT.getType())) {
+      final ActionClosePreparedStatementRequest request = FlightSqlUtils.unpackAndParseOrThrow(action.getBody(),
+          ActionClosePreparedStatementRequest.class);
+      closePreparedStatement(request, context, listener);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Creates a prepared statement on the server and returns a handle and metadata for in a
+   * {@link ActionCreatePreparedStatementResult} object in a {@link Result}
+   * object.
+   *
+   * @param request  The sql command to generate the prepared statement.
+   * @param context  Per-call context.
+   * @param listener A stream of responses.
+   */
+  void createPreparedStatement(ActionCreatePreparedStatementRequest request, CallContext context,
+                               StreamListener<Result> listener);
+
+  /**
+   * Closes a prepared statement on the server. No result is expected.
+   *
+   * @param request  The sql command to generate the prepared statement.
+   * @param context  Per-call context.
+   * @param listener A stream of responses.
+   */
+  void closePreparedStatement(ActionClosePreparedStatementRequest request, CallContext context,
+                              StreamListener<Result> listener);
+
+  /**
+   * Gets information about a particular SQL query based data stream.
+   *
+   * @param command    The sql command to generate the data stream.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoStatement(CommandStatementQuery command, CallContext context,
+                                    FlightDescriptor descriptor);
+
+  /**
+   * Gets information about a particular prepared statement data stream.
+   *
+   * @param command    The prepared statement to generate the data stream.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoPreparedStatement(CommandPreparedStatementQuery command,
+                                            CallContext context, FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about a particular SQL query based data stream.
+   *
+   * @param command    The sql command to generate the data stream.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Schema for the stream.
+   */
+  SchemaResult getSchemaStatement(CommandStatementQuery command, CallContext context,
+                                  FlightDescriptor descriptor);
+
+  /**
+   * Returns data for a SQL query based data stream.
+   *
+   * @param command  The sql command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamStatement(CommandStatementQuery command, CallContext context, Ticket ticket,
+                          ServerStreamListener listener);
+
+  /**
+   * Returns data for a particular prepared statement query instance.
+   *
+   * @param command  The prepared statement to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamPreparedStatement(CommandPreparedStatementQuery command, CallContext context,
+                                  Ticket ticket, ServerStreamListener listener);
+
+  /**
+   * Accepts uploaded data for a particular SQL query based data stream.
+   * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+   *
+   * @param command      The sql command to generate the data stream.
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The result data stream.
+   * @return A runnable to process the stream.
+   */
+  Runnable acceptPutStatement(CommandStatementUpdate command, CallContext context,
+                              FlightStream flightStream, StreamListener<PutResult> ackStream);
+
+  /**
+   * Accepts uploaded data for a particular prepared statement data stream.
+   * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+   *
+   * @param command      The prepared statement to generate the data stream.
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The result data stream.
+   * @return A runnable to process the stream.
+   */
+  Runnable acceptPutPreparedStatementUpdate(CommandPreparedStatementUpdate command,
+                                            CallContext context, FlightStream flightStream,
+                                            StreamListener<PutResult> ackStream);
+
+  /**
+   * Accepts uploaded parameter values for a particular prepared statement query.
+   *
+   * @param command      The prepared statement the parameter values will bind to.
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The result data stream.
+   * @return A runnable to process the stream.
+   */
+  Runnable acceptPutPreparedStatementQuery(CommandPreparedStatementQuery command,
+                                           CallContext context, FlightStream flightStream,
+                                           StreamListener<PutResult> ackStream);
+
+  /**
+   * Returns the SQL Info of the server by returning a
+   * {@link CommandGetSqlInfo} in a {@link Result}.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoSqlInfo(CommandGetSqlInfo request, CallContext context,
+                                  FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get SQL info data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaSqlInfo() {
+    return new SchemaResult(Schemas.GET_SQL_INFO_SCHEMA);
+  }
+
+  /**
+   * Returns data for SQL info based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamSqlInfo(CommandGetSqlInfo command, CallContext context, Ticket ticket,
+                        ServerStreamListener listener);
+
+  /**
+   * Returns the available catalogs by returning a stream of
+   * {@link CommandGetCatalogs} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoCatalogs(CommandGetCatalogs request, CallContext context,
+                                   FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get catalogs data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaCatalogs() {
+    return new SchemaResult(Schemas.GET_CATALOGS_SCHEMA);
+  }
+
+  /**
+   * Returns data for catalogs based data stream.
+   *
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamCatalogs(CallContext context, Ticket ticket,
+                         ServerStreamListener listener);
+
+  /**
+   * Returns the available schemas by returning a stream of
+   * {@link CommandGetSchemas} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoSchemas(CommandGetSchemas request, CallContext context,
+                                  FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get schemas data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaSchemas() {
+    return new SchemaResult(Schemas.GET_SCHEMAS_SCHEMA);
+  }
+
+  /**
+   * Returns data for schemas based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamSchemas(CommandGetSchemas command, CallContext context, Ticket ticket,
+                        ServerStreamListener listener);
+
+  /**
+   * Returns the available tables by returning a stream of
+   * {@link CommandGetTables} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoTables(CommandGetTables request, CallContext context,
+                                 FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get tables data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaTables() {
+    return new SchemaResult(Schemas.GET_TABLES_SCHEMA);
+  }
+
+  /**
+   * Returns data for tables based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamTables(CommandGetTables command, CallContext context, Ticket ticket,
+                       ServerStreamListener listener);
+
+  /**
+   * Returns the available table types by returning a stream of
+   * {@link CommandGetTableTypes} objects in {@link Result} objects.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoTableTypes(CommandGetTableTypes request, CallContext context,
+                                     FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get table types data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaTableTypes() {
+    return new SchemaResult(Schemas.GET_TABLE_TYPES_SCHEMA);
+  }
+
+  /**
+   * Returns data for table types based data stream.
+   *
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamTableTypes(CallContext context, Ticket ticket, ServerStreamListener listener);
+
+  /**
+   * Returns the available primary keys by returning a stream of
+   * {@link CommandGetPrimaryKeys} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoPrimaryKeys(CommandGetPrimaryKeys request, CallContext context,
+                                      FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get primary keys data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaPrimaryKeys() {
+    final List<Field> fields = Arrays.asList(
+        Field.nullable("catalog_name", MinorType.VARCHAR.getType()),
+        Field.nullable("schema_name", MinorType.VARCHAR.getType()),
+        Field.nullable("table_name", MinorType.VARCHAR.getType()),
+        Field.nullable("column_name", MinorType.VARCHAR.getType()),
+        Field.nullable("key_sequence", MinorType.INT.getType()),
+        Field.nullable("key_name", MinorType.VARCHAR.getType()));
+
+    return new SchemaResult(new Schema(fields));
+  }
+
+  /**
+   * Returns data for primary keys based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamPrimaryKeys(CommandGetPrimaryKeys command, CallContext context, Ticket ticket,
+                            ServerStreamListener listener);
+
+  /**
+   * Retrieves a description of the foreign key columns that reference the given table's primary key columns
+   * {@link CommandGetExportedKeys} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoExportedKeys(CommandGetExportedKeys request, CallContext context,
+                                       FlightDescriptor descriptor);
+
+  /**
+   * Retrieves a description of the primary key columns that are referenced by given table's foreign key columns
+   * {@link CommandGetImportedKeys} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoImportedKeys(CommandGetImportedKeys request, CallContext context,
+                                       FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get imported and exported keys data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaForImportedAndExportedKeys() {
+    return new SchemaResult(Schemas.GET_IMPORTED_AND_EXPORTED_KEYS_SCHEMA);
+  }
+
+  /**
+   * Returns data for foreign keys based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamExportedKeys(CommandGetExportedKeys command, CallContext context, Ticket ticket,
+                             ServerStreamListener listener);
+
+  /**
+   * Returns data for foreign keys based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamImportedKeys(CommandGetImportedKeys command, CallContext context, Ticket ticket,
+                             ServerStreamListener listener);
+
+  /**
+   * Default schema templates for the {@link FlightSqlProducer}.
+   */
+  final class Schemas {

Review comment:
       static final class?

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/sql/FlightSqlExample.java
##########
@@ -0,0 +1,1768 @@
+/*
+ * 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.flight.sql;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.base.Strings.emptyToNull;
+import static com.google.common.base.Strings.isNullOrEmpty;
+import static com.google.protobuf.Any.pack;
+import static com.google.protobuf.ByteString.copyFrom;
+import static java.lang.String.format;
+import static java.util.Collections.singletonList;
+import static java.util.Objects.isNull;
+import static java.util.Optional.empty;
+import static java.util.UUID.randomUUID;
+import static java.util.stream.StreamSupport.stream;
+import static org.apache.arrow.adapter.jdbc.JdbcToArrow.sqlToArrowVectorIterator;
+import static org.apache.arrow.adapter.jdbc.JdbcToArrowUtils.jdbcToArrowSchema;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.Date;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Calendar;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.adapter.jdbc.ArrowVectorIterator;
+import org.apache.arrow.adapter.jdbc.JdbcFieldInfo;
+import org.apache.arrow.adapter.jdbc.JdbcToArrowConfig;
+import org.apache.arrow.adapter.jdbc.JdbcToArrowUtils;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.Criteria;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightEndpoint;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightRuntimeException;
+import org.apache.arrow.flight.FlightStatusCode;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.Location;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.BitVector;
+import org.apache.arrow.vector.DateDayVector;
+import org.apache.arrow.vector.DateMilliVector;
+import org.apache.arrow.vector.Decimal256Vector;
+import org.apache.arrow.vector.DecimalVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.Float4Vector;
+import org.apache.arrow.vector.Float8Vector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.LargeVarCharVector;
+import org.apache.arrow.vector.SmallIntVector;
+import org.apache.arrow.vector.TimeMicroVector;
+import org.apache.arrow.vector.TimeMilliVector;
+import org.apache.arrow.vector.TimeNanoVector;
+import org.apache.arrow.vector.TimeSecVector;
+import org.apache.arrow.vector.TimeStampMicroTZVector;
+import org.apache.arrow.vector.TimeStampMilliTZVector;
+import org.apache.arrow.vector.TimeStampNanoTZVector;
+import org.apache.arrow.vector.TimeStampSecTZVector;
+import org.apache.arrow.vector.TimeStampVector;
+import org.apache.arrow.vector.TinyIntVector;
+import org.apache.arrow.vector.UInt1Vector;
+import org.apache.arrow.vector.UInt2Vector;
+import org.apache.arrow.vector.UInt4Vector;
+import org.apache.arrow.vector.UInt8Vector;
+import org.apache.arrow.vector.VarBinaryVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorLoader;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.VectorUnloader;
+import org.apache.arrow.vector.complex.DenseUnionVector;
+import org.apache.arrow.vector.holders.NullableIntHolder;
+import org.apache.arrow.vector.holders.NullableVarCharHolder;
+import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.Text;
+import org.apache.commons.dbcp2.ConnectionFactory;
+import org.apache.commons.dbcp2.DriverManagerConnectionFactory;
+import org.apache.commons.dbcp2.PoolableConnection;
+import org.apache.commons.dbcp2.PoolableConnectionFactory;
+import org.apache.commons.dbcp2.PoolingDataSource;
+import org.apache.commons.pool2.ObjectPool;
+import org.apache.commons.pool2.impl.GenericObjectPool;
+import org.slf4j.Logger;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Message;
+import com.google.protobuf.ProtocolStringList;
+
+import io.grpc.Status;

Review comment:
       Ditto - use CallStatus, not io.grpc.Status.

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java
##########
@@ -0,0 +1,704 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.ActionType;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightProducer;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+import org.apache.arrow.vector.types.Types.MinorType;
+import org.apache.arrow.vector.types.UnionMode;
+import org.apache.arrow.vector.types.pojo.ArrowType.Union;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+import io.grpc.Status;
+
+/**
+ * API to Implement an Arrow Flight SQL producer.
+ */
+public interface FlightSqlProducer extends FlightProducer, AutoCloseable {
+  /**
+   * Depending on the provided command, method either:
+   * 1. Return information about a SQL query, or
+   * 2. Return information about a prepared statement. In this case, parameters binding is allowed.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return information about the given SQL query, or the given prepared statement.
+   */
+  @Override
+  default FlightInfo getFlightInfo(CallContext context, FlightDescriptor descriptor) {
+    final Any command = FlightSqlUtils.parseOrThrow(descriptor.getCommand());
+
+    if (command.is(CommandStatementQuery.class)) {
+      return getFlightInfoStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      return getFlightInfoPreparedStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      return getFlightInfoCatalogs(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetCatalogs.class), context, descriptor);
+    } else if (command.is(CommandGetSchemas.class)) {
+      return getFlightInfoSchemas(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetSchemas.class), context, descriptor);
+    } else if (command.is(CommandGetTables.class)) {
+      return getFlightInfoTables(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetTables.class), context, descriptor);
+    } else if (command.is(CommandGetTableTypes.class)) {
+      return getFlightInfoTableTypes(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetTableTypes.class), context, descriptor);
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      return getFlightInfoSqlInfo(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetSqlInfo.class), context, descriptor);
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      return getFlightInfoPrimaryKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetPrimaryKeys.class), context, descriptor);
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      return getFlightInfoExportedKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetExportedKeys.class), context, descriptor);
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      return getFlightInfoImportedKeys(
+          FlightSqlUtils.unpackOrThrow(command, CommandGetImportedKeys.class), context, descriptor);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Returns the schema of the result produced by the SQL query.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return the result set schema.
+   */
+  @Override
+  default SchemaResult getSchema(CallContext context, FlightDescriptor descriptor) {
+    final Any command = FlightSqlUtils.parseOrThrow(descriptor.getCommand());
+
+    if (command.is(CommandStatementQuery.class)) {
+      return getSchemaStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, descriptor);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      return getSchemaCatalogs();
+    } else if (command.is(CommandGetSchemas.class)) {
+      return getSchemaSchemas();
+    } else if (command.is(CommandGetTables.class)) {
+      return getSchemaTables();
+    } else if (command.is(CommandGetTableTypes.class)) {
+      return getSchemaTableTypes();
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      return getSchemaSqlInfo();
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      return getSchemaPrimaryKeys();
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      return getSchemaForImportedAndExportedKeys();
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      return getSchemaForImportedAndExportedKeys();
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Depending on the provided command, method either:
+   * 1. Return data for a stream produced by executing the provided SQL query, or
+   * 2. Return data for a prepared statement. In this case, parameters binding is allowed.
+   *
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  @Override
+  default void getStream(CallContext context, Ticket ticket, ServerStreamListener listener) {
+    final Any command;
+
+    try {
+      command = Any.parseFrom(ticket.getBytes());
+    } catch (InvalidProtocolBufferException e) {
+      listener.error(e);
+      return;
+    }
+
+    if (command.is(CommandStatementQuery.class)) {
+      getStreamStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementQuery.class), context, ticket, listener);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      getStreamPreparedStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class), context, ticket, listener);
+    } else if (command.is(CommandGetCatalogs.class)) {
+      getStreamCatalogs(context, ticket, listener);
+    } else if (command.is(CommandGetSchemas.class)) {
+      getStreamSchemas(FlightSqlUtils.unpackOrThrow(command, CommandGetSchemas.class), context, ticket, listener);
+    } else if (command.is(CommandGetTables.class)) {
+      getStreamTables(FlightSqlUtils.unpackOrThrow(command, CommandGetTables.class), context, ticket, listener);
+    } else if (command.is(CommandGetTableTypes.class)) {
+      getStreamTableTypes(context, ticket, listener);
+    } else if (command.is(CommandGetSqlInfo.class)) {
+      getStreamSqlInfo(FlightSqlUtils.unpackOrThrow(command, CommandGetSqlInfo.class), context, ticket, listener);
+    } else if (command.is(CommandGetPrimaryKeys.class)) {
+      getStreamPrimaryKeys(FlightSqlUtils.unpackOrThrow(command, CommandGetPrimaryKeys.class),
+          context, ticket, listener);
+    } else if (command.is(CommandGetExportedKeys.class)) {
+      getStreamExportedKeys(FlightSqlUtils.unpackOrThrow(command, CommandGetExportedKeys.class),
+          context, ticket, listener);
+    } else if (command.is(CommandGetImportedKeys.class)) {
+      getStreamImportedKeys(FlightSqlUtils.unpackOrThrow(command, CommandGetImportedKeys.class),
+          context, ticket, listener);
+    } else {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+  }
+
+  /**
+   * Depending on the provided command, method either:
+   * 1. Execute provided SQL query as an update statement, or
+   * 2. Execute provided update SQL query prepared statement. In this case, parameters binding
+   * is allowed, or
+   * 3. Binds parameters to the provided prepared statement.
+   *
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The data stream listener for update result acknowledgement.
+   * @return a Runnable to process the stream.
+   */
+  @Override
+  default Runnable acceptPut(CallContext context, FlightStream flightStream, StreamListener<PutResult> ackStream) {
+    final Any command = FlightSqlUtils.parseOrThrow(flightStream.getDescriptor().getCommand());
+
+    if (command.is(CommandStatementUpdate.class)) {
+      return acceptPutStatement(
+          FlightSqlUtils.unpackOrThrow(command, CommandStatementUpdate.class),
+          context, flightStream, ackStream);
+    } else if (command.is(CommandPreparedStatementUpdate.class)) {
+      return acceptPutPreparedStatementUpdate(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementUpdate.class),
+          context, flightStream, ackStream);
+    } else if (command.is(CommandPreparedStatementQuery.class)) {
+      return acceptPutPreparedStatementQuery(
+          FlightSqlUtils.unpackOrThrow(command, CommandPreparedStatementQuery.class),
+          context, flightStream, ackStream);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Lists all available Flight SQL actions.
+   *
+   * @param context  Per-call context.
+   * @param listener An interface for sending data back to the client.
+   */
+  @Override
+  default void listActions(CallContext context, StreamListener<ActionType> listener) {
+    FlightSqlUtils.FLIGHT_SQL_ACTIONS.forEach(listener::onNext);
+    listener.onCompleted();
+  }
+
+  /**
+   * Performs the requested Flight SQL action.
+   *
+   * @param context  Per-call context.
+   * @param action   Client-supplied parameters.
+   * @param listener A stream of responses.
+   */
+  @Override
+  default void doAction(CallContext context, Action action, StreamListener<Result> listener) {
+    final String actionType = action.getType();
+    if (actionType.equals(FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType())) {
+      final ActionCreatePreparedStatementRequest request = FlightSqlUtils.unpackAndParseOrThrow(action.getBody(),
+          ActionCreatePreparedStatementRequest.class);
+      createPreparedStatement(request, context, listener);
+    } else if (actionType.equals(FlightSqlUtils.FLIGHT_SQL_CLOSEPREPAREDSTATEMENT.getType())) {
+      final ActionClosePreparedStatementRequest request = FlightSqlUtils.unpackAndParseOrThrow(action.getBody(),
+          ActionClosePreparedStatementRequest.class);
+      closePreparedStatement(request, context, listener);
+    }
+
+    throw Status.INVALID_ARGUMENT.asRuntimeException();
+  }
+
+  /**
+   * Creates a prepared statement on the server and returns a handle and metadata for in a
+   * {@link ActionCreatePreparedStatementResult} object in a {@link Result}
+   * object.
+   *
+   * @param request  The sql command to generate the prepared statement.
+   * @param context  Per-call context.
+   * @param listener A stream of responses.
+   */
+  void createPreparedStatement(ActionCreatePreparedStatementRequest request, CallContext context,
+                               StreamListener<Result> listener);
+
+  /**
+   * Closes a prepared statement on the server. No result is expected.
+   *
+   * @param request  The sql command to generate the prepared statement.
+   * @param context  Per-call context.
+   * @param listener A stream of responses.
+   */
+  void closePreparedStatement(ActionClosePreparedStatementRequest request, CallContext context,
+                              StreamListener<Result> listener);
+
+  /**
+   * Gets information about a particular SQL query based data stream.
+   *
+   * @param command    The sql command to generate the data stream.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoStatement(CommandStatementQuery command, CallContext context,
+                                    FlightDescriptor descriptor);
+
+  /**
+   * Gets information about a particular prepared statement data stream.
+   *
+   * @param command    The prepared statement to generate the data stream.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoPreparedStatement(CommandPreparedStatementQuery command,
+                                            CallContext context, FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about a particular SQL query based data stream.
+   *
+   * @param command    The sql command to generate the data stream.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Schema for the stream.
+   */
+  SchemaResult getSchemaStatement(CommandStatementQuery command, CallContext context,
+                                  FlightDescriptor descriptor);
+
+  /**
+   * Returns data for a SQL query based data stream.
+   *
+   * @param command  The sql command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamStatement(CommandStatementQuery command, CallContext context, Ticket ticket,
+                          ServerStreamListener listener);
+
+  /**
+   * Returns data for a particular prepared statement query instance.
+   *
+   * @param command  The prepared statement to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamPreparedStatement(CommandPreparedStatementQuery command, CallContext context,
+                                  Ticket ticket, ServerStreamListener listener);
+
+  /**
+   * Accepts uploaded data for a particular SQL query based data stream.
+   * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+   *
+   * @param command      The sql command to generate the data stream.
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The result data stream.
+   * @return A runnable to process the stream.
+   */
+  Runnable acceptPutStatement(CommandStatementUpdate command, CallContext context,
+                              FlightStream flightStream, StreamListener<PutResult> ackStream);
+
+  /**
+   * Accepts uploaded data for a particular prepared statement data stream.
+   * <p>`PutResult`s must be in the form of a {@link DoPutUpdateResult}.
+   *
+   * @param command      The prepared statement to generate the data stream.
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The result data stream.
+   * @return A runnable to process the stream.
+   */
+  Runnable acceptPutPreparedStatementUpdate(CommandPreparedStatementUpdate command,
+                                            CallContext context, FlightStream flightStream,
+                                            StreamListener<PutResult> ackStream);
+
+  /**
+   * Accepts uploaded parameter values for a particular prepared statement query.
+   *
+   * @param command      The prepared statement the parameter values will bind to.
+   * @param context      Per-call context.
+   * @param flightStream The data stream being uploaded.
+   * @param ackStream    The result data stream.
+   * @return A runnable to process the stream.
+   */
+  Runnable acceptPutPreparedStatementQuery(CommandPreparedStatementQuery command,
+                                           CallContext context, FlightStream flightStream,
+                                           StreamListener<PutResult> ackStream);
+
+  /**
+   * Returns the SQL Info of the server by returning a
+   * {@link CommandGetSqlInfo} in a {@link Result}.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoSqlInfo(CommandGetSqlInfo request, CallContext context,
+                                  FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get SQL info data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaSqlInfo() {
+    return new SchemaResult(Schemas.GET_SQL_INFO_SCHEMA);
+  }
+
+  /**
+   * Returns data for SQL info based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamSqlInfo(CommandGetSqlInfo command, CallContext context, Ticket ticket,
+                        ServerStreamListener listener);
+
+  /**
+   * Returns the available catalogs by returning a stream of
+   * {@link CommandGetCatalogs} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoCatalogs(CommandGetCatalogs request, CallContext context,
+                                   FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get catalogs data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaCatalogs() {
+    return new SchemaResult(Schemas.GET_CATALOGS_SCHEMA);
+  }
+
+  /**
+   * Returns data for catalogs based data stream.
+   *
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamCatalogs(CallContext context, Ticket ticket,
+                         ServerStreamListener listener);
+
+  /**
+   * Returns the available schemas by returning a stream of
+   * {@link CommandGetSchemas} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoSchemas(CommandGetSchemas request, CallContext context,
+                                  FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get schemas data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaSchemas() {
+    return new SchemaResult(Schemas.GET_SCHEMAS_SCHEMA);
+  }
+
+  /**
+   * Returns data for schemas based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamSchemas(CommandGetSchemas command, CallContext context, Ticket ticket,
+                        ServerStreamListener listener);
+
+  /**
+   * Returns the available tables by returning a stream of
+   * {@link CommandGetTables} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoTables(CommandGetTables request, CallContext context,
+                                 FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get tables data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaTables() {
+    return new SchemaResult(Schemas.GET_TABLES_SCHEMA);
+  }
+
+  /**
+   * Returns data for tables based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamTables(CommandGetTables command, CallContext context, Ticket ticket,
+                       ServerStreamListener listener);
+
+  /**
+   * Returns the available table types by returning a stream of
+   * {@link CommandGetTableTypes} objects in {@link Result} objects.
+   *
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoTableTypes(CommandGetTableTypes request, CallContext context,
+                                     FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get table types data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaTableTypes() {
+    return new SchemaResult(Schemas.GET_TABLE_TYPES_SCHEMA);
+  }
+
+  /**
+   * Returns data for table types based data stream.
+   *
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamTableTypes(CallContext context, Ticket ticket, ServerStreamListener listener);
+
+  /**
+   * Returns the available primary keys by returning a stream of
+   * {@link CommandGetPrimaryKeys} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoPrimaryKeys(CommandGetPrimaryKeys request, CallContext context,
+                                      FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get primary keys data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaPrimaryKeys() {
+    final List<Field> fields = Arrays.asList(
+        Field.nullable("catalog_name", MinorType.VARCHAR.getType()),
+        Field.nullable("schema_name", MinorType.VARCHAR.getType()),
+        Field.nullable("table_name", MinorType.VARCHAR.getType()),
+        Field.nullable("column_name", MinorType.VARCHAR.getType()),
+        Field.nullable("key_sequence", MinorType.INT.getType()),
+        Field.nullable("key_name", MinorType.VARCHAR.getType()));
+
+    return new SchemaResult(new Schema(fields));
+  }
+
+  /**
+   * Returns data for primary keys based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamPrimaryKeys(CommandGetPrimaryKeys command, CallContext context, Ticket ticket,
+                            ServerStreamListener listener);
+
+  /**
+   * Retrieves a description of the foreign key columns that reference the given table's primary key columns
+   * {@link CommandGetExportedKeys} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoExportedKeys(CommandGetExportedKeys request, CallContext context,
+                                       FlightDescriptor descriptor);
+
+  /**
+   * Retrieves a description of the primary key columns that are referenced by given table's foreign key columns
+   * {@link CommandGetImportedKeys} objects in {@link Result} objects.
+   *
+   * @param request    request filter parameters.
+   * @param context    Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfoImportedKeys(CommandGetImportedKeys request, CallContext context,
+                                       FlightDescriptor descriptor);
+
+  /**
+   * Gets schema about the get imported and exported keys data stream.
+   *
+   * @return Schema for the stream.
+   */
+  default SchemaResult getSchemaForImportedAndExportedKeys() {
+    return new SchemaResult(Schemas.GET_IMPORTED_AND_EXPORTED_KEYS_SCHEMA);
+  }
+
+  /**
+   * Returns data for foreign keys based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamExportedKeys(CommandGetExportedKeys command, CallContext context, Ticket ticket,
+                             ServerStreamListener listener);
+
+  /**
+   * Returns data for foreign keys based data stream.
+   *
+   * @param command  The command to generate the data stream.
+   * @param context  Per-call context.
+   * @param ticket   The application-defined ticket identifying this stream.
+   * @param listener An interface for sending data back to the client.
+   */
+  void getStreamImportedKeys(CommandGetImportedKeys command, CallContext context, Ticket ticket,
+                             ServerStreamListener listener);
+
+  /**
+   * Default schema templates for the {@link FlightSqlProducer}.
+   */
+  final class Schemas {
+    public static final Schema GET_TABLES_SCHEMA = new Schema(Arrays.asList(
+        Field.nullable("catalog_name", MinorType.VARCHAR.getType()),
+        Field.nullable("schema_name", MinorType.VARCHAR.getType()),
+        Field.nullable("table_name", MinorType.VARCHAR.getType()),
+        Field.nullable("table_type", MinorType.VARCHAR.getType()),
+        Field.nullable("table_schema", MinorType.VARBINARY.getType())));
+    public static final Schema GET_TABLES_SCHEMA_NO_SCHEMA = new Schema(Arrays.asList(
+        Field.nullable("catalog_name", MinorType.VARCHAR.getType()),
+        Field.nullable("schema_name", MinorType.VARCHAR.getType()),
+        Field.nullable("table_name", MinorType.VARCHAR.getType()),
+        Field.nullable("table_type", MinorType.VARCHAR.getType())));
+    public static final Schema GET_CATALOGS_SCHEMA = new Schema(
+        Collections.singletonList(new Field("catalog_name", FieldType.nullable(MinorType.VARCHAR.getType()), null)));
+    public static final Schema GET_TABLE_TYPES_SCHEMA =
+        new Schema(Collections.singletonList(Field.nullable("table_type", MinorType.VARCHAR.getType())));
+    public static final Schema GET_SCHEMAS_SCHEMA = new Schema(
+        Arrays.asList(Field.nullable("catalog_name", MinorType.VARCHAR.getType()),
+            Field.nullable("schema_name", MinorType.VARCHAR.getType())));
+    public static final Schema GET_IMPORTED_AND_EXPORTED_KEYS_SCHEMA = new Schema(Arrays.asList(
+        Field.nullable("pk_catalog_name", MinorType.VARCHAR.getType()),
+        Field.nullable("pk_schema_name", MinorType.VARCHAR.getType()),
+        Field.nullable("pk_table_name", MinorType.VARCHAR.getType()),
+        Field.nullable("pk_column_name", MinorType.VARCHAR.getType()),
+        Field.nullable("fk_catalog_name", MinorType.VARCHAR.getType()),
+        Field.nullable("fk_schema_name", MinorType.VARCHAR.getType()),
+        Field.nullable("fk_table_name", MinorType.VARCHAR.getType()),
+        Field.nullable("fk_column_name", MinorType.VARCHAR.getType()),
+        Field.nullable("key_sequence", MinorType.INT.getType()),
+        Field.nullable("fk_key_name", MinorType.VARCHAR.getType()),
+        Field.nullable("pk_key_name", MinorType.VARCHAR.getType()),
+        Field.nullable("update_rule", MinorType.INT.getType()),
+        Field.nullable("delete_rule", MinorType.INT.getType())));
+    public static final Schema GET_SQL_INFO_SCHEMA =
+        new Schema(Arrays.asList(
+            Field.nullable("info_name", MinorType.INT.getType()),
+            new Field("value",
+                // dense_union<string_value: string, int_value: int32, bigint_value: int64, int32_bitmask: int32>
+                new FieldType(true, new Union(UnionMode.Dense, new int[] {0, 1, 2, 3}), /*dictionary=*/null),
+                Arrays.asList(
+                    Field.nullable("string_value", MinorType.VARCHAR.getType()),
+                    Field.nullable("int_value", MinorType.INT.getType()),
+                    Field.nullable("bigint_value", MinorType.BIGINT.getType()),
+                    Field.nullable("int32_bitmask", MinorType.INT.getType())))));
+
+    private Schemas() {
+      // Prevent instantiation.
+    }
+  }
+
+  /**
+   * Reserved options for the SQL command `GetSqlInfo` used by {@link FlightSqlProducer}.
+   */
+  final class SqlInfo {

Review comment:
       static final class with private constructor?

##########
File path: java/flight/flight-sql/src/test/java/org/apache/arrow/flight/sql/StatementContext.java
##########
@@ -0,0 +1,92 @@
+/*
+ * 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.flight.sql;
+
+import java.io.Serializable;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.Objects;
+import java.util.Optional;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.util.AutoCloseables;
+import org.apache.arrow.util.Preconditions;
+
+/**
+ * Context for {@link T} to be persisted in memory in between {@link FlightSqlProducer} calls.
+ *
+ * @param <T> the {@link Statement} to be persisted.
+ */
+public final class StatementContext<T extends Statement> implements AutoCloseable, Serializable {

Review comment:
       (Notably I don't think we should encourage use of Serializable.)

##########
File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java
##########
@@ -0,0 +1,504 @@
+/*
+ * 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.flight.sql;
+
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionClosePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementRequest;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetCatalogs;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetExportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetImportedKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetPrimaryKeys;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSchemas;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetSqlInfo;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTableTypes;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandGetTables;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementQuery;
+import static org.apache.arrow.flight.sql.impl.FlightSql.CommandStatementUpdate;
+import static org.apache.arrow.flight.sql.impl.FlightSql.DoPutUpdateResult;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.sql.SQLException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.Nullable;
+
+import org.apache.arrow.flight.Action;
+import org.apache.arrow.flight.CallOption;
+import org.apache.arrow.flight.FlightClient;
+import org.apache.arrow.flight.FlightDescriptor;
+import org.apache.arrow.flight.FlightInfo;
+import org.apache.arrow.flight.FlightStream;
+import org.apache.arrow.flight.PutResult;
+import org.apache.arrow.flight.Result;
+import org.apache.arrow.flight.SchemaResult;
+import org.apache.arrow.flight.SyncPutListener;
+import org.apache.arrow.flight.Ticket;
+import org.apache.arrow.flight.sql.impl.FlightSql;
+import org.apache.arrow.flight.sql.impl.FlightSql.ActionCreatePreparedStatementResult;
+import org.apache.arrow.flight.sql.impl.FlightSql.CommandPreparedStatementQuery;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.StringValue;
+
+import io.grpc.Status;
+
+/**
+ * Flight client with Flight SQL semantics.
+ */
+public class FlightSqlClient {
+  private FlightClient client;
+
+  public FlightSqlClient(FlightClient client) {
+    this.client = client;
+  }
+
+  /**
+   * Execute a query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo execute(String query) {
+    final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder();
+    builder.setQuery(query);
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Execute an update query on the server.
+   *
+   * @param query The query to execute.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public long executeUpdate(String query) {
+    final CommandStatementUpdate.Builder builder = CommandStatementUpdate.newBuilder();
+    builder.setQuery(query);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    final SyncPutListener putListener = new SyncPutListener();
+    client.startPut(descriptor, VectorSchemaRoot.of(), putListener);
+
+    try {
+      final PutResult read = putListener.read();
+      try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+        final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+        return doPutUpdateResult.getRecordCount();
+      }
+    } catch (InterruptedException | ExecutionException | InvalidProtocolBufferException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  /**
+   * Request a list of catalogs.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getCatalogs() {
+    final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of schemas.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern) {
+    final CommandGetSchemas.Builder builder = CommandGetSchemas.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Get schema for a stream.
+   *
+   * @param descriptor The descriptor for the stream.
+   * @param options    RPC-layer hints for this call.
+   */
+  public SchemaResult getSchema(FlightDescriptor descriptor, CallOption... options) {
+    return this.client.getSchema(descriptor, options);
+  }
+
+  /**
+   * Retrieve a stream from the server.
+   *
+   * @param ticket  The ticket granting access to the data stream.
+   * @param options RPC-layer hints for this call.
+   */
+  public FlightStream getStream(Ticket ticket, CallOption... options) {
+    return this.client.getStream(ticket, options);
+  }
+
+  /**
+   * Request a set of Flight SQL metadata.
+   *
+   * @param info The set of metadata to retrieve. None to retrieve all metadata.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getSqlInfo(final @Nullable int... info) {
+    final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder();
+    for (final int pieceOfInfo : Objects.isNull(info) ? new int[0] : info) {
+      builder.addInfo(pieceOfInfo);
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of tables.
+   *
+   * @param catalog             The catalog.
+   * @param schemaFilterPattern The schema filter pattern.
+   * @param tableFilterPattern  The table filter pattern.
+   * @param tableTypes          The table types to include.
+   * @param includeSchema       True to include the schema upon return, false to not include the schema.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTables(final @Nullable String catalog, final @Nullable String schemaFilterPattern,
+                              final @Nullable String tableFilterPattern, final List<String> tableTypes,
+                              final boolean includeSchema) {
+    final CommandGetTables.Builder builder = CommandGetTables.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schemaFilterPattern != null) {
+      builder.setSchemaFilterPattern(StringValue.newBuilder().setValue(schemaFilterPattern).build());
+    }
+
+    if (tableFilterPattern != null) {
+      builder.setTableNameFilterPattern(StringValue.newBuilder().setValue(tableFilterPattern).build());
+    }
+
+    if (tableTypes != null) {
+      builder.addAllTableTypes(tableTypes);
+    }
+    builder.setIncludeSchema(includeSchema);
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request the primary keys for a table.
+   *
+   * @param catalog The catalog.
+   * @param schema  The schema.
+   * @param table   The table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getPrimaryKeys(final @Nullable String catalog, final @Nullable String schema,
+                                   final @Nullable String table) {
+    final CommandGetPrimaryKeys.Builder builder = CommandGetPrimaryKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    if (table != null) {
+      builder.setTable(StringValue.newBuilder().setValue(table).build());
+    }
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which exports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The foreign key table catalog.
+   * @param schema  The foreign key table schema.
+   * @param table   The foreign key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getExportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetExportedKeys.Builder builder = CommandGetExportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request to get info about keys on a table. The table, which imports the foreign keys, parameter must be specified.
+   *
+   * @param catalog The primary key table catalog.
+   * @param schema  The primary key table schema.
+   * @param table   The primary key table.
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getImportedKeys(String catalog, String schema, String table) {
+    if (null == table) {
+      throw Status.INVALID_ARGUMENT.asRuntimeException();
+    }
+
+    final CommandGetImportedKeys.Builder builder = CommandGetImportedKeys.newBuilder();
+
+    if (catalog != null) {
+      builder.setCatalog(StringValue.newBuilder().setValue(catalog).build());
+    }
+
+    if (schema != null) {
+      builder.setSchema(StringValue.newBuilder().setValue(schema).build());
+    }
+
+    builder.setTable(table).build();
+
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Request a list of table types.
+   *
+   * @return a FlightInfo object representing the stream(s) to fetch.
+   */
+  public FlightInfo getTableTypes() {
+    final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder();
+    final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray());
+    return client.getInfo(descriptor);
+  }
+
+  /**
+   * Create a prepared statement on the server.
+   *
+   * @param query The query to prepare.
+   * @return The representation of the prepared statement which exists on the server.
+   */
+  public PreparedStatement prepare(String query) {
+    return new PreparedStatement(client, query);
+  }
+
+  /**
+   * Helper class to encapsulate Flight SQL prepared statement logic.
+   */
+  public static class PreparedStatement implements Closeable {
+    private final FlightClient client;
+    private final ActionCreatePreparedStatementResult preparedStatementResult;
+    private AtomicLong invocationCount;
+    private boolean isClosed;
+    private Schema resultSetSchema = null;
+    private Schema parameterSchema = null;
+    private VectorSchemaRoot parameterBindingRoot;
+
+    /**
+     * Constructor.
+     *
+     * @param client The client. FlightSqlPreparedStatement does not maintain this resource.
+     * @param sql    The query.
+     */
+    public PreparedStatement(FlightClient client, String sql) {
+      this.client = client;
+
+      final Iterator<Result> preparedStatementResults = client.doAction(new Action(
+          FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType(),
+          Any.pack(ActionCreatePreparedStatementRequest
+              .newBuilder()
+              .setQuery(sql)
+              .build())
+              .toByteArray()));
+
+      preparedStatementResult = FlightSqlUtils.unpackAndParseOrThrow(
+          preparedStatementResults.next().getBody(),
+          ActionCreatePreparedStatementResult.class);
+
+      invocationCount = new AtomicLong(0);
+      isClosed = false;
+    }
+
+    /**
+     * Set the {@link VectorSchemaRoot} containing the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     * @param parameterBindingRoot  a {@link VectorSchemaRoot} object contain the values to be used in the
+     *                              PreparedStatement setters.
+     */
+    public void setParameters(VectorSchemaRoot parameterBindingRoot) {
+      this.parameterBindingRoot = parameterBindingRoot;
+    }
+
+    /**
+     * Empty the {@link VectorSchemaRoot} that contains the parameter binding from a preparedStatemnt
+     * operation.
+     *
+     */
+    public void clearParameters() {
+      this.parameterBindingRoot = null;
+    }
+
+    /**
+     * Returns the Schema of the resultset.
+     *
+     * @return the Schema of the resultset.
+     */
+    public Schema getResultSetSchema() {
+      if (resultSetSchema == null && preparedStatementResult.getDatasetSchema() != null) {
+        resultSetSchema = Schema.deserialize(preparedStatementResult.getDatasetSchema().asReadOnlyByteBuffer());
+      }
+      return resultSetSchema;
+    }
+
+    /**
+     * Returns the Schema of the parameters.
+     *
+     * @return the Schema of the parameters.
+     */
+    public Schema getParameterSchema() {
+      if (parameterSchema == null && preparedStatementResult.getParameterSchema() != null) {
+        parameterSchema = Schema.deserialize(preparedStatementResult.getParameterSchema().asReadOnlyByteBuffer());
+      }
+      return parameterSchema;
+    }
+
+    /**
+     * Executes the prepared statement query on the server.
+     *
+     * @return a FlightInfo object representing the stream(s) to fetch.
+     * @throws IOException if the PreparedStatement is closed.
+     */
+    public FlightInfo execute() throws IOException {
+      if (isClosed) {
+        throw new IllegalStateException("Prepared statement has already been closed on the server.");
+      }
+
+      final FlightDescriptor descriptor = FlightDescriptor
+          .command(Any.pack(CommandPreparedStatementQuery.newBuilder()
+              .setClientExecutionHandle(
+                  ByteString.copyFrom(ByteBuffer.allocate(Long.BYTES).putLong(invocationCount.getAndIncrement())))
+              .setPreparedStatementHandle(preparedStatementResult.getPreparedStatementHandle())
+              .build())
+              .toByteArray());
+
+      if (parameterBindingRoot != null) {
+        final SyncPutListener putListener = new SyncPutListener();
+
+        FlightClient.ClientStreamListener listener =
+            client.startPut(descriptor, this.parameterBindingRoot, putListener);
+
+        listener.putNext();
+        listener.completed();
+      }
+
+      return client.getInfo(descriptor);
+    }
+
+    /**
+     * Executes the prepared statement update on the server.
+     */
+    public long executeUpdate() throws SQLException {
+      if (isClosed) {
+        throw new IllegalStateException("Prepared statement has already been closed on the server.");
+      }
+
+      final FlightDescriptor descriptor = FlightDescriptor
+          .command(Any.pack(FlightSql.CommandPreparedStatementUpdate.newBuilder()
+              .setClientExecutionHandle(
+                  ByteString.copyFrom(ByteBuffer.allocate(Long.BYTES).putLong(invocationCount.getAndIncrement())))
+              .setPreparedStatementHandle(preparedStatementResult.getPreparedStatementHandle())
+              .build())
+              .toByteArray());
+
+      if (this.parameterBindingRoot == null) {
+        this.parameterBindingRoot = VectorSchemaRoot.of();
+      }
+
+      final SyncPutListener putListener = new SyncPutListener();
+      final FlightClient.ClientStreamListener listener =
+          client.startPut(descriptor, this.parameterBindingRoot, putListener);
+
+      listener.putNext();
+      listener.completed();
+
+      try {
+        final PutResult read = putListener.read();
+        try (final ArrowBuf metadata = read.getApplicationMetadata()) {
+          final FlightSql.DoPutUpdateResult doPutUpdateResult =
+              FlightSql.DoPutUpdateResult.parseFrom(metadata.nioBuffer());
+          return doPutUpdateResult.getRecordCount();
+        }
+      } catch (InterruptedException | InvalidProtocolBufferException | ExecutionException e) {
+        throw new SQLException(e);

Review comment:
       We should be consistent about what exceptions we use. We don't use SQLException for anything else.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org