feat(rest): add scan plan endpoints to REST catalog client (#614)

* Introduced 4 endpoints: planning table scan, fetching a scan result,
cancelling a scan plan and fetching scan planning tasks (endpoint.h)
* Introduced corresponding request and response objects (types.h)
* Implemented handlers (catalog.h). The handlers construct a request
object and return a response object
- Handlers use newly introduced helper methods to serialize and
deserialize request response json objects
* Added unit tests for all the above helper methods and handler methods
* Added integration tests for the handlers

Author: gsandeep1241

src/iceberg/catalog/rest/endpoint.h

@@ -128,6 +128,26 @@ class ICEBERG_REST_EXPORT Endpoint {
     return {HttpMethod::kPost, "/v1/{prefix}/transactions/commit"};
   }
 
+  // Scan planning endpoints
+  static Endpoint PlanTableScan() {
+    return {HttpMethod::kPost, "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan"};
+  }
+
+  static Endpoint FetchPlanningResult() {
+    return {HttpMethod::kGet,
+            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}"};
+  }
+
+  static Endpoint CancelPlanning() {
+    return {HttpMethod::kDelete,
+            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan-id}"};
+  }
+
+  static Endpoint FetchScanTasks() {
+    return {HttpMethod::kPost,
+            "/v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks"};
+  }
+
  private:
   Endpoint(HttpMethod method, std::string_view path) : method_(method), path_(path) {}
 

src/iceberg/catalog/rest/error_handlers.cc

@@ -30,6 +30,9 @@ namespace {
 constexpr std::string_view kIllegalArgumentException = "IllegalArgumentException";
 constexpr std::string_view kNoSuchNamespaceException = "NoSuchNamespaceException";
 constexpr std::string_view kNamespaceNotEmptyException = "NamespaceNotEmptyException";
+constexpr std::string_view kNoSuchTableException = "NoSuchTableException";
+constexpr std::string_view kNoSuchPlanIdException = "NoSuchPlanIdException";
+constexpr std::string_view kNoSuchPlanTaskException = "NoSuchPlanTaskException";
 
 }  // namespace
 
@@ -183,4 +186,52 @@ Status ViewCommitErrorHandler::Accept(const ErrorResponse& error) const {
   return DefaultErrorHandler::Accept(error);
 }
 
+const std::shared_ptr<PlanErrorHandler>& PlanErrorHandler::Instance() {
+  static const std::shared_ptr<PlanErrorHandler> instance{new PlanErrorHandler()};
+  return instance;
+}
+
+Status PlanErrorHandler::Accept(const ErrorResponse& error) const {
+  switch (error.code) {
+    case 404:
+      if (error.type == kNoSuchNamespaceException) {
+        return NoSuchNamespace(error.message);
+      }
+      if (error.type == kNoSuchTableException) {
+        return NoSuchTable(error.message);
+      }
+      if (error.type == kNoSuchPlanIdException) {
+        return NoSuchPlanId(error.message);
+      }
+      return NotFound(error.message);
+    case 406:
+      return NotSupported(error.message);
+  }
+
+  return DefaultErrorHandler::Accept(error);
+}
+
+const std::shared_ptr<PlanTaskErrorHandler>& PlanTaskErrorHandler::Instance() {
+  static const std::shared_ptr<PlanTaskErrorHandler> instance{new PlanTaskErrorHandler()};
+  return instance;
+}
+
+Status PlanTaskErrorHandler::Accept(const ErrorResponse& error) const {
+  switch (error.code) {
+    case 404:
+      if (error.type == kNoSuchNamespaceException) {
+        return NoSuchNamespace(error.message);
+      }
+      if (error.type == kNoSuchTableException) {
+        return NoSuchTable(error.message);
+      }
+      if (error.type == kNoSuchPlanTaskException) {
+        return NoSuchPlanTask(error.message);
+      }
+      return NotFound(error.message);
+  }
+
+  return DefaultErrorHandler::Accept(error);
+}
+
 }  // namespace iceberg::rest

src/iceberg/catalog/rest/error_handlers.h

@@ -127,4 +127,26 @@ class ICEBERG_REST_EXPORT ViewCommitErrorHandler final : public DefaultErrorHand
   constexpr ViewCommitErrorHandler() = default;
 };
 
+/// \brief Plan operation error handler.
+class ICEBERG_REST_EXPORT PlanErrorHandler final : public DefaultErrorHandler {
+ public:
+  static const std::shared_ptr<PlanErrorHandler>& Instance();
+
+  Status Accept(const ErrorResponse& error) const override;
+
+ private:
+  constexpr PlanErrorHandler() = default;
+};
+
+/// \brief Fetch scan tasks operation error handler.
+class ICEBERG_REST_EXPORT PlanTaskErrorHandler final : public DefaultErrorHandler {
+ public:
+  static const std::shared_ptr<PlanTaskErrorHandler>& Instance();
+
+  Status Accept(const ErrorResponse& error) const override;
+
+ private:
+  constexpr PlanTaskErrorHandler() = default;
+};
+
 }  // namespace iceberg::rest