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:52 UTC
(auron) 01/01: init gpu
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