This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch 
gh-readonly-queue/main/pr-6151-8166586fa767008a45665358d200ce23ec955f64
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git

commit 1182a1c614bbeac883aa1e982fca6a9d249ae34c
Author: Andy Grove <[email protected]>
AuthorDate: Thu Sep 24 09:40:28 2026 +0000

    fix: plan Iceberg writes with Spark's own operator when Comet is disabled 
(#6151)
    
    * fix: plan Iceberg writes with Spark's own operator when Comet is disabled
    
    IcebergWriteStrategy only checked the split-operator flag, so with
    spark.comet.enabled=false it still planned Comet's IcebergCommitExec and
    IcebergWriteExec. Gate it on isCometLoaded, as CometExecRule does, so
    disabling Comet restores Spark's V2 write plan.
    
    Closes #6142.
    
    * test: keep Iceberg Comet-disabled assertions inside withSQLConf for Scala 
2.12
    
    withSQLConf returns Unit before Spark 4.0, so binding its result broke
    test compilation on the Spark 3.4 and 3.5 builds.
---
 .../comet/iceberg/IcebergWriteStrategy.scala       |  6 +++++-
 .../comet/CometIcebergWriteActionSuite.scala       | 22 ++++++++++++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

diff --git 
a/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala 
b/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala
index c501c90097..8ceb4e6e48 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.execution.{SparkPlan, 
SparkStrategy}
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
 
 import org.apache.comet.CometConf
+import org.apache.comet.CometSparkSessionExtensions.isCometLoaded
 
 /**
  * Spark Strategy that intercepts Iceberg V2 copy-on-write logical writes and 
emits Comet's
@@ -35,7 +36,10 @@ import org.apache.comet.CometConf
 case class IcebergWriteStrategy(session: SparkSession) extends SparkStrategy {
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
-    if 
(!CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.get(session.sessionState.conf))
 {
+    val conf = session.sessionState.conf
+    // Planner strategies run whether or not Comet is enabled, so check it 
here too: with Comet
+    // off, Spark must plan its own V2 write operator.
+    if (!isCometLoaded(conf) || 
!CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.get(conf)) {
       return Nil
     }
 
diff --git 
a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala 
b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
index 3ab8998291..3edb69f3e2 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
@@ -77,6 +77,28 @@ class CometIcebergWriteActionSuite
     }
   }
 
+  test("spark.comet.enabled=false keeps Spark's own write plan with the split 
flag on") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+    withIcebergCatalog { warehouseDir =>
+      createTable(warehouseDir, "comet_disabled", partitionSpec = "")
+      // withSQLConf returns Unit before Spark 4.0, so the assertions run 
inside it.
+      withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+        val snapshot = captureWrite("comet_disabled") {
+          spark.sql(
+            "INSERT INTO cat.db.comet_disabled VALUES " +
+              "(1, 'us-east', 10.5), (2, 'us-west', 20.3), (3, 'eu', 30.7)")
+        }
+        assert(snapshot.snapshotDelta == 1L, s"expected 1 commit, got 
${snapshot.snapshotDelta}")
+        val (commits, writes) = collectIcebergWriteOps(snapshot.plans)
+        assert(
+          commits.isEmpty && writes.isEmpty,
+          "expected Spark's own write plan with Comet disabled. Plans:\n" +
+            snapshot.plans.mkString("\n--\n"))
+      }
+      assertRows("comet_disabled", expectedIds = Seq(1, 2, 3))
+    }
+  }
+
   test("AppendData partitioned INSERT INTO routes through two-op") {
     assume(icebergAvailable, "Iceberg not available in classpath")
     withIcebergCatalog { warehouseDir =>


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to