Core: Propagate server error message in failed remote scan planning responses

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

@@ -219,7 +219,9 @@ private CloseableIterable<FileScanTask> planTableScan(PlanTableScanRequest planT
         return fetchPlanningResult();
       case FAILED:
         throw new IllegalStateException(
-            String.format("Received status: %s for planId: %s", PlanStatus.FAILED, planId));
+            String.format(
+                "Received status: %s for planId: %s. Server error: %s",
+                PlanStatus.FAILED, planId, response.errorMessage()));
       case CANCELLED:
         throw new IllegalStateException(
             String.format("Received status: %s for planId: %s", PlanStatus.CANCELLED, planId));
@@ -289,7 +291,8 @@ private CloseableIterable<FileScanTask> fetchPlanningResult() {
                 } else if (response.planStatus() != PlanStatus.COMPLETED) {
                   throw new IllegalStateException(
                       String.format(
-                          "Invalid planStatus: %s for planId: %s", response.planStatus(), id));
+                          "Invalid planStatus: %s for planId: %s. Server error: %s",
+                          response.planStatus(), id, response.errorMessage()));
                 }
 
                 result.set(response);

core/src/main/java/org/apache/iceberg/rest/responses/ErrorResponseParser.java

@@ -46,9 +46,13 @@ public static String toJson(ErrorResponse errorResponse, boolean pretty) {
   public static void toJson(ErrorResponse errorResponse, JsonGenerator generator)
       throws IOException {
     generator.writeStartObject();
+    writeErrorField(errorResponse, generator);
+    generator.writeEndObject();
+  }
 
+  public static void writeErrorField(ErrorResponse errorResponse, JsonGenerator generator)
+      throws IOException {
     generator.writeObjectFieldStart(ERROR);
-
     generator.writeStringField(MESSAGE, errorResponse.message());
     generator.writeStringField(TYPE, errorResponse.type());
     generator.writeNumberField(CODE, errorResponse.code());
@@ -57,8 +61,6 @@ public static void toJson(ErrorResponse errorResponse, JsonGenerator generator)
     }
 
     generator.writeEndObject();
-
-    generator.writeEndObject();
   }
 
   /**

core/src/main/java/org/apache/iceberg/rest/responses/FetchPlanningResultResponse.java

@@ -31,17 +31,20 @@
 
 public class FetchPlanningResultResponse extends BaseScanTaskResponse {
   private final PlanStatus planStatus;
+  private final ErrorResponse errorResponse;
   private final List<Credential> credentials;
 
   private FetchPlanningResultResponse(
       PlanStatus planStatus,
+      ErrorResponse errorResponse,
       List<String> planTasks,
       List<FileScanTask> fileScanTasks,
       List<DeleteFile> deleteFiles,
       Map<Integer, PartitionSpec> specsById,
       List<Credential> credentials) {
     super(planTasks, fileScanTasks, deleteFiles, specsById);
     this.planStatus = planStatus;
+    this.errorResponse = errorResponse;
     this.credentials = credentials;
     validate();
   }
@@ -50,6 +53,14 @@ public PlanStatus planStatus() {
     return planStatus;
   }
 
+  public ErrorResponse errorResponse() {
+    return errorResponse;
+  }
+
+  public String errorMessage() {
+    return errorResponse != null ? errorResponse.message() : null;
+  }
+
   public List<Credential> credentials() {
     return credentials != null ? credentials : ImmutableList.of();
   }
@@ -76,13 +87,19 @@ public static class Builder
     private Builder() {}
 
     private PlanStatus planStatus;
+    private ErrorResponse errorResponse;
     private final List<Credential> credentials = Lists.newArrayList();
 
     public Builder withPlanStatus(PlanStatus status) {
       this.planStatus = status;
       return this;
     }
 
+    public Builder withErrorResponse(ErrorResponse response) {
+      this.errorResponse = response;
+      return this;
+    }
+
     public Builder withCredentials(List<Credential> credentialsToAdd) {
       credentials.addAll(credentialsToAdd);
       return this;
@@ -91,7 +108,13 @@ public Builder withCredentials(List<Credential> credentialsToAdd) {
     @Override
     public FetchPlanningResultResponse build() {
       return new FetchPlanningResultResponse(
-          planStatus, planTasks(), fileScanTasks(), deleteFiles(), specsById(), credentials);
+          planStatus,
+          errorResponse,
+          planTasks(),
+          fileScanTasks(),
+          deleteFiles(),
+          specsById(),
+          credentials);
     }
   }
 }

core/src/main/java/org/apache/iceberg/rest/responses/FetchPlanningResultResponseParser.java

@@ -38,6 +38,7 @@ public class FetchPlanningResultResponseParser {
   private static final String STATUS = "status";
   private static final String PLAN_TASKS = "plan-tasks";
   private static final String STORAGE_CREDENTIALS = "storage-credentials";
+  private static final String ERROR = "error";
 
   private FetchPlanningResultResponseParser() {}
 
@@ -58,6 +59,11 @@ public static void toJson(FetchPlanningResultResponse response, JsonGenerator ge
         "Cannot serialize fileScanTasks in fetchingPlanningResultResponse without specsById");
     gen.writeStartObject();
     gen.writeStringField(STATUS, response.planStatus().status());
+
+    if (response.errorResponse() != null) {
+      ErrorResponseParser.writeErrorField(response.errorResponse(), gen);
+    }
+
     if (response.planTasks() != null) {
       JsonUtil.writeStringArray(PLAN_TASKS, response.planTasks(), gen);
     }
@@ -90,6 +96,11 @@ public static FetchPlanningResultResponse fromJson(
         json != null && !json.isEmpty(), "Invalid fetchPlanningResult response: null or empty");
 
     PlanStatus planStatus = PlanStatus.fromName(JsonUtil.getString(STATUS, json));
+    ErrorResponse errorResponse = null;
+    if (json.has(ERROR) && json.get(ERROR).isObject()) {
+      errorResponse = ErrorResponseParser.fromJson(json);
+    }
+
     List<String> planTasks = JsonUtil.getStringListOrNull(PLAN_TASKS, json);
     List<DeleteFile> deleteFiles = TableScanResponseParser.parseDeleteFiles(json, specsById);
     List<FileScanTask> fileScanTasks =
@@ -98,6 +109,7 @@ public static FetchPlanningResultResponse fromJson(
     FetchPlanningResultResponse.Builder builder =
         FetchPlanningResultResponse.builder()
             .withPlanStatus(planStatus)
+            .withErrorResponse(errorResponse)
             .withPlanTasks(planTasks)
             .withFileScanTasks(fileScanTasks)
             .withSpecsById(specsById);

core/src/main/java/org/apache/iceberg/rest/responses/PlanTableScanResponse.java

@@ -33,11 +33,13 @@
 public class PlanTableScanResponse extends BaseScanTaskResponse {
   private final PlanStatus planStatus;
   private final String planId;
+  private final ErrorResponse errorResponse;
   private final List<Credential> credentials;
 
   private PlanTableScanResponse(
       PlanStatus planStatus,
       String planId,
+      ErrorResponse errorResponse,
       List<String> planTasks,
       List<FileScanTask> fileScanTasks,
       List<DeleteFile> deleteFiles,
@@ -46,6 +48,7 @@ private PlanTableScanResponse(
     super(planTasks, fileScanTasks, deleteFiles, specsById);
     this.planStatus = planStatus;
     this.planId = planId;
+    this.errorResponse = errorResponse;
     this.credentials = credentials;
     validate();
   }
@@ -58,6 +61,14 @@ public String planId() {
     return planId;
   }
 
+  public ErrorResponse errorResponse() {
+    return errorResponse;
+  }
+
+  public String errorMessage() {
+    return errorResponse != null ? errorResponse.message() : null;
+  }
+
   public List<Credential> credentials() {
     return credentials != null ? credentials : ImmutableList.of();
   }
@@ -108,6 +119,7 @@ public static Builder builder() {
   public static class Builder extends BaseScanTaskResponse.Builder<Builder, PlanTableScanResponse> {
     private PlanStatus planStatus;
     private String planId;
+    private ErrorResponse errorResponse;
     private final List<Credential> credentials = Lists.newArrayList();
 
     /**
@@ -127,6 +139,11 @@ public Builder withPlanId(String id) {
       return this;
     }
 
+    public Builder withErrorResponse(ErrorResponse response) {
+      this.errorResponse = response;
+      return this;
+    }
+
     public Builder withCredentials(List<Credential> credentialsToAdd) {
       credentials.addAll(credentialsToAdd);
       return this;
@@ -137,6 +154,7 @@ public PlanTableScanResponse build() {
       return new PlanTableScanResponse(
           planStatus,
           planId,
+          errorResponse,
           planTasks(),
           fileScanTasks(),
           deleteFiles(),

core/src/main/java/org/apache/iceberg/rest/responses/PlanTableScanResponseParser.java

@@ -39,6 +39,7 @@ public class PlanTableScanResponseParser {
   private static final String PLAN_ID = "plan-id";
   private static final String PLAN_TASKS = "plan-tasks";
   private static final String STORAGE_CREDENTIALS = "storage-credentials";
+  private static final String ERROR = "error";
 
   private PlanTableScanResponseParser() {}
 
@@ -60,6 +61,10 @@ public static void toJson(PlanTableScanResponse response, JsonGenerator gen) thr
     gen.writeStartObject();
     gen.writeStringField(STATUS, response.planStatus().status());
 
+    if (response.errorResponse() != null) {
+      ErrorResponseParser.writeErrorField(response.errorResponse(), gen);
+    }
+
     if (response.planId() != null) {
       gen.writeStringField(PLAN_ID, response.planId());
     }
@@ -98,6 +103,11 @@ public static PlanTableScanResponse fromJson(
         "Cannot parse planTableScan response from empty or null object");
 
     PlanStatus planStatus = PlanStatus.fromName(JsonUtil.getString(STATUS, json));
+    ErrorResponse errorResponse = null;
+    if (json.has(ERROR) && json.get(ERROR).isObject()) {
+      errorResponse = ErrorResponseParser.fromJson(json);
+    }
+
     String planId = JsonUtil.getStringOrNull(PLAN_ID, json);
     List<String> planTasks = JsonUtil.getStringListOrNull(PLAN_TASKS, json);
     List<DeleteFile> deleteFiles = TableScanResponseParser.parseDeleteFiles(json, specsById);
@@ -108,6 +118,7 @@ public static PlanTableScanResponse fromJson(
         PlanTableScanResponse.builder()
             .withPlanId(planId)
             .withPlanStatus(planStatus)
+            .withErrorResponse(errorResponse)
             .withPlanTasks(planTasks)
             .withFileScanTasks(fileScanTasks)
             .withSpecsById(specsById);

core/src/test/java/org/apache/iceberg/rest/responses/TestFetchPlanningResultResponseParser.java

@@ -330,4 +330,45 @@ public void roundTripSerdeWithCredentials() {
     assertThat(FetchPlanningResultResponseParser.toJson(copyResponse, true))
         .isEqualTo(expectedJson);
   }
+
+  @Test
+  public void roundTripSerdeWithFailedStatusAndErrorResponse() {
+    ErrorResponse errorResponse =
+        ErrorResponse.builder()
+            .withMessage("Scan planning failed: table too large to plan")
+            .withType("IllegalStateException")
+            .responseCode(500)
+            .build();
+
+    FetchPlanningResultResponse response =
+        FetchPlanningResultResponse.builder()
+            .withPlanStatus(PlanStatus.FAILED)
+            .withErrorResponse(errorResponse)
+            .build();
+
+    String expectedJson =
+        "{\"status\":\"failed\","
+            + "\"error\":{\"message\":\"Scan planning failed: table too large to plan\","
+            + "\"type\":\"IllegalStateException\",\"code\":500}}";
+    String json = FetchPlanningResultResponseParser.toJson(response);
+    assertThat(json).isEqualTo(expectedJson);
+
+    FetchPlanningResultResponse fromResponse =
+        FetchPlanningResultResponseParser.fromJson(json, PARTITION_SPECS_BY_ID, false);
+    assertThat(fromResponse.planStatus()).isEqualTo(PlanStatus.FAILED);
+    assertThat(fromResponse.errorResponse()).isNotNull();
+    assertThat(fromResponse.errorResponse().message())
+        .isEqualTo("Scan planning failed: table too large to plan");
+    assertThat(fromResponse.errorResponse().type()).isEqualTo("IllegalStateException");
+    assertThat(fromResponse.errorResponse().code()).isEqualTo(500);
+  }
+
+  @Test
+  public void parseFailedStatusWithoutErrorObject() {
+    String json = "{\"status\":\"failed\"}";
+    FetchPlanningResultResponse response =
+        FetchPlanningResultResponseParser.fromJson(json, PARTITION_SPECS_BY_ID, false);
+    assertThat(response.planStatus()).isEqualTo(PlanStatus.FAILED);
+    assertThat(response.errorResponse()).isNull();
+  }
 }

core/src/test/java/org/apache/iceberg/rest/responses/TestPlanTableScanResponseParser.java

@@ -648,4 +648,46 @@ public void roundTripSerdeWithValidStatusAndFileScanTasksAndCredentials() {
 
     assertThat(PlanTableScanResponseParser.toJson(copyResponse, true)).isEqualTo(expectedJson);
   }
+
+  @Test
+  public void roundTripSerdeWithFailedStatusAndErrorResponse() {
+    ErrorResponse errorResponse =
+        ErrorResponse.builder()
+            .withMessage("Scan planning failed: table too large to plan")
+            .withType("IllegalStateException")
+            .responseCode(500)
+            .build();
+
+    PlanTableScanResponse response =
+        PlanTableScanResponse.builder()
+            .withPlanStatus(PlanStatus.FAILED)
+            .withErrorResponse(errorResponse)
+            .withSpecsById(PARTITION_SPECS_BY_ID)
+            .build();
+
+    String expectedJson =
+        "{\"status\":\"failed\","
+            + "\"error\":{\"message\":\"Scan planning failed: table too large to plan\","
+            + "\"type\":\"IllegalStateException\",\"code\":500}}";
+    String json = PlanTableScanResponseParser.toJson(response);
+    assertThat(json).isEqualTo(expectedJson);
+
+    PlanTableScanResponse fromResponse =
+        PlanTableScanResponseParser.fromJson(json, PARTITION_SPECS_BY_ID, false);
+    assertThat(fromResponse.planStatus()).isEqualTo(PlanStatus.FAILED);
+    assertThat(fromResponse.errorResponse()).isNotNull();
+    assertThat(fromResponse.errorResponse().message())
+        .isEqualTo("Scan planning failed: table too large to plan");
+    assertThat(fromResponse.errorResponse().type()).isEqualTo("IllegalStateException");
+    assertThat(fromResponse.errorResponse().code()).isEqualTo(500);
+  }
+
+  @Test
+  public void parseFailedStatusWithoutErrorObject() {
+    String json = "{\"status\":\"failed\"}";
+    PlanTableScanResponse response =
+        PlanTableScanResponseParser.fromJson(json, PARTITION_SPECS_BY_ID, false);
+    assertThat(response.planStatus()).isEqualTo(PlanStatus.FAILED);
+    assertThat(response.errorResponse()).isNull();
+  }
 }