Add cucascade::io::datasource engine wrappers and Python bindings

cmake/cuCascadeConfig.cmake.in

@@ -38,6 +38,7 @@ if(cuCascade_io_FOUND AND ("io" IN_LIST cuCascade_FIND_COMPONENTS
   pkg_check_modules(LIBURING REQUIRED IMPORTED_TARGET liburing)
   pkg_check_modules(CURL REQUIRED IMPORTED_TARGET libcurl)
   find_dependency(OpenSSL REQUIRED)
+  find_dependency(kvikio REQUIRED CONFIG)
 
   # Best-effort check that the moodycamel headers the installed io headers
   # include are reachable. They are not shipped (the consumer provides the same

include/cucascade/cudf/datasource.hpp

@@ -20,11 +20,13 @@
 
 #include <cucascade/io/cache/prefetching_cache.hpp>
 #include <cucascade/io/io_context.hpp>
+#include <cucascade/io/types.hpp>
 
 #include <cudf/io/datasource.hpp>
 #include <cudf/io/text/byte_range_info.hpp>
 
 #include <span>
+#include <vector>
 
 namespace cucascade::io {
 
@@ -93,16 +95,13 @@ class datasource : public cudf::io::datasource {
 
   std::unique_ptr<datasource::buffer> device_read(size_t offset,
                                                   size_t size,
-                                                  rmm::cuda_stream_view stream) override;
-  size_t device_read(size_t offset,
-                     size_t size,
-                     uint8_t* dst,
-                     rmm::cuda_stream_view stream) override;
+                                                  cuda::stream_ref stream) override;
+  size_t device_read(size_t offset, size_t size, uint8_t* dst, cuda::stream_ref stream) override;
 
   std::future<size_t> device_read_async(size_t offset,
                                         size_t size,
                                         uint8_t* dst,
-                                        rmm::cuda_stream_view stream) override;
+                                        cuda::stream_ref stream) override;
 
   // ---- Advisory IO ---------------------------------------------------------
 
@@ -138,6 +137,21 @@ class datasource : public cudf::io::datasource {
   /// @c disposable call at consume time.
   void fadvise(std::span<const cudf::io::text::byte_range_info> ranges, std::optional<int> dev_id);
 
+  /**
+   * @brief Submit all byte ranges as a single vectorized host read.
+   *
+   * Uses the ioctx's scatter-read backend to fetch all segments in as few
+   * HTTP requests as possible, writing each range directly into the
+   * caller-supplied destination pointer in its @p segment.
+   *
+   * @param segments File offsets, sizes, and destination buffers.
+   * @return A future that resolves when every segment has been written.
+   */
+  [[nodiscard]] std::future<size_t> host_read_ranges_async(std::span<io_object_segment> segments);
+
+  [[nodiscard]] std::future<size_t> host_read_ranges_async(
+    std::vector<io_object_segment>& segments);
+
   void prefetch(cache::prefetching_stage site);
 
  private:

include/cucascade/cudf/rest_datasource_engine.hpp

@@ -0,0 +1,98 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cucascade/cudf/datasource.hpp>
+#include <cucascade/memory/fixed_size_host_memory_resource.hpp>
+#include <cucascade/memory/memory_reservation_manager.hpp>
+#include <cucascade/memory/numa_region_pinned_host_allocator.hpp>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+
+namespace cucascade::io {
+
+class ioctx;
+
+/**
+ * @brief Self-contained REST datasource engine for S3/HTTP object-store reads.
+ *
+ * Owns a NUMA-local pinned host staging pool and a pool of libcurl reactor
+ * threads backed by SigV4 presigned-URL signing. Callers open individual
+ * @c datasource instances via @c open(); each datasource shares the engine's
+ * @c ioctx and memory pool but carries its own per-scan @c prefetching_handle.
+ *
+ * The engine must outlive every datasource it produces.
+ *
+ * @code{.cpp}
+ * auto engine = cucascade::io::rest_datasource_engine::make_s3(...);
+ * auto ds = engine->open("s3://my-bucket/data/lineitem.parquet");
+ * ds->fadvise(byte_ranges, device_id);
+ * @endcode
+ */
+class rest_datasource_engine {
+ public:
+  static constexpr std::size_t default_block_size    = 1UL << 20;
+  static constexpr std::size_t default_pool_capacity = 20UL * 128UL * (1UL << 20);
+
+  /**
+   * @brief Construct an S3-backed REST engine with SigV4 presigned-URL signing.
+   *
+   * Credentials are static for the lifetime of the engine. For short-lived STS
+   * tokens, reconstruct the engine before the token expires.
+   *
+   * @param access_key_id     AWS access key ID.
+   * @param secret_access_key AWS secret access key.
+   * @param session_token     STS session token; empty for long-lived credentials.
+   * @param region            AWS region (e.g. @c "us-east-1").
+   * @param endpoint          S3-compatible endpoint host (e.g. @c "s3.amazonaws.com"
+   *                          or a MinIO host:port). Leave empty to derive from region.
+   * @param n_reactors        Number of libcurl reactor threads.
+   * @param tls_verify        Whether to verify TLS peer certificates.
+   * @param pool_capacity     Total capacity of the pinned host staging pool in bytes.
+   * @param block_size        Fixed block size in bytes for the staging pool.
+   * @return A ready-to-use engine. The engine's @c ioctx is started before returning.
+   */
+  explicit rest_datasource_engine(std::string access_key_id,
+                                  std::string secret_access_key,
+                                  std::string session_token,
+                                  std::string region,
+                                  std::string endpoint,
+                                  std::size_t n_reactors      = 4,
+                                  bool        tls_verify      = true,
+                                  std::size_t pool_capacity   = default_pool_capacity,
+                                  std::size_t block_size      = default_block_size,
+                                  std::size_t max_connections = 16,
+                                  std::size_t chunk_size      = 8UL << 20,
+                                  std::size_t max_n_chunks    = 16,
+                                  bool        enable_cache    = false);
+
+  ~rest_datasource_engine();
+
+  rest_datasource_engine(rest_datasource_engine const&)            = delete;
+  rest_datasource_engine& operator=(rest_datasource_engine const&) = delete;
+
+  /**
+   * @brief Open a datasource for the given S3 URI.
+   *
+   * Issues an HTTP HEAD request to resolve the object size.
+   *
+   * @param path S3 URI of the form @c "s3://bucket/key".
+   * @return A @c datasource bound to this engine's @c ioctx. The returned
+   *         datasource must not outlive this engine.
+   * @throw std::runtime_error if the HEAD request fails or the URI is malformed.
+   */
+  [[nodiscard]] std::unique_ptr<datasource> open(std::string path) const;
+
+ private:
+  cucascade::memory::numa_region_pinned_host_memory_resource          _upstream;
+  cucascade::memory::fixed_size_host_memory_resource                  _host_mr;
+  std::shared_ptr<ioctx>                                              _io_ctx;
+  std::unique_ptr<cucascade::memory::memory_reservation_manager>      _reservation_manager;
+};
+
+}  // namespace cucascade::io

include/cucascade/cudf/uring_datasource_engine.hpp

@@ -0,0 +1,82 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include <cucascade/cudf/datasource.hpp>
+#include <cucascade/memory/fixed_size_host_memory_resource.hpp>
+#include <cucascade/memory/numa_region_pinned_host_allocator.hpp>
+#include <cucascade/io/uring/uring_ioctx.hpp>
+#include <cucascade/io/uring/uring_reactor.hpp>
+
+#include <cstddef>
+#include <memory>
+#include <string>
+
+namespace cucascade::io {
+
+/**
+ * @brief Self-contained io_uring datasource engine for local NVMe reads.
+ *
+ * Owns the full stack needed for O_DIRECT io_uring reads: a NUMA-local pinned
+ * host memory pool, a pool of io_uring reactor threads, and an @c ioctx.
+ * Callers open individual @c datasource instances via @c open(); each datasource
+ * shares the engine's @c ioctx and memory pool but carries its own per-scan
+ * @c prefetching_handle.
+ *
+ * The engine must outlive every datasource it produces.
+ *
+ * @code{.cpp}
+ * cucascade::io::uring_datasource_engine engine;
+ * auto ds = engine.open("/mnt/nvme/data/lineitem.parquet");
+ * ds->fadvise(byte_ranges, device_id);
+ * // ... read through ds as a cudf::io::datasource ...
+ * @endcode
+ */
+class uring_datasource_engine {
+ public:
+  static constexpr std::size_t default_block_size    = 1UL << 20;   ///< 1 MiB
+  static constexpr std::size_t default_pool_capacity = 20UL * 128UL * (1UL << 20);  ///< ~2.5 GiB
+
+  /**
+   * @brief Construct a uring datasource engine.
+   *
+   * @param n_reactors    Number of io_uring reactor threads.
+   * @param pool_capacity Total capacity of the pinned host staging pool in bytes.
+   * @param block_size    Size of each fixed-size block in the staging pool in bytes.
+   *                      Must be a power of two and at least the alignment required
+   *                      by O_DIRECT on the target filesystem.
+   * @param use_odirect   Whether to open files with @c O_DIRECT (bypasses page cache).
+   * @param numa_node     NUMA node from which to allocate the pinned staging pool.
+   */
+  explicit uring_datasource_engine(std::size_t n_reactors   = 2,
+                                   std::size_t pool_capacity = default_pool_capacity,
+                                   std::size_t block_size    = default_block_size,
+                                   bool        use_odirect   = true,
+                                   int         numa_node     = 0);
+
+  ~uring_datasource_engine();
+
+  uring_datasource_engine(uring_datasource_engine const&)            = delete;
+  uring_datasource_engine& operator=(uring_datasource_engine const&) = delete;
+
+  /**
+   * @brief Open a datasource for the given local file path.
+   *
+   * @param path Absolute or relative path to a local file.
+   * @return A @c datasource bound to this engine's @c ioctx. The returned
+   *         datasource must not outlive this engine.
+   * @throw std::runtime_error if the file cannot be opened.
+   */
+  [[nodiscard]] std::unique_ptr<datasource> open(std::string path) const;
+
+ private:
+  cucascade::memory::numa_region_pinned_host_memory_resource _upstream;
+  cucascade::memory::fixed_size_host_memory_resource         _host_mr;
+  std::shared_ptr<uring::uring_reactor::reactor_context>     _reactor_ctx;
+  std::shared_ptr<ioctx>                                     _io_ctx;
+};
+
+}  // namespace cucascade::io

python/CMakeLists.txt

@@ -0,0 +1,61 @@
+# =============================================================================
+# cmake-format: off
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+# cmake-format: on
+# =============================================================================
+
+cmake_minimum_required(VERSION 4.0 FATAL_ERROR)
+
+project(
+  cucascade_python
+  VERSION 0.1.0
+  LANGUAGES CXX)
+
+set(CMAKE_CXX_STANDARD 20)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+find_package(PkgConfig REQUIRED)
+pkg_check_modules(LIBURING REQUIRED IMPORTED_TARGET liburing)
+pkg_check_modules(CURL REQUIRED IMPORTED_TARGET libcurl)
+find_package(OpenSSL REQUIRED)
+
+find_package(cudf REQUIRED)
+find_package(cuCascade REQUIRED COMPONENTS cudf io)
+
+set(rapids-cmake-dir
+    ""
+    CACHE
+      PATH
+      "Optional path to an already-fetched rapids-cmake source tree (skips download)"
+)
+if(rapids-cmake-dir)
+  list(APPEND CMAKE_MODULE_PATH "${rapids-cmake-dir}/rapids-cmake")
+else()
+  set(rapids-cmake-version "24.12")
+  include(FetchContent)
+  FetchContent_Declare(
+    rapids-cmake
+    GIT_REPOSITORY https://github.com/rapidsai/rapids-cmake.git
+    GIT_TAG "branch-${rapids-cmake-version}"
+    GIT_SHALLOW TRUE)
+  FetchContent_MakeAvailable(rapids-cmake)
+  list(APPEND CMAKE_MODULE_PATH "${rapids-cmake_SOURCE_DIR}/rapids-cmake")
+endif()
+
+execute_process(
+  COMMAND
+    "${Python_EXECUTABLE}" -c
+    "import pylibcudf, pathlib; print(pathlib.Path(pylibcudf.__file__).parent.parent)"
+  OUTPUT_VARIABLE PYLIBCUDF_SOURCE_DIR
+  OUTPUT_STRIP_TRAILING_WHITESPACE COMMAND_ERROR_IS_FATAL ANY)
+
+include(rapids-cython-core)
+rapids_cython_init()
+
+set(CYTHON_FLAGS
+    "${CYTHON_FLAGS} -I${PYLIBCUDF_SOURCE_DIR}"
+    CACHE STRING "" FORCE)
+message(STATUS "cucascade: pylibcudf source dir: ${PYLIBCUDF_SOURCE_DIR}")
+
+add_subdirectory(cucascade)