feat(rest): add session-aware REST catalog

Add SessionContext and SessionCatalog APIs, make RestCatalog a session-aware
root, and bind REST catalog operations through default, contextual, and
table-scoped catalog views.

Route contextual and table auth through AuthManager, preserve table response
config for scoped catalogs, and fail closed on unsupported table FileIO config.

Author: wgtmac

src/iceberg/CMakeLists.txt

@@ -21,6 +21,7 @@ set(ICEBERG_SOURCES
     arrow_c_data_util.cc
     arrow_c_data_guard_internal.cc
     catalog/memory/in_memory_catalog.cc
+    catalog/session_context.cc
     delete_file_index.cc
     expression/aggregate.cc
     expression/binder.cc

src/iceberg/catalog/meson.build

@@ -17,6 +17,11 @@
 
 subdir('memory')
 
+install_headers(
+    ['session_catalog.h', 'session_context.h'],
+    subdir: 'iceberg/catalog',
+)
+
 if get_option('rest').enabled()
     subdir('rest')
 endif

src/iceberg/catalog/rest/auth/auth_manager.cc

@@ -38,8 +38,7 @@ Result<std::shared_ptr<AuthSession>> AuthManager::InitSession(
 }
 
 Result<std::shared_ptr<AuthSession>> AuthManager::ContextualSession(
-    [[maybe_unused]] const std::unordered_map<std::string, std::string>& context,
-    std::shared_ptr<AuthSession> parent) {
+    [[maybe_unused]] const SessionContext& context, std::shared_ptr<AuthSession> parent) {
   // By default, return the parent session as-is
   return parent;
 }

src/iceberg/catalog/rest/auth/auth_manager.h

@@ -70,13 +70,13 @@ class ICEBERG_REST_EXPORT AuthManager {
   /// This method is used by SessionCatalog to create sessions for different contexts
   /// (e.g., different users or tenants).
   ///
-  /// \param context Context properties (e.g., user credentials, tenant info).
+  /// \param context Session context (e.g., session ID, identity, credentials,
+  /// properties).
   /// \param parent Catalog session to inherit from or return as-is.
   /// \return A context-specific session, or the parent session if no context-specific
   /// session is needed, or an error if session creation fails.
   virtual Result<std::shared_ptr<AuthSession>> ContextualSession(
-      const std::unordered_map<std::string, std::string>& context,
-      std::shared_ptr<AuthSession> parent);
+      const SessionContext& context, std::shared_ptr<AuthSession> parent);
 
   /// \brief Create or reuse a session scoped to a single table/view.
   ///

src/iceberg/catalog/rest/auth/oauth2_util.cc

@@ -21,6 +21,7 @@
 
 #include <nlohmann/json.hpp>
 
+#include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/error_handlers.h"
 #include "iceberg/catalog/rest/http_client.h"

src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h

@@ -126,8 +126,7 @@ class ICEBERG_REST_EXPORT SigV4AuthManager : public AuthManager {
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::shared_ptr<AuthSession>> ContextualSession(
-      const std::unordered_map<std::string, std::string>& context,
-      std::shared_ptr<AuthSession> parent) override;
+      const SessionContext& context, std::shared_ptr<AuthSession> parent) override;
 
   Result<std::shared_ptr<AuthSession>> TableSession(
       const TableIdentifier& table,

src/iceberg/catalog/rest/auth/sigv4_manager.cc

@@ -19,6 +19,7 @@
 
 #include "iceberg/catalog/rest/auth/auth_manager_internal.h"
 #include "iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h"
+#include "iceberg/catalog/session_context.h"
 #include "iceberg/result.h"
 
 #if ICEBERG_SIGV4_ENABLED
@@ -146,6 +147,19 @@ std::unordered_map<std::string, std::string> MergeProperties(
   return merged;
 }
 
+Result<std::unordered_map<std::string, std::string>> ContextProperties(
+    const SessionContext& context) {
+  auto merged = context.properties;
+  for (const auto& [key, value] : context.credentials) {
+    auto [it, inserted] = merged.emplace(key, value);
+    if (!inserted && it->second != value) {
+      return InvalidArgument("Session context has conflicting values for property '{}'",
+                             key);
+    }
+  }
+  return merged;
+}
+
 /// Matches Java RESTSigV4AuthSession: canonical headers carry
 /// Base64(SHA256(body)), canonical request trailer uses hex.
 class RestSigV4Signer : public Aws::Client::AWSAuthV4Signer {
@@ -386,19 +400,19 @@ Result<std::shared_ptr<AuthSession>> SigV4AuthManager::CatalogSession(
 }
 
 Result<std::shared_ptr<AuthSession>> SigV4AuthManager::ContextualSession(
-    const std::unordered_map<std::string, std::string>& context,
-    std::shared_ptr<AuthSession> parent) {
+    const SessionContext& context, std::shared_ptr<AuthSession> parent) {
   auto sigv4_parent = std::dynamic_pointer_cast<SigV4AuthSession>(std::move(parent));
   ICEBERG_PRECHECK(sigv4_parent != nullptr,
                    "SigV4AuthManager parent must be a SigV4AuthSession");
 
   ICEBERG_ASSIGN_OR_RAISE(auto delegate_session, delegate_->ContextualSession(
                                                      context, sigv4_parent->delegate()));
 
-  auto merged = MergeProperties(catalog_properties_, context);
+  ICEBERG_ASSIGN_OR_RAISE(auto context_properties, ContextProperties(context));
+  auto merged = MergeProperties(catalog_properties_, context_properties);
   ICEBERG_ASSIGN_OR_RAISE(
-      auto credentials,
-      ResolveCredentialsProvider(context, sigv4_parent->credentials_provider()));
+      auto credentials, ResolveCredentialsProvider(context_properties,
+                                                   sigv4_parent->credentials_provider()));
   return WrapSession(std::move(delegate_session), merged, std::move(credentials));
 }