Spark 3.5: Use fanout writers for unsorted tables by default

apache · Sep 23, 2023 · 53704f3 · 53704f3
1 parent f2ce4ef
commit 53704f3
Show file tree

Hide file tree

Showing 3 changed files with 111 additions and 318 deletions.
diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/SparkWriteConf.java
@@ -179,7 +179,7 @@ public boolean fanoutWriterEnabled() {
         .booleanConf()
         .option(SparkWriteOptions.FANOUT_ENABLED)
         .tableProperty(TableProperties.SPARK_WRITE_PARTITIONED_FANOUT_ENABLED)
-        .defaultValue(TableProperties.SPARK_WRITE_PARTITIONED_FANOUT_ENABLED_DEFAULT)
+        .defaultValue(true)
         .parse();
   }
 

diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
@@ -98,7 +98,7 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
   private final Schema writeSchema;
   private final StructType dsSchema;
   private final Map<String, String> extraSnapshotMetadata;
-  private final boolean partitionedFanoutEnabled;
+  private final boolean useFanoutWriters;
   private final SparkWriteRequirements writeRequirements;
   private final Map<String, String> writeProperties;
 
@@ -126,7 +126,7 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
     this.writeSchema = writeSchema;
     this.dsSchema = dsSchema;
     this.extraSnapshotMetadata = writeConf.extraSnapshotMetadata();
-    this.partitionedFanoutEnabled = writeConf.fanoutWriterEnabled();
+    this.useFanoutWriters = writeConf.fanoutWriterEnabled() && !writeRequirements.hasOrdering();
     this.writeRequirements = writeRequirements;
     this.outputSpecId = writeConf.outputSpecId();
     this.writeProperties = writeConf.writeProperties();
@@ -188,7 +188,7 @@ private WriterFactory createWriterFactory() {
         targetFileSize,
         writeSchema,
         dsSchema,
-        partitionedFanoutEnabled,
+        useFanoutWriters,
         writeProperties);
   }
 
@@ -617,7 +617,7 @@ private static class WriterFactory implements DataWriterFactory, StreamingDataWr
     private final long targetFileSize;
     private final Schema writeSchema;
     private final StructType dsSchema;
-    private final boolean partitionedFanoutEnabled;
+    private final boolean useFanoutWriters;
     private final String queryId;
     private final Map<String, String> writeProperties;
 
@@ -629,15 +629,15 @@ protected WriterFactory(
         long targetFileSize,
         Schema writeSchema,
         StructType dsSchema,
-        boolean partitionedFanoutEnabled,
+        boolean useFanoutWriters,
         Map<String, String> writeProperties) {
       this.tableBroadcast = tableBroadcast;
       this.format = format;
       this.outputSpecId = outputSpecId;
       this.targetFileSize = targetFileSize;
       this.writeSchema = writeSchema;
       this.dsSchema = dsSchema;
-      this.partitionedFanoutEnabled = partitionedFanoutEnabled;
+      this.useFanoutWriters = useFanoutWriters;
       this.queryId = queryId;
       this.writeProperties = writeProperties;
     }
@@ -678,7 +678,7 @@ public DataWriter<InternalRow> createWriter(int partitionId, long taskId, long e
             writeSchema,
             dsSchema,
             targetFileSize,
-            partitionedFanoutEnabled);
+            useFanoutWriters);
       }
     }
   }
@@ -742,8 +742,8 @@ private PartitionedDataWriter(
         Schema dataSchema,
         StructType dataSparkType,
         long targetFileSize,
-        boolean fanoutEnabled) {
-      if (fanoutEnabled) {
+        boolean fanout) {
+      if (fanout) {
         this.delegate = new FanoutDataWriter<>(writerFactory, fileFactory, io, targetFileSize);
       } else {
         this.delegate = new ClusteredDataWriter<>(writerFactory, fileFactory, io, targetFileSize);