[Spark] Remove sorts from delete from queries reading just one partition and already sorted data. (apache#45) (apache#49)

(cherry picked from commit 07cc6e4)

Author: puchengy

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

@@ -47,7 +47,8 @@ abstract class BaseScan<ThisT, T extends ScanTask, G extends ScanTaskGroup<T>>
           "record_count",
           "partition",
           "key_metadata",
-          "split_offsets");
+          "split_offsets",
+          "sort_order_id");
 
   private static final List<String> STATS_COLUMNS =
       ImmutableList.of(

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

@@ -216,6 +216,9 @@ private TableProperties() {}
   public static final String SPLIT_OPEN_FILE_COST = "read.split.open-file-cost";
   public static final long SPLIT_OPEN_FILE_COST_DEFAULT = 4 * 1024 * 1024; // 4MB
 
+  public static final String FILE_AS_SPLIT = "read.split.file-as-split";
+  public static final boolean FILE_AS_SPLIT_DEFAULT = false;
+
   public static final String PARQUET_VECTORIZATION_ENABLED = "read.parquet.vectorization.enabled";
   public static final boolean PARQUET_VECTORIZATION_ENABLED_DEFAULT = true;
 

spark/v3.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestPinterestCopyOnWriteDelete.java

@@ -179,6 +179,15 @@ public long splitOpenFileCost() {
         .parse();
   }
 
+  public boolean fileAsSplit() {
+    return confParser
+        .booleanConf()
+        .option(SparkReadOptions.FILE_AS_SPLIT)
+        .tableProperty(TableProperties.FILE_AS_SPLIT)
+        .defaultValue(TableProperties.FILE_AS_SPLIT_DEFAULT)
+        .parse();
+  }
+
   /**
    * Enables reading a timestamp without time zone as a timestamp with time zone.
    *

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadConf.java

@@ -47,6 +47,9 @@ private SparkReadOptions() {}
   // Overrides the table's read.split.planning-lookback
   public static final String LOOKBACK = "lookback";
 
+  // Use input file as one split
+  public static final String FILE_AS_SPLIT = "file-as-split";
+
   // Overrides the table's read.split.open-file-cost
   public static final String FILE_OPEN_COST = "file-open-cost";
 

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/SparkReadOptions.java

@@ -24,6 +24,7 @@
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.apache.iceberg.BaseCombinedScanTask;
 import org.apache.iceberg.CombinedScanTask;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.MetadataColumns;
@@ -51,6 +52,7 @@ class SparkCopyOnWriteScan extends SparkScan implements SupportsRuntimeFiltering
 
   private final TableScan scan;
   private final Snapshot snapshot;
+  private final SparkReadConf readConf;
 
   // lazy variables
   private List<FileScanTask> files = null; // lazy cache of files
@@ -79,6 +81,7 @@ class SparkCopyOnWriteScan extends SparkScan implements SupportsRuntimeFiltering
 
     this.scan = scan;
     this.snapshot = snapshot;
+    this.readConf = readConf;
 
     if (scan == null) {
       this.files = Collections.emptyList();
@@ -153,12 +156,19 @@ synchronized List<FileScanTask> files() {
   @Override
   protected synchronized List<CombinedScanTask> tasks() {
     if (tasks == null) {
-      CloseableIterable<FileScanTask> splitFiles =
-          TableScanUtil.splitFiles(
-              CloseableIterable.withNoopClose(files()), scan.targetSplitSize());
-      CloseableIterable<CombinedScanTask> scanTasks =
-          TableScanUtil.planTasks(
-              splitFiles, scan.targetSplitSize(), scan.splitLookback(), scan.splitOpenFileCost());
+      CloseableIterable<CombinedScanTask> scanTasks;
+      if (readConf.fileAsSplit()) {
+        scanTasks =
+            CloseableIterable.transform(
+                CloseableIterable.withNoopClose(files()), BaseCombinedScanTask::new);
+      } else {
+        CloseableIterable<FileScanTask> splitFiles =
+            TableScanUtil.splitFiles(
+                CloseableIterable.withNoopClose(files()), scan.targetSplitSize());
+        scanTasks =
+            TableScanUtil.planTasks(
+                splitFiles, scan.targetSplitSize(), scan.splitLookback(), scan.splitOpenFileCost());
+      }
       tasks = Lists.newArrayList(scanTasks);
     }
 
@@ -201,6 +211,10 @@ public String toString() {
         table(), expectedSchema().asStruct(), filterExpressions(), caseSensitive());
   }
 
+  public boolean fileAsSplit() {
+    return readConf.fileAsSplit();
+  }
+
   private Long currentSnapshotId() {
     Snapshot currentSnapshot = table().currentSnapshot();
     return currentSnapshot != null ? currentSnapshot.snapshotId() : null;

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java

@@ -634,6 +634,7 @@ public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long e
               .dataFileFormat(format)
               .dataSchema(writeSchema)
               .dataSparkType(dsSchema)
+              .dataSortOrder(table.sortOrder())
               .build();
 
       if (spec.isUnpartitioned()) {

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg.spark.source;
 
+import java.util.Arrays;
 import org.apache.iceberg.DistributionMode;
 import org.apache.iceberg.IsolationLevel;
 import org.apache.iceberg.PartitionSpec;
@@ -34,9 +35,11 @@
 import org.apache.iceberg.spark.SparkUtil;
 import org.apache.iceberg.spark.SparkWriteConf;
 import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.util.SortOrderUtil;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.connector.distributions.Distribution;
 import org.apache.spark.sql.connector.distributions.Distributions;
+import org.apache.spark.sql.connector.distributions.OrderedDistribution;
 import org.apache.spark.sql.connector.expressions.SortOrder;
 import org.apache.spark.sql.connector.iceberg.write.RowLevelOperation.Command;
 import org.apache.spark.sql.connector.read.Scan;
@@ -163,6 +166,22 @@ public Write build() {
       ordering = NO_ORDERING;
     }
 
+    // In case of CopyOnWrite operation with scan using file as split and OrderedDistribution
+    // * skip ordering by partition, iff, all input data files are in same partition and has same
+    // spec as current
+    //   table spec
+    // * skip ordering by table sort order, iff, all input files are already sorted by table's
+    // current sort order
+    if (copyOnWriteScan != null
+        && copyOnWriteScan.fileAsSplit()
+        && distribution instanceof OrderedDistribution) {
+      if (skipOrderingAndDistribution((OrderedDistribution) distribution)) {
+        LOG.info(
+            "Skipping distribution/ordering: input files are already in required distribution/ordering");
+        ordering = NO_ORDERING;
+        distribution = Distributions.unspecified();
+      }
+    }
     return new SparkWrite(
         spark, table, writeConf, writeInfo, appId, writeSchema, dsSchema, distribution, ordering) {
 
@@ -265,4 +284,26 @@ private static Schema validateOrMergeWriteSchema(
 
     return writeSchema;
   }
+
+  private boolean skipOrderingAndDistribution(OrderedDistribution distribution) {
+    // check if all input files have same partitioning as current table partitioning
+    if (!copyOnWriteScan.files().stream()
+        .allMatch(x -> x.file().specId() == table.spec().specId())) {
+      return false;
+    }
+
+    // check if all input files are sorted on table's current sort order
+    if (!copyOnWriteScan.files().stream()
+        .allMatch(
+            x ->
+                x.file().sortOrderId() != null
+                    && x.file().sortOrderId() == table.sortOrder().orderId())) {
+      return false;
+    }
+
+    // check if required ordering is same as table's default ordering
+    return Arrays.equals(
+        distribution.ordering(),
+        SparkDistributionAndOrderingUtil.convert(SortOrderUtil.buildSortOrder(table)));
+  }
 }