You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/03/01 15:57:49 UTC
[GitHub] [arrow] pitrou commented on a change in pull request #12465: ARROW-15282: [C++][FlightRPC] Split data methods from the underlying transport
pitrou commented on a change in pull request #12465:
URL: https://github.com/apache/arrow/pull/12465#discussion_r816863819
##########
File path: cpp/src/arrow/flight/server.h
##########
@@ -184,6 +184,10 @@ class ARROW_FLIGHT_EXPORT FlightServerBase {
/// domain socket).
int port() const;
+ /// \brief Get the address that the Flight server is listening on.
Review comment:
Should this return a `std::vector<Location>` instead, in case the server is listening on several addresses?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
Review comment:
```suggestion
/// \brief A transport-specific interface for reading/writing Arrow data.
```
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
Review comment:
What is Finish in this context?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
Review comment:
You mean WriteData?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
+ /// FlightPayload&).
+ virtual Status WritePutMetadata(const Buffer& payload);
+ /// \brief Indicate that there are no more writes on this stream.
+ ///
+ /// This is only a hint for the underlying transport and may not
+ /// actually do anything.
+ virtual Status WritesDone();
+};
+
+/// \brief A transport-specific interface for reading/writing Arrow
+/// data for a client.
+class ARROW_FLIGHT_EXPORT ClientDataStream : public TransportDataStream {
+ public:
+ /// \brief Attempt to read a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Read(FlightData*).
+ virtual bool ReadPutMetadata(std::shared_ptr<Buffer>* out);
+ /// \brief Finish the call, returning the final server status.
+ ///
+ /// Implies WritesDone().
+ virtual Status Finish() = 0;
Review comment:
Is this equivalent to `Finish(Status::OK())`?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
+ /// FlightPayload&).
+ virtual Status WritePutMetadata(const Buffer& payload);
+ /// \brief Indicate that there are no more writes on this stream.
+ ///
+ /// This is only a hint for the underlying transport and may not
+ /// actually do anything.
+ virtual Status WritesDone();
+};
+
+/// \brief A transport-specific interface for reading/writing Arrow
+/// data for a client.
+class ARROW_FLIGHT_EXPORT ClientDataStream : public TransportDataStream {
+ public:
+ /// \brief Attempt to read a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Read(FlightData*).
+ virtual bool ReadPutMetadata(std::shared_ptr<Buffer>* out);
+ /// \brief Finish the call, returning the final server status.
+ ///
+ /// Implies WritesDone().
+ virtual Status Finish() = 0;
+ /// \brief Attempt to cancel the call.
+ ///
+ /// This is only a hint and may not take effect immediately. The
+ /// client should still finish the call with Finish() as usual.
+ virtual void TryCancel() {}
+ /// \brief Finish the call, combining client-side status with server status.
+ Status Finish(Status st);
+};
+
+/// An implementation of a Flight client for a particular transport.
+///
+/// Transports should override the methods they are capable of
+/// supporting. The default method implementations return an error.
+class ARROW_FLIGHT_EXPORT ClientTransportImpl {
+ public:
+ virtual ~ClientTransportImpl() = default;
+
+ /// Initialize the client.
+ virtual Status Init(const FlightClientOptions& options, const Location& location,
+ const arrow::internal::Uri& uri) = 0;
+ /// Close the client. Once this returns, the client is no longer usable.
+ virtual Status Close() = 0;
+
+ virtual Status Authenticate(const FlightCallOptions& options,
+ std::unique_ptr<ClientAuthHandler> auth_handler);
+ virtual arrow::Result<std::pair<std::string, std::string>> AuthenticateBasicToken(
+ const FlightCallOptions& options, const std::string& username,
+ const std::string& password);
+ virtual Status DoAction(const FlightCallOptions& options, const Action& action,
+ std::unique_ptr<ResultStream>* results);
+ virtual Status ListActions(const FlightCallOptions& options,
+ std::vector<ActionType>* actions);
+ virtual Status GetFlightInfo(const FlightCallOptions& options,
+ const FlightDescriptor& descriptor,
+ std::unique_ptr<FlightInfo>* info);
+ virtual Status GetSchema(const FlightCallOptions& options,
+ const FlightDescriptor& descriptor,
+ std::unique_ptr<SchemaResult>* schema_result);
+ virtual Status ListFlights(const FlightCallOptions& options, const Criteria& criteria,
+ std::unique_ptr<FlightListing>* listing);
+ virtual Status DoGet(const FlightCallOptions& options, const Ticket& ticket,
+ std::unique_ptr<ClientDataStream>* stream);
+ virtual Status DoPut(const FlightCallOptions& options,
+ std::unique_ptr<ClientDataStream>* stream);
+ virtual Status DoExchange(const FlightCallOptions& options,
+ std::unique_ptr<ClientDataStream>* stream);
+};
+
+/// The implementation of the Flight service, which implements RPC
+/// method handlers that work in terms of the generic interfaces
+/// here.
Review comment:
So, basically this is a server-side connection handler?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
+ /// FlightPayload&).
+ virtual Status WritePutMetadata(const Buffer& payload);
+ /// \brief Indicate that there are no more writes on this stream.
+ ///
+ /// This is only a hint for the underlying transport and may not
+ /// actually do anything.
+ virtual Status WritesDone();
+};
+
+/// \brief A transport-specific interface for reading/writing Arrow
+/// data for a client.
+class ARROW_FLIGHT_EXPORT ClientDataStream : public TransportDataStream {
+ public:
+ /// \brief Attempt to read a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Read(FlightData*).
+ virtual bool ReadPutMetadata(std::shared_ptr<Buffer>* out);
+ /// \brief Finish the call, returning the final server status.
+ ///
+ /// Implies WritesDone().
+ virtual Status Finish() = 0;
+ /// \brief Attempt to cancel the call.
+ ///
+ /// This is only a hint and may not take effect immediately. The
+ /// client should still finish the call with Finish() as usual.
+ virtual void TryCancel() {}
+ /// \brief Finish the call, combining client-side status with server status.
+ Status Finish(Status st);
+};
+
+/// An implementation of a Flight client for a particular transport.
+///
+/// Transports should override the methods they are capable of
+/// supporting. The default method implementations return an error.
+class ARROW_FLIGHT_EXPORT ClientTransportImpl {
+ public:
+ virtual ~ClientTransportImpl() = default;
+
+ /// Initialize the client.
+ virtual Status Init(const FlightClientOptions& options, const Location& location,
+ const arrow::internal::Uri& uri) = 0;
+ /// Close the client. Once this returns, the client is no longer usable.
+ virtual Status Close() = 0;
+
+ virtual Status Authenticate(const FlightCallOptions& options,
+ std::unique_ptr<ClientAuthHandler> auth_handler);
+ virtual arrow::Result<std::pair<std::string, std::string>> AuthenticateBasicToken(
+ const FlightCallOptions& options, const std::string& username,
+ const std::string& password);
+ virtual Status DoAction(const FlightCallOptions& options, const Action& action,
+ std::unique_ptr<ResultStream>* results);
+ virtual Status ListActions(const FlightCallOptions& options,
+ std::vector<ActionType>* actions);
+ virtual Status GetFlightInfo(const FlightCallOptions& options,
+ const FlightDescriptor& descriptor,
+ std::unique_ptr<FlightInfo>* info);
+ virtual Status GetSchema(const FlightCallOptions& options,
+ const FlightDescriptor& descriptor,
+ std::unique_ptr<SchemaResult>* schema_result);
+ virtual Status ListFlights(const FlightCallOptions& options, const Criteria& criteria,
+ std::unique_ptr<FlightListing>* listing);
+ virtual Status DoGet(const FlightCallOptions& options, const Ticket& ticket,
+ std::unique_ptr<ClientDataStream>* stream);
+ virtual Status DoPut(const FlightCallOptions& options,
+ std::unique_ptr<ClientDataStream>* stream);
+ virtual Status DoExchange(const FlightCallOptions& options,
+ std::unique_ptr<ClientDataStream>* stream);
+};
+
+/// The implementation of the Flight service, which implements RPC
+/// method handlers that work in terms of the generic interfaces
+/// here.
+///
+/// Transport implementations should implement the necessary
+/// interfaces and call methods of this service.
+class ARROW_FLIGHT_EXPORT FlightServiceImpl {
+ public:
+ explicit FlightServiceImpl(FlightServerBase* base,
+ std::shared_ptr<MemoryManager> memory_manager)
+ : service_(base), memory_manager_(std::move(memory_manager)) {}
+ Status DoGet(const ServerCallContext& context, const Ticket& request,
+ TransportDataStream* stream);
+ Status DoPut(const ServerCallContext& context, TransportDataStream* stream);
+ Status DoExchange(const ServerCallContext& context, TransportDataStream* stream);
+ FlightServerBase* base() const { return service_; }
+
+ private:
+ FlightServerBase* service_;
Review comment:
Bad naming here IMHO.
##########
File path: cpp/src/arrow/flight/client.cc
##########
@@ -596,198 +427,256 @@ class GrpcStreamReader : public FlightStreamReader {
return ReadAll(table, stop_token_);
}
using FlightStreamReader::ReadAll;
- void Cancel() override { rpc_->context.TryCancel(); }
+ void Cancel() override { stream_->TryCancel(); }
private:
- std::unique_lock<std::mutex> TakeGuard() {
- return read_mutex_ ? std::unique_lock<std::mutex>(*read_mutex_)
- : std::unique_lock<std::mutex>();
- }
-
Status OverrideWithServerError(Status&& st) {
if (st.ok()) {
return std::move(st);
}
return stream_->Finish(std::move(st));
}
- friend class GrpcIpcMessageReader<Reader>;
- std::shared_ptr<ClientRpc> rpc_;
- std::shared_ptr<MemoryManager> memory_manager_;
- // Guard reads with a lock to prevent Finish()/Close() from being
- // called on the writer while the reader has a pending
- // read. Nullable, as DoGet() doesn't need this.
- std::shared_ptr<std::mutex> read_mutex_;
+ std::shared_ptr<internal::ClientDataStream> stream_;
ipc::IpcReadOptions options_;
StopToken stop_token_;
- std::shared_ptr<FinishableStream<Reader, internal::FlightData>> stream_;
- std::shared_ptr<internal::PeekableFlightDataReader<std::shared_ptr<Reader>>>
- peekable_reader_;
+ std::shared_ptr<internal::PeekableFlightDataReader> peekable_reader_;
std::shared_ptr<ipc::RecordBatchReader> batch_reader_;
std::shared_ptr<Buffer> app_metadata_;
};
-// The next two classes implement writing to a FlightData stream.
-// Similarly to the read side, we want to reuse the implementation of
-// RecordBatchWriter. As a result, these two classes are intertwined
-// in order to pass application metadata "through" RecordBatchWriter.
-// In order to get application-specific metadata to the
-// IpcPayloadWriter, DoPutPayloadWriter takes a pointer to
-// GrpcStreamWriter. GrpcStreamWriter updates a metadata field on
-// write; DoPutPayloadWriter reads that metadata field to determine
-// what to write.
-
-template <typename ProtoReadT, typename FlightReadT>
-class DoPutPayloadWriter;
-
-template <typename ProtoReadT, typename FlightReadT>
-class GrpcStreamWriter : public FlightStreamWriter {
+FlightMetadataReader::~FlightMetadataReader() = default;
+
+/// \brief The base of the ClientDataStream implementation for gRPC.
+template <typename Stream, typename ReadPayload>
+class FinishableDataStream : public internal::ClientDataStream {
public:
- ~GrpcStreamWriter() override = default;
+ FinishableDataStream(std::shared_ptr<ClientRpc> rpc, std::shared_ptr<Stream> stream,
+ std::shared_ptr<MemoryManager> memory_manager)
+ : rpc_(std::move(rpc)),
+ stream_(std::move(stream)),
+ memory_manager_(memory_manager ? std::move(memory_manager)
+ : CPUDevice::Instance()->default_memory_manager()),
+ finished_(false) {}
- using GrpcStream = grpc::ClientReaderWriter<pb::FlightData, ProtoReadT>;
+ Status Finish() override {
+ if (finished_) {
+ return server_status_;
+ }
- explicit GrpcStreamWriter(
- const FlightDescriptor& descriptor, std::shared_ptr<ClientRpc> rpc,
- int64_t write_size_limit_bytes, const ipc::IpcWriteOptions& options,
- std::shared_ptr<FinishableWritableStream<GrpcStream, FlightReadT>> writer)
- : app_metadata_(nullptr),
- batch_writer_(nullptr),
- writer_(std::move(writer)),
- rpc_(std::move(rpc)),
- write_size_limit_bytes_(write_size_limit_bytes),
- options_(options),
- descriptor_(descriptor),
- writer_closed_(false) {}
+ // Drain the read side, as otherwise gRPC Finish() will hang. We
+ // only call Finish() when the client closes the writer or the
+ // reader finishes, so it's OK to assume the client no longer
+ // wants to read and drain the read side. (If the client wants to
+ // indicate that it is done writing, but not done reading, it
+ // should use DoneWriting.
+ ReadPayload message;
+ while (internal::ReadPayload(stream_.get(), &message)) {
+ // Drain the read side to avoid gRPC hanging in Finish()
+ }
- static Status Open(
- const FlightDescriptor& descriptor, std::shared_ptr<Schema> schema,
- const ipc::IpcWriteOptions& options, std::shared_ptr<ClientRpc> rpc,
- int64_t write_size_limit_bytes,
- std::shared_ptr<FinishableWritableStream<GrpcStream, FlightReadT>> writer,
- std::unique_ptr<FlightStreamWriter>* out);
+ server_status_ = internal::FromGrpcStatus(stream_->Finish(), &rpc_->context);
+ if (!server_status_.ok()) {
+ server_status_ = Status::FromDetailAndArgs(
+ server_status_.code(), server_status_.detail(), server_status_.message(),
+ ". gRPC client debug context: ", rpc_->context.debug_error_string());
+ }
+ finished_ = true;
- Status CheckStarted() {
- if (!batch_writer_) {
- return Status::Invalid("Writer not initialized. Call Begin() with a schema.");
+ return server_status_;
+ }
+ void TryCancel() override { rpc_->context.TryCancel(); }
+
+ std::shared_ptr<ClientRpc> rpc_;
+ std::shared_ptr<Stream> stream_;
+ std::shared_ptr<MemoryManager> memory_manager_;
+ bool finished_;
+ Status server_status_;
+};
+
+/// \brief A ClientDataStream implementation for gRPC that manages a
+/// mutex to protect from concurrent reads/writes, and drains the
+/// read side on finish.
+template <typename Stream, typename ReadPayload>
+class WritableDataStream : public FinishableDataStream<Stream, ReadPayload> {
+ public:
+ using Base = FinishableDataStream<Stream, ReadPayload>;
+ WritableDataStream(std::shared_ptr<ClientRpc> rpc, std::shared_ptr<Stream> stream,
+ std::shared_ptr<MemoryManager> memory_manager)
+ : Base(std::move(rpc), std::move(stream), std::move(memory_manager)),
+ read_mutex_(),
+ finish_mutex_(),
+ done_writing_(false) {}
+
+ Status WritesDone() override {
+ // This is only used by the writer side of a stream, so it need
+ // not be protected with a lock.
+ if (done_writing_) {
+ return Status::OK();
+ }
+ done_writing_ = true;
+ if (!stream_->WritesDone()) {
+ // Error happened, try to close the stream to get more detailed info
+ return internal::ClientDataStream::Finish(MakeFlightError(
+ FlightStatusCode::Internal, "Could not flush pending record batches"));
}
return Status::OK();
}
- Status Begin(const std::shared_ptr<Schema>& schema,
- const ipc::IpcWriteOptions& options) override {
- if (batch_writer_) {
- return Status::Invalid("This writer has already been started.");
+ Status Finish() override {
+ // This may be used concurrently by reader/writer side of a
+ // stream, so it needs to be protected.
+ std::lock_guard<std::mutex> guard(finish_mutex_);
+
+ // Now that we're shared between a reader and writer, we need to
+ // protect ourselves from being called while there's an
+ // outstanding read.
+ std::unique_lock<std::mutex> read_guard(read_mutex_, std::try_to_lock);
+ if (!read_guard.owns_lock()) {
+ return MakeFlightError(FlightStatusCode::Internal,
+ "Cannot close stream with pending read operation.");
}
- std::unique_ptr<ipc::internal::IpcPayloadWriter> payload_writer(
- new DoPutPayloadWriter<ProtoReadT, FlightReadT>(
- descriptor_, std::move(rpc_), write_size_limit_bytes_, writer_, this));
- // XXX: this does not actually write the message to the stream.
- // See Close().
- ARROW_ASSIGN_OR_RAISE(batch_writer_, ipc::internal::OpenRecordBatchWriter(
- std::move(payload_writer), schema, options));
- return Status::OK();
+
+ // Try to flush pending writes. Don't use our WritesDone() to
+ // avoid recursion.
+ bool finished_writes = done_writing_ || stream_->WritesDone();
+ done_writing_ = true;
+
+ Status st = Base::Finish();
+ if (!finished_writes) {
+ return Status::FromDetailAndArgs(
+ st.code(), st.detail(), st.message(),
+ ". Additionally, could not finish writing record batches before closing");
+ }
+ return st;
}
- Status Begin(const std::shared_ptr<Schema>& schema) override {
- return Begin(schema, options_);
+ using Base::stream_;
+ std::mutex read_mutex_;
+ std::mutex finish_mutex_;
+ bool done_writing_;
+};
+
+class GrpcClientGetStream
+ : public FinishableDataStream<grpc::ClientReader<pb::FlightData>,
+ internal::FlightData> {
+ public:
+ using FinishableDataStream::FinishableDataStream;
+
+ bool ReadData(internal::FlightData* data) override {
+ bool success = internal::ReadPayload(stream_.get(), data);
+ if (ARROW_PREDICT_FALSE(!success)) return false;
+ if (data->body &&
+ ARROW_PREDICT_FALSE(!data->body->device()->Equals(*memory_manager_->device()))) {
Review comment:
Is the `Equals` check actually required? AFAIR, `ViewOrCopy` should just be a no-op from CPU to CPU (and hopefully also for a given CUDA device onto itself).
##########
File path: cpp/src/arrow/flight/server.cc
##########
@@ -442,20 +196,69 @@ class GrpcAddCallHeaders : public AddCallHeaders {
grpc::ServerContext* context_;
};
+class GetDataStream : public internal::TransportDataStream {
+ public:
+ explicit GetDataStream(ServerWriter<pb::FlightData>* writer) : writer_(writer) {}
+
+ Status WriteData(const FlightPayload& payload) override {
+ return internal::WritePayload(payload, writer_);
+ }
+
+ private:
+ ServerWriter<pb::FlightData>* writer_;
+};
+
+class PutDataStream final : public internal::TransportDataStream {
+ public:
+ explicit PutDataStream(grpc::ServerReaderWriter<pb::PutResult, pb::FlightData>* stream)
+ : stream_(stream) {}
+
+ bool ReadData(internal::FlightData* data) override {
+ return internal::ReadPayload(&*stream_, data);
+ }
+ Status WritePutMetadata(const Buffer& metadata) override {
+ pb::PutResult message{};
+ message.set_app_metadata(metadata.data(), metadata.size());
+ if (stream_->Write(message)) {
+ return Status::OK();
+ }
+ return Status::IOError("Unknown error writing metadata.");
+ }
+
+ private:
+ grpc::ServerReaderWriter<pb::PutResult, pb::FlightData>* stream_;
+};
+
+class ExchangeDataStream final : public internal::TransportDataStream {
+ public:
+ explicit ExchangeDataStream(
+ grpc::ServerReaderWriter<pb::FlightData, pb::FlightData>* stream)
+ : stream_(stream) {}
+
+ bool ReadData(internal::FlightData* data) override {
+ return internal::ReadPayload(&*stream_, data);
+ }
+ Status WriteData(const FlightPayload& payload) override {
+ return internal::WritePayload(payload, stream_);
+ }
+
+ private:
+ grpc::ServerReaderWriter<pb::FlightData, pb::FlightData>* stream_;
+};
+
// This class glues an implementation of FlightServerBase together with the
// gRPC service definition, so the latter is not exposed in the public API
-class FlightServiceImpl : public FlightService::Service {
+class FlightGrpcServiceImpl : public FlightService::Service {
Review comment:
Hmm, can you call this something else? It's really too close to `FlightServiceImpl` not to produce any confusion.
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
+ /// FlightPayload&).
+ virtual Status WritePutMetadata(const Buffer& payload);
+ /// \brief Indicate that there are no more writes on this stream.
+ ///
+ /// This is only a hint for the underlying transport and may not
+ /// actually do anything.
+ virtual Status WritesDone();
+};
+
+/// \brief A transport-specific interface for reading/writing Arrow
+/// data for a client.
+class ARROW_FLIGHT_EXPORT ClientDataStream : public TransportDataStream {
+ public:
+ /// \brief Attempt to read a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Read(FlightData*).
+ virtual bool ReadPutMetadata(std::shared_ptr<Buffer>* out);
+ /// \brief Finish the call, returning the final server status.
+ ///
+ /// Implies WritesDone().
+ virtual Status Finish() = 0;
+ /// \brief Attempt to cancel the call.
+ ///
+ /// This is only a hint and may not take effect immediately. The
+ /// client should still finish the call with Finish() as usual.
+ virtual void TryCancel() {}
+ /// \brief Finish the call, combining client-side status with server status.
Review comment:
I'm not sure what combining means here? Is the status input sent to the server?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
+ /// FlightPayload&).
+ virtual Status WritePutMetadata(const Buffer& payload);
+ /// \brief Indicate that there are no more writes on this stream.
+ ///
+ /// This is only a hint for the underlying transport and may not
+ /// actually do anything.
+ virtual Status WritesDone();
+};
+
+/// \brief A transport-specific interface for reading/writing Arrow
+/// data for a client.
+class ARROW_FLIGHT_EXPORT ClientDataStream : public TransportDataStream {
+ public:
+ /// \brief Attempt to read a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Read(FlightData*).
+ virtual bool ReadPutMetadata(std::shared_ptr<Buffer>* out);
Review comment:
Hmm, is there a reason why `ReadPutMetadata` is defined here but `WritePutMetadata` on the base class?
##########
File path: cpp/src/arrow/flight/transport_impl.h
##########
@@ -0,0 +1,220 @@
+// 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.
+
+// Internal (but not private) interface for implementing alternate
+// transports in Flight.
+//
+// EXPERIMENTAL. Subject to change.
+
+#pragma once
+
+#include <chrono>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/flight/type_fwd.h"
+#include "arrow/flight/visibility.h"
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+namespace ipc {
+class Message;
+}
+namespace flight {
+namespace internal {
+
+/// Internal, not user-visible type used for memory-efficient reads
+struct FlightData {
+ /// Used only for puts, may be null
+ std::unique_ptr<FlightDescriptor> descriptor;
+
+ /// Non-length-prefixed Message header as described in format/Message.fbs
+ std::shared_ptr<Buffer> metadata;
+
+ /// Application-defined metadata
+ std::shared_ptr<Buffer> app_metadata;
+
+ /// Message body
+ std::shared_ptr<Buffer> body;
+
+ /// Open IPC message from the metadata and body
+ ::arrow::Result<std::unique_ptr<ipc::Message>> OpenMessage();
+};
+
+/// \brief An transport-specific interface for reading/writing Arrow data.
+///
+/// New transports will implement this to read/write IPC payloads to
+/// the underlying stream.
+class ARROW_FLIGHT_EXPORT TransportDataStream {
+ public:
+ virtual ~TransportDataStream() = default;
+ /// \brief Attemnpt to read the next FlightData message.
+ ///
+ /// \return success true if data was populated, false if there was
+ /// an error. For clients, the error can be retrieved from Finish.
+ virtual bool ReadData(FlightData* data);
+ /// \brief Attempt to write a FlightPayload.
+ virtual Status WriteData(const FlightPayload& payload);
+ /// \brief Attempt to write a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Write(const
+ /// FlightPayload&).
+ virtual Status WritePutMetadata(const Buffer& payload);
+ /// \brief Indicate that there are no more writes on this stream.
+ ///
+ /// This is only a hint for the underlying transport and may not
+ /// actually do anything.
+ virtual Status WritesDone();
+};
+
+/// \brief A transport-specific interface for reading/writing Arrow
+/// data for a client.
+class ARROW_FLIGHT_EXPORT ClientDataStream : public TransportDataStream {
+ public:
+ /// \brief Attempt to read a non-data message.
+ ///
+ /// Only implemented for DoPut; mutually exclusive with Read(FlightData*).
+ virtual bool ReadPutMetadata(std::shared_ptr<Buffer>* out);
+ /// \brief Finish the call, returning the final server status.
+ ///
+ /// Implies WritesDone().
+ virtual Status Finish() = 0;
+ /// \brief Attempt to cancel the call.
+ ///
+ /// This is only a hint and may not take effect immediately. The
+ /// client should still finish the call with Finish() as usual.
+ virtual void TryCancel() {}
+ /// \brief Finish the call, combining client-side status with server status.
+ Status Finish(Status st);
+};
+
+/// An implementation of a Flight client for a particular transport.
+///
+/// Transports should override the methods they are capable of
+/// supporting. The default method implementations return an error.
+class ARROW_FLIGHT_EXPORT ClientTransportImpl {
+ public:
+ virtual ~ClientTransportImpl() = default;
+
+ /// Initialize the client.
+ virtual Status Init(const FlightClientOptions& options, const Location& location,
+ const arrow::internal::Uri& uri) = 0;
+ /// Close the client. Once this returns, the client is no longer usable.
+ virtual Status Close() = 0;
+
+ virtual Status Authenticate(const FlightCallOptions& options,
+ std::unique_ptr<ClientAuthHandler> auth_handler);
+ virtual arrow::Result<std::pair<std::string, std::string>> AuthenticateBasicToken(
+ const FlightCallOptions& options, const std::string& username,
+ const std::string& password);
+ virtual Status DoAction(const FlightCallOptions& options, const Action& action,
+ std::unique_ptr<ResultStream>* results);
+ virtual Status ListActions(const FlightCallOptions& options,
+ std::vector<ActionType>* actions);
+ virtual Status GetFlightInfo(const FlightCallOptions& options,
+ const FlightDescriptor& descriptor,
+ std::unique_ptr<FlightInfo>* info);
+ virtual Status GetSchema(const FlightCallOptions& options,
+ const FlightDescriptor& descriptor,
+ std::unique_ptr<SchemaResult>* schema_result);
+ virtual Status ListFlights(const FlightCallOptions& options, const Criteria& criteria,
+ std::unique_ptr<FlightListing>* listing);
+ virtual Status DoGet(const FlightCallOptions& options, const Ticket& ticket,
+ std::unique_ptr<ClientDataStream>* stream);
+ virtual Status DoPut(const FlightCallOptions& options,
+ std::unique_ptr<ClientDataStream>* stream);
+ virtual Status DoExchange(const FlightCallOptions& options,
+ std::unique_ptr<ClientDataStream>* stream);
+};
+
+/// The implementation of the Flight service, which implements RPC
+/// method handlers that work in terms of the generic interfaces
+/// here.
+///
+/// Transport implementations should implement the necessary
+/// interfaces and call methods of this service.
+class ARROW_FLIGHT_EXPORT FlightServiceImpl {
+ public:
+ explicit FlightServiceImpl(FlightServerBase* base,
+ std::shared_ptr<MemoryManager> memory_manager)
+ : service_(base), memory_manager_(std::move(memory_manager)) {}
+ Status DoGet(const ServerCallContext& context, const Ticket& request,
+ TransportDataStream* stream);
Review comment:
`stream` is an out-parameter? Or is it initialized by the caller? Should `DoGet` take a `TransportDataStream` factory instead?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org