ARROW-15974: [C++] Migrate flight/types.h header definitions to use Result<>

cpp/src/arrow/flight/client.cc

@@ -72,12 +72,21 @@ std::shared_ptr<FlightWriteSizeStatusDetail> FlightWriteSizeStatusDetail::Unwrap
 
 FlightClientOptions FlightClientOptions::Defaults() { return FlightClientOptions(); }
 
+arrow::Result<std::shared_ptr<Table>> FlightStreamReader::ToTable(
+    const StopToken& stop_token) {
+  ARROW_ASSIGN_OR_RAISE(auto batches, ToRecordBatches(stop_token));
+  ARROW_ASSIGN_OR_RAISE(auto schema, GetSchema());
+  return Table::FromRecordBatches(schema, std::move(batches));
+}
+
+Status FlightStreamReader::ReadAll(std::vector<std::shared_ptr<RecordBatch>>* batches,
+                                   const StopToken& stop_token) {
+  return ToRecordBatches(stop_token).Value(batches);
+}
+
 Status FlightStreamReader::ReadAll(std::shared_ptr<Table>* table,
                                    const StopToken& stop_token) {
-  std::vector<std::shared_ptr<RecordBatch>> batches;
-  RETURN_NOT_OK(ReadAll(&batches, stop_token));
-  ARROW_ASSIGN_OR_RAISE(auto schema, GetSchema());
-  return Table::FromRecordBatches(schema, std::move(batches)).Value(table);
+  return ToTable(stop_token).Value(table);
 }
 
 /// \brief An ipc::MessageReader adapting the Flight ClientDataStream interface.
@@ -169,57 +178,60 @@ class ClientStreamReader : public FlightStreamReader {
     RETURN_NOT_OK(EnsureDataStarted());
     return batch_reader_->schema();
   }
-  Status Next(FlightStreamChunk* out) override {
+  arrow::Result<FlightStreamChunk> Next() override {
+    FlightStreamChunk out;
     internal::FlightData* data;
     peekable_reader_->Peek(&data);
     if (!data) {
-      out->app_metadata = nullptr;
-      out->data = nullptr;
-      return stream_->Finish(Status::OK());
+      out.app_metadata = nullptr;
+      out.data = nullptr;
+      RETURN_NOT_OK(stream_->Finish(Status::OK()));
+      return out;
     }
 
     if (!data->metadata) {
       // Metadata-only (data->metadata is the IPC header)
-      out->app_metadata = data->app_metadata;
-      out->data = nullptr;
+      out.app_metadata = data->app_metadata;
+      out.data = nullptr;
       peekable_reader_->Next(&data);
-      return Status::OK();
+      return out;
     }
 
     if (!batch_reader_) {
       RETURN_NOT_OK(EnsureDataStarted());
       // Re-peek here since EnsureDataStarted() advances the stream
-      return Next(out);
+      return Next();
     }
-    auto status = batch_reader_->ReadNext(&out->data);
+    auto status = batch_reader_->ReadNext(&out.data);
     if (ARROW_PREDICT_FALSE(!status.ok())) {
       return stream_->Finish(std::move(status));
     }
-    out->app_metadata = std::move(app_metadata_);
-    return Status::OK();
+    out.app_metadata = std::move(app_metadata_);
+    return out;
   }
-  Status ReadAll(std::vector<std::shared_ptr<RecordBatch>>* batches) override {
-    return ReadAll(batches, stop_token_);
+  arrow::Result<std::vector<std::shared_ptr<RecordBatch>>> ToRecordBatches() override {
+    return ToRecordBatches(stop_token_);
   }
-  Status ReadAll(std::vector<std::shared_ptr<RecordBatch>>* batches,
-                 const StopToken& stop_token) override {
+  arrow::Result<std::vector<std::shared_ptr<RecordBatch>>> ToRecordBatches(
+      const StopToken& stop_token) override {
+    std::vector<std::shared_ptr<RecordBatch>> batches;
     FlightStreamChunk chunk;
 
     while (true) {
       if (stop_token.IsStopRequested()) {
         Cancel();
         return stop_token.Poll();
       }
-      RETURN_NOT_OK(Next(&chunk));
+      ARROW_ASSIGN_OR_RAISE(chunk, Next());
       if (!chunk.data) break;
-      batches->emplace_back(std::move(chunk.data));
+      batches.emplace_back(std::move(chunk.data));
     }
-    return Status::OK();
+    return batches;
   }
-  Status ReadAll(std::shared_ptr<Table>* table) override {
-    return ReadAll(table, stop_token_);
+  arrow::Result<std::shared_ptr<Table>> ToTable() override {
+    return ToTable(stop_token_);
   }
-  using FlightStreamReader::ReadAll;
+  using FlightStreamReader::ToTable;
   void Cancel() override { stream_->TryCancel(); }
 
  private:

cpp/src/arrow/flight/client.h

@@ -133,11 +133,22 @@ class ARROW_FLIGHT_EXPORT FlightStreamReader : public MetadataRecordBatchReader
  public:
   /// \brief Try to cancel the call.
   virtual void Cancel() = 0;
-  using MetadataRecordBatchReader::ReadAll;
+
+  using MetadataRecordBatchReader::ToRecordBatches;
   /// \brief Consume entire stream as a vector of record batches
-  virtual Status ReadAll(std::vector<std::shared_ptr<RecordBatch>>* batches,
-                         const StopToken& stop_token) = 0;
+  virtual arrow::Result<std::vector<std::shared_ptr<RecordBatch>>> ToRecordBatches(
+      const StopToken& stop_token) = 0;
+
+  using MetadataRecordBatchReader::ReadAll;
+  ARROW_DEPRECATED("Deprecated in 8.0.0. Use ToRecordBatches instead.")
+  Status ReadAll(std::vector<std::shared_ptr<RecordBatch>>* batches,
+                 const StopToken& stop_token);
+
+  using MetadataRecordBatchReader::ToTable;
   /// \brief Consume entire stream as a Table
+  arrow::Result<std::shared_ptr<Table>> ToTable(const StopToken& stop_token);
+
+  ARROW_DEPRECATED("Deprecated in 8.0.0. Use ToTable instead.")
   Status ReadAll(std::shared_ptr<Table>* table, const StopToken& stop_token);
 };
 

cpp/src/arrow/flight/flight_test.cc

@@ -1445,19 +1445,18 @@ TEST_F(TestBasicHeaderAuthMiddleware, ValidCredentials) { RunValidClientAuth();
 TEST_F(TestBasicHeaderAuthMiddleware, InvalidCredentials) { RunInvalidClientAuth(); }
 
 class ForeverFlightListing : public FlightListing {
-  Status Next(std::unique_ptr<FlightInfo>* info) override {
+  arrow::Result<std::unique_ptr<FlightInfo>> Next() override {
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
-    *info = arrow::internal::make_unique<FlightInfo>(ExampleFlightInfo()[0]);
-    return Status::OK();
+    return arrow::internal::make_unique<FlightInfo>(ExampleFlightInfo()[0]);
   }
 };
 
 class ForeverResultStream : public ResultStream {
-  Status Next(std::unique_ptr<Result>* result) override {
+  arrow::Result<std::unique_ptr<Result>> Next() override {
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
-    *result = arrow::internal::make_unique<Result>();
-    (*result)->body = Buffer::FromString("foo");
-    return Status::OK();
+    auto result = arrow::internal::make_unique<Result>();
+    result->body = Buffer::FromString("foo");
+    return result;
   }
 };
 

cpp/src/arrow/flight/transport_server.cc

@@ -109,31 +109,32 @@ class TransportMessageReader final : public FlightMessageReader {
     return batch_reader_->schema();
   }
 
-  Status Next(FlightStreamChunk* out) override {
+  arrow::Result<FlightStreamChunk> Next() override {
+    FlightStreamChunk out;
     internal::FlightData* data;
     peekable_reader_->Peek(&data);
     if (!data) {
-      out->app_metadata = nullptr;
-      out->data = nullptr;
-      return Status::OK();
+      out.app_metadata = nullptr;
+      out.data = nullptr;
+      return out;
     }
 
     if (!data->metadata) {
       // Metadata-only (data->metadata is the IPC header)
-      out->app_metadata = data->app_metadata;
-      out->data = nullptr;
+      out.app_metadata = data->app_metadata;
+      out.data = nullptr;
       peekable_reader_->Next(&data);
-      return Status::OK();
+      return out;
     }
 
     if (!batch_reader_) {
       RETURN_NOT_OK(EnsureDataStarted());
       // re-peek here since EnsureDataStarted() advances the stream
-      return Next(out);
+      return Next();
     }
-    RETURN_NOT_OK(batch_reader_->ReadNext(&out->data));
-    out->app_metadata = std::move(app_metadata_);
-    return Status::OK();
+    RETURN_NOT_OK(batch_reader_->ReadNext(&out.data));
+    out.app_metadata = std::move(app_metadata_);
+    return out;
   }
 
  private:

cpp/src/arrow/flight/types.cc

@@ -142,10 +142,15 @@ Status FlightPayload::Validate() const {
   return Status::OK();
 }
 
+arrow::Result<std::shared_ptr<Schema>> SchemaResult::GetSchema(
+    ipc::DictionaryMemo* dictionary_memo) const {
+  io::BufferReader schema_reader(raw_schema_);
+  return ipc::ReadSchema(&schema_reader, dictionary_memo);
+}
+
 Status SchemaResult::GetSchema(ipc::DictionaryMemo* dictionary_memo,
                                std::shared_ptr<Schema>* out) const {
-  io::BufferReader schema_reader(raw_schema_);
-  return ipc::ReadSchema(&schema_reader, dictionary_memo).Value(out);
+  return GetSchema(dictionary_memo).Value(out);
 }
 
 arrow::Result<std::string> FlightDescriptor::SerializeToString() const {
@@ -233,17 +238,20 @@ arrow::Result<FlightInfo> FlightInfo::Make(const Schema& schema,
   return FlightInfo(data);
 }
 
-Status FlightInfo::GetSchema(ipc::DictionaryMemo* dictionary_memo,
-                             std::shared_ptr<Schema>* out) const {
+arrow::Result<std::shared_ptr<Schema>> FlightInfo::GetSchema(
+    ipc::DictionaryMemo* dictionary_memo) const {
   if (reconstructed_schema_) {
-    *out = schema_;
-    return Status::OK();
+    return schema_;
   }
   io::BufferReader schema_reader(data_.schema);
   RETURN_NOT_OK(ipc::ReadSchema(&schema_reader, dictionary_memo).Value(&schema_));
   reconstructed_schema_ = true;
-  *out = schema_;
-  return Status::OK();
+  return schema_;
+}
+
+Status FlightInfo::GetSchema(ipc::DictionaryMemo* dictionary_memo,
+                             std::shared_ptr<Schema>* out) const {
+  return GetSchema(dictionary_memo).Value(out);
 }
 
 arrow::Result<std::string> FlightInfo::SerializeToString() const {
@@ -284,35 +292,55 @@ Status FlightInfo::Deserialize(const std::string& serialized,
 
 Location::Location() { uri_ = std::make_shared<arrow::internal::Uri>(); }
 
+Status FlightListing::Next(std::unique_ptr<FlightInfo>* info) {
+  return Next().Value(info);
+}
+
+arrow::Result<Location> Location::Parse(const std::string& uri_string) {
+  Location location;
+  RETURN_NOT_OK(location.uri_->Parse(uri_string));
+  return location;
+}
+
 Status Location::Parse(const std::string& uri_string, Location* location) {
-  return location->uri_->Parse(uri_string);
+  return Parse(uri_string).Value(location);
 }
 
-Status Location::ForGrpcTcp(const std::string& host, const int port, Location* location) {
+arrow::Result<Location> Location::ForGrpcTcp(const std::string& host, const int port) {
   std::stringstream uri_string;
   uri_string << "grpc+tcp://" << host << ':' << port;
-  return Location::Parse(uri_string.str(), location);
+  return Location::Parse(uri_string.str());
 }
 
-Status Location::ForGrpcTls(const std::string& host, const int port, Location* location) {
+Status Location::ForGrpcTcp(const std::string& host, const int port, Location* location) {
+  return ForGrpcTcp(host, port).Value(location);
+}
+
+arrow::Result<Location> Location::ForGrpcTls(const std::string& host, const int port) {
   std::stringstream uri_string;
   uri_string << "grpc+tls://" << host << ':' << port;
-  return Location::Parse(uri_string.str(), location);
+  return Location::Parse(uri_string.str());
 }
 
-Status Location::ForGrpcUnix(const std::string& path, Location* location) {
+Status Location::ForGrpcTls(const std::string& host, const int port, Location* location) {
+  return ForGrpcTls(host, port).Value(location);
+}
+
+arrow::Result<Location> Location::ForGrpcUnix(const std::string& path) {
   std::stringstream uri_string;
   uri_string << "grpc+unix://" << path;
-  return Location::Parse(uri_string.str(), location);
+  return Location::Parse(uri_string.str());
+}
+
+Status Location::ForGrpcUnix(const std::string& path, Location* location) {
+  return ForGrpcUnix(path).Value(location);
 }
 
 arrow::Result<Location> Location::ForScheme(const std::string& scheme,
                                             const std::string& host, const int port) {
-  Location location;
   std::stringstream uri_string;
   uri_string << scheme << "://" << host << ':' << port;
-  RETURN_NOT_OK(Location::Parse(uri_string.str(), &location));
-  return location;
+  return Location::Parse(uri_string.str());
 }
 
 std::string Location::ToString() const { return uri_->ToString(); }
@@ -337,23 +365,36 @@ bool ActionType::Equals(const ActionType& other) const {
   return type == other.type && description == other.description;
 }
 
-Status MetadataRecordBatchReader::ReadAll(
-    std::vector<std::shared_ptr<RecordBatch>>* batches) {
-  FlightStreamChunk chunk;
+Status ResultStream::Next(std::unique_ptr<Result>* info) { return Next().Value(info); }
+
+Status MetadataRecordBatchReader::Next(FlightStreamChunk* next) {
+  return Next().Value(next);
+}
 
+arrow::Result<std::vector<std::shared_ptr<RecordBatch>>>
+MetadataRecordBatchReader::ToRecordBatches() {
+  std::vector<std::shared_ptr<RecordBatch>> batches;
   while (true) {
-    RETURN_NOT_OK(Next(&chunk));
+    ARROW_ASSIGN_OR_RAISE(FlightStreamChunk chunk, Next());
     if (!chunk.data) break;
-    batches->emplace_back(std::move(chunk.data));
+    batches.emplace_back(std::move(chunk.data));
   }
-  return Status::OK();
+  return batches;
 }
 
-Status MetadataRecordBatchReader::ReadAll(std::shared_ptr<Table>* table) {
-  std::vector<std::shared_ptr<RecordBatch>> batches;
-  RETURN_NOT_OK(ReadAll(&batches));
+Status MetadataRecordBatchReader::ReadAll(
+    std::vector<std::shared_ptr<RecordBatch>>* batches) {
+  return ToRecordBatches().Value(batches);
+}
+
+arrow::Result<std::shared_ptr<Table>> MetadataRecordBatchReader::ToTable() {
+  ARROW_ASSIGN_OR_RAISE(auto batches, ToRecordBatches());
   ARROW_ASSIGN_OR_RAISE(auto schema, GetSchema());
-  return Table::FromRecordBatches(schema, std::move(batches)).Value(table);
+  return Table::FromRecordBatches(schema, std::move(batches));
+}
+
+Status MetadataRecordBatchReader::ReadAll(std::shared_ptr<Table>* table) {
+  return ToTable().Value(table);
 }
 
 Status MetadataRecordBatchWriter::Begin(const std::shared_ptr<Schema>& schema) {
@@ -402,25 +443,21 @@ SimpleFlightListing::SimpleFlightListing(const std::vector<FlightInfo>& flights)
 SimpleFlightListing::SimpleFlightListing(std::vector<FlightInfo>&& flights)
     : position_(0), flights_(std::move(flights)) {}
 
-Status SimpleFlightListing::Next(std::unique_ptr<FlightInfo>* info) {
+arrow::Result<std::unique_ptr<FlightInfo>> SimpleFlightListing::Next() {
   if (position_ >= static_cast<int>(flights_.size())) {
-    *info = nullptr;
-    return Status::OK();
+    return nullptr;
   }
-  *info = std::unique_ptr<FlightInfo>(new FlightInfo(std::move(flights_[position_++])));
-  return Status::OK();
+  return std::unique_ptr<FlightInfo>(new FlightInfo(std::move(flights_[position_++])));
 }
 
 SimpleResultStream::SimpleResultStream(std::vector<Result>&& results)
     : results_(std::move(results)), position_(0) {}
 
-Status SimpleResultStream::Next(std::unique_ptr<Result>* result) {
+arrow::Result<std::unique_ptr<Result>> SimpleResultStream::Next() {
   if (position_ >= results_.size()) {
-    *result = nullptr;
-    return Status::OK();
+    return nullptr;
   }
-  *result = std::unique_ptr<Result>(new Result(std::move(results_[position_++])));
-  return Status::OK();
+  return std::unique_ptr<Result>(new Result(std::move(results_[position_++])));
 }
 
 arrow::Result<BasicAuth> BasicAuth::Deserialize(arrow::util::string_view serialized) {