You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by ra...@apache.org on 2025/06/16 07:30:10 UTC
(arrow-skyhook) branch main updated (245feb8 -> 3652416)
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
from 245feb8 chore: Add .asf.yaml
new 89572bc ARROW-13607: [C++] Add Skyhook to Arrow
new f2f8b00 ARROW-15300: [C++] Update Skyhook for async dataset interfaces
new d196efa ARROW-15984: [C++] Change RecordBatchReader API to use Result<>
new 1437ff0 ARROW-16585: [C++] Add support for absolute CMAKE_INSTALL_*DIR
new 8a4e55b ARROW-15927: [C++][Skyhook] Add skyhook example (#12620)
new a011988 ARROW-17814: [C++] Remove make_unique reimplementation (#14204)
new 78e8bf7 GH-14990: [C++][Skyhook] Follow FileFormat API change (#15086)
new 8747d6e GH-34269: [C++] Fix include file name (#34285)
new e614565 GH-34630: [C++] Second block of refactoring to move acero out of libarrow (#34575)
new d70c798 GH-37485: [C++][Skyhook] Don't use deprecated BufferReader API (#37486)
new 424c863 GH-38401: [C++] Re-generate flatbuffers C++ for Skyhook (#38405)
new cc456d6 GH-38979: [C++] Fix spelling (#38980)
new f72fc46 GH-39823: [C++] Allow building cpp/src/arrow/**/*.cc without waiting bundled libraries (#39824)
new 73af16c GH-40882: [C++] Suppress shorten-64-to-32 warnings in CUDA/Skyhook codes (#40883)
new c84b1d1 GH-41478: [C++] Clean up more redundant move warnings (#41487)
new 65328d0 GH-45537: [CI][C++] Add missing includes (iwyu) to file_skyhook.cc (#45538)
new 076d4c9 GH-45694: [C++] Bump vendored flatbuffers to 24.3.6 (#45687)
new 3652416 GH-46011: [C++] Hide DCHECK family from public headers (#46015)
The 18 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
CMakeLists.txt | 93 +++++++++++++
.asf.yaml => client/CMakeLists.txt | 26 +---
client/file_skyhook.cc | 170 +++++++++++++++++++++++
client/file_skyhook.h | 107 +++++++++++++++
cls/cls_skyhook.cc | 267 +++++++++++++++++++++++++++++++++++++
cls/cls_skyhook_test.cc | 208 +++++++++++++++++++++++++++++
protocol/ScanRequest.fbs | 29 ++++
protocol/ScanRequest_generated.h | 176 ++++++++++++++++++++++++
protocol/rados_protocol.cc | 102 ++++++++++++++
protocol/rados_protocol.h | 104 +++++++++++++++
protocol/skyhook_protocol.cc | 139 +++++++++++++++++++
protocol/skyhook_protocol.h | 116 ++++++++++++++++
protocol/skyhook_protocol_test.cc | 71 ++++++++++
.asf.yaml => skyhook.pc.in | 32 ++---
14 files changed, 1591 insertions(+), 49 deletions(-)
create mode 100644 CMakeLists.txt
copy .asf.yaml => client/CMakeLists.txt (53%)
create mode 100644 client/file_skyhook.cc
create mode 100644 client/file_skyhook.h
create mode 100644 cls/cls_skyhook.cc
create mode 100644 cls/cls_skyhook_test.cc
create mode 100644 protocol/ScanRequest.fbs
create mode 100644 protocol/ScanRequest_generated.h
create mode 100644 protocol/rados_protocol.cc
create mode 100644 protocol/rados_protocol.h
create mode 100644 protocol/skyhook_protocol.cc
create mode 100644 protocol/skyhook_protocol.h
create mode 100644 protocol/skyhook_protocol_test.cc
copy .asf.yaml => skyhook.pc.in (53%)
(arrow-skyhook) 17/18: GH-45694: [C++] Bump vendored flatbuffers to 24.3.6 (#45687)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 076d4c93d9736ae33fb354908d8e9b4db09b1e72
Author: William Ayd <wi...@icloud.com>
AuthorDate: Mon Mar 10 17:41:28 2025 -0400
GH-45694: [C++] Bump vendored flatbuffers to 24.3.6 (#45687)
### Rationale for this change
This bumps the vendored copy of flatbuffers to a version that also exists in the Meson wrapdb, which will subsequently make using the Meson build system easier
### What changes are included in this PR?
I have re-vendored flatbuffers keeping the namespacing modifications for arrow inplace. I have also regenerated the appropriate header files using flatc version 24.3.6
### Are these changes tested?
In CI
### Are there any user-facing changes?
I do not believe so
* GitHub Issue: #45694
Authored-by: Will Ayd <wi...@icloud.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
protocol/ScanRequest_generated.h | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/protocol/ScanRequest_generated.h b/protocol/ScanRequest_generated.h
index ecf6ff7..798548e 100644
--- a/protocol/ScanRequest_generated.h
+++ b/protocol/ScanRequest_generated.h
@@ -8,9 +8,9 @@
// Ensure the included flatbuffers.h is the same version as when this file was
// generated, otherwise it may not be compatible.
-static_assert(FLATBUFFERS_VERSION_MAJOR == 23 &&
- FLATBUFFERS_VERSION_MINOR == 5 &&
- FLATBUFFERS_VERSION_REVISION == 26,
+static_assert(FLATBUFFERS_VERSION_MAJOR == 24 &&
+ FLATBUFFERS_VERSION_MINOR == 3 &&
+ FLATBUFFERS_VERSION_REVISION == 6,
"Non-compatible flatbuffers version included");
namespace org {
(arrow-skyhook) 02/18: ARROW-15300: [C++] Update Skyhook for async dataset interfaces
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit f2f8b00f209adaf964a33b2ec5e869a995877b89
Author: David Li <li...@gmail.com>
AuthorDate: Wed Jan 12 16:56:35 2022 -0500
ARROW-15300: [C++] Update Skyhook for async dataset interfaces
After the synchronous dataset interfaces were removed in ARROW-13554, we need to update Skyhook to use the async interfaces. For now, we just use the same synchronous path as before, but on an I/O thread.
Closes #12123 from lidavidm/arrow-15300
Authored-by: David Li <li...@gmail.com>
Signed-off-by: David Li <li...@gmail.com>
---
CMakeLists.txt | 7 ++--
client/file_skyhook.cc | 109 +++++++++++++++++++++---------------------------
client/file_skyhook.h | 11 +++--
cls/cls_skyhook_test.cc | 1 +
4 files changed, 57 insertions(+), 71 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 22a414c..992c467 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -63,11 +63,12 @@ add_dependencies(cls_skyhook ${ARROW_SKYHOOK_CLS_LIBRARIES})
# define the test builds
if(ARROW_TEST_LINKAGE STREQUAL "static")
- set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_dataset_static ${ARROW_TEST_STATIC_LINK_LIBS})
+ set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_skyhook_client_static arrow_dataset_static
+ ${ARROW_TEST_STATIC_LINK_LIBS})
else()
- set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_dataset_shared ${ARROW_TEST_SHARED_LINK_LIBS})
+ set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_skyhook_client_shared arrow_dataset_shared
+ ${ARROW_TEST_SHARED_LINK_LIBS})
endif()
-list(APPEND ARROW_SKYHOOK_TEST_LINK_LIBS ${ARROW_SKYHOOK_CLIENT_LIBRARIES})
# build the cls and protocol tests
add_arrow_test(cls_test
diff --git a/client/file_skyhook.cc b/client/file_skyhook.cc
index f8b57f4..cc26280 100644
--- a/client/file_skyhook.cc
+++ b/client/file_skyhook.cc
@@ -22,63 +22,11 @@
#include "arrow/dataset/file_base.h"
#include "arrow/dataset/file_ipc.h"
#include "arrow/dataset/file_parquet.h"
+#include "arrow/util/checked_cast.h"
#include "arrow/util/compression.h"
namespace skyhook {
-/// A ScanTask to scan a file fragment in Skyhook format.
-class SkyhookScanTask : public arrow::dataset::ScanTask {
- public:
- SkyhookScanTask(std::shared_ptr<arrow::dataset::ScanOptions> options,
- std::shared_ptr<arrow::dataset::Fragment> fragment,
- arrow::dataset::FileSource source,
- std::shared_ptr<skyhook::SkyhookDirectObjectAccess> doa,
- skyhook::SkyhookFileType::type file_format,
- arrow::compute::Expression partition_expression)
- : ScanTask(std::move(options), std::move(fragment)),
- source_(std::move(source)),
- doa_(std::move(doa)),
- file_format_(file_format),
- partition_expression_(std::move(partition_expression)) {}
-
- arrow::Result<arrow::RecordBatchIterator> Execute() override {
- /// Retrieve the size of the file using POSIX `stat`.
- struct stat st {};
- RETURN_NOT_OK(doa_->Stat(source_.path(), st));
-
- /// Create a ScanRequest instance.
- skyhook::ScanRequest req;
- req.filter_expression = options_->filter;
- req.partition_expression = partition_expression_;
- req.projection_schema = options_->projected_schema;
- req.dataset_schema = options_->dataset_schema;
- req.file_size = st.st_size;
- req.file_format = file_format_;
-
- /// Serialize the ScanRequest into a ceph bufferlist.
- ceph::bufferlist request;
- RETURN_NOT_OK(skyhook::SerializeScanRequest(req, &request));
-
- /// Execute the Ceph object class method `scan_op`.
- ceph::bufferlist result;
- RETURN_NOT_OK(doa_->Exec(st.st_ino, "scan_op", request, result));
-
- /// Read RecordBatches from the result bufferlist. Since, this step might use
- /// threads for decompressing compressed batches, to avoid running into
- /// [ARROW-12597], we switch off threaded decompression to avoid nested threading
- /// scenarios when scan tasks are executed in parallel by the CpuThreadPool.
- arrow::RecordBatchVector batches;
- RETURN_NOT_OK(skyhook::DeserializeTable(result, !options_->use_threads, &batches));
- return arrow::MakeVectorIterator(std::move(batches));
- }
-
- protected:
- arrow::dataset::FileSource source_;
- std::shared_ptr<skyhook::SkyhookDirectObjectAccess> doa_;
- skyhook::SkyhookFileType::type file_format_;
- arrow::compute::Expression partition_expression_;
-};
-
class SkyhookFileFormat::Impl {
public:
Impl(std::shared_ptr<RadosConnCtx> ctx, std::string file_format)
@@ -95,12 +43,10 @@ class SkyhookFileFormat::Impl {
return arrow::Status::OK();
}
- arrow::Result<arrow::dataset::ScanTaskIterator> ScanFile(
+ arrow::Result<arrow::RecordBatchGenerator> ScanBatchesAsync(
+ const std::shared_ptr<const SkyhookFileFormat>& format,
const std::shared_ptr<arrow::dataset::ScanOptions>& options,
const std::shared_ptr<arrow::dataset::FileFragment>& file) const {
- /// Make sure client-side filtering and projection is turned off.
- file->apply_compute = false;
-
/// Convert string file format name to Enum.
skyhook::SkyhookFileType::type file_format;
if (file_format_ == "parquet") {
@@ -111,9 +57,46 @@ class SkyhookFileFormat::Impl {
return arrow::Status::Invalid("Unsupported file format ", file_format_);
}
- arrow::dataset::ScanTaskVector v{std::make_shared<SkyhookScanTask>(
- options, file, file->source(), doa_, file_format, file->partition_expression())};
- return arrow::MakeVectorIterator(v);
+ auto fut = arrow::DeferNotOk(options->io_context.executor()->Submit(
+ [file, file_format, format,
+ options]() -> arrow::Result<arrow::RecordBatchGenerator> {
+ auto self = format->impl_.get();
+
+ /// Retrieve the size of the file using POSIX `stat`.
+ struct stat st {};
+ RETURN_NOT_OK(self->doa_->Stat(file->source().path(), st));
+
+ /// Create a ScanRequest instance.
+ skyhook::ScanRequest req;
+ req.filter_expression = options->filter;
+ req.partition_expression = file->partition_expression();
+ req.projection_schema = options->projected_schema;
+ req.dataset_schema = options->dataset_schema;
+ req.file_size = st.st_size;
+ req.file_format = file_format;
+
+ /// Serialize the ScanRequest into a ceph bufferlist.
+ ceph::bufferlist request;
+ RETURN_NOT_OK(skyhook::SerializeScanRequest(req, &request));
+
+ /// Execute the Ceph object class method `scan_op`.
+ ceph::bufferlist result;
+ RETURN_NOT_OK(self->doa_->Exec(st.st_ino, "scan_op", request, result));
+
+ /// Read RecordBatches from the result bufferlist. Since, this step might use
+ /// threads for decompressing compressed batches, to avoid running into
+ /// [ARROW-12597], we switch off threaded decompression to avoid nested
+ /// threading scenarios when scan tasks are executed in parallel by the
+ /// CpuThreadPool.
+ arrow::RecordBatchVector batches;
+ RETURN_NOT_OK(
+ skyhook::DeserializeTable(result, !options->use_threads, &batches));
+ auto gen = arrow::MakeVectorGenerator(std::move(batches));
+ // Keep Ceph client alive
+ arrow::RecordBatchGenerator gen_with_client = [format, gen]() { return gen(); };
+ return gen_with_client;
+ }));
+ return arrow::MakeFromFuture(std::move(fut));
}
arrow::Result<std::shared_ptr<arrow::Schema>> Inspect(
@@ -160,10 +143,12 @@ arrow::Result<std::shared_ptr<arrow::Schema>> SkyhookFileFormat::Inspect(
return impl_->Inspect(source);
}
-arrow::Result<arrow::dataset::ScanTaskIterator> SkyhookFileFormat::ScanFile(
+arrow::Result<arrow::RecordBatchGenerator> SkyhookFileFormat::ScanBatchesAsync(
const std::shared_ptr<arrow::dataset::ScanOptions>& options,
const std::shared_ptr<arrow::dataset::FileFragment>& file) const {
- return impl_->ScanFile(options, file);
+ return impl_->ScanBatchesAsync(
+ arrow::internal::checked_pointer_cast<const SkyhookFileFormat>(shared_from_this()),
+ options, file);
}
std::shared_ptr<arrow::dataset::FileWriteOptions>
diff --git a/client/file_skyhook.h b/client/file_skyhook.h
index 52a19f5..bec41fb 100644
--- a/client/file_skyhook.h
+++ b/client/file_skyhook.h
@@ -16,11 +16,14 @@
// under the License.
#pragma once
-#include "arrow/api.h"
+#include <memory>
+#include <string>
+
#include "arrow/dataset/file_parquet.h"
#include "arrow/dataset/scanner.h"
#include "arrow/dataset/type_fwd.h"
#include "arrow/dataset/visibility.h"
+#include "arrow/type_fwd.h"
namespace skyhook {
@@ -77,11 +80,7 @@ class SkyhookFileFormat : public arrow::dataset::FileFormat {
arrow::Result<std::shared_ptr<arrow::Schema>> Inspect(
const arrow::dataset::FileSource& source) const override;
- /// \brief Scan a file fragment.
- /// \param[in] options The ScanOptions to use.
- /// \param[in] file The file fragment to scan.
- /// \return An iterator of ScanTasks.
- arrow::Result<arrow::dataset::ScanTaskIterator> ScanFile(
+ arrow::Result<arrow::RecordBatchGenerator> ScanBatchesAsync(
const std::shared_ptr<arrow::dataset::ScanOptions>& options,
const std::shared_ptr<arrow::dataset::FileFragment>& file) const override;
diff --git a/cls/cls_skyhook_test.cc b/cls/cls_skyhook_test.cc
index 461cdd6..ebb1f4e 100644
--- a/cls/cls_skyhook_test.cc
+++ b/cls/cls_skyhook_test.cc
@@ -21,6 +21,7 @@
#include "arrow/dataset/file_base.h"
#include "arrow/filesystem/api.h"
#include "arrow/io/api.h"
+#include "arrow/table.h"
#include "arrow/testing/gtest_util.h"
#include "arrow/util/checked_cast.h"
#include "arrow/util/iterator.h"
(arrow-skyhook) 07/18: GH-14990: [C++][Skyhook] Follow FileFormat API change (#15086)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 78e8bf7e89d6d6d59ea8ff75b742f21ddad35087
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Mon Dec 26 05:36:53 2022 +0900
GH-14990: [C++][Skyhook] Follow FileFormat API change (#15086)
`arrow::dataset::FileFormat`'s constructor API was changed in ARROW-17288/GH-14663.
* Closes: #14990
Authored-by: Sutou Kouhei <ko...@clear-code.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
client/file_skyhook.cc | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/client/file_skyhook.cc b/client/file_skyhook.cc
index cc26280..725b04a 100644
--- a/client/file_skyhook.cc
+++ b/client/file_skyhook.cc
@@ -14,6 +14,7 @@
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
+
#include "skyhook/client/file_skyhook.h"
#include "skyhook/protocol/rados_protocol.h"
#include "skyhook/protocol/skyhook_protocol.h"
@@ -132,7 +133,7 @@ arrow::Result<std::shared_ptr<SkyhookFileFormat>> SkyhookFileFormat::Make(
SkyhookFileFormat::SkyhookFileFormat(std::shared_ptr<RadosConnCtx> ctx,
std::string file_format)
- : impl_(new Impl(std::move(ctx), std::move(file_format))) {}
+ : FileFormat(nullptr), impl_(new Impl(std::move(ctx), std::move(file_format))) {}
SkyhookFileFormat::~SkyhookFileFormat() = default;
(arrow-skyhook) 03/18: ARROW-15984: [C++] Change RecordBatchReader API to use Result<>
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit d196efa8b35274fb094f75974aae8068e5afb90d
Author: Tobias Zagorni <to...@zagorni.eu>
AuthorDate: Thu Mar 24 13:54:44 2022 +0100
ARROW-15984: [C++] Change RecordBatchReader API to use Result<>
Closes #12678 from zagto/recordbatchreader-result
Authored-by: Tobias Zagorni <to...@zagorni.eu>
Signed-off-by: Antoine Pitrou <an...@python.org>
---
protocol/skyhook_protocol.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/protocol/skyhook_protocol.cc b/protocol/skyhook_protocol.cc
index c261048..f83b124 100644
--- a/protocol/skyhook_protocol.cc
+++ b/protocol/skyhook_protocol.cc
@@ -116,7 +116,7 @@ arrow::Status DeserializeTable(ceph::bufferlist& bl, bool use_threads,
options.use_threads = use_threads;
ARROW_ASSIGN_OR_RAISE(
auto reader, arrow::ipc::RecordBatchStreamReader::Open(buffer_reader, options));
- ARROW_RETURN_NOT_OK(reader->ReadAll(batches));
+ ARROW_ASSIGN_OR_RAISE(*batches, reader->ToRecordBatches());
return arrow::Status::OK();
}
(arrow-skyhook) 01/18: ARROW-13607: [C++] Add Skyhook to Arrow
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 89572bcd1f1661fe858c06cfcafd85acfca6d85f
Author: Jayjeet Chakraborty <jc...@rediffmail.com>
AuthorDate: Fri Oct 22 08:17:08 2021 -1000
ARROW-13607: [C++] Add Skyhook to Arrow
Closes #10913 from JayjeetAtGithub/skyhook/pr_1
Authored-by: Jayjeet Chakraborty <jc...@rediffmail.com>
Signed-off-by: Weston Pace <we...@gmail.com>
---
CMakeLists.txt | 87 +++++++++++++
client/CMakeLists.txt | 18 +++
client/file_skyhook.cc | 182 ++++++++++++++++++++++++++
client/file_skyhook.h | 108 +++++++++++++++
cls/cls_skyhook.cc | 267 ++++++++++++++++++++++++++++++++++++++
cls/cls_skyhook_test.cc | 207 +++++++++++++++++++++++++++++
protocol/ScanRequest.fbs | 29 +++++
protocol/ScanRequest_generated.h | 167 ++++++++++++++++++++++++
protocol/rados_protocol.cc | 99 ++++++++++++++
protocol/rados_protocol.h | 103 +++++++++++++++
protocol/skyhook_protocol.cc | 136 +++++++++++++++++++
protocol/skyhook_protocol.h | 116 +++++++++++++++++
protocol/skyhook_protocol_test.cc | 71 ++++++++++
skyhook.pc.in | 26 ++++
14 files changed, 1616 insertions(+)
diff --git a/CMakeLists.txt b/CMakeLists.txt
new file mode 100644
index 0000000..22a414c
--- /dev/null
+++ b/CMakeLists.txt
@@ -0,0 +1,87 @@
+# 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 limitationsn
+# under the License.
+
+# add the client subdirectory
+add_subdirectory(client)
+
+# define the targets to build
+add_custom_target(arrow_skyhook_client)
+add_custom_target(cls_skyhook)
+
+# define the dependencies
+find_package(librados REQUIRED)
+set(ARROW_SKYHOOK_LINK_STATIC arrow_dataset_static librados::rados)
+set(ARROW_SKYHOOK_LINK_SHARED arrow_dataset_shared librados::rados)
+
+# define the client and cls sources
+set(ARROW_SKYHOOK_CLIENT_SOURCES client/file_skyhook.cc protocol/rados_protocol.cc
+ protocol/skyhook_protocol.cc)
+set(ARROW_SKYHOOK_CLS_SOURCES cls/cls_skyhook.cc protocol/rados_protocol.cc
+ protocol/skyhook_protocol.cc)
+
+# define the client library
+add_arrow_lib(arrow_skyhook_client
+ PKG_CONFIG_NAME
+ skyhook
+ SOURCES
+ ${ARROW_SKYHOOK_CLIENT_SOURCES}
+ OUTPUTS
+ ARROW_SKYHOOK_CLIENT_LIBRARIES
+ SHARED_LINK_LIBS
+ ${ARROW_SKYHOOK_LINK_SHARED}
+ STATIC_LINK_LIBS
+ ${ARROW_SKYHOOK_LINK_STATIC})
+
+# define the cls library
+add_arrow_lib(cls_skyhook
+ SOURCES
+ ${ARROW_SKYHOOK_CLS_SOURCES}
+ OUTPUTS
+ ARROW_SKYHOOK_CLS_LIBRARIES
+ SHARED_LINK_LIBS
+ ${ARROW_SKYHOOK_LINK_SHARED}
+ STATIC_LINK_LIBS
+ ${ARROW_SKYHOOK_LINK_STATIC})
+
+# finish building the project
+add_dependencies(arrow_skyhook_client ${ARROW_SKYHOOK_CLIENT_LIBRARIES})
+add_dependencies(cls_skyhook ${ARROW_SKYHOOK_CLS_LIBRARIES})
+
+# define the test builds
+if(ARROW_TEST_LINKAGE STREQUAL "static")
+ set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_dataset_static ${ARROW_TEST_STATIC_LINK_LIBS})
+else()
+ set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_dataset_shared ${ARROW_TEST_SHARED_LINK_LIBS})
+endif()
+list(APPEND ARROW_SKYHOOK_TEST_LINK_LIBS ${ARROW_SKYHOOK_CLIENT_LIBRARIES})
+
+# build the cls and protocol tests
+add_arrow_test(cls_test
+ SOURCES
+ cls/cls_skyhook_test.cc
+ EXTRA_LINK_LIBS
+ ${ARROW_SKYHOOK_TEST_LINK_LIBS}
+ PREFIX
+ "skyhook")
+
+add_arrow_test(protocol_test
+ SOURCES
+ protocol/skyhook_protocol_test.cc
+ EXTRA_LINK_LIBS
+ ${ARROW_SKYHOOK_TEST_LINK_LIBS}
+ PREFIX
+ "skyhook")
diff --git a/client/CMakeLists.txt b/client/CMakeLists.txt
new file mode 100644
index 0000000..6255d9a
--- /dev/null
+++ b/client/CMakeLists.txt
@@ -0,0 +1,18 @@
+# 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.
+
+arrow_install_all_headers("skyhook/client")
diff --git a/client/file_skyhook.cc b/client/file_skyhook.cc
new file mode 100644
index 0000000..f8b57f4
--- /dev/null
+++ b/client/file_skyhook.cc
@@ -0,0 +1,182 @@
+// 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 "skyhook/client/file_skyhook.h"
+#include "skyhook/protocol/rados_protocol.h"
+#include "skyhook/protocol/skyhook_protocol.h"
+
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/file_base.h"
+#include "arrow/dataset/file_ipc.h"
+#include "arrow/dataset/file_parquet.h"
+#include "arrow/util/compression.h"
+
+namespace skyhook {
+
+/// A ScanTask to scan a file fragment in Skyhook format.
+class SkyhookScanTask : public arrow::dataset::ScanTask {
+ public:
+ SkyhookScanTask(std::shared_ptr<arrow::dataset::ScanOptions> options,
+ std::shared_ptr<arrow::dataset::Fragment> fragment,
+ arrow::dataset::FileSource source,
+ std::shared_ptr<skyhook::SkyhookDirectObjectAccess> doa,
+ skyhook::SkyhookFileType::type file_format,
+ arrow::compute::Expression partition_expression)
+ : ScanTask(std::move(options), std::move(fragment)),
+ source_(std::move(source)),
+ doa_(std::move(doa)),
+ file_format_(file_format),
+ partition_expression_(std::move(partition_expression)) {}
+
+ arrow::Result<arrow::RecordBatchIterator> Execute() override {
+ /// Retrieve the size of the file using POSIX `stat`.
+ struct stat st {};
+ RETURN_NOT_OK(doa_->Stat(source_.path(), st));
+
+ /// Create a ScanRequest instance.
+ skyhook::ScanRequest req;
+ req.filter_expression = options_->filter;
+ req.partition_expression = partition_expression_;
+ req.projection_schema = options_->projected_schema;
+ req.dataset_schema = options_->dataset_schema;
+ req.file_size = st.st_size;
+ req.file_format = file_format_;
+
+ /// Serialize the ScanRequest into a ceph bufferlist.
+ ceph::bufferlist request;
+ RETURN_NOT_OK(skyhook::SerializeScanRequest(req, &request));
+
+ /// Execute the Ceph object class method `scan_op`.
+ ceph::bufferlist result;
+ RETURN_NOT_OK(doa_->Exec(st.st_ino, "scan_op", request, result));
+
+ /// Read RecordBatches from the result bufferlist. Since, this step might use
+ /// threads for decompressing compressed batches, to avoid running into
+ /// [ARROW-12597], we switch off threaded decompression to avoid nested threading
+ /// scenarios when scan tasks are executed in parallel by the CpuThreadPool.
+ arrow::RecordBatchVector batches;
+ RETURN_NOT_OK(skyhook::DeserializeTable(result, !options_->use_threads, &batches));
+ return arrow::MakeVectorIterator(std::move(batches));
+ }
+
+ protected:
+ arrow::dataset::FileSource source_;
+ std::shared_ptr<skyhook::SkyhookDirectObjectAccess> doa_;
+ skyhook::SkyhookFileType::type file_format_;
+ arrow::compute::Expression partition_expression_;
+};
+
+class SkyhookFileFormat::Impl {
+ public:
+ Impl(std::shared_ptr<RadosConnCtx> ctx, std::string file_format)
+ : ctx_(std::move(ctx)), file_format_(std::move(file_format)) {}
+
+ ~Impl() = default;
+
+ arrow::Status Init() {
+ /// Connect to the RADOS cluster and instantiate a `SkyhookDirectObjectAccess`
+ /// instance.
+ auto connection = std::make_shared<skyhook::rados::RadosConn>(ctx_);
+ RETURN_NOT_OK(connection->Connect());
+ doa_ = std::make_shared<skyhook::SkyhookDirectObjectAccess>(connection);
+ return arrow::Status::OK();
+ }
+
+ arrow::Result<arrow::dataset::ScanTaskIterator> ScanFile(
+ const std::shared_ptr<arrow::dataset::ScanOptions>& options,
+ const std::shared_ptr<arrow::dataset::FileFragment>& file) const {
+ /// Make sure client-side filtering and projection is turned off.
+ file->apply_compute = false;
+
+ /// Convert string file format name to Enum.
+ skyhook::SkyhookFileType::type file_format;
+ if (file_format_ == "parquet") {
+ file_format = skyhook::SkyhookFileType::type::PARQUET;
+ } else if (file_format_ == "ipc") {
+ file_format = skyhook::SkyhookFileType::type::IPC;
+ } else {
+ return arrow::Status::Invalid("Unsupported file format ", file_format_);
+ }
+
+ arrow::dataset::ScanTaskVector v{std::make_shared<SkyhookScanTask>(
+ options, file, file->source(), doa_, file_format, file->partition_expression())};
+ return arrow::MakeVectorIterator(v);
+ }
+
+ arrow::Result<std::shared_ptr<arrow::Schema>> Inspect(
+ const arrow::dataset::FileSource& source) const {
+ std::shared_ptr<arrow::dataset::FileFormat> file_format;
+ /// Convert string file format name to Arrow FileFormat.
+ if (file_format_ == "parquet") {
+ file_format = std::make_shared<arrow::dataset::ParquetFileFormat>();
+ } else if (file_format_ == "ipc") {
+ file_format = std::make_shared<arrow::dataset::IpcFileFormat>();
+ } else {
+ return arrow::Status::Invalid("Unsupported file format ", file_format_);
+ }
+ std::shared_ptr<arrow::Schema> schema;
+ ARROW_ASSIGN_OR_RAISE(schema, file_format->Inspect(source));
+ return schema;
+ }
+
+ private:
+ std::shared_ptr<skyhook::SkyhookDirectObjectAccess> doa_;
+ std::shared_ptr<RadosConnCtx> ctx_;
+ std::string file_format_;
+};
+
+arrow::Result<std::shared_ptr<SkyhookFileFormat>> SkyhookFileFormat::Make(
+ std::shared_ptr<RadosConnCtx> ctx, std::string file_format) {
+ auto format =
+ std::make_shared<SkyhookFileFormat>(std::move(ctx), std::move(file_format));
+ /// Establish connection to the Ceph cluster.
+ RETURN_NOT_OK(format->Init());
+ return format;
+}
+
+SkyhookFileFormat::SkyhookFileFormat(std::shared_ptr<RadosConnCtx> ctx,
+ std::string file_format)
+ : impl_(new Impl(std::move(ctx), std::move(file_format))) {}
+
+SkyhookFileFormat::~SkyhookFileFormat() = default;
+
+arrow::Status SkyhookFileFormat::Init() { return impl_->Init(); }
+
+arrow::Result<std::shared_ptr<arrow::Schema>> SkyhookFileFormat::Inspect(
+ const arrow::dataset::FileSource& source) const {
+ return impl_->Inspect(source);
+}
+
+arrow::Result<arrow::dataset::ScanTaskIterator> SkyhookFileFormat::ScanFile(
+ const std::shared_ptr<arrow::dataset::ScanOptions>& options,
+ const std::shared_ptr<arrow::dataset::FileFragment>& file) const {
+ return impl_->ScanFile(options, file);
+}
+
+std::shared_ptr<arrow::dataset::FileWriteOptions>
+SkyhookFileFormat::DefaultWriteOptions() {
+ return nullptr;
+}
+
+arrow::Result<std::shared_ptr<arrow::dataset::FileWriter>> SkyhookFileFormat::MakeWriter(
+ std::shared_ptr<arrow::io::OutputStream> destination,
+ std::shared_ptr<arrow::Schema> schema,
+ std::shared_ptr<arrow::dataset::FileWriteOptions> options,
+ arrow::fs::FileLocator destination_locator) const {
+ return arrow::Status::NotImplemented("Skyhook writer not yet implemented.");
+}
+
+} // namespace skyhook
diff --git a/client/file_skyhook.h b/client/file_skyhook.h
new file mode 100644
index 0000000..52a19f5
--- /dev/null
+++ b/client/file_skyhook.h
@@ -0,0 +1,108 @@
+// 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.
+#pragma once
+
+#include "arrow/api.h"
+#include "arrow/dataset/file_parquet.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/dataset/type_fwd.h"
+#include "arrow/dataset/visibility.h"
+
+namespace skyhook {
+
+/// \addtogroup dataset-file-formats
+///
+/// @{
+
+/// \struct RadosConnCtx
+/// \brief A struct to hold the parameters required
+/// for connecting to a RADOS cluster.
+struct RadosConnCtx {
+ std::string ceph_config_path;
+ std::string ceph_data_pool;
+ std::string ceph_user_name;
+ std::string ceph_cluster_name;
+ std::string ceph_cls_name;
+
+ RadosConnCtx(std::string ceph_config_path, std::string ceph_data_pool,
+ std::string ceph_user_name, std::string ceph_cluster_name,
+ std::string ceph_cls_name)
+ : ceph_config_path(std::move(ceph_config_path)),
+ ceph_data_pool(std::move(ceph_data_pool)),
+ ceph_user_name(std::move(ceph_user_name)),
+ ceph_cluster_name(std::move(ceph_cluster_name)),
+ ceph_cls_name(std::move(ceph_cls_name)) {}
+};
+
+/// \class SkyhookFileFormat
+/// \brief A FileFormat implementation that offloads fragment
+/// scan operations to the Ceph OSDs. For more details, see the
+/// Skyhook paper, https://arxiv.org/pdf/2105.09894.pdf.
+class SkyhookFileFormat : public arrow::dataset::FileFormat {
+ public:
+ static arrow::Result<std::shared_ptr<SkyhookFileFormat>> Make(
+ std::shared_ptr<RadosConnCtx> ctx, std::string file_format);
+ SkyhookFileFormat(std::shared_ptr<RadosConnCtx> ctx, std::string file_format);
+
+ ~SkyhookFileFormat() override;
+
+ std::string type_name() const override { return "skyhook"; }
+
+ bool Equals(const arrow::dataset::FileFormat& other) const override {
+ return type_name() == other.type_name();
+ }
+
+ arrow::Result<bool> IsSupported(
+ const arrow::dataset::FileSource& source) const override {
+ return true;
+ }
+
+ /// \brief Return the schema of the file fragment.
+ /// \param[in] source The source of the file fragment.
+ /// \return The schema of the file fragment.
+ arrow::Result<std::shared_ptr<arrow::Schema>> Inspect(
+ const arrow::dataset::FileSource& source) const override;
+
+ /// \brief Scan a file fragment.
+ /// \param[in] options The ScanOptions to use.
+ /// \param[in] file The file fragment to scan.
+ /// \return An iterator of ScanTasks.
+ arrow::Result<arrow::dataset::ScanTaskIterator> ScanFile(
+ const std::shared_ptr<arrow::dataset::ScanOptions>& options,
+ const std::shared_ptr<arrow::dataset::FileFragment>& file) const override;
+
+ /// \brief Create a writer for this format.
+ arrow::Result<std::shared_ptr<arrow::dataset::FileWriter>> MakeWriter(
+ std::shared_ptr<arrow::io::OutputStream> destination,
+ std::shared_ptr<arrow::Schema> schema,
+ std::shared_ptr<arrow::dataset::FileWriteOptions> options,
+ arrow::fs::FileLocator destination_locator) const override;
+
+ /// \brief Get default write options for this format.
+ std::shared_ptr<arrow::dataset::FileWriteOptions> DefaultWriteOptions() override;
+
+ private:
+ class Impl;
+ std::unique_ptr<Impl> impl_;
+
+ /// \brief Initialize the SkyhookFileFormat by connecting to RADOS.
+ arrow::Status Init();
+};
+
+/// @}
+
+} // namespace skyhook
diff --git a/cls/cls_skyhook.cc b/cls/cls_skyhook.cc
new file mode 100644
index 0000000..5f50dd0
--- /dev/null
+++ b/cls/cls_skyhook.cc
@@ -0,0 +1,267 @@
+// 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 <rados/objclass.h>
+
+#include <memory>
+
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/dataset.h"
+#include "arrow/dataset/file_ipc.h"
+#include "arrow/dataset/file_parquet.h"
+#include "arrow/io/interfaces.h"
+#include "arrow/result.h"
+#include "arrow/util/logging.h"
+
+#include "skyhook/protocol/skyhook_protocol.h"
+
+CLS_VER(1, 0)
+CLS_NAME(skyhook)
+
+cls_handle_t h_class;
+cls_method_handle_t h_scan_op;
+
+/// \brief Log skyhook errors using RADOS object class SDK's logger.
+void LogSkyhookError(const std::string& msg) { CLS_LOG(0, "error: %s", msg.c_str()); }
+
+/// \class RandomAccessObject
+/// \brief An interface to provide a file-like view over RADOS objects.
+class RandomAccessObject : public arrow::io::RandomAccessFile {
+ public:
+ explicit RandomAccessObject(cls_method_context_t hctx, int64_t file_size) {
+ hctx_ = hctx;
+ content_length_ = file_size;
+ chunks_ = std::vector<std::shared_ptr<ceph::bufferlist>>();
+ }
+
+ ~RandomAccessObject() override { DCHECK_OK(Close()); }
+
+ /// Check if the file stream is closed.
+ arrow::Status CheckClosed() const {
+ if (closed_) {
+ return arrow::Status::Invalid("Operation on closed stream");
+ }
+ return arrow::Status::OK();
+ }
+
+ /// Check if the position of the object is valid.
+ arrow::Status CheckPosition(int64_t position, const char* action) const {
+ if (position < 0) {
+ return arrow::Status::Invalid("Cannot ", action, " from negative position");
+ }
+ if (position > content_length_) {
+ return arrow::Status::IOError("Cannot ", action, " past end of file");
+ }
+ return arrow::Status::OK();
+ }
+
+ arrow::Result<int64_t> ReadAt(int64_t position, int64_t nbytes, void* out) override {
+ return arrow::Status::NotImplemented(
+ "ReadAt has not been implemented in RandomAccessObject");
+ }
+
+ /// Read a specified number of bytes from a specified position.
+ arrow::Result<std::shared_ptr<arrow::Buffer>> ReadAt(int64_t position,
+ int64_t nbytes) override {
+ RETURN_NOT_OK(CheckClosed());
+ RETURN_NOT_OK(CheckPosition(position, "read"));
+
+ // No need to allocate more than the remaining number of bytes
+ nbytes = std::min(nbytes, content_length_ - position);
+
+ if (nbytes > 0) {
+ std::shared_ptr<ceph::bufferlist> bl = std::make_shared<ceph::bufferlist>();
+ cls_cxx_read(hctx_, position, nbytes, bl.get());
+ chunks_.push_back(bl);
+ return std::make_shared<arrow::Buffer>((uint8_t*)bl->c_str(), bl->length());
+ }
+ return std::make_shared<arrow::Buffer>("");
+ }
+
+ /// Read a specified number of bytes from the current position.
+ arrow::Result<std::shared_ptr<arrow::Buffer>> Read(int64_t nbytes) override {
+ ARROW_ASSIGN_OR_RAISE(auto buffer, ReadAt(pos_, nbytes));
+ pos_ += buffer->size();
+ return std::move(buffer);
+ }
+
+ /// Read a specified number of bytes from the current position into an output stream.
+ arrow::Result<int64_t> Read(int64_t nbytes, void* out) override {
+ ARROW_ASSIGN_OR_RAISE(int64_t bytes_read, ReadAt(pos_, nbytes, out));
+ pos_ += bytes_read;
+ return bytes_read;
+ }
+
+ /// Return the size of the file.
+ arrow::Result<int64_t> GetSize() override {
+ RETURN_NOT_OK(CheckClosed());
+ return content_length_;
+ }
+
+ /// Sets the file-pointer offset, measured from the beginning of the
+ /// file, at which the next read or write occurs.
+ arrow::Status Seek(int64_t position) override {
+ RETURN_NOT_OK(CheckClosed());
+ RETURN_NOT_OK(CheckPosition(position, "seek"));
+
+ pos_ = position;
+ return arrow::Status::OK();
+ }
+
+ /// Returns the file-pointer offset.
+ arrow::Result<int64_t> Tell() const override {
+ RETURN_NOT_OK(CheckClosed());
+ return pos_;
+ }
+
+ /// Mark the file as closed.
+ arrow::Status Close() override {
+ closed_ = true;
+ return arrow::Status::OK();
+ }
+
+ bool closed() const override { return closed_; }
+
+ private:
+ cls_method_context_t hctx_;
+ bool closed_ = false;
+ int64_t pos_ = 0;
+ int64_t content_length_ = -1;
+ std::vector<std::shared_ptr<ceph::bufferlist>> chunks_;
+};
+
+/// \brief Driver function to execute the Scan operations.
+/// \param[in] hctx RADOS object context.
+/// \param[in] req The scan request received from the client.
+/// \param[in] format The file format instance to use in the scan.
+/// \param[in] fragment_scan_options The fragment scan options to use to customize the
+/// scan.
+/// \return Table.
+arrow::Result<std::shared_ptr<arrow::Table>> DoScan(
+ cls_method_context_t hctx, const skyhook::ScanRequest& req,
+ const std::shared_ptr<arrow::dataset::FileFormat>& format,
+ const std::shared_ptr<arrow::dataset::FragmentScanOptions>& fragment_scan_options) {
+ auto file = std::make_shared<RandomAccessObject>(hctx, req.file_size);
+ arrow::dataset::FileSource source(file);
+ ARROW_ASSIGN_OR_RAISE(
+ auto fragment, format->MakeFragment(std::move(source), req.partition_expression));
+ auto options = std::make_shared<arrow::dataset::ScanOptions>();
+ auto builder = std::make_shared<arrow::dataset::ScannerBuilder>(
+ req.dataset_schema, std::move(fragment), std::move(options));
+
+ ARROW_RETURN_NOT_OK(builder->Filter(req.filter_expression));
+ ARROW_RETURN_NOT_OK(builder->Project(req.projection_schema->field_names()));
+ ARROW_RETURN_NOT_OK(builder->UseThreads(true));
+ ARROW_RETURN_NOT_OK(builder->FragmentScanOptions(fragment_scan_options));
+
+ ARROW_ASSIGN_OR_RAISE(auto scanner, builder->Finish());
+ ARROW_ASSIGN_OR_RAISE(auto table, scanner->ToTable());
+ return table;
+}
+
+/// \brief Scan RADOS objects containing Arrow IPC data.
+/// \param[in] hctx The RADOS object context.
+/// \param[in] req The scan request received from the client.
+/// \return Table.
+static arrow::Result<std::shared_ptr<arrow::Table>> ScanIpcObject(
+ cls_method_context_t hctx, skyhook::ScanRequest req) {
+ auto format = std::make_shared<arrow::dataset::IpcFileFormat>();
+ auto fragment_scan_options = std::make_shared<arrow::dataset::IpcFragmentScanOptions>();
+
+ ARROW_ASSIGN_OR_RAISE(auto result_table, DoScan(hctx, req, std::move(format),
+ std::move(fragment_scan_options)));
+ return result_table;
+}
+
+/// \brief Scan RADOS objects containing Parquet binary data.
+/// \param[in] hctx The RADOS object context.
+/// \param[in] req The scan request received from the client.
+/// \return Table.
+static arrow::Result<std::shared_ptr<arrow::Table>> ScanParquetObject(
+ cls_method_context_t hctx, skyhook::ScanRequest req) {
+ auto format = std::make_shared<arrow::dataset::ParquetFileFormat>();
+ auto fragment_scan_options =
+ std::make_shared<arrow::dataset::ParquetFragmentScanOptions>();
+
+ ARROW_ASSIGN_OR_RAISE(auto result_table, DoScan(hctx, req, std::move(format),
+ std::move(fragment_scan_options)));
+ return result_table;
+}
+
+/// \brief The scan operation to execute on the Ceph OSD nodes. The scan request is
+/// deserialized, the object is scanned, and the resulting table is serialized
+/// and sent back to the client.
+/// \param[in] hctx The RADOS object context.
+/// \param[in] in A bufferlist containing serialized Scan request.
+/// \param[out] out A bufferlist to store the serialized resultant table.
+/// \return Exit code.
+static int scan_op(cls_method_context_t hctx, ceph::bufferlist* in,
+ ceph::bufferlist* out) {
+ // Components required to construct a File fragment.
+ arrow::Status s;
+ skyhook::ScanRequest req;
+
+ // Deserialize the scan request.
+ if (!(s = skyhook::DeserializeScanRequest(*in, &req)).ok()) {
+ LogSkyhookError(s.message());
+ return SCAN_REQ_DESER_ERR_CODE;
+ }
+
+ // Scan the object.
+ std::shared_ptr<arrow::Table> table;
+ arrow::Result<std::shared_ptr<arrow::Table>> maybe_table;
+ switch (req.file_format) {
+ case skyhook::SkyhookFileType::type::PARQUET:
+ maybe_table = ScanParquetObject(hctx, std::move(req));
+ if (!maybe_table.ok()) {
+ LogSkyhookError("Could not scan parquet object: " +
+ maybe_table.status().ToString());
+ return SCAN_ERR_CODE;
+ }
+ table = *maybe_table;
+ break;
+ case skyhook::SkyhookFileType::type::IPC:
+ maybe_table = ScanIpcObject(hctx, std::move(req));
+ if (!maybe_table.ok()) {
+ LogSkyhookError("Could not scan IPC object: " + maybe_table.status().ToString());
+ return SCAN_ERR_CODE;
+ }
+ table = *maybe_table;
+ break;
+ default:
+ table = nullptr;
+ }
+ if (!table) {
+ LogSkyhookError("Unsupported file format");
+ return SCAN_ERR_CODE;
+ }
+
+ // Serialize the resultant table to send back to the client.
+ ceph::bufferlist bl;
+ if (!(s = skyhook::SerializeTable(table, &bl)).ok()) {
+ LogSkyhookError(s.message());
+ return SCAN_RES_SER_ERR_CODE;
+ }
+
+ *out = std::move(bl);
+ return 0;
+}
+
+void __cls_init() {
+ /// Register the skyhook object classes with the OSD.
+ cls_register("skyhook", &h_class);
+ cls_register_cxx_method(h_class, "scan_op", CLS_METHOD_RD, scan_op, &h_scan_op);
+}
diff --git a/cls/cls_skyhook_test.cc b/cls/cls_skyhook_test.cc
new file mode 100644
index 0000000..461cdd6
--- /dev/null
+++ b/cls/cls_skyhook_test.cc
@@ -0,0 +1,207 @@
+// 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 "skyhook/client/file_skyhook.h"
+
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/dataset.h"
+#include "arrow/dataset/file_base.h"
+#include "arrow/filesystem/api.h"
+#include "arrow/io/api.h"
+#include "arrow/testing/gtest_util.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/iterator.h"
+#include "gtest/gtest.h"
+
+#include "parquet/arrow/reader.h"
+#include "parquet/arrow/writer.h"
+
+std::shared_ptr<skyhook::SkyhookFileFormat> GetSkyhookFormat() {
+ // The constants below should match the parameters with
+ // which the Ceph cluster is configured in integration_skyhook.sh.
+ // Currently, all the default values have been used.
+ std::string ceph_config_path = "/etc/ceph/ceph.conf";
+ std::string ceph_data_pool = "cephfs_data";
+ std::string ceph_user_name = "client.admin";
+ std::string ceph_cluster_name = "ceph";
+ std::string ceph_cls_name = "skyhook";
+ std::shared_ptr<skyhook::RadosConnCtx> rados_ctx =
+ std::make_shared<skyhook::RadosConnCtx>(ceph_config_path, ceph_data_pool,
+ ceph_user_name, ceph_cluster_name,
+ ceph_cls_name);
+ EXPECT_OK_AND_ASSIGN(auto format,
+ skyhook::SkyhookFileFormat::Make(rados_ctx, "parquet"));
+ return format;
+}
+
+std::shared_ptr<arrow::dataset::ParquetFileFormat> GetParquetFormat() {
+ return std::make_shared<arrow::dataset::ParquetFileFormat>();
+}
+
+std::shared_ptr<arrow::dataset::Dataset> GetDatasetFromDirectory(
+ std::shared_ptr<arrow::fs::FileSystem> fs,
+ std::shared_ptr<arrow::dataset::FileFormat> format, std::string dir) {
+ arrow::fs::FileSelector s;
+ s.base_dir = std::move(dir);
+ s.recursive = true;
+
+ arrow::dataset::FileSystemFactoryOptions options;
+ options.partitioning = std::make_shared<arrow::dataset::HivePartitioning>(
+ arrow::schema({arrow::field("payment_type", arrow::int32()),
+ arrow::field("VendorID", arrow::int32())}));
+ EXPECT_OK_AND_ASSIGN(auto factory, arrow::dataset::FileSystemDatasetFactory::Make(
+ std::move(fs), s, std::move(format), options));
+
+ arrow::dataset::InspectOptions inspect_options;
+ arrow::dataset::FinishOptions finish_options;
+ EXPECT_OK_AND_ASSIGN(auto schema, factory->Inspect(inspect_options));
+ EXPECT_OK_AND_ASSIGN(auto dataset, factory->Finish(finish_options));
+ return dataset;
+}
+
+std::shared_ptr<arrow::fs::FileSystem> GetFileSystemFromUri(const std::string& uri,
+ std::string* path) {
+ EXPECT_OK_AND_ASSIGN(auto fs, arrow::fs::FileSystemFromUri(uri, path));
+ return fs;
+}
+
+std::shared_ptr<arrow::dataset::Dataset> GetDatasetFromPath(
+ std::shared_ptr<arrow::fs::FileSystem> fs,
+ std::shared_ptr<arrow::dataset::FileFormat> format, std::string path) {
+ EXPECT_OK_AND_ASSIGN(auto info, fs->GetFileInfo(path));
+ return GetDatasetFromDirectory(std::move(fs), std::move(format), std::move(path));
+}
+
+std::shared_ptr<arrow::dataset::Scanner> GetScannerFromDataset(
+ const std::shared_ptr<arrow::dataset::Dataset>& dataset,
+ std::vector<std::string> columns, arrow::compute::Expression filter,
+ bool use_threads) {
+ EXPECT_OK_AND_ASSIGN(auto scanner_builder, dataset->NewScan());
+
+ if (!columns.empty()) {
+ ARROW_EXPECT_OK(scanner_builder->Project(std::move(columns)));
+ }
+
+ ARROW_EXPECT_OK(scanner_builder->Filter(std::move(filter)));
+ ARROW_EXPECT_OK(scanner_builder->UseThreads(use_threads));
+ EXPECT_OK_AND_ASSIGN(auto scanner, scanner_builder->Finish());
+ return scanner;
+}
+
+TEST(TestSkyhookCLS, SelectEntireDataset) {
+ std::string path;
+ auto fs = GetFileSystemFromUri("file:///mnt/cephfs/nyc", &path);
+ std::vector<std::string> columns;
+
+ auto parquet_format = GetParquetFormat();
+ auto dataset = GetDatasetFromPath(fs, parquet_format, path);
+ auto scanner =
+ GetScannerFromDataset(dataset, columns, arrow::compute::literal(true), true);
+ EXPECT_OK_AND_ASSIGN(auto table_parquet, scanner->ToTable());
+
+ auto skyhook_format = GetSkyhookFormat();
+ dataset = GetDatasetFromPath(fs, skyhook_format, path);
+ scanner = GetScannerFromDataset(dataset, columns, arrow::compute::literal(true), true);
+ EXPECT_OK_AND_ASSIGN(auto table_skyhook_parquet, scanner->ToTable());
+
+ ASSERT_EQ(table_parquet->Equals(*table_skyhook_parquet), 1);
+ ASSERT_EQ(table_parquet->num_rows(), table_skyhook_parquet->num_rows());
+}
+
+TEST(TestSkyhookCLS, SelectFewRows) {
+ std::string path;
+ auto fs = GetFileSystemFromUri("file:///mnt/cephfs/nyc", &path);
+ std::vector<std::string> columns;
+ auto filter = arrow::compute::greater(arrow::compute::field_ref("payment_type"),
+ arrow::compute::literal(2));
+ auto parquet_format = GetParquetFormat();
+ auto dataset = GetDatasetFromPath(fs, parquet_format, path);
+ auto scanner = GetScannerFromDataset(dataset, columns, filter, true);
+ EXPECT_OK_AND_ASSIGN(auto table_parquet, scanner->ToTable());
+
+ auto skyhook_format = GetSkyhookFormat();
+ dataset = GetDatasetFromPath(fs, skyhook_format, path);
+ scanner = GetScannerFromDataset(dataset, columns, filter, true);
+ EXPECT_OK_AND_ASSIGN(auto table_skyhook_parquet, scanner->ToTable());
+
+ ASSERT_EQ(table_parquet->Equals(*table_skyhook_parquet), 1);
+ ASSERT_EQ(table_parquet->num_rows(), table_skyhook_parquet->num_rows());
+}
+
+TEST(TestSkyhookCLS, SelectFewColumns) {
+ std::string path;
+ auto fs = GetFileSystemFromUri("file:///mnt/cephfs/nyc", &path);
+ std::vector<std::string> columns = {"fare_amount", "total_amount"};
+
+ auto parquet_format = GetParquetFormat();
+ auto dataset = GetDatasetFromPath(fs, parquet_format, path);
+ auto scanner =
+ GetScannerFromDataset(dataset, columns, arrow::compute::literal(true), true);
+ EXPECT_OK_AND_ASSIGN(auto table_parquet, scanner->ToTable());
+
+ auto skyhook_format = GetSkyhookFormat();
+ dataset = GetDatasetFromPath(fs, skyhook_format, path);
+ scanner = GetScannerFromDataset(dataset, columns, arrow::compute::literal(true), true);
+ EXPECT_OK_AND_ASSIGN(auto table_skyhook_parquet, scanner->ToTable());
+
+ ASSERT_EQ(table_parquet->Equals(*table_skyhook_parquet), 1);
+ ASSERT_EQ(table_parquet->num_rows(), table_skyhook_parquet->num_rows());
+}
+
+TEST(TestSkyhookCLS, SelectRowsAndColumnsOnPartitionKey) {
+ std::string path;
+ auto fs = GetFileSystemFromUri("file:///mnt/cephfs/nyc", &path);
+ std::vector<std::string> columns = {"fare_amount", "VendorID", "payment_type"};
+ auto filter = arrow::compute::greater(arrow::compute::field_ref("payment_type"),
+ arrow::compute::literal(2));
+
+ auto parquet_format = GetParquetFormat();
+ auto dataset = GetDatasetFromPath(fs, parquet_format, path);
+ auto scanner = GetScannerFromDataset(dataset, columns, filter, true);
+ EXPECT_OK_AND_ASSIGN(auto table_parquet, scanner->ToTable());
+
+ auto skyhook_format = GetSkyhookFormat();
+ dataset = GetDatasetFromPath(fs, skyhook_format, path);
+ scanner = GetScannerFromDataset(dataset, columns, filter, true);
+ EXPECT_OK_AND_ASSIGN(auto table_skyhook_parquet, scanner->ToTable());
+
+ ASSERT_EQ(table_parquet->Equals(*table_skyhook_parquet), 1);
+ ASSERT_EQ(table_parquet->num_rows(), table_skyhook_parquet->num_rows());
+}
+
+TEST(TestSkyhookCLS, SelectRowsAndColumnsOnlyOnPartitionKey) {
+ std::string path;
+ auto fs = GetFileSystemFromUri("file:///mnt/cephfs/nyc", &path);
+ std::vector<std::string> columns = {"total_amount", "VendorID", "payment_type"};
+ auto filter = arrow::compute::and_(
+ arrow::compute::greater(arrow::compute::field_ref("payment_type"),
+ arrow::compute::literal(2)),
+ arrow::compute::greater(arrow::compute::field_ref("VendorID"),
+ arrow::compute::literal(1)));
+
+ auto parquet_format = GetParquetFormat();
+ auto dataset = GetDatasetFromPath(fs, parquet_format, path);
+ auto scanner = GetScannerFromDataset(dataset, columns, filter, true);
+ EXPECT_OK_AND_ASSIGN(auto table_parquet, scanner->ToTable());
+
+ auto skyhook_format = GetSkyhookFormat();
+ dataset = GetDatasetFromPath(fs, skyhook_format, path);
+ scanner = GetScannerFromDataset(dataset, columns, filter, true);
+ EXPECT_OK_AND_ASSIGN(auto table_skyhook_parquet, scanner->ToTable());
+
+ ASSERT_EQ(table_parquet->Equals(*table_skyhook_parquet), 1);
+ ASSERT_EQ(table_parquet->num_rows(), table_skyhook_parquet->num_rows());
+}
diff --git a/protocol/ScanRequest.fbs b/protocol/ScanRequest.fbs
new file mode 100644
index 0000000..870d603
--- /dev/null
+++ b/protocol/ScanRequest.fbs
@@ -0,0 +1,29 @@
+// 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.
+
+namespace org.apache.arrow.flatbuf;
+
+table ScanRequest {
+ file_size: long;
+ file_format: short;
+ filter: [ubyte];
+ partition: [ubyte];
+ dataset_schema: [ubyte];
+ projection_schema: [ubyte];
+}
+
+root_type ScanRequest;
diff --git a/protocol/ScanRequest_generated.h b/protocol/ScanRequest_generated.h
new file mode 100644
index 0000000..884857a
--- /dev/null
+++ b/protocol/ScanRequest_generated.h
@@ -0,0 +1,167 @@
+// automatically generated by the FlatBuffers compiler, do not modify
+
+
+#ifndef FLATBUFFERS_GENERATED_SCANREQUEST_ORG_APACHE_ARROW_FLATBUF_H_
+#define FLATBUFFERS_GENERATED_SCANREQUEST_ORG_APACHE_ARROW_FLATBUF_H_
+
+#include "flatbuffers/flatbuffers.h"
+
+namespace org {
+namespace apache {
+namespace arrow {
+namespace flatbuf {
+
+struct ScanRequest;
+
+struct ScanRequest FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
+ enum FlatBuffersVTableOffset FLATBUFFERS_VTABLE_UNDERLYING_TYPE {
+ VT_FILE_SIZE = 4,
+ VT_FILE_FORMAT = 6,
+ VT_FILTER = 8,
+ VT_PARTITION = 10,
+ VT_DATASET_SCHEMA = 12,
+ VT_PROJECTION_SCHEMA = 14
+ };
+ int64_t file_size() const {
+ return GetField<int64_t>(VT_FILE_SIZE, 0);
+ }
+ int16_t file_format() const {
+ return GetField<int16_t>(VT_FILE_FORMAT, 0);
+ }
+ const flatbuffers::Vector<uint8_t> *filter() const {
+ return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_FILTER);
+ }
+ const flatbuffers::Vector<uint8_t> *partition() const {
+ return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_PARTITION);
+ }
+ const flatbuffers::Vector<uint8_t> *dataset_schema() const {
+ return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_DATASET_SCHEMA);
+ }
+ const flatbuffers::Vector<uint8_t> *projection_schema() const {
+ return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_PROJECTION_SCHEMA);
+ }
+ bool Verify(flatbuffers::Verifier &verifier) const {
+ return VerifyTableStart(verifier) &&
+ VerifyField<int64_t>(verifier, VT_FILE_SIZE) &&
+ VerifyField<int16_t>(verifier, VT_FILE_FORMAT) &&
+ VerifyOffset(verifier, VT_FILTER) &&
+ verifier.VerifyVector(filter()) &&
+ VerifyOffset(verifier, VT_PARTITION) &&
+ verifier.VerifyVector(partition()) &&
+ VerifyOffset(verifier, VT_DATASET_SCHEMA) &&
+ verifier.VerifyVector(dataset_schema()) &&
+ VerifyOffset(verifier, VT_PROJECTION_SCHEMA) &&
+ verifier.VerifyVector(projection_schema()) &&
+ verifier.EndTable();
+ }
+};
+
+struct ScanRequestBuilder {
+ flatbuffers::FlatBufferBuilder &fbb_;
+ flatbuffers::uoffset_t start_;
+ void add_file_size(int64_t file_size) {
+ fbb_.AddElement<int64_t>(ScanRequest::VT_FILE_SIZE, file_size, 0);
+ }
+ void add_file_format(int16_t file_format) {
+ fbb_.AddElement<int16_t>(ScanRequest::VT_FILE_FORMAT, file_format, 0);
+ }
+ void add_filter(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> filter) {
+ fbb_.AddOffset(ScanRequest::VT_FILTER, filter);
+ }
+ void add_partition(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> partition) {
+ fbb_.AddOffset(ScanRequest::VT_PARTITION, partition);
+ }
+ void add_dataset_schema(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> dataset_schema) {
+ fbb_.AddOffset(ScanRequest::VT_DATASET_SCHEMA, dataset_schema);
+ }
+ void add_projection_schema(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> projection_schema) {
+ fbb_.AddOffset(ScanRequest::VT_PROJECTION_SCHEMA, projection_schema);
+ }
+ explicit ScanRequestBuilder(flatbuffers::FlatBufferBuilder &_fbb)
+ : fbb_(_fbb) {
+ start_ = fbb_.StartTable();
+ }
+ ScanRequestBuilder &operator=(const ScanRequestBuilder &);
+ flatbuffers::Offset<ScanRequest> Finish() {
+ const auto end = fbb_.EndTable(start_);
+ auto o = flatbuffers::Offset<ScanRequest>(end);
+ return o;
+ }
+};
+
+inline flatbuffers::Offset<ScanRequest> CreateScanRequest(
+ flatbuffers::FlatBufferBuilder &_fbb,
+ int64_t file_size = 0,
+ int16_t file_format = 0,
+ flatbuffers::Offset<flatbuffers::Vector<uint8_t>> filter = 0,
+ flatbuffers::Offset<flatbuffers::Vector<uint8_t>> partition = 0,
+ flatbuffers::Offset<flatbuffers::Vector<uint8_t>> dataset_schema = 0,
+ flatbuffers::Offset<flatbuffers::Vector<uint8_t>> projection_schema = 0) {
+ ScanRequestBuilder builder_(_fbb);
+ builder_.add_file_size(file_size);
+ builder_.add_projection_schema(projection_schema);
+ builder_.add_dataset_schema(dataset_schema);
+ builder_.add_partition(partition);
+ builder_.add_filter(filter);
+ builder_.add_file_format(file_format);
+ return builder_.Finish();
+}
+
+inline flatbuffers::Offset<ScanRequest> CreateScanRequestDirect(
+ flatbuffers::FlatBufferBuilder &_fbb,
+ int64_t file_size = 0,
+ int16_t file_format = 0,
+ const std::vector<uint8_t> *filter = nullptr,
+ const std::vector<uint8_t> *partition = nullptr,
+ const std::vector<uint8_t> *dataset_schema = nullptr,
+ const std::vector<uint8_t> *projection_schema = nullptr) {
+ auto filter__ = filter ? _fbb.CreateVector<uint8_t>(*filter) : 0;
+ auto partition__ = partition ? _fbb.CreateVector<uint8_t>(*partition) : 0;
+ auto dataset_schema__ = dataset_schema ? _fbb.CreateVector<uint8_t>(*dataset_schema) : 0;
+ auto projection_schema__ = projection_schema ? _fbb.CreateVector<uint8_t>(*projection_schema) : 0;
+ return org::apache::arrow::flatbuf::CreateScanRequest(
+ _fbb,
+ file_size,
+ file_format,
+ filter__,
+ partition__,
+ dataset_schema__,
+ projection_schema__);
+}
+
+inline const org::apache::arrow::flatbuf::ScanRequest *GetScanRequest(const void *buf) {
+ return flatbuffers::GetRoot<org::apache::arrow::flatbuf::ScanRequest>(buf);
+}
+
+inline const org::apache::arrow::flatbuf::ScanRequest *GetSizePrefixedScanRequest(const void *buf) {
+ return flatbuffers::GetSizePrefixedRoot<org::apache::arrow::flatbuf::ScanRequest>(buf);
+}
+
+inline bool VerifyScanRequestBuffer(
+ flatbuffers::Verifier &verifier) {
+ return verifier.VerifyBuffer<org::apache::arrow::flatbuf::ScanRequest>(nullptr);
+}
+
+inline bool VerifySizePrefixedScanRequestBuffer(
+ flatbuffers::Verifier &verifier) {
+ return verifier.VerifySizePrefixedBuffer<org::apache::arrow::flatbuf::ScanRequest>(nullptr);
+}
+
+inline void FinishScanRequestBuffer(
+ flatbuffers::FlatBufferBuilder &fbb,
+ flatbuffers::Offset<org::apache::arrow::flatbuf::ScanRequest> root) {
+ fbb.Finish(root);
+}
+
+inline void FinishSizePrefixedScanRequestBuffer(
+ flatbuffers::FlatBufferBuilder &fbb,
+ flatbuffers::Offset<org::apache::arrow::flatbuf::ScanRequest> root) {
+ fbb.FinishSizePrefixed(root);
+}
+
+} // namespace flatbuf
+} // namespace arrow
+} // namespace apache
+} // namespace org
+
+#endif // FLATBUFFERS_GENERATED_SCANREQUEST_ORG_APACHE_ARROW_FLATBUF_H_
diff --git a/protocol/rados_protocol.cc b/protocol/rados_protocol.cc
new file mode 100644
index 0000000..cb1acec
--- /dev/null
+++ b/protocol/rados_protocol.cc
@@ -0,0 +1,99 @@
+// 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 "skyhook/protocol/rados_protocol.h"
+
+#include "arrow/util/io_util.h"
+
+#include <iostream>
+#include <vector>
+
+namespace skyhook {
+namespace rados {
+
+template <typename... Args>
+arrow::Status GetStatusFromReturnCode(int code, Args&&... args) {
+ if (code)
+ return arrow::internal::StatusFromErrno(code, arrow::StatusCode::Invalid,
+ std::forward<Args>(args)...);
+ return arrow::Status::OK();
+}
+
+arrow::Status IoCtxInterface::read(const std::string& oid, ceph::bufferlist& bl,
+ size_t len, uint64_t offset) {
+ return GetStatusFromReturnCode(ioCtx->read(oid, bl, len, offset),
+ "ioctx->read failed.");
+}
+
+arrow::Status IoCtxInterface::exec(const std::string& oid, const char* cls,
+ const char* method, ceph::bufferlist& in,
+ ceph::bufferlist& out) {
+ return GetStatusFromReturnCode(ioCtx->exec(oid, cls, method, in, out),
+ "ioctx->exec failed.");
+}
+
+arrow::Status IoCtxInterface::stat(const std::string& oid, uint64_t* psize) {
+ return GetStatusFromReturnCode(ioCtx->stat(oid, psize, NULL), "ioctx->stat failed.");
+}
+
+arrow::Status RadosInterface::init2(const char* const name, const char* const clustername,
+ uint64_t flags) {
+ return GetStatusFromReturnCode(cluster->init2(name, clustername, flags),
+ "rados->init failed.");
+}
+
+arrow::Status RadosInterface::ioctx_create(const char* name, IoCtxInterface* pioctx) {
+ librados::IoCtx ioCtx;
+ int ret = cluster->ioctx_create(name, ioCtx);
+ pioctx->setIoCtx(&ioCtx);
+ return GetStatusFromReturnCode(ret, "rados->ioctx_create failed.");
+}
+
+arrow::Status RadosInterface::conf_read_file(const char* const path) {
+ return GetStatusFromReturnCode(cluster->conf_read_file(path),
+ "rados->conf_read_file failed.");
+}
+
+arrow::Status RadosInterface::connect() {
+ return GetStatusFromReturnCode(cluster->connect(), "rados->connect failed.");
+}
+
+void RadosInterface::shutdown() { cluster->shutdown(); }
+
+RadosConn::~RadosConn() { Shutdown(); }
+
+arrow::Status RadosConn::Connect() {
+ if (connected) {
+ return arrow::Status::OK();
+ }
+
+ ARROW_RETURN_NOT_OK(
+ rados->init2(ctx->ceph_user_name.c_str(), ctx->ceph_cluster_name.c_str(), 0));
+ ARROW_RETURN_NOT_OK(rados->conf_read_file(ctx->ceph_config_path.c_str()));
+ ARROW_RETURN_NOT_OK(rados->connect());
+ ARROW_RETURN_NOT_OK(rados->ioctx_create(ctx->ceph_data_pool.c_str(), io_ctx.get()));
+ return arrow::Status::OK();
+}
+
+void RadosConn::Shutdown() {
+ if (connected) {
+ rados->shutdown();
+ connected = false;
+ }
+}
+
+} // namespace rados
+} // namespace skyhook
diff --git a/protocol/rados_protocol.h b/protocol/rados_protocol.h
new file mode 100644
index 0000000..3e5fac7
--- /dev/null
+++ b/protocol/rados_protocol.h
@@ -0,0 +1,103 @@
+// 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.
+#pragma once
+
+#include <rados/librados.hpp>
+
+#include "arrow/status.h"
+#include "arrow/util/make_unique.h"
+
+#include "skyhook/client/file_skyhook.h"
+
+namespace skyhook {
+namespace rados {
+
+class IoCtxInterface {
+ public:
+ IoCtxInterface() { ioCtx = arrow::internal::make_unique<librados::IoCtx>(); }
+ /// \brief Read from a RADOS object.
+ ///
+ /// \param[in] oid the ID of the object to read.
+ /// \param[in] bl a bufferlist to hold the contents of the read object.
+ /// \param[in] len the length of data to read from the object.
+ /// \param[in] offset the offset to read from in the object.
+ arrow::Status read(const std::string& oid, ceph::bufferlist& bl, size_t len,
+ uint64_t offset);
+ /// \brief Executes a Ceph Object Class method.
+ ///
+ /// \param[in] oid the object ID on which to invoke the CLS function.
+ /// \param[in] cls the name of the object class.
+ /// \param[in] method the name of the object class method.
+ /// \param[in] in a bufferlist to send data to the object class method.
+ /// \param[in] out a bufferlist to recieve data from the object class method.
+ arrow::Status exec(const std::string& oid, const char* cls, const char* method,
+ ceph::bufferlist& in, ceph::bufferlist& out);
+ /// \brief Execute POSIX stat on a RADOS object.
+ ///
+ /// \param[in] oid the object ID on which to call stat.
+ /// \param[out] psize hold the size of the object.
+ arrow::Status stat(const std::string& oid, uint64_t* psize);
+ /// \brief Set the `librados::IoCtx` instance inside a IoCtxInterface instance.
+ void setIoCtx(librados::IoCtx* ioCtx_) { *ioCtx = *ioCtx_; }
+
+ private:
+ std::unique_ptr<librados::IoCtx> ioCtx;
+};
+
+class RadosInterface {
+ public:
+ RadosInterface() { cluster = arrow::internal::make_unique<librados::Rados>(); }
+ /// Initializes a cluster handle.
+ arrow::Status init2(const char* const name, const char* const clustername,
+ uint64_t flags);
+ /// Create an I/O context
+ arrow::Status ioctx_create(const char* name, IoCtxInterface* pioctx);
+ /// Read the Ceph config file.
+ arrow::Status conf_read_file(const char* const path);
+ /// Connect to the Ceph cluster.
+ arrow::Status connect();
+ /// Close connection to the Ceph cluster.
+ void shutdown();
+
+ private:
+ std::unique_ptr<librados::Rados> cluster;
+};
+
+/// Connect to a Ceph cluster and hold the connection
+/// information for use in later stages.
+class RadosConn {
+ public:
+ explicit RadosConn(std::shared_ptr<skyhook::RadosConnCtx> ctx)
+ : ctx(std::move(ctx)),
+ rados(arrow::internal::make_unique<RadosInterface>()),
+ io_ctx(arrow::internal::make_unique<IoCtxInterface>()),
+ connected(false) {}
+ ~RadosConn();
+ /// Connect to the Ceph cluster.
+ arrow::Status Connect();
+ /// Shutdown the connection to the Ceph
+ /// cluster if already connected.
+ void Shutdown();
+
+ std::shared_ptr<skyhook::RadosConnCtx> ctx;
+ std::unique_ptr<RadosInterface> rados;
+ std::unique_ptr<IoCtxInterface> io_ctx;
+ bool connected;
+};
+
+} // namespace rados
+} // namespace skyhook
diff --git a/protocol/skyhook_protocol.cc b/protocol/skyhook_protocol.cc
new file mode 100644
index 0000000..c261048
--- /dev/null
+++ b/protocol/skyhook_protocol.cc
@@ -0,0 +1,136 @@
+// 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 "skyhook/protocol/skyhook_protocol.h"
+
+#include <flatbuffers/flatbuffers.h>
+
+#include "ScanRequest_generated.h"
+#include "arrow/io/api.h"
+#include "arrow/ipc/api.h"
+#include "arrow/result.h"
+#include "arrow/util/io_util.h"
+
+namespace skyhook {
+
+namespace flatbuf = org::apache::arrow::flatbuf;
+
+arrow::Status SerializeScanRequest(ScanRequest& req, ceph::bufferlist* bl) {
+ ARROW_ASSIGN_OR_RAISE(auto filter_expression,
+ arrow::compute::Serialize(req.filter_expression));
+ ARROW_ASSIGN_OR_RAISE(auto partition_expression,
+ arrow::compute::Serialize(req.partition_expression));
+ ARROW_ASSIGN_OR_RAISE(auto projection_schema,
+ arrow::ipc::SerializeSchema(*req.projection_schema));
+ ARROW_ASSIGN_OR_RAISE(auto dataset_schema,
+ arrow::ipc::SerializeSchema(*req.dataset_schema));
+
+ flatbuffers::FlatBufferBuilder builder(1024);
+ auto filter_expression_vector =
+ builder.CreateVector(filter_expression->data(), filter_expression->size());
+ auto partition_expression_vector =
+ builder.CreateVector(partition_expression->data(), partition_expression->size());
+ auto projected_schema_vector =
+ builder.CreateVector(projection_schema->data(), projection_schema->size());
+ auto dataset_schema_vector =
+ builder.CreateVector(dataset_schema->data(), dataset_schema->size());
+
+ auto request = flatbuf::CreateScanRequest(
+ builder, req.file_size, static_cast<int>(req.file_format), filter_expression_vector,
+ partition_expression_vector, dataset_schema_vector, projected_schema_vector);
+ builder.Finish(request);
+ uint8_t* buf = builder.GetBufferPointer();
+ int size = builder.GetSize();
+
+ bl->append(reinterpret_cast<const char*>(buf), size);
+ return arrow::Status::OK();
+}
+
+arrow::Status DeserializeScanRequest(ceph::bufferlist& bl, ScanRequest* req) {
+ auto request = flatbuf::GetScanRequest((uint8_t*)bl.c_str());
+
+ ARROW_ASSIGN_OR_RAISE(auto filter_expression,
+ arrow::compute::Deserialize(std::make_shared<arrow::Buffer>(
+ request->filter()->data(), request->filter()->size())));
+ req->filter_expression = filter_expression;
+
+ ARROW_ASSIGN_OR_RAISE(auto partition_expression,
+ arrow::compute::Deserialize(std::make_shared<arrow::Buffer>(
+ request->partition()->data(), request->partition()->size())));
+ req->partition_expression = partition_expression;
+
+ arrow::ipc::DictionaryMemo empty_memo;
+ arrow::io::BufferReader projection_schema_reader(request->projection_schema()->data(),
+ request->projection_schema()->size());
+ arrow::io::BufferReader dataset_schema_reader(request->dataset_schema()->data(),
+ request->dataset_schema()->size());
+
+ ARROW_ASSIGN_OR_RAISE(req->projection_schema,
+ arrow::ipc::ReadSchema(&projection_schema_reader, &empty_memo));
+ ARROW_ASSIGN_OR_RAISE(req->dataset_schema,
+ arrow::ipc::ReadSchema(&dataset_schema_reader, &empty_memo));
+
+ req->file_size = request->file_size();
+ req->file_format = (SkyhookFileType::type)request->file_format();
+ return arrow::Status::OK();
+}
+
+arrow::Status SerializeTable(const std::shared_ptr<arrow::Table>& table,
+ ceph::bufferlist* bl) {
+ ARROW_ASSIGN_OR_RAISE(auto buffer_output_stream,
+ arrow::io::BufferOutputStream::Create());
+
+ auto options = arrow::ipc::IpcWriteOptions::Defaults();
+ auto codec = arrow::Compression::LZ4_FRAME;
+
+ ARROW_ASSIGN_OR_RAISE(options.codec, arrow::util::Codec::Create(codec));
+ ARROW_ASSIGN_OR_RAISE(auto writer, arrow::ipc::MakeStreamWriter(
+ buffer_output_stream, table->schema(), options));
+
+ ARROW_RETURN_NOT_OK(writer->WriteTable(*table));
+ ARROW_RETURN_NOT_OK(writer->Close());
+
+ ARROW_ASSIGN_OR_RAISE(auto buffer, buffer_output_stream->Finish());
+ bl->append(reinterpret_cast<const char*>(buffer->data()), buffer->size());
+ return arrow::Status::OK();
+}
+
+arrow::Status DeserializeTable(ceph::bufferlist& bl, bool use_threads,
+ arrow::RecordBatchVector* batches) {
+ auto buffer = std::make_shared<arrow::Buffer>((uint8_t*)bl.c_str(), bl.length());
+ auto buffer_reader = std::make_shared<arrow::io::BufferReader>(buffer);
+ auto options = arrow::ipc::IpcReadOptions::Defaults();
+ options.use_threads = use_threads;
+ ARROW_ASSIGN_OR_RAISE(
+ auto reader, arrow::ipc::RecordBatchStreamReader::Open(buffer_reader, options));
+ ARROW_RETURN_NOT_OK(reader->ReadAll(batches));
+ return arrow::Status::OK();
+}
+
+arrow::Status ExecuteObjectClassFn(const std::shared_ptr<rados::RadosConn>& connection,
+ const std::string& oid, const std::string& fn,
+ ceph::bufferlist& in, ceph::bufferlist& out) {
+ int e = arrow::internal::ErrnoFromStatus(connection->io_ctx->exec(
+ oid.c_str(), connection->ctx->ceph_cls_name.c_str(), fn.c_str(), in, out));
+
+ if (e == SCAN_ERR_CODE) return arrow::Status::Invalid(SCAN_ERR_MSG);
+ if (e == SCAN_REQ_DESER_ERR_CODE) return arrow::Status::Invalid(SCAN_REQ_DESER_ERR_MSG);
+ if (e == SCAN_RES_SER_ERR_CODE) return arrow::Status::Invalid(SCAN_RES_SER_ERR_MSG);
+ if (e != 0) return arrow::Status::Invalid(SCAN_UNKNOWN_ERR_MSG);
+ return arrow::Status::OK();
+}
+
+} // namespace skyhook
diff --git a/protocol/skyhook_protocol.h b/protocol/skyhook_protocol.h
new file mode 100644
index 0000000..b4f6d6e
--- /dev/null
+++ b/protocol/skyhook_protocol.h
@@ -0,0 +1,116 @@
+// 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.
+#pragma once
+
+#include "skyhook/protocol/rados_protocol.h"
+
+#include <sys/stat.h>
+#include <sstream>
+
+#include "arrow/compute/exec/expression.h"
+#include "arrow/record_batch.h"
+#include "arrow/table.h"
+#include "arrow/type.h"
+
+#define SCAN_UNKNOWN_ERR_MSG "something went wrong while scanning file fragment"
+#define SCAN_ERR_CODE 25
+#define SCAN_ERR_MSG "failed to scan file fragment"
+#define SCAN_REQ_DESER_ERR_CODE 26
+#define SCAN_REQ_DESER_ERR_MSG "failed to deserialize scan request"
+#define SCAN_RES_SER_ERR_CODE 27
+#define SCAN_RES_SER_ERR_MSG "failed to serialize result table"
+
+namespace skyhook {
+
+/// An enum to represent the different
+/// types of file formats that Skyhook supports.
+struct SkyhookFileType {
+ enum type { PARQUET, IPC };
+};
+
+/// A struct encapsulating all the parameters
+/// required to be serialized in the form of flatbuffers for
+/// sending to the cls.
+struct ScanRequest {
+ arrow::compute::Expression filter_expression;
+ arrow::compute::Expression partition_expression;
+ std::shared_ptr<arrow::Schema> projection_schema;
+ std::shared_ptr<arrow::Schema> dataset_schema;
+ int64_t file_size;
+ SkyhookFileType::type file_format;
+};
+
+/// Utility functions to serialize and deserialize scan requests and result Arrow tables.
+arrow::Status SerializeScanRequest(ScanRequest& req, ceph::bufferlist* bl);
+arrow::Status DeserializeScanRequest(ceph::bufferlist& bl, ScanRequest* req);
+arrow::Status SerializeTable(const std::shared_ptr<arrow::Table>& table,
+ ceph::bufferlist* bl);
+arrow::Status DeserializeTable(ceph::bufferlist& bl, bool use_threads,
+ arrow::RecordBatchVector* batches);
+
+/// Utility function to invoke a RADOS object class function on an RADOS object.
+arrow::Status ExecuteObjectClassFn(const std::shared_ptr<rados::RadosConn>& connection,
+ const std::string& oid, const std::string& fn,
+ ceph::bufferlist& in, ceph::bufferlist& out);
+
+/// An interface for translating the name of a file in CephFS to its
+/// corresponding object ID in RADOS assuming 1:1 mapping between a file
+/// and it's underlying object.
+class SkyhookDirectObjectAccess {
+ public:
+ explicit SkyhookDirectObjectAccess(std::shared_ptr<rados::RadosConn> connection)
+ : connection_(std::move(connection)) {}
+
+ ~SkyhookDirectObjectAccess() = default;
+
+ /// Execute a POSIX stat on a file.
+ arrow::Status Stat(const std::string& path, struct stat& st) {
+ struct stat file_st;
+ if (stat(path.c_str(), &file_st) < 0)
+ return arrow::Status::Invalid("stat returned non-zero exit code.");
+ st = file_st;
+ return arrow::Status::OK();
+ }
+
+ /// Convert a file inode to RADOS object ID.
+ std::string ConvertInodeToOID(uint64_t inode) {
+ std::stringstream ss;
+ /// In Ceph, the underlying stripes that make up a file are
+ /// named in the format [hex(inode)].[8-bit-binary(stripe_index)].
+ ss << std::hex << inode;
+
+ /// Since in Skyhook, we ensure a single stripe per file,
+ /// we can assume the stripe index to be always 0 and hence
+ /// hardcode it's 8-bit binary form.
+ std::string oid(ss.str() + ".00000000");
+ return oid;
+ }
+
+ /// Execute an object class method. It uses the `librados::exec` api to
+ /// perform object clsass method calls on the storage node and
+ /// stores the result in an output bufferlist.
+ arrow::Status Exec(uint64_t inode, const std::string& fn, ceph::bufferlist& in,
+ ceph::bufferlist& out) {
+ std::string oid = ConvertInodeToOID(inode);
+ return ExecuteObjectClassFn(connection_, oid, fn, in, out);
+ }
+
+ private:
+ std::shared_ptr<rados::RadosConn> connection_;
+};
+
+} // namespace skyhook
diff --git a/protocol/skyhook_protocol_test.cc b/protocol/skyhook_protocol_test.cc
new file mode 100644
index 0000000..1d3af3e
--- /dev/null
+++ b/protocol/skyhook_protocol_test.cc
@@ -0,0 +1,71 @@
+// 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 "skyhook/protocol/skyhook_protocol.h"
+
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/test_util.h"
+#include "arrow/table.h"
+#include "arrow/testing/gtest_util.h"
+
+std::shared_ptr<arrow::Table> CreateTable() {
+ auto schema = arrow::schema({
+ {arrow::field("a", arrow::uint8())},
+ {arrow::field("b", arrow::uint32())},
+ });
+
+ std::shared_ptr<arrow::Table> table;
+ return TableFromJSON(schema, {R"([{"a": null, "b": 5},
+ {"a": 1, "b": 3},
+ {"a": 3, "b": null},
+ {"a": null, "b": null},
+ {"a": 2, "b": 5},
+ {"a": 1, "b": 5}
+ ])"});
+}
+
+TEST(TestSkyhookProtocol, SerDeserScanRequest) {
+ ceph::bufferlist bl;
+ skyhook::ScanRequest req;
+ req.filter_expression = arrow::compute::literal(true);
+ req.partition_expression = arrow::compute::literal(false);
+ req.projection_schema = arrow::schema({arrow::field("a", arrow::int64())});
+ req.dataset_schema = arrow::schema({arrow::field("a", arrow::int64())});
+ req.file_size = 1000000;
+ req.file_format = skyhook::SkyhookFileType::type::IPC;
+ ASSERT_OK(skyhook::SerializeScanRequest(req, &bl));
+
+ skyhook::ScanRequest req_;
+ ASSERT_OK(skyhook::DeserializeScanRequest(bl, &req_));
+ ASSERT_TRUE(req.filter_expression.Equals(req_.filter_expression));
+ ASSERT_TRUE(req.partition_expression.Equals(req_.partition_expression));
+ ASSERT_TRUE(req.projection_schema->Equals(req_.projection_schema));
+ ASSERT_TRUE(req.dataset_schema->Equals(req_.dataset_schema));
+ ASSERT_EQ(req.file_size, req_.file_size);
+ ASSERT_EQ(req.file_format, req_.file_format);
+}
+
+TEST(TestSkyhookProtocol, SerDeserTable) {
+ std::shared_ptr<arrow::Table> table = CreateTable();
+ ceph::bufferlist bl;
+ ASSERT_OK(skyhook::SerializeTable(table, &bl));
+
+ arrow::RecordBatchVector batches;
+ ASSERT_OK(skyhook::DeserializeTable(bl, false, &batches));
+ ASSERT_OK_AND_ASSIGN(auto materialized_table, arrow::Table::FromRecordBatches(batches));
+
+ ASSERT_TRUE(table->Equals(*materialized_table));
+}
diff --git a/skyhook.pc.in b/skyhook.pc.in
new file mode 100644
index 0000000..a3a4da5
--- /dev/null
+++ b/skyhook.pc.in
@@ -0,0 +1,26 @@
+# 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.
+
+prefix=@CMAKE_INSTALL_PREFIX@
+libdir=${prefix}/@CMAKE_INSTALL_LIBDIR@
+includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@
+
+Name: Skyhook
+Description: Skyhook is a plugin for offloading computations into Ceph.
+Version: @SKYHOOK_VERSION@
+Requires: arrow_dataset
+Libs: -L${libdir} -larrow_skyhook_client
\ No newline at end of file
(arrow-skyhook) 10/18: GH-37485: [C++][Skyhook] Don't use deprecated BufferReader API (#37486)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit d70c79897f3bab4be060bc53197f174f3d1d77ec
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Fri Sep 1 00:09:53 2023 +0900
GH-37485: [C++][Skyhook] Don't use deprecated BufferReader API (#37486)
### Rationale for this change
GH-37360 deprecated `BufferReader(const uint8_t*, int64_t)`.
### What changes are included in this PR?
Use `BufferReader(std::shared_ptr<Buffer>)` instead.
### Are these changes tested?
Yes.
### Are there any user-facing changes?
No.
* Closes: #37485
Authored-by: Sutou Kouhei <ko...@clear-code.com>
Signed-off-by: Antoine Pitrou <an...@python.org>
---
protocol/skyhook_protocol.cc | 10 ++++++----
1 file changed, 6 insertions(+), 4 deletions(-)
diff --git a/protocol/skyhook_protocol.cc b/protocol/skyhook_protocol.cc
index f83b124..3b1234c 100644
--- a/protocol/skyhook_protocol.cc
+++ b/protocol/skyhook_protocol.cc
@@ -73,10 +73,12 @@ arrow::Status DeserializeScanRequest(ceph::bufferlist& bl, ScanRequest* req) {
req->partition_expression = partition_expression;
arrow::ipc::DictionaryMemo empty_memo;
- arrow::io::BufferReader projection_schema_reader(request->projection_schema()->data(),
- request->projection_schema()->size());
- arrow::io::BufferReader dataset_schema_reader(request->dataset_schema()->data(),
- request->dataset_schema()->size());
+ auto projection_schema_buffer = std::make_shared<arrow::Buffer>(
+ request->projection_schema()->data(), request->projection_schema()->size());
+ arrow::io::BufferReader projection_schema_reader(std::move(projection_schema_buffer));
+ auto dataset_schema_buffer = std::make_shared<arrow::Buffer>(
+ request->dataset_schema()->data(), request->dataset_schema()->size());
+ arrow::io::BufferReader dataset_schema_reader(std::move(dataset_schema_buffer));
ARROW_ASSIGN_OR_RAISE(req->projection_schema,
arrow::ipc::ReadSchema(&projection_schema_reader, &empty_memo));
(arrow-skyhook) 06/18: ARROW-17814: [C++] Remove make_unique reimplementation (#14204)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit a01198810b25b08b2567d47b4dd2b6893717e264
Author: Gang Wu <us...@gmail.com>
AuthorDate: Fri Sep 23 00:48:52 2022 +0800
ARROW-17814: [C++] Remove make_unique reimplementation (#14204)
Use std::make_unique instead.
Authored-by: Gang Wu <us...@gmail.com>
Signed-off-by: Antoine Pitrou <an...@python.org>
---
protocol/rados_protocol.h | 11 ++++++-----
1 file changed, 6 insertions(+), 5 deletions(-)
diff --git a/protocol/rados_protocol.h b/protocol/rados_protocol.h
index 3e5fac7..6a0b2ea 100644
--- a/protocol/rados_protocol.h
+++ b/protocol/rados_protocol.h
@@ -19,16 +19,17 @@
#include <rados/librados.hpp>
#include "arrow/status.h"
-#include "arrow/util/make_unique.h"
#include "skyhook/client/file_skyhook.h"
+#include <memory>
+
namespace skyhook {
namespace rados {
class IoCtxInterface {
public:
- IoCtxInterface() { ioCtx = arrow::internal::make_unique<librados::IoCtx>(); }
+ IoCtxInterface() { ioCtx = std::make_unique<librados::IoCtx>(); }
/// \brief Read from a RADOS object.
///
/// \param[in] oid the ID of the object to read.
@@ -60,7 +61,7 @@ class IoCtxInterface {
class RadosInterface {
public:
- RadosInterface() { cluster = arrow::internal::make_unique<librados::Rados>(); }
+ RadosInterface() { cluster = std::make_unique<librados::Rados>(); }
/// Initializes a cluster handle.
arrow::Status init2(const char* const name, const char* const clustername,
uint64_t flags);
@@ -83,8 +84,8 @@ class RadosConn {
public:
explicit RadosConn(std::shared_ptr<skyhook::RadosConnCtx> ctx)
: ctx(std::move(ctx)),
- rados(arrow::internal::make_unique<RadosInterface>()),
- io_ctx(arrow::internal::make_unique<IoCtxInterface>()),
+ rados(std::make_unique<RadosInterface>()),
+ io_ctx(std::make_unique<IoCtxInterface>()),
connected(false) {}
~RadosConn();
/// Connect to the Ceph cluster.
(arrow-skyhook) 13/18: GH-39823: [C++] Allow building cpp/src/arrow/**/*.cc without waiting bundled libraries (#39824)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit f72fc468569d3dd3d2066ec884c7fc759ad86699
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Thu Feb 29 09:44:15 2024 +0900
GH-39823: [C++] Allow building cpp/src/arrow/**/*.cc without waiting bundled libraries (#39824)
### Rationale for this change
If we can build most of `cpp/src/arrow/**/*.cc` before all bundled libraries are built, we can reduce build time.
### What changes are included in this PR?
* Remove the `toolchain` internal CMake target
* Remove `ARROW_SHARED_LINK_LIBS`
* Remove `ARROW_STATIC_LINK_LIBS`
* Move the following variables to `cpp/src/arrow/CMakeLists.txt`
* `ARROW_SHARED_PRIVATE_LINK_LIBS`
* `ARROW_SHARED_INSTALL_INTERFACE_LIBS`
* `ARROW_STATIC_INSTALL_INTERFACE_LIBS`
* `ARROW_TEST_LINK_TOOLCHAIN`
* `ARROW_TEST_SHARED_LINK_LIBS`
* `ARROW_TEST_STATIC_LINK_LIBS`
* `ARROW_SYSTEM_LINK_LIBS`
* Add internal `OBJECT` libraries that have minimal dependencies
* Remove unused `cpp/src/arrow/util/benchmark_main.cc`
### Are these changes tested?
Yes.
### Are there any user-facing changes?
No.
* Closes: #39823
* GitHub Issue: #39823
Authored-by: Sutou Kouhei <ko...@clear-code.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
CMakeLists.txt | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 67fddec..1b41e34 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -43,8 +43,11 @@ add_arrow_lib(arrow_skyhook
ARROW_SKYHOOK_CLIENT_LIBRARIES
SHARED_LINK_LIBS
${ARROW_SKYHOOK_LINK_SHARED}
+ SHARED_PRIVATE_LINK_LIBS
+ arrow::flatbuffers
STATIC_LINK_LIBS
- ${ARROW_SKYHOOK_LINK_STATIC})
+ ${ARROW_SKYHOOK_LINK_STATIC}
+ arrow::flatbuffers)
# define the cls library
add_arrow_lib(cls_skyhook
@@ -54,6 +57,8 @@ add_arrow_lib(cls_skyhook
ARROW_SKYHOOK_CLS_LIBRARIES
SHARED_LINK_LIBS
${ARROW_SKYHOOK_LINK_SHARED}
+ SHARED_PRIVATE_LINK_LIBS
+ arrow::flatbuffers
STATIC_LINK_LIBS
${ARROW_SKYHOOK_LINK_STATIC})
(arrow-skyhook) 14/18: GH-40882: [C++] Suppress shorten-64-to-32 warnings in CUDA/Skyhook codes (#40883)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 73af16c8fd3993d56e7aa219e742d4b256e76dde
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Tue Apr 2 06:17:03 2024 +0900
GH-40882: [C++] Suppress shorten-64-to-32 warnings in CUDA/Skyhook codes (#40883)
### Rationale for this change
```text
cpp/src/arrow/gpu/cuda_memory.cc:497:72: error: implicit conversion loses integer precision: 'int64_t' (aka 'long') to 'int' [-Werror,-Wshorten-64-to-32]
ARROW_ASSIGN_OR_RAISE(auto device, arrow::cuda::CudaDevice::Make(device_id));
~~~~~ ^~~~~~~~~
```
```text
cpp/src/arrow/gpu/cuda_memory.cc:508:68: error: implicit conversion loses integer precision: 'int64_t' (aka 'long') to 'int' [-Werror,-Wshorten-64-to-32]
ARROW_ASSIGN_OR_RAISE(auto device, arrow::cuda::CudaDevice::Make(device_id));
~~~~~ ^~~~~~~~~
```
```text
cpp/src/skyhook/protocol/skyhook_protocol.cc:109:69: error: implicit conversion loses integer precision: 'int64_t' (aka 'long') to 'unsigned int' [-Werror,-Wshorten-64-to-32]
bl->append(reinterpret_cast<const char*>(buffer->data()), buffer->size());
~~~~~~ ~~~~~~~~^~~~~~
```
```text
cpp/src/skyhook/cls/cls_skyhook.cc:87:37: error: implicit conversion loses integer precision: 'int64_t' (aka 'long') to 'int' [-Werror,-Wshorten-64-to-32]
cls_cxx_read(hctx_, position, nbytes, bl.get());
~~~~~~~~~~~~ ^~~~~~
cpp/src/skyhook/cls/cls_skyhook.cc:87:27: error: implicit conversion loses integer precision: 'int64_t' (aka 'long') to 'int' [-Werror,-Wshorten-64-to-32]
cls_cxx_read(hctx_, position, nbytes, bl.get());
~~~~~~~~~~~~ ^~~~~~~~
```
```text
cpp/src/skyhook/protocol/skyhook_protocol.cc:109:69: error: implicit conversion loses integer precision: 'int64_t' (aka 'long') to 'unsigned int' [-Werror,-Wshorten-64-to-32]
bl->append(reinterpret_cast<const char*>(buffer->data()), buffer->size());
~~~~~~
```
### What changes are included in this PR?
Add casts explicitly.
### Are these changes tested?
Yes.
### Are there any user-facing changes?
No.
* GitHub Issue: #40882
Authored-by: Sutou Kouhei <ko...@clear-code.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
cls/cls_skyhook.cc | 2 +-
protocol/skyhook_protocol.cc | 3 ++-
2 files changed, 3 insertions(+), 2 deletions(-)
diff --git a/cls/cls_skyhook.cc b/cls/cls_skyhook.cc
index 24f80c7..e021cb3 100644
--- a/cls/cls_skyhook.cc
+++ b/cls/cls_skyhook.cc
@@ -84,7 +84,7 @@ class RandomAccessObject : public arrow::io::RandomAccessFile {
if (nbytes > 0) {
std::shared_ptr<ceph::bufferlist> bl = std::make_shared<ceph::bufferlist>();
- cls_cxx_read(hctx_, position, nbytes, bl.get());
+ cls_cxx_read(hctx_, static_cast<int>(position), static_cast<int>(nbytes), bl.get());
chunks_.push_back(bl);
return std::make_shared<arrow::Buffer>((uint8_t*)bl->c_str(), bl->length());
}
diff --git a/protocol/skyhook_protocol.cc b/protocol/skyhook_protocol.cc
index 3b1234c..b91a9bf 100644
--- a/protocol/skyhook_protocol.cc
+++ b/protocol/skyhook_protocol.cc
@@ -106,7 +106,8 @@ arrow::Status SerializeTable(const std::shared_ptr<arrow::Table>& table,
ARROW_RETURN_NOT_OK(writer->Close());
ARROW_ASSIGN_OR_RAISE(auto buffer, buffer_output_stream->Finish());
- bl->append(reinterpret_cast<const char*>(buffer->data()), buffer->size());
+ bl->append(reinterpret_cast<const char*>(buffer->data()),
+ static_cast<unsigned int>(buffer->size()));
return arrow::Status::OK();
}
(arrow-skyhook) 05/18: ARROW-15927: [C++][Skyhook] Add skyhook example (#12620)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 8a4e55beeb6c5823d5530b2a560d895895a1804d
Author: Jayjeet Chakraborty <jc...@rediffmail.com>
AuthorDate: Mon Aug 1 06:24:11 2022 -0700
ARROW-15927: [C++][Skyhook] Add skyhook example (#12620)
Add an example for the `SkyhookFileFormat` extension of Dataset API.
Authored-by: Jayjeet Chakraborty <jc...@rediffmail.com>
Signed-off-by: David Li <li...@gmail.com>
---
CMakeLists.txt | 10 +++++-----
protocol/rados_protocol.cc | 5 ++++-
skyhook.pc.in | 2 +-
3 files changed, 10 insertions(+), 7 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 992c467..0019251 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -19,7 +19,7 @@
add_subdirectory(client)
# define the targets to build
-add_custom_target(arrow_skyhook_client)
+add_custom_target(arrow_skyhook)
add_custom_target(cls_skyhook)
# define the dependencies
@@ -34,7 +34,7 @@ set(ARROW_SKYHOOK_CLS_SOURCES cls/cls_skyhook.cc protocol/rados_protocol.cc
protocol/skyhook_protocol.cc)
# define the client library
-add_arrow_lib(arrow_skyhook_client
+add_arrow_lib(arrow_skyhook
PKG_CONFIG_NAME
skyhook
SOURCES
@@ -58,15 +58,15 @@ add_arrow_lib(cls_skyhook
${ARROW_SKYHOOK_LINK_STATIC})
# finish building the project
-add_dependencies(arrow_skyhook_client ${ARROW_SKYHOOK_CLIENT_LIBRARIES})
+add_dependencies(arrow_skyhook ${ARROW_SKYHOOK_CLIENT_LIBRARIES})
add_dependencies(cls_skyhook ${ARROW_SKYHOOK_CLS_LIBRARIES})
# define the test builds
if(ARROW_TEST_LINKAGE STREQUAL "static")
- set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_skyhook_client_static arrow_dataset_static
+ set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_skyhook_static arrow_dataset_static
${ARROW_TEST_STATIC_LINK_LIBS})
else()
- set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_skyhook_client_shared arrow_dataset_shared
+ set(ARROW_SKYHOOK_TEST_LINK_LIBS arrow_skyhook_shared arrow_dataset_shared
${ARROW_TEST_SHARED_LINK_LIBS})
endif()
diff --git a/protocol/rados_protocol.cc b/protocol/rados_protocol.cc
index cb1acec..eb52078 100644
--- a/protocol/rados_protocol.cc
+++ b/protocol/rados_protocol.cc
@@ -58,7 +58,9 @@ arrow::Status RadosInterface::init2(const char* const name, const char* const cl
arrow::Status RadosInterface::ioctx_create(const char* name, IoCtxInterface* pioctx) {
librados::IoCtx ioCtx;
int ret = cluster->ioctx_create(name, ioCtx);
- pioctx->setIoCtx(&ioCtx);
+ if (!ret) {
+ pioctx->setIoCtx(&ioCtx);
+ }
return GetStatusFromReturnCode(ret, "rados->ioctx_create failed.");
}
@@ -85,6 +87,7 @@ arrow::Status RadosConn::Connect() {
ARROW_RETURN_NOT_OK(rados->conf_read_file(ctx->ceph_config_path.c_str()));
ARROW_RETURN_NOT_OK(rados->connect());
ARROW_RETURN_NOT_OK(rados->ioctx_create(ctx->ceph_data_pool.c_str(), io_ctx.get()));
+ connected = true;
return arrow::Status::OK();
}
diff --git a/skyhook.pc.in b/skyhook.pc.in
index 8f7acfa..5568d63 100644
--- a/skyhook.pc.in
+++ b/skyhook.pc.in
@@ -23,4 +23,4 @@ Name: Skyhook
Description: Skyhook is a plugin for offloading computations into Ceph.
Version: @SKYHOOK_VERSION@
Requires: arrow_dataset
-Libs: -L${libdir} -larrow_skyhook_client
+Libs: -L${libdir} -larrow_skyhook
(arrow-skyhook) 09/18: GH-34630: [C++] Second block of refactoring to move acero out of libarrow (#34575)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit e6145658387a4128da99c5c576d69b111ef4d577
Author: Davide Pasetto <dp...@gmail.com>
AuthorDate: Sat Mar 18 11:22:15 2023 -0400
GH-34630: [C++] Second block of refactoring to move acero out of libarrow (#34575)
### Rationale for this change
This is the second and last block of refactoring before moving acero out
of libarrow. This PR removes the remaining dependencies from libarrow
into the compute/exec directory.
### What changes are included in this PR?
In detail:
* compute/exec/expression is moved inside compute since it is needed in
several kernels
* all compute/exec specific include files are removed from compute/api.h
* light_array_exec_test.cc i(introduced in the previous refactoring) s
merged back into light_array_test
* testing generators are refactored to remove the need for
exec/options.h
* some utilities are moved from compute/exec/util to compute/util
* updated python and r libraries usage of moved files
### Are these changes tested?
Yes using existing tests (some of which have been updated due to
refactoring).
### Are there any user-facing changes?
no
* Issue: #15280
* Closes: #34630
---
client/file_skyhook.cc | 2 +-
cls/cls_skyhook.cc | 2 +-
cls/cls_skyhook_test.cc | 2 +-
protocol/skyhook_protocol.h | 2 +-
protocol/skyhook_protocol_test.cc | 2 +-
5 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/client/file_skyhook.cc b/client/file_skyhook.cc
index 725b04a..5af1b2b 100644
--- a/client/file_skyhook.cc
+++ b/client/file_skyhook.cc
@@ -19,7 +19,7 @@
#include "skyhook/protocol/rados_protocol.h"
#include "skyhook/protocol/skyhook_protocol.h"
-#include "arrow/compute/exec/expression.h"
+#include "arrow/compute/expression.h"
#include "arrow/dataset/file_base.h"
#include "arrow/dataset/file_ipc.h"
#include "arrow/dataset/file_parquet.h"
diff --git a/cls/cls_skyhook.cc b/cls/cls_skyhook.cc
index 5f50dd0..24f80c7 100644
--- a/cls/cls_skyhook.cc
+++ b/cls/cls_skyhook.cc
@@ -18,7 +18,7 @@
#include <memory>
-#include "arrow/compute/exec/expression.h"
+#include "arrow/compute/expression.h"
#include "arrow/dataset/dataset.h"
#include "arrow/dataset/file_ipc.h"
#include "arrow/dataset/file_parquet.h"
diff --git a/cls/cls_skyhook_test.cc b/cls/cls_skyhook_test.cc
index ebb1f4e..ce4250c 100644
--- a/cls/cls_skyhook_test.cc
+++ b/cls/cls_skyhook_test.cc
@@ -16,7 +16,7 @@
// under the License.
#include "skyhook/client/file_skyhook.h"
-#include "arrow/compute/exec/expression.h"
+#include "arrow/compute/expression.h"
#include "arrow/dataset/dataset.h"
#include "arrow/dataset/file_base.h"
#include "arrow/filesystem/api.h"
diff --git a/protocol/skyhook_protocol.h b/protocol/skyhook_protocol.h
index b4f6d6e..37dda8a 100644
--- a/protocol/skyhook_protocol.h
+++ b/protocol/skyhook_protocol.h
@@ -21,7 +21,7 @@
#include <sys/stat.h>
#include <sstream>
-#include "arrow/compute/exec/expression.h"
+#include "arrow/compute/expression.h"
#include "arrow/record_batch.h"
#include "arrow/table.h"
#include "arrow/type.h"
diff --git a/protocol/skyhook_protocol_test.cc b/protocol/skyhook_protocol_test.cc
index 463767a..23c78e2 100644
--- a/protocol/skyhook_protocol_test.cc
+++ b/protocol/skyhook_protocol_test.cc
@@ -16,7 +16,7 @@
// under the License.
#include "skyhook/protocol/skyhook_protocol.h"
-#include "arrow/compute/exec/expression.h"
+#include "arrow/compute/expression.h"
#include "arrow/dataset/test_util_internal.h"
#include "arrow/table.h"
#include "arrow/testing/gtest_util.h"
(arrow-skyhook) 15/18: GH-41478: [C++] Clean up more redundant move warnings (#41487)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit c84b1d1d2bd133fa6846f8323ba270d02e020eba
Author: William Ayd <wi...@icloud.com>
AuthorDate: Fri May 24 12:07:38 2024 -0400
GH-41478: [C++] Clean up more redundant move warnings (#41487)
### Rationale for this change
Minor warning cleanup for downstream libraries trying to get warning-free builds
### What changes are included in this PR?
Removed redundant std::move from return statements
### Are these changes tested?
Builds cleanly
### Are there any user-facing changes?
No
* GitHub Issue: #41478
Authored-by: Will Ayd <wi...@icloud.com>
Signed-off-by: Benjamin Kietzman <be...@gmail.com>
---
cls/cls_skyhook.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/cls/cls_skyhook.cc b/cls/cls_skyhook.cc
index e021cb3..632b82f 100644
--- a/cls/cls_skyhook.cc
+++ b/cls/cls_skyhook.cc
@@ -95,7 +95,7 @@ class RandomAccessObject : public arrow::io::RandomAccessFile {
arrow::Result<std::shared_ptr<arrow::Buffer>> Read(int64_t nbytes) override {
ARROW_ASSIGN_OR_RAISE(auto buffer, ReadAt(pos_, nbytes));
pos_ += buffer->size();
- return std::move(buffer);
+ return buffer;
}
/// Read a specified number of bytes from the current position into an output stream.
(arrow-skyhook) 11/18: GH-38401: [C++] Re-generate flatbuffers C++ for Skyhook (#38405)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 424c863c45d92d421abbe89aa371eb4783f8996f
Author: Felipe Oliveira Carvalho <fe...@gmail.com>
AuthorDate: Mon Oct 23 18:17:24 2023 -0300
GH-38401: [C++] Re-generate flatbuffers C++ for Skyhook (#38405)
### Rationale for this change
Vendored flatbuffer headers have been upgraded so the checked-in flatc-generated files need to be upgraded.
### What changes are included in this PR?
- Changes to the shell script that re-generates flatbuffers
- Check-in of the newly generated `ScanRequest_generated.h`
### Are these changes tested?
By integration tests.
* Closes: #38401
Authored-by: Felipe Oliveira Carvalho <fe...@gmail.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
protocol/ScanRequest_generated.h | 85 ++++++++++++++++++++++------------------
1 file changed, 47 insertions(+), 38 deletions(-)
diff --git a/protocol/ScanRequest_generated.h b/protocol/ScanRequest_generated.h
index 884857a..ecf6ff7 100644
--- a/protocol/ScanRequest_generated.h
+++ b/protocol/ScanRequest_generated.h
@@ -6,14 +6,23 @@
#include "flatbuffers/flatbuffers.h"
+// Ensure the included flatbuffers.h is the same version as when this file was
+// generated, otherwise it may not be compatible.
+static_assert(FLATBUFFERS_VERSION_MAJOR == 23 &&
+ FLATBUFFERS_VERSION_MINOR == 5 &&
+ FLATBUFFERS_VERSION_REVISION == 26,
+ "Non-compatible flatbuffers version included");
+
namespace org {
namespace apache {
namespace arrow {
namespace flatbuf {
struct ScanRequest;
+struct ScanRequestBuilder;
-struct ScanRequest FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
+struct ScanRequest FLATBUFFERS_FINAL_CLASS : private ::flatbuffers::Table {
+ typedef ScanRequestBuilder Builder;
enum FlatBuffersVTableOffset FLATBUFFERS_VTABLE_UNDERLYING_TYPE {
VT_FILE_SIZE = 4,
VT_FILE_FORMAT = 6,
@@ -28,22 +37,22 @@ struct ScanRequest FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
int16_t file_format() const {
return GetField<int16_t>(VT_FILE_FORMAT, 0);
}
- const flatbuffers::Vector<uint8_t> *filter() const {
- return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_FILTER);
+ const ::flatbuffers::Vector<uint8_t> *filter() const {
+ return GetPointer<const ::flatbuffers::Vector<uint8_t> *>(VT_FILTER);
}
- const flatbuffers::Vector<uint8_t> *partition() const {
- return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_PARTITION);
+ const ::flatbuffers::Vector<uint8_t> *partition() const {
+ return GetPointer<const ::flatbuffers::Vector<uint8_t> *>(VT_PARTITION);
}
- const flatbuffers::Vector<uint8_t> *dataset_schema() const {
- return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_DATASET_SCHEMA);
+ const ::flatbuffers::Vector<uint8_t> *dataset_schema() const {
+ return GetPointer<const ::flatbuffers::Vector<uint8_t> *>(VT_DATASET_SCHEMA);
}
- const flatbuffers::Vector<uint8_t> *projection_schema() const {
- return GetPointer<const flatbuffers::Vector<uint8_t> *>(VT_PROJECTION_SCHEMA);
+ const ::flatbuffers::Vector<uint8_t> *projection_schema() const {
+ return GetPointer<const ::flatbuffers::Vector<uint8_t> *>(VT_PROJECTION_SCHEMA);
}
- bool Verify(flatbuffers::Verifier &verifier) const {
+ bool Verify(::flatbuffers::Verifier &verifier) const {
return VerifyTableStart(verifier) &&
- VerifyField<int64_t>(verifier, VT_FILE_SIZE) &&
- VerifyField<int16_t>(verifier, VT_FILE_FORMAT) &&
+ VerifyField<int64_t>(verifier, VT_FILE_SIZE, 8) &&
+ VerifyField<int16_t>(verifier, VT_FILE_FORMAT, 2) &&
VerifyOffset(verifier, VT_FILTER) &&
verifier.VerifyVector(filter()) &&
VerifyOffset(verifier, VT_PARTITION) &&
@@ -57,46 +66,46 @@ struct ScanRequest FLATBUFFERS_FINAL_CLASS : private flatbuffers::Table {
};
struct ScanRequestBuilder {
- flatbuffers::FlatBufferBuilder &fbb_;
- flatbuffers::uoffset_t start_;
+ typedef ScanRequest Table;
+ ::flatbuffers::FlatBufferBuilder &fbb_;
+ ::flatbuffers::uoffset_t start_;
void add_file_size(int64_t file_size) {
fbb_.AddElement<int64_t>(ScanRequest::VT_FILE_SIZE, file_size, 0);
}
void add_file_format(int16_t file_format) {
fbb_.AddElement<int16_t>(ScanRequest::VT_FILE_FORMAT, file_format, 0);
}
- void add_filter(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> filter) {
+ void add_filter(::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> filter) {
fbb_.AddOffset(ScanRequest::VT_FILTER, filter);
}
- void add_partition(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> partition) {
+ void add_partition(::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> partition) {
fbb_.AddOffset(ScanRequest::VT_PARTITION, partition);
}
- void add_dataset_schema(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> dataset_schema) {
+ void add_dataset_schema(::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> dataset_schema) {
fbb_.AddOffset(ScanRequest::VT_DATASET_SCHEMA, dataset_schema);
}
- void add_projection_schema(flatbuffers::Offset<flatbuffers::Vector<uint8_t>> projection_schema) {
+ void add_projection_schema(::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> projection_schema) {
fbb_.AddOffset(ScanRequest::VT_PROJECTION_SCHEMA, projection_schema);
}
- explicit ScanRequestBuilder(flatbuffers::FlatBufferBuilder &_fbb)
+ explicit ScanRequestBuilder(::flatbuffers::FlatBufferBuilder &_fbb)
: fbb_(_fbb) {
start_ = fbb_.StartTable();
}
- ScanRequestBuilder &operator=(const ScanRequestBuilder &);
- flatbuffers::Offset<ScanRequest> Finish() {
+ ::flatbuffers::Offset<ScanRequest> Finish() {
const auto end = fbb_.EndTable(start_);
- auto o = flatbuffers::Offset<ScanRequest>(end);
+ auto o = ::flatbuffers::Offset<ScanRequest>(end);
return o;
}
};
-inline flatbuffers::Offset<ScanRequest> CreateScanRequest(
- flatbuffers::FlatBufferBuilder &_fbb,
+inline ::flatbuffers::Offset<ScanRequest> CreateScanRequest(
+ ::flatbuffers::FlatBufferBuilder &_fbb,
int64_t file_size = 0,
int16_t file_format = 0,
- flatbuffers::Offset<flatbuffers::Vector<uint8_t>> filter = 0,
- flatbuffers::Offset<flatbuffers::Vector<uint8_t>> partition = 0,
- flatbuffers::Offset<flatbuffers::Vector<uint8_t>> dataset_schema = 0,
- flatbuffers::Offset<flatbuffers::Vector<uint8_t>> projection_schema = 0) {
+ ::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> filter = 0,
+ ::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> partition = 0,
+ ::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> dataset_schema = 0,
+ ::flatbuffers::Offset<::flatbuffers::Vector<uint8_t>> projection_schema = 0) {
ScanRequestBuilder builder_(_fbb);
builder_.add_file_size(file_size);
builder_.add_projection_schema(projection_schema);
@@ -107,8 +116,8 @@ inline flatbuffers::Offset<ScanRequest> CreateScanRequest(
return builder_.Finish();
}
-inline flatbuffers::Offset<ScanRequest> CreateScanRequestDirect(
- flatbuffers::FlatBufferBuilder &_fbb,
+inline ::flatbuffers::Offset<ScanRequest> CreateScanRequestDirect(
+ ::flatbuffers::FlatBufferBuilder &_fbb,
int64_t file_size = 0,
int16_t file_format = 0,
const std::vector<uint8_t> *filter = nullptr,
@@ -130,32 +139,32 @@ inline flatbuffers::Offset<ScanRequest> CreateScanRequestDirect(
}
inline const org::apache::arrow::flatbuf::ScanRequest *GetScanRequest(const void *buf) {
- return flatbuffers::GetRoot<org::apache::arrow::flatbuf::ScanRequest>(buf);
+ return ::flatbuffers::GetRoot<org::apache::arrow::flatbuf::ScanRequest>(buf);
}
inline const org::apache::arrow::flatbuf::ScanRequest *GetSizePrefixedScanRequest(const void *buf) {
- return flatbuffers::GetSizePrefixedRoot<org::apache::arrow::flatbuf::ScanRequest>(buf);
+ return ::flatbuffers::GetSizePrefixedRoot<org::apache::arrow::flatbuf::ScanRequest>(buf);
}
inline bool VerifyScanRequestBuffer(
- flatbuffers::Verifier &verifier) {
+ ::flatbuffers::Verifier &verifier) {
return verifier.VerifyBuffer<org::apache::arrow::flatbuf::ScanRequest>(nullptr);
}
inline bool VerifySizePrefixedScanRequestBuffer(
- flatbuffers::Verifier &verifier) {
+ ::flatbuffers::Verifier &verifier) {
return verifier.VerifySizePrefixedBuffer<org::apache::arrow::flatbuf::ScanRequest>(nullptr);
}
inline void FinishScanRequestBuffer(
- flatbuffers::FlatBufferBuilder &fbb,
- flatbuffers::Offset<org::apache::arrow::flatbuf::ScanRequest> root) {
+ ::flatbuffers::FlatBufferBuilder &fbb,
+ ::flatbuffers::Offset<org::apache::arrow::flatbuf::ScanRequest> root) {
fbb.Finish(root);
}
inline void FinishSizePrefixedScanRequestBuffer(
- flatbuffers::FlatBufferBuilder &fbb,
- flatbuffers::Offset<org::apache::arrow::flatbuf::ScanRequest> root) {
+ ::flatbuffers::FlatBufferBuilder &fbb,
+ ::flatbuffers::Offset<org::apache::arrow::flatbuf::ScanRequest> root) {
fbb.FinishSizePrefixed(root);
}
(arrow-skyhook) 16/18: GH-45537: [CI][C++] Add missing includes (iwyu) to file_skyhook.cc (#45538)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 65328d0ee6404f62e04d4d64acf0f767e675547d
Author: Raúl Cumplido <ra...@gmail.com>
AuthorDate: Sun Feb 16 01:31:11 2025 +0100
GH-45537: [CI][C++] Add missing includes (iwyu) to file_skyhook.cc (#45538)
### Rationale for this change
The job is failing because it was using some includes from a different header file `cpp/src/arrow/acero/options.h`. The file removed some of those includes and now they were missing.
### What changes are included in this PR?
Add missing includes
### Are these changes tested?
Yes, via archery
### Are there any user-facing changes?
No
* GitHub Issue: #45537
Authored-by: Raúl Cumplido <ra...@gmail.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
client/file_skyhook.cc | 2 ++
1 file changed, 2 insertions(+)
diff --git a/client/file_skyhook.cc b/client/file_skyhook.cc
index 5af1b2b..18f0b0e 100644
--- a/client/file_skyhook.cc
+++ b/client/file_skyhook.cc
@@ -23,6 +23,8 @@
#include "arrow/dataset/file_base.h"
#include "arrow/dataset/file_ipc.h"
#include "arrow/dataset/file_parquet.h"
+#include "arrow/record_batch.h"
+#include "arrow/util/async_generator.h"
#include "arrow/util/checked_cast.h"
#include "arrow/util/compression.h"
(arrow-skyhook) 08/18: GH-34269: [C++] Fix include file name (#34285)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 8747d6eaee266665b776ef5b0da795a6ec7db79f
Author: Abe Tomoaki <ab...@enzou.tokyo>
AuthorDate: Wed Feb 22 22:28:15 2023 +0900
GH-34269: [C++] Fix include file name (#34285)
### Rationale for this change
I got a "arrow/dataset/test_util.h: No such file or directory" error when building, so I changed it to the correct file name.
### What changes are included in this PR?
Changed include file name.
### Are these changes tested?
I checked that I was able to build because I had an error in the build.
### Are there any user-facing changes?
* Closes: #34269
Authored-by: abetomo <ab...@enzou.tokyo>
Signed-off-by: Weston Pace <we...@gmail.com>
---
protocol/skyhook_protocol_test.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/protocol/skyhook_protocol_test.cc b/protocol/skyhook_protocol_test.cc
index 1d3af3e..463767a 100644
--- a/protocol/skyhook_protocol_test.cc
+++ b/protocol/skyhook_protocol_test.cc
@@ -17,7 +17,7 @@
#include "skyhook/protocol/skyhook_protocol.h"
#include "arrow/compute/exec/expression.h"
-#include "arrow/dataset/test_util.h"
+#include "arrow/dataset/test_util_internal.h"
#include "arrow/table.h"
#include "arrow/testing/gtest_util.h"
(arrow-skyhook) 12/18: GH-38979: [C++] Fix spelling (#38980)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit cc456d6c020a021a70edb1789cee25b68b74c106
Author: Josh Soref <21...@users.noreply.github.com>
AuthorDate: Sun Dec 10 07:16:18 2023 -0500
GH-38979: [C++] Fix spelling (#38980)
### Rationale for this change
### What changes are included in this PR?
Spelling fixes to cpp/src/
### Are these changes tested?
### Are there any user-facing changes?
* Closes: #38979
Authored-by: Josh Soref <21...@users.noreply.github.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
CMakeLists.txt | 2 +-
protocol/rados_protocol.h | 2 +-
protocol/skyhook_protocol.h | 2 +-
3 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 0019251..67fddec 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -12,7 +12,7 @@
# 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 limitationsn
+# specific language governing permissions and limitations
# under the License.
# add the client subdirectory
diff --git a/protocol/rados_protocol.h b/protocol/rados_protocol.h
index 6a0b2ea..52b0ed4 100644
--- a/protocol/rados_protocol.h
+++ b/protocol/rados_protocol.h
@@ -44,7 +44,7 @@ class IoCtxInterface {
/// \param[in] cls the name of the object class.
/// \param[in] method the name of the object class method.
/// \param[in] in a bufferlist to send data to the object class method.
- /// \param[in] out a bufferlist to recieve data from the object class method.
+ /// \param[in] out a bufferlist to receive data from the object class method.
arrow::Status exec(const std::string& oid, const char* cls, const char* method,
ceph::bufferlist& in, ceph::bufferlist& out);
/// \brief Execute POSIX stat on a RADOS object.
diff --git a/protocol/skyhook_protocol.h b/protocol/skyhook_protocol.h
index 37dda8a..4097b4e 100644
--- a/protocol/skyhook_protocol.h
+++ b/protocol/skyhook_protocol.h
@@ -101,7 +101,7 @@ class SkyhookDirectObjectAccess {
}
/// Execute an object class method. It uses the `librados::exec` api to
- /// perform object clsass method calls on the storage node and
+ /// perform object class method calls on the storage node and
/// stores the result in an output bufferlist.
arrow::Status Exec(uint64_t inode, const std::string& fn, ceph::bufferlist& in,
ceph::bufferlist& out) {
(arrow-skyhook) 18/18: GH-46011: [C++] Hide DCHECK family from public headers (#46015)
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 365241620515a6b71aba5d10af9e8e102bb54f4c
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Fri Apr 4 11:24:06 2025 +0900
GH-46011: [C++] Hide DCHECK family from public headers (#46015)
### Rationale for this change
`arrow/util/logging.h` exports `DCHECK()` family macros as internal macros. They are conflicted with macros provided by glog when glog headers are included after `arrow/util/logging.h`.
### What changes are included in this PR?
* Move `DCHECK()` family macros to `arrow/util/logging_internal.h`
* Use `arrow/util/logging_internal.h` not `.../logging.h` in `*.cc`
* Use `ARROW_DCHECK*()` macros instead of `DCHECK*()` macros in headers including `*_internal.h`
* Use `ARROW_DCHECK*()` macros instead of `DCHECK*()` macros in examples
* Use `ARROW_DCHECK*()` macros instead of `DCHECK*()` macros in `python/`
### Are these changes tested?
Yes.
### Are there any user-facing changes?
Yes.
* GitHub Issue: #46011
Authored-by: Sutou Kouhei <ko...@clear-code.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
cls/cls_skyhook.cc | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/cls/cls_skyhook.cc b/cls/cls_skyhook.cc
index 632b82f..cf41e98 100644
--- a/cls/cls_skyhook.cc
+++ b/cls/cls_skyhook.cc
@@ -24,7 +24,7 @@
#include "arrow/dataset/file_parquet.h"
#include "arrow/io/interfaces.h"
#include "arrow/result.h"
-#include "arrow/util/logging.h"
+#include "arrow/util/logging_internal.h"
#include "skyhook/protocol/skyhook_protocol.h"
(arrow-skyhook) 04/18: ARROW-16585: [C++] Add support for absolute CMAKE_INSTALL_*DIR
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
raulcd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-skyhook.git
commit 1437ff0511c3202c2200ce40240c1279df146dd2
Author: Sutou Kouhei <ko...@clear-code.com>
AuthorDate: Thu May 19 10:56:18 2022 +0900
ARROW-16585: [C++] Add support for absolute CMAKE_INSTALL_*DIR
Closes #13182 from kou/cpp-pkg-config-path
Authored-by: Sutou Kouhei <ko...@clear-code.com>
Signed-off-by: Sutou Kouhei <ko...@clear-code.com>
---
skyhook.pc.in | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/skyhook.pc.in b/skyhook.pc.in
index a3a4da5..8f7acfa 100644
--- a/skyhook.pc.in
+++ b/skyhook.pc.in
@@ -16,11 +16,11 @@
# under the License.
prefix=@CMAKE_INSTALL_PREFIX@
-libdir=${prefix}/@CMAKE_INSTALL_LIBDIR@
-includedir=${prefix}/@CMAKE_INSTALL_INCLUDEDIR@
+includedir=@ARROW_PKG_CONFIG_INCLUDEDIR@
+libdir=@ARROW_PKG_CONFIG_LIBDIR@
Name: Skyhook
Description: Skyhook is a plugin for offloading computations into Ceph.
Version: @SKYHOOK_VERSION@
Requires: arrow_dataset
-Libs: -L${libdir} -larrow_skyhook_client
\ No newline at end of file
+Libs: -L${libdir} -larrow_skyhook_client