feat(hive): HmsClient connection lifecycle and URI parsing

Add the first iceberg_hive code that actually talks to a Hive Metastore
over Thrift, mirroring the structure of iceberg-rust's HmsCatalog::new
while keeping every Thrift type out of the public header via a pImpl.

  * HmsEndpoint + ParseHmsUris(): tolerant URI parser that accepts
    `thrift://host:port`, bare `host:port`, missing-port (defaults to
    9083), and comma-separated lists for HA failover, with whitespace
    around each segment stripped. Returns InvalidArgument for empty
    hosts, non-numeric or out-of-range ports, and empty list segments.

  * HmsClient::Connect(): wires TSocket -> TBufferedTransport /
    TFramedTransport (selected by HiveCatalogProperties::ThriftTransport)
    -> TBinaryProtocol -> ThriftHiveMetastoreClient. Connect / socket
    timeouts come from the properties. Connection failures are caught
    and translated into ErrorKind::kIOError so callers never see raw
    Thrift exceptions. The dtor best-effort-closes the transport.

  * HmsClient::Impl holds the Thrift state in destruction-order
    (client -> protocol -> transport -> socket) so teardown is clean.

  * src/iceberg/test/hms_client_test.cc adds 15 GoogleTest cases:
    11 covering ParseHmsUris (single, multi-HA, default port, scheme
    prefix, whitespace, empty/bad host/port edges) and 4 covering
    HmsClient::Connect's error paths (missing URI, bad URI, invalid
    transport mode, unreachable HMS).

  * src/iceberg/test/CMakeLists.txt gains an add_hive_iceberg_test()
    helper (mirroring add_rest_iceberg_test) and the hive_catalog_test
    target gated on ICEBERG_BUILD_HIVE.

`ctest --test-dir build --output-on-failure` now reports 17/17 passing
(16 previous + 1 new hive_catalog_test with 15 cases inside).

Part of the iceberg-cpp HiveCatalog port (C06).

Author: MisterRaindrop

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/hms_client.cc

@@ -0,0 +1,194 @@
+/*
+ * 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());
+
+  // HA failover beyond the first endpoint is left for a later commit.
+  const HmsEndpoint& endpoint = endpoints.front();
+  const int connect_timeout_ms = config.Get(HiveCatalogProperties::kConnectTimeoutMs);
+  const int socket_timeout_ms = config.Get(HiveCatalogProperties::kSocketTimeoutMs);
+
+  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

src/iceberg/catalog/hive/hms_client.h

@@ -0,0 +1,90 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "iceberg/catalog/hive/hive_catalog_properties.h"
+#include "iceberg/catalog/hive/iceberg_hive_export.h"
+#include "iceberg/result.h"
+
+/// \file iceberg/catalog/hive/hms_client.h
+/// \brief Thin wrapper around the generated Hive Metastore Thrift client.
+///
+/// Thrift types are kept out of the public interface via a pImpl.
+
+namespace iceberg::hive {
+
+/// \brief Hive's well-known metastore port, used when an HMS URI omits one.
+inline constexpr int kDefaultHmsPort = 9083;
+
+/// \brief A single host:port pair parsed from an HMS URI.
+///
+/// HMS URIs commonly take one of the forms:
+///   * `thrift://host:port`        (Java HiveCatalog convention)
+///   * `host:port`                 (iceberg-rust HmsCatalog convention)
+///   * comma-separated list of either form for HA failover
+///
+/// `port` defaults to 9083 (Hive's well-known metastore port) when the
+/// caller omits an explicit port.
+struct ICEBERG_HIVE_EXPORT HmsEndpoint {
+  std::string host;
+  int port = kDefaultHmsPort;
+};
+
+/// \brief Parse an HMS URI string into one or more endpoints.
+///
+/// Each comma-separated segment is treated as an independent endpoint;
+/// surrounding whitespace and an optional `thrift://` scheme prefix are
+/// stripped. Returns an InvalidArgument error if any segment fails to
+/// produce a non-empty host and a port within (0, 65535].
+ICEBERG_HIVE_EXPORT Result<std::vector<HmsEndpoint>> ParseHmsUris(std::string_view uri);
+
+/// \brief A live connection to a Hive Metastore over Thrift.
+///
+/// Construction goes through `Connect`, which parses the URI, selects the
+/// transport and opens it so that configuration errors surface as
+/// `iceberg::Error` rather than C++ exceptions.
+class ICEBERG_HIVE_EXPORT HmsClient {
+ public:
+  ~HmsClient();
+
+  HmsClient(const HmsClient&) = delete;
+  HmsClient& operator=(const HmsClient&) = delete;
+  HmsClient(HmsClient&&) = delete;
+  HmsClient& operator=(HmsClient&&) = delete;
+
+  /// \brief Connect to the Hive Metastore described by `config`.
+  ///
+  /// The first endpoint listed in `config.Uri()` is used; HA failover
+  /// to subsequent endpoints is left to a future commit.
+  static Result<std::unique_ptr<HmsClient>> Connect(const HiveCatalogProperties& config);
+
+ private:
+  class Impl;
+  std::unique_ptr<Impl> impl_;
+
+  explicit HmsClient(std::unique_ptr<Impl> impl);
+};
+
+}  // namespace iceberg::hive

src/iceberg/test/CMakeLists.txt

@@ -316,3 +316,27 @@ if(ICEBERG_BUILD_REST)
                           util/docker_compose_util.cc)
   endif()
 endif()
+
+if(ICEBERG_BUILD_HIVE)
+  function(add_hive_iceberg_test test_name)
+    set(options)
+    set(oneValueArgs)
+    set(multiValueArgs SOURCES)
+    cmake_parse_arguments(ARG
+                          "${options}"
+                          "${oneValueArgs}"
+                          "${multiValueArgs}"
+                          ${ARGN})
+
+    add_executable(${test_name})
+    target_include_directories(${test_name} PRIVATE "${CMAKE_BINARY_DIR}/iceberg/test/")
+    target_sources(${test_name} PRIVATE ${ARG_SOURCES})
+    target_link_libraries(${test_name} PRIVATE GTest::gmock_main iceberg_hive_static)
+    if(MSVC_TOOLCHAIN)
+      target_compile_options(${test_name} PRIVATE /bigobj)
+    endif()
+    add_test(NAME ${test_name} COMMAND ${test_name})
+  endfunction()
+
+  add_hive_iceberg_test(hive_catalog_test SOURCES hms_client_test.cc)
+endif()