Core: Request/Response models and parsers for REST Scan Planning

core/src/main/java/org/apache/iceberg/ContentFileParser.java

@@ -27,7 +27,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.JsonUtil;
 
-class ContentFileParser {
+public class ContentFileParser {
   private static final String SPEC_ID = "spec-id";
   private static final String CONTENT = "content";
   private static final String FILE_PATH = "file-path";
@@ -56,12 +56,12 @@ private static boolean hasPartitionData(StructLike partitionData) {
     return partitionData != null && partitionData.size() > 0;
   }
 
-  static String toJson(ContentFile<?> contentFile, PartitionSpec spec) {
+  public static String toJson(ContentFile<?> contentFile, PartitionSpec spec) {
     return JsonUtil.generate(
         generator -> ContentFileParser.toJson(contentFile, spec, generator), false);
   }
 
-  static void toJson(ContentFile<?> contentFile, PartitionSpec spec, JsonGenerator generator)
+  public static void toJson(ContentFile<?> contentFile, PartitionSpec spec, JsonGenerator generator)
       throws IOException {
     Preconditions.checkArgument(contentFile != null, "Invalid content file: null");
     Preconditions.checkArgument(spec != null, "Invalid partition spec: null");
@@ -134,13 +134,18 @@ static void toJson(ContentFile<?> contentFile, PartitionSpec spec, JsonGenerator
     generator.writeEndObject();
   }
 
-  static ContentFile<?> fromJson(JsonNode jsonNode, PartitionSpec spec) {
+  public static ContentFile<?> fromJson(JsonNode jsonNode, PartitionSpec spec) {
+    return fromJson(jsonNode, spec == null ? null : Map.of(spec.specId(), spec));
+  }
+
+  public static ContentFile<?> fromJson(JsonNode jsonNode, Map<Integer, PartitionSpec> specsById) {
     Preconditions.checkArgument(jsonNode != null, "Invalid JSON node for content file: null");
     Preconditions.checkArgument(
         jsonNode.isObject(), "Invalid JSON node for content file: non-object (%s)", jsonNode);
-    Preconditions.checkArgument(spec != null, "Invalid partition spec: null");
-
+    Preconditions.checkArgument(specsById != null, "Invalid partition spec: null");
     int specId = JsonUtil.getInt(SPEC_ID, jsonNode);
+    PartitionSpec spec = specsById.get(specId);
+    Preconditions.checkArgument(spec != null, "Invalid partition specId: %s", specId);
     FileContent fileContent = FileContent.valueOf(JsonUtil.getString(CONTENT, jsonNode));
     String filePath = JsonUtil.getString(FILE_PATH, jsonNode);
     FileFormat fileFormat = FileFormat.fromString(JsonUtil.getString(FILE_FORMAT, jsonNode));

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

@@ -0,0 +1,48 @@
+/*
+ * 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 java.util.Locale;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+public enum PlanStatus {
+  COMPLETED("completed"),
+  SUBMITTED("submitted"),
+  CANCELLED("cancelled"),
+  FAILED("failed");
+
+  private final String status;
+
+  PlanStatus(String status) {
+    this.status = status;
+  }
+
+  public String status() {
+    return status;
+  }
+
+  public static PlanStatus fromName(String status) {
+    Preconditions.checkArgument(status != null, "Status is null");
+    try {
+      return PlanStatus.valueOf(status.toUpperCase(Locale.ROOT));
+    } catch (IllegalArgumentException e) {
+      throw new IllegalArgumentException(String.format("Invalid status name: %s", status), e);
+    }
+  }
+}

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

@@ -0,0 +1,109 @@
+/*
+ * 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.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.BaseFileScanTask;
+import org.apache.iceberg.ContentFileParser;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionSpecParser;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.ExpressionParser;
+import org.apache.iceberg.expressions.ResidualEvaluator;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.JsonUtil;
+
+class RESTFileScanTaskParser {
+  private static final String DATA_FILE = "data-file";
+  private static final String DELETE_FILE_REFERENCES = "delete-file-references";
+  private static final String RESIDUAL_FILTER = "residual-filter";
+
+  private RESTFileScanTaskParser() {}
+
+  public static void toJson(
+      FileScanTask fileScanTask,
+      Set<Integer> deleteFileReferences,
+      PartitionSpec partitionSpec,
+      JsonGenerator generator)
+      throws IOException {
+    Preconditions.checkArgument(fileScanTask != null, "Invalid file scan task: null");
+    Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
+
+    generator.writeStartObject();
+    generator.writeFieldName(DATA_FILE);
+    ContentFileParser.toJson(fileScanTask.file(), partitionSpec, generator);
+    if (deleteFileReferences != null) {
+      JsonUtil.writeIntegerArray(DELETE_FILE_REFERENCES, deleteFileReferences, generator);
+    }
+
+    if (fileScanTask.residual() != null) {
+      generator.writeFieldName(RESIDUAL_FILTER);
+      ExpressionParser.toJson(fileScanTask.residual(), generator);
+    }
+
+    generator.writeEndObject();
+  }
+
+  public static FileScanTask fromJson(
+      JsonNode jsonNode,
+      List<DeleteFile> allDeleteFiles,
+      Map<Integer, PartitionSpec> specsById,
+      boolean isCaseSensitive) {
+    Preconditions.checkArgument(jsonNode != null, "Invalid JSON node for file scan task: null");
+    Preconditions.checkArgument(
+        jsonNode.isObject(), "Invalid JSON node for file scan task: non-object (%s)", jsonNode);
+
+    DataFile dataFile =
+        (DataFile) ContentFileParser.fromJson(JsonUtil.get(DATA_FILE, jsonNode), specsById);
+    int specId = dataFile.specId();
+
+    DeleteFile[] deleteFiles = null;
+    if (jsonNode.has(DELETE_FILE_REFERENCES)) {
+      List<Integer> indices = JsonUtil.getIntegerList(DELETE_FILE_REFERENCES, jsonNode);
+      Preconditions.checkArgument(
+          Collections.max(indices) < allDeleteFiles.size(),
+          "Invalid delete file references: %s, expected indices < %s",
+          indices,
+          allDeleteFiles.size());
+      deleteFiles = indices.stream().map(allDeleteFiles::get).toArray(DeleteFile[]::new);
+    }
+
+    Expression filter = null;
+    if (jsonNode.has(RESIDUAL_FILTER)) {
+      filter = ExpressionParser.fromJson(jsonNode.get(RESIDUAL_FILTER));
+    }
+
+    String schemaString = SchemaParser.toJson(specsById.get(specId).schema());
+    String specString = PartitionSpecParser.toJson(specsById.get(specId));
+    ResidualEvaluator boundResidual =
+        ResidualEvaluator.of(specsById.get(specId), filter, isCaseSensitive);
+
+    return new BaseFileScanTask(dataFile, deleteFiles, schemaString, specString, boundResidual);
+  }
+}