From 6642933fd9fa04e198a54278f2993cc039133278 Mon Sep 17 00:00:00 2001 From: Steve Lord Date: Mon, 22 Jan 2024 10:07:27 -0800 Subject: [PATCH] DX-61034: Arrow changes from GH-34865 PR --- cpp/src/arrow/flight/client.cc | 25 ----- cpp/src/arrow/flight/client.h | 8 +- cpp/src/arrow/flight/types.cc | 37 ++------ cpp/src/arrow/flight/types.h | 94 ++----------------- .../integration/tests/IntegrationTest.java | 5 - 5 files changed, 15 insertions(+), 154 deletions(-) diff --git a/cpp/src/arrow/flight/client.cc b/cpp/src/arrow/flight/client.cc index 2b8bf96ffe66b..dce0224432f52 100644 --- a/cpp/src/arrow/flight/client.cc +++ b/cpp/src/arrow/flight/client.cc @@ -703,31 +703,6 @@ Status FlightClient::DoExchange(const FlightCallOptions& options, return Status::OK(); } -// PHOXME implement these... -arrow::Result FlightClient::SetSessionOptions( - const FlightCallOptions& options, const SetSessionOptionsRequest& request) { - RETURN_NOT_OK(CheckOpen()); - /* - */ -} - -arrow::Result FlightClient::GetSessionOptions( - const FlightCallOptions& options) { - RETURN_NOT_OK(CheckOpen()); - /* - */ -} - -arrow::Result<> FlightClient::CloseSession(const FlightCallOptions& options) { - RETURN_NOT_OK(CheckOpen()); - /* - */ -} - -::arrow::Result -FlightClient::SetSessionOptions( - const FlightCallOptions& options, - const SetSessionOptionsRequest& request) { ::arrow::Result FlightClient::SetSessionOptions( const FlightCallOptions& options, const SetSessionOptionsRequest& request) { RETURN_NOT_OK(CheckOpen()); diff --git a/cpp/src/arrow/flight/client.h b/cpp/src/arrow/flight/client.h index 7d614b8292f6d..e30dc982ec437 100644 --- a/cpp/src/arrow/flight/client.h +++ b/cpp/src/arrow/flight/client.h @@ -411,13 +411,7 @@ class ARROW_FLIGHT_EXPORT FlightClient { *reader = std::move(output.reader); return Status::OK(); } -/// \\brief Set server session option(s) by key/value. Sessions are generally -/// persisted via HTTP cookies. -/// \param[in] options Per-RPC options -/// \param[in] request The server session options to set -::arrow::Result SetSessionOptions( - const FlightCallOptions& options, - const SetSessionOptionsRequest& request); + /// \\brief Set server session option(s) by key/value. Sessions are generally /// persisted via HTTP cookies. /// \param[in] options Per-RPC options diff --git a/cpp/src/arrow/flight/types.cc b/cpp/src/arrow/flight/types.cc index be939691a3c16..b0da07fbf2bcc 100644 --- a/cpp/src/arrow/flight/types.cc +++ b/cpp/src/arrow/flight/types.cc @@ -831,35 +831,6 @@ std::string ActionType::ToString() const { description, "'>"); } -<<<<<<< HEAD -======= -const ActionType ActionType::kCancelFlightInfo = - ActionType{"CancelFlightInfo", - "Explicitly cancel a running FlightInfo.\n" - "Request Message: CancelFlightInfoRequest\n" - "Response Message: CancelFlightInfoResult"}; -const ActionType ActionType::kRenewFlightEndpoint = - ActionType{"RenewFlightEndpoint", - "Extend expiration time of the given FlightEndpoint.\n" - "Request Message: RenewFlightEndpointRequest\n" - "Response Message: Renewed FlightEndpoint"}; -const ActionType ActionType::kSetSessionOptions = - ActionType{"SetSessionOptions", - "Set client session options by name/value pairs.\n" - "Request Message: SetSessionOptionsRequest\n" - "Response Message: SetSessionOptionsResult"}; -const ActionType ActionType::kGetSessionOptions = - ActionType{"GetSessionOptions", - "Get current client session options\n" - "Request Message: GetSessionOptionsRequest\n" - "Response Message: GetSessionOptionsResult"}; -const ActionType ActionType::kCloseSession = - ActionType{"CloseSession", - "Explicitly close/invalidate the cookie-specified client session.\n" - "Request Message: CloseSessionRequest\n" - "Response Message: CloseSessionResult"}; - ->>>>>>> 5df37ac44 (WIP: High-level design review ONLY) bool ActionType::Equals(const ActionType& other) const { return type == other.type && description == other.description; } @@ -1015,6 +986,14 @@ Status MetadataRecordBatchReader::Next(FlightStreamChunk* next) { return Next().Value(next); } +Status ResultStream::Drain() { + while (true) { + ARROW_ASSIGN_OR_RAISE(auto result, Next()); + if (!result) break; + } + return Status::OK(); +} + arrow::Result>> MetadataRecordBatchReader::ToRecordBatches() { std::vector> batches; diff --git a/cpp/src/arrow/flight/types.h b/cpp/src/arrow/flight/types.h index 4044dcb46675d..f1e897501ba63 100644 --- a/cpp/src/arrow/flight/types.h +++ b/cpp/src/arrow/flight/types.h @@ -21,6 +21,7 @@ #include #include +#include #include #include #include @@ -601,94 +602,6 @@ class ARROW_FLIGHT_EXPORT FlightInfo { mutable bool reconstructed_schema_; }; -/// \brief The information to process a long-running query. -class ARROW_FLIGHT_EXPORT PollInfo { - public: - /// The currently available results so far. - std::unique_ptr info = NULLPTR; - /// The descriptor the client should use on the next try. If unset, - /// the query is complete. - std::optional descriptor = std::nullopt; - /// Query progress. Must be in [0.0, 1.0] but need not be - /// monotonic or nondecreasing. If unknown, do not set. - std::optional progress = std::nullopt; - /// Expiration time for this request. After this passes, the server - /// might not accept the poll descriptor anymore (and the query may - /// be cancelled). This may be updated on a call to PollFlightInfo. - std::optional expiration_time = std::nullopt; - - PollInfo() - : info(NULLPTR), - descriptor(std::nullopt), - progress(std::nullopt), - expiration_time(std::nullopt) {} - - explicit PollInfo(std::unique_ptr info, - std::optional descriptor, - std::optional progress, - std::optional expiration_time) - : info(std::move(info)), - descriptor(std::move(descriptor)), - progress(progress), - expiration_time(expiration_time) {} - - explicit PollInfo(const PollInfo& other) - : info(other.info ? std::make_unique(*other.info) : NULLPTR), - descriptor(other.descriptor), - progress(other.progress), - expiration_time(other.expiration_time) {} - - /// \brief Get the wire-format representation of this type. - /// - /// Useful when interoperating with non-Flight systems (e.g. REST - /// services) that may want to return Flight types. - arrow::Result SerializeToString() const; - - /// \brief Parse the wire-format representation of this type. - /// - /// Useful when interoperating with non-Flight systems (e.g. REST - /// services) that may want to return Flight types. - static arrow::Result> Deserialize( - std::string_view serialized); - - std::string ToString() const; - - /// Compare two PollInfo for equality. This will compare the - /// serialized schema representations, NOT the logical equality of - /// the schemas. - bool Equals(const PollInfo& other) const; - - friend bool operator==(const PollInfo& left, const PollInfo& right) { - return left.Equals(right); - } - friend bool operator!=(const PollInfo& left, const PollInfo& right) { - return !(left == right); - } -}; - -/// \brief The request of the CancelFlightInfoRequest action. -struct ARROW_FLIGHT_EXPORT CancelFlightInfoRequest { - std::unique_ptr info; - - std::string ToString() const; - bool Equals(const CancelFlightInfoRequest& other) const; - - friend bool operator==(const CancelFlightInfoRequest& left, - const CancelFlightInfoRequest& right) { - return left.Equals(right); - } - friend bool operator!=(const CancelFlightInfoRequest& left, - const CancelFlightInfoRequest& right) { - return !(left == right); - } - - /// \brief Serialize this message to its wire-format representation. - arrow::Result SerializeToString() const; - - /// \brief Deserialize this message from its wire-format representation. - static arrow::Result Deserialize(std::string_view serialized); -}; - /// \brief Variant supporting all possible value types for {Set,Get}SessionOptions /// /// By convention, an attempt to set a valueless (std::monostate) SessionOptionValue @@ -907,6 +820,11 @@ class ARROW_FLIGHT_EXPORT ResultStream { ARROW_DEPRECATED("Deprecated in 8.0.0. Use Result-returning overload instead.") Status Next(std::unique_ptr* info); + + /// \brief Read and drop the remaining messages to get the error (if any) from a server. + /// \return Status OK if this is no error from a server, any other status if a + /// server returns an error. + Status Drain(); }; /// \brief A holder for a RecordBatch with associated Flight metadata. diff --git a/java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java b/java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java index 867eca5948773..627a9ee23e41b 100644 --- a/java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java +++ b/java/flight/flight-integration-tests/src/test/java/org/apache/arrow/flight/integration/tests/IntegrationTest.java @@ -48,11 +48,6 @@ void flightSqlExtension() throws Exception { testScenario("flight_sql:extension"); } - @Test - void appMetadataFlightInfoEndpoint() throws Exception { - testScenario("app_metadata_flight_info_endpoint"); - } - @Test void sessionOptions() throws Exception { testScenario("session_options");