Comments and stuff

Author: Sandeep Gottimukkala

src/iceberg/catalog.h

@@ -191,17 +191,39 @@ class ICEBERG_EXPORT Catalog {
   virtual Result<std::shared_ptr<Table>> RegisterTable(
       const TableIdentifier& identifier, const std::string& metadata_file_location) = 0;
 
+  /// \brief Initiate a scan planning operation for the given table.
+  ///
+  /// \param table The table to scan.
+  /// \param context The scan context containing snapshot, filter, and other options.
+  /// \return A PlanTableScanResponse with the plan status and initial scan tasks.
   virtual Result<rest::PlanTableScanResponse> PlanTableScan(
       const Table& table, const internal::TableScanContext& context) {
     return NotImplemented("PlanTableScan is not supported by this catalog");
   }
+
+  /// \brief Fetch the current status and results of an asynchronous scan plan.
+  ///
+  /// \param table The table being scanned.
+  /// \param plan_id The plan ID returned by PlanTableScan.
+  /// \return A FetchPlanningResultResponse with the current plan status and tasks.
   virtual Result<rest::FetchPlanningResultResponse> FetchPlanningResult(
       const Table& table, const std::string& plan_id) {
     return NotImplemented("FetchPlanningResult is not supported by this catalog");
   }
+
+  /// \brief Cancel an in-progress scan planning operation.
+  ///
+  /// \param table The table being scanned.
+  /// \param plan_id The plan ID returned by PlanTableScan.
   virtual Status CancelPlanning(const Table& table, const std::string& plan_id) {
     return NotImplemented("CancelPlanning is not supported by this catalog");
   }
+
+  /// \brief Fetch the scan tasks for a given plan task token.
+  ///
+  /// \param table The table being scanned.
+  /// \param plan_task The plan task token returned in a scan plan response.
+  /// \return A FetchScanTasksResponse with the file scan tasks.
   virtual Result<rest::FetchScanTasksResponse> FetchScanTasks(
       const Table& table, const std::string& plan_task) {
     return NotImplemented("FetchScanTasks is not supported by this catalog");

src/iceberg/catalog/rest/error_handlers.h

@@ -127,6 +127,7 @@ class ICEBERG_REST_EXPORT ViewCommitErrorHandler final : public DefaultErrorHand
   constexpr ViewCommitErrorHandler() = default;
 };
 
+/// \brief Scan plan operation error handler.
 class ICEBERG_REST_EXPORT ScanPlanErrorHandler final : public DefaultErrorHandler {
  public:
   static const std::shared_ptr<ScanPlanErrorHandler>& Instance();

src/iceberg/catalog/rest/json_serde.cc

@@ -605,6 +605,7 @@ Result<PlanTableScanResponse> PlanTableScanResponseFromJson(
   ICEBERG_ASSIGN_OR_RAISE(response.plan_id, GetJsonValue<std::string>(json, kPlanId));
   ICEBERG_RETURN_UNEXPECTED(
       BaseScanTaskResponseFromJson(json, &response, partition_specs_by_id, schema));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
   return response;
 }
 
@@ -618,6 +619,7 @@ Result<FetchPlanningResultResponse> FetchPlanningResultResponseFromJson(
   response.plan_status = PlanStatus(PlanStatus::FromString(status_str));
   ICEBERG_RETURN_UNEXPECTED(
       BaseScanTaskResponseFromJson(json, &response, partition_specs_by_id, schema));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
   return response;
 }
 
@@ -628,6 +630,7 @@ Result<FetchScanTasksResponse> FetchScanTasksResponseFromJson(
   FetchScanTasksResponse response;
   ICEBERG_RETURN_UNEXPECTED(
       BaseScanTaskResponseFromJson(json, &response, partition_specs_by_id, schema));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
   return response;
 }
 

src/iceberg/catalog/rest/resource_paths.h

@@ -81,8 +81,14 @@ class ICEBERG_REST_EXPORT ResourcePaths {
   /// \brief Get the /v1/{prefix}/transactions/commit endpoint path.
   Result<std::string> CommitTransaction() const;
 
+  /// \brief Get the /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan endpoint path.
   Result<std::string> ScanPlan(const TableIdentifier& ident) const;
+
+  /// \brief Get the /v1/{prefix}/namespaces/{namespace}/tables/{table}/plan/{plan_id}
+  /// endpoint path.
   Result<std::string> ScanPlan(const TableIdentifier& ident, const std::string& plan_id) const;
+
+  /// \brief Get the /v1/{prefix}/namespaces/{namespace}/tables/{table}/tasks endpoint path.
   Result<std::string> ScanTask(const TableIdentifier& ident) const;
 
  private:

src/iceberg/catalog/rest/types.cc

@@ -118,6 +118,52 @@ bool CommitTableResponse::operator==(const CommitTableResponse& other) const {
   return true;
 }
 
+bool PlanTableScanRequest::operator==(const PlanTableScanRequest& other) const {
+  return snapshot_id == other.snapshot_id && select == other.select &&
+         filter == other.filter && case_sensitive == other.case_sensitive &&
+         use_snapshot_schema == other.use_snapshot_schema &&
+         start_snapshot_id == other.start_snapshot_id &&
+         end_snapshot_id == other.end_snapshot_id && statsFields == other.statsFields &&
+         min_rows_required == other.min_rows_required;
+}
+
+bool BaseScanTaskResponse::operator==(const BaseScanTaskResponse& other) const {
+  if (plan_tasks != other.plan_tasks) return false;
+  if (delete_files != other.delete_files) return false;
+  if (file_scan_tasks.size() != other.file_scan_tasks.size()) return false;
+  for (size_t i = 0; i < file_scan_tasks.size(); ++i) {
+    const auto& a = file_scan_tasks[i];
+    const auto& b = other.file_scan_tasks[i];
+    if (!a.data_file() != !b.data_file()) return false;
+    if (a.data_file() && *a.data_file() != *b.data_file()) return false;
+    if (a.delete_files().size() != b.delete_files().size()) return false;
+    for (size_t j = 0; j < a.delete_files().size(); ++j) {
+      if (!a.delete_files()[j] != !b.delete_files()[j]) return false;
+      if (a.delete_files()[j] && *a.delete_files()[j] != *b.delete_files()[j]) return false;
+    }
+    if (a.residual_filter() != b.residual_filter()) return false;
+  }
+  return true;
+}
+
+bool PlanTableScanResponse::operator==(const PlanTableScanResponse& other) const {
+  return BaseScanTaskResponse::operator==(other) && plan_status == other.plan_status &&
+         plan_id == other.plan_id;
+}
+
+bool FetchPlanningResultResponse::operator==(const FetchPlanningResultResponse& other) const {
+  return BaseScanTaskResponse::operator==(other) &&
+         plan_status.ToString() == other.plan_status.ToString();
+}
+
+bool FetchScanTasksRequest::operator==(const FetchScanTasksRequest& other) const {
+  return planTask == other.planTask;
+}
+
+bool FetchScanTasksResponse::operator==(const FetchScanTasksResponse& other) const {
+  return BaseScanTaskResponse::operator==(other);
+}
+
 Status OAuthTokenResponse::Validate() const {
   if (access_token.empty()) {
     return ValidationFailed("OAuth2 token response missing required 'access_token'");

src/iceberg/json_serde.cc

@@ -253,35 +253,6 @@ constexpr std::string_view kReferencedDataFile = "referenced-data-file";
 constexpr std::string_view kContentOffset = "content-offset";
 constexpr std::string_view kContentSizeInBytes = "content-size-in-bytes";
 
-// Decode a base64-encoded string to raw bytes.
-std::vector<uint8_t> Base64Decode(std::string_view encoded) {
-  static const std::array<int, 256> kDecodeTable = [] {
-    std::array<int, 256> table;
-    table.fill(-1);
-    for (int i = 0; i < 26; i++) table[static_cast<unsigned char>('A') + i] = i;
-    for (int i = 0; i < 26; i++) table[static_cast<unsigned char>('a') + i] = 26 + i;
-    for (int i = 0; i < 10; i++) table[static_cast<unsigned char>('0') + i] = 52 + i;
-    table[static_cast<unsigned char>('+')] = 62;
-    table[static_cast<unsigned char>('/')] = 63;
-    return table;
-  }();
-
-  std::vector<uint8_t> decoded;
-  decoded.reserve(encoded.size() * 3 / 4);
-  int val = 0, bits = -8;
-  for (unsigned char c : encoded) {
-    if (c == '=') break;
-    const int d = kDecodeTable[c];
-    if (d == -1) continue;
-    val = (val << 6) + d;
-    bits += 6;
-    if (bits >= 0) {
-      decoded.push_back(static_cast<uint8_t>((val >> bits) & 0xFF));
-      bits -= 8;
-    }
-  }
-  return decoded;
-}
 
 }  // namespace
 
@@ -1866,21 +1837,20 @@ Result<DataFile> DataFileFromJson(
   ICEBERG_RETURN_UNEXPECTED(parse_int_map(kNullValueCounts, df.null_value_counts));
   ICEBERG_RETURN_UNEXPECTED(parse_int_map(kNanValueCounts, df.nan_value_counts));
 
-  // Parse BinaryMap: {"keys": [int, ...], "values": [base64string, ...]}
+  // Parse BinaryMap: {"keys": [int, ...], "values": [base64 binary, ...]}
   auto parse_binary_map = [&](std::string_view key,
                               std::map<int32_t, std::vector<uint8_t>>& target) -> Status {
     if (!json.contains(key) || json.at(key).is_null()) return {};
     ICEBERG_ASSIGN_OR_RAISE(auto map_json, GetJsonValue<nlohmann::json>(json, key));
     ICEBERG_ASSIGN_OR_RAISE(auto keys,
-                            GetTypedJsonValue<std::vector<int32_t>>(map_json.at("keys")));
-    ICEBERG_ASSIGN_OR_RAISE(
-        auto values,
-        GetTypedJsonValue<std::vector<std::string>>(map_json.at("values")));
+                            GetJsonValue<std::vector<int32_t>>(map_json, "keys"));
+    ICEBERG_ASSIGN_OR_RAISE(auto values,
+                            GetJsonValue<std::vector<std::vector<uint8_t>>>(map_json, "values"));
     if (keys.size() != values.size()) {
       return JsonParseError("'{}' binary map keys and values have different lengths", key);
     }
     for (size_t i = 0; i < keys.size(); ++i) {
-      target[keys[i]] = Base64Decode(values[i]);
+      target[keys[i]] = values[i];
     }
     return {};
   };
@@ -1889,9 +1859,8 @@ Result<DataFile> DataFileFromJson(
   ICEBERG_RETURN_UNEXPECTED(parse_binary_map(kUpperBounds, df.upper_bounds));
 
   if (json.contains(kKeyMetadata) && !json.at(kKeyMetadata).is_null()) {
-    ICEBERG_ASSIGN_OR_RAISE(auto key_meta_str,
-                            GetJsonValue<std::string>(json, kKeyMetadata));
-    df.key_metadata = Base64Decode(key_meta_str);
+    ICEBERG_ASSIGN_OR_RAISE(df.key_metadata,
+                            GetJsonValue<std::vector<uint8_t>>(json, kKeyMetadata));
   }
   if (json.contains(kSplitOffsets) && !json.at(kSplitOffsets).is_null()) {
     ICEBERG_ASSIGN_OR_RAISE(df.split_offsets,