Add basic Arrow Flight docs

docs/source/conf.py

@@ -416,7 +416,16 @@ except ImportError:
     from unittest import mock
     pyarrow.cuda = sys.modules['pyarrow.cuda'] = mock.Mock()
 
+try:
+    import pyarrow.flight
+    flight_enabled = True
+except ImportError:
+    flight_enabled = False
+    pyarrow.flight = sys.modules['pyarrow.flight'] = mock.Mock()
+
+
 def setup(app):
     # Use a config value to indicate whether CUDA API docs can be generated.
     # This will also rebuild appropriately when the value changes.
     app.add_config_value('cuda_enabled', cuda_enabled, 'env')
+    app.add_config_value('flight_enabled', flight_enabled, 'env')

docs/source/cpp/api.rst

@@ -30,3 +30,4 @@ API Reference
    api/table
    api/utilities
    api/cuda
+   api/flight

docs/source/cpp/api/flight.rst

@@ -0,0 +1,126 @@
+.. 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.
+
+================
+Arrow Flight RPC
+================
+
+.. warning:: Flight is currently unstable. APIs are subject to change,
+             though we don't expect drastic changes.
+
+.. warning:: Flight is currently only available when built from source
+             appropriately.
+
+Common Types
+============
+
+.. doxygenstruct:: arrow::flight::Action
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::ActionType
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::Criteria
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::FlightDescriptor
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::FlightEndpoint
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::FlightInfo
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::FlightPayload
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::FlightListing
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::Location
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::PutResult
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::Result
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::ResultStream
+   :project: arrow_cpp
+   :members:
+
+.. doxygenstruct:: arrow::flight::Ticket
+   :project: arrow_cpp
+   :members:
+
+Clients
+=======
+
+.. doxygenclass:: arrow::flight::FlightClient
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::FlightCallOptions
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::ClientAuthHandler
+   :project: arrow_cpp
+   :members:
+
+.. doxygentypedef:: arrow::flight::TimeoutDuration
+   :project: arrow_cpp
+
+Servers
+=======
+
+.. doxygenclass:: arrow::flight::FlightServerBase
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::FlightDataStream
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::FlightMessageReader
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::RecordBatchStream
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::ServerAuthHandler
+   :project: arrow_cpp
+   :members:
+
+.. doxygenclass:: arrow::flight::ServerCallContext
+   :project: arrow_cpp
+   :members:

docs/source/format/Flight.rst

@@ -0,0 +1,106 @@
+.. 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.
+
+Arrow Flight RPC
+================
+
+Arrow Flight is a RPC framework for high-performance data services
+based on Arrow data, and is built on top of gRPC_ and the :doc:`IPC
+format <IPC>`.
+
+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).
+
+.. _gRPC: https://grpc.io/
+
+RPC Methods
+-----------
+
+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. Thus, one Flight client can
+connect to any Flight service and perform basic operations.
+
+Data streams are identified by descriptors, which are either a path or
+an arbitrary binary command. A client that wishes to download the data
+would:
+
+#. 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.
+#. Call ``GetFlightInfo(FlightDescriptor)`` to get a ``FlightInfo``
+   message containing details on where the data is located (as well as
+   other metadata, like the schema and possibly an estimate of the
+   dataset size).
+
+   Flight does not require that data live on the same server as
+   metadata: this call may list other servers to connect to. The
+   ``FlightInfo`` message includes a ``Ticket``, an opaque binary
+   token that the server uses to identify the exact data set being
+   requested.
+#. Connect to other servers (if needed).
+#. Call ``DoGet(Ticket)`` to get back a stream of Arrow record
+   batches.
+
+To upload data, a client would:
+
+#. Construct or acquire a ``FlightDescriptor``, as before.
+#. Call ``DoPut(FlightData)`` and upload a stream of Arrow record
+   batches. They would also include the ``FlightDescriptor`` with the
+   first message.
+
+See `Protocol Buffer Definitions`_ for full details on the methods and
+messages involved.
+
+Authentication
+~~~~~~~~~~~~~~
+
+Flight supports application-implemented authentication
+methods. Authentication, if enabled, has two phases: at connection
+time, the client and server can exchange any number of messages. Then,
+the client can provide a token alongside each call, and the server can
+validate that token.
+
+Applications may use any part of this; for instance, they may ignore
+the initial handshake and send an externally acquired token on each
+call, or they may establish trust during the handshake and not
+validate a token for each call. (Note that the latter is not secure if
+you choose to deploy a layer 7 load balancer, as is common with gRPC.)
+
+External Resources
+------------------
+
+- https://arrow.apache.org/blog/2018/10/09/0.11.0-release/
+- https://www.slideshare.net/JacquesNadeau5/apache-arrow-flight-overview
+
+Protocol Buffer Definitions
+---------------------------
+
+.. literalinclude:: ../../../format/Flight.proto
+   :language: protobuf
+   :linenos:

docs/source/index.rst

@@ -43,6 +43,7 @@ such topics as:
    format/Layout
    format/Metadata
    format/IPC
+   format/Flight
 
 .. _toc.usage:
 

docs/source/python/api.rst

@@ -30,6 +30,7 @@ API Reference
    api/files
    api/tables
    api/ipc
+   api/flight
    api/formats
    api/plasma
    api/cuda

docs/source/python/api/flight.rst

@@ -0,0 +1,82 @@
+.. 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.
+
+.. currentmodule:: pyarrow.flight
+
+Arrow Flight
+============
+
+.. ifconfig:: not flight_enabled
+
+   .. error::
+      This documentation was built without Flight enabled.  The Flight
+      API docs are not available.
+
+.. NOTE We still generate those API docs (with empty docstrings)
+.. when Flight is disabled and `pyarrow.flight` mocked (see conf.py).
+.. Otherwise we'd get autodoc warnings, see https://github.com/sphinx-doc/sphinx/issues/4770
+
+.. warning:: Flight is currently unstable. APIs are subject to change,
+             though we don't expect drastic changes.
+
+.. warning:: Flight is currently not distributed as part of wheels or
+             in Conda - it is only available when built from source
+             appropriately.
+
+Common Types
+------------
+
+.. autosummary::
+   :toctree: ../generated/
+
+    Action
+    ActionType
+    DescriptorType
+    FlightDescriptor
+    FlightEndpoint
+    FlightInfo
+    Location
+    Ticket
+    Result
+
+Flight Client
+-------------
+
+.. autosummary::
+   :toctree: ../generated/
+
+    FlightCallOptions
+    FlightClient
+
+Flight Server
+-------------
+
+.. autosummary::
+   :toctree: ../generated/
+
+    FlightServerBase
+    GeneratorStream
+    RecordBatchStream
+
+Authentication
+--------------
+
+.. autosummary::
+   :toctree: ../generated/
+
+    ClientAuthHandler
+    ServerAuthHandler

java/flight/src/main/java/org/apache/arrow/flight/FlightClient.java

@@ -75,6 +75,9 @@ public class FlightClient implements AutoCloseable {
   private final MethodDescriptor<Flight.Ticket, ArrowMessage> doGetDescriptor;
   private final MethodDescriptor<ArrowMessage, Flight.PutResult> doPutDescriptor;
 
+  /**
+   * Create a Flight client from an allocator and a gRPC channel.
+   */
   private FlightClient(BufferAllocator incomingAllocator, ManagedChannel channel) {
     this.allocator = incomingAllocator.newChildAllocator("flight-client", 0, Long.MAX_VALUE);
     this.channel = channel;
@@ -153,11 +156,11 @@ public class FlightClient implements AutoCloseable {
 
   /**
    * Create or append a descriptor with another stream.
-   * @param descriptor FlightDescriptor
-   * @param root VectorSchemaRoot
+   * @param descriptor FlightDescriptor the descriptor for the data
+   * @param root VectorSchemaRoot the root containing data
    * @param metadataListener A handler for metadata messages from the server.
    * @param options RPC-layer hints for this call.
-   * @return ClientStreamListener
+   * @return ClientStreamListener an interface to control uploading data
    */
   public ClientStreamListener startPut(FlightDescriptor descriptor, VectorSchemaRoot root,
       StreamListener<PutResult> metadataListener, CallOption... options) {
@@ -168,7 +171,7 @@ public class FlightClient implements AutoCloseable {
     final io.grpc.CallOptions callOptions = CallOptions.wrapStub(asyncStub, options).getCallOptions();
     ClientCallStreamObserver<ArrowMessage> observer = (ClientCallStreamObserver<ArrowMessage>)
         ClientCalls.asyncBidiStreamingCall(
-                authInterceptor.interceptCall(doPutDescriptor, callOptions, channel), resultObserver);
+            authInterceptor.interceptCall(doPutDescriptor, callOptions, channel), resultObserver);
     // send the schema to start.
     ArrowMessage message = new ArrowMessage(descriptor.toProtocol(), root.getSchema());
     observer.onNext(message);
@@ -200,7 +203,7 @@ public class FlightClient implements AutoCloseable {
   public FlightStream getStream(Ticket ticket, CallOption... options) {
     final io.grpc.CallOptions callOptions = CallOptions.wrapStub(asyncStub, options).getCallOptions();
     ClientCall<Flight.Ticket, ArrowMessage> call =
-            authInterceptor.interceptCall(doGetDescriptor, callOptions, channel);
+        authInterceptor.interceptCall(doGetDescriptor, callOptions, channel);
     FlightStream stream = new FlightStream(
         allocator,
         PENDING_REQUESTS,
@@ -238,6 +241,7 @@ public class FlightClient implements AutoCloseable {
   }
 
   private static class SetStreamObserver implements StreamObserver<Flight.PutResult> {
+
     private final CompletableFuture<Void> result;
     private final StreamListener<PutResult> listener;
 
@@ -275,6 +279,7 @@ public class FlightClient implements AutoCloseable {
   }
 
   private static class PutObserver implements ClientStreamListener {
+
     private final ClientCallStreamObserver<ArrowMessage> observer;
     private final VectorUnloader unloader;
     private final CompletableFuture<Void> futureResult;
@@ -326,21 +331,32 @@ public class FlightClient implements AutoCloseable {
    */
   public interface ClientStreamListener {
 
+    /**
+     * Send the current data in the corresponding {@link VectorSchemaRoot} to the server.
+     */
     void putNext();
 
     void putNext(byte[] appMetadata);
 
+    /**
+     * Indicate an error to the server. Terminates the stream; do not call {@link #completed()}.
+     */
     void error(Throwable ex);
 
     /** Indicate the stream is finished on the client side. */
     void completed();
 
-    /** Wait for the stream to finish on the server side. */
+    /**
+     * Wait for the stream to finish on the server side. You must call this to be notified of any errors that may have
+     * happened during the upload.
+     */
     void getResult();
 
   }
 
-
+  /**
+   * Shut down this client.
+   */
   public void close() throws InterruptedException {
     channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
     allocator.close();

java/flight/src/main/java/org/apache/arrow/flight/FlightProducer.java

@@ -25,63 +25,99 @@ import org.apache.arrow.vector.VectorSchemaRoot;
 public interface FlightProducer {
 
   /**
-   * Return data for the given stream.
+   * Return data for a 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 getStream(CallContext context, Ticket ticket,
-      ServerStreamListener listener);
+  void getStream(CallContext context, Ticket ticket, ServerStreamListener listener);
 
-  /** List available flights given some criteria. */
+  /**
+   * List available data streams on this service.
+   *
+   * @param context Per-call context.
+   * @param criteria Application-defined criteria for filtering streams.
+   * @param listener An interface for sending data back to the client.
+   */
   void listFlights(CallContext context, Criteria criteria,
       StreamListener<FlightInfo> listener);
 
-  /** Get information on a specific flight. */
-  FlightInfo getFlightInfo(CallContext context,
-      FlightDescriptor descriptor);
+  /**
+   * Get information about a particular data stream.
+   *
+   * @param context Per-call context.
+   * @param descriptor The descriptor identifying the data stream.
+   * @return Metadata about the stream.
+   */
+  FlightInfo getFlightInfo(CallContext context, FlightDescriptor descriptor);
 
-  /** Accept uploaded data for a flight. */
-  Runnable acceptPut(CallContext context,
+  /**
+   * Accept uploaded data for a particular stream.
+   *
+   * @param context Per-call context.
+   * @param flightStream The data stream being uploaded.
+   */
+  public Runnable acceptPut(CallContext context,
       FlightStream flightStream, StreamListener<PutResult> ackStream);
 
+  /**
+   * Generic handler for application-defined RPCs.
+   *
+   * @param context Per-call context.
+   * @param action Client-supplied parameters.
+   * @param listener A stream of responses.
+   */
   void doAction(CallContext context, Action action,
       StreamListener<Result> listener);
 
-  void listActions(CallContext context,
-      StreamListener<ActionType> listener);
+  /**
+   * List available application-defined RPCs.
+   * @param context Per-call context.
+   * @param listener An interface for sending data back to the client.
+   */
+  void listActions(CallContext context, StreamListener<ActionType> listener);
 
   /**
-   * Listener for creating a stream on the server side.
+   * An interface for sending Arrow data back to a client.
    */
-  interface ServerStreamListener {
+  public interface ServerStreamListener {
 
     /**
-     * Check if the client has cancelled their request.
+     * Check whether the call has been cancelled. If so, stop sending data.
      */
     boolean isCancelled();
 
     /**
-     * Check if the client is ready to receive further data.
+     * A hint indicating whether the client is ready to receive data without excessive buffering.
      */
     boolean isReady();
 
     /**
-     * Begin sending data from the specified root.
+     * Start sending data, using the schema of the given {@link VectorSchemaRoot}.
+     *
+     * <p>This method must be called before all others.
      */
     void start(VectorSchemaRoot root);
 
-    /** Send the current contents of the root to the client. */
+    /**
+     * Send the current contents of the associated {@link VectorSchemaRoot}.
+     */
     void putNext();
 
     /**
-     * Send the current contents of the root to the client, along with application-defined metadata on the side.
+     * Send the current contents of the associated {@link VectorSchemaRoot} alongside application-defined metadata.
      */
     void putNext(byte[] metadata);
 
     /**
-     *  Indicate an error to the client. Do not call {@link #completed} afterwards.
+     * Indicate an error to the client. Terminates the stream; do not call {@link #completed()} afterwards.
      */
     void error(Throwable ex);
 
-    /** Indicate that the stream is finished. */
+    /**
+     * Indicate that transmission is finished.
+     */
     void completed();
 
   }
@@ -93,10 +129,21 @@ public interface FlightProducer {
    */
   interface StreamListener<T> {
 
+    /**
+     * Send the next value to the client.
+     */
     void onNext(T val);
 
+    /**
+     * Indicate an error to the client.
+     *
+     * <p>Terminates the stream; do not call {@link #onCompleted()}.
+     */
     void onError(Throwable t);
 
+    /**
+     * Indicate that the transmission is finished.
+     */
     void onCompleted();
 
   }