feat(io): S3 LIST + footer probe + kvikIO backend, and io-layer fixes

CMakeLists.txt

@@ -158,6 +158,11 @@ if(NOT CUCASCADE_TOPOLOGY_ONLY)
     pkg_check_modules(CURL REQUIRED IMPORTED_TARGET libcurl)
     find_package(OpenSSL REQUIRED)
 
+    # kvikIO — backs the local-file fallback ioctx (kvikio_context). Used
+    # directly (not via cudf) so the io library stays cudf-free. Not swappable:
+    # unlike moodycamel/invocable below there is no in-tree stand-in to replace.
+    find_package(kvikio REQUIRED CONFIG)
+
     # cucascade_io_thirdparty carries the swappable moodycamel + invocable
     # (abseil) usage requirements from a single place; the io object library,
     # its installable static/shared variants, and their in-tree consumers
@@ -402,7 +407,7 @@ if(CUCASCADE_BUILD_IO)
   # side by cuCascadeConfig.cmake (same names), mirroring the Numa::Numa
   # approach.
   set(CUCASCADE_IO_LINK_LIBS PkgConfig::LIBURING PkgConfig::CURL
-                             OpenSSL::Crypto)
+                             OpenSSL::Crypto kvikio::kvikio)
 
   target_link_libraries(
     cucascade_io_objects PUBLIC cucascade_objects ${CUCASCADE_IO_LINK_LIBS}

include/cucascade/cudf/datasource.hpp

@@ -159,4 +159,19 @@ class datasource : public cudf::io::datasource {
 [[nodiscard]] std::unique_ptr<datasource> open_datasource(std::shared_ptr<ioctx> io_ctx,
                                                           std::string path);
 
+/// As above, forwarding @p hint to the backend's io_object resolution so it can,
+/// e.g., prefetch a parquet footer in the same round-trip as the size
+/// (@c open_hint::parquet_footer_probe).  Backends that cannot act on the hint
+/// fall back to the plain open.
+[[nodiscard]] std::unique_ptr<datasource> open_datasource(std::shared_ptr<ioctx> io_ctx,
+                                                          std::string path,
+                                                          open_hint hint);
+
+/// As above, with the object's size already known (e.g. from an S3
+/// ListObjectsV2 response), so a backend that can act on it skips its size
+/// discovery entirely (no HEAD for object stores).
+[[nodiscard]] std::unique_ptr<datasource> open_datasource(std::shared_ptr<ioctx> io_ctx,
+                                                          std::string path,
+                                                          std::uint64_t known_size);
+
 }  // namespace cucascade::io

include/cucascade/io/cache/metadata_store.hpp

@@ -20,13 +20,34 @@
 
 #include <cucascade/io/types.hpp>
 
+#include <cstddef>
+#include <functional>
 #include <memory>
 #include <shared_mutex>
 #include <string>
+#include <string_view>
 #include <unordered_map>
 
 namespace cucascade::io::cache {
 
+namespace detail {
+
+/// Transparent hasher so the store can be looked up by @c std::string_view (or
+/// @c const char*) without materialising a @c std::string.  Paired with
+/// @c std::equal_to<> below, this enables C++20 heterogeneous lookup on the
+/// underlying @c unordered_map — without both, a string_view-taking getter
+/// would just construct a temporary key on every call and be strictly worse
+/// than taking @c std::string const&.
+struct string_hash {
+  using is_transparent = void;
+  [[nodiscard]] std::size_t operator()(std::string_view sv) const noexcept
+  {
+    return std::hash<std::string_view>{}(sv);
+  }
+};
+
+}  // namespace detail
+
 /**
  * @brief Thread-safe per-file metadata cache, keyed by an io_object's
  *        raw_file_cache_id().
@@ -58,9 +79,19 @@ class metadata_store {
   /// miss.
   [[nodiscard]] std::shared_ptr<io_object_metadata> get_metadata(io_object const& obj) const;
 
+  /// As above but keyed directly by @c raw_file_cache_id() — for callers that
+  /// know the path but have not built an io_object yet.  Returns nullptr on miss.
+  /// Looked up heterogeneously, so passing a @c string_view or a string literal
+  /// allocates nothing.
+  [[nodiscard]] std::shared_ptr<io_object_metadata> get_metadata(std::string_view cache_key) const;
+
  private:
   mutable std::shared_mutex _mtx;
-  std::unordered_map<std::string, std::shared_ptr<io_object_metadata>> _by_key;
+  std::unordered_map<std::string,
+                     std::shared_ptr<io_object_metadata>,
+                     detail::string_hash,
+                     std::equal_to<>>
+    _by_key;
 };
 
 }  // namespace cucascade::io::cache

include/cucascade/io/config.hpp

@@ -18,6 +18,7 @@
 #pragma once
 
 #include <cucascade/io/cache/config.hpp>
+#include <cucascade/io/kvikio/config.hpp>
 #include <cucascade/io/object_store_config.hpp>
 #include <cucascade/io/rest/config.hpp>
 #include <cucascade/io/uring/config.hpp>
@@ -34,6 +35,7 @@ namespace cucascade::io {
  * Sub-configs:
  *  - @c local   — uring reactor tunables (local-disk IO path).
  *  - @c rest    — REST reactor tunables (S3/object-store IO path).
+ *  - @c kvikio  — kvikIO fallback tunables (local-disk catch-all path).
  *  - @c cache   — prefetching cache tunables.
  *  - @c object_store — object-store credentials and endpoint.
  */
@@ -57,6 +59,12 @@ struct io_config {
   /// retry policy, etc.
   rest::config rest{};
 
+  /// kvikIO fallback configuration — thread-pool size, task/bounce sizing,
+  /// O_DIRECT, compat mode.  All fields default to "unset", leaving kvikIO's
+  /// own env-var-seeded defaults in place.  Note these are process-global once
+  /// applied; see @ref kvikio_config.
+  kvikio_config kvikio{};
+
   /// Prefetching cache configuration — in-flight budget, pool sizing,
   /// dispose-after-use policy.
   cache::config cache{};

include/cucascade/io/datasource_factory.hpp

@@ -38,17 +38,13 @@ namespace cucascade::io {
 // ---------------------------------------------------------------------------
 
 /**
- * @brief Thread-safe registry mapping URI schemes to @c ioctx instances.
+ * @brief Thread-safe registry of @c ioctx backends, resolved by full path.
  *
- * The engine constructs a registry at startup and populates it with one
- * @c ioctx per backend (uring / gds / s3 / rdma_s3). The factory looks
- * up the correct backend by URI scheme at datasource-creation time.
- *
- * Scheme matching is case-insensitive: @c register_ioctx and @c lookup both
- * lowercase the scheme before storing / searching, matching the
- * normalization done by @c cucascade::io::parse (RFC 3986 §3.1). Callers may
- * register / look up with any casing — @c register_ioctx("S3", ...) and
- * @c lookup("s3") refer to the same entry.
+ * The engine constructs a registry at startup and registers one entry per
+ * backend (kvikio / uring / restful), each carrying a path-capability checker.
+ * At datasource-creation time @c lookup_path runs the checkers against a full
+ * path (the checkers parse the URI / stat the filesystem themselves) and picks
+ * the backend, preferring an explicit backend over the kvikio catch-all.
  *
  * All operations are safe under concurrent reads; mutations take an exclusive
  * lock but are expected only at engine bootstrap / shutdown.
@@ -75,17 +71,21 @@ class io_context_registry {
   using factory_type        = std::function<std::shared_ptr<io::ioctx>(const config_type&)>;
 
   /**
-   * @brief Register an ioctx for a scheme. Replaces any prior registration
-   *        for the same scheme.
+   * @brief Register an ioctx backend. Replaces any prior registration for the
+   *        same type.
    *
-   * The scheme is lowercased before storage; subsequent @c lookup calls
-   * with any casing of the same scheme resolve to this entry.
-   * @param type    Opaque identifier for the ioctx type. Used by the engine to
-   *                identify the backend.
+   * @param type    Backend identifier (uring / restful / kvikio).
+   * @param checker Decides whether this backend claims a given path.
+   * @param factory Constructs the backend's ioctx; invoked by @c make_ioctx.
    */
   void register_ioctx(io_context_type type, scheme_checker_type checker, factory_type factory);
 
-  std::optional<io_context_type> lookup(std::string_view scheme) const noexcept;
+  /// Resolve the backend for a full @p path (not a bare scheme — the checkers
+  /// parse the URI / stat the filesystem themselves).  Explicit backends
+  /// (uring / restful) take precedence over the kvikio catch-all, so `s3://`
+  /// never resolves to kvikio and a local file routes to uring before the
+  /// universal fallback.  std::nullopt when nothing matches.
+  std::optional<io_context_type> lookup_path(std::string_view path) const noexcept;
 
   std::shared_ptr<ioctx> make_ioctx(io_context_type type) const noexcept;
 

include/cucascade/io/io_context.hpp

@@ -27,6 +27,7 @@
 #include <rmm/cuda_stream_view.hpp>
 
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <optional>
 #include <span>
@@ -36,7 +37,15 @@
 
 namespace cucascade::io {
 
-enum class io_context_type { uring, restful };
+enum class io_context_type { uring, restful, kvikio };
+
+/// Hint passed to @c open_io_object so a backend can tailor how it resolves an
+/// object's metadata.  @c generic resolves the size however is cheapest for the
+/// scheme (a HEAD for object stores).  @c parquet_footer_probe asks the backend
+/// to resolve the size *and* stash the object's trailing bytes in one
+/// round-trip (a suffix-range GET), so the parquet footer reads that follow are
+/// served locally instead of costing extra round-trips.
+enum class open_hint { generic, parquet_footer_probe };
 
 namespace cache {
 class prefetching_cache;
@@ -102,6 +111,22 @@ class ioctx : public std::enable_shared_from_this<ioctx> {
     return create_io_object(std::move(path));
   }
 
+  /// As above, forwarding @p hint to the backend's io_object resolution so it
+  /// can, e.g., prefetch a parquet footer in the same round-trip as the size.
+  [[nodiscard]] std::shared_ptr<io_object> open_io_object(std::string path, open_hint hint)
+  {
+    return create_io_object(std::move(path), hint);
+  }
+
+  /// As above, with the object's size already known (e.g. from an S3
+  /// ListObjectsV2 response), so a backend that can act on it skips its size
+  /// discovery entirely (no HEAD for object stores).
+  [[nodiscard]] std::shared_ptr<io_object> open_io_object(std::string path,
+                                                          std::uint64_t known_size)
+  {
+    return create_io_object(std::move(path), known_size);
+  }
+
   /// Whether this backend can serve reads for @p path.  Backends should
   /// validate scheme/protocol support and any backend-specific
   /// preconditions (e.g. file existence for local-disk backends).
@@ -263,6 +288,20 @@ class ioctx : public std::enable_shared_from_this<ioctx> {
   /// on unsupported / unreachable paths.
   virtual std::shared_ptr<io_object> create_io_object(std::string path) = 0;
 
+  /// Hinted variant.  The base implementation ignores @p hint and delegates to
+  /// the required @c create_io_object(path); a backend that can act on the hint
+  /// (e.g. rest_ioctx's suffix-range footer probe) overrides this.  Kept a
+  /// distinct virtual — not a defaulted argument on the pure virtual above — so
+  /// the hint dispatches on the dynamic type instead of binding statically.
+  virtual std::shared_ptr<io_object> create_io_object(std::string path, open_hint hint);
+
+  /// Known-size variant.  The base implementation ignores @p known_size and
+  /// delegates to the required @c create_io_object(path); a backend whose size
+  /// discovery would otherwise cost a round-trip overrides this to build the
+  /// io_object without one.  Same distinct-virtual rationale as the hint
+  /// variant above.
+  virtual std::shared_ptr<io_object> create_io_object(std::string path, std::uint64_t known_size);
+
   /// Owned by this ioctx.  Built by @ref initialize_cache, destroyed
   /// by @ref shutdown_cache (or the ioctx destructor as a safety net,
   /// though callers are expected to drive the lifecycle explicitly so

include/cucascade/io/io_request.hpp

@@ -147,9 +147,7 @@ struct device_cpy_request {
 
   // Issue every copy on @p stream (a batch when there is more than one), then
   // record @p event once after the last so a single wait covers them all.
-  cudaError_t copy_async(uint8_t* host_buffer,
-                         [[maybe_unused]] size_t bytes,
-                         cudaEvent_t event = nullptr) noexcept
+  cudaError_t copy_async(uint8_t* host_buffer, size_t bytes, cudaEvent_t event = nullptr) noexcept
   {
     assert(host_buffer != nullptr && "Caller must provide a valid host buffer for the copy.");
     rmm::cuda_set_device_raii device_guard(rmm::cuda_device_id{device_id});
@@ -159,8 +157,24 @@ struct device_cpy_request {
              "Caller must provide a valid device destination buffer for the copy.");
       assert((c.src != nullptr || c.src_off + c.size <= bytes) &&
              "Caller must ensure the copy fits in the host buffer.");
-      uint8_t* src_ptr = c.src != nullptr ? c.src : host_buffer + c.src_off;
-      err              = cudaMemcpyAsync(c.dst, src_ptr, c.size, cudaMemcpyHostToDevice, stream);
+      // Resolve the host source.  The asserts above are compiled out in release,
+      // so validate *before* forming the pointer: for a bounce-staged copy
+      // (c.src == nullptr) the source is host_buffer + c.src_off, but a null
+      // host_buffer or an out-of-range [src_off, src_off + size) would otherwise
+      // produce UB (nullptr + offset) or a wild in-range pointer that the
+      // near-null check below cannot catch.  A null-buffer segment must reach
+      // here as c.src == nullptr, never as a non-null "nullptr + offset" pointer.
+      uint8_t* src_ptr = nullptr;
+      if (c.src != nullptr) {
+        src_ptr = c.src;
+      } else if (host_buffer != nullptr && c.src_off <= bytes && c.size <= bytes - c.src_off) {
+        src_ptr = host_buffer + c.src_off;
+      }
+      if (c.dst == nullptr || src_ptr == nullptr ||
+          reinterpret_cast<std::uintptr_t>(src_ptr) < 4096U) {
+        return cudaErrorInvalidValue;
+      }
+      err = cudaMemcpyAsync(c.dst, src_ptr, c.size, cudaMemcpyHostToDevice, stream);
       if (err != cudaSuccess) { return err; }
     }
     if (event != nullptr) { err = cudaEventRecord(event, stream); }