You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by ri...@apache.org on 2025/10/20 03:29:51 UTC
(auron) branch gpu created (now 0d71e374)
This is an automated email from the ASF dual-hosted git repository.
richox pushed a change to branch gpu
in repository https://gitbox.apache.org/repos/asf/auron.git
at 0d71e374 init gpu
This branch includes the following new commits:
new 0d71e374 init gpu
The 1 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.
(auron) 01/01: init gpu
Posted by ri...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
richox pushed a commit to branch gpu
in repository https://gitbox.apache.org/repos/asf/auron.git
commit 0d71e374df2e971f9df5877db982514bb06eaaa5
Author: zhangli20 <ri...@gmail.com>
AuthorDate: Thu Oct 9 16:33:01 2025 +0800
init gpu
---
Cargo.lock | 14 +
Cargo.toml | 1 +
native-engine/auron-cudf-bridge/Cargo.toml | 15 +
.../auron-cudf-bridge/bridge-cpp/.clang-format | 20 ++
.../auron-cudf-bridge/bridge-cpp/CMakeLists.txt | 56 +++
.../auron-cudf-bridge/bridge-cpp/CMakePresets.json | 15 +
.../auron-cudf-bridge/bridge-cpp/build.sh | 20 ++
.../bridge-cpp/compile_commands.json | 1 +
.../auron-cudf-bridge/bridge-cpp/src/column.cpp | 67 ++++
.../auron-cudf-bridge/bridge-cpp/src/column.hpp | 38 ++
.../bridge-cpp/src/column_ffi.cpp | 35 ++
.../bridge-cpp/src/datasource.cpp | 116 ++++++
.../bridge-cpp/src/datasource.hpp | 44 +++
.../bridge-cpp/src/datasource_ffi.cpp | 93 +++++
.../auron-cudf-bridge/bridge-cpp/src/expr.cpp | 93 +++++
.../auron-cudf-bridge/bridge-cpp/src/expr.hpp | 33 ++
.../auron-cudf-bridge/bridge-cpp/src/expr_ffi.cpp | 93 +++++
.../bridge-cpp/src/ffi_ptr_types.cpp | 22 ++
.../bridge-cpp/src/ffi_ptr_types.hpp | 30 ++
.../bridge-cpp/src/precompile/headers.hpp | 46 +++
.../auron-cudf-bridge/bridge-cpp/src/table.cpp | 89 +++++
.../auron-cudf-bridge/bridge-cpp/src/table.hpp | 51 +++
.../auron-cudf-bridge/bridge-cpp/src/table_ffi.cpp | 76 ++++
.../src/test/test_apply_boolean_mask.cpp | 194 ++++++++++
.../bridge-cpp/src/test/test_arrow_interop.cpp | 67 ++++
.../bridge-cpp/src/test/test_compute_column.cpp | 389 +++++++++++++++++++++
.../src/test/test_parquet_datasource.cpp | 36 ++
.../auron-cudf-bridge/bridge-cpp/src/test/util.hpp | 105 ++++++
.../bridge-cpp/src/util/arrow.hpp | 56 +++
.../auron-cudf-bridge/bridge-cpp/src/util/ffi.hpp | 76 ++++
.../bridge-cpp/src/util/types.hpp | 68 ++++
.../auron-cudf-bridge/bridge-cpp/vcpkg.json | 29 ++
.../vcpkg_ports/cuda-headers/get-cuda-headers.sh | 28 ++
.../vcpkg_ports/cuda-headers/portfile.cmake | 7 +
.../bridge-cpp/vcpkg_ports/cuda-headers/vcpkg.json | 9 +
.../vcpkg_ports/libcudf-headers/portfile.cmake | 12 +
.../vcpkg_ports/libcudf-headers/vcpkg.json | 11 +
.../bridge-cpp/vcpkg_ports/libcudf/portfile.cmake | 49 +++
.../bridge-cpp/vcpkg_ports/libcudf/vcpkg.json | 12 +
.../vcpkg_ports/librmm-headers/portfile.cmake | 13 +
.../vcpkg_ports/librmm-headers/vcpkg.json | 11 +
native-engine/auron-cudf-bridge/build.rs | 60 ++++
.../auron-cudf-bridge/src/datafusion/mod.rs | 33 ++
.../src/datafusion/parquet_scan.rs | 166 +++++++++
native-engine/auron-cudf-bridge/src/ffi/column.rs | 72 ++++
.../auron-cudf-bridge/src/ffi/datasource.rs | 173 +++++++++
native-engine/auron-cudf-bridge/src/ffi/expr.rs | 222 ++++++++++++
native-engine/auron-cudf-bridge/src/ffi/mod.rs | 103 ++++++
native-engine/auron-cudf-bridge/src/ffi/table.rs | 147 ++++++++
native-engine/auron-cudf-bridge/src/lib.rs | 17 +
native-engine/auron-cudf-bridge/test/item.parquet | Bin 0 -> 5926585 bytes
51 files changed, 3233 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 79f53f4d..86e99bb4 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -370,6 +370,20 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "auron-cudf-bridge"
+version = "0.1.0"
+dependencies = [
+ "arrow",
+ "datafusion",
+ "datafusion-ext-commons",
+ "futures",
+ "log",
+ "num",
+ "parquet",
+ "paste",
+]
+
[[package]]
name = "auron-jni-bridge"
version = "0.1.0"
diff --git a/Cargo.toml b/Cargo.toml
index 2874d7ca..f71cb0c0 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -23,6 +23,7 @@ members = [
"native-engine/datafusion-ext-plans",
"native-engine/auron",
"native-engine/auron-jni-bridge",
+ "native-engine/auron-cudf-bridge",
"native-engine/auron-serde",
]
diff --git a/native-engine/auron-cudf-bridge/Cargo.toml b/native-engine/auron-cudf-bridge/Cargo.toml
new file mode 100644
index 00000000..70aa2546
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/Cargo.toml
@@ -0,0 +1,15 @@
+[package]
+name = "auron-cudf-bridge"
+version = "0.1.0"
+edition = "2021"
+resolver = "1"
+
+[dependencies]
+arrow = { workspace = true }
+datafusion = { workspace = true }
+datafusion-ext-commons = { workspace = true }
+futures = "0.3.31"
+parquet = { workspace = true }
+log = "0.4.27"
+num = "0.4.3"
+paste = "1.0.15"
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/.clang-format b/native-engine/auron-cudf-bridge/bridge-cpp/.clang-format
new file mode 100644
index 00000000..5dc170d7
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/.clang-format
@@ -0,0 +1,20 @@
+BasedOnStyle: Chromium
+
+## custom style
+AccessModifierOffset: 0
+AlignAfterOpenBracket: AlwaysBreak
+AllowAllParametersOfDeclarationOnNextLine: true
+AllowShortBlocksOnASingleLine: false
+AllowShortFunctionsOnASingleLine: false
+BinPackArguments: true
+BinPackParameters: OnePerLine
+BreakConstructorInitializers: AfterColon
+BreakTemplateDeclarations: true
+ColumnLimit: 100
+ContinuationIndentWidth: 4
+IncludeBlocks: Merge
+IndentAccessModifiers: false
+IndentCaseBlocks: false
+IndentWidth: 4
+InsertBraces: true
+InsertNewlineAtEOF: true
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/CMakeLists.txt b/native-engine/auron-cudf-bridge/bridge-cpp/CMakeLists.txt
new file mode 100644
index 00000000..5e4b8b72
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/CMakeLists.txt
@@ -0,0 +1,56 @@
+cmake_minimum_required(VERSION 3.15)
+project(auron-cudf-bridge)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+set(VCPKG_BUILD_TYPE release)
+set(VCPKG_OVERLAY_PORTS "${CMAKE_SOURCE_DIR}/vcpkg_ports")
+
+include_directories("${CMAKE_SOURCE_DIR}/")
+include_directories("${CMAKE_SOURCE_DIR}/vcpkg_installed/${VCPKG_TARGET_TRIPLET}/include")
+
+file(GLOB_RECURSE SOURCES CONFIGURE_DEPENDS "src/*.cpp")
+list(FILTER SOURCES EXCLUDE REGEX ".*_test\\.cpp$")
+add_library(${PROJECT_NAME} STATIC ${SOURCES})
+
+# disable including CUDA under MacSs
+set(HAS_CUDA TRUE)
+if("${CMAKE_SYSTEM_NAME}" MATCHES "Darwin")
+ message(WARNING "auron-cudf-bridge cannot be built on MacOS, this build command is only used for providing C++ headers for easier developing")
+ set(HAS_CUDA FALSE)
+endif()
+
+find_package(Arrow CONFIG REQUIRED)
+find_package(spdlog CONFIG REQUIRED)
+find_package(GTest REQUIRED)
+if (${HAS_CUDA})
+ find_package(CUDAToolkit REQUIRED)
+ find_package(cudf REQUIRED)
+endif()
+
+target_link_libraries(${PROJECT_NAME} PRIVATE Arrow::arrow_static)
+target_link_libraries(${PROJECT_NAME} PRIVATE spdlog::spdlog)
+if (${HAS_CUDA})
+ target_precompile_headers(${PROJECT_NAME} PUBLIC "${CMAKE_SOURCE_DIR}/src/precompile/headers.hpp")
+ target_link_libraries(${PROJECT_NAME} PRIVATE CUDA::cudart)
+ target_link_libraries(${PROJECT_NAME} PRIVATE cudf::cudf)
+endif()
+
+# tests
+enable_testing()
+aux_source_directory("${CMAKE_SOURCE_DIR}/src/test" TEST_SOURCES)
+foreach(TEST_CPP ${TEST_SOURCES})
+ if(TEST_CPP MATCHES "test_.*\\.cpp$")
+ get_filename_component(TEST_BIN ${TEST_CPP} NAME_WE)
+ add_executable("${TEST_BIN}" "${TEST_CPP}")
+ target_link_libraries("${TEST_BIN}" PRIVATE ${PROJECT_NAME})
+ target_link_libraries("${TEST_BIN}" PRIVATE Arrow::arrow_static)
+ target_link_libraries("${TEST_BIN}" PRIVATE spdlog::spdlog)
+ target_link_libraries("${TEST_BIN}" PRIVATE GTest::gtest_main)
+ if (${HAS_CUDA})
+ target_link_libraries("${TEST_BIN}" PRIVATE CUDA::cudart)
+ target_link_libraries("${TEST_BIN}" PRIVATE cudf::cudf)
+ endif()
+ gtest_discover_tests("${TEST_BIN}" WORKING_DIRECTORY "${CMAKE_SOURCE_DIR}")
+ endif()
+endforeach()
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/CMakePresets.json b/native-engine/auron-cudf-bridge/bridge-cpp/CMakePresets.json
new file mode 100644
index 00000000..c3946264
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/CMakePresets.json
@@ -0,0 +1,15 @@
+{
+ "version": 2,
+ "configurePresets": [
+ {
+ "name": "vcpkg",
+ "generator": "Ninja",
+ "binaryDir": "${sourceDir}/build",
+ "cacheVariables": {
+ "CMAKE_TOOLCHAIN_FILE": "$env{VCPKG_ROOT}/scripts/buildsystems/vcpkg.cmake",
+ "VCPKG_INSTALLED_DIR": "${sourceDir}/vcpkg_installed",
+ "VCPKG_INSTALL_OPTIONS": "--allow-unsupported"
+ }
+ }
+ ]
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/build.sh b/native-engine/auron-cudf-bridge/bridge-cpp/build.sh
new file mode 100755
index 00000000..68064ab8
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/build.sh
@@ -0,0 +1,20 @@
+#!/bin/bash
+
+set -x
+set -e
+
+ROOT_DIR="$(cd `dirname "$0"` && pwd)"
+
+if [ ! -d "$VCPKG_ROOT" ]; then
+ echo "invalid VCPKG_ROOT: $VCPKG_ROOT" >&2
+ exit -1
+fi
+
+cmake --preset=vcpkg \
+ -DCMAKE_BUILD_TYPE=Release \
+ -DVCPKG_BUILD_TYPE=release \
+ -DVCPKG_OVERLAY_PORTS=./vcpkg_ports \
+ -DCMAKE_EXPORT_COMPILE_COMMANDS=1
+
+cmake --build build
+cmake --build build --target test
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/compile_commands.json b/native-engine/auron-cudf-bridge/bridge-cpp/compile_commands.json
new file mode 120000
index 00000000..25eb4b2b
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/compile_commands.json
@@ -0,0 +1 @@
+build/compile_commands.json
\ No newline at end of file
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/column.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/column.cpp
new file mode 100644
index 00000000..9ad6cb87
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/column.cpp
@@ -0,0 +1,67 @@
+// 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 "src/column.hpp"
+#include "src/util/arrow.hpp"
+
+// OwnedColumn implementations
+OwnedColumn::OwnedColumn(std::unique_ptr<cudf::column> owned) :
+ owned(std::make_optional(std::move(owned))) {
+}
+
+OwnedColumn::OwnedColumn(
+ const std::shared_ptr<OwnedColumn>& holder, const cudf::column_view& referenced) :
+ referenced(std::make_pair(holder, referenced)) {
+}
+
+OwnedColumn::OwnedColumn(OwnedColumn&& moving) noexcept :
+ owned(std::move(moving.owned)), referenced(std::move(moving.referenced)) {
+}
+
+OwnedColumn OwnedColumn::import_from_arrow(const arrow::Array& array) {
+ auto ffi_schema = ArrowFFISchema{};
+ auto ffi_array = ArrowFFIArray{};
+ ARROW_OK_OR_THROW(arrow::ExportArray(array, &ffi_array, &ffi_schema), "Failed to export array");
+ return import_from_arrow_ffi(&ffi_schema, &ffi_array);
+}
+
+OwnedColumn OwnedColumn::import_from_arrow_ffi(
+ const ArrowSchema* ffi_schema, const ArrowArray* ffi_array) {
+ return OwnedColumn(cudf::from_arrow_column(ffi_schema, ffi_array));
+}
+
+std::shared_ptr<arrow::Array> OwnedColumn::export_to_arrow(
+ const std::shared_ptr<arrow::DataType>& data_type) const {
+ auto host_array = cudf::to_arrow_host(view());
+ auto device_array = ARROW_VALUE_OR_THROW(
+ arrow::ImportDeviceArray(host_array.get(), data_type),
+ "error importing arrow device array");
+ return device_array;
+}
+
+void OwnedColumn::export_to_arrow_ffi(ArrowSchema* ffi_schema, ArrowArray* ffi_array) const {
+ auto host_array = cudf::to_arrow_host(view());
+ auto device_array = ARROW_VALUE_OR_THROW(
+ arrow::ImportDeviceArray(host_array.get(), ffi_schema),
+ "error importing arrow device array");
+ ARROW_OK_OR_THROW(arrow::ExportArray(*device_array, ffi_array), "error exporting arrow array");
+}
+
+cudf::column_view OwnedColumn::view() const {
+ if (owned.has_value()) {
+ return owned.value()->view();
+ }
+ return referenced.value().second;
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/column.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/column.hpp
new file mode 100644
index 00000000..911cff3b
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/column.hpp
@@ -0,0 +1,38 @@
+// 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
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+struct OwnedColumn {
+ std::optional<std::unique_ptr<cudf::column>> owned;
+ std::optional<std::pair<std::shared_ptr<OwnedColumn>, cudf::column_view>> referenced;
+
+ explicit OwnedColumn(std::unique_ptr<cudf::column> owned);
+ OwnedColumn(const std::shared_ptr<OwnedColumn>& holder, const cudf::column_view& referenced);
+ OwnedColumn(OwnedColumn&& moving) noexcept;
+
+ static OwnedColumn import_from_arrow(const arrow::Array& array);
+ static OwnedColumn import_from_arrow_ffi(
+ const ArrowSchema* ffi_schema, const ArrowArray* ffi_array);
+
+ std::shared_ptr<arrow::Array> export_to_arrow(
+ const std::shared_ptr<arrow::DataType>& data_type) const;
+ void export_to_arrow_ffi(ArrowSchema* ffi_schema, ArrowArray* ffi_array) const;
+
+ cudf::column_view view() const;
+};
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/column_ffi.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/column_ffi.cpp
new file mode 100644
index 00000000..a4f47c7b
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/column_ffi.cpp
@@ -0,0 +1,35 @@
+// 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 "src/column.hpp"
+#include "src/ffi_ptr_types.hpp"
+#include "src/util/ffi.hpp"
+
+extern "C" FFIResult<FFI_ColumnPtr> ffi_column_from_arrow(ArrowSchema* schema, ArrowArray* array) {
+ try {
+ auto column = OwnedColumn::import_from_arrow_ffi(schema, array);
+ return FFIResult(true, new FFIObjWrapper(std::make_unique<OwnedColumn>(std::move(column))));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, (FFI_ColumnPtr) nullptr));
+}
+
+extern "C" FFIResult<void> ffi_column_to_arrow(
+ FFI_ColumnPtr column, ArrowSchema* ffi_schema, ArrowArray* ffi_array) {
+ try {
+ column->inner->export_to_arrow_ffi(ffi_schema, ffi_array);
+ return FFIResult<void>(true);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult<void>(false));
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource.cpp
new file mode 100644
index 00000000..b8052942
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource.cpp
@@ -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.
+
+#include "src/datasource.hpp"
+#include <cudf/io/types.hpp>
+
+ParquetScanTableStream::ParquetScanTableStream(
+ std::shared_ptr<cudf::io::chunked_parquet_reader> reader,
+ std::unique_ptr<cudf::io::datasource> datasource) :
+ reader(reader),
+ datasource(std::move(datasource)) {
+ // according to cudf docs, first chunk must be existed
+ if (auto first_table = reader->read_chunk().tbl; first_table->num_rows() > 0) {
+ next_table = std::make_optional(std::move(first_table));
+ }
+}
+
+bool ParquetScanTableStream::has_next() {
+ if (!next_table.has_value() && reader->has_next()) {
+ next_table = std::make_optional(std::move(reader->read_chunk().tbl));
+ }
+ return next_table.has_value();
+}
+
+OwnedTable ParquetScanTableStream::next() {
+ auto p = std::move(next_table).value();
+ next_table = std::nullopt;
+ return OwnedTable(std::move(p));
+}
+
+inline static std::vector<cudf::io::reader_column_schema> build_parquet_reader_column_schema(
+ arrow::Schema const& output_schema, cudf::io::parquet_metadata const& metadata) {
+ auto column_schema = std::vector<cudf::io::reader_column_schema>();
+ auto& fields = output_schema.fields();
+
+ for (auto const& col : metadata.schema().root().children()) {
+ auto field = std::find_if(fields.begin(), fields.end(), [&](auto f) {
+ return strcasecmp(f->name().c_str(), col.name().c_str()) == 0;
+ });
+ if (field == fields.end()) {
+ column_schema.emplace_back();
+ } else if (field->get()->type()->id() == arrow::Type::STRING) {
+ column_schema.emplace_back().set_convert_binary_to_strings(true);
+ } else if (field->get()->type()->id() == arrow::Type::STRUCT) {
+ throw std::runtime_error("complex type not implemented");
+ } else {
+ column_schema.emplace_back();
+ }
+ }
+ return column_schema;
+}
+
+std::unique_ptr<TableStream> create_parquet_scan_table_stream(
+ std::unique_ptr<cudf::io::datasource> datasource,
+ size_t read_offset,
+ size_t read_size,
+ const std::shared_ptr<arrow::Schema>& output_schema,
+ std::shared_ptr<ExprBuilder> predicate_filter_expr) {
+ auto datasource_total_size = datasource->size();
+ auto source_info = cudf::io::source_info(&*datasource);
+ auto metadata = cudf::io::read_parquet_metadata(source_info);
+
+ // calculate row group ids to read
+ auto read_row_group_ids = std::vector<cudf::size_type>();
+ auto num_row_groups = metadata.num_rowgroups();
+ auto file_total_byte_size = static_cast<uint64_t>(0);
+ for (auto i = 0; i < num_row_groups; i++) {
+ auto rg_total_byte_size = metadata.rowgroup_metadata()[i].at("total_byte_size");
+ file_total_byte_size += rg_total_byte_size;
+ }
+ auto start = static_cast<uint64_t>(0);
+ for (auto i = 0; i < num_row_groups; i++) {
+ auto rg_total_byte_size = metadata.rowgroup_metadata()[i].at("total_byte_size");
+ start += rg_total_byte_size;
+ auto mapped_end = start * datasource_total_size / file_total_byte_size;
+ if (mapped_end > read_offset && mapped_end <= read_offset + read_size) {
+ read_row_group_ids.push_back(i);
+ }
+ }
+ if (read_row_group_ids.empty()) {
+ return TableStream::empty_stream(); // no row groups to read
+ }
+ spdlog::info("reading parquet row groups: {}", read_row_group_ids);
+ spdlog::info("reading parquet fields: {}", output_schema->field_names());
+
+ auto reader_column_schema = build_parquet_reader_column_schema(*output_schema, metadata);
+ auto options_builder =
+ cudf::io::parquet_reader_options_builder(source_info)
+ .row_groups(std::vector<std::vector<cudf::size_type>>{std::move(read_row_group_ids)})
+ .set_column_schema(reader_column_schema)
+ .columns(output_schema->field_names())
+ .allow_mismatched_pq_schemas(true);
+
+ // set optional predicate filter
+ auto const& pred = predicate_filter_expr->expr();
+ if (typeid(pred) == typeid(cudf::ast::operation)) {
+ options_builder.filter(pred);
+ }
+ auto options = options_builder.build();
+
+ auto chunk_bytes_size = 16777216;
+ auto reader = std::make_shared<cudf::io::chunked_parquet_reader>(chunk_bytes_size, options);
+ return std::make_unique<ParquetScanTableStream>(reader, std::move(datasource));
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource.hpp
new file mode 100644
index 00000000..27a8e126
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource.hpp
@@ -0,0 +1,44 @@
+// 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 "src/expr.hpp"
+#include "src/table.hpp"
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+#include <cudf/io/datasource.hpp>
+#include <memory>
+
+struct ParquetScanTableStream : TableStream {
+ std::shared_ptr<cudf::io::chunked_parquet_reader> reader;
+ std::unique_ptr<cudf::io::datasource> datasource;
+ std::optional<std::unique_ptr<cudf::table>> next_table;
+
+ explicit ParquetScanTableStream(
+ std::shared_ptr<cudf::io::chunked_parquet_reader> reader,
+ std::unique_ptr<cudf::io::datasource> datasource);
+
+ inline bool has_next() override;
+ inline OwnedTable next() override;
+};
+
+std::unique_ptr<TableStream> create_parquet_scan_table_stream(
+ std::unique_ptr<cudf::io::datasource> datasource,
+ size_t read_offset,
+ size_t read_size,
+ const std::shared_ptr<arrow::Schema>& output_schema,
+ std::shared_ptr<ExprBuilder> predicate_filter_expr);
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource_ffi.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource_ffi.cpp
new file mode 100644
index 00000000..bfe1d41a
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/datasource_ffi.cpp
@@ -0,0 +1,93 @@
+// 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 "src/datasource.hpp"
+#include "src/ffi_ptr_types.hpp"
+#include "src/util/arrow.hpp"
+#include "src/util/ffi.hpp"
+
+struct alignas(8) FFI_CudfDataSource {
+ using DtorFn = void (*)(FFI_CudfDataSource*);
+ using GetDataSizeFn = size_t (*)(const FFI_CudfDataSource*);
+ using ReadFn = FFIResult<size_t> (*)(
+ FFI_CudfDataSource* self, size_t offset, uint8_t* buf, size_t buf_size);
+
+ DtorFn dtor;
+ GetDataSizeFn data_size;
+ ReadFn read;
+ void* raw;
+
+ inline auto to_cudf_io_datasource() -> std::unique_ptr<cudf::io::datasource> {
+ struct Wrapper : cudf::io::datasource {
+ FFI_CudfDataSource inner;
+
+ explicit inline Wrapper(FFI_CudfDataSource wrapped) : inner(wrapped) {
+ }
+
+ inline ~Wrapper() override {
+ inner.dtor(&inner);
+ }
+
+ inline std::unique_ptr<buffer> host_read(size_t offset, size_t size) {
+ auto buffer = std::vector<uint8_t>(size);
+ auto read = (size_t) 0;
+
+ while (read < size) {
+ auto read_size = host_read(offset + read, size - read, &buffer[read]);
+ if (read_size == 0) {
+ throw std::runtime_error("got unexpected EOF while reading datasource");
+ }
+ read += read_size;
+ }
+ return owning_buffer<std::vector<uint8_t>>::create(std::move(buffer));
+ }
+
+ inline size_t host_read(size_t offset, size_t size, uint8_t* dst) override {
+ auto read_size_ret = inner.read(&inner, offset, dst, size);
+ if (!read_size_ret.ok) {
+ throw std::runtime_error("error reading datasource");
+ }
+ return read_size_ret.value;
+ }
+
+ inline size_t size() const override {
+ return inner.data_size(&inner);
+ }
+ };
+ return std::make_unique<Wrapper>(*this);
+ }
+};
+
+extern "C" FFIResult<FFI_TableStreamPtr> ffi_create_parquet_scan_table_stream(
+ FFI_CudfDataSource* datasource,
+ size_t read_offset,
+ size_t read_size,
+ ArrowSchema* ffi_schema,
+ FFI_ExprBuilderPtr predicate_filter_expr) {
+ try {
+ // Convert FFI datasource to cudf datasource
+ auto cudf_datasource = datasource->to_cudf_io_datasource();
+
+ // Convert Arrow schema from FFI
+ auto output_schema =
+ ARROW_VALUE_OR_THROW(arrow::ImportSchema(ffi_schema), "error importing arrow schema");
+
+ // Call the actual function
+ auto stream = create_parquet_scan_table_stream(
+ std::move(cudf_datasource), read_offset, read_size, output_schema, predicate_filter_expr->inner);
+ return FFIResult(true, new FFIObjWrapper(std::move(stream)));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, (FFI_TableStreamPtr) nullptr));
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/expr.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/expr.cpp
new file mode 100644
index 00000000..13b2d586
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/expr.cpp
@@ -0,0 +1,93 @@
+// 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 "src/expr.hpp"
+
+size_t ExprBuilder::add_column_reference(size_t column_idx) {
+ const auto expr_id = tree.size();
+ tree.emplace<cudf::ast::column_reference>(column_idx);
+ return expr_id;
+}
+
+size_t ExprBuilder::add_column_name_reference(std::string const& name) {
+ const auto expr_id = tree.size();
+ tree.emplace<cudf::ast::column_name_reference>(name);
+ return expr_id;
+}
+
+size_t ExprBuilder::add_unary_operation(size_t arg1_expr_id, int op) {
+ const auto& arg1 = tree.at(arg1_expr_id);
+ const auto expr_id = tree.size();
+ tree.emplace<cudf::ast::operation>(static_cast<cudf::ast::ast_operator>(op), arg1);
+ return expr_id;
+}
+
+size_t ExprBuilder::add_binary_operation(size_t arg1_expr_id, size_t arg2_expr_id, int op) {
+ auto const& arg1 = tree.at(arg1_expr_id);
+ auto const& arg2 = tree.at(arg2_expr_id);
+ auto expr_id = tree.size();
+ tree.emplace<cudf::ast::operation>((cudf::ast::ast_operator)op, arg1, arg2);
+ return expr_id;
+}
+
+size_t ExprBuilder::add_literal(std::shared_ptr<cudf::scalar> scalar) {
+ auto& scalar_value = *scalar.get();
+ auto&& type_id = typeid(scalar_value);
+ auto expr_id = tree.size();
+
+#define handle_numeric_type(c_ty) \
+ if (type_id == typeid(cudf::numeric_scalar<c_ty>)) { \
+ tree.emplace<cudf::ast::literal>(dynamic_cast<cudf::numeric_scalar<c_ty>&>(scalar_value)); \
+ scalar_pool.push_back((scalar)); \
+ return expr_id; \
+ }
+ handle_numeric_type(int8_t);
+ handle_numeric_type(int16_t);
+ handle_numeric_type(int32_t);
+ handle_numeric_type(int64_t);
+ handle_numeric_type(uint8_t);
+ handle_numeric_type(uint16_t);
+ handle_numeric_type(uint32_t);
+ handle_numeric_type(uint64_t);
+ handle_numeric_type(float);
+ handle_numeric_type(double);
+
+ if (type_id == typeid(cudf::string_scalar)) {
+ tree.emplace<cudf::ast::literal>(dynamic_cast<cudf::string_scalar&>(scalar_value));
+ scalar_pool.push_back(scalar);
+ return expr_id;
+ }
+ throw std::runtime_error(std::string("unsupported scalar type: ") + type_id.name());
+}
+
+cudf::ast::expression const& ExprBuilder::expr() const {
+ return tree.back();
+}
+
+std::shared_ptr<ExprBuilder> ExprBuilder::boolean_lit_expr_builder(bool value) {
+ static auto true_expr_builder = ({
+ auto true_scalar = std::make_shared<cudf::numeric_scalar<int>>(1, true);
+ auto builder = std::make_shared<ExprBuilder>();
+ builder->add_literal(std::static_pointer_cast<cudf::scalar>(true_scalar));
+ builder;
+ });
+ static auto false_expr_builder = ({
+ auto false_scalar = std::make_shared<cudf::numeric_scalar<int>>(0, true);
+ auto builder = std::make_shared<ExprBuilder>();
+ builder->add_literal(std::static_pointer_cast<cudf::scalar>(false_scalar));
+ builder;
+ });
+ return value ? true_expr_builder : false_expr_builder;
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/expr.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/expr.hpp
new file mode 100644
index 00000000..8fafca6e
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/expr.hpp
@@ -0,0 +1,33 @@
+// 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
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+struct ExprBuilder {
+ cudf::ast::tree tree;
+ std::vector<std::shared_ptr<cudf::scalar>> scalar_pool;
+
+ size_t add_column_reference(size_t column_idx);
+ size_t add_column_name_reference(std::string const& name);
+ size_t add_unary_operation(size_t arg1_expr_id, int op);
+ size_t add_binary_operation(size_t arg1_expr_id, size_t arg2_expr_id, int op);
+ size_t add_literal(std::shared_ptr<cudf::scalar> scalar);
+ cudf::ast::expression const& expr() const;
+
+ static std::shared_ptr<ExprBuilder> boolean_lit_expr_builder(bool value);
+};
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/expr_ffi.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/expr_ffi.cpp
new file mode 100644
index 00000000..236869b3
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/expr_ffi.cpp
@@ -0,0 +1,93 @@
+// 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 "src/expr.hpp"
+#include "src/ffi_ptr_types.hpp"
+#include "src/util/ffi.hpp"
+
+#define FFI_DEF_NUMERIC_SCALAR(rust_ty, c_ty) \
+ extern "C" FFI_ScalarPtr ffi_##rust_ty##_scalar(c_ty value) { \
+ try { \
+ auto scalar = std::make_shared<cudf::numeric_scalar<c_ty>>(value); \
+ return new FFIObjWrapper((std::shared_ptr<cudf::scalar>&&)std::move(scalar)); \
+ } \
+ CATCH_EXCEPTIONS_AND_RETURN(nullptr); \
+ }
+
+FFI_DEF_NUMERIC_SCALAR(i8, int8_t);
+FFI_DEF_NUMERIC_SCALAR(i16, int16_t);
+FFI_DEF_NUMERIC_SCALAR(i32, int32_t);
+FFI_DEF_NUMERIC_SCALAR(i64, int64_t);
+FFI_DEF_NUMERIC_SCALAR(u8, uint8_t);
+FFI_DEF_NUMERIC_SCALAR(u16, uint16_t);
+FFI_DEF_NUMERIC_SCALAR(u32, uint32_t);
+FFI_DEF_NUMERIC_SCALAR(u64, uint64_t);
+FFI_DEF_NUMERIC_SCALAR(f32, float);
+FFI_DEF_NUMERIC_SCALAR(f64, double);
+
+extern "C" FFI_ScalarPtr ffi_string_scalar(char const* value) {
+ try {
+ auto scalar = std::make_shared<cudf::string_scalar>(std::string(value));
+ return new FFIObjWrapper((std::shared_ptr<cudf::scalar>&&)std::move(scalar));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(nullptr);
+}
+
+extern "C" FFI_ExprBuilderPtr ffi_new_expr_builder() {
+ try {
+ return new FFIObjWrapper(std::make_shared<ExprBuilder>());
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(nullptr);
+}
+
+extern "C" size_t ffi_expr_builder_add_literal(
+ FFI_ExprBuilderPtr expr_builder, FFI_ScalarPtr scalar_value) {
+ try {
+ return expr_builder->inner->add_literal(scalar_value->inner);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(-1);
+}
+
+extern "C" size_t ffi_expr_builder_add_column_reference(
+ FFI_ExprBuilderPtr expr_builder, size_t column_idx) {
+ try {
+ return expr_builder->inner->add_column_reference(column_idx);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(-1);
+}
+
+extern "C" size_t ffi_expr_builder_add_column_name_reference(
+ FFI_ExprBuilderPtr expr_builder, char const* name) {
+ try {
+ return expr_builder->inner->add_column_name_reference(name);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(-1);
+}
+
+extern "C" size_t ffi_expr_builder_add_unary_expr(
+ FFI_ExprBuilderPtr expr_builder, size_t arg1_expr_id, int32_t op) {
+ try {
+ return expr_builder->inner->add_unary_operation(arg1_expr_id, op);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(-1);
+}
+
+extern "C" size_t ffi_expr_builder_add_binary_expr(
+ FFI_ExprBuilderPtr expr_builder, size_t arg1_expr_id, size_t arg2_expr_id, int32_t op) {
+ try {
+ return expr_builder->inner->add_binary_operation(arg1_expr_id, arg2_expr_id, op);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(-1);
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/ffi_ptr_types.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/ffi_ptr_types.cpp
new file mode 100644
index 00000000..59b2fe12
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/ffi_ptr_types.cpp
@@ -0,0 +1,22 @@
+// 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 "src/ffi_ptr_types.hpp"
+
+FFI_IMPLEMENT_OBJ_WRAPPED_TYPE(FFI_ExprBuilderPtr, std::shared_ptr<ExprBuilder>);
+FFI_IMPLEMENT_OBJ_WRAPPED_TYPE(FFI_ScalarPtr, std::shared_ptr<cudf::scalar>);
+FFI_IMPLEMENT_OBJ_WRAPPED_TYPE(FFI_ColumnPtr, std::unique_ptr<OwnedColumn>);
+FFI_IMPLEMENT_OBJ_WRAPPED_TYPE(FFI_TablePtr, std::unique_ptr<OwnedTable>);
+FFI_IMPLEMENT_OBJ_WRAPPED_TYPE(FFI_TableStreamPtr, std::unique_ptr<TableStream>);
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/ffi_ptr_types.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/ffi_ptr_types.hpp
new file mode 100644
index 00000000..493125be
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/ffi_ptr_types.hpp
@@ -0,0 +1,30 @@
+// 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 "src/column.hpp"
+#include "src/expr.hpp"
+#include "src/table.hpp"
+#include "src/util/ffi.hpp"
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+FFI_DECLARE_OBJ_WRAPPED_TYPE(FFI_ExprBuilderPtr, std::shared_ptr<ExprBuilder>);
+FFI_DECLARE_OBJ_WRAPPED_TYPE(FFI_ScalarPtr, std::shared_ptr<cudf::scalar>);
+FFI_DECLARE_OBJ_WRAPPED_TYPE(FFI_ColumnPtr, std::unique_ptr<OwnedColumn>);
+FFI_DECLARE_OBJ_WRAPPED_TYPE(FFI_TablePtr, std::unique_ptr<OwnedTable>);
+FFI_DECLARE_OBJ_WRAPPED_TYPE(FFI_TableStreamPtr, std::unique_ptr<TableStream>);
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/precompile/headers.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/precompile/headers.hpp
new file mode 100644
index 00000000..c5db2d62
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/precompile/headers.hpp
@@ -0,0 +1,46 @@
+// 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 <algorithm>
+#include <memory>
+#include <optional>
+#include <regex>
+#include <stdexcept>
+#include <typeinfo>
+
+// include spdlog
+#include <spdlog/fmt/ranges.h>
+#include <spdlog/spdlog.h>
+
+// include arrow
+#include <arrow/builder.h>
+#include <arrow/c/abi.h>
+#include <arrow/c/bridge.h>
+#include <arrow/record_batch.h>
+#include <arrow/result.h>
+#include <arrow/type.h>
+
+// include libcudf
+#include <cudf/ast/expressions.hpp>
+#include <cudf/interop.hpp>
+#include <cudf/io/datasource.hpp>
+#include <cudf/io/parquet.hpp>
+#include <cudf/scalar/scalar.hpp>
+#include <cudf/stream_compaction.hpp>
+#include <cudf/table/table.hpp>
+#include <cudf/table/table_view.hpp>
+#include <cudf/utilities/error.hpp>
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/table.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/table.cpp
new file mode 100644
index 00000000..8fa96953
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/table.cpp
@@ -0,0 +1,89 @@
+// 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 "src/table.hpp"
+#include "src/util/arrow.hpp"
+
+OwnedTable::OwnedTable(std::unique_ptr<cudf::table> owned) :
+ owned(std::make_optional(std::move(owned))) {
+}
+
+OwnedTable::OwnedTable(
+ const std::shared_ptr<OwnedTable>& holder, const cudf::table_view& referenced) :
+ referenced(std::make_pair(holder, referenced)) {
+}
+
+OwnedTable::OwnedTable(OwnedTable&& moving) noexcept :
+ owned(std::move(moving.owned)), referenced(std::move(moving.referenced)) {
+}
+
+OwnedTable OwnedTable::import_from_arrow(const arrow::RecordBatch& record_batch) {
+ auto ffi_schema = ArrowFFISchema{};
+ auto ffi_array = ArrowFFIArray{};
+ ARROW_OK_OR_THROW(
+ arrow::ExportRecordBatch(record_batch, &ffi_array, &ffi_schema),
+ "Failed to export record batch");
+ return import_from_arrow_ffi(&ffi_schema, &ffi_array);
+}
+
+OwnedTable OwnedTable::import_from_arrow_ffi(
+ const ArrowSchema* ffi_schema, const ArrowArray* ffi_array) {
+ return OwnedTable(cudf::from_arrow(ffi_schema, ffi_array));
+}
+
+std::shared_ptr<arrow::RecordBatch> OwnedTable::export_to_arrow(const arrow::Schema& schema) const {
+ auto data_type = std::make_shared<arrow::StructType>(schema.fields());
+ auto device_array = ARROW_VALUE_OR_THROW(
+ arrow::ImportDeviceArray(cudf::to_arrow_host(view()).get(), data_type),
+ "error importing arrow device array");
+ return ARROW_VALUE_OR_THROW(
+ arrow::RecordBatch::FromStructArray(device_array),
+ "error creating record batch from struct array");
+}
+
+void OwnedTable::export_to_arrow_ffi(ArrowSchema* ffi_schema, ArrowArray* ffi_array) const {
+ auto device_array = ARROW_VALUE_OR_THROW(
+ arrow::ImportDeviceArray(cudf::to_arrow_host(view()).get(), ffi_schema),
+ "error importing arrow device array");
+ ARROW_OK_OR_THROW(arrow::ExportArray(*device_array, ffi_array), "error exporting arrow array");
+}
+
+OwnedColumn OwnedTable::compute_column(const cudf::ast::expression& expr) const {
+ return OwnedColumn(cudf::compute_column(view(), expr));
+}
+OwnedTable OwnedTable::apply_boolean_mask(const cudf::column_view& mask) const {
+ return OwnedTable(cudf::apply_boolean_mask(view(), mask));
+}
+
+cudf::table_view OwnedTable::view() const {
+ if (owned.has_value()) {
+ return owned.value()->view();
+ }
+ return referenced.value().second;
+}
+
+// TableStream implementations
+std::unique_ptr<TableStream> TableStream::empty_stream() {
+ struct EmptyTableStream : TableStream {
+ bool has_next() override {
+ return false;
+ }
+
+ OwnedTable next() override {
+ throw std::runtime_error("unreachable: EmptyTableStream has no tables");
+ }
+ };
+ return std::make_unique<EmptyTableStream>();
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/table.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/table.hpp
new file mode 100644
index 00000000..3bdb86ce
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/table.hpp
@@ -0,0 +1,51 @@
+// 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 "src/column.hpp"
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+struct OwnedTable {
+ std::optional<std::unique_ptr<cudf::table>> owned;
+ std::optional<std::pair<std::shared_ptr<OwnedTable>, cudf::table_view>> referenced;
+
+ explicit OwnedTable(std::unique_ptr<cudf::table> owned);
+ OwnedTable(const std::shared_ptr<OwnedTable>& holder, const cudf::table_view& referenced);
+ OwnedTable(OwnedTable&& moving) noexcept;
+
+ static OwnedTable import_from_arrow(const arrow::RecordBatch& record_batch);
+ static OwnedTable import_from_arrow_ffi(
+ const ArrowSchema* ffi_schema, const ArrowArray* ffi_array);
+
+ std::shared_ptr<arrow::RecordBatch> export_to_arrow(const arrow::Schema& schema) const;
+ void export_to_arrow_ffi(ArrowSchema* ffi_schema, ArrowArray* ffi_array) const;
+
+ OwnedColumn compute_column(const cudf::ast::expression& expr) const;
+ OwnedTable apply_boolean_mask(const cudf::column_view& mask) const;
+ cudf::table_view view() const;
+};
+
+struct TableStream {
+ TableStream() = default;
+ virtual ~TableStream() = default;
+
+ virtual bool has_next() = 0;
+ virtual OwnedTable next() = 0;
+
+ static std::unique_ptr<TableStream> empty_stream();
+};
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/table_ffi.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/table_ffi.cpp
new file mode 100644
index 00000000..ce6245a4
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/table_ffi.cpp
@@ -0,0 +1,76 @@
+// 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 "src/ffi_ptr_types.hpp"
+#include "src/table.hpp"
+#include "src/util/ffi.hpp"
+
+extern "C" FFIResult<bool> ffi_table_stream_has_next(FFI_TableStreamPtr table_stream) {
+ try {
+ return FFIResult(true, table_stream->inner->has_next());
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, false));
+}
+
+extern "C" FFIResult<FFI_TablePtr> ffi_table_stream_next(FFI_TableStreamPtr table_stream) {
+ try {
+ auto next = table_stream->inner->next();
+ return FFIResult(true, new FFIObjWrapper(std::make_unique<OwnedTable>(std::move(next))));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, (FFI_TablePtr) nullptr));
+}
+
+extern "C" FFIResult<FFI_TablePtr> ffi_table_from_arrow(ArrowSchema* schema, ArrowArray* array) {
+ try {
+ auto table = OwnedTable::import_from_arrow_ffi(schema, array);
+ return FFIResult(true, new FFIObjWrapper(std::make_unique<OwnedTable>(std::move(table))));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, (FFI_TablePtr) nullptr));
+}
+
+extern "C" FFIResult<void> ffi_table_to_arrow(
+ FFI_TablePtr table, ArrowSchema* ffi_schema, ArrowArray* ffi_array) {
+ try {
+ table->inner->export_to_arrow_ffi(ffi_schema, ffi_array);
+ return FFIResult<void>(true);
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult<void>(false));
+}
+
+extern "C" FFIResult<FFI_ColumnPtr> ffi_table_compute_column(
+ FFI_TablePtr table, FFI_ExprBuilderPtr expr) {
+ try {
+ auto column = table->inner->compute_column(expr->inner->expr());
+ return FFIResult(true, new FFIObjWrapper(std::make_unique<OwnedColumn>(std::move(column))));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, (FFI_ColumnPtr) nullptr));
+}
+
+extern "C" FFIResult<FFI_TablePtr> ffi_table_apply_boolean_mask(
+ FFI_TablePtr table, FFI_ColumnPtr mask) {
+ try {
+ auto result = table->inner->apply_boolean_mask(mask->inner->view());
+ return FFIResult(true, new FFIObjWrapper(std::make_unique<OwnedTable>(std::move(result))));
+ }
+ CATCH_EXCEPTIONS_AND_RETURN(FFIResult(false, (FFI_TablePtr) nullptr));
+}
+
+extern "C" size_t ffi_table_num_rows(FFI_TablePtr table) {
+ return table->inner->view().num_rows();
+}
+
+extern "C" size_t ffi_table_num_columns(FFI_TablePtr table) {
+ return table->inner->view().num_columns();
+}
\ No newline at end of file
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_apply_boolean_mask.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_apply_boolean_mask.cpp
new file mode 100644
index 00000000..2804e2f9
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_apply_boolean_mask.cpp
@@ -0,0 +1,194 @@
+// Licensed to the 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 "src/table.hpp"
+#include "src/test/util.hpp"
+
+struct FilterTest : testing::Test {
+ void SetUp() override {
+ }
+ void TearDown() override {
+ }
+};
+
+TEST_F(FilterTest, test_apply_boolean_mask_basic) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create a boolean mask: [true, false, true] - should keep rows 0 and 2
+ auto mask_builder = std::make_shared<arrow::BooleanBuilder>();
+ auto mask_values = std::vector<bool>{true, false, true};
+ ARROW_OK_OR_THROW(mask_builder->AppendValues(mask_values), "Failed to append mask values");
+ auto mask_array = ARROW_VALUE_OR_THROW(mask_builder->Finish(), "Failed to finish mask array");
+
+ // Import mask into OwnedColumn and get column view
+ auto mask_column = OwnedColumn::import_from_arrow(*mask_array);
+ auto mask_view = mask_column.view();
+
+ // Apply boolean mask
+ auto filtered_table = owned_table.apply_boolean_mask(mask_view);
+
+ // Verify the filtered table properties
+ auto filtered_view = filtered_table.view();
+ ASSERT_EQ(filtered_view.num_rows(), 2); // Should have 2 rows (rows 0 and 2)
+ ASSERT_EQ(filtered_view.num_columns(), 3); // Should have same number of columns
+
+ // Export back to Arrow to verify data integrity
+ auto exported = filtered_table.export_to_arrow(*record_batch->schema());
+
+ // Verify the filtered data
+ ASSERT_EQ(exported->num_rows(), 2);
+ ASSERT_EQ(exported->num_columns(), 3);
+
+ // Check that we have the correct rows (0 and 2 from original data)
+ // Original data: id=[1,2,3], name=["Alice","Bob","Charlie"], value=[1.5,2.5,3.5]
+ // Expected filtered: id=[1,3], name=["Alice","Charlie"], value=[1.5,3.5]
+
+ // Check id column
+ auto id_array = std::static_pointer_cast<arrow::Int64Array>(exported->column(0));
+ ASSERT_EQ(id_array->Value(0), 1);
+ ASSERT_EQ(id_array->Value(1), 3);
+
+ // Check name column
+ auto name_array = std::static_pointer_cast<arrow::StringArray>(exported->column(1));
+ ASSERT_EQ(name_array->GetString(0), "Alice");
+ ASSERT_EQ(name_array->GetString(1), "Charlie");
+
+ // Check value column
+ auto value_array = std::static_pointer_cast<arrow::DoubleArray>(exported->column(2));
+ ASSERT_DOUBLE_EQ(value_array->Value(0), 1.5);
+ ASSERT_DOUBLE_EQ(value_array->Value(1), 3.5);
+}
+
+TEST_F(FilterTest, test_apply_boolean_mask_all_true) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create a boolean mask with all true values: [true, true, true]
+ auto mask_builder = std::make_shared<arrow::BooleanBuilder>();
+ auto mask_values = std::vector<bool>{true, true, true};
+ ARROW_OK_OR_THROW(mask_builder->AppendValues(mask_values), "Failed to append mask values");
+ auto mask_array = ARROW_VALUE_OR_THROW(mask_builder->Finish(), "Failed to finish mask array");
+
+ // Import mask into OwnedColumn and get column view
+ auto mask_column = OwnedColumn::import_from_arrow(*mask_array);
+ auto mask_view = mask_column.view();
+
+ // Apply boolean mask
+ auto filtered_table = owned_table.apply_boolean_mask(mask_view);
+
+ // Verify the filtered table properties
+ auto filtered_view = filtered_table.view();
+ ASSERT_EQ(filtered_view.num_rows(), 3); // Should have all 3 rows
+ ASSERT_EQ(filtered_view.num_columns(), 3); // Should have same number of columns
+
+ // Export back to Arrow to verify data integrity
+ auto exported = filtered_table.export_to_arrow(*record_batch->schema());
+
+ // Verify that all data is preserved
+ ASSERT_EQ(exported->num_rows(), record_batch->num_rows());
+ ASSERT_EQ(exported->num_columns(), record_batch->num_columns());
+ for (int i = 0; i < record_batch->num_columns(); ++i) {
+ ASSERT_TRUE(exported->column(i)->Equals(record_batch->column(i)))
+ << "Column " << i << " data does not match";
+ }
+}
+
+TEST_F(FilterTest, test_apply_boolean_mask_all_false) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create a boolean mask with all false values: [false, false, false]
+ auto mask_builder = std::make_shared<arrow::BooleanBuilder>();
+ auto mask_values = std::vector<bool>{false, false, false};
+ ARROW_OK_OR_THROW(mask_builder->AppendValues(mask_values), "Failed to append mask values");
+ auto mask_array = ARROW_VALUE_OR_THROW(mask_builder->Finish(), "Failed to finish mask array");
+
+ // Import mask into OwnedColumn and get column view
+ auto mask_column = OwnedColumn::import_from_arrow(*mask_array);
+ auto mask_view = mask_column.view();
+
+ // Apply boolean mask
+ auto filtered_table = owned_table.apply_boolean_mask(mask_view);
+
+ // Verify the filtered table properties
+ auto filtered_view = filtered_table.view();
+ ASSERT_EQ(filtered_view.num_rows(), 0); // Should have no rows
+ ASSERT_EQ(filtered_view.num_columns(), 3); // Should have same number of columns
+
+ // Export back to Arrow to verify structure
+ auto exported = filtered_table.export_to_arrow(*record_batch->schema());
+
+ // Verify that no data is present but schema is preserved
+ ASSERT_EQ(exported->num_rows(), 0);
+ ASSERT_EQ(exported->num_columns(), 3);
+}
+
+TEST_F(FilterTest, test_apply_boolean_mask_single_row) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create a boolean mask that selects only the middle row: [false, true, false]
+ auto mask_builder = std::make_shared<arrow::BooleanBuilder>();
+ auto mask_values = std::vector<bool>{false, true, false};
+ ARROW_OK_OR_THROW(mask_builder->AppendValues(mask_values), "Failed to append mask values");
+ auto mask_array = ARROW_VALUE_OR_THROW(mask_builder->Finish(), "Failed to finish mask array");
+
+ // Import mask into OwnedColumn and get column view
+ auto mask_column = OwnedColumn::import_from_arrow(*mask_array);
+ auto mask_view = mask_column.view();
+
+ // Apply boolean mask
+ auto filtered_table = owned_table.apply_boolean_mask(mask_view);
+
+ // Verify the filtered table properties
+ auto filtered_view = filtered_table.view();
+ ASSERT_EQ(filtered_view.num_rows(), 1); // Should have 1 row
+ ASSERT_EQ(filtered_view.num_columns(), 3); // Should have same number of columns
+
+ // Export back to Arrow to verify data integrity
+ auto exported = filtered_table.export_to_arrow(*record_batch->schema());
+
+ // Verify the filtered data (should be row 1 from original data)
+ // Original data: id=[1,2,3], name=["Alice","Bob","Charlie"], value=[1.5,2.5,3.5]
+ // Expected filtered: id=[2], name=["Bob"], value=[2.5]
+
+ ASSERT_EQ(exported->num_rows(), 1);
+ ASSERT_EQ(exported->num_columns(), 3);
+
+ // Check id column
+ auto id_array = std::static_pointer_cast<arrow::Int64Array>(exported->column(0));
+ ASSERT_EQ(id_array->Value(0), 2);
+
+ // Check name column
+ auto name_array = std::static_pointer_cast<arrow::StringArray>(exported->column(1));
+ ASSERT_EQ(name_array->GetString(0), "Bob");
+
+ // Check value column
+ auto value_array = std::static_pointer_cast<arrow::DoubleArray>(exported->column(2));
+ ASSERT_DOUBLE_EQ(value_array->Value(0), 2.5);
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_arrow_interop.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_arrow_interop.cpp
new file mode 100644
index 00000000..8c9dd3ab
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_arrow_interop.cpp
@@ -0,0 +1,67 @@
+// 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 "src/table.hpp"
+#include "src/test/util.hpp"
+
+struct ArrowInteropTest : testing::Test {
+ void SetUp() override {
+ }
+ void TearDown() override {
+ }
+};
+
+TEST_F(ArrowInteropTest, test_arrow_batch_import_export) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Verify the table properties
+ auto table_view = owned_table.view();
+ ASSERT_EQ(table_view.num_rows(), 3);
+ ASSERT_EQ(table_view.num_columns(), 3);
+
+ // Export back from OwnedTable to Arrow
+ auto exported = owned_table.export_to_arrow(*record_batch->schema());
+
+ // Verify the round-trip data integrity
+ ASSERT_EQ(exported->num_rows(), record_batch->num_rows());
+ ASSERT_EQ(exported->num_columns(), record_batch->num_columns());
+ for (int i = 0; i < record_batch->num_columns(); ++i) {
+ ASSERT_TRUE(exported->column(i)->Equals(record_batch->column(i)))
+ << "Column " << i << " data does not match";
+ }
+}
+
+TEST_F(ArrowInteropTest, test_arrow_array_import_export) {
+ // Create a simple Arrow array
+ auto array = create_simple_array();
+
+ // Import into OwnedColumn
+ auto owned_column = OwnedColumn::import_from_arrow(*array);
+
+ // Verify the column properties
+ auto column_view = owned_column.view();
+ ASSERT_EQ(column_view.size(), array->length());
+
+ // Export back from OwnedColumn to Arrow
+ auto exported = owned_column.export_to_arrow(array->type());
+
+ // Verify the round-trip data integrity
+ ASSERT_EQ(exported->length(), array->length());
+ ASSERT_TRUE(exported->Equals(array)) << "Array data does not match after round-trip";
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_compute_column.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_compute_column.cpp
new file mode 100644
index 00000000..aee6c175
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_compute_column.cpp
@@ -0,0 +1,389 @@
+// Licensed to the 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 "src/expr.hpp"
+#include "src/table.hpp"
+#include "src/test/util.hpp"
+#include "src/util/types.hpp"
+
+struct ColumnTest : testing::Test {
+ void SetUp() override {
+ }
+ void TearDown() override {
+ }
+};
+
+TEST_F(ColumnTest, test_compute_column_simple_reference) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression to reference column 0 (id column)
+ ExprBuilder expr_builder;
+ expr_builder.add_column_reference(0);
+
+ // Compute column using the expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::INT64); // Should be int64 type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the data matches the original id column
+ auto id_array = std::static_pointer_cast<arrow::Int64Array>(exported_array);
+ ASSERT_EQ(id_array->Value(0), 1);
+ ASSERT_EQ(id_array->Value(1), 2);
+ ASSERT_EQ(id_array->Value(2), 3);
+}
+
+TEST_F(ColumnTest, test_compute_column_literal_expression) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression with a literal value (42)
+ ExprBuilder expr_builder;
+ auto literal_scalar = std::make_shared<cudf::numeric_scalar<int64_t>>(42, true);
+ expr_builder.add_literal(literal_scalar);
+
+ // Compute column using the literal expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::INT64); // Should be int64 type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify all values are 42
+ auto result_array = std::static_pointer_cast<arrow::Int64Array>(exported_array);
+ ASSERT_EQ(result_array->Value(0), 42);
+ ASSERT_EQ(result_array->Value(1), 42);
+ ASSERT_EQ(result_array->Value(2), 42);
+}
+
+TEST_F(ColumnTest, test_compute_column_binary_addition) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression: id + 10 (column 0 + literal 10)
+ ExprBuilder expr_builder;
+ auto col_ref_id = expr_builder.add_column_reference(0); // id column
+ auto literal_scalar = std::make_shared<cudf::numeric_scalar<int64_t>>(10, true);
+ auto literal_id = expr_builder.add_literal(literal_scalar);
+ expr_builder.add_binary_operation(
+ col_ref_id, literal_id, static_cast<int>(cudf::ast::ast_operator::ADD));
+
+ // Compute column using the addition expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::INT64); // Should be int64 type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values (original id + 10)
+ // Original data: id=[1,2,3], expected result: [11,12,13]
+ auto result_array = std::static_pointer_cast<arrow::Int64Array>(exported_array);
+ ASSERT_EQ(result_array->Value(0), 11);
+ ASSERT_EQ(result_array->Value(1), 12);
+ ASSERT_EQ(result_array->Value(2), 13);
+}
+
+TEST_F(ColumnTest, test_compute_column_binary_multiplication) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression: value * 2 (column 2 * literal 2)
+ ExprBuilder expr_builder;
+ auto col_ref_id = expr_builder.add_column_reference(2); // value column (double)
+ auto literal_scalar = std::make_shared<cudf::numeric_scalar<double>>(2.0, true);
+ auto literal_id = expr_builder.add_literal(literal_scalar);
+ expr_builder.add_binary_operation(
+ col_ref_id, literal_id, static_cast<int>(cudf::ast::ast_operator::MUL));
+
+ // Compute column using the multiplication expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::FLOAT64); // Should be double type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values (original value * 2)
+ // Original data: value=[1.5,2.5,3.5], expected result: [3.0,5.0,7.0]
+ auto result_array = std::static_pointer_cast<arrow::DoubleArray>(exported_array);
+ ASSERT_DOUBLE_EQ(result_array->Value(0), 3.0);
+ ASSERT_DOUBLE_EQ(result_array->Value(1), 5.0);
+ ASSERT_DOUBLE_EQ(result_array->Value(2), 7.0);
+}
+
+TEST_F(ColumnTest, test_compute_column_double_to_double_operation) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression: value + 1.0 (column 2 + literal 1.0)
+ // Both operands are double type, so this should work without casting
+ ExprBuilder expr_builder;
+ auto value_col_ref = expr_builder.add_column_reference(2); // value column (double)
+ auto literal_scalar = std::make_shared<cudf::numeric_scalar<double>>(1.0, true);
+ auto literal_id = expr_builder.add_literal(literal_scalar);
+ expr_builder.add_binary_operation(
+ value_col_ref, literal_id, static_cast<int>(cudf::ast::ast_operator::ADD));
+
+ // Compute column using the double-to-double addition expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::FLOAT64); // Should be double type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values (value + 1.0)
+ // Original data: value=[1.5,2.5,3.5], expected result: [2.5,3.5,4.5]
+ auto result_array = std::static_pointer_cast<arrow::DoubleArray>(exported_array);
+ ASSERT_DOUBLE_EQ(result_array->Value(0), 2.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(1), 3.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(2), 4.5);
+}
+
+TEST_F(ColumnTest, test_compute_column_int_to_int_operation) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression: id + id (column 0 + column 0)
+ // Both operands are int64 type, so this should work without casting
+ ExprBuilder expr_builder;
+ auto id_col_ref1 = expr_builder.add_column_reference(0); // id column (int64)
+ auto id_col_ref2 = expr_builder.add_column_reference(0); // id column (int64)
+ expr_builder.add_binary_operation(
+ id_col_ref1, id_col_ref2, static_cast<int>(cudf::ast::ast_operator::ADD));
+
+ // Compute column using the int-to-int addition expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::INT64); // Should be int64 type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values (id + id)
+ // Original data: id=[1,2,3], expected result: [2,4,6]
+ auto result_array = std::static_pointer_cast<arrow::Int64Array>(exported_array);
+ ASSERT_EQ(result_array->Value(0), 2);
+ ASSERT_EQ(result_array->Value(1), 4);
+ ASSERT_EQ(result_array->Value(2), 6);
+}
+
+TEST_F(ColumnTest, test_compute_column_simple_expression) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create simple expression: value * 3.0
+ ExprBuilder expr_builder;
+
+ auto value_col_ref = expr_builder.add_column_reference(2); // value column
+ auto literal_3 = std::make_shared<cudf::numeric_scalar<double>>(3.0, true);
+ auto literal_3_id = expr_builder.add_literal(literal_3);
+ expr_builder.add_binary_operation(
+ value_col_ref, literal_3_id, static_cast<int>(cudf::ast::ast_operator::MUL));
+
+ // Compute column using the simple expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::FLOAT64); // Should be double type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values: value * 3.0
+ // Original data: value=[1.5,2.5,3.5]
+ // Expected result: [4.5, 7.5, 10.5]
+ auto result_array = std::static_pointer_cast<arrow::DoubleArray>(exported_array);
+ ASSERT_DOUBLE_EQ(result_array->Value(0), 4.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(1), 7.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(2), 10.5);
+}
+
+TEST_F(ColumnTest, test_compute_column_column_to_column_operation) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression: id + value (column 0 + column 2)
+ // Note: This will require type casting since id is int64 and value is double
+ ExprBuilder expr_builder;
+ auto id_col_ref = expr_builder.add_unary_operation(
+ expr_builder.add_column_reference(0), // id column (int64)
+ (int)cudf::ast::ast_operator::CAST_TO_FLOAT64);
+ auto value_col_ref = expr_builder.add_column_reference(2); // value column (double)
+ expr_builder.add_binary_operation(
+ id_col_ref, value_col_ref, static_cast<int>(cudf::ast::ast_operator::ADD));
+
+ // Compute column using the column-to-column addition expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::FLOAT64); // Should be double type (promoted)
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values (id + value)
+ // Original data: id=[1,2,3], value=[1.5,2.5,3.5], expected result: [2.5,4.5,6.5]
+ auto result_array = std::static_pointer_cast<arrow::DoubleArray>(exported_array);
+ ASSERT_DOUBLE_EQ(result_array->Value(0), 2.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(1), 4.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(2), 6.5);
+}
+
+TEST_F(ColumnTest, test_compute_column_complex_expression) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create complex expression: (id * 2) + (value * 3)
+ ExprBuilder expr_builder;
+
+ // First part: id * 2
+ auto id_col_ref = expr_builder.add_column_reference(0); // id column
+ auto literal_2 = std::make_shared<cudf::numeric_scalar<int64_t>>(2, true);
+ auto literal_2_id = expr_builder.add_literal(literal_2);
+ auto id_times_2 = expr_builder.add_binary_operation(
+ id_col_ref, literal_2_id, static_cast<int>(cudf::ast::ast_operator::MUL));
+ auto id_times_2_float64 =
+ expr_builder.add_unary_operation(id_times_2, (int)cudf::ast::ast_operator::CAST_TO_FLOAT64);
+
+ // Second part: value * 3
+ auto value_col_ref = expr_builder.add_column_reference(2); // value column
+ auto literal_3 = std::make_shared<cudf::numeric_scalar<double>>(3.0, true);
+ auto literal_3_id = expr_builder.add_literal(literal_3);
+ auto value_times_3 = expr_builder.add_binary_operation(
+ value_col_ref, literal_3_id, static_cast<int>(cudf::ast::ast_operator::MUL));
+
+ // Final expression: (id * 2) + (value * 3)
+ expr_builder.add_binary_operation(
+ id_times_2_float64, value_times_3, static_cast<int>(cudf::ast::ast_operator::ADD));
+
+ // Compute column using the complex expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::FLOAT64); // Should be double type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values: (id * 2) + (value * 3)
+ // Original data: id=[1,2,3], value=[1.5,2.5,3.5]
+ // Expected result: [1*2 + 1.5*3, 2*2 + 2.5*3, 3*2 + 3.5*3] = [2+4.5, 4+7.5, 6+10.5] =
+ // [6.5, 11.5, 16.5]
+ auto result_array = std::static_pointer_cast<arrow::DoubleArray>(exported_array);
+ ASSERT_DOUBLE_EQ(result_array->Value(0), 6.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(1), 11.5);
+ ASSERT_DOUBLE_EQ(result_array->Value(2), 16.5);
+}
+
+TEST_F(ColumnTest, test_compute_column_comparison_expression) {
+ // Create a simple Arrow record batch
+ auto record_batch = create_simple_record_batch();
+
+ // Import into OwnedTable
+ auto owned_table = OwnedTable::import_from_arrow(*record_batch);
+
+ // Create expression: id > 2 (column 0 > literal 2)
+ ExprBuilder expr_builder;
+ auto col_ref_id = expr_builder.add_column_reference(0); // id column
+ auto literal_scalar = std::make_shared<cudf::numeric_scalar<int64_t>>(2, true);
+ auto literal_id = expr_builder.add_literal(literal_scalar);
+ expr_builder.add_binary_operation(
+ col_ref_id, literal_id, static_cast<int>(cudf::ast::ast_operator::GREATER));
+
+ // Compute column using the comparison expression
+ auto computed_column = owned_table.compute_column(expr_builder.expr());
+
+ // Verify the computed column properties
+ auto column_view = computed_column.view();
+ ASSERT_EQ(column_view.size(), 3); // Should have 3 rows
+ ASSERT_EQ(column_view.type().id(), cudf::type_id::BOOL8); // Should be boolean type
+
+ // Export back to Arrow to verify data integrity
+ auto arrow_type = cudf_type_to_arrow_type(column_view.type().id());
+ auto exported_array = computed_column.export_to_arrow(arrow_type);
+
+ // Verify the computed values (id > 2)
+ // Original data: id=[1,2,3], expected result: [false,false,true]
+ auto result_array = std::static_pointer_cast<arrow::BooleanArray>(exported_array);
+ ASSERT_FALSE(result_array->Value(0)); // 1 > 2 = false
+ ASSERT_FALSE(result_array->Value(1)); // 2 > 2 = false
+ ASSERT_TRUE(result_array->Value(2)); // 3 > 2 = true
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_parquet_datasource.cpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_parquet_datasource.cpp
new file mode 100644
index 00000000..7e84b533
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/test_parquet_datasource.cpp
@@ -0,0 +1,36 @@
+// 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 "src/datasource.hpp"
+#include "src/test/util.hpp"
+
+struct ParquetDatasourceTest : testing::Test {
+ void SetUp() override {
+ }
+ void TearDown() override {
+ }
+};
+
+TEST_F(ParquetDatasourceTest, test_read_tables) {
+ const auto table_stream = create_tpcds_item_table_stream();
+ auto row_count = 0;
+
+ while (table_stream->has_next()) {
+ const auto& table_view = table_stream->next().view();
+ row_count += table_view.num_rows();
+ ASSERT_EQ(table_view.num_columns(), 22);
+ }
+ ASSERT_EQ(row_count, 102000);
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/test/util.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/util.hpp
new file mode 100644
index 00000000..5b51bff8
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/test/util.hpp
@@ -0,0 +1,105 @@
+// 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 <gtest/gtest.h>
+#include "src/datasource.hpp"
+#include "src/util/arrow.hpp"
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+// Helper function to create a TPCDS item table schema
+inline static std::shared_ptr<arrow::Schema> create_tpcds_item_table_schema() {
+ static std::shared_ptr<arrow::Schema> schema = arrow::schema({
+ arrow::field("i_item_sk", arrow::int64()), // item surrogate key
+ arrow::field("i_item_id", arrow::utf8()), // item business key
+ arrow::field("i_rec_start_date", arrow::date32()), // record start date
+ arrow::field("i_rec_end_date", arrow::date32()), // record end date
+ arrow::field("i_item_desc", arrow::utf8()), // item description
+ arrow::field("i_current_price", arrow::decimal128(7, 2)), // current price
+ arrow::field("i_wholesale_cost", arrow::decimal128(7, 2)), // wholesale cost
+ arrow::field("i_brand_id", arrow::int32()), // brand id
+ arrow::field("i_brand", arrow::utf8()), // brand name
+ arrow::field("i_class_id", arrow::int32()), // class id
+ arrow::field("i_class", arrow::utf8()), // class name
+ arrow::field("i_category_id", arrow::int32()), // category id
+ arrow::field("i_category", arrow::utf8()), // category name
+ arrow::field("i_manufact_id", arrow::int32()), // manufacturer id
+ arrow::field("i_manufact", arrow::utf8()), // manufacturer name
+ arrow::field("i_size", arrow::utf8()), // size
+ arrow::field("i_formulation", arrow::utf8()), // formulation
+ arrow::field("i_color", arrow::utf8()), // color
+ arrow::field("i_units", arrow::utf8()), // units
+ arrow::field("i_container", arrow::utf8()), // container
+ arrow::field("i_manager_id", arrow::int32()), // manager id
+ arrow::field("i_product_name", arrow::utf8()) // product name
+ });
+ return schema;
+}
+
+// Helper function to create a TPCDS item table stream
+inline static std::unique_ptr<TableStream> create_tpcds_item_table_stream() {
+ const auto schema = create_tpcds_item_table_schema();
+ const auto path = std::string{"../test/item.parquet"};
+ auto datasource = cudf::io::datasource::create(path);
+ auto table_stream = create_parquet_scan_table_stream(
+ std::move(datasource), 0, 1 << 30, schema, ExprBuilder::boolean_lit_expr_builder(true));
+ return std::move(table_stream);
+}
+
+// Helper function to create a simple Arrow schema
+inline static std::shared_ptr<arrow::Schema> create_simple_schema() {
+ return arrow::schema(
+ {arrow::field("id", arrow::int64()), arrow::field("name", arrow::utf8()),
+ arrow::field("value", arrow::float64())});
+}
+
+// Helper function to create a simple Arrow record batch
+inline static std::shared_ptr<arrow::RecordBatch> create_simple_record_batch() {
+ auto schema = create_simple_schema();
+
+ // Create arrays
+ auto id_builder = std::make_shared<arrow::Int64Builder>();
+ auto name_builder = std::make_shared<arrow::StringBuilder>();
+ auto value_builder = std::make_shared<arrow::DoubleBuilder>();
+
+ // Add data
+ ARROW_OK_OR_THROW(id_builder->AppendValues({1, 2, 3}), "Failed to append id values");
+ ARROW_OK_OR_THROW(
+ name_builder->AppendValues({"Alice", "Bob", "Charlie"}), "Failed to append name values");
+ ARROW_OK_OR_THROW(
+ value_builder->AppendValues({1.5, 2.5, 3.5}), "Failed to append value values");
+
+ // Finish arrays
+ auto id_array = ARROW_VALUE_OR_THROW(id_builder->Finish(), "Failed to finish id array");
+ auto name_array = ARROW_VALUE_OR_THROW(name_builder->Finish(), "Failed to finish name array");
+ auto value_array =
+ ARROW_VALUE_OR_THROW(value_builder->Finish(), "Failed to finish value array");
+
+ return arrow::RecordBatch::Make(schema, 3, {id_array, name_array, value_array});
+}
+
+// Helper function to create a simple Arrow array
+inline static std::shared_ptr<arrow::Array> create_simple_array() {
+ auto builder = std::make_shared<arrow::Int64Builder>();
+
+ // Add data
+ ARROW_OK_OR_THROW(builder->AppendValues({10, 20, 30, 40, 50}), "Failed to append array values");
+
+ // Finish array
+ return ARROW_VALUE_OR_THROW(builder->Finish(), "Failed to finish array");
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/util/arrow.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/util/arrow.hpp
new file mode 100644
index 00000000..7cfaa23d
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/util/arrow.hpp
@@ -0,0 +1,56 @@
+// 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
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+#define ARROW_OK_OR_THROW(result, msg) \
+ do { \
+ auto result__ = std::move(result); \
+ if (!result__.ok()) { \
+ auto ss__ = std::stringstream{}; \
+ ss__ << "arrow error({" << __func__ << "}): " << (msg); \
+ throw std::runtime_error{ss__.str()}; \
+ } \
+ } while (0)
+
+#define ARROW_VALUE_OR_THROW(result, msg) \
+ ({ \
+ auto result__ = std::move(result); \
+ if (!result__.ok()) { \
+ auto ss__ = std::stringstream{}; \
+ ss__ << "arrow error({" << __func__ << "}): " << (msg); \
+ throw std::runtime_error{ss__.str()}; \
+ } \
+ result__.MoveValueUnsafe(); \
+ })
+
+struct ArrowFFISchema : ArrowSchema {
+ ~ArrowFFISchema() {
+ if (this->release) {
+ this->release(this);
+ }
+ }
+};
+
+struct ArrowFFIArray : ArrowArray {
+ ~ArrowFFIArray() {
+ if (this->release) {
+ this->release(this);
+ }
+ }
+};
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/util/ffi.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/util/ffi.hpp
new file mode 100644
index 00000000..dd906fdf
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/util/ffi.hpp
@@ -0,0 +1,76 @@
+// 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
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+template <typename T>
+struct alignas(8) FFIObjWrapper {
+ T inner;
+
+ FFIObjWrapper(T inner) : inner(std::move(inner)) {
+ }
+};
+
+template <typename T>
+struct alignas(8) FFIResult {
+ bool ok;
+ T value;
+
+ explicit FFIResult(bool ok, T value) : ok(ok), value(std::move(value)) {
+ }
+};
+
+template <>
+struct alignas(8) FFIResult<void> {
+ explicit FFIResult(bool ok) : ok(ok) {
+ }
+ bool ok;
+};
+
+#define FFI_DECLARE_OBJ_WRAPPED_TYPE(name, ty) \
+ using name = FFIObjWrapper<ty>*; \
+ extern "C" void ffi_##name##_delete(name ptr);
+
+#define FFI_IMPLEMENT_OBJ_WRAPPED_TYPE(name, ty) \
+ extern "C" void ffi_##name##_delete(name ptr) { \
+ delete ptr; \
+ }
+
+#define CATCH_EXCEPTIONS_AND_RETURN(ret) \
+ catch (cudf::logic_error & err) { \
+ spdlog::error("cudf logic error({}): {}", __func__, err.what()); \
+ return (ret); \
+ } \
+ catch (cudf::data_type_error & err) { \
+ spdlog::error("cudf data type error({}): {}", __func__, err.what()); \
+ std::cerr << err.stacktrace(); \
+ return (ret); \
+ } \
+ catch (cudf::cuda_error & err) { \
+ spdlog::error("cudf cuda error({}): {}", __func__, err.what()); \
+ std::cerr << err.stacktrace(); \
+ return (ret); \
+ } \
+ catch (std::runtime_error & err) { \
+ spdlog::error("runtime error({}): {}", __func__, err.what()); \
+ return (ret); \
+ } \
+ catch (std::exception & err) { \
+ spdlog::error("exception({}): {}", __func__, err.what()); \
+ return (ret); \
+ }
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/src/util/types.hpp b/native-engine/auron-cudf-bridge/bridge-cpp/src/util/types.hpp
new file mode 100644
index 00000000..455d1744
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/src/util/types.hpp
@@ -0,0 +1,68 @@
+// Licensed to the 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
+
+// use precompiled headers
+#include "src/precompile/headers.hpp"
+
+/**
+ * Convert cudf type to corresponding Arrow type.
+ * @param cudf_type The cudf type_id to convert
+ * @return Shared pointer to Arrow DataType, or nullptr if unsupported
+ */
+std::shared_ptr<arrow::DataType> cudf_type_to_arrow_type(cudf::type_id cudf_type);
+
+// Implementation
+inline std::shared_ptr<arrow::DataType> cudf_type_to_arrow_type(cudf::type_id cudf_type) {
+ switch (cudf_type) {
+ case cudf::type_id::INT8:
+ return arrow::int8();
+ case cudf::type_id::INT16:
+ return arrow::int16();
+ case cudf::type_id::INT32:
+ return arrow::int32();
+ case cudf::type_id::INT64:
+ return arrow::int64();
+ case cudf::type_id::UINT8:
+ return arrow::uint8();
+ case cudf::type_id::UINT16:
+ return arrow::uint16();
+ case cudf::type_id::UINT32:
+ return arrow::uint32();
+ case cudf::type_id::UINT64:
+ return arrow::uint64();
+ case cudf::type_id::FLOAT32:
+ return arrow::float32();
+ case cudf::type_id::FLOAT64:
+ return arrow::float64();
+ case cudf::type_id::BOOL8:
+ return arrow::boolean();
+ case cudf::type_id::TIMESTAMP_DAYS:
+ return arrow::date32();
+ case cudf::type_id::TIMESTAMP_SECONDS:
+ return arrow::timestamp(arrow::TimeUnit::SECOND);
+ case cudf::type_id::TIMESTAMP_MILLISECONDS:
+ return arrow::timestamp(arrow::TimeUnit::MILLI);
+ case cudf::type_id::TIMESTAMP_MICROSECONDS:
+ return arrow::timestamp(arrow::TimeUnit::MICRO);
+ case cudf::type_id::TIMESTAMP_NANOSECONDS:
+ return arrow::timestamp(arrow::TimeUnit::NANO);
+ case cudf::type_id::STRING:
+ return arrow::utf8();
+ default:
+ return nullptr; // Unsupported type
+ }
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg.json b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg.json
new file mode 100644
index 00000000..444bbc01
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg.json
@@ -0,0 +1,29 @@
+{
+ "name": "auron-cudf-bridge",
+ "version": "0.1.0",
+ "dependencies": [
+ "arrow",
+ "spdlog",
+ "gtest",
+ {
+ "name": "cuda",
+ "platform": "!osx"
+ },
+ {
+ "name": "libcudf",
+ "platform": "!osx"
+ },
+ {
+ "name": "cuda-headers",
+ "platform": "osx"
+ },
+ {
+ "name": "libcudf-headers",
+ "platform": "osx"
+ },
+ {
+ "name": "librmm-headers",
+ "platform": "osx"
+ }
+ ]
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/get-cuda-headers.sh b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/get-cuda-headers.sh
new file mode 100644
index 00000000..e2ece63a
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/get-cuda-headers.sh
@@ -0,0 +1,28 @@
+#!/bin/bash
+
+
+set -e
+set -x
+
+echo "current directory: $PWD"
+
+INSTALL_PATH="$1"
+RPM_FILENAME="cuda-repo-rhel7-12-4-local-12.4.0_550.54.14-1.x86_64.rpm"
+RPM_URL="https://developer.download.nvidia.com/compute/cuda/12.4.0/local_installers/${RPM_FILENAME}"
+RPM_MD5="a9fb260d41093fb201cc77c1dc3a7ae9"
+
+xrpm() {
+ rpm2cpio "$1" | cpio -idmv
+}
+
+if [ -f "$RPM_FILENAME" ]; then
+ if [ "$(md5sum --quiet "$RPM_FILENAME")" != "$RPM_MD5" ]; then
+ rm -rf "$RPM_FILENAME"
+ fi
+fi
+wget -c "$RPM_URL"
+xrpm "$RPM_FILENAME"
+for x in ./var/cuda-*-local/*.rpm; do xrpm "$x"; done
+
+mkdir -p "$INSTALL_PATH/include/"
+cp -r "./usr/local/cuda"-*/include/* "$INSTALL_PATH/include/"
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/portfile.cmake b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/portfile.cmake
new file mode 100644
index 00000000..28739ad5
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/portfile.cmake
@@ -0,0 +1,7 @@
+set(ENV{INSTALL_PREFIX} "${CURRENT_PACKAGES_DIR}")
+
+vcpkg_execute_build_process(
+ COMMAND "sh" "${CMAKE_CURRENT_LIST_DIR}/get-cuda-headers.sh" "${CURRENT_PACKAGES_DIR}"
+ WORKING_DIRECTORY "${SOURCE_PATH}"
+ LOGNAME "build-${TARGET_TRIPLET}"
+)
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/vcpkg.json b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/vcpkg.json
new file mode 100644
index 00000000..f82b0eab
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/cuda-headers/vcpkg.json
@@ -0,0 +1,9 @@
+{
+ "name": "cuda-headers",
+ "version": "12.4.0",
+ "description": "CUDA headers",
+ "homepage": "https://developer.nvidia.com/cuda-downloads",
+ "license": "private",
+ "dependencies": [],
+ "supports": "linux | windows | osx"
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf-headers/portfile.cmake b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf-headers/portfile.cmake
new file mode 100644
index 00000000..0f159b83
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf-headers/portfile.cmake
@@ -0,0 +1,12 @@
+vcpkg_from_github(
+ OUT_SOURCE_PATH SOURCE_PATH
+ REPO rapidsai/cudf
+ REF "branch-25.06"
+ SHA512 e32b3ff160a46fcc612f63e3ce20ed04829646bf8478cf975c31a4d877bfd35f444698e3490721099ebab846bd78e21bfbdc4b56364c62a9fdfe32be5d8cee84
+ HEAD_REF main
+)
+
+set(ENV{INSTALL_PREFIX} "${CURRENT_PACKAGES_DIR}")
+file(
+ INSTALL "${SOURCE_PATH}/cpp/include"
+ DESTINATION "${CURRENT_PACKAGES_DIR}")
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf-headers/vcpkg.json b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf-headers/vcpkg.json
new file mode 100644
index 00000000..19dd6c19
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf-headers/vcpkg.json
@@ -0,0 +1,11 @@
+{
+ "name": "libcudf-headers",
+ "version": "25.6",
+ "description": "GPU DataFrame Library for Apache Arrow",
+ "homepage": "https://github.com/rapidsai/cudf",
+ "license": "Apache-2.0",
+ "dependencies": [
+ "vcpkg-cmake"
+ ],
+ "supports": "linux | windows | osx"
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf/portfile.cmake b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf/portfile.cmake
new file mode 100644
index 00000000..070d81c4
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf/portfile.cmake
@@ -0,0 +1,49 @@
+vcpkg_from_github(
+ OUT_SOURCE_PATH SOURCE_PATH
+ REPO rapidsai/cudf
+ REF "branch-25.06"
+ SHA512 e32b3ff160a46fcc612f63e3ce20ed04829646bf8478cf975c31a4d877bfd35f444698e3490721099ebab846bd78e21bfbdc4b56364c62a9fdfe32be5d8cee84
+ HEAD_REF main
+)
+
+# check for nvcc
+execute_process(
+ COMMAND nvcc --version
+ OUTPUT_VARIABLE NVCC_VERSION_OUTPUT
+ ERROR_VARIABLE NVCC_VERSION_ERROR
+ RESULT_VARIABLE NVCC_VERSION_RESULT
+)
+if(NOT ${NVCC_VERSION_RESULT} EQUAL 0)
+ message(FATAL_ERROR "execute nvcc --version failed: ${NVCC_VERSION_ERROR}")
+endif()
+
+set(BUILD_ARGS --disable_large_strings --cmake-args="-DCUDA_STATIC_RUNTIME=OFF -DBUILD_SHARED_LIBS=OFF")
+set(ENV{INSTALL_PREFIX} "${CURRENT_PACKAGES_DIR}")
+
+# build
+vcpkg_execute_required_process(
+ COMMAND "${SOURCE_PATH}/build.sh" libcudf ${BUILD_ARGS} -n
+ WORKING_DIRECTORY "${SOURCE_PATH}"
+ LOGNAME "build-${TARGET_TRIPLET}"
+)
+
+# patch for installation
+vcpkg_execute_required_process(
+ COMMAND sed -i "230,231s/^/#/" "./cpp/build/_deps/rapids-cmake-src/rapids-cmake/cpm/nvcomp.cmake"
+ WORKING_DIRECTORY "${SOURCE_PATH}"
+ LOGNAME "build-${TARGET_TRIPLET}"
+)
+file(MAKE_DIRECTORY "${CURRENT_PACKAGES_DIR}/usr/lib64")
+
+# install
+vcpkg_execute_required_process(
+ COMMAND "${SOURCE_PATH}/build.sh" libcudf ${BUILD_ARGS}
+ WORKING_DIRECTORY "${SOURCE_PATH}"
+ LOGNAME "build-${TARGET_TRIPLET}"
+)
+
+# fix some file conflicts
+file(RENAME "${CURRENT_PACKAGES_DIR}/include/zdict.h" "${CURRENT_PACKAGES_DIR}/include/cudf-zdict.h")
+file(RENAME "${CURRENT_PACKAGES_DIR}/include/zstd.h" "${CURRENT_PACKAGES_DIR}/include/cudf-zstd.h")
+file(RENAME "${CURRENT_PACKAGES_DIR}/include/zstd_errors.h" "${CURRENT_PACKAGES_DIR}/include/cudf-zstd_errors.h")
+
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf/vcpkg.json b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf/vcpkg.json
new file mode 100644
index 00000000..2115af52
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/libcudf/vcpkg.json
@@ -0,0 +1,12 @@
+{
+ "name": "libcudf",
+ "version": "25.6",
+ "description": "GPU DataFrame Library for Apache Arrow",
+ "homepage": "https://github.com/rapidsai/cudf",
+ "license": "Apache-2.0",
+ "dependencies": [
+ "vcpkg-cmake",
+ "cuda"
+ ],
+ "supports": "linux | windows | osx"
+}
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/librmm-headers/portfile.cmake b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/librmm-headers/portfile.cmake
new file mode 100644
index 00000000..06a21a90
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/librmm-headers/portfile.cmake
@@ -0,0 +1,13 @@
+vcpkg_from_github(
+ OUT_SOURCE_PATH SOURCE_PATH
+ REPO rapidsai/rmm
+ REF "branch-25.06"
+ SHA512 24c7b5c2ec9593b306c462deaab49f3ddd7fa9cb75ab86f331ec9436a2303351129621cc5cea1366366a53976296dccc2b38aa78607da5b1e4b32ae44bfbd7d9
+ HEAD_REF main
+)
+
+set(ENV{INSTALL_PREFIX} "${CURRENT_PACKAGES_DIR}")
+file(
+ INSTALL "${SOURCE_PATH}/cpp/include"
+ DESTINATION "${CURRENT_PACKAGES_DIR}"
+ PATTERN "include/doxygen_groups.h" EXCLUDE)
diff --git a/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/librmm-headers/vcpkg.json b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/librmm-headers/vcpkg.json
new file mode 100644
index 00000000..5ff6ac1e
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/bridge-cpp/vcpkg_ports/librmm-headers/vcpkg.json
@@ -0,0 +1,11 @@
+{
+ "name": "librmm-headers",
+ "version": "25.6",
+ "description": "RAPIDS Memory Manager",
+ "homepage": "https://github.com/rapidsai/librmm",
+ "license": "Apache-2.0",
+ "dependencies": [
+ "vcpkg-cmake"
+ ],
+ "supports": "linux | windows | osx"
+}
diff --git a/native-engine/auron-cudf-bridge/build.rs b/native-engine/auron-cudf-bridge/build.rs
new file mode 100644
index 00000000..da558685
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/build.rs
@@ -0,0 +1,60 @@
+// 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.
+
+use std::error::Error;
+
+fn main() -> Result<(), Box<dyn Error>> {
+ let manifest_dir = env!("CARGO_MANIFEST_DIR");
+ let cuda_path = env!("CUDA_PATH");
+ let vcpkg_installed_root_path = format!("{manifest_dir}/bridge-cpp/vcpkg_installed/x64-linux");
+
+ // dynamic link cuda
+ println!("cargo::rustc-link-search={cuda_path}/targets/x86_64-linux/lib");
+ println!("cargo::rustc-link-lib=dylib=rt");
+ println!("cargo::rustc-link-lib=dylib=cudart");
+ println!("cargo::rustc-link-lib=dylib=nvcomp");
+ println!("cargo::rustc-link-lib=dylib=rapids_logger");
+ println!("cargo::rustc-link-lib=dylib=kvikio");
+
+ // dynamic link libstdc++
+ println!("cargo::rustc-link-lib=dylib=stdc++");
+
+ // static link vcpkg
+ println!("cargo::rustc-link-search={vcpkg_installed_root_path}/lib/");
+ println!("cargo::rustc-link-search={vcpkg_installed_root_path}/lib64/");
+ println!("cargo::rustc-link-lib=static=arrow");
+ println!("cargo::rustc-link-lib=static=brotlicommon");
+ println!("cargo::rustc-link-lib=static=brotlidec");
+ println!("cargo::rustc-link-lib=static=brotlienc");
+ println!("cargo::rustc-link-lib=static=bz2");
+ println!("cargo::rustc-link-lib=static=crypto");
+ println!("cargo::rustc-link-lib=static=cudf");
+ println!("cargo::rustc-link-lib=static=fmt");
+ println!("cargo::rustc-link-lib=static=lz4");
+ println!("cargo::rustc-link-lib=static=nanoarrow");
+ println!("cargo::rustc-link-lib=static=parquet");
+ println!("cargo::rustc-link-lib=static=snappy");
+ println!("cargo::rustc-link-lib=static=spdlog");
+ println!("cargo::rustc-link-lib=static=ssl");
+ println!("cargo::rustc-link-lib=static=thrift");
+ println!("cargo::rustc-link-lib=static=z");
+ println!("cargo::rustc-link-lib=static=zstd");
+
+ // static link cudf-bridge
+ println!("cargo::rustc-link-search={manifest_dir}/bridge-cpp/build");
+ println!("cargo::rustc-link-lib=static=auron-cudf-bridge");
+ Ok(())
+}
+
diff --git a/native-engine/auron-cudf-bridge/src/datafusion/mod.rs b/native-engine/auron-cudf-bridge/src/datafusion/mod.rs
new file mode 100644
index 00000000..38a74e35
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/datafusion/mod.rs
@@ -0,0 +1,33 @@
+// 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.
+
+use std::pin::Pin;
+use arrow::datatypes::SchemaRef;
+use datafusion::error::Result;
+use futures::Stream;
+use crate::ffi::table::CudfTable;
+
+pub mod parquet_scan;
+
+pub trait CudfDataStream: Stream<Item=Result<CudfTable>> {
+ fn schema(&self) -> SchemaRef;
+}
+
+pub type SendableCudfDataStream = Pin<Box<dyn CudfDataStream + Send>>;
+
+pub trait CudfDataStreamProvider {
+ fn schema(&self) -> Result<SchemaRef>;
+ fn execute(&self) -> Result<SendableCudfDataStream>;
+}
\ No newline at end of file
diff --git a/native-engine/auron-cudf-bridge/src/datafusion/parquet_scan.rs b/native-engine/auron-cudf-bridge/src/datafusion/parquet_scan.rs
new file mode 100644
index 00000000..e5388fd5
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/datafusion/parquet_scan.rs
@@ -0,0 +1,166 @@
+// 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.
+
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use arrow::datatypes::SchemaRef;
+use arrow::ffi::FFI_ArrowSchema;
+use datafusion::error::Result;
+use futures::Stream;
+use crate::datafusion::{CudfDataStream, CudfDataStreamProvider, SendableCudfDataStream};
+use crate::ffi::expr::CudfExprBuilder;
+use crate::ffi::{ffi_create_parquet_scan_table_stream, ffi_table_stream_has_next, ffi_table_stream_next, AsRawPtrAddr};
+use crate::ffi::datasource::{CudfDataSourceProvider, IntoFFICudfDataSource};
+use crate::ffi::table::{CudfTable, CudfTableStream};
+
+pub struct CudfParquetScanStreamProvider {
+ datasource_provider: Arc<dyn CudfDataSourceProvider>,
+ path: String,
+ predicate_filter_expr: Arc<CudfExprBuilder>,
+ read_offset: usize,
+ read_size: usize,
+ file_size: Option<usize>,
+ schema: SchemaRef,
+}
+
+impl CudfDataStreamProvider for CudfParquetScanStreamProvider {
+ fn schema(&self) -> Result<SchemaRef> {
+ Ok(self.schema.clone())
+ }
+
+ fn execute(&self) -> Result<SendableCudfDataStream> {
+ let datasource = self.datasource_provider.provide(&self.path, self.file_size)?;
+ let mut ffi_datasource = datasource.into_ffi();
+ let mut ffi_schema = FFI_ArrowSchema::try_from(self.schema.as_ref())?;
+
+ let raw_table_stream = unsafe {
+ ffi_create_parquet_scan_table_stream(
+ &mut ffi_datasource,
+ self.read_offset,
+ self.read_size,
+ &mut ffi_schema,
+ self.predicate_filter_expr.as_raw_ptr_addr(),
+ )
+ .into_result(format!("error creating parquet scan table stream"))?
+ };
+ let table_stream = CudfTableStream::try_from_raw_table_stream(raw_table_stream)?;
+ Ok(Box::pin(TableCudfDataStream(table_stream, self.schema.clone())))
+ }
+}
+
+struct TableCudfDataStream(CudfTableStream, SchemaRef);
+
+impl CudfDataStream for TableCudfDataStream {
+ fn schema(&self) -> SchemaRef {
+ self.1.clone()
+ }
+}
+
+impl Stream for TableCudfDataStream {
+ type Item = Result<CudfTable>;
+
+ fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ unsafe fn sync_next(stream: &CudfTableStream) -> Result<Option<CudfTable>> {
+ let has_next = ffi_table_stream_has_next(stream.as_raw_ptr_addr())
+ .into_result(format!("TableStream.has_next() for parquet scan error"))?;
+ if !has_next {
+ return Ok(None);
+ }
+
+ let next = ffi_table_stream_next(stream.as_raw_ptr_addr())
+ .into_result(format!("TableStream.has_next() for parquet scan error"))?;
+ Ok(Some(CudfTable::from_raw_table(next)))
+ }
+ Poll::Ready(unsafe { sync_next(&self.0).transpose() })
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::sync::Arc;
+ use arrow::datatypes::{DataType, Field, Schema};
+ use crate::ffi::datasource::CudfLocalDataSourceProvider;
+ use crate::ffi::expr::{CudfExprBuilder, CudfScalar};
+
+ // Helper function to create TPCDS item table schema (matching C++ util.hpp)
+ fn create_tpcds_item_table_schema() -> SchemaRef {
+ Arc::new(Schema::new(vec![
+ Field::new("i_item_sk", DataType::Int64, false), // item surrogate key
+ Field::new("i_item_id", DataType::Utf8, false), // item business key
+ Field::new("i_rec_start_date", DataType::Date32, true), // record start date
+ Field::new("i_rec_end_date", DataType::Date32, true), // record end date
+ Field::new("i_item_desc", DataType::Utf8, true), // item description
+ Field::new("i_current_price", DataType::Decimal128(7, 2), true), // current price
+ Field::new("i_wholesale_cost", DataType::Decimal128(7, 2), true), // wholesale cost
+ Field::new("i_brand_id", DataType::Int32, true), // brand id
+ Field::new("i_brand", DataType::Utf8, true), // brand name
+ Field::new("i_class_id", DataType::Int32, true), // class id
+ Field::new("i_class", DataType::Utf8, true), // class name
+ Field::new("i_category_id", DataType::Int32, true), // category id
+ Field::new("i_category", DataType::Utf8, true), // category name
+ Field::new("i_manufact_id", DataType::Int32, true), // manufacturer id
+ Field::new("i_manufact", DataType::Utf8, true), // manufacturer name
+ Field::new("i_size", DataType::Utf8, true), // size
+ Field::new("i_formulation", DataType::Utf8, true), // formulation
+ Field::new("i_color", DataType::Utf8, true), // color
+ Field::new("i_units", DataType::Utf8, true), // units
+ Field::new("i_container", DataType::Utf8, true), // container
+ Field::new("i_manager_id", DataType::Int32, true), // manager id
+ Field::new("i_product_name", DataType::Utf8, true), // product name
+ ]))
+ }
+
+ // Helper function to create a boolean literal expression (true)
+ fn create_boolean_true_expr() -> Arc<CudfExprBuilder> {
+ let expr_builder = CudfExprBuilder::new();
+ let true_scalar = CudfScalar::new_bool(true);
+ expr_builder.add_literal(&true_scalar);
+ Arc::new(expr_builder)
+ }
+
+ #[test]
+ fn test_parquet_scan_read_tables() -> Result<()> {
+ let datasource_provider = Arc::new(CudfLocalDataSourceProvider);
+ let path = "test/item.parquet".to_string();
+ let schema = create_tpcds_item_table_schema();
+ let predicate_filter_expr = create_boolean_true_expr();
+
+ let provider = CudfParquetScanStreamProvider {
+ datasource_provider,
+ path,
+ predicate_filter_expr,
+ read_offset: 0,
+ read_size: 1 << 30, // 1GB
+ file_size: None, // Will be determined by the datasource provider
+ schema: schema.clone(),
+ };
+
+ // Execute the stream and collect results
+ let mut stream = provider.execute()?;
+ let mut total_row_count = 0;
+ let mut table_count = 0;
+
+ let noop_waker = futures::task::noop_waker();
+ let mut noop_ctx = Context::from_waker(&noop_waker);
+ while let Poll::Ready(Some(result)) = stream.as_mut().poll_next(&mut noop_ctx) {
+ let table = result?;
+ total_row_count += table.num_rows();
+ }
+ assert_eq!(total_row_count, 102400);
+ Ok(())
+ }
+}
diff --git a/native-engine/auron-cudf-bridge/src/ffi/column.rs b/native-engine/auron-cudf-bridge/src/ffi/column.rs
new file mode 100644
index 00000000..91ae58c2
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/ffi/column.rs
@@ -0,0 +1,72 @@
+// 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.
+
+use arrow::{
+ array::{make_array, ArrayRef},
+ datatypes::DataType,
+ error::Result,
+ ffi::{from_ffi_and_data_type, FFI_ArrowArray, FFI_ArrowSchema},
+};
+
+use crate::ffi::*;
+
+pub struct CudfColumn {
+ column: FFI_ColumnPtr,
+}
+
+impl AsRawPtrAddr for CudfColumn {
+ fn as_raw_ptr_addr(&self) -> usize {
+ self.column.as_raw_ptr_addr()
+ }
+}
+
+impl CudfColumn {
+ /// Creates a new `CudfColumn` from a raw `FFI_ColumnPtr`.
+ pub fn from_raw_column(column: FFI_ColumnPtr) -> Self {
+ Self { column }
+ }
+
+ /// Converts an Arrow `ArrayRef` to a `CudfColumn`.
+ pub fn from_arrow_array(array: &ArrayRef) -> Result<Self> {
+ let mut ffi_array = FFI_ArrowArray::empty();
+ let mut ffi_schema = FFI_ArrowSchema::try_from(array.data_type().clone())?;
+ let raw_column = unsafe {
+ ffi_column_from_arrow(&mut ffi_schema, &mut ffi_array)
+ .into_result(format!("error converting arrow array to cudf::column"))?
+ };
+ Ok(Self::from_raw_column(raw_column))
+ }
+
+ /// Converts a `CudfColumn` to an Arrow `ArrayRef`.
+ pub fn to_arrow_array(&self, data_type: &DataType) -> Result<ArrayRef> {
+ let mut ffi_schema = FFI_ArrowSchema::try_from(data_type.clone())?;
+ let mut ffi_array = FFI_ArrowArray::empty();
+ let array_data = unsafe {
+ ffi_column_to_arrow(self.as_raw_ptr_addr(), &mut ffi_schema, &mut ffi_array)
+ .into_result(format!("error converting cudf::column to arrow array"))?;
+ from_ffi_and_data_type(ffi_array, data_type.clone())?
+ };
+ Ok(make_array(array_data))
+ }
+}
+
+extern "C" {
+ fn ffi_column_from_arrow(schema: *mut FFI_ArrowSchema, array: *mut FFI_ArrowArray) -> FFI_ColumnResult;
+ fn ffi_column_to_arrow(
+ column_ptr_addr: usize,
+ ffi_schema: *mut FFI_ArrowSchema,
+ ffi_array: *mut FFI_ArrowArray,
+ ) -> FFI_VoidResult;
+}
diff --git a/native-engine/auron-cudf-bridge/src/ffi/datasource.rs b/native-engine/auron-cudf-bridge/src/ffi/datasource.rs
new file mode 100644
index 00000000..7d808e69
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/ffi/datasource.rs
@@ -0,0 +1,173 @@
+// 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.
+
+use std::{
+ fs::File,
+ io::{Read, Seek},
+};
+use std::sync::Arc;
+use arrow::ffi::FFI_ArrowSchema;
+use parquet::file::reader::Length;
+use datafusion::common::Result;
+use datafusion_ext_commons::hadoop_fs::{FsDataInputWrapper, FsProvider};
+use crate::ffi::{FFI_CudfDataSource, FFI_TableStreamPtrResult, FFI_USizeResult};
+
+pub trait CudfDataSource {
+ fn data_size(&self) -> usize;
+ fn read(&mut self, offset: usize, buf: &mut [u8]) -> Result<usize>;
+}
+
+pub trait IntoFFICudfDataSource {
+ fn into_ffi(self) -> FFI_CudfDataSource;
+}
+
+impl IntoFFICudfDataSource for Box<dyn CudfDataSource> {
+ fn into_ffi(self) -> FFI_CudfDataSource {
+ extern "C" fn dtor(this: *mut FFI_CudfDataSource) {
+ log::warn!("XXX FFI_CudfDataSource dtor");
+ unsafe {
+ let this_mut = this.as_mut().expect("FFI_CudfDataSource is null");
+ let raw_mut = this_mut
+ .raw
+ .as_mut()
+ .expect("FFI_CudfDataSource.raw is null");
+ let _ = Box::from_raw(raw_mut as *mut Box<dyn CudfDataSource>);
+ }
+ }
+
+ extern "C" fn data_size(this: *const FFI_CudfDataSource) -> usize {
+ log::warn!("XXX FFIDataSource.data_size");
+ unsafe {
+ let this_ref = this.as_ref().expect("FFI_CudfDataSource is null");
+ let raw_ref = this_ref
+ .raw
+ .as_ref()
+ .expect("FFI_CudfDataSource.raw is null");
+ raw_ref.data_size()
+ }
+ }
+
+ extern "C" fn read(
+ this: *mut FFI_CudfDataSource,
+ offset: usize,
+ buf: *mut u8,
+ buf_size: usize,
+ ) -> FFI_USizeResult {
+ log::warn!("XXX FFIDataSource.read offset={offset}, buf_size={buf_size}");
+ unsafe {
+ let this_mut = this.as_mut().expect("FFI_CudfDataSource is null");
+ let raw_mut = this_mut
+ .raw
+ .as_mut()
+ .expect("FFI_CudfDataSource.raw is null");
+ let buf_mut = std::slice::from_raw_parts_mut(buf, buf_size);
+ match raw_mut.read(offset, buf_mut) {
+ Ok(read_size) => FFI_USizeResult {
+ ok: true,
+ value: read_size,
+ },
+ Err(_err) => FFI_USizeResult {
+ ok: false,
+ value: 0,
+ },
+ }
+ }
+ }
+
+ FFI_CudfDataSource {
+ dtor,
+ data_size,
+ read,
+ raw: Box::into_raw(Box::new(self)),
+ }
+ }
+}
+
+pub trait CudfDataSourceProvider {
+ fn provide(&self, path: &str, file_size: Option<usize>) -> Result<Box<dyn CudfDataSource>>;
+}
+
+pub struct CudfLocalDataSourceProvider;
+pub struct CudfFSDataInputSourceProvider(Arc<FsProvider>);
+
+impl CudfDataSourceProvider for CudfLocalDataSourceProvider {
+ fn provide(&self, path: &str, file_size: Option<usize>) -> Result<Box<dyn CudfDataSource>> {
+ struct DataSourceImpl(File, usize);
+ impl CudfDataSource for DataSourceImpl {
+ fn data_size(&self) -> usize {
+ self.1
+ }
+
+ fn read(&mut self, offset: usize, buf: &mut [u8]) -> Result<usize> {
+ self.0.seek(std::io::SeekFrom::Start(offset as u64))?;
+ let read_size = self.0.read(buf)?;
+ Ok(read_size)
+ }
+ }
+
+ let file = File::open(path)?;
+ let file_size = file_size.unwrap_or_else(|| file.len() as usize);
+ Ok(Box::new(DataSourceImpl(file, file_size)))
+ }
+}
+
+impl CudfDataSourceProvider for CudfFSDataInputSourceProvider {
+ fn provide(&self, path: &str, file_size: Option<usize>) -> Result<Box<dyn CudfDataSource>> {
+ struct DataSourceImpl(Arc<FsDataInputWrapper>, usize);
+ impl CudfDataSource for DataSourceImpl {
+ fn data_size(&self) -> usize {
+ self.1
+ }
+
+ fn read(&mut self, offset: usize, buf: &mut [u8]) -> Result<usize> {
+ let read_len = buf.len().min(self.1.saturating_sub(offset));
+ self.0.read_fully(offset as u64, &mut buf[..read_len])?;
+ Ok(read_len)
+ }
+ }
+
+ let fs = self.0.provide(path)?;
+ let input = fs.open(path)?;
+ let file_size = file_size
+ .expect("CudfFSDataInputSourceProvider does not support empty file size");
+ Ok(Box::new(DataSourceImpl(input, file_size)))
+ }
+}
+
+
+#[repr(C, align(8))]
+#[allow(non_camel_case_types)]
+pub struct FFI_CudfDataSource {
+ pub dtor: extern "C" fn(this: *mut FFI_CudfDataSource) -> (),
+ pub data_size: extern "C" fn(this: *const FFI_CudfDataSource) -> usize,
+ pub read: extern "C" fn(
+ this: *mut FFI_CudfDataSource,
+ offset: usize,
+ buf: *mut u8,
+ buf_size: usize,
+ ) -> FFI_USizeResult,
+
+ pub raw: *mut Box<dyn CudfDataSource>,
+}
+
+extern "C" {
+ fn ffi_create_parquet_scan_table_stream(
+ datasource: *mut FFI_CudfDataSource,
+ read_offset: usize,
+ read_size: usize,
+ ffi_read_schema: *mut FFI_ArrowSchema,
+ predicate_filter_expr_addr_ptr: usize,
+ ) -> FFI_TableStreamPtrResult;
+}
\ No newline at end of file
diff --git a/native-engine/auron-cudf-bridge/src/ffi/expr.rs b/native-engine/auron-cudf-bridge/src/ffi/expr.rs
new file mode 100644
index 00000000..c6f1107f
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/ffi/expr.rs
@@ -0,0 +1,222 @@
+// 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.
+
+use std::{any::TypeId, ffi::CString, str::FromStr};
+
+use num::ToPrimitive;
+
+use crate::ffi::*;
+
+pub type CudfExprOperator = FFI_AstOperator;
+
+pub struct CudfExprBuilder {
+ expr_builder: FFI_ExprBuilderPtr,
+}
+
+impl AsRawPtrAddr for CudfExprBuilder {
+ fn as_raw_ptr_addr(&self) -> usize {
+ self.expr_builder.as_raw_ptr_addr()
+ }
+}
+
+impl CudfExprBuilder {
+ pub fn new() -> Self {
+ Self {
+ expr_builder: unsafe { ffi_new_expr_builder() },
+ }
+ }
+
+ pub fn add_literal(&self, scalar: &CudfScalar) -> usize {
+ unsafe { ffi_expr_builder_add_literal(self.as_raw_ptr_addr(), scalar.as_raw_ptr_addr()) }
+ }
+
+ pub fn add_column_ref(&self, column_idx: usize) -> usize {
+ unsafe { ffi_expr_builder_add_column_reference(self.as_raw_ptr_addr(), column_idx) }
+ }
+
+ pub fn add_column_name_ref(&self, name: &str) -> usize {
+ let name_c_str = CString::from_str(name).unwrap();
+ unsafe {
+ ffi_expr_builder_add_column_name_reference(self.as_raw_ptr_addr(), name_c_str.as_ptr())
+ }
+ }
+
+ pub fn add_unary(&self, arg1_expr_id: usize, op: CudfExprOperator) -> usize {
+ unsafe { ffi_expr_builder_add_unary_expr(self.as_raw_ptr_addr(), arg1_expr_id, op) }
+ }
+
+ pub fn add_binary(
+ &self,
+ arg1_expr_id: usize,
+ arg2_expr_id: usize,
+ op: CudfExprOperator,
+ ) -> usize {
+ unsafe {
+ ffi_expr_builder_add_binary_expr(self.as_raw_ptr_addr(), arg1_expr_id, arg2_expr_id, op)
+ }
+ }
+}
+
+pub struct CudfScalar {
+ scalar: FFI_ScalarPtr,
+}
+
+impl AsRawPtrAddr for CudfScalar {
+ fn as_raw_ptr_addr(&self) -> usize {
+ self.scalar.as_raw_ptr_addr()
+ }
+}
+
+impl CudfScalar {
+ pub fn new_prim<T: Copy + ToPrimitive + 'static>(value: T) -> Self {
+ let typeid = TypeId::of::<T>();
+ let scalar = unsafe {
+ match () {
+ _ if typeid == TypeId::of::<i8>() => ffi_i8_scalar(value.to_i8().unwrap()),
+ _ if typeid == TypeId::of::<i16>() => ffi_i16_scalar(value.to_i16().unwrap()),
+ _ if typeid == TypeId::of::<i32>() => ffi_i32_scalar(value.to_i32().unwrap()),
+ _ if typeid == TypeId::of::<i64>() => ffi_i64_scalar(value.to_i64().unwrap()),
+ _ if typeid == TypeId::of::<u8>() => ffi_u8_scalar(value.to_u8().unwrap()),
+ _ if typeid == TypeId::of::<u16>() => ffi_u16_scalar(value.to_u16().unwrap()),
+ _ if typeid == TypeId::of::<u32>() => ffi_u32_scalar(value.to_u32().unwrap()),
+ _ if typeid == TypeId::of::<u64>() => ffi_u64_scalar(value.to_u64().unwrap()),
+ _ if typeid == TypeId::of::<f32>() => ffi_f32_scalar(value.to_f32().unwrap()),
+ _ if typeid == TypeId::of::<f64>() => ffi_f64_scalar(value.to_f64().unwrap()),
+ _ => unreachable!("unimplemented fixed withd scalar"),
+ }
+ };
+ Self { scalar }
+ }
+
+ pub fn new_string(value: &str) -> Self {
+ let c_str = CString::from_str(value).unwrap();
+ let scalar = unsafe { ffi_string_scalar(c_str.as_ptr()) };
+ Self { scalar }
+ }
+
+ pub fn new_bool(value: bool) -> Self {
+ let scalar = unsafe { ffi_i8_scalar(if value { 1 } else { 0 }) };
+ Self { scalar }
+ }
+}
+
+#[repr(i32)]
+#[allow(non_camel_case_types)]
+#[allow(unused)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum FFI_AstOperator {
+ // this enum is copied from <cudf/ast/expressions.hpp>
+ // Binary operators
+ ADD = 0, // < operator +
+ SUB, // < operator -
+ MUL, // < operator *
+ DIV, // < operator / using common type of lhs and rhs
+ TRUE_DIV, // < operator / after promoting type to floating point
+ FLOOR_DIV, // < operator / after promoting to 64 bit floating point and then
+ // < flooring the result
+ MOD, // < operator %
+ PYMOD, // < operator % using Python's sign rules for negatives
+ POW, // < lhs ^ rhs
+ EQUAL, // < operator ==
+ NULL_EQUAL, /* < operator == with Spark rules: NULL_EQUAL(null, null) is true,
+ * NULL_EQUAL(null, */
+ // < valid) is false, and
+ // < NULL_EQUAL(valid, valid) == EQUAL(valid, valid)
+ NOT_EQUAL, // < operator !=
+ LESS, // < operator <
+ GREATER, // < operator >
+ LESS_EQUAL, // < operator <=
+ GREATER_EQUAL, // < operator >=
+ BITWISE_AND, // < operator &
+ BITWISE_OR, // < operator |
+ BITWISE_XOR, // < operator ^
+ LOGICAL_AND, // < operator &&
+ NULL_LOGICAL_AND, // < operator && with Spark rules: NULL_LOGICAL_AND(null, null) is null,
+ // < NULL_LOGICAL_AND(null, true) is
+ // < null, NULL_LOGICAL_AND(null, false) is false, and NULL_LOGICAL_AND(valid,
+ // < valid) == LOGICAL_AND(valid, valid)
+ LOGICAL_OR, // < operator ||
+ NULL_LOGICAL_OR, // < operator || with Spark rules: NULL_LOGICAL_OR(null, null) is null,
+ // < NULL_LOGICAL_OR(null, true) is true,
+ // < NULL_LOGICAL_OR(null, false) is null, and NULL_LOGICAL_OR(valid, valid) ==
+ // < LOGICAL_OR(valid, valid)
+ // Unary operators
+ IDENTITY, // < Identity function
+ IS_NULL, // < Check if operand is null
+ SIN, // < Trigonometric sine
+ COS, // < Trigonometric cosine
+ TAN, // < Trigonometric tangent
+ ARCSIN, // < Trigonometric sine inverse
+ ARCCOS, // < Trigonometric cosine inverse
+ ARCTAN, // < Trigonometric tangent inverse
+ SINH, // < Hyperbolic sine
+ COSH, // < Hyperbolic cosine
+ TANH, // < Hyperbolic tangent
+ ARCSINH, // < Hyperbolic sine inverse
+ ARCCOSH, // < Hyperbolic cosine inverse
+ ARCTANH, // < Hyperbolic tangent inverse
+ EXP, // < Exponential (base e, Euler number)
+ LOG, // < Natural Logarithm (base e)
+ SQRT, // < Square-root (x^0.5)
+ CBRT, // < Cube-root (x^(1.0/3))
+ CEIL, // < Smallest integer value not less than arg
+ FLOOR, // < largest integer value not greater than arg
+ ABS, // < Absolute value
+ RINT, // < Rounds the floating-point argument arg to an integer value
+ BIT_INVERT, // < Bitwise Not (~)
+ NOT, // < Logical Not (!)
+ CAST_TO_INT64, // < Cast value to int64_t
+ CAST_TO_UINT64, // < Cast value to uint64_t
+ CAST_TO_FLOAT64, // < Cast value to double
+}
+
+extern "C" {
+ pub fn ffi_new_expr_builder() -> FFI_ExprBuilderPtr;
+ pub fn ffi_i8_scalar(value: i8) -> FFI_ScalarPtr;
+ pub fn ffi_i16_scalar(value: i16) -> FFI_ScalarPtr;
+ pub fn ffi_i32_scalar(value: i32) -> FFI_ScalarPtr;
+ pub fn ffi_i64_scalar(value: i64) -> FFI_ScalarPtr;
+ pub fn ffi_u8_scalar(value: u8) -> FFI_ScalarPtr;
+ pub fn ffi_u16_scalar(value: u16) -> FFI_ScalarPtr;
+ pub fn ffi_u32_scalar(value: u32) -> FFI_ScalarPtr;
+ pub fn ffi_u64_scalar(value: u64) -> FFI_ScalarPtr;
+ pub fn ffi_f32_scalar(value: f32) -> FFI_ScalarPtr;
+ pub fn ffi_f64_scalar(value: f64) -> FFI_ScalarPtr;
+ pub fn ffi_string_scalar(value: *const std::ffi::c_char) -> FFI_ScalarPtr;
+
+ pub fn ffi_expr_builder_add_literal(
+ expr_builder_ptr_addr: usize,
+ scalar_ptr_addr: usize,
+ ) -> usize;
+ pub fn ffi_expr_builder_add_column_reference(
+ expr_builder_ptr_addr: usize,
+ column_idx: usize,
+ ) -> usize;
+ pub fn ffi_expr_builder_add_column_name_reference(
+ expr_builder_ptr_addr: usize,
+ name: *const std::ffi::c_char,
+ ) -> usize;
+ pub fn ffi_expr_builder_add_unary_expr(
+ expr_builder_ptr_addr: usize,
+ arg1_expr_id: usize,
+ op: FFI_AstOperator,
+ ) -> usize;
+ pub fn ffi_expr_builder_add_binary_expr(
+ expr_builder_ptr_addr: usize,
+ arg1_expr_id: usize,
+ arg2_expr_id: usize,
+ op: FFI_AstOperator,
+ ) -> usize;
+}
\ No newline at end of file
diff --git a/native-engine/auron-cudf-bridge/src/ffi/mod.rs b/native-engine/auron-cudf-bridge/src/ffi/mod.rs
new file mode 100644
index 00000000..a5ea0b9d
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/ffi/mod.rs
@@ -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.
+
+pub mod column;
+pub mod expr;
+pub mod datasource;
+pub mod table;
+
+use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
+use paste::paste;
+use crate::ffi::datasource::CudfDataSource;
+
+pub trait AsRawPtrAddr {
+ fn as_raw_ptr_addr(&self) -> usize;
+}
+
+macro_rules! define_ffi_obj_wrapped_type {
+ ($tyname:ident) => {
+ #[repr(C, align(8))]
+ #[allow(non_camel_case_types)]
+ pub struct $tyname {
+ raw_ptr_addr: usize,
+ }
+
+ impl $tyname {
+ #[allow(unused)]
+ pub fn as_raw_ptr_addr(&self) -> usize {
+ return self.raw_ptr_addr;
+ }
+ }
+
+ impl Drop for $tyname {
+ fn drop(&mut self) {
+ unsafe {
+ paste! {
+ extern "C" {
+ fn [<ffi_ $tyname _delete>] (raw_ptr_addr: usize) -> ();
+ }
+ [<ffi_ $tyname _delete>] (self.raw_ptr_addr);
+ }
+ }
+ }
+ }
+ };
+}
+
+macro_rules! define_ffi_result_type {
+ ($tyname:ident, $t:ty) => {
+ #[repr(C, align(8))]
+ #[allow(non_camel_case_types)]
+ pub struct $tyname {
+ pub ok: bool,
+ pub value: $t,
+ }
+
+ impl $tyname {
+ #[allow(unused)]
+ pub fn into_result(self, errmsg: String) -> arrow::error::Result<$t> {
+ if !self.ok {
+ return Err(arrow::error::ArrowError::ComputeError(errmsg));
+ }
+ Ok(self.value)
+ }
+ }
+ };
+}
+
+define_ffi_obj_wrapped_type!(FFI_TableStreamPtr);
+define_ffi_obj_wrapped_type!(FFI_TablePtr);
+define_ffi_obj_wrapped_type!(FFI_ColumnPtr);
+define_ffi_obj_wrapped_type!(FFI_ScalarPtr);
+define_ffi_obj_wrapped_type!(FFI_ExprBuilderPtr);
+
+define_ffi_result_type!(FFI_VoidResult, ());
+define_ffi_result_type!(FFI_BoolResult, bool);
+define_ffi_result_type!(FFI_TableStreamPtrResult, FFI_TableStreamPtr);
+define_ffi_result_type!(FFI_TableResult, FFI_TablePtr);
+define_ffi_result_type!(FFI_ColumnResult, FFI_ColumnPtr);
+define_ffi_result_type!(FFI_USizeResult, usize);
+
+pub fn slice_to_const_ptr_and_len<T>(vec: &[T]) -> (*const T, usize) {
+ let len = vec.len();
+ let ptr = vec.as_ptr();
+ (ptr, len)
+}
+
+pub fn slice_to_mut_ptr_and_len<T>(vec: &mut [T]) -> (*mut T, usize) {
+ let len = vec.len();
+ let ptr = vec.as_mut_ptr();
+ (ptr, len)
+}
diff --git a/native-engine/auron-cudf-bridge/src/ffi/table.rs b/native-engine/auron-cudf-bridge/src/ffi/table.rs
new file mode 100644
index 00000000..0854b5da
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/ffi/table.rs
@@ -0,0 +1,147 @@
+// 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.
+
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use arrow::{
+ array::{make_array, AsArray, RecordBatch},
+ datatypes::{DataType, SchemaRef},
+ ffi::{from_ffi_and_data_type, FFI_ArrowArray, FFI_ArrowSchema},
+};
+use datafusion::error::Result;
+use futures::Stream;
+use datafusion_ext_commons::df_execution_err;
+use crate::ffi::*;
+
+pub struct CudfTableStream {
+ table_stream: FFI_TableStreamPtr,
+}
+
+impl AsRawPtrAddr for CudfTableStream {
+ fn as_raw_ptr_addr(&self) -> usize {
+ self.table_stream.as_raw_ptr_addr()
+ }
+}
+
+impl CudfTableStream {
+ pub fn try_from_raw_table_stream(table_stream: FFI_TableStreamPtr) -> Result<Self> {
+ if table_stream.as_raw_ptr_addr() == 0 {
+ return df_execution_err!("table stream is null");
+ }
+ Ok(Self { table_stream })
+ }
+
+ fn next_impl(&mut self) -> Result<Option<CudfTable>> {
+ let has_next = unsafe {
+ ffi_table_stream_has_next(self.as_raw_ptr_addr())
+ .into_result(format!("error invoking table_stream.has_next()"))?
+ };
+ if has_next {
+ let raw_table = unsafe {
+ ffi_table_stream_next(self.as_raw_ptr_addr())
+ .into_result(format!("error invoking table_stream.next()"))?
+ };
+ return Ok(Some(CudfTable::from_raw_table(raw_table)));
+ }
+ Ok(None)
+ }
+}
+
+impl Iterator for CudfTableStream {
+ type Item = Result<CudfTable>;
+
+ fn next(&mut self) -> Option<Self::Item> {
+ self.next_impl().transpose()
+ }
+}
+
+pub struct CudfTable {
+ table: FFI_TablePtr,
+}
+
+impl AsRawPtrAddr for CudfTable {
+ fn as_raw_ptr_addr(&self) -> usize {
+ self.table.as_raw_ptr_addr()
+ }
+}
+
+impl Stream for CudfTableStream {
+ type Item = Result<CudfTable>;
+
+ fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+ // TODO: support async polling in the future
+ Poll::Ready(self.as_mut().next_impl().transpose())
+ }
+}
+
+impl CudfTable {
+ pub fn from_raw_table(table: FFI_TablePtr) -> Self {
+ Self { table }
+ }
+
+ pub fn from_arrow_record_batch
+ (batch: &RecordBatch) -> Result<Self> {
+ let mut ffi_array = FFI_ArrowArray::empty();
+ let mut ffi_schema = FFI_ArrowSchema::try_from(batch.schema().as_ref().clone())?;
+ let raw_table = unsafe {
+ ffi_table_from_arrow(&mut ffi_schema, &mut ffi_array)
+ .into_result(format!("error converting arrow record batch to cudf::table"))?
+ };
+ Ok(Self::from_raw_table(raw_table))
+ }
+
+ pub fn to_arrow_record_batch(&self, schema: SchemaRef) -> Result<RecordBatch> {
+ let schema_type = DataType::Struct(schema.fields().clone());
+ let mut ffi_schema = FFI_ArrowSchema::try_from(schema_type.clone())?;
+ let mut ffi_array = FFI_ArrowArray::empty();
+ let array_data = unsafe {
+ ffi_table_to_arrow(self.as_raw_ptr_addr(), &mut ffi_schema, &mut ffi_array)
+ .into_result(format!("error converting cudf::table to arrow record batch"))?;
+ from_ffi_and_data_type(ffi_array, schema_type)?
+ };
+ Ok(RecordBatch::from(make_array(array_data).as_struct().clone()))
+ }
+
+ pub fn num_rows(&self) -> usize {
+ unsafe { ffi_table_num_rows(self.as_raw_ptr_addr()) }
+ }
+
+ pub fn num_columns(&self) -> usize {
+ unsafe { ffi_table_num_columns(self.as_raw_ptr_addr()) }
+ }
+}
+
+extern "C" {
+ fn ffi_table_stream_has_next(table_stream_ptr_addr: usize) -> FFI_BoolResult;
+ fn ffi_table_stream_next(table_stream_ptr_addr: usize) -> FFI_TableResult;
+
+ fn ffi_table_from_arrow(
+ schema: *mut FFI_ArrowSchema,
+ array: *mut FFI_ArrowArray,
+ ) -> FFI_TableResult;
+
+ fn ffi_table_to_arrow(
+ table_ptr_addr: usize,
+ ffi_schema: *mut FFI_ArrowSchema,
+ ffi_array: *mut FFI_ArrowArray,
+ ) -> FFI_VoidResult;
+
+ fn ffi_table_num_rows(table_ptr_addr: usize) -> usize;
+ fn ffi_table_num_columns(table_ptr_addr: usize) -> usize;
+ fn ffi_table_compute_column(
+ table_ptr_addr: usize, expr_builder_ptr_addr: usize) -> FFI_ColumnResult;
+ fn ffi_table_apply_boolean_mask(
+ table_ptr_addr: usize, boolean_mask_column_ptr_addr: usize) -> FFI_TableResult;
+}
diff --git a/native-engine/auron-cudf-bridge/src/lib.rs b/native-engine/auron-cudf-bridge/src/lib.rs
new file mode 100644
index 00000000..8f3a4d5e
--- /dev/null
+++ b/native-engine/auron-cudf-bridge/src/lib.rs
@@ -0,0 +1,17 @@
+// 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.
+
+pub mod ffi;
+pub mod datafusion;
diff --git a/native-engine/auron-cudf-bridge/test/item.parquet b/native-engine/auron-cudf-bridge/test/item.parquet
new file mode 100644
index 00000000..692abad7
Binary files /dev/null and b/native-engine/auron-cudf-bridge/test/item.parquet differ