[CORE][REST]: Add context aware response parsing

.palantir/revapi.yml

@@ -1285,6 +1285,12 @@ acceptedBreaks:
     - code: "java.field.removedWithConstant"
       old: "field org.apache.iceberg.TableProperties.ROW_LINEAGE"
       justification: "Removing deprecations for 1.10.0"
+    - code: "java.method.abstractMethodAdded"
+      new: "method <T extends org.apache.iceberg.rest.RESTResponse> T org.apache.iceberg.rest.BaseHTTPClient::execute(org.apache.iceberg.rest.HTTPRequest,\
+        \ java.lang.Class<T>, java.util.function.Consumer<org.apache.iceberg.rest.responses.ErrorResponse>,\
+        \ java.util.function.Consumer<java.util.Map<java.lang.String, java.lang.String>>,\
+        \ org.apache.iceberg.rest.ParserContext)"
+      justification: "Add context aware parsing"
     - code: "java.method.removed"
       old: "method boolean org.apache.iceberg.TableMetadata::rowLineageEnabled()"
       justification: "Removing deprecations for 1.10.0"

core/src/main/java/org/apache/iceberg/rest/BaseHTTPClient.java

@@ -77,6 +77,18 @@ public <T extends RESTResponse> T get(
     return execute(request, responseType, errorHandler, h -> {});
   }
 
+  @Override
+  public <T extends RESTResponse> T get(
+      String path,
+      Map<String, String> queryParams,
+      Class<T> responseType,
+      Map<String, String> headers,
+      Consumer<ErrorResponse> errorHandler,
+      ParserContext parserContext) {
+    HTTPRequest request = buildRequest(HTTPMethod.GET, path, queryParams, headers, null);
+    return execute(request, responseType, errorHandler, h -> {}, parserContext);
+  }
+
   @Override
   public <T extends RESTResponse> T post(
       String path,
@@ -100,6 +112,19 @@ public <T extends RESTResponse> T post(
     return execute(request, responseType, errorHandler, responseHeaders);
   }
 
+  @Override
+  public <T extends RESTResponse> T post(
+      String path,
+      RESTRequest body,
+      Class<T> responseType,
+      Map<String, String> headers,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders,
+      ParserContext parserContext) {
+    HTTPRequest request = buildRequest(HTTPMethod.POST, path, null, headers, body);
+    return execute(request, responseType, errorHandler, responseHeaders, parserContext);
+  }
+
   @Override
   public <T extends RESTResponse> T postForm(
       String path,
@@ -123,4 +148,11 @@ protected abstract <T extends RESTResponse> T execute(
       Class<T> responseType,
       Consumer<ErrorResponse> errorHandler,
       Consumer<Map<String, String>> responseHeaders);
+
+  protected abstract <T extends RESTResponse> T execute(
+      HTTPRequest request,
+      Class<T> responseType,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders,
+      ParserContext parserContext);
 }

core/src/main/java/org/apache/iceberg/rest/HTTPClient.java

@@ -20,11 +20,13 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.net.URI;
 import java.nio.charset.StandardCharsets;
 import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import org.apache.hc.client5.http.auth.AuthScope;
@@ -101,6 +103,7 @@ public class HTTPClient extends BaseHTTPClient {
   private final ObjectMapper mapper;
   private final AuthSession authSession;
   private final boolean isRootClient;
+  private final ConcurrentMap<Class<?>, ObjectReader> objectReaderCache = Maps.newConcurrentMap();
 
   private HTTPClient(
       URI baseUri,
@@ -303,6 +306,17 @@ protected <T extends RESTResponse> T execute(
       Class<T> responseType,
       Consumer<ErrorResponse> errorHandler,
       Consumer<Map<String, String>> responseHeaders) {
+    return execute(
+        req, responseType, errorHandler, responseHeaders, ParserContext.builder().build());
+  }
+
+  @Override
+  protected <T extends RESTResponse> T execute(
+      HTTPRequest req,
+      Class<T> responseType,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders,
+      ParserContext parserContext) {
     HttpUriRequestBase request = new HttpUriRequestBase(req.method().name(), req.requestUri());
 
     req.headers().entries().forEach(e -> request.addHeader(e.name(), e.value()));
@@ -341,7 +355,11 @@ protected <T extends RESTResponse> T execute(
       }
 
       try {
-        return mapper.readValue(responseBody, responseType);
+        ObjectReader reader = objectReaderCache.computeIfAbsent(responseType, mapper::readerFor);
+        if (parserContext != null && !parserContext.isEmpty()) {
+          reader = reader.with(parserContext.toInjectableValues());
+        }
+        return reader.readValue(responseBody);
       } catch (JsonProcessingException e) {
         throw new RESTException(
             e,

core/src/main/java/org/apache/iceberg/rest/ParserContext.java

@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.iceberg.rest;
+
+import com.fasterxml.jackson.databind.InjectableValues;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.hadoop.util.Preconditions;
+
+class ParserContext {
+
+  private final Map<String, Object> data;
+
+  private ParserContext(Builder builder) {
+    this.data = Collections.unmodifiableMap(builder.data);
+  }
+
+  public boolean isEmpty() {
+    return data.isEmpty();
+  }
+
+  public InjectableValues toInjectableValues() {
+    return new InjectableValues.Std(data);
+  }
+
+  static Builder builder() {
+    return new Builder();
+  }
+
+  static class Builder {
+    private Map<String, Object> data;
+
+    private Builder() {
+      this.data = Collections.emptyMap();
+    }
+
+    public Builder add(String key, Object value) {
+      Preconditions.checkNotNull(key, "Key cannot be null");
+      Preconditions.checkNotNull(value, "Value cannot be null");
+      this.data.put(key, value);
+      return this;
+    }
+
+    public ParserContext build() {
+      return new ParserContext(this);
+    }
+  }
+}

core/src/main/java/org/apache/iceberg/rest/RESTClient.java

@@ -97,6 +97,19 @@ default <T extends RESTResponse> T get(
     return get(path, queryParams, responseType, headers.get(), errorHandler);
   }
 
+  default <T extends RESTResponse> T get(
+      String path,
+      Map<String, String> queryParams,
+      Class<T> responseType,
+      Map<String, String> headers,
+      Consumer<ErrorResponse> errorHandler,
+      ParserContext parserContext) {
+    if (parserContext != null) {
+      throw new UnsupportedOperationException("Parser context is not supported");
+    }
+    return get(path, queryParams, responseType, headers, errorHandler);
+  }
+
   <T extends RESTResponse> T get(
       String path,
       Map<String, String> queryParams,
@@ -123,6 +136,20 @@ default <T extends RESTResponse> T post(
     return post(path, body, responseType, headers.get(), errorHandler, responseHeaders);
   }
 
+  default <T extends RESTResponse> T post(
+      String path,
+      RESTRequest body,
+      Class<T> responseType,
+      Map<String, String> headers,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders,
+      ParserContext parserContext) {
+    if (parserContext != null) {
+      throw new UnsupportedOperationException("Parser context is not supported");
+    }
+    return post(path, body, responseType, headers, errorHandler, responseHeaders);
+  }
+
   default <T extends RESTResponse> T post(
       String path,
       RESTRequest body,

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -606,6 +606,17 @@ protected <T extends RESTResponse> T execute(
       Class<T> responseType,
       Consumer<ErrorResponse> errorHandler,
       Consumer<Map<String, String>> responseHeaders) {
+    return execute(
+        request, responseType, errorHandler, responseHeaders, ParserContext.builder().build());
+  }
+
+  @Override
+  protected <T extends RESTResponse> T execute(
+      HTTPRequest request,
+      Class<T> responseType,
+      Consumer<ErrorResponse> errorHandler,
+      Consumer<Map<String, String>> responseHeaders,
+      ParserContext parserContext) {
     ErrorResponse.Builder errorBuilder = ErrorResponse.builder();
     Pair<Route, Map<String, String>> routeAndVars = Route.from(request.method(), request.path());
     if (routeAndVars != null) {