API: Refactor FileScanTask

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

@@ -0,0 +1,40 @@
+/*
+ * 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.Collections;
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+public class BaseInputSplit<T extends ScanTask> implements InputSplit<T> {
+  private final List<T> tasks;
+
+  public BaseInputSplit(List<T> tasks) {
+    Preconditions.checkNotNull(tasks, "tasks cannot be null");
+    this.tasks = Collections.unmodifiableList(Lists.newArrayList(tasks));
+  }
+
+  @Override
+  public Collection<T> files() {
+    return tasks;
+  }
+}

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

@@ -19,18 +19,10 @@
 
 package org.apache.iceberg;
 
-import java.util.Collection;
-
 /**
  * A scan task made of several ranges from files.
  */
-public interface CombinedScanTask extends ScanTask {
-  /**
-   * Return the {@link FileScanTask tasks} in this combined task.
-   * @return a Collection of FileScanTask instances.
-   */
-  Collection<FileScanTask> files();
-
+public interface CombinedScanTask extends InputSplit<FileScanTask> {
   @Override
   default CombinedScanTask asCombinedScanTask() {
     return this;

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

@@ -0,0 +1,63 @@
+/*
+ * 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;
+
+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 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();
+}

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

@@ -20,64 +20,37 @@
 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 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();
+  @Override
+  default boolean isAdjacent(FileScanTask other) {
+    return false;
+  }
 
-  /**
-   * 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();
+  @Override
+  default FileScanTask combine(FileScanTask other) {
+    throw new UnsupportedOperationException(this.getClass().getName() + " can't be combined with another task");
+  }
 
-  /**
-   * 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 totalSizeBytes() {
+    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 long totalFilesCount() {
+    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/InputSplit.java

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+/**
+ * A scan task that may include partial input files, multiple input files or both.
+ *
+ * @param <T> the type of scan tasks
+ */
+public interface InputSplit<T extends ScanTask> extends ScanTask {
+  Collection<T> files();
+}

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

@@ -29,7 +29,7 @@
  * Scan objects are immutable and can be shared between threads. Refinement methods, like
  * {@link #select(Collection)} and {@link #filter(Expression)}, create new TableScan instances.
  */
-public interface Scan<T extends Scan<T>> {
+public interface Scan<ThisT, T extends ScanTask, S extends InputSplit<T>> {
   /**
    * Create a new scan from this scan's configuration that will override the {@link Table}'s behavior based
    * on the incoming pair. Unknown properties will be ignored.
@@ -38,15 +38,15 @@ public interface Scan<T extends Scan<T>> {
    * @param value value to override with
    * @return a new scan based on this with overridden behavior
    */
-  T option(String property, String value);
+  ThisT option(String property, String value);
 
   /**
    * Create a new scan from this with the schema as its projection.
    *
    * @param schema a projection schema
    * @return a new scan based on this with the given projection
    */
-  T project(Schema schema);
+  ThisT project(Schema schema);
 
   /**
    * Create a new scan from this that, if data columns where selected
@@ -55,7 +55,7 @@ public interface Scan<T extends Scan<T>> {
    *
    * @return a new scan based on this with case sensitivity as stated
    */
-  T caseSensitive(boolean caseSensitive);
+  ThisT caseSensitive(boolean caseSensitive);
 
   /**
    * Create a new scan from this that loads the column stats with each data file.
@@ -64,7 +64,7 @@ public interface Scan<T extends Scan<T>> {
    *
    * @return a new scan based on this that loads column stats.
    */
-  T includeColumnStats();
+  ThisT includeColumnStats();
 
   /**
    * Create a new scan from this that will read the given data columns. This produces
@@ -74,22 +74,22 @@ public interface Scan<T extends Scan<T>> {
    * @param columns column names from the table's schema
    * @return a new scan based on this with the given projection columns
    */
-  T select(Collection<String> columns);
+  ThisT select(Collection<String> columns);
 
   /**
    * Create a new scan from the results of this filtered by the {@link Expression}.
    *
    * @param expr a filter expression
    * @return a new scan based on this with results filtered by the expression
    */
-  T filter(Expression expr);
+  ThisT filter(Expression expr);
 
   /**
    * Create a new scan from this that applies data filtering to files but not to rows in those files.
    *
    * @return a new scan based on this that does not filter rows in files.
    */
-  T ignoreResiduals();
+  ThisT ignoreResiduals();
 
   /**
    * Create a new scan to use a particular executor to plan. The default worker pool will be
@@ -98,7 +98,7 @@ public interface Scan<T extends Scan<T>> {
    * @param executorService the provided executor
    * @return a table scan that uses the provided executor to access manifests
    */
-  T planWith(ExecutorService executorService);
+  ThisT planWith(ExecutorService executorService);
 
   /**
    * Returns this scan's projection {@link Schema}.
@@ -113,26 +113,23 @@ public interface Scan<T extends Scan<T>> {
   Schema schema();
 
   /**
-   * Plan the {@link FileScanTask files} that will be read by this scan.
+   * Plan tasks for this scan without trying to balance the work.
    * <p>
-   * Each file has a residual expression that should be applied to filter the file's rows.
-   * <p>
-   * This simple plan returns file scans for each file from position 0 to the file's length. For
-   * planning that will combine small files, split large files, and attempt to balance work, use
-   * {@link #planTasks()} instead.
+   * Use {@link #planTasks()} for planning that will attempt to balance the work
+   * by combining small or splitting large files.
    *
-   * @return an Iterable of file tasks that are required by this scan
+   * @return an Iterable of tasks required by this scan
    */
-  CloseableIterable<FileScanTask> planFiles();
+  CloseableIterable<T> planFiles();
 
   /**
-   * Plan the {@link CombinedScanTask tasks} for this scan.
+   * Plan input split tasks for this scan and balance the work.
    * <p>
-   * Tasks created by this method may read partial input files, multiple input files, or both.
+   * Tasks created by this method may read partial input files, multiple input files or both.
    *
-   * @return an Iterable of tasks for this scan
+   * @return an Iterable of input splits required by this scan
    */
-  CloseableIterable<CombinedScanTask> planTasks();
+  CloseableIterable<S> planTasks();
 
   /**
    * Returns the target split size for this scan.