lidavidm commented on a change in pull request #10906: URL: https://github.com/apache/arrow/pull/10906#discussion_r690626507
########## File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java ########## @@ -0,0 +1,567 @@ +/* + * 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.nio.ByteBuffer; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +import org.apache.arrow.flight.Action; +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallStatus; +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.util.Preconditions; +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; + +/** + * Flight client with Flight SQL semantics. + */ +public class FlightSqlClient { + private final FlightClient client; + + public FlightSqlClient(final FlightClient client) { + this.client = Objects.requireNonNull(client, "Client cannot be null!"); + } + + /** + * Execute a query on the server. + * + * @param query The query to execute. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo execute(final String query, final CallOption... options) { + final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder(); + builder.setQuery(query); + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * Execute an update query on the server. + * + * @param query The query to execute. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public long executeUpdate(final String query, final CallOption... options) { + 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, options); + + try { + final PutResult read = putListener.read(); + try (final ArrowBuf metadata = read.getApplicationMetadata()) { + final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer()); + return doPutUpdateResult.getRecordCount(); + } + } catch (final InterruptedException | ExecutionException e) { + throw CallStatus.CANCELLED.withCause(e).toRuntimeException(); + } catch (final InvalidProtocolBufferException e) { + throw CallStatus.INVALID_ARGUMENT.withCause(e).toRuntimeException(); Review comment: nit: this seems more like `INTERNAL` since the server sent the wrong/an invalid Protobuf ########## 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: Do we want to order by column_name? ########## File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlProducer.java ########## @@ -0,0 +1,720 @@ +/* + * 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.CallStatus; +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; +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; + +/** + * 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 CallStatus.INVALID_ARGUMENT.withDescription("The defined request is invalid.").toRuntimeException(); + } + + /** + * 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 CallStatus.INVALID_ARGUMENT.withDescription("Invalid command provided.").toRuntimeException(); + } + + /** + * 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 CallStatus.INVALID_ARGUMENT.withDescription("The defined request is invalid.").toRuntimeException(); + } + } + + /** + * 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 CallStatus.INVALID_ARGUMENT.withDescription("The defined request is invalid.").toRuntimeException(); + } + + /** + * 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 CallStatus.INVALID_ARGUMENT.withDescription("Invalid action provided.").toRuntimeException(); + } + + /** + * 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", new ArrowType.Int(8, false)), + Field.nullable("delete_rule", new ArrowType.Int(8, false)))); + public static final Schema GET_SQL_INFO_SCHEMA = + new Schema(Arrays.asList( + Field.nullable("info_name", new ArrowType.Int(32, false)), + 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 { + public static final int FLIGHT_SQL_SERVER_NAME = 0; + public static final int FLIGHT_SQL_SERVER_VERSION = 1; + public static final int FLIGHT_SQL_SERVER_ARROW_VERSION = 2; + public static final int FLIGHT_SQL_SERVER_READ_ONLY = 3; + public static final int SQL_DDL_CATALOG = 500; + public static final int SQL_DDL_SCHEMA = 501; + public static final int SQL_DDL_TABLE = 502; + public static final int SQL_IDENTIFIER_CASE = 503; + public static final int SQL_IDENTIFIER_QUOTE_CHAR = 504; + public static final int SQL_QUOTED_IDENTIFIER_CASE = 505; + } + + /** + * Update/delete rules for {@link FlightSqlProducer#getStreamImportedKeys} and + * {@link FlightSqlProducer#getStreamExportedKeys}. + */ + final class UpdateDeleteRules { Review comment: If I'm not mistaken, this isn't used anywhere? If it is to be used, might it make more sense as an enum? (If it's to appear in an API anywhere.) ########## File path: format/FlightSql.proto ########## @@ -0,0 +1,449 @@ +/* + * 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"; +import "google/protobuf/descriptor.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: uint32, + * 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 { + option (experimental) = true; + + /* + * 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 - Range [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 for future use. + * 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 { + option (experimental) = true; +} + +/* + * 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 { + option (experimental) = true; + + /* + * 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 (schema of the table as described in Schema.fbs::Schema, it is serialized as an IPC message.) + * > + * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type. + */ +message CommandGetTables { + option (experimental) = true; + + /* + * 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 { + option (experimental) = true; +} + +/* + * 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_name: utf8 + * key_sequence: int, + * > + * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence. + */ +message CommandGetPrimaryKeys { + option (experimental) = true; + + // 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: uint1, + * delete_rule: uint1 + * > + * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence. Review comment: Which catalog_name, which schema_name, etc.? ########## 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: Got it, thanks. ########## File path: java/flight/flight-sql/src/main/java/org/apache/arrow/flight/sql/FlightSqlClient.java ########## @@ -0,0 +1,567 @@ +/* + * 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.nio.ByteBuffer; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Collectors; + +import org.apache.arrow.flight.Action; +import org.apache.arrow.flight.CallOption; +import org.apache.arrow.flight.CallStatus; +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.util.Preconditions; +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; + +/** + * Flight client with Flight SQL semantics. + */ +public class FlightSqlClient { + private final FlightClient client; + + public FlightSqlClient(final FlightClient client) { + this.client = Objects.requireNonNull(client, "Client cannot be null!"); + } + + /** + * Execute a query on the server. + * + * @param query The query to execute. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo execute(final String query, final CallOption... options) { + final CommandStatementQuery.Builder builder = CommandStatementQuery.newBuilder(); + builder.setQuery(query); + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * Execute an update query on the server. + * + * @param query The query to execute. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public long executeUpdate(final String query, final CallOption... options) { + 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, options); + + try { + final PutResult read = putListener.read(); + try (final ArrowBuf metadata = read.getApplicationMetadata()) { + final DoPutUpdateResult doPutUpdateResult = DoPutUpdateResult.parseFrom(metadata.nioBuffer()); + return doPutUpdateResult.getRecordCount(); + } + } catch (final InterruptedException | ExecutionException e) { + throw CallStatus.CANCELLED.withCause(e).toRuntimeException(); + } catch (final InvalidProtocolBufferException e) { + throw CallStatus.INVALID_ARGUMENT.withCause(e).toRuntimeException(); + } + } + + /** + * Request a list of catalogs. + * + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getCatalogs(final CallOption... options) { + final CommandGetCatalogs.Builder builder = CommandGetCatalogs.newBuilder(); + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * Request a list of schemas. + * + * @param catalog The catalog. + * @param schemaFilterPattern The schema filter pattern. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getSchemas(final String catalog, final String schemaFilterPattern, final CallOption... options) { + 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, options); + } + + /** + * 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 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 client.getStream(ticket, options); + } + + /** + * Request a set of Flight SQL metadata. + * + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getSqlInfo(final int... info) { + return getSqlInfo(info, new CallOption[0]); + } + + /** + * Request a set of Flight SQL metadata. + * + * @param info The set of metadata to retrieve. None to retrieve all metadata. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getSqlInfo(final int[] info, final CallOption... options) { + return getSqlInfo(Arrays.stream(info).boxed().collect(Collectors.toList()), options); + } + + /** + * Request a set of Flight SQL metadata. + * + * @param info The set of metadata to retrieve. None to retrieve all metadata. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getSqlInfo(final List<Integer> info, final CallOption... options) { + final CommandGetSqlInfo.Builder builder = CommandGetSqlInfo.newBuilder(); + builder.addAllInfo(info); + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * 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. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getTables(final String catalog, final String schemaFilterPattern, + final String tableFilterPattern, final List<String> tableTypes, + final boolean includeSchema, final CallOption... options) { + 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, options); + } + + /** + * Request the primary keys for a table. + * + * @param catalog The catalog. + * @param schema The schema. + * @param table The table. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getPrimaryKeys(final String catalog, final String schema, + final String table, final CallOption... options) { + 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, options); + } + + /** + * Retrieves a description about the foreign key columns that reference the primary key columns of the given table. + * + * @param catalog The foreign key table catalog. + * @param schema The foreign key table schema. + * @param table The foreign key table. Cannot be null. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getExportedKeys(String catalog, String schema, String table, final CallOption... options) { + Objects.requireNonNull(table, "Table cannot be null."); + + 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()); + } + + Objects.requireNonNull(table); + builder.setTable(table).build(); + + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * Retrieves the foreign key columns for the given table. + * + * @param catalog The primary key table catalog. + * @param schema The primary key table schema. + * @param table The primary key table. Cannot be null. + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getImportedKeys(final String catalog, final String schema, final String table, + final CallOption... options) { + Objects.requireNonNull(table, "Table cannot be null."); + + 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()); + } + + Objects.requireNonNull(table); + builder.setTable(table).build(); + + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * Request a list of table types. + * + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo getTableTypes(final CallOption... options) { + final CommandGetTableTypes.Builder builder = CommandGetTableTypes.newBuilder(); + final FlightDescriptor descriptor = FlightDescriptor.command(Any.pack(builder.build()).toByteArray()); + return client.getInfo(descriptor, options); + } + + /** + * Create a prepared statement on the server. + * + * @param query The query to prepare. + * @param options RPC-layer hints for this call. + * @return The representation of the prepared statement which exists on the server. + */ + public PreparedStatement prepare(final String query, final CallOption... options) { + return new PreparedStatement(client, query, options); + } + + /** + * Helper class to encapsulate Flight SQL prepared statement logic. + */ + public static class PreparedStatement implements AutoCloseable { + private final FlightClient client; + private final ActionCreatePreparedStatementResult preparedStatementResult; + private final AtomicLong invocationCount; + private VectorSchemaRoot parameterBindingRoot; + private boolean isClosed; + private Schema resultSetSchema; + private Schema parameterSchema; + + /** + * Constructor. + * + * @param client The client. PreparedStatement does not maintain this resource. + * @param sql The query. + * @param options RPC-layer hints for this call. + */ + public PreparedStatement(final FlightClient client, final String sql, final CallOption... options) { + this.client = client; + final Action action = new Action( + FlightSqlUtils.FLIGHT_SQL_CREATEPREPAREDSTATEMENT.getType(), + Any.pack(ActionCreatePreparedStatementRequest + .newBuilder() + .setQuery(sql) + .build()) + .toByteArray()); + final Iterator<Result> preparedStatementResults = client.doAction(action, options); + + preparedStatementResult = FlightSqlUtils.unpackAndParseOrThrow( + preparedStatementResults.next().getBody(), + ActionCreatePreparedStatementResult.class); + + invocationCount = new AtomicLong(0); + isClosed = false; + } + + /** + * Set the {@link #parameterBindingRoot} containing the parameter binding from a {@link PreparedStatement} + * operation. + * + * @param parameterBindingRoot a {@code VectorSchemaRoot} object containing the values to be used in the + * {@code PreparedStatement} setters. + */ + public void setParameters(final VectorSchemaRoot parameterBindingRoot) { + if (this.parameterBindingRoot != null) { + if (this.parameterBindingRoot.equals(parameterBindingRoot)) { + return; + } + this.parameterBindingRoot.close(); + } + this.parameterBindingRoot = + Objects.requireNonNull(parameterBindingRoot, "Parameter binding root cannot be null!"); + } + + /** + * Empty the {@link #parameterBindingRoot}, which contains the parameter binding from + * a {@link PreparedStatement} operation. + */ + public void clearParameters() { + if (parameterBindingRoot != null) { + parameterBindingRoot.close(); + } + } + + /** + * Returns the Schema of the resultset. + * + * @return the Schema of the resultset. + */ + public Schema getResultSetSchema() { + if (resultSetSchema == null) { + final ByteString bytes = preparedStatementResult.getDatasetSchema(); + if (bytes.isEmpty()) { + return new Schema(Collections.emptyList()); + } + resultSetSchema = Schema.deserialize(bytes.asReadOnlyByteBuffer()); + } + return resultSetSchema; + } + + /** + * Returns the Schema of the parameters. + * + * @return the Schema of the parameters. + */ + public Schema getParameterSchema() { + if (parameterSchema == null) { + final ByteString bytes = preparedStatementResult.getParameterSchema(); + if (bytes.isEmpty()) { + return new Schema(Collections.emptyList()); + } + parameterSchema = Schema.deserialize(bytes.asReadOnlyByteBuffer()); + } + return parameterSchema; + } + + /** + * Executes the prepared statement query on the server. + * + * @param options RPC-layer hints for this call. + * @return a FlightInfo object representing the stream(s) to fetch. + */ + public FlightInfo execute(final CallOption... options) throws SQLException { + checkOpen(); + + 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 && parameterBindingRoot.getRowCount() > 0) { + final SyncPutListener putListener = new SyncPutListener(); + + FlightClient.ClientStreamListener listener = + client.startPut(descriptor, parameterBindingRoot, putListener, options); + + listener.putNext(); + listener.completed(); + } + + return client.getInfo(descriptor, options); + } + + /** + * Checks whether this client is open. + * + * @throws IllegalStateException if client is closed. + */ + protected final void checkOpen() { + Preconditions.checkState(!isClosed, "Statement closed"); + } + + /** + * Executes the prepared statement update on the server. + * + * @param options RPC-layer hints for this call. + */ Review comment: ```suggestion * @return the count of updated records */ ``` ########## File path: format/FlightSql.proto ########## @@ -0,0 +1,449 @@ +/* + * 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"; +import "google/protobuf/descriptor.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: uint32, + * 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 { + option (experimental) = true; + + /* + * 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 - Range [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 for future use. + * 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 { + option (experimental) = true; +} + +/* + * 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 { + option (experimental) = true; + + /* + * 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 (schema of the table as described in Schema.fbs::Schema, it is serialized as an IPC message.) + * > + * The returned data should be ordered by catalog_name, schema_name, table_name, then table_type. + */ +message CommandGetTables { + option (experimental) = true; + + /* + * 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 { + option (experimental) = true; +} + +/* + * 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_name: utf8 + * key_sequence: int, + * > + * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence. + */ +message CommandGetPrimaryKeys { + option (experimental) = true; + + // 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: uint1, + * delete_rule: uint1 + * > + * The returned data should be ordered by catalog_name, schema_name, table_name, key_name, then key_sequence. Review comment: Ditto the question above about whether we should order by all of the fields (except the _rule ones I suppose)? -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
