Spark 3.5: Set useCommitCoordinator to false in batch writes (#9017)

Co-authored-by: Huaxin Gao <huaxin.gao@apple.com>
apache · Nov 10, 2023 · 6a9c182 · 6a9c182
1 parent 774d0e8
commit 6a9c182
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 0 deletions.
diff --git a/...v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewrite.java b/...v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeletesRewrite.java
@@ -136,6 +136,11 @@ public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
           writeProperties);
     }
 
+    @Override
+    public boolean useCommitCoordinator() {
+      return false;
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       PositionDeletesRewriteCoordinator coordinator = PositionDeletesRewriteCoordinator.get();

diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
@@ -171,6 +171,11 @@ public DeltaWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
       return new PositionDeltaWriteFactory(tableBroadcast, command, context, writeProperties);
     }
 
+    @Override
+    public boolean useCommitCoordinator() {
+      return false;
+    }
+
     @Override
     public void commit(WriterCommitMessage[] messages) {
       RowDelta rowDelta = table.newRowDelta();

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
@@ -271,6 +271,11 @@ public DataWriterFactory createBatchWriterFactory(PhysicalWriteInfo info) {
       return createWriterFactory();
     }
 
+    @Override
+    public boolean useCommitCoordinator() {
+      return false;
+    }
+
     @Override
     public void abort(WriterCommitMessage[] messages) {
       SparkWrite.this.abort(messages);