feat(hive): add HmsClient connection lifecycle and URI parsing

.github/workflows/cpp-linter.yml

@@ -79,7 +79,8 @@ jobs:
             -DICEBERG_BUILD_SQL_CATALOG=ON \
             -DICEBERG_SQL_SQLITE=ON \
             -DICEBERG_SQL_POSTGRESQL=ON \
-            -DICEBERG_SQL_MYSQL=ON
+            -DICEBERG_SQL_MYSQL=ON \
+            -DICEBERG_BUILD_HIVE=ON
           cmake --build .
       - name: Show sccache stats
         shell: bash

.github/workflows/test.yml

@@ -86,6 +86,50 @@ jobs:
           CC: gcc-14
           CXX: g++-14
         run: ci/scripts/build_example.sh $(pwd)/example
+  hive:
+    if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }}
+    name: AMD64 Ubuntu 26.04 Hive
+    runs-on: ubuntu-26.04
+    timeout-minutes: 45
+    strategy:
+      fail-fast: false
+    env:
+      SCCACHE_DIR: ${{ github.workspace }}/.sccache
+      SCCACHE_CACHE_SIZE: "2G"
+      ICEBERG_EXTRA_CMAKE_ARGS: "-DICEBERG_BUILD_HIVE=ON"
+    steps:
+      - name: Checkout iceberg-cpp
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          persist-credentials: false
+      - name: Install dependencies
+        shell: bash
+        run: sudo apt-get update && sudo apt-get install -y libcurl4-openssl-dev
+      - name: Restore sccache cache
+        uses: actions/cache/restore@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        with:
+          path: ${{ github.workspace }}/.sccache
+          key: sccache-test-hive-${{ github.run_id }}
+          restore-keys: |
+            sccache-test-hive-
+            sccache-test-ubuntu-
+      - name: Setup sccache
+        uses: mozilla-actions/sccache-action@9e7fa8a12102821edf02ca5dbea1acd0f89a2696 # v0.0.10
+      - name: Build and test Iceberg with Hive
+        shell: bash
+        env:
+          CC: gcc-14
+          CXX: g++-14
+        run: ci/scripts/build_iceberg.sh $(pwd) OFF ON
+      - name: Show sccache stats
+        shell: bash
+        run: sccache --show-stats
+      - name: Save sccache cache
+        if: github.ref == 'refs/heads/main'
+        uses: actions/cache/save@27d5ce7f107fe9357f9df03efb73ab90386fccae # v5.0.5
+        with:
+          path: ${{ github.workspace }}/.sccache
+          key: sccache-test-hive-${{ github.run_id }}
   macos:
     if: ${{ github.event_name != 'pull_request' || github.event.pull_request.draft == false }}
     name: AArch64 macOS 26

src/iceberg/catalog/hive/CMakeLists.txt

@@ -54,7 +54,7 @@ if(NOT TARGET thrift::thrift)
                       "-DICEBERG_BUNDLE_THRIFT=OFF against a system Thrift install.")
 endif()
 
-set(ICEBERG_HIVE_SOURCES hive_catalog.cc hive_catalog_properties.cc
+set(ICEBERG_HIVE_SOURCES hive_catalog.cc hive_catalog_properties.cc hms_client.cc
                          ${ICEBERG_HIVE_THRIFT_GEN_SOURCES})
 
 set(ICEBERG_HIVE_STATIC_BUILD_INTERFACE_LIBS)

src/iceberg/catalog/hive/hive_catalog_properties.cc

@@ -22,10 +22,24 @@
 #include <string>
 #include <string_view>
 
+#include "iceberg/util/macros.h"
 #include "iceberg/util/string_util.h"
 
 namespace iceberg::hive {
 
+namespace {
+
+/// Parse a millisecond timeout property, rejecting malformed or negative values.
+Result<int> ParseTimeoutMs(const std::string& value, std::string_view property) {
+  ICEBERG_ASSIGN_OR_RAISE(auto ms, StringUtils::ParseNumber<int>(value));
+  if (ms < 0) {
+    return InvalidArgument("Invalid Hive {}: '{}'.", property, value);
+  }
+  return ms;
+}
+
+}  // namespace
+
 HiveCatalogProperties HiveCatalogProperties::default_properties() { return {}; }
 
 HiveCatalogProperties HiveCatalogProperties::FromMap(
@@ -54,4 +68,12 @@ Result<HiveThriftTransport> HiveCatalogProperties::ThriftTransport() const {
   return InvalidArgument("Invalid Hive thrift transport: '{}'.", Get(kThriftTransport));
 }
 
+Result<int> HiveCatalogProperties::ConnectTimeoutMs() const {
+  return ParseTimeoutMs(Get(kConnectTimeoutMs), kConnectTimeoutMs.key());
+}
+
+Result<int> HiveCatalogProperties::SocketTimeoutMs() const {
+  return ParseTimeoutMs(Get(kSocketTimeoutMs), kSocketTimeoutMs.key());
+}
+
 }  // namespace iceberg::hive

src/iceberg/catalog/hive/hive_catalog_properties.h

@@ -73,11 +73,14 @@ class ICEBERG_HIVE_EXPORT HiveCatalogProperties
   /// \brief Thrift framing for the HMS connection ("buffered" or "framed").
   inline static Entry<std::string> kThriftTransport{"thrift-transport", "buffered"};
 
-  /// \brief HMS connect timeout, in milliseconds.
-  inline static Entry<int> kConnectTimeoutMs{"connect-timeout-ms", 30000};
+  /// \brief HMS connect timeout, in milliseconds. Stored as a string so a
+  /// malformed value surfaces as an error from `ConnectTimeoutMs()` rather
+  /// than throwing during `Get`.
+  inline static Entry<std::string> kConnectTimeoutMs{"connect-timeout-ms", "30000"};
 
-  /// \brief HMS socket / RPC timeout, in milliseconds.
-  inline static Entry<int> kSocketTimeoutMs{"socket-timeout-ms", 60000};
+  /// \brief HMS socket / RPC timeout, in milliseconds. Stored as a string for
+  /// the same reason as `kConnectTimeoutMs`.
+  inline static Entry<std::string> kSocketTimeoutMs{"socket-timeout-ms", "60000"};
 
   /// \brief When true, wrap the commit path with HMS `lock` / `unlock` for
   /// extra safety on top of the metadata_location CAS. Defaults to false
@@ -99,6 +102,14 @@ class ICEBERG_HIVE_EXPORT HiveCatalogProperties
   /// is case-insensitive to match the conventions used by other Iceberg
   /// language ports.
   Result<HiveThriftTransport> ThriftTransport() const;
+
+  /// \brief Parse `kConnectTimeoutMs` into a non-negative millisecond value.
+  /// Returns InvalidArgument for malformed or negative values.
+  Result<int> ConnectTimeoutMs() const;
+
+  /// \brief Parse `kSocketTimeoutMs` into a non-negative millisecond value.
+  /// Returns InvalidArgument for malformed or negative values.
+  Result<int> SocketTimeoutMs() const;
 };
 
 }  // namespace iceberg::hive

src/iceberg/catalog/hive/hms_client.cc

@@ -0,0 +1,199 @@
+/*
+ * 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 "iceberg/catalog/hive/hms_client.h"
+
+#include <cctype>
+#include <charconv>
+#include <memory>
+#include <string>
+#include <string_view>
+#include <system_error>
+#include <utility>
+#include <vector>
+
+#include <thrift/Thrift.h>
+#include <thrift/protocol/TBinaryProtocol.h>
+#include <thrift/transport/TBufferTransports.h>
+#include <thrift/transport/TSocket.h>
+#include <thrift/transport/TTransportException.h>
+
+#include "ThriftHiveMetastore.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::hive {
+
+namespace {
+
+constexpr std::string_view kThriftPrefix = "thrift://";
+
+std::string_view StripScheme(std::string_view spec) {
+  if (spec.starts_with(kThriftPrefix)) {
+    return spec.substr(kThriftPrefix.size());
+  }
+  return spec;
+}
+
+std::string_view Trim(std::string_view spec) {
+  while (!spec.empty() && (std::isspace(static_cast<unsigned char>(spec.front())) != 0)) {
+    spec.remove_prefix(1);
+  }
+  while (!spec.empty() && (std::isspace(static_cast<unsigned char>(spec.back())) != 0)) {
+    spec.remove_suffix(1);
+  }
+  return spec;
+}
+
+Result<HmsEndpoint> ParseSingleEndpoint(std::string_view spec) {
+  spec = Trim(spec);
+  spec = StripScheme(spec);
+  spec = Trim(spec);
+  if (spec.empty()) {
+    return InvalidArgument("Empty HMS endpoint in URI list.");
+  }
+
+  HmsEndpoint endpoint;
+  const auto colon = spec.rfind(':');
+  if (colon == std::string_view::npos) {
+    endpoint.host = std::string(spec);
+    endpoint.port = kDefaultHmsPort;
+    return endpoint;
+  }
+
+  endpoint.host = std::string(spec.substr(0, colon));
+  if (endpoint.host.empty()) {
+    return InvalidArgument("HMS endpoint has empty host: '{}'.", spec);
+  }
+
+  const auto port_str = spec.substr(colon + 1);
+  if (port_str.empty()) {
+    endpoint.port = kDefaultHmsPort;
+    return endpoint;
+  }
+
+  int port = 0;
+  const auto* const port_end = port_str.data() + port_str.size();
+  const auto [ptr, ec] = std::from_chars(port_str.data(), port_end, port);
+  if (ec != std::errc() || ptr != port_end || port <= 0 || port > 65535) {
+    return InvalidArgument("Invalid HMS port in endpoint '{}'.", spec);
+  }
+  endpoint.port = port;
+  return endpoint;
+}
+
+}  // namespace
+
+Result<std::vector<HmsEndpoint>> ParseHmsUris(std::string_view uri) {
+  std::vector<HmsEndpoint> endpoints;
+  if (Trim(uri).empty()) {
+    return InvalidArgument("HMS URI is empty.");
+  }
+
+  std::size_t pos = 0;
+  while (pos <= uri.size()) {
+    const auto comma = uri.find(',', pos);
+    const auto piece = uri.substr(
+        pos, comma == std::string_view::npos ? std::string_view::npos : comma - pos);
+    ICEBERG_ASSIGN_OR_RAISE(auto endpoint, ParseSingleEndpoint(piece));
+    endpoints.push_back(std::move(endpoint));
+    if (comma == std::string_view::npos) {
+      break;
+    }
+    pos = comma + 1;
+  }
+  return endpoints;
+}
+
+// Fields are declared in dependency order (socket <- transport <- protocol
+// <- client) so the client tears down before the transport it borrows.
+class HmsClient::Impl {
+ public:
+  std::shared_ptr<apache::thrift::transport::TSocket> socket;
+  std::shared_ptr<apache::thrift::transport::TTransport> transport;
+  std::shared_ptr<apache::thrift::protocol::TProtocol> protocol;
+  std::unique_ptr<Apache::Hadoop::Hive::ThriftHiveMetastoreClient> client;
+};
+
+HmsClient::HmsClient(std::unique_ptr<Impl> impl) : impl_(std::move(impl)) {}
+
+HmsClient::~HmsClient() {
+  if (impl_ && impl_->transport && impl_->transport->isOpen()) {
+    try {
+      impl_->transport->close();
+    } catch (const apache::thrift::TException&) {
+      // Best-effort close on teardown; ignore exceptions.
+    }
+  }
+}
+
+Result<std::unique_ptr<HmsClient>> HmsClient::Connect(
+    const HiveCatalogProperties& config) {
+  ICEBERG_ASSIGN_OR_RAISE(auto uri, config.Uri());
+  ICEBERG_ASSIGN_OR_RAISE(auto endpoints, ParseHmsUris(uri));
+  ICEBERG_ASSIGN_OR_RAISE(auto transport_mode, config.ThriftTransport());
+
+  if (endpoints.size() > 1) {
+    return InvalidArgument(
+        "Multi-endpoint HMS URIs are not yet supported; HA failover is not "
+        "implemented. Configure a single endpoint. Got {} endpoints.",
+        endpoints.size());
+  }
+  const HmsEndpoint& endpoint = endpoints.front();
+  ICEBERG_ASSIGN_OR_RAISE(auto connect_timeout_ms, config.ConnectTimeoutMs());
+  ICEBERG_ASSIGN_OR_RAISE(auto socket_timeout_ms, config.SocketTimeoutMs());
+
+  auto socket =
+      std::make_shared<apache::thrift::transport::TSocket>(endpoint.host, endpoint.port);
+  socket->setConnTimeout(connect_timeout_ms);
+  socket->setRecvTimeout(socket_timeout_ms);
+  socket->setSendTimeout(socket_timeout_ms);
+
+  std::shared_ptr<apache::thrift::transport::TTransport> transport;
+  switch (transport_mode) {
+    case HiveThriftTransport::kBuffered:
+      transport = std::make_shared<apache::thrift::transport::TBufferedTransport>(socket);
+      break;
+    case HiveThriftTransport::kFramed:
+      transport = std::make_shared<apache::thrift::transport::TFramedTransport>(socket);
+      break;
+  }
+
+  auto protocol = std::make_shared<apache::thrift::protocol::TBinaryProtocol>(transport);
+  auto client =
+      std::make_unique<Apache::Hadoop::Hive::ThriftHiveMetastoreClient>(protocol);
+
+  try {
+    transport->open();
+  } catch (const apache::thrift::transport::TTransportException& e) {
+    return IOError("Failed to connect to HMS at {}:{} : {}", endpoint.host, endpoint.port,
+                   e.what());
+  } catch (const apache::thrift::TException& e) {
+    return IOError("Thrift error contacting HMS at {}:{} : {}", endpoint.host,
+                   endpoint.port, e.what());
+  }
+
+  auto impl = std::make_unique<HmsClient::Impl>();
+  impl->socket = std::move(socket);
+  impl->transport = std::move(transport);
+  impl->protocol = std::move(protocol);
+  impl->client = std::move(client);
+  return std::unique_ptr<HmsClient>(new HmsClient(std::move(impl)));
+}
+
+}  // namespace iceberg::hive