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/01/18 22:10:15 UTC
[GitHub] [arrow-cookbook] lidavidm opened a new pull request #124: [C++][Flight] Add basic Flight service
lidavidm opened a new pull request #124:
URL: https://github.com/apache/arrow-cookbook/pull/124
Also fixes a bug(?) with merging outputs.
--
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
[GitHub] [arrow-cookbook] lidavidm commented on a change in pull request #124: [C++][Flight] Add basic Flight service
Posted by GitBox <gi...@apache.org>.
lidavidm commented on a change in pull request #124:
URL: https://github.com/apache/arrow-cookbook/pull/124#discussion_r802847613
##########
File path: cpp/code/flight.cc
##########
@@ -0,0 +1,300 @@
+// 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.
+
+#include <arrow/buffer.h>
+#include <arrow/filesystem/filesystem.h>
+#include <arrow/filesystem/localfs.h>
+#include <arrow/flight/client.h>
+#include <arrow/flight/server.h>
+#include <arrow/pretty_print.h>
+#include <arrow/result.h>
+#include <arrow/status.h>
+#include <arrow/table.h>
+#include <arrow/type.h>
+#include <gtest/gtest.h>
+#include <parquet/arrow/reader.h>
+#include <parquet/arrow/writer.h>
+
+#include <algorithm>
+#include <memory>
+#include <numeric>
+#include <vector>
+
+#include "common.h"
+
+class ParquetStorageService : public arrow::flight::FlightServerBase {
+ public:
+ const arrow::flight::ActionType kActionDropDataset{"drop_dataset",
+ "Delete a dataset."};
+
+ explicit ParquetStorageService(std::shared_ptr<arrow::fs::FileSystem> root)
+ : root_(std::move(root)) {}
+
+ arrow::Status ListFlights(
+ const arrow::flight::ServerCallContext&, const arrow::flight::Criteria*,
+ std::unique_ptr<arrow::flight::FlightListing>* listings) {
+ arrow::fs::FileSelector selector;
+ selector.base_dir = "/";
+ ARROW_ASSIGN_OR_RAISE(auto listing, root_->GetFileInfo(selector));
+
+ std::vector<arrow::flight::FlightInfo> flights;
+ for (const auto& file_info : listing) {
+ if (!file_info.IsFile() || file_info.extension() != "parquet") continue;
+ ARROW_ASSIGN_OR_RAISE(auto info, MakeFlightInfo(file_info));
+ flights.push_back(std::move(info));
+ }
+
+ *listings = std::unique_ptr<arrow::flight::FlightListing>(
+ new arrow::flight::SimpleFlightListing(std::move(flights)));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status GetFlightInfo(
+ const arrow::flight::ServerCallContext&,
+ const arrow::flight::FlightDescriptor& descriptor,
+ std::unique_ptr<arrow::flight::FlightInfo>* info) {
+ ARROW_ASSIGN_OR_RAISE(auto file_info, FileInfoFromDescriptor(descriptor));
+ ARROW_ASSIGN_OR_RAISE(auto flight_info, MakeFlightInfo(file_info));
+ *info = std::unique_ptr<arrow::flight::FlightInfo>(
+ new arrow::flight::FlightInfo(std::move(flight_info)));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoPut(
+ const arrow::flight::ServerCallContext&,
+ std::unique_ptr<arrow::flight::FlightMessageReader> reader,
+ std::unique_ptr<arrow::flight::FlightMetadataWriter>) {
+ ARROW_ASSIGN_OR_RAISE(auto file_info,
+ FileInfoFromDescriptor(reader->descriptor()));
+ ARROW_ASSIGN_OR_RAISE(auto sink, root_->OpenOutputStream(file_info.path()));
+ std::shared_ptr<arrow::Table> table;
+ ARROW_RETURN_NOT_OK(reader->ReadAll(&table));
+
+ ARROW_RETURN_NOT_OK(parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), sink, /*chunk_size=*/65536));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoGet(
+ const arrow::flight::ServerCallContext&,
+ const arrow::flight::Ticket& request,
+ std::unique_ptr<arrow::flight::FlightDataStream>* stream) {
+ ARROW_ASSIGN_OR_RAISE(auto input, root_->OpenInputFile(request.ticket));
+ std::unique_ptr<parquet::arrow::FileReader> reader;
+ ARROW_RETURN_NOT_OK(parquet::arrow::OpenFile(
+ std::move(input), arrow::default_memory_pool(), &reader));
+
+ std::shared_ptr<arrow::Table> table;
+ ARROW_RETURN_NOT_OK(reader->ReadTable(&table));
+ // Note that we can't directly pass TableBatchReader to
+ // RecordBatchStream because TableBatchReader keeps a non-owning
+ // reference to the underlying Table, which would then get freed
+ // when we exit this function
+ std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+ arrow::TableBatchReader batch_reader(*table);
+ ARROW_RETURN_NOT_OK(batch_reader.ReadAll(&batches));
+
+ ARROW_ASSIGN_OR_RAISE(
+ auto owning_reader,
+ arrow::RecordBatchReader::Make(std::move(batches), table->schema()));
+ *stream = std::unique_ptr<arrow::flight::FlightDataStream>(
+ new arrow::flight::RecordBatchStream(owning_reader));
+
+ return arrow::Status::OK();
+ }
+
+ arrow::Status ListActions(
+ const arrow::flight::ServerCallContext&,
+ std::vector<arrow::flight::ActionType>* actions) override {
+ *actions = {kActionDropDataset};
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoAction(const arrow::flight::ServerCallContext&,
+ const arrow::flight::Action& action,
+ std::unique_ptr<arrow::flight::ResultStream>* result) {
+ if (action.type == kActionDropDataset.type) {
+ *result = std::unique_ptr<arrow::flight::ResultStream>(
+ new arrow::flight::SimpleResultStream({}));
+ return DoActionDropDataset(action.body->ToString());
+ }
+ return arrow::Status::NotImplemented("Unknown action type: ", action.type);
+ }
+
+ private:
+ arrow::Result<arrow::flight::FlightInfo> MakeFlightInfo(
+ const arrow::fs::FileInfo& file_info) {
+ ARROW_ASSIGN_OR_RAISE(auto input, root_->OpenInputFile(file_info));
+ std::unique_ptr<parquet::arrow::FileReader> reader;
+ ARROW_RETURN_NOT_OK(parquet::arrow::OpenFile(
+ std::move(input), arrow::default_memory_pool(), &reader));
+
+ std::shared_ptr<arrow::Schema> schema;
+ ARROW_RETURN_NOT_OK(reader->GetSchema(&schema));
+
+ auto descriptor =
+ arrow::flight::FlightDescriptor::Path({file_info.base_name()});
+
+ arrow::flight::FlightEndpoint endpoint;
+ endpoint.ticket.ticket = file_info.base_name();
+ arrow::flight::Location location;
+ ARROW_RETURN_NOT_OK(
+ arrow::flight::Location::ForGrpcTcp("localhost", port(), &location));
+ endpoint.locations.push_back(location);
+
+ int64_t total_records = reader->parquet_reader()->metadata()->num_rows();
+ int64_t total_bytes = file_info.size();
+
+ return arrow::flight::FlightInfo::Make(*schema, descriptor, {endpoint},
+ total_records, total_bytes);
+ }
+
+ arrow::Result<arrow::fs::FileInfo> FileInfoFromDescriptor(
+ const arrow::flight::FlightDescriptor& descriptor) {
+ if (descriptor.type != arrow::flight::FlightDescriptor::PATH) {
+ return arrow::Status::Invalid("Must provide PATH-type FlightDescriptor");
+ } else if (descriptor.path.size() != 1) {
+ return arrow::Status::Invalid(
+ "Must provide PATH-type FlightDescriptor with one path component");
+ }
+ return root_->GetFileInfo(descriptor.path[0]);
+ }
+
+ arrow::Status DoActionDropDataset(const std::string& key) {
+ return root_->DeleteFile(key);
+ }
+
+ std::shared_ptr<arrow::fs::FileSystem> root_;
+}; // end ParquetStorageService
+
+TEST(ParquetStorageServiceTest, PutGetDelete) {
+ StartRecipe("ParquetStorageService::StartServer");
+ auto fs = std::make_shared<arrow::fs::LocalFileSystem>();
+ ASSERT_OK(fs->CreateDir("./flight_datasets/"));
Review comment:
Updated (finally, sorry for the delay).
--
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
[GitHub] [arrow-cookbook] pitrou commented on a change in pull request #124: [C++][Flight] Add basic Flight service
Posted by GitBox <gi...@apache.org>.
pitrou commented on a change in pull request #124:
URL: https://github.com/apache/arrow-cookbook/pull/124#discussion_r797650266
##########
File path: cpp/code/flight.cc
##########
@@ -0,0 +1,300 @@
+// 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.
+
+#include <arrow/buffer.h>
+#include <arrow/filesystem/filesystem.h>
+#include <arrow/filesystem/localfs.h>
+#include <arrow/flight/client.h>
+#include <arrow/flight/server.h>
+#include <arrow/pretty_print.h>
+#include <arrow/result.h>
+#include <arrow/status.h>
+#include <arrow/table.h>
+#include <arrow/type.h>
+#include <gtest/gtest.h>
+#include <parquet/arrow/reader.h>
+#include <parquet/arrow/writer.h>
+
+#include <algorithm>
+#include <memory>
+#include <numeric>
+#include <vector>
+
+#include "common.h"
+
+class ParquetStorageService : public arrow::flight::FlightServerBase {
+ public:
+ const arrow::flight::ActionType kActionDropDataset{"drop_dataset",
+ "Delete a dataset."};
+
+ explicit ParquetStorageService(std::shared_ptr<arrow::fs::FileSystem> root)
+ : root_(std::move(root)) {}
+
+ arrow::Status ListFlights(
+ const arrow::flight::ServerCallContext&, const arrow::flight::Criteria*,
+ std::unique_ptr<arrow::flight::FlightListing>* listings) {
+ arrow::fs::FileSelector selector;
+ selector.base_dir = "/";
+ ARROW_ASSIGN_OR_RAISE(auto listing, root_->GetFileInfo(selector));
+
+ std::vector<arrow::flight::FlightInfo> flights;
+ for (const auto& file_info : listing) {
+ if (!file_info.IsFile() || file_info.extension() != "parquet") continue;
+ ARROW_ASSIGN_OR_RAISE(auto info, MakeFlightInfo(file_info));
+ flights.push_back(std::move(info));
+ }
+
+ *listings = std::unique_ptr<arrow::flight::FlightListing>(
+ new arrow::flight::SimpleFlightListing(std::move(flights)));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status GetFlightInfo(
+ const arrow::flight::ServerCallContext&,
+ const arrow::flight::FlightDescriptor& descriptor,
+ std::unique_ptr<arrow::flight::FlightInfo>* info) {
+ ARROW_ASSIGN_OR_RAISE(auto file_info, FileInfoFromDescriptor(descriptor));
+ ARROW_ASSIGN_OR_RAISE(auto flight_info, MakeFlightInfo(file_info));
+ *info = std::unique_ptr<arrow::flight::FlightInfo>(
+ new arrow::flight::FlightInfo(std::move(flight_info)));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoPut(
+ const arrow::flight::ServerCallContext&,
+ std::unique_ptr<arrow::flight::FlightMessageReader> reader,
+ std::unique_ptr<arrow::flight::FlightMetadataWriter>) {
+ ARROW_ASSIGN_OR_RAISE(auto file_info,
+ FileInfoFromDescriptor(reader->descriptor()));
+ ARROW_ASSIGN_OR_RAISE(auto sink, root_->OpenOutputStream(file_info.path()));
+ std::shared_ptr<arrow::Table> table;
+ ARROW_RETURN_NOT_OK(reader->ReadAll(&table));
+
+ ARROW_RETURN_NOT_OK(parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), sink, /*chunk_size=*/65536));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoGet(
+ const arrow::flight::ServerCallContext&,
+ const arrow::flight::Ticket& request,
+ std::unique_ptr<arrow::flight::FlightDataStream>* stream) {
+ ARROW_ASSIGN_OR_RAISE(auto input, root_->OpenInputFile(request.ticket));
+ std::unique_ptr<parquet::arrow::FileReader> reader;
+ ARROW_RETURN_NOT_OK(parquet::arrow::OpenFile(
+ std::move(input), arrow::default_memory_pool(), &reader));
+
+ std::shared_ptr<arrow::Table> table;
+ ARROW_RETURN_NOT_OK(reader->ReadTable(&table));
+ // Note that we can't directly pass TableBatchReader to
+ // RecordBatchStream because TableBatchReader keeps a non-owning
+ // reference to the underlying Table, which would then get freed
+ // when we exit this function
+ std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+ arrow::TableBatchReader batch_reader(*table);
+ ARROW_RETURN_NOT_OK(batch_reader.ReadAll(&batches));
+
+ ARROW_ASSIGN_OR_RAISE(
+ auto owning_reader,
+ arrow::RecordBatchReader::Make(std::move(batches), table->schema()));
+ *stream = std::unique_ptr<arrow::flight::FlightDataStream>(
+ new arrow::flight::RecordBatchStream(owning_reader));
+
+ return arrow::Status::OK();
+ }
+
+ arrow::Status ListActions(
+ const arrow::flight::ServerCallContext&,
+ std::vector<arrow::flight::ActionType>* actions) override {
+ *actions = {kActionDropDataset};
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoAction(const arrow::flight::ServerCallContext&,
+ const arrow::flight::Action& action,
+ std::unique_ptr<arrow::flight::ResultStream>* result) {
+ if (action.type == kActionDropDataset.type) {
+ *result = std::unique_ptr<arrow::flight::ResultStream>(
+ new arrow::flight::SimpleResultStream({}));
+ return DoActionDropDataset(action.body->ToString());
+ }
+ return arrow::Status::NotImplemented("Unknown action type: ", action.type);
+ }
+
+ private:
+ arrow::Result<arrow::flight::FlightInfo> MakeFlightInfo(
+ const arrow::fs::FileInfo& file_info) {
+ ARROW_ASSIGN_OR_RAISE(auto input, root_->OpenInputFile(file_info));
+ std::unique_ptr<parquet::arrow::FileReader> reader;
+ ARROW_RETURN_NOT_OK(parquet::arrow::OpenFile(
+ std::move(input), arrow::default_memory_pool(), &reader));
+
+ std::shared_ptr<arrow::Schema> schema;
+ ARROW_RETURN_NOT_OK(reader->GetSchema(&schema));
+
+ auto descriptor =
+ arrow::flight::FlightDescriptor::Path({file_info.base_name()});
+
+ arrow::flight::FlightEndpoint endpoint;
+ endpoint.ticket.ticket = file_info.base_name();
+ arrow::flight::Location location;
+ ARROW_RETURN_NOT_OK(
+ arrow::flight::Location::ForGrpcTcp("localhost", port(), &location));
+ endpoint.locations.push_back(location);
+
+ int64_t total_records = reader->parquet_reader()->metadata()->num_rows();
+ int64_t total_bytes = file_info.size();
+
+ return arrow::flight::FlightInfo::Make(*schema, descriptor, {endpoint},
+ total_records, total_bytes);
+ }
+
+ arrow::Result<arrow::fs::FileInfo> FileInfoFromDescriptor(
+ const arrow::flight::FlightDescriptor& descriptor) {
+ if (descriptor.type != arrow::flight::FlightDescriptor::PATH) {
+ return arrow::Status::Invalid("Must provide PATH-type FlightDescriptor");
+ } else if (descriptor.path.size() != 1) {
+ return arrow::Status::Invalid(
+ "Must provide PATH-type FlightDescriptor with one path component");
+ }
+ return root_->GetFileInfo(descriptor.path[0]);
+ }
+
+ arrow::Status DoActionDropDataset(const std::string& key) {
+ return root_->DeleteFile(key);
+ }
+
+ std::shared_ptr<arrow::fs::FileSystem> root_;
+}; // end ParquetStorageService
+
+TEST(ParquetStorageServiceTest, PutGetDelete) {
+ StartRecipe("ParquetStorageService::StartServer");
+ auto fs = std::make_shared<arrow::fs::LocalFileSystem>();
+ ASSERT_OK(fs->CreateDir("./flight_datasets/"));
Review comment:
Wouldn't it be better to define a helper function returning a `Status`, and use `ARROW_RETURN_NOT_OK` / `ARROW_ASSIGN_OR_RAISE` instead of asserts?
Related, though the solution there is different: https://github.com/apache/arrow-cookbook/pull/52
--
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
[GitHub] [arrow-cookbook] lidavidm commented on pull request #124: [C++][Flight] Add basic Flight service
Posted by GitBox <gi...@apache.org>.
lidavidm commented on pull request #124:
URL: https://github.com/apache/arrow-cookbook/pull/124#issuecomment-1016698370
<details>
<summary>Rendered</summary>
</details>
--
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
[GitHub] [arrow-cookbook] lidavidm commented on a change in pull request #124: [C++][Flight] Add basic Flight service
Posted by GitBox <gi...@apache.org>.
lidavidm commented on a change in pull request #124:
URL: https://github.com/apache/arrow-cookbook/pull/124#discussion_r797657847
##########
File path: cpp/code/flight.cc
##########
@@ -0,0 +1,300 @@
+// 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.
+
+#include <arrow/buffer.h>
+#include <arrow/filesystem/filesystem.h>
+#include <arrow/filesystem/localfs.h>
+#include <arrow/flight/client.h>
+#include <arrow/flight/server.h>
+#include <arrow/pretty_print.h>
+#include <arrow/result.h>
+#include <arrow/status.h>
+#include <arrow/table.h>
+#include <arrow/type.h>
+#include <gtest/gtest.h>
+#include <parquet/arrow/reader.h>
+#include <parquet/arrow/writer.h>
+
+#include <algorithm>
+#include <memory>
+#include <numeric>
+#include <vector>
+
+#include "common.h"
+
+class ParquetStorageService : public arrow::flight::FlightServerBase {
+ public:
+ const arrow::flight::ActionType kActionDropDataset{"drop_dataset",
+ "Delete a dataset."};
+
+ explicit ParquetStorageService(std::shared_ptr<arrow::fs::FileSystem> root)
+ : root_(std::move(root)) {}
+
+ arrow::Status ListFlights(
+ const arrow::flight::ServerCallContext&, const arrow::flight::Criteria*,
+ std::unique_ptr<arrow::flight::FlightListing>* listings) {
+ arrow::fs::FileSelector selector;
+ selector.base_dir = "/";
+ ARROW_ASSIGN_OR_RAISE(auto listing, root_->GetFileInfo(selector));
+
+ std::vector<arrow::flight::FlightInfo> flights;
+ for (const auto& file_info : listing) {
+ if (!file_info.IsFile() || file_info.extension() != "parquet") continue;
+ ARROW_ASSIGN_OR_RAISE(auto info, MakeFlightInfo(file_info));
+ flights.push_back(std::move(info));
+ }
+
+ *listings = std::unique_ptr<arrow::flight::FlightListing>(
+ new arrow::flight::SimpleFlightListing(std::move(flights)));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status GetFlightInfo(
+ const arrow::flight::ServerCallContext&,
+ const arrow::flight::FlightDescriptor& descriptor,
+ std::unique_ptr<arrow::flight::FlightInfo>* info) {
+ ARROW_ASSIGN_OR_RAISE(auto file_info, FileInfoFromDescriptor(descriptor));
+ ARROW_ASSIGN_OR_RAISE(auto flight_info, MakeFlightInfo(file_info));
+ *info = std::unique_ptr<arrow::flight::FlightInfo>(
+ new arrow::flight::FlightInfo(std::move(flight_info)));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoPut(
+ const arrow::flight::ServerCallContext&,
+ std::unique_ptr<arrow::flight::FlightMessageReader> reader,
+ std::unique_ptr<arrow::flight::FlightMetadataWriter>) {
+ ARROW_ASSIGN_OR_RAISE(auto file_info,
+ FileInfoFromDescriptor(reader->descriptor()));
+ ARROW_ASSIGN_OR_RAISE(auto sink, root_->OpenOutputStream(file_info.path()));
+ std::shared_ptr<arrow::Table> table;
+ ARROW_RETURN_NOT_OK(reader->ReadAll(&table));
+
+ ARROW_RETURN_NOT_OK(parquet::arrow::WriteTable(
+ *table, arrow::default_memory_pool(), sink, /*chunk_size=*/65536));
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoGet(
+ const arrow::flight::ServerCallContext&,
+ const arrow::flight::Ticket& request,
+ std::unique_ptr<arrow::flight::FlightDataStream>* stream) {
+ ARROW_ASSIGN_OR_RAISE(auto input, root_->OpenInputFile(request.ticket));
+ std::unique_ptr<parquet::arrow::FileReader> reader;
+ ARROW_RETURN_NOT_OK(parquet::arrow::OpenFile(
+ std::move(input), arrow::default_memory_pool(), &reader));
+
+ std::shared_ptr<arrow::Table> table;
+ ARROW_RETURN_NOT_OK(reader->ReadTable(&table));
+ // Note that we can't directly pass TableBatchReader to
+ // RecordBatchStream because TableBatchReader keeps a non-owning
+ // reference to the underlying Table, which would then get freed
+ // when we exit this function
+ std::vector<std::shared_ptr<arrow::RecordBatch>> batches;
+ arrow::TableBatchReader batch_reader(*table);
+ ARROW_RETURN_NOT_OK(batch_reader.ReadAll(&batches));
+
+ ARROW_ASSIGN_OR_RAISE(
+ auto owning_reader,
+ arrow::RecordBatchReader::Make(std::move(batches), table->schema()));
+ *stream = std::unique_ptr<arrow::flight::FlightDataStream>(
+ new arrow::flight::RecordBatchStream(owning_reader));
+
+ return arrow::Status::OK();
+ }
+
+ arrow::Status ListActions(
+ const arrow::flight::ServerCallContext&,
+ std::vector<arrow::flight::ActionType>* actions) override {
+ *actions = {kActionDropDataset};
+ return arrow::Status::OK();
+ }
+
+ arrow::Status DoAction(const arrow::flight::ServerCallContext&,
+ const arrow::flight::Action& action,
+ std::unique_ptr<arrow::flight::ResultStream>* result) {
+ if (action.type == kActionDropDataset.type) {
+ *result = std::unique_ptr<arrow::flight::ResultStream>(
+ new arrow::flight::SimpleResultStream({}));
+ return DoActionDropDataset(action.body->ToString());
+ }
+ return arrow::Status::NotImplemented("Unknown action type: ", action.type);
+ }
+
+ private:
+ arrow::Result<arrow::flight::FlightInfo> MakeFlightInfo(
+ const arrow::fs::FileInfo& file_info) {
+ ARROW_ASSIGN_OR_RAISE(auto input, root_->OpenInputFile(file_info));
+ std::unique_ptr<parquet::arrow::FileReader> reader;
+ ARROW_RETURN_NOT_OK(parquet::arrow::OpenFile(
+ std::move(input), arrow::default_memory_pool(), &reader));
+
+ std::shared_ptr<arrow::Schema> schema;
+ ARROW_RETURN_NOT_OK(reader->GetSchema(&schema));
+
+ auto descriptor =
+ arrow::flight::FlightDescriptor::Path({file_info.base_name()});
+
+ arrow::flight::FlightEndpoint endpoint;
+ endpoint.ticket.ticket = file_info.base_name();
+ arrow::flight::Location location;
+ ARROW_RETURN_NOT_OK(
+ arrow::flight::Location::ForGrpcTcp("localhost", port(), &location));
+ endpoint.locations.push_back(location);
+
+ int64_t total_records = reader->parquet_reader()->metadata()->num_rows();
+ int64_t total_bytes = file_info.size();
+
+ return arrow::flight::FlightInfo::Make(*schema, descriptor, {endpoint},
+ total_records, total_bytes);
+ }
+
+ arrow::Result<arrow::fs::FileInfo> FileInfoFromDescriptor(
+ const arrow::flight::FlightDescriptor& descriptor) {
+ if (descriptor.type != arrow::flight::FlightDescriptor::PATH) {
+ return arrow::Status::Invalid("Must provide PATH-type FlightDescriptor");
+ } else if (descriptor.path.size() != 1) {
+ return arrow::Status::Invalid(
+ "Must provide PATH-type FlightDescriptor with one path component");
+ }
+ return root_->GetFileInfo(descriptor.path[0]);
+ }
+
+ arrow::Status DoActionDropDataset(const std::string& key) {
+ return root_->DeleteFile(key);
+ }
+
+ std::shared_ptr<arrow::fs::FileSystem> root_;
+}; // end ParquetStorageService
+
+TEST(ParquetStorageServiceTest, PutGetDelete) {
+ StartRecipe("ParquetStorageService::StartServer");
+ auto fs = std::make_shared<arrow::fs::LocalFileSystem>();
+ ASSERT_OK(fs->CreateDir("./flight_datasets/"));
Review comment:
Ah, yes, having that helper in #52 would clean things up here. A helper function would work fine right now as well.
--
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
[GitHub] [arrow-cookbook] pitrou merged pull request #124: [C++][Flight] Add basic Flight service
Posted by GitBox <gi...@apache.org>.
pitrou merged pull request #124:
URL: https://github.com/apache/arrow-cookbook/pull/124
--
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