You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by we...@apache.org on 2019/03/20 02:14:26 UTC
[arrow] branch master updated: ARROW-4947: [Flight/C++] Remove
redundant schema parameter to Flight client DoGet
This is an automated email from the ASF dual-hosted git repository.
wesm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new 36b13fa ARROW-4947: [Flight/C++] Remove redundant schema parameter to Flight client DoGet
36b13fa is described below
commit 36b13fa81a4b250ad79fbdc116065aa6406e44dc
Author: David Li <Da...@twosigma.com>
AuthorDate: Tue Mar 19 21:14:16 2019 -0500
ARROW-4947: [Flight/C++] Remove redundant schema parameter to Flight client DoGet
Not a high priority for review - fine to land for 0.14
Author: David Li <Da...@twosigma.com>
Closes #3958 from lihalite/no-redundant-parameter and squashes the following commits:
52ec0b4f5 <David Li> Remove redundant schema parameter to Flight client DoGet
---
cpp/src/arrow/flight/client.cc | 30 ++++++++++++++++---------
cpp/src/arrow/flight/client.h | 5 +----
cpp/src/arrow/flight/flight-benchmark.cc | 2 +-
cpp/src/arrow/flight/flight-test.cc | 2 +-
cpp/src/arrow/flight/test-integration-client.cc | 2 +-
python/examples/flight/client.py | 2 +-
python/pyarrow/_flight.pyx | 5 ++---
python/pyarrow/includes/libarrow_flight.pxd | 2 +-
python/pyarrow/tests/test_flight.py | 9 ++++----
9 files changed, 31 insertions(+), 28 deletions(-)
diff --git a/cpp/src/arrow/flight/client.cc b/cpp/src/arrow/flight/client.cc
index 5cebd9f..30e9b61 100644
--- a/cpp/src/arrow/flight/client.cc
+++ b/cpp/src/arrow/flight/client.cc
@@ -86,13 +86,10 @@ class FlightStreamReader : public RecordBatchReader {
// Validate IPC message
RETURN_NOT_OK(ipc::Message::Open(data.metadata, data.body, &message));
- // The first message is a schema; read it and then try to read a
- // record batch.
- if (message->type() == ipc::Message::Type::SCHEMA) {
- RETURN_NOT_OK(ipc::ReadSchema(*message, &schema_));
- return ReadNext(out);
- } else if (message->type() == ipc::Message::Type::RECORD_BATCH) {
+ if (message->type() == ipc::Message::Type::RECORD_BATCH) {
return ipc::ReadRecordBatch(*message, schema_, out);
+ } else if (message->type() == ipc::Message::Type::SCHEMA) {
+ return Status(StatusCode::Invalid, "Flight stream changed schema midway");
} else {
return Status(StatusCode::Invalid, "Unrecognized message in Flight stream");
}
@@ -287,16 +284,27 @@ class FlightClient::FlightClientImpl {
return Status::OK();
}
- Status DoGet(const Ticket& ticket, const std::shared_ptr<Schema>& schema,
- std::unique_ptr<RecordBatchReader>* out) {
+ Status DoGet(const Ticket& ticket, std::unique_ptr<RecordBatchReader>* out) {
pb::Ticket pb_ticket;
internal::ToProto(ticket, &pb_ticket);
- // ClientRpc rpc;
std::unique_ptr<ClientRpc> rpc(new ClientRpc);
std::unique_ptr<grpc::ClientReader<pb::FlightData>> stream(
stub_->DoGet(&rpc->context, pb_ticket));
+ // First message must be the schema
+ std::shared_ptr<Schema> schema;
+ internal::FlightData data;
+ if (!stream->Read(reinterpret_cast<pb::FlightData*>(&data))) {
+ return Status(StatusCode::Invalid, "No data in Flight stream");
+ }
+ std::unique_ptr<ipc::Message> message;
+ RETURN_NOT_OK(ipc::Message::Open(data.metadata, data.body, &message));
+ if (message->type() != ipc::Message::Type::SCHEMA) {
+ return Status(StatusCode::Invalid, "Flight stream did not start with schema");
+ }
+ RETURN_NOT_OK(ipc::ReadSchema(*message, &schema));
+
*out = std::unique_ptr<RecordBatchReader>(
new FlightStreamReader(std::move(rpc), schema, std::move(stream)));
return Status::OK();
@@ -379,9 +387,9 @@ Status FlightClient::ListFlights(const Criteria& criteria,
return impl_->ListFlights(criteria, listing);
}
-Status FlightClient::DoGet(const Ticket& ticket, const std::shared_ptr<Schema>& schema,
+Status FlightClient::DoGet(const Ticket& ticket,
std::unique_ptr<RecordBatchReader>* stream) {
- return impl_->DoGet(ticket, schema, stream);
+ return impl_->DoGet(ticket, stream);
}
Status FlightClient::DoPut(const FlightDescriptor& descriptor,
diff --git a/cpp/src/arrow/flight/client.h b/cpp/src/arrow/flight/client.h
index e88b73f..6277c15 100644
--- a/cpp/src/arrow/flight/client.h
+++ b/cpp/src/arrow/flight/client.h
@@ -89,12 +89,9 @@ class ARROW_EXPORT FlightClient {
/// \brief Given a flight ticket and schema, request to be sent the
/// stream. Returns record batch stream reader
/// \param[in] ticket The flight ticket to use
- /// \param[in] schema the schema of the stream data as computed by
- /// GetFlightInfo
/// \param[out] stream the returned RecordBatchReader
/// \return Status
- Status DoGet(const Ticket& ticket, const std::shared_ptr<Schema>& schema,
- std::unique_ptr<RecordBatchReader>* stream);
+ Status DoGet(const Ticket& ticket, std::unique_ptr<RecordBatchReader>* stream);
/// \brief Upload data to a Flight described by the given
/// descriptor. The caller must call Close() on the returned stream
diff --git a/cpp/src/arrow/flight/flight-benchmark.cc b/cpp/src/arrow/flight/flight-benchmark.cc
index 0f3eb6b..775662c 100644
--- a/cpp/src/arrow/flight/flight-benchmark.cc
+++ b/cpp/src/arrow/flight/flight-benchmark.cc
@@ -98,7 +98,7 @@ Status RunPerformanceTest(const int port) {
token.ParseFromString(endpoint.ticket.ticket);
std::unique_ptr<RecordBatchReader> reader;
- RETURN_NOT_OK(client->DoGet(endpoint.ticket, schema, &reader));
+ RETURN_NOT_OK(client->DoGet(endpoint.ticket, &reader));
std::shared_ptr<RecordBatch> batch;
diff --git a/cpp/src/arrow/flight/flight-test.cc b/cpp/src/arrow/flight/flight-test.cc
index d1ab0aa..8a1adb2 100644
--- a/cpp/src/arrow/flight/flight-test.cc
+++ b/cpp/src/arrow/flight/flight-test.cc
@@ -204,7 +204,7 @@ TEST_F(TestFlightClient, DoGet) {
AssertSchemaEqual(*expected_schema, *schema);
std::unique_ptr<RecordBatchReader> stream;
- ASSERT_OK(client_->DoGet(ticket, schema, &stream));
+ ASSERT_OK(client_->DoGet(ticket, &stream));
BatchVector expected_batches;
const int num_batches = 5;
diff --git a/cpp/src/arrow/flight/test-integration-client.cc b/cpp/src/arrow/flight/test-integration-client.cc
index b3fbb1b..a4a1510 100644
--- a/cpp/src/arrow/flight/test-integration-client.cc
+++ b/cpp/src/arrow/flight/test-integration-client.cc
@@ -90,7 +90,7 @@ arrow::Status ConsumeFlightLocation(const arrow::flight::Location& location,
arrow::flight::FlightClient::Connect(location.host, location.port, &read_client));
std::unique_ptr<arrow::RecordBatchReader> stream;
- RETURN_NOT_OK(read_client->DoGet(ticket, schema, &stream));
+ RETURN_NOT_OK(read_client->DoGet(ticket, &stream));
return ReadToTable(stream, retrieved_data);
}
diff --git a/python/examples/flight/client.py b/python/examples/flight/client.py
index df9acc1..6d5b452 100644
--- a/python/examples/flight/client.py
+++ b/python/examples/flight/client.py
@@ -84,7 +84,7 @@ def get_flight(args, client):
for location in endpoint.locations:
print(location)
get_client = pyarrow.flight.FlightClient.connect(location)
- reader = get_client.do_get(endpoint.ticket, info.schema)
+ reader = get_client.do_get(endpoint.ticket)
df = reader.read_pandas()
print(df)
diff --git a/python/pyarrow/_flight.pyx b/python/pyarrow/_flight.pyx
index c79c738..d8a5aea 100644
--- a/python/pyarrow/_flight.pyx
+++ b/python/pyarrow/_flight.pyx
@@ -372,17 +372,16 @@ cdef class FlightClient:
return result
- def do_get(self, ticket: Ticket, schema: Schema):
+ def do_get(self, ticket: Ticket):
"""Request the data for a flight."""
cdef:
# TODO: introduce unwrap
CTicket c_ticket
- shared_ptr[CSchema] c_schema = pyarrow_unwrap_schema(schema)
unique_ptr[CRecordBatchReader] reader
c_ticket.ticket = ticket.ticket
with nogil:
- check_status(self.client.get().DoGet(c_ticket, c_schema, &reader))
+ check_status(self.client.get().DoGet(c_ticket, &reader))
result = FlightRecordBatchReader()
result.reader.reset(reader.release())
return result
diff --git a/python/pyarrow/includes/libarrow_flight.pxd b/python/pyarrow/includes/libarrow_flight.pxd
index faf58cb..78e7143 100644
--- a/python/pyarrow/includes/libarrow_flight.pxd
+++ b/python/pyarrow/includes/libarrow_flight.pxd
@@ -132,7 +132,7 @@ cdef extern from "arrow/flight/api.h" namespace "arrow" nogil:
CStatus GetFlightInfo(CFlightDescriptor& descriptor,
unique_ptr[CFlightInfo]* info)
- CStatus DoGet(CTicket& ticket, shared_ptr[CSchema]& schema,
+ CStatus DoGet(CTicket& ticket,
unique_ptr[CRecordBatchReader]* stream)
CStatus DoPut(CFlightDescriptor& descriptor,
shared_ptr[CSchema]& schema,
diff --git a/python/pyarrow/tests/test_flight.py b/python/pyarrow/tests/test_flight.py
index 93c6a87..9e28084 100644
--- a/python/pyarrow/tests/test_flight.py
+++ b/python/pyarrow/tests/test_flight.py
@@ -115,7 +115,7 @@ def test_flight_do_get():
with flight_server(ConstantFlightServer) as server_port:
client = flight.FlightClient.connect('localhost', server_port)
- data = client.do_get(flight.Ticket(b''), table.schema).read_all()
+ data = client.do_get(flight.Ticket(b'')).read_all()
assert data.equals(table)
@@ -137,7 +137,7 @@ def test_flight_large_message():
# Write a single giant chunk
writer.write_table(data, 10 * 1024 * 1024)
writer.close()
- result = client.do_get(flight.Ticket(b''), data.schema).read_all()
+ result = client.do_get(flight.Ticket(b'')).read_all()
assert result.equals(data)
@@ -153,7 +153,7 @@ def test_flight_generator_stream():
data.schema)
writer.write_table(data)
writer.close()
- result = client.do_get(flight.Ticket(b''), data.schema).read_all()
+ result = client.do_get(flight.Ticket(b'')).read_all()
assert result.equals(data)
@@ -161,6 +161,5 @@ def test_flight_invalid_generator_stream():
"""Try streaming data with mismatched schemas."""
with flight_server(InvalidStreamFlightServer) as server_port:
client = flight.FlightClient.connect('localhost', server_port)
- schema = InvalidStreamFlightServer.schema
with pytest.raises(pa.ArrowException):
- client.do_get(flight.Ticket(b''), schema).read_all()
+ client.do_get(flight.Ticket(b'')).read_all()