get rebase to compile

Author: Rahil Chertara

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

@@ -274,55 +274,6 @@ static PartitionData partitionDataFromRawValue(JsonNode rawPartitionValue, Parti
     return partitionData;
   }
 
-  static void unboundContentFileToJson(
-      ContentFile<?> contentFile, PartitionSpec spec, JsonGenerator generator) throws IOException {
-    Preconditions.checkArgument(contentFile != null, "Invalid content file: null");
-    Preconditions.checkArgument(spec != null, "Invalid partition spec: null");
-    Preconditions.checkArgument(generator != null, "Invalid JSON generator: null");
-    Preconditions.checkArgument(
-        contentFile.specId() == spec.specId(),
-        "Invalid partition spec id from content file: expected = %s, actual = %s",
-        spec.specId(),
-        contentFile.specId());
-
-    generator.writeStartObject();
-    // ignore the ordinal position (ContentFile#pos) of the file in a manifest,
-    // as it isn't used and BaseFile constructor doesn't support it.
-
-    generator.writeNumberField(SPEC_ID, contentFile.specId());
-    generator.writeStringField(CONTENT, contentFile.content().name());
-    generator.writeStringField(FILE_PATH, contentFile.path().toString());
-    generator.writeStringField(FILE_FORMAT, contentFile.format().name());
-
-    if (contentFile.partition() != null) {
-      generator.writeFieldName(PARTITION);
-      SingleValueParser.toJson(spec.partitionType(), contentFile.partition(), generator);
-    }
-
-    generator.writeNumberField(FILE_SIZE, contentFile.fileSizeInBytes());
-
-    metricsToJson(contentFile, generator);
-
-    if (contentFile.keyMetadata() != null) {
-      generator.writeFieldName(KEY_METADATA);
-      SingleValueParser.toJson(DataFile.KEY_METADATA.type(), contentFile.keyMetadata(), generator);
-    }
-
-    if (contentFile.splitOffsets() != null) {
-      JsonUtil.writeLongArray(SPLIT_OFFSETS, contentFile.splitOffsets(), generator);
-    }
-
-    if (contentFile.equalityFieldIds() != null) {
-      JsonUtil.writeIntegerArray(EQUALITY_IDS, contentFile.equalityFieldIds(), generator);
-    }
-
-    if (contentFile.sortOrderId() != null) {
-      generator.writeNumberField(SORT_ORDER_ID, contentFile.sortOrderId());
-    }
-
-    generator.writeEndObject();
-  }
-
   private static void metricsToJson(ContentFile<?> contentFile, JsonGenerator generator)
       throws IOException {
     generator.writeNumberField(RECORD_COUNT, contentFile.recordCount());

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

@@ -20,16 +20,15 @@
 
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.ResidualEvaluator;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 
-class UnboundBaseFileScanTask extends BaseFileScanTask {
-  private UnboundGenericDataFile unboundDataFile;
-  private UnboundGenericDeleteFile[] unboundDeleteFiles;
+public class UnboundBaseFileScanTask extends BaseFileScanTask {
+  private DataFile unboundDataFile;
+  private DeleteFile[] unboundDeleteFiles;
   private Expression filter;
 
-  UnboundBaseFileScanTask(
-      UnboundGenericDataFile unboundDataFile,
-      UnboundGenericDeleteFile[] unboundDeleteFiles,
-      Expression filter) {
+  public UnboundBaseFileScanTask(
+      DataFile unboundDataFile, DeleteFile[] unboundDeleteFiles, Expression filter) {
     super(unboundDataFile, unboundDeleteFiles, null, null, ResidualEvaluator.unpartitioned(filter));
     this.unboundDataFile = unboundDataFile;
     this.unboundDeleteFiles = unboundDeleteFiles;
@@ -46,11 +45,74 @@ public PartitionSpec spec() {
     throw new UnsupportedOperationException("spec() is not supported in UnboundBaseFileScanTask");
   }
 
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("unboundDataFile", unboundDataFile)
+        .add("unboundDeleteFiles", unboundDeleteFiles)
+        .add("filter", filter)
+        .toString();
+  }
+
   public FileScanTask bind(PartitionSpec spec, boolean caseSensitive) {
-    GenericDataFile boundDataFile = unboundDataFile.bindToSpec(spec);
+    // TODO before creating a new task
+    // need to ensure that dataFile is refreshed with correct partitionData using  spec
+    // need to ensure deleteFiles is refreshed with spec info
+    // need to ensure residual refreshed with spec.
+
+    Metrics dataFileMetrics =
+        new Metrics(
+            unboundDataFile.recordCount(),
+            unboundDataFile.columnSizes(),
+            unboundDataFile.valueCounts(),
+            unboundDataFile.nullValueCounts(),
+            unboundDataFile.nanValueCounts());
+    PartitionData partitionData = new PartitionData(spec.partitionType());
+
+    GenericDataFile boundDataFile =
+        new GenericDataFile(
+            spec.specId(),
+            (String) unboundDataFile.path(),
+            unboundDataFile.format(),
+            partitionData,
+            unboundDataFile.fileSizeInBytes(),
+            dataFileMetrics,
+            unboundDataFile.keyMetadata(),
+            unboundDataFile.splitOffsets(),
+            unboundDataFile.sortOrderId());
+
     DeleteFile[] boundDeleteFiles = new DeleteFile[unboundDeleteFiles.length];
     for (int i = 0; i < unboundDeleteFiles.length; i++) {
-      boundDeleteFiles[i] = unboundDeleteFiles[i].bindToSpec(spec);
+      DeleteFile deleteFile = unboundDeleteFiles[i];
+      Metrics deleteFileMetrics =
+          new Metrics(
+              deleteFile.recordCount(),
+              deleteFile.columnSizes(),
+              deleteFile.valueCounts(),
+              deleteFile.nullValueCounts(),
+              deleteFile.nanValueCounts());
+
+      int[] equalityDeletes = null;
+      if (deleteFile.equalityFieldIds() != null) {
+        equalityDeletes =
+            deleteFile.equalityFieldIds().stream().mapToInt(Integer::intValue).toArray();
+      }
+
+      DeleteFile genericDeleteFile =
+          new GenericDeleteFile(
+              spec.specId(),
+              deleteFile.content(),
+              (String) deleteFile.path(),
+              deleteFile.format(),
+              partitionData,
+              deleteFile.fileSizeInBytes(),
+              deleteFileMetrics,
+              equalityDeletes,
+              deleteFile.sortOrderId(),
+              deleteFile.splitOffsets(),
+              deleteFile.keyMetadata());
+
+      boundDeleteFiles[i] = genericDeleteFile;
     }
 
     String schemaString = SchemaParser.toJson(spec.schema());

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

@@ -25,9 +25,8 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.rest.PlanStatus;
-import org.apache.iceberg.rest.RESTResponse;
 
-public class FetchPlanningResultResponse implements RESTResponse {
+public class FetchPlanningResultResponse implements TableScanResponse {
   private final PlanStatus planStatus;
   private final List<String> planTasks;
   private final List<FileScanTask> fileScanTasks;

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

@@ -24,9 +24,8 @@
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.rest.RESTResponse;
 
-public class FetchScanTasksResponse implements RESTResponse {
+public class FetchScanTasksResponse implements TableScanResponse {
   private final List<String> planTasks;
   private final List<FileScanTask> fileScanTasks;
   private final List<DeleteFile> deleteFiles;

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

@@ -25,10 +25,10 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.rest.PlanStatus;
-import org.apache.iceberg.rest.RESTResponse;
 
-public class PlanTableScanResponse implements RESTResponse {
+public class PlanTableScanResponse implements TableScanResponse {
   private final PlanStatus planStatus;
   private final String planId;
   private final List<String> planTasks;
@@ -48,8 +48,7 @@ private PlanTableScanResponse(
     this.planTasks = planTasks;
     this.fileScanTasks = fileScanTasks;
     this.deleteFiles = deleteFiles;
-    this.specsById = specsById;
-    validate();
+    this.specsById = specsById != null ? specsById : Maps.newHashMap();
   }
 
   public PlanStatus planStatus() {

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

@@ -0,0 +1,23 @@
+/*
+ * 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.responses;
+
+import org.apache.iceberg.rest.RESTResponse;
+
+public interface TableScanResponse extends RESTResponse {}

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

@@ -536,7 +536,9 @@ public <T extends RESTResponse> T handleRequest(
         {
           TableIdentifier ident = tableIdentFromPathVars(vars);
           PlanTableScanRequest request = castRequest(PlanTableScanRequest.class, body);
-          TableScan tableScan = catalog.loadTable(ident).newScan();
+          Table table = catalog.loadTable(ident);
+          TableScan tableScan = table.newScan();
+
           if (request.snapshotId() != null) {
             tableScan.useSnapshot(request.snapshotId());
           }
@@ -558,9 +560,10 @@ public <T extends RESTResponse> T handleRequest(
           if (ident.equals(TABLE_COMPLETED_WITH_FILE_SCAN_TASK)) {
             return castResponse(
                 responseType,
-                new PlanTableScanResponse.Builder()
+                PlanTableScanResponse.builder()
                     .withPlanStatus(PlanStatus.COMPLETED)
                     .withFileScanTasks(fileScanTasks)
+                    .withSpecsById(table.specs())
                     .build());
           }
 
@@ -571,9 +574,10 @@ public <T extends RESTResponse> T handleRequest(
             planToFileScanTasks.put(planId, fileScanTasks);
             return castResponse(
                 responseType,
-                new PlanTableScanResponse.Builder()
+                PlanTableScanResponse.builder()
                     .withPlanId(planId)
                     .withPlanStatus(PlanStatus.SUBMITTED)
+                    // .withSpecsById(table.specs())
                     .build());
           }
 
@@ -585,9 +589,10 @@ public <T extends RESTResponse> T handleRequest(
             planTasks.forEach(task -> planToFileScanTasks.put(task, fileScanTasks));
             return castResponse(
                 responseType,
-                new PlanTableScanResponse.Builder()
+                PlanTableScanResponse.builder()
                     .withPlanStatus(PlanStatus.COMPLETED)
                     .withPlanTasks(planTasks)
+                    // .withSpecsById(table.specs())
                     .build());
           }
 
@@ -608,7 +613,7 @@ public <T extends RESTResponse> T handleRequest(
 
             return castResponse(
                 responseType,
-                new PlanTableScanResponse.Builder()
+                PlanTableScanResponse.builder()
                     .withPlanStatus(PlanStatus.COMPLETED)
                     .withPlanTasks(outerPlanTasks)
                     .build());
@@ -623,7 +628,7 @@ public <T extends RESTResponse> T handleRequest(
             String planId = planIDFromPathVars(vars);
             return castResponse(
                 responseType,
-                new FetchPlanningResultResponse.Builder()
+                FetchPlanningResultResponse.builder()
                     .withPlanStatus(PlanStatus.fromName("completed"))
                     .withFileScanTasks(planToFileScanTasks.get(planId))
                     .build());
@@ -638,7 +643,7 @@ public <T extends RESTResponse> T handleRequest(
           if (ident.equals(TABLE_COMPLETED_WITH_PLAN_TASK)) {
             return castResponse(
                 responseType,
-                new FetchScanTasksResponse.Builder()
+                FetchScanTasksResponse.builder()
                     .withFileScanTasks(planToFileScanTasks.get(request.planTask()))
                     .build());
           }
@@ -649,9 +654,7 @@ public <T extends RESTResponse> T handleRequest(
               String innerPlanTask = planToPlanTasks.remove(request.planTask());
               return castResponse(
                   responseType,
-                  new FetchScanTasksResponse.Builder()
-                      .withPlanTasks(List.of(innerPlanTask))
-                      .build());
+                  FetchScanTasksResponse.builder().withPlanTasks(List.of(innerPlanTask)).build());
             }
 
             if (planToFileScanTasks.containsKey(request.planTask())) {
@@ -660,7 +663,7 @@ public <T extends RESTResponse> T handleRequest(
                   planToFileScanTasks.remove(request.planTask());
               return castResponse(
                   responseType,
-                  new FetchScanTasksResponse.Builder()
+                  FetchScanTasksResponse.builder()
                       .withFileScanTasks(fileScanTasksFromPlanTask)
                       .build());
             }