Arrow Flight RPC

Arrow Flight is an RPC framework for high-performance data services based on Arrow data, and is built on top of gRPC and the IPC format.

Flight is organized around streams of Arrow record batches, being either downloaded from or uploaded to another service. A set of metadata methods offers discovery and introspection of streams, as well as the ability to implement application-specific methods.

Methods and message wire formats are defined by Protobuf, enabling interoperability with clients that may support gRPC and Arrow separately, but not Flight. However, Flight implementations include further optimizations to avoid overhead in usage of Protobuf (mostly around avoiding excessive memory copies).

RPC Methods and Request Patterns

Flight defines a set of RPC methods for uploading/downloading data, retrieving metadata about a data stream, listing available data streams, and for implementing application-specific RPC methods. A Flight service implements some subset of these methods, while a Flight client can call any of these methods.

Data streams are identified by descriptors (the FlightDescriptor message), which are either a path or an arbitrary binary command. For instance, the descriptor may encode a SQL query, a path to a file on a distributed file system, or even a pickled Python object; the application can use this message as it sees fit.

Thus, one Flight client can connect to any service and perform basic operations. To facilitate this, Flight services are expected to support some common request patterns, described next. Of course, applications may ignore compatibility and simply treat the Flight RPC methods as low-level building blocks for their own purposes.

See Protocol Buffer Definitions for full details on the methods and messages involved.

Downloading Data

A client that wishes to download the data would:

../_images/DoGet.mmd.svg

Retrieving data via DoGet.

  1. Construct or acquire a FlightDescriptor for the data set they are interested in.

    A client may know what descriptor they want already, or they may use methods like ListFlights to discover them.

  2. Call GetFlightInfo(FlightDescriptor) to get a FlightInfo message.

    Flight does not require that data live on the same server as metadata. Hence, FlightInfo contains details on where the data is located, so the client can go fetch the data from an appropriate server. This is encoded as a series of FlightEndpoint messages inside FlightInfo. Each endpoint represents some location that contains a subset of the response data.

    An endpoint contains a list of locations (server addresses) where this data can be retrieved from, and a Ticket, an opaque binary token that the server will use to identify the data being requested. There is no ordering defined on endpoints or the data within, so if the dataset is sorted, applications should return data in a single endpoint.

    The response also contains other metadata, like the schema, and optionally an estimate of the dataset size.

  3. Consume each endpoint returned by the server.

    To consume an endpoint, the client should connect to one of the locations in the endpoint, then call DoGet(Ticket) with the ticket in the endpoint. This will give the client a stream of Arrow record batches.

    If the server wishes to indicate that the data is on the local server and not a different location, then it can return an empty list of locations. The client can then reuse the existing connection to the original server to fetch data. Otherwise, the client must connect to one of the indicated locations.

    In this way, the locations inside an endpoint can also be thought of as performing look-aside load balancing or service discovery functions. And the endpoints can represent data that is partitioned or otherwise distributed.

    The client must consume all endpoints to retrieve the complete data set. The client can consume endpoints in any order, or even in parallel, or distribute the endpoints among multiple machines for consumption; this is up to the application to implement.

Uploading Data

To upload data, a client would:

../_images/DoPut.mmd.svg

Uploading data via DoPut.

  1. Construct or acquire a FlightDescriptor, as before.

  2. Call DoPut(FlightData) and upload a stream of Arrow record batches.

    The FlightDescriptor is included with the first message so the server can identify the dataset.

DoPut allows the server to send response messages back to the client with custom metadata. This can be used to implement things like resumable writes (e.g. the server can periodically send a message indicating how many rows have been committed so far).

Exchanging Data

Some use cases may require uploading and downloading data within a single call. While this can be emulated with multiple calls, this may be difficult if the application is stateful. For instance, the application may wish to implement a call where the client uploads data and the server responds with a transformation of that data; this would require being stateful if implemented using DoGet and DoPut. Instead, DoExchange allows this to be implemented as a single call. A client would:

../_images/DoExchange.mmd.svg

Complex data flow with DoExchange.

  1. Construct or acquire a FlightDescriptor, as before.

  2. Call DoExchange(FlightData).

    The FlightDescriptor is included with the first message, as with DoPut. At this point, both the client and the server may simultaneously stream data to the other side.

Authentication

Flight supports a variety of authentication methods that applications can customize for their needs.

“Handshake” authentication

This is implemented in two parts. At connection time, the client calls the Handshake RPC method, and the application-defined authentication handler can exchange any number of messages with its counterpart on the server. The handler then provides a binary token. The Flight client will then include this token in the headers of all future calls, which is validated by the server authentication handler.

Applications may use any part of this; for instance, they may ignore the initial handshake and send an externally acquired token (e.g. a bearer token) on each call, or they may establish trust during the handshake and not validate a token for each call, treating the connection as stateful (a “login” pattern).

Warning

Unless a token is validated on every call, this pattern is not secure, especially in the presence of a layer 7 load balancer, as is common with gRPC, or if gRPC transparently reconnects the client.

Header-based/middleware-based authentication

Clients may include custom headers with calls. Custom middleware can then be implemented to validate and accept/reject calls on the server side.

Mutual TLS (mTLS)

The client provides a certificate during connection establishment which is verified by the server. The application does not need to implement any authentication code, but must provision and distribute certificates.

This may only be available in certain implementations, and is only available when TLS is also enabled.

Some Flight implementations may expose the underlying gRPC API as well, in which case any authentication method supported by gRPC is available.

Transport Implementations

Flight is primarily defined in terms of its Protobuf and gRPC specification below, but Arrow implementations may also support alternative transports (see Flight RPC). In that case, implementations should use the following URI schemes for the given transport implemenatations:

Transport

URI Scheme

gRPC (plaintext)

grpc: or grpc+tcp:

gRPC (TLS)

grpc+tls:

gRPC (Unix domain socket)

grpc+unix:

UCX (plaintext)

ucx:

Error Handling

Arrow Flight defines its own set of error codes. The implementation differs between languages (e.g. in C++, Unimplemented is a general Arrow error status while it’s a Flight-specific exception in Java), but the following set is exposed:

Error Code

Description

UNKNOWN

An unknown error. The default if no other error applies.

INTERNAL

An error internal to the service implementation occurred.

INVALID_ARGUMENT

The client passed an invalid argument to the RPC.

TIMED_OUT

The operation exceeded a timeout or deadline.

NOT_FOUND

The requested resource (action, data stream) was not found.

ALREADY_EXISTS

The resource already exists.

CANCELLED

The operation was cancelled (either by the client or the server).

UNAUTHENTICATED

The client is not authenticated.

UNAUTHORIZED

The client is authenticated, but does not have permissions for the requested operation.

UNIMPLEMENTED

The RPC is not implemented.

UNAVAILABLE

The server is not available. May be emitted by the client for connectivity reasons.

External Resources

Protocol Buffer Definitions

  1/*
  2 * Licensed to the Apache Software Foundation (ASF) under one
  3 * or more contributor license agreements.  See the NOTICE file
  4 * distributed with this work for additional information
  5 * regarding copyright ownership.  The ASF licenses this file
  6 * to you under the Apache License, Version 2.0 (the
  7 * "License"); you may not use this file except in compliance
  8 * with the License.  You may obtain a copy of the License at
  9 * <p>
 10 * http://www.apache.org/licenses/LICENSE-2.0
 11 * <p>
 12 * Unless required by applicable law or agreed to in writing, software
 13 * distributed under the License is distributed on an "AS IS" BASIS,
 14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15 * See the License for the specific language governing permissions and
 16 * limitations under the License.
 17 */
 18
 19syntax = "proto3";
 20
 21option java_package = "org.apache.arrow.flight.impl";
 22option go_package = "github.com/apache/arrow/go/arrow/flight/internal/flight";
 23option csharp_namespace = "Apache.Arrow.Flight.Protocol";
 24
 25package arrow.flight.protocol;
 26
 27/*
 28 * A flight service is an endpoint for retrieving or storing Arrow data. A
 29 * flight service can expose one or more predefined endpoints that can be
 30 * accessed using the Arrow Flight Protocol. Additionally, a flight service
 31 * can expose a set of actions that are available.
 32 */
 33service FlightService {
 34
 35  /*
 36   * Handshake between client and server. Depending on the server, the
 37   * handshake may be required to determine the token that should be used for
 38   * future operations. Both request and response are streams to allow multiple
 39   * round-trips depending on auth mechanism.
 40   */
 41  rpc Handshake(stream HandshakeRequest) returns (stream HandshakeResponse) {}
 42
 43  /*
 44   * Get a list of available streams given a particular criteria. Most flight
 45   * services will expose one or more streams that are readily available for
 46   * retrieval. This api allows listing the streams available for
 47   * consumption. A user can also provide a criteria. The criteria can limit
 48   * the subset of streams that can be listed via this interface. Each flight
 49   * service allows its own definition of how to consume criteria.
 50   */
 51  rpc ListFlights(Criteria) returns (stream FlightInfo) {}
 52
 53  /*
 54   * For a given FlightDescriptor, get information about how the flight can be
 55   * consumed. This is a useful interface if the consumer of the interface
 56   * already can identify the specific flight to consume. This interface can
 57   * also allow a consumer to generate a flight stream through a specified
 58   * descriptor. For example, a flight descriptor might be something that
 59   * includes a SQL statement or a Pickled Python operation that will be
 60   * executed. In those cases, the descriptor will not be previously available
 61   * within the list of available streams provided by ListFlights but will be
 62   * available for consumption for the duration defined by the specific flight
 63   * service.
 64   */
 65  rpc GetFlightInfo(FlightDescriptor) returns (FlightInfo) {}
 66
 67  /*
 68   * For a given FlightDescriptor, get the Schema as described in Schema.fbs::Schema
 69   * This is used when a consumer needs the Schema of flight stream. Similar to
 70   * GetFlightInfo this interface may generate a new flight that was not previously
 71   * available in ListFlights.
 72   */
 73   rpc GetSchema(FlightDescriptor) returns (SchemaResult) {}
 74
 75  /*
 76   * Retrieve a single stream associated with a particular descriptor
 77   * associated with the referenced ticket. A Flight can be composed of one or
 78   * more streams where each stream can be retrieved using a separate opaque
 79   * ticket that the flight service uses for managing a collection of streams.
 80   */
 81  rpc DoGet(Ticket) returns (stream FlightData) {}
 82
 83  /*
 84   * Push a stream to the flight service associated with a particular
 85   * flight stream. This allows a client of a flight service to upload a stream
 86   * of data. Depending on the particular flight service, a client consumer
 87   * could be allowed to upload a single stream per descriptor or an unlimited
 88   * number. In the latter, the service might implement a 'seal' action that
 89   * can be applied to a descriptor once all streams are uploaded.
 90   */
 91  rpc DoPut(stream FlightData) returns (stream PutResult) {}
 92
 93  /*
 94   * Open a bidirectional data channel for a given descriptor. This
 95   * allows clients to send and receive arbitrary Arrow data and
 96   * application-specific metadata in a single logical stream. In
 97   * contrast to DoGet/DoPut, this is more suited for clients
 98   * offloading computation (rather than storage) to a Flight service.
 99   */
100  rpc DoExchange(stream FlightData) returns (stream FlightData) {}
101
102  /*
103   * Flight services can support an arbitrary number of simple actions in
104   * addition to the possible ListFlights, GetFlightInfo, DoGet, DoPut
105   * operations that are potentially available. DoAction allows a flight client
106   * to do a specific action against a flight service. An action includes
107   * opaque request and response objects that are specific to the type action
108   * being undertaken.
109   */
110  rpc DoAction(Action) returns (stream Result) {}
111
112  /*
113   * A flight service exposes all of the available action types that it has
114   * along with descriptions. This allows different flight consumers to
115   * understand the capabilities of the flight service.
116   */
117  rpc ListActions(Empty) returns (stream ActionType) {}
118
119}
120
121/*
122 * The request that a client provides to a server on handshake.
123 */
124message HandshakeRequest {
125
126  /*
127   * A defined protocol version
128   */
129  uint64 protocol_version = 1;
130
131  /*
132   * Arbitrary auth/handshake info.
133   */
134  bytes payload = 2;
135}
136
137message HandshakeResponse {
138
139  /*
140   * A defined protocol version
141   */
142  uint64 protocol_version = 1;
143
144  /*
145   * Arbitrary auth/handshake info.
146   */
147  bytes payload = 2;
148}
149
150/*
151 * A message for doing simple auth.
152 */
153message BasicAuth {
154  string username = 2;
155  string password = 3;
156}
157
158message Empty {}
159
160/*
161 * Describes an available action, including both the name used for execution
162 * along with a short description of the purpose of the action.
163 */
164message ActionType {
165  string type = 1;
166  string description = 2;
167}
168
169/*
170 * A service specific expression that can be used to return a limited set
171 * of available Arrow Flight streams.
172 */
173message Criteria {
174  bytes expression = 1;
175}
176
177/*
178 * An opaque action specific for the service.
179 */
180message Action {
181  string type = 1;
182  bytes body = 2;
183}
184
185/*
186 * An opaque result returned after executing an action.
187 */
188message Result {
189  bytes body = 1;
190}
191
192/*
193 * Wrap the result of a getSchema call
194 */
195message SchemaResult {
196  // The schema of the dataset in its IPC form:
197  //   4 bytes - an optional IPC_CONTINUATION_TOKEN prefix
198  //   4 bytes - the byte length of the payload
199  //   a flatbuffer Message whose header is the Schema
200  bytes schema = 1;
201}
202
203/*
204 * The name or tag for a Flight. May be used as a way to retrieve or generate
205 * a flight or be used to expose a set of previously defined flights.
206 */
207message FlightDescriptor {
208
209  /*
210   * Describes what type of descriptor is defined.
211   */
212  enum DescriptorType {
213
214    // Protobuf pattern, not used.
215    UNKNOWN = 0;
216
217    /*
218     * A named path that identifies a dataset. A path is composed of a string
219     * or list of strings describing a particular dataset. This is conceptually
220     *  similar to a path inside a filesystem.
221     */
222    PATH = 1;
223
224    /*
225     * An opaque command to generate a dataset.
226     */
227    CMD = 2;
228  }
229
230  DescriptorType type = 1;
231
232  /*
233   * Opaque value used to express a command. Should only be defined when
234   * type = CMD.
235   */
236  bytes cmd = 2;
237
238  /*
239   * List of strings identifying a particular dataset. Should only be defined
240   * when type = PATH.
241   */
242  repeated string path = 3;
243}
244
245/*
246 * The access coordinates for retrieval of a dataset. With a FlightInfo, a
247 * consumer is able to determine how to retrieve a dataset.
248 */
249message FlightInfo {
250  // The schema of the dataset in its IPC form:
251  //   4 bytes - an optional IPC_CONTINUATION_TOKEN prefix
252  //   4 bytes - the byte length of the payload
253  //   a flatbuffer Message whose header is the Schema
254  bytes schema = 1;
255
256  /*
257   * The descriptor associated with this info.
258   */
259  FlightDescriptor flight_descriptor = 2;
260
261  /*
262   * A list of endpoints associated with the flight. To consume the
263   * whole flight, all endpoints (and hence all Tickets) must be
264   * consumed. Endpoints can be consumed in any order.
265   *
266   * In other words, an application can use multiple endpoints to
267   * represent partitioned data.
268   *
269   * There is no ordering defined on endpoints. Hence, if the returned
270   * data has an ordering, it should be returned in a single endpoint.
271   */
272  repeated FlightEndpoint endpoint = 3;
273
274  // Set these to -1 if unknown.
275  int64 total_records = 4;
276  int64 total_bytes = 5;
277}
278
279/*
280 * A particular stream or split associated with a flight.
281 */
282message FlightEndpoint {
283
284  /*
285   * Token used to retrieve this stream.
286   */
287  Ticket ticket = 1;
288
289  /*
290   * A list of URIs where this ticket can be redeemed via DoGet().
291   *
292   * If the list is empty, the expectation is that the ticket can only
293   * be redeemed on the current service where the ticket was
294   * generated.
295   *
296   * If the list is not empty, the expectation is that the ticket can
297   * be redeemed at any of the locations, and that the data returned
298   * will be equivalent. In this case, the ticket may only be redeemed
299   * at one of the given locations, and not (necessarily) on the
300   * current service.
301   *
302   * In other words, an application can use multiple locations to
303   * represent redundant and/or load balanced services.
304   */
305  repeated Location location = 2;
306}
307
308/*
309 * A location where a Flight service will accept retrieval of a particular
310 * stream given a ticket.
311 */
312message Location {
313  string uri = 1;
314}
315
316/*
317 * An opaque identifier that the service can use to retrieve a particular
318 * portion of a stream.
319 *
320 * Tickets are meant to be single use. It is an error/application-defined
321 * behavior to reuse a ticket.
322 */
323message Ticket {
324  bytes ticket = 1;
325}
326
327/*
328 * A batch of Arrow data as part of a stream of batches.
329 */
330message FlightData {
331
332  /*
333   * The descriptor of the data. This is only relevant when a client is
334   * starting a new DoPut stream.
335   */
336  FlightDescriptor flight_descriptor = 1;
337
338  /*
339   * Header for message data as described in Message.fbs::Message.
340   */
341  bytes data_header = 2;
342
343  /*
344   * Application-defined metadata.
345   */
346  bytes app_metadata = 3;
347
348  /*
349   * The actual batch of Arrow data. Preferably handled with minimal-copies
350   * coming last in the definition to help with sidecar patterns (it is
351   * expected that some implementations will fetch this field off the wire
352   * with specialized code to avoid extra memory copies).
353   */
354  bytes data_body = 1000;
355}
356
357/**
358 * The response message associated with the submission of a DoPut.
359 */
360message PutResult {
361  bytes app_metadata = 1;
362}