API: Refactor FileScanTask

api/src/main/java/org/apache/iceberg/BaseScanTaskGroup.java

@@ -0,0 +1,62 @@
+/*
+ * 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;
+
+import java.util.Collection;
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+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.ImmutableList;
+
+public class BaseScanTaskGroup<T extends ScanTask> implements ScanTaskGroup<T> {
+  private final Object[] tasks;
+  private transient volatile List<T> taskList;
+
+  public BaseScanTaskGroup(Collection<T> tasks) {
+    Preconditions.checkNotNull(tasks, "tasks cannot be null");
+    this.tasks = tasks.toArray();
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public Collection<T> tasks() {
+    if (taskList == null) {
+      synchronized (this) {
+        if (taskList == null) {
+          ImmutableList.Builder<T> listBuilder = ImmutableList.builderWithExpectedSize(tasks.length);
+          for (Object task : tasks) {
+            listBuilder.add((T) task);
+          }
+          taskList = listBuilder.build();
+        }
+      }
+    }
+
+    return taskList;
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("tasks", Joiner.on(", ").join(tasks))
+        .toString();
+  }
+}

api/src/main/java/org/apache/iceberg/CombinedScanTask.java

@@ -24,13 +24,18 @@
 /**
  * A scan task made of several ranges from files.
  */
-public interface CombinedScanTask extends ScanTask {
+public interface CombinedScanTask extends ScanTaskGroup<FileScanTask> {
   /**
    * Return the {@link FileScanTask tasks} in this combined task.
    * @return a Collection of FileScanTask instances.
    */
   Collection<FileScanTask> files();
 
+  @Override
+  default Collection<FileScanTask> tasks() {
+    return files();
+  }
+
   @Override
   default CombinedScanTask asCombinedScanTask() {
     return this;

api/src/main/java/org/apache/iceberg/ContentScanTask.java

@@ -0,0 +1,67 @@
+/*
+ * 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;
+
+import org.apache.iceberg.expressions.Expression;
+
+/**
+ * A scan task over a range of bytes in a content file.
+ *
+ * @param <F> the Java class of the content file
+ */
+public interface ContentScanTask<F extends ContentFile<F>> extends ScanTask {
+  /**
+   * The {@link ContentFile file} to scan.
+   *
+   * @return the file to scan
+   */
+  F file();
+
+  /**
+   * The {@link PartitionSpec spec} used to store this file.
+   *
+   * @return the partition spec from this file's manifest
+   */
+  PartitionSpec spec();
+
+  /**
+   * The starting position of this scan range in the file.
+   *
+   * @return the start position of this scan range
+   */
+  long start();
+
+  /**
+   * The number of bytes to scan from the {@link #start()} position in the file.
+   *
+   * @return the length of this scan range in bytes
+   */
+  long length();
+
+  /**
+   * Returns the residual expression that should be applied to rows in this file scan.
+   * <p>
+   * The residual expression for a file is a filter expression created by partially evaluating the scan's filter
+   * using the file's partition data.
+   *
+   * @return a residual expression to apply to rows from this scan
+   */
+  Expression residual();
+}

api/src/main/java/org/apache/iceberg/FileScanTask.java

@@ -20,64 +20,27 @@
 package org.apache.iceberg;
 
 import java.util.List;
-import org.apache.iceberg.expressions.Expression;
 
 /**
- * A scan task over a range of a single file.
+ * A scan task over a range of bytes in a single data file.
  */
-public interface FileScanTask extends ScanTask {
-  /**
-   * The {@link DataFile file} to scan.
-   *
-   * @return the file to scan
-   */
-  DataFile file();
-
+public interface FileScanTask extends ContentScanTask<DataFile>, SplittableScanTask<FileScanTask> {
   /**
    * A list of {@link DeleteFile delete files} to apply when reading the task's data file.
    *
    * @return a list of delete files to apply
    */
   List<DeleteFile> deletes();
 
-  /**
-   * The {@link PartitionSpec spec} used to store this file.
-   *
-   * @return the partition spec from this file's manifest
-   */
-  PartitionSpec spec();
-
-  /**
-   * The starting position of this scan range in the file.
-   *
-   * @return the start position of this scan range
-   */
-  long start();
-
-  /**
-   * The number of bytes to scan from the {@link #start()} position in the file.
-   *
-   * @return the length of this scan range in bytes
-   */
-  long length();
-
-  /**
-   * Returns the residual expression that should be applied to rows in this file scan.
-   * <p>
-   * The residual expression for a file is a filter expression created from the scan's filter, inclusive
-   * any predicates that are true or false for the entire file removed, based on the file's
-   * partition data.
-   *
-   * @return a residual expression to apply to rows from this scan
-   */
-  Expression residual();
+  @Override
+  default long sizeBytes() {
+    return length() + deletes().stream().mapToLong(ContentFile::fileSizeInBytes).sum();
+  }
 
-  /**
-   * Splits this scan task into component {@link FileScanTask scan tasks}, each of {@code splitSize} size
-   * @param splitSize The size of a component scan task
-   * @return an Iterable of {@link FileScanTask scan tasks}
-   */
-  Iterable<FileScanTask> split(long splitSize);
+  @Override
+  default int filesCount() {
+    return 1 + deletes().size();
+  }
 
   @Override
   default boolean isFileScanTask() {

api/src/main/java/org/apache/iceberg/IncrementalAppendScan.java

@@ -23,7 +23,7 @@
 /**
  * API for configuring an incremental table scan for appends only snapshots
  */
-public interface IncrementalAppendScan extends Scan<IncrementalAppendScan> {
+public interface IncrementalAppendScan extends Scan<IncrementalAppendScan, FileScanTask, CombinedScanTask> {
 
   /**
    * Refine the incremental scan with the start snapshot inclusive.

api/src/main/java/org/apache/iceberg/MergeableScanTask.java

@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+/**
+ * A scan task that can be potentially merged with other scan tasks.
+ *
+ * @param <ThisT> the child Java API class
+ */
+public interface MergeableScanTask<ThisT> extends ScanTask {
+  /**
+   * Checks if this task can merge with a given task.
+   *
+   * @param other another task
+   * @return whether the tasks can be merged
+   */
+  boolean canMerge(ScanTask other);
+
+  /**
+   * Merges this task with a given task.
+   * <p>
+   * Note this method will be called only if {@link #canMerge(ScanTask)} returns true.
+   *
+   * @param other another task
+   * @return a new merged task
+   */
+  ThisT merge(ScanTask other);
+}