[WIP/DNM] remote-io testing

ci/build_wheel_cudf_streaming.sh

@@ -12,11 +12,13 @@ dependency_file_key_suffix="cudf_streaming"
 
 RAPIDS_PY_CUDA_SUFFIX="$(rapids-wheel-ctk-name-gen "${RAPIDS_CUDA_VERSION}")"
 
-# Downloads libcudf_streaming wheel from this current build,
-# then ensures 'cudf_streaming' wheel builds always use the 'libcudf_streaming' just built in the same CI run.
+# Downloads libcudf, pylibcudf, and libcudf_streaming wheels from the current build.
+# Then ensures 'cudf_streaming' wheel builds always use wheels built in the same CI run.
 LIBCUDF_STREAMING_WHEELHOUSE=$(RAPIDS_PY_WHEEL_NAME="libcudf_streaming_${RAPIDS_PY_CUDA_SUFFIX}" rapids-download-wheels-from-github cpp)
+LIBCUDF_WHEELHOUSE=$(RAPIDS_PY_WHEEL_NAME="libcudf_${RAPIDS_PY_CUDA_SUFFIX}" rapids-download-wheels-from-github cpp)
 PYLIBCUDF_WHEELHOUSE=$(rapids-download-from-github "$(rapids-package-name "wheel_python" pylibcudf --stable --cuda "$RAPIDS_CUDA_VERSION")")
 echo "libcudf-streaming-${RAPIDS_PY_CUDA_SUFFIX} @ file://$(echo "${LIBCUDF_STREAMING_WHEELHOUSE}"/libcudf_streaming_*.whl)" >> "${PIP_CONSTRAINT}"
+echo "libcudf-${RAPIDS_PY_CUDA_SUFFIX} @ file://$(echo "${LIBCUDF_WHEELHOUSE}"/libcudf_*.whl)" >> "${PIP_CONSTRAINT}"
 echo "pylibcudf-${RAPIDS_PY_CUDA_SUFFIX} @ file://$(echo "${PYLIBCUDF_WHEELHOUSE}"/pylibcudf_*.whl)" >> "${PIP_CONSTRAINT}"
 
 rapids-logger "Generating build requirements"

cpp/include/cudf/io/config_utils.hpp

@@ -1,11 +1,13 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2024-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2024-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 #pragma once
 
 #include <cudf/utilities/export.hpp>
 
+#include <cstddef>
+
 namespace CUDF_EXPORT cudf {
 namespace io {
 //! `KvikIO`
@@ -67,5 +69,27 @@ namespace integrated_memory_optimization {
 
 /** @} */  // end of group
 }  // namespace integrated_memory_optimization
+
+//! Parquet
+namespace parquet_reader {
+
+/**
+ * @brief Returns the Parquet reader's footer speculative read size in bytes.
+ *
+ * Controlled by the `LIBCUDF_PARQUET_METADATA_SIZE_HINT` environment variable.
+ * Defaults to 64 KiB.
+ *
+ * When the footer is smaller than the speculative read size, the footer metadata
+ * is loaded in a single read, which is especially useful for high-latency, remote
+ * storage systems. When the footer is larger than the speculative read size, the
+ * footer metadata will be loaded in two reads.
+ *
+ * Set `LIBCUDF_PARQUET_METADATA_SIZE_HINT=0` to disable speculative reads.
+ *
+ * @return Number of bytes to speculatively read from the end of the source.
+ */
+[[nodiscard]] std::size_t metadata_size_hint();
+
+}  // namespace parquet_reader
 }  // namespace io
 }  // namespace CUDF_EXPORT cudf

cpp/include/cudf/io/datasource.hpp

@@ -13,7 +13,7 @@
 #include <rmm/cuda_stream_view.hpp>
 
 #include <future>
-#include <memory>
+#include <optional>
 
 namespace CUDF_EXPORT cudf {
 //! IO interfaces
@@ -98,11 +98,14 @@ class datasource {
    * @param[in] offset Starting byte offset from which data will be read (the default is zero)
    * @param[in] max_size_estimate Upper estimate of the data range that will be read (the default is
    * zero, which means the whole file after `offset`)
+   * @param[in] known_size Optional known file size in bytes. When set for remote URLs, the IO
+   * backend may skip querying the remote server for file size at open time.
    * @return Constructed datasource object
    */
   static std::unique_ptr<datasource> create(std::string const& filepath,
-                                            size_t offset            = 0,
-                                            size_t max_size_estimate = 0);
+                                            size_t offset                         = 0,
+                                            size_t max_size_estimate              = 0,
+                                            std::optional<std::size_t> known_size = std::nullopt);
 
   /**
    * @brief Creates a source from a host memory buffer.

cpp/include/cudf/io/detail/parquet.hpp

@@ -18,6 +18,7 @@
 
 #include <rmm/cuda_stream_view.hpp>
 
+#include <cstddef>
 #include <string>
 #include <vector>
 

cpp/include/cudf/io/types.hpp

@@ -1,5 +1,5 @@
 /*
- * SPDX-FileCopyrightText: Copyright (c) 2019-2025, NVIDIA CORPORATION.
+ * SPDX-FileCopyrightText: Copyright (c) 2019-2026, NVIDIA CORPORATION.
  * SPDX-License-Identifier: Apache-2.0
  */
 
@@ -310,6 +310,17 @@ constexpr inline auto is_byte_like_type()
          std::is_same_v<non_cv_T, std::byte>;
 }
 
+/**
+ * @brief A file path with an optional known size in bytes.
+ *
+ * When `size` is set for a remote URL, the IO backend may skip querying the remote server for file
+ * size at open time.
+ */
+struct filepath_source {
+  std::string path;                   ///< Path or URL of the input file
+  std::optional<std::size_t> size{};  ///< Known file size; omit to query size at open time
+};
+
 /**
  * @brief Source information for read interfaces
  */
@@ -325,8 +336,13 @@ struct source_info {
    * @param file_paths Input files paths
    */
   explicit source_info(std::vector<std::string> file_paths)
-    : _type(io_type::FILEPATH), _num_sources(file_paths.size()), _filepaths(std::move(file_paths))
+    : _type(io_type::FILEPATH), _num_sources(file_paths.size())
   {
+    _filepath_sources.reserve(file_paths.size());
+    for (auto& path : file_paths) {
+      _filepath_sources.push_back({std::move(path), std::nullopt});
+    }
+    rebuild_filepaths();
   }
 
   /**
@@ -335,10 +351,21 @@ struct source_info {
    * @param file_path Single input file
    */
   explicit source_info(std::string file_path)
-    : _type(io_type::FILEPATH), _num_sources(1), _filepaths({std::move(file_path)})
+    : source_info(std::vector<std::string>{std::move(file_path)})
   {
   }
 
+  /**
+   * @brief Construct a new source info object from filepath sources with optional known sizes
+   *
+   * @param sources Input filepath sources
+   */
+  explicit source_info(std::vector<filepath_source> sources)
+    : _type(io_type::FILEPATH), _num_sources(sources.size()), _filepath_sources(std::move(sources))
+  {
+    rebuild_filepaths();
+  }
+
   /**
    * @brief Construct a new source info object for multiple buffers in host memory
    *
@@ -424,6 +451,12 @@ struct source_info {
    * @return The type of the input
    */
   [[nodiscard]] auto type() const { return _type; }
+  /**
+   * @brief Get the filepath sources of the input
+   *
+   * @return The filepath sources of the input
+   */
+  [[nodiscard]] auto const& filepath_sources() const { return _filepath_sources; }
   /**
    * @brief Get the filepaths of the input
    *
@@ -457,8 +490,18 @@ struct source_info {
   [[nodiscard]] auto num_sources() const { return _num_sources; }
 
  private:
+  void rebuild_filepaths()
+  {
+    _filepaths.clear();
+    _filepaths.reserve(_filepath_sources.size());
+    for (auto const& source : _filepath_sources) {
+      _filepaths.push_back(source.path);
+    }
+  }
+
   io_type _type       = io_type::VOID;
   size_t _num_sources = 0;
+  std::vector<filepath_source> _filepath_sources;
   std::vector<std::string> _filepaths;
   std::vector<cudf::host_span<std::byte const>> _host_buffers;
   std::vector<cudf::device_span<std::byte const>> _device_buffers;

cpp/src/io/functions.cpp

@@ -162,24 +162,26 @@ std::vector<std::unique_ptr<cudf::io::datasource>> make_datasources(source_info
   switch (info.type()) {
     case io_type::FILEPATH: {
       std::vector<std::unique_ptr<cudf::io::datasource>> sources;
-      sources.reserve(info.filepaths().size());
+      sources.reserve(info.filepath_sources().size());
       // Creating sources in a single thread is faster for a small number of sources
       auto const pool_use_threshold =
         cudf::detail::getenv_or("LIBCUDF_DATASOURCE_PARALLEL_CREATION_THRESHOLD", 8ul);
-      if (info.filepaths().size() >= pool_use_threshold) {
+      if (info.filepath_sources().size() >= pool_use_threshold) {
         std::vector<std::future<std::unique_ptr<cudf::io::datasource>>> source_tasks;
-        source_tasks.reserve(info.filepaths().size());
-        for (auto const& path : info.filepaths()) {
-          source_tasks.emplace_back(cudf::detail::host_worker_pool().submit_task(
-            [=] { return cudf::io::datasource::create(path, offset, max_size_estimate); }));
+        source_tasks.reserve(info.filepath_sources().size());
+        for (auto const& fs : info.filepath_sources()) {
+          source_tasks.emplace_back(cudf::detail::host_worker_pool().submit_task([=] {
+            return cudf::io::datasource::create(fs.path, offset, max_size_estimate, fs.size);
+          }));
         }
         std::transform(
           source_tasks.begin(), source_tasks.end(), std::back_inserter(sources), [](auto& task) {
             return task.get();
           });
       } else {
-        for (auto const& filepath : info.filepaths()) {
-          sources.emplace_back(cudf::io::datasource::create(filepath, offset, max_size_estimate));
+        for (auto const& fs : info.filepath_sources()) {
+          sources.emplace_back(
+            cudf::io::datasource::create(fs.path, offset, max_size_estimate, fs.size));
         }
       }
       return sources;

cpp/src/io/parquet/io_utils/parquet_io_utils.cpp

@@ -10,6 +10,7 @@
 #include <cudf/detail/utilities/cuda_memcpy.hpp>
 #include <cudf/detail/utilities/host_worker_pool.hpp>
 #include <cudf/detail/utilities/integer_utils.hpp>
+#include <cudf/io/config_utils.hpp>
 #include <cudf/io/datasource.hpp>
 #include <cudf/io/parquet.hpp>
 #include <cudf/io/parquet_io_utils.hpp>
@@ -22,12 +23,18 @@
 #include <cuda/iterator>
 #include <cuda/std/tuple>
 
+#include <algorithm>
+#include <array>
+#include <cstring>
+#include <format>
 #include <functional>
 #include <mutex>
 #include <numeric>
 #include <stdexcept>
+#include <string>
 #include <tuple>
 #include <type_traits>
+#include <vector>
 
 /**
  * @file parquet_io_utils.cpp
@@ -86,23 +93,88 @@ auto dispatch_fetch_tasks(std::size_t num_sources, Task fetch_task)
 std::vector<std::unique_ptr<cudf::io::datasource::buffer>> fetch_footers_to_host_impl(
   cudf::host_span<std::reference_wrapper<cudf::io::datasource> const> datasources)
 {
+  // Look up runtime configuration once, as late as possible.
+  auto const metadata_size_hint = cudf::io::parquet_reader::metadata_size_hint();
   // Helper to fetch footer from a datasource
-  auto const fetch_footer = [](cudf::io::datasource& datasource) {
+  auto const fetch_footer = [metadata_size_hint](cudf::io::datasource& datasource) {
     constexpr auto header_len = sizeof(file_header_s);
     constexpr auto ender_len  = sizeof(file_ender_s);
     size_t const len          = datasource.size();
     CUDF_EXPECTS(len > header_len + ender_len, "Incorrect data source");
 
-    auto header_buffer = datasource.host_read(0, header_len);
-    auto const header  = reinterpret_cast<file_header_s const*>(header_buffer->data());
-    auto ender_buffer  = datasource.host_read(len - ender_len, ender_len);
-    auto const ender   = reinterpret_cast<file_ender_s const*>(ender_buffer->data());
-    CUDF_EXPECTS(header->magic == detail::parquet_magic, "Corrupted header");
-    CUDF_EXPECTS(ender->magic == detail::parquet_magic, "Corrupted footer");
-    CUDF_EXPECTS(ender->footer_len != 0 && ender->footer_len <= (len - header_len - ender_len),
+    auto const speculative_read_size =
+      std::min(len, std::max(metadata_size_hint, static_cast<size_t>(ender_len)));
+    auto const speculative_read_offset = len - speculative_read_size;
+
+    auto speculative_buffer = datasource.host_read(speculative_read_offset, speculative_read_size);
+    CUDF_EXPECTS(speculative_buffer->size() >= speculative_read_size,
+                 std::format("Failed to read Parquet speculative metadata bytes: "
+                             "requested_offset={}, requested_size={}, bytes_read={}, "
+                             "required_size={}",
+                             speculative_read_offset,
+                             speculative_read_size,
+                             speculative_buffer->size(),
+                             speculative_read_size));
+
+    file_ender_s ender{};
+    std::memcpy(
+      &ender, speculative_buffer->data() + speculative_buffer->size() - ender_len, ender_len);
+
+    if (speculative_read_offset == 0) {
+      file_header_s header{};
+      std::memcpy(&header, speculative_buffer->data(), header_len);
+      CUDF_EXPECTS(header.magic == detail::parquet_magic, "Corrupted header");
+    };
+
+    CUDF_EXPECTS(ender.magic == detail::parquet_magic, "Corrupted footer");
+    CUDF_EXPECTS(ender.footer_len != 0 && ender.footer_len <= (len - header_len - ender_len),
                  "Incorrect footer length");
 
-    return datasource.host_read(len - ender->footer_len - ender_len, ender->footer_len);
+    auto const footer_offset = len - ender.footer_len - ender_len;
+    if (footer_offset >= speculative_read_offset) {
+      auto const footer_start_offset = footer_offset - speculative_read_offset;
+      CUDF_EXPECTS(
+        footer_start_offset + ender.footer_len <= speculative_buffer->size(),
+        std::format("Speculative metadata read did not include full footer bytes: "
+                    "file_size={}, metadata_size_hint={}, speculative_read_offset={}, "
+                    "speculative_read_size={}, bytes_read={}, footer_offset={}, footer_len={}",
+                    len,
+                    metadata_size_hint,
+                    speculative_read_offset,
+                    speculative_read_size,
+                    speculative_buffer->size(),
+                    footer_offset,
+                    ender.footer_len));
+      std::vector<uint8_t> footer_bytes(ender.footer_len);
+      std::memcpy(
+        footer_bytes.data(), speculative_buffer->data() + footer_start_offset, ender.footer_len);
+      return cudf::io::datasource::buffer::create(std::move(footer_bytes));
+    }
+
+    // Footer starts before the speculative read range. Read the missing prefix, then stitch.
+    auto const missing_prefix_size = speculative_read_offset - footer_offset;
+    auto missing_prefix            = datasource.host_read(footer_offset, missing_prefix_size);
+    CUDF_EXPECTS(missing_prefix->size() == missing_prefix_size,
+                 std::format("Failed to read the missing footer prefix bytes: "
+                             "requested_offset={}, requested_size={}, bytes_read={}, file_size={}",
+                             footer_offset,
+                             missing_prefix_size,
+                             missing_prefix->size(),
+                             len));
+    std::vector<uint8_t> footer_bytes(ender.footer_len);
+    std::memcpy(footer_bytes.data(), missing_prefix->data(), missing_prefix_size);
+    auto const footer_suffix_size = ender.footer_len - missing_prefix_size;
+    CUDF_EXPECTS(speculative_buffer->size() >= footer_suffix_size,
+                 std::format("Failed to read Parquet speculative metadata suffix bytes: "
+                             "requested_offset={}, requested_size={}, bytes_read={}, "
+                             "required_size={}",
+                             speculative_read_offset,
+                             speculative_read_size,
+                             speculative_buffer->size(),
+                             footer_suffix_size));
+    std::memcpy(
+      footer_bytes.data() + missing_prefix_size, speculative_buffer->data(), footer_suffix_size);
+    return cudf::io::datasource::buffer::create(std::move(footer_bytes));
   };
 
   return dispatch_fetch_tasks(datasources.size(), [&](std::size_t source_idx) {
@@ -338,8 +410,6 @@ fetch_byte_ranges_to_device_async_impl(
 std::unique_ptr<cudf::io::datasource::buffer> fetch_footer_to_host(cudf::io::datasource& datasource)
 {
   CUDF_FUNC_RANGE();
-
-  // Wrap the input into an array and delegate to the multi-source implementation
   std::array<std::reference_wrapper<cudf::io::datasource>, 1> datasources{std::ref(datasource)};
   auto footer_buffers = fetch_footers_to_host_impl({datasources.data(), datasources.size()});
   return std::move(footer_buffers.front());

cpp/src/io/parquet/reader_impl_helpers.hpp

@@ -15,6 +15,7 @@
 
 #include <cstddef>
 #include <functional>
+#include <optional>
 #include <string>
 #include <string_view>
 #include <tuple>

cpp/src/io/utilities/config_utils.cpp

@@ -82,4 +82,16 @@ namespace integrated_memory_optimization {
 }
 
 }  // namespace integrated_memory_optimization
+
+namespace parquet_reader {
+
+[[nodiscard]] std::size_t metadata_size_hint()
+{
+  static constexpr auto default_metadata_size_hint = std::size_t{64} * 1024;
+  static auto const metadata_size_hint             = cudf::detail::getenv_or<std::size_t>(
+    "LIBCUDF_PARQUET_METADATA_SIZE_HINT", default_metadata_size_hint);
+  return metadata_size_hint;
+}
+
+}  // namespace parquet_reader
 }  // namespace cudf::io