Support WebHDFS (1/2): C++ implementation

conda/recipes/libkvikio/conda_build_config.yaml

@@ -11,4 +11,4 @@ c_stdlib_version:
   - "2.28"
 
 libcurl_version:
-  - "==8.5.0"
+  - "8.5.0"

conda/recipes/libkvikio/recipe.yaml

@@ -65,7 +65,7 @@ cache:
       - ${{ stdlib("c") }}
     host:
       - cuda-version =${{ cuda_version }}
-      - libcurl ${{ libcurl_version }}
+      - libcurl ==${{ libcurl_version }}
       - if: should_use_cufile
         then:
         - libcufile-dev
@@ -91,7 +91,7 @@ outputs:
         - ${{ compiler("c") }}
       host:
         - cuda-version =${{ cuda_version }}
-        - libcurl ${{ libcurl_version }}
+        - libcurl ==${{ libcurl_version }}
       run:
         - if: x86_64
           then:
@@ -108,7 +108,6 @@ outputs:
       ignore_run_exports:
         by_name:
           - cuda-version
-          - libcurl
           - if: should_use_cufile
             then:
             - libcufile
@@ -138,6 +137,7 @@ outputs:
         - ${{ pin_subpackage("libkvikio", exact=True) }}
         - cuda-version =${{ cuda_version }}
         - cuda-cudart-dev
+        - libcurl ==${{ libcurl_version }}
         - if: should_use_cufile
           then:
           - libcufile-dev
@@ -156,7 +156,6 @@ outputs:
         by_name:
           - cuda-cudart
           - cuda-version
-          - libcurl
           - libnuma
           - if: should_use_cufile
             then:

cpp/CMakeLists.txt

@@ -163,7 +163,9 @@ set(SOURCES
 )
 
 if(KvikIO_REMOTE_SUPPORT)
-  list(APPEND SOURCES "src/remote_handle.cpp" "src/shim/libcurl.cpp")
+  list(APPEND SOURCES "src/hdfs.cpp" "src/remote_handle.cpp" "src/detail/remote_handle.cpp"
+       "src/shim/libcurl.cpp"
+  )
 endif()
 
 add_library(kvikio ${SOURCES})

cpp/include/kvikio/detail/remote_handle.hpp

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed 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 <cstring>
+
+namespace kvikio::detail {
+/**
+ * @brief Callback for `CURLOPT_WRITEFUNCTION` that copies received data into a `std::string`.
+ *
+ * @param data Received data
+ * @param size Curl internal implementation always sets this parameter to 1
+ * @param num_bytes Number of bytes received
+ * @param userdata Must be cast from `std::string*`
+ * @return The number of bytes consumed by the callback
+ */
+std::size_t callback_get_string_response(char* data,
+                                         std::size_t size,
+                                         std::size_t num_bytes,
+                                         void* userdata);
+}  // namespace kvikio::detail

cpp/include/kvikio/hdfs.hpp

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed 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 <optional>
+
+#include <kvikio/remote_handle.hpp>
+
+namespace kvikio {
+
+/**
+ * @brief A remote endpoint for Apache Hadoop WebHDFS.
+ *
+ * If KvikIO is run within a Docker, the argument `--network host` needs to be passed to the `docker
+ * run` command.
+ */
+class WebHdfsEndpoint : public RemoteEndpoint {
+ private:
+  std::string _url;
+  std::optional<std::string> _username;
+
+ public:
+  /**
+   * @brief Create an WebHDFS endpoint from a url.
+   *
+   * @param url The WebHDFS HTTP/HTTPS url to the remote file.
+   */
+  explicit WebHdfsEndpoint(std::string url);
+
+  /**
+   * @brief Create an WebHDFS endpoint from the host, port, file path and optionally username.
+   *
+   * @param host Host
+   * @param port Port
+   * @param remote_file_path Remote file path
+   * @param username User name
+   */
+  explicit WebHdfsEndpoint(std::string host,
+                           std::string port,
+                           std::string remote_file_path,
+                           std::optional<std::string> username = std::nullopt);
+
+  ~WebHdfsEndpoint() override = default;
+  void setopt(CurlHandle& curl) override;
+  std::string str() const override;
+  std::size_t get_file_size() override;
+  void setup_range_request(CurlHandle& curl, std::size_t file_offset, std::size_t size) override;
+};
+}  // namespace kvikio

cpp/include/kvikio/remote_handle.hpp

@@ -68,6 +68,12 @@ class RemoteEndpoint {
    * @return The file size
    */
   virtual std::size_t get_file_size() = 0;
+
+  /**
+   * @brief Set up the range request in order to read part of a file given the file offset and read
+   * size.
+   */
+  virtual void setup_range_request(CurlHandle& curl, std::size_t file_offset, std::size_t size) = 0;
 };
 
 /**
@@ -89,6 +95,7 @@ class HttpEndpoint : public RemoteEndpoint {
   void setopt(CurlHandle& curl) override;
   std::string str() const override;
   std::size_t get_file_size() override;
+  void setup_range_request(CurlHandle& curl, std::size_t file_offset, std::size_t size) override;
 };
 
 /**
@@ -198,6 +205,7 @@ class S3Endpoint : public RemoteEndpoint {
   void setopt(CurlHandle& curl) override;
   std::string str() const override;
   std::size_t get_file_size() override;
+  void setup_range_request(CurlHandle& curl, std::size_t file_offset, std::size_t size) override;
 };
 
 /**
@@ -215,6 +223,7 @@ class S3EndpointWithPresignedUrl : public RemoteEndpoint {
   void setopt(CurlHandle& curl) override;
   std::string str() const override;
   std::size_t get_file_size() override;
+  void setup_range_request(CurlHandle& curl, std::size_t file_offset, std::size_t size) override;
 };
 
 /**

cpp/src/detail/remote_handle.cpp

@@ -0,0 +1,32 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed 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 <string>
+
+#include <kvikio/detail/remote_handle.hpp>
+
+namespace kvikio::detail {
+std::size_t callback_get_string_response(char* data,
+                                         std::size_t size,
+                                         std::size_t num_bytes,
+                                         void* userdata)
+{
+  auto new_data_size = size * num_bytes;
+  auto* response     = reinterpret_cast<std::string*>(userdata);
+  response->append(data, new_data_size);
+  return new_data_size;
+}
+}  // namespace kvikio::detail

cpp/src/hdfs.cpp

@@ -0,0 +1,131 @@
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION.
+ *
+ * Licensed 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 <regex>
+
+#include <kvikio/detail/remote_handle.hpp>
+#include <kvikio/error.hpp>
+#include <kvikio/hdfs.hpp>
+#include <kvikio/nvtx.hpp>
+#include <kvikio/remote_handle.hpp>
+#include <kvikio/shim/libcurl.hpp>
+
+namespace kvikio {
+
+WebHdfsEndpoint::WebHdfsEndpoint(std::string url)
+{
+  // todo: Use libcurl URL API for more secure and idiomatic parsing.
+  // Split the URL into two parts: one without query and one with.
+  std::regex const pattern{R"(^([^?]+)\?([^#]*))"};
+  // Regex meaning:
+  // ^: From the start of the line
+  // [^?]+: Matches non-question-mark characters one or more times. The question mark ushers in the
+  // URL query component.
+  // \?: Matches the question mark, which needs to be escaped.
+  // [^#]*: Matches the non-pound characters zero or more times. The pound sign ushers in the URL
+  // fragment component. It is very likely that this part does not exist.
+  std::smatch match_results;
+  bool found = std::regex_search(url, match_results, pattern);
+  // If the match is not found, the URL contains no query.
+  if (!found) {
+    _url = url;
+    return;
+  }
+
+  _url       = match_results[1].str();
+  auto query = match_results[2].str();
+
+  {
+    // Extract user name if provided. In WebHDFS, user name is specified as the key=value pair in
+    // the query
+    std::regex const pattern{R"(user.name=([^&]+))"};
+    // Regex meaning:
+    // [^&]+: Matches the non-ampersand character one or more times. The ampersand delimits
+    // different parameters.
+    std::smatch match_results;
+    if (std::regex_search(query, match_results, pattern)) { _username = match_results[1].str(); }
+  }
+}
+
+WebHdfsEndpoint::WebHdfsEndpoint(std::string host,
+                                 std::string port,
+                                 std::string file_path,
+                                 std::optional<std::string> username)
+  : _username{std::move(username)}
+{
+  std::stringstream ss;
+  ss << "http://" << host << ":" << port << "/webhdfs/v1" << file_path;
+  _url = ss.str();
+}
+
+std::string WebHdfsEndpoint::str() const { return _url; }
+
+void WebHdfsEndpoint::setopt(CurlHandle& curl)
+{
+  KVIKIO_NVTX_FUNC_RANGE();
+  curl.setopt(CURLOPT_URL, _url.c_str());
+  curl.setopt(CURLOPT_FOLLOWLOCATION, 1L);
+}
+
+std::size_t WebHdfsEndpoint::get_file_size()
+{
+  KVIKIO_NVTX_FUNC_RANGE();
+
+  std::stringstream ss;
+  ss << _url << "?";
+  if (_username.has_value()) { ss << "user.name=" << _username.value() << "&"; }
+  ss << "op=GETFILESTATUS";
+
+  auto curl = create_curl_handle();
+  curl.setopt(CURLOPT_URL, ss.str().c_str());
+  curl.setopt(CURLOPT_FOLLOWLOCATION, 1L);
+
+  std::string response;
+  curl.setopt(CURLOPT_WRITEDATA, static_cast<void*>(&response));
+  curl.setopt(CURLOPT_WRITEFUNCTION, detail::callback_get_string_response);
+
+  curl.perform();
+
+  long http_status_code{};
+  curl.getinfo(CURLINFO_RESPONSE_CODE, &http_status_code);
+  KVIKIO_EXPECT(http_status_code == 200, "HTTP response is not successful.");
+
+  // The response is in JSON format. The file size is given by `"length":<file_size>`.
+  std::regex const pattern{R"("length"\s*:\s*(\d+)[^\d])"};
+  // Regex meaning:
+  // \s*: Matches the space character zero or more times.
+  // \d+: Matches the digit one or more times.
+  // [^\d]: Matches a non-digit character.
+  std::smatch match_results;
+  bool found = std::regex_search(response, match_results, pattern);
+  KVIKIO_EXPECT(
+    found, "Regular expression search failed. Cannot extract file length from the JSON response.");
+  return std::stoull(match_results[1].str());
+}
+
+void WebHdfsEndpoint::setup_range_request(CurlHandle& curl,
+                                          std::size_t file_offset,
+                                          std::size_t size)
+{
+  // WebHDFS does not support CURLOPT_RANGE. The range is specified as query parameters in the URL.
+  KVIKIO_NVTX_FUNC_RANGE();
+  std::stringstream ss;
+  ss << _url << "?";
+  if (_username.has_value()) { ss << "user.name=" << _username.value() << "&"; }
+  ss << "op=OPEN&offset=" << file_offset << "&length=" << size;
+  curl.setopt(CURLOPT_URL, ss.str().c_str());
+}
+}  // namespace kvikio