feat(rest): support snapshot loading mode (#543)

Author: HeartLinked

src/iceberg/catalog/rest/catalog_properties.cc

@@ -19,6 +19,8 @@
 
 #include "iceberg/catalog/rest/catalog_properties.h"
 
+#include <algorithm>
+#include <string>
 #include <string_view>
 
 namespace iceberg::rest {
@@ -47,4 +49,16 @@ Result<std::string_view> RestCatalogProperties::Uri() const {
   return it->second;
 }
 
+Result<SnapshotMode> RestCatalogProperties::SnapshotLoadingMode() const {
+  std::string upper = StringUtils::ToUpper(Get(kSnapshotLoadingMode));
+  if (upper == "ALL") {
+    return SnapshotMode::kAll;
+  } else if (upper == "REFS") {
+    return SnapshotMode::kRefs;
+  } else {
+    return InvalidArgument("Invalid snapshot loading mode: '{}'.",
+                           Get(kSnapshotLoadingMode));
+  }
+}
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/catalog_properties.h

@@ -31,6 +31,9 @@
 
 namespace iceberg::rest {
 
+/// \brief Snapshot loading mode for REST catalog.
+enum class SnapshotMode : uint8_t { kAll, kRefs };
+
 /// \brief Configuration class for a REST Catalog.
 class ICEBERG_REST_EXPORT RestCatalogProperties
     : public ConfigBase<RestCatalogProperties> {
@@ -46,6 +49,8 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   inline static Entry<std::string> kWarehouse{"warehouse", ""};
   /// \brief The optional prefix for REST API paths.
   inline static Entry<std::string> kPrefix{"prefix", ""};
+  /// \brief The snapshot loading mode (ALL or REFS).
+  inline static Entry<std::string> kSnapshotLoadingMode{"snapshot-loading-mode", "ALL"};
   /// \brief The prefix for HTTP headers.
   inline static constexpr std::string_view kHeaderPrefix = "header.";
 
@@ -62,6 +67,12 @@ class ICEBERG_REST_EXPORT RestCatalogProperties
   /// \brief Get the URI of the REST catalog server.
   /// \return The URI if configured, or an error if not set or empty.
   Result<std::string_view> Uri() const;
+
+  /// \brief Get the snapshot loading mode.
+  /// \return SnapshotMode::kAll if configured as "ALL", SnapshotMode::kRefs if
+  ///         "REFS", or an error if the value is invalid. Parsing is
+  ///         case-insensitive to match Java behavior.
+  Result<SnapshotMode> SnapshotLoadingMode() const;
 };
 
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/rest_catalog.cc

@@ -161,29 +161,35 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
       paths, ResourcePaths::Make(std::string(TrimTrailingSlash(final_uri)),
                                  final_config.Get(RestCatalogProperties::kPrefix)));
 
+  // Get snapshot loading mode
+  ICEBERG_ASSIGN_OR_RAISE(auto snapshot_mode, final_config.SnapshotLoadingMode());
+
   auto client = std::make_unique<HttpClient>(final_config.ExtractHeaders());
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
                           auth_manager->CatalogSession(*client, final_config.configs()));
 
-  return std::shared_ptr<RestCatalog>(new RestCatalog(
-      std::move(final_config), std::move(file_io), std::move(client), std::move(paths),
-      std::move(endpoints), std::move(auth_manager), std::move(catalog_session)));
+  return std::shared_ptr<RestCatalog>(
+      new RestCatalog(std::move(final_config), std::move(file_io), std::move(client),
+                      std::move(paths), std::move(endpoints), std::move(auth_manager),
+                      std::move(catalog_session), snapshot_mode));
 }
 
 RestCatalog::RestCatalog(RestCatalogProperties config, std::shared_ptr<FileIO> file_io,
                          std::unique_ptr<HttpClient> client,
                          std::unique_ptr<ResourcePaths> paths,
                          std::unordered_set<Endpoint> endpoints,
                          std::unique_ptr<auth::AuthManager> auth_manager,
-                         std::shared_ptr<auth::AuthSession> catalog_session)
+                         std::shared_ptr<auth::AuthSession> catalog_session,
+                         SnapshotMode snapshot_mode)
     : config_(std::move(config)),
       file_io_(std::move(file_io)),
       client_(std::move(client)),
       paths_(std::move(paths)),
       name_(config_.Get(RestCatalogProperties::kName)),
       supported_endpoints_(std::move(endpoints)),
       auth_manager_(std::move(auth_manager)),
-      catalog_session_(std::move(catalog_session)) {
+      catalog_session_(std::move(catalog_session)),
+      snapshot_mode_(snapshot_mode) {
   ICEBERG_DCHECK(catalog_session_ != nullptr, "catalog_session must not be null");
 }
 
@@ -442,9 +448,17 @@ Result<std::string> RestCatalog::LoadTableInternal(
     const TableIdentifier& identifier) const {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::LoadTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
+
+  std::unordered_map<std::string, std::string> params;
+  if (snapshot_mode_ == SnapshotMode::kRefs) {
+    params["snapshots"] = "refs";
+  } else {
+    params["snapshots"] = "all";
+  }
+
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Get(path, /*params=*/{}, /*headers=*/{}, *TableErrorHandler::Instance(),
+      client_->Get(path, params, /*headers=*/{}, *TableErrorHandler::Instance(),
                    *catalog_session_));
   return response.body();
 }
@@ -453,7 +467,6 @@ Result<std::shared_ptr<Table>> RestCatalog::LoadTable(const TableIdentifier& ide
   ICEBERG_ASSIGN_OR_RAISE(const auto body, LoadTableInternal(identifier));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(body));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));
-
   return Table::Make(identifier, std::move(load_result.metadata),
                      std::move(load_result.metadata_location), file_io_,
                      shared_from_this());

src/iceberg/catalog/rest/rest_catalog.h

@@ -109,7 +109,8 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
               std::unique_ptr<HttpClient> client, std::unique_ptr<ResourcePaths> paths,
               std::unordered_set<Endpoint> endpoints,
               std::unique_ptr<auth::AuthManager> auth_manager,
-              std::shared_ptr<auth::AuthSession> catalog_session);
+              std::shared_ptr<auth::AuthSession> catalog_session,
+              SnapshotMode snapshot_mode);
 
   Result<std::string> LoadTableInternal(const TableIdentifier& identifier) const;
 
@@ -127,6 +128,7 @@ class ICEBERG_REST_EXPORT RestCatalog : public Catalog,
   std::unordered_set<Endpoint> supported_endpoints_;
   std::unique_ptr<auth::AuthManager> auth_manager_;
   std::shared_ptr<auth::AuthSession> catalog_session_;
+  SnapshotMode snapshot_mode_;
 };
 
 }  // namespace iceberg::rest

src/iceberg/test/CMakeLists.txt

@@ -225,7 +225,7 @@ if(ICEBERG_BUILD_REST)
   if(ICEBERG_BUILD_REST_INTEGRATION_TESTS)
     add_rest_iceberg_test(rest_catalog_integration_test
                           SOURCES
-                          rest_catalog_test.cc
+                          rest_catalog_integration_test.cc
                           util/cmd_util.cc
                           util/docker_compose_util.cc)
   endif()

src/iceberg/test/meson.build

@@ -120,7 +120,7 @@ if get_option('rest').enabled()
             iceberg_tests += {
                 'rest_integration_test': {
                     'sources': files(
-                        'rest_catalog_test.cc',
+                        'rest_catalog_integration_test.cc',
                         'util/cmd_util.cc',
                         'util/docker_compose_util.cc',
                     ),