You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/10/05 14:31:27 UTC

[PR] feat: enable the Iceberg split-operator write plan by default [datafusion-comet]

andygrove opened a new pull request, #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664

   ## Which issue does this PR close?
   
   Part of #5644. This is step 1 of its rollout, the split-operator default. Step 2, the native-writer default, stays open there.
   
   ## Rationale for this change
   
   #5644 makes the Iceberg write path the default in two steps: the split-operator plan first, then the native writer one release later. The split plan changes only the plan shape. iceberg-java still writes and commits the data files, so running it by default for a release gives the plan shape real use before the native writer, which builds on it, is turned on. With 1.2.0 targeted for late October or early November (#6550), it needs to land now to get nightly runs first.
   
   What blocked it is fixed: #6142 (the planner strategy ignored `spark.comet.enabled`) and #6586 (Spark 4.2 catalog transactions, fixed by #6587). The Iceberg Spark test jobs have run with the split plan on since it was added in #4658.
   
   ## What changes are included in this PR?
   
   - `spark.comet.write.iceberg.splitOperator.enabled` defaults to `true` on every Spark version. It moves from the testing config category to query execution, since it is now the switch that restores Spark's own operator, and its description says so.
   - Docs: the Iceberg writes user guide (intro, configuration example, and the fallback list, which now also names `spark.comet.enabled=false`), `iceberg.md`, the operators table, a 1.2.0 entry in the upgrade guide, the Iceberg writes contributor guide, and the Iceberg write review skill.
   - A test in `CometIcebergWriteActionSuite` that unsets the flag and checks that an Iceberg append plans `IcebergCommit` over `IcebergWrite`.
   
   ## How are these changes tested?
   
   - The new test failed before the default changed, with no `IcebergCommitExec` in the plan.
   - The existing test that `spark.comet.enabled=false` keeps Spark's own write plan still passes.
   - Every Iceberg suite that does not need MinIO, plus `CometConfSuite` and the Iceberg SQL file tests, passes locally on Spark 4.1 (379 tests, one canceled by an existing SPARK-55626 assumption) and on Spark 3.5 (365 tests, 11 canceled by Spark 4.0+, Spark 4.1+ and Iceberg version assumptions). Most of these suites create their tables with `INSERT`, which now goes through the split plan.
   - The Iceberg Spark test diffs already set the flag, so those jobs run the same plan as before. `run-iceberg-tests` and `run-all-spark-profiles` are applied for the other Spark versions.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6110923455

   @unikdahal you have worked on the split Iceberg write path (#6693 and others). If you have time, would you be able to review this one? Thanks.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6060918310

   @viirya `CometIcebergWriteBenchmark` for the four arms, at 1cb149f6e on an Apple M3 Ultra (local filesystem, 4M rows, `local[5]`, average of five runs):
   
   | Case | Spark | Comet scan, iceberg-java writer | Comet scan, native writer | Native vs iceberg-java writer |
   | --- | --: | --: | --: | --: |
   | Unpartitioned `INSERT` | 1426 ms | 1250 ms | 606 ms | 2.1x |
   | Partitioned `INSERT`, clustered writer | 1948 ms | 1496 ms | 1040 ms | 1.4x |
   | Partitioned `INSERT`, fanout writer | 1922 ms | 1787 ms | 951 ms | 1.9x |
   | Copy-on-write `DELETE` | 2293 ms | 2116 ms | 1559 ms | 1.4x |
   
   The native writer is faster than iceberg-java's in all four.
   
   On the footer read: for every file it writes natively, the task reads the file's footer back before it returns its commit message, so that iceberg-java's own code computes the file's metrics. Through `S3FileIO` that is two ranged GETs per file, the 8-byte tail and then the footer, made one file after another. iceberg-java takes the same metrics from the footer it still holds in memory. A task that writes a few large files hardly notices, but a fanout task that writes many small files pays two request round trips for each of them. The user guide, under Native Parquet write eligibility, and the 1.2.0 upgrade entry now say so, and #6772 tracks removing the reads by handing each footer from the native writer to the JVM.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6149238597

   Thanks @mbutrovich !


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6028046022

   Thanks @comphead, these were all still describing the old behavior. `datasources.md`, `gluten_comparison.md`, the roadmap's Iceberg section and the operator tables in `understanding-comet-plans.md` now describe the single, default-on setting, and none of them call the writer or the split-plan operators experimental any more. I've replied on each inline thread with what changed there.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6082875162

   @viirya One correction to my earlier comment: the Spark 4.2 label run doesn't exercise the Iceberg write path. Iceberg isn't on the test classpath there, so the Iceberg suites are canceled in the 4.2 `[scans]` bucket. The strategy change and the new default-flag tests did run and pass on Spark 3.4 (the merge-on-read test needs 3.5), 3.5 and 4.0.
   
   The two red jobs are the 4.1 `[scans]` listener race that #6809 fixes, which this branch gets the next time `main` is merged, and a Maven Central timeout in the 3.5 `[exec]` bucket, where no test ran.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6070795063

   @viirya I merged `main` in (b8cadbad4), so #6740 is in and the Spark 4.2 expressions failure should clear. The only conflict was `gluten_comparison.md`, where `main` had just rewritten the Iceberg paragraph and still called the native writer experimental and off by default. I rewrote that sentence for the new default and fixed the same claim in the FAQ.
   
   I added `run-all-spark-profiles` back, so the Comet suites run on Spark 3.4, 3.5, 4.0 and 4.2 for this head. That covers the strategy change and, on 3.5 and later, the merge-on-read test. The suites listed in the description also still pass locally on the merged head on 4.1.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "unikdahal (via GitHub)" <gi...@apache.org>.
unikdahal commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4252723097


##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala:
##########
@@ -31,16 +31,21 @@ import org.apache.comet.CometSparkSessionExtensions.isCometLoaded
 import org.apache.comet.shims.ShimCometMergeRows
 
 /**
- * Spark strategy for Comet's split Iceberg V2 writer. WriteDelta is modeled and dispatched here,
- * but position-delta rows are still executed by Iceberg's JVM DeltaWriter.
+ * Spark strategy for Comet's split Iceberg V2 writer. WriteDelta is modeled and dispatched here
+ * when the testing split flag is on, but position-delta rows are still executed by Iceberg's JVM
+ * DeltaWriter.
  */
 case class IcebergWriteStrategy(session: SparkSession) extends SparkStrategy {
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
     val conf = session.sessionState.conf
+    // The native write flag plans the split operator on its own. The split flag plans it with
+    // the native writer off, which only tests do.
+    val splitEnabled = CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.get(conf) ||

Review Comment:
   One concern with the upgrade path here.
   
   In 1.1 we documented enabling "spark.comet.write.iceberg.splitOperator.enabled" together with native writes, so users may still have this set after upgrading.
   
   With the OR here, setting "spark.comet.write.iceberg.enabled=false" would still use the split write path rather than restoring Spark's original write operator. On Spark 3.5+ it would also enable the split "WriteDelta" path.
   
   Should we either make the new flag authoritative, or mention in the migration guide that the old split flag needs to be removed/unset? A test for this combination would probably be useful too.



##########
docs/source/contributor-guide/iceberg-spark-tests.md:
##########
@@ -31,13 +31,11 @@ Here is an overview of the changes that the diffs make to Iceberg:
   uses a native Iceberg scan, these classes fail to compile and must be removed.
 - Configure test base classes (`TestBase`, `ExtensionsTestBase`, `ScanTestBase`, etc.) to load the Comet Spark
   plugin and shuffle manager
-- Enable the Iceberg write split-operator plan (`spark.comet.write.iceberg.splitOperator.enabled`) alongside the
-  native scan in every Comet-configured session. The flag is off by default for users, so Iceberg's own suites
-  are the only place the split plan (`IcebergCommit -> IcebergWrite`) is exercised against Iceberg's write,
-  commit, and row-level-operation tests. See [#5259]
-- Enable Comet's native (iceberg-rust) Parquet writer (`spark.comet.write.iceberg.enabled`) in the same sessions.
-  The native writer is experimental and off by default for users, so this is where it runs against Iceberg's
-  write, commit, and row-level-operation tests.
+- Enable Comet's Iceberg write path, the split-operator plan (`IcebergCommit -> IcebergWrite`) and the native
+  (iceberg-rust) Parquet writer, alongside the native scan in every Comet-configured session, so that Iceberg's
+  write, commit, and row-level-operation tests run through it. `spark.comet.write.iceberg.enabled` switches both
+  and is on by default; the diffs set it explicitly, together with the testing-only
+  `spark.comet.write.iceberg.splitOperator.enabled`, which it makes redundant. See [#5259]

Review Comment:
   I don't think this flag is completely redundant. It is redundant for the normal write path, but it is still needed to exercise the split "WriteDelta" path for MOR writes on Spark 3.5+.
   
   Might be worth clarifying that here.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6070500247

   Thanks for the benchmark numbers and the footer-read docs. Both answer what I asked. `94915161e` is a good catch: with #6693 on `main`, the new default would have sent every merge-on-read write through the split plan for no benefit. Since `run-all-spark-profiles` was removed on 10/07, the strategy change and the new merge-on-read test have only run on Spark 4.1. Could you add the label back for one more round before merge? #6740 is in now, so merging `main` should also clear the Spark 4.2 expressions failure.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4200024207


##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -125,21 +125,24 @@ object CometConf extends ShimCometConf {
     conf("spark.comet.write.iceberg.splitOperator.enabled")
       .category(CATEGORY_TESTING)
       .doc(
-        "Whether to rewrite Iceberg V2 writes from Spark's combined V2 write/commit operator " +
-          "into Comet's two-operator shape: a file writer exec (inside AQE) and a committer " +
-          "(outside AQE).")
+        "Whether to plan Iceberg writes as Comet's file writer and committer even when " +
+          "`spark.comet.write.iceberg.enabled` is false, so that Iceberg's own writer writes " +
+          "every data file inside Comet's plan. Used by tests to compare the two writers " +
+          "under the same plan.")
       .booleanConf
       .createWithDefault(false)
 
   val COMET_ICEBERG_NATIVE_WRITE_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.write.iceberg.enabled")
-      .category(CATEGORY_TESTING)
+      .category(CATEGORY_EXEC)
       .doc(
-        "Whether to delegate the executor-side Parquet write to Comet's native (iceberg-rust) " +
-          "writer when the table's properties allow it. Requires " +
-          "`spark.comet.write.iceberg.splitOperator.enabled = true`. Off by default.")
+        "Whether Comet plans Iceberg writes and writes their data files natively. Comet " +
+          "replaces Spark's combined V2 write operator with a file writer (inside AQE) under a " +
+          "committer (outside AQE), and writes the data files of each eligible write with its " +
+          "native (iceberg-rust) writer. Other writes use Iceberg's own writer, and Iceberg " +
+          "commits every write. Set this to false to plan Spark's own V2 write operator.")
       .booleanConf
-      .createWithDefault(false)
+      .createWithDefault(true)

Review Comment:
   [P2] Preserve the write operator during transition reversion before enabling this default. With `spark.sql.adaptive.enabled=false`, `spark.comet.exec.transitionRevert.enabled=true` and `spark.comet.exec.transitionRevert.maxTransitions=0`, an eligible Iceberg `INSERT ... SELECT` now selects `CometIcebergWriteExec` without either write flag being set. Spark inserts a `ColumnarToRowExec` beneath it, which triggers reversion before redundant transitions are removed. Because the native writer’s `originalPlan` is its child, reversion removes the writer entirely. `IcebergCommitExec` then attempts to deserialize table data as commit messages and the write fails. Previously, these settings retained Spark’s working V2 writer. The underlying defect is tracked in #5719, but this default materially expands its exposure to users who never enabled experimental Iceberg writes. Restore an `IcebergWriteExec` when reverting, or safely exclude this combination from native conversion, and add a re
 gression covering the default flags.
   
   Evidence: Bounded reproduction: `/tmp/6664-current-review-075caeqj/WriteRevertProbe.scala`, with results in `probe-full-pipeline.log`. Using exact-head classes, Spark’s transition insertion and `CometRule.postColumnarRules` transformed `IcebergCommit -> CometIcebergWrite -> ColumnarToRow -> CometProject -> CometLocalTableScan` into `IcebergCommit -> Project -> Project -> LocalTableScan`. Output: `COUNT=1`, `WRITERS_LEFT=0`, followed by `java.io.EOFException` for one ordinary binary row. The equivalent stock `AppendDataExec` control reported `COMMIT_ROWS=1` and `STOCK_SUCCEEDED=true`. Disabling reversion retained the native writer. This was a JVM plan/commit harness with a test BatchWrite, without native storage I/O.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4200267381


##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,28 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,

Review Comment:
   The 1.1.0 name for this setting was `spark.comet.iceberg.write.enabled`, and the [1.1.0 user guide](https://github.com/apache/datafusion-comet/blob/1.1.0/docs/source/user-guide/latest/iceberg-writes.md#L83) told people to set that name. #6494 renamed it to `spark.comet.write.iceberg.enabled` after the 1.1 branch was cut. Comet ignores the old name now, and nothing in this entry mentions it.
   
   That matters here because the entry says the setting "now defaults to `true`" and offers `spark.comet.write.iceberg.enabled=false` as the way back to 1.1.0 behavior. A deployment that set `spark.comet.iceberg.write.enabled=false` in 1.1.0 would get native writes after upgrading. This PR is also the one that moves the key out of `testing`, so it seems like the right place to say it.
   
   Could the entry name the old key and say Comet ignores it, the way the 1.1.0 section does for the testing keys it removed?



##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala:
##########
@@ -38,9 +38,13 @@ case class IcebergWriteStrategy(session: SparkSession) extends SparkStrategy {
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
     val conf = session.sessionState.conf
+    // The native write flag plans the split operator on its own. The split flag plans it with
+    // the native writer off, which only tests do.
+    val splitEnabled = CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.get(conf) ||
+      CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.get(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)) {
+    if (!isCometLoaded(conf) || !splitEnabled) {

Review Comment:
   Should `spark.comet.exec.enabled=false` also keep Spark's own write operator? This strategy is injected unconditionally in `CometSparkSessionExtensions`, and it only checks `isCometLoaded`, the two write flags and plan-only mode. `isCometLoaded` covers `spark.comet.enabled`, off-heap memory, the shuffle manager and a few other startup conditions, but not `spark.comet.exec.enabled`. So with the new default, someone who runs Comet for scans or shuffle only would get `IcebergCommit` over a JVM `IcebergWrite` after upgrading. The operators page tells them `spark.comet.exec.enabled=false` turns off native execution, and the new fallback list in `iceberg-writes.md` names only `spark.comet.enabled=false`. `CometIcebergWriteActionSuite` covers that case (line 122) but not this one.
   
   I read this from the code and haven't run it. The description says the split plan alone gives users nothing over Spark's operator, so I'd lean toward adding `COMET_EXEC_ENABLED` to this check, with a test next to the `spark.comet.enabled=false` one. If you'd rather keep the split plan in that mode, could the fallback list say so? It could also mention plan-only mode, which this method checks too.



##########
docs/source/user-guide/latest/iceberg-writes.md:
##########
@@ -92,10 +96,8 @@ spark.sql.catalog.<name>=org.apache.iceberg.spark.SparkCatalog
 spark.sql.catalog.<name>.type=hadoop                          # or hive / glue / rest / ...
 spark.sql.catalog.<name>.warehouse=...
 
-# Split-operator plan (experimental, off by default)
-spark.comet.write.iceberg.splitOperator.enabled=true
-
-# Native Parquet writer (experimental, off by default; requires the split plan)
+# Split-operator plan and native Parquet writer (on by default since Comet 1.2.0; false plans
+# Spark's own write operator)
 spark.comet.write.iceberg.enabled=true

Review Comment:
   This block still reads as an enable step. [Line 89](https://github.com/apache/datafusion-comet/blob/1dc0d67df2ee8b15dc5b8704598dfb71e2d22831/docs/source/user-guide/latest/iceberg-writes.md#L89) introduces it as "plus the write-side toggle", and this line sets `spark.comet.write.iceberg.enabled=true`, which is now the default. Copying the block changes nothing for this setting. Of the write-related lines, only `spark.comet.exec.localTableScan.enabled=true` changes behavior.
   
   Could the example show `spark.comet.write.iceberg.enabled=false` as the opt-out instead? Or the line could go, with the default stated in the text above.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4224875529


##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -117,25 +120,114 @@ class CometIcebergWriteActionSuite
     }
   }
 
-  test("spark.comet.enabled=false keeps Spark's own write plan with the split flag on") {
+  // Turning off Comet, or only its native execution as an application that uses Comet just for
+  // scans or shuffle does, keeps Spark's own write operator.
+  Seq(CometConf.COMET_ENABLED, CometConf.COMET_EXEC_ENABLED).foreach { flag =>
+    test(s"${flag.key}=false keeps Spark's own write plan with the split flag on") {
+      assume(icebergAvailable, "Iceberg not available in classpath")
+      withIcebergCatalog { warehouseDir =>
+        val table = flag.key.replace('.', '_')
+        createTable(warehouseDir, table, partitionSpec = "")
+        // withSQLConf returns Unit before Spark 4.0, so the assertions run inside it.
+        withSQLConf(flag.key -> "false") {
+          val snapshot = captureWrite(table) {
+            spark.sql(s"INSERT INTO cat.db.$table 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,
+            s"expected Spark's own write plan with ${flag.key}=false. Plans:\n" +
+              snapshot.plans.mkString("\n--\n"))
+        }
+        assertRows(table, expectedIds = Seq(1, 2, 3))
+      }
+    }
+  }
+
+  // The suite pins both write flags, so this test unsets them to see what an application that
+  // sets neither gets. A Parquet scan is a native input, so the write is eligible.
+  // https://github.com/apache/datafusion-comet/issues/5644
+  test("an eligible Iceberg write runs natively under the split plan when no flag is set") {
     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)")
+      withTempPath { dir =>
+        spark
+          .range(10)
+          .selectExpr("CAST(id AS INT) AS id", "'eu' AS region", "CAST(id AS DOUBLE) AS amount")
+          .write
+          .parquet(dir.getCanonicalPath)
+        createTable(warehouseDir, "write_defaults", partitionSpec = "")
+        val snapshot = captureWrite("write_defaults") {
+          withSessionConf(
+            CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None,
+            CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) {
+            spark.read
+              .parquet(dir.getCanonicalPath)
+              .writeTo(s"$catalog.$ns.write_defaults")
+              .append()
+          }
         }
         assert(snapshot.snapshotDelta == 1L, s"expected 1 commit, got ${snapshot.snapshotDelta}")
-        val (commits, writes) = collectIcebergWriteOps(snapshot.plans)
+        val (commits, _) = collectIcebergWriteOps(snapshot.plans)
+        val nativeWrites = snapshot.plans.flatMap { plan =>
+          collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e }
+        }
         assert(
-          commits.isEmpty && writes.isEmpty,
-          "expected Spark's own write plan with Comet disabled. Plans:\n" +
+          commits.nonEmpty && nativeWrites.nonEmpty,
+          "expected an IcebergCommitExec over a CometIcebergWriteExec. Plans:\n" +
             snapshot.plans.mkString("\n--\n"))
+        assertRows("write_defaults", expectedIds = 0 until 10)
+      }
+    }
+  }
+
+  // With no flag set, an eligible write becomes a CometIcebergWriteExec, so reverting a
+  // transition-heavy stage has to put a JVM writer back rather than drop the write.
+  // https://github.com/apache/datafusion-comet/issues/5719
+  for (adaptive <- Seq(false, true)) {
+    test(s"transition-heavy fallback keeps the write when no flag is set with AQE=$adaptive") {
+      assume(icebergAvailable, "Iceberg not available in classpath")
+      withIcebergCatalog { warehouseDir =>
+        withTempPath { dir =>
+          spark
+            .range(3)
+            .selectExpr(
+              "CAST(id + 1 AS INT) AS id",
+              "'eu' AS region",
+              "CAST(id AS DOUBLE) AS amount")
+            .write
+            .parquet(dir.getCanonicalPath)
+          val table = s"transition_defaults_${if (adaptive) "aqe" else "no_aqe"}"
+          createTable(warehouseDir, table, partitionSpec = "")
+          // withSQLConf returns Unit before Spark 4.0, so the assertions run inside it.
+          withSQLConf(
+            SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+            CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "true",
+            CometConf.COMET_EXEC_TRANSITION_REVERT_MAX_TRANSITIONS.key -> "0") {
+            val snapshot = captureWrite(table) {
+              withSessionConf(
+                CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None,
+                CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) {
+                spark.read.parquet(dir.getCanonicalPath).writeTo(s"$catalog.$ns.$table").append()
+              }
+            }
+            assertExactlyOneCommit(snapshot)
+            // Also shows the stage was reverted: otherwise the native writer would still be here.
+            val nativeWrites = snapshot.plans.flatMap { plan =>
+              collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e }
+            }
+            assert(
+              nativeWrites.isEmpty,

Review Comment:
   Thanks, this reads well. Running the same write under the default threshold first makes the reverted run meaningful, and checking for the JVM `IcebergWriteExec` too closes the gap. Resolving.



##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,31 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,
+`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite`
+under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of
+each eligible write natively with iceberg-rust. A write that is not eligible still uses
+iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows,
+but natively written files differ from iceberg-java's in the ways listed under
+[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI
+show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator,
+as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the
+testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the
+new default unless it sets `spark.comet.write.iceberg.enabled=false`.
+`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to
+`false` does not turn the split operator off. See [Iceberg Writes](iceberg-writes.md).
+
+The native writer's buffers count against Comet's off-heap memory pool, where iceberg-java's buffers
+sit on the JVM heap. A fanout write keeps a data file open for every partition a task writes to, and
+each open file holds the row group it is writing, so a task that writes to many partitions can need
+more memory than the pool grants it. The task then fails with
+`Additional allocation failed for IcebergWriteExec`. Disabling the fanout writer

Review Comment:
   You're right about `useFanoutWriter`, and I had the scope too narrow. Since apache/iceberg#8621, `writeOrdering` drops the local sort whenever the table is unsorted and fanout is on by default, so any unsorted partitioned table gets the fanout writer whatever its distribution mode. Keeping fanout on iceberg-java would take most partitioned writes off the native path, so closing partitions early is the better fix. I'll review #6773 separately, and I'd like to keep this thread open until it lands, since it's listed as a blocker.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "mbutrovich (via GitHub)" <gi...@apache.org>.
mbutrovich commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4260266960


##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -111,6 +111,45 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,
+`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite`
+under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of
+each eligible write natively with iceberg-rust. A write that is not eligible still uses
+iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows,
+but natively written files differ from iceberg-java's in the ways listed under
+[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI
+show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator,
+as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the
+testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the
+new default unless it sets `spark.comet.write.iceberg.enabled=false`.
+`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to

Review Comment:
   Should this say the split flag remains a testing setting? "Is now" reads as if 1.1.0 treated it as a user setting, but it was already in the `testing` category in 1.1.0 ([`CometConf.scala` at 1.1.0](https://github.com/apache/datafusion-comet/blob/992c806a7e38c2e88bd018aa5774164b0850e1fa/spark/src/main/scala/org/apache/comet/CometConf.scala#L124-L126)). The rest of the paragraph covers the part that did change, that the 1.1.0 user guide told people to set it.
   
   ```suggestion
   `spark.comet.write.iceberg.splitOperator.enabled` remains a testing setting, and setting it to
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6022484839

   cc @jordepic 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "jordepic (via GitHub)" <gi...@apache.org>.
jordepic commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6023324662

   Exciting!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4201730609


##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala:
##########
@@ -38,9 +38,13 @@ case class IcebergWriteStrategy(session: SparkSession) extends SparkStrategy {
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
     val conf = session.sessionState.conf
+    // The native write flag plans the split operator on its own. The split flag plans it with
+    // the native writer off, which only tests do.
+    val splitEnabled = CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.get(conf) ||
+      CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.get(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)) {
+    if (!isCometLoaded(conf) || !splitEnabled) {

Review Comment:
   Agreed, the split plan with iceberg-java's writer gives a scan- or shuffle-only deployment nothing, so it shouldn't change their plans. The strategy now also checks `spark.comet.exec.enabled`, and the `spark.comet.enabled=false` test runs for both settings. The fallback list in `iceberg-writes.md` names `spark.comet.exec.enabled=false` and plan-only mode now too.
   



##########
docs/source/user-guide/latest/iceberg-writes.md:
##########
@@ -92,10 +96,8 @@ spark.sql.catalog.<name>=org.apache.iceberg.spark.SparkCatalog
 spark.sql.catalog.<name>.type=hadoop                          # or hive / glue / rest / ...
 spark.sql.catalog.<name>.warehouse=...
 
-# Split-operator plan (experimental, off by default)
-spark.comet.write.iceberg.splitOperator.enabled=true
-
-# Native Parquet writer (experimental, off by default; requires the split plan)
+# Split-operator plan and native Parquet writer (on by default since Comet 1.2.0; false plans
+# Spark's own write operator)
 spark.comet.write.iceberg.enabled=true

Review Comment:
   Good point. I dropped the line from the block, and the text above it now says the setting is on by default and that `false` plans Spark's own operator. The local table scan setting is the only write-related line left in the example.
   



##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,28 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,

Review Comment:
   Thanks, I'd missed that the rename landed after the 1.1 branch was cut. The entry now says 1.1.0 named the setting `spark.comet.iceberg.write.enabled`, that it was a testing setting Comet now ignores, and that a deployment that set it to `false` needs `spark.comet.write.iceberg.enabled=false` to keep iceberg-java's writer.
   



##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -125,21 +125,24 @@ object CometConf extends ShimCometConf {
     conf("spark.comet.write.iceberg.splitOperator.enabled")
       .category(CATEGORY_TESTING)
       .doc(
-        "Whether to rewrite Iceberg V2 writes from Spark's combined V2 write/commit operator " +
-          "into Comet's two-operator shape: a file writer exec (inside AQE) and a committer " +
-          "(outside AQE).")
+        "Whether to plan Iceberg writes as Comet's file writer and committer even when " +
+          "`spark.comet.write.iceberg.enabled` is false, so that Iceberg's own writer writes " +
+          "every data file inside Comet's plan. Used by tests to compare the two writers " +
+          "under the same plan.")
       .booleanConf
       .createWithDefault(false)
 
   val COMET_ICEBERG_NATIVE_WRITE_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.write.iceberg.enabled")
-      .category(CATEGORY_TESTING)
+      .category(CATEGORY_EXEC)
       .doc(
-        "Whether to delegate the executor-side Parquet write to Comet's native (iceberg-rust) " +
-          "writer when the table's properties allow it. Requires " +
-          "`spark.comet.write.iceberg.splitOperator.enabled = true`. Off by default.")
+        "Whether Comet plans Iceberg writes and writes their data files natively. Comet " +
+          "replaces Spark's combined V2 write operator with a file writer (inside AQE) under a " +
+          "committer (outside AQE), and writes the data files of each eligible write with its " +
+          "native (iceberg-rust) writer. Other writes use Iceberg's own writer, and Iceberg " +
+          "commits every write. Set this to false to plan Spark's own V2 write operator.")
       .booleanConf
-      .createWithDefault(false)
+      .createWithDefault(true)

Review Comment:
   This is #5719, and #5957 fixed it by making the reverted stage restore an `IcebergWriteExec` instead of dropping the write. This branch now includes #5957. I added `transition-heavy fallback keeps the write when no flag is set`, which writes from a Parquet source with transition reversion on, `maxTransitions=0` and no write flag set, under both AQE settings. Without #5957 it fails with the same `EOFException` at commit, and with it the write goes through iceberg-java's writer and commits the expected rows.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4253014839


##########
spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteStrategy.scala:
##########
@@ -31,16 +31,21 @@ import org.apache.comet.CometSparkSessionExtensions.isCometLoaded
 import org.apache.comet.shims.ShimCometMergeRows
 
 /**
- * Spark strategy for Comet's split Iceberg V2 writer. WriteDelta is modeled and dispatched here,
- * but position-delta rows are still executed by Iceberg's JVM DeltaWriter.
+ * Spark strategy for Comet's split Iceberg V2 writer. WriteDelta is modeled and dispatched here
+ * when the testing split flag is on, but position-delta rows are still executed by Iceberg's JVM
+ * DeltaWriter.
  */
 case class IcebergWriteStrategy(session: SparkSession) extends SparkStrategy {
 
   override def apply(plan: LogicalPlan): Seq[SparkPlan] = {
     val conf = session.sessionState.conf
+    // The native write flag plans the split operator on its own. The split flag plans it with
+    // the native writer off, which only tests do.
+    val splitEnabled = CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.get(conf) ||

Review Comment:
   The old key can stay set after the upgrade, and the split flag is now a testing setting that the suites depend on: they set it with the write flag off to get the split plan with iceberg-java's writer as a baseline. So I kept the write flag from overriding it and documented the combination in cb90a1b5f. The 1.2.0 upgrade entry now says a deployment that set the split flag in 1.1.0 has to unset it, as well as set `spark.comet.write.iceberg.enabled=false`, to get Spark's own operator back. It also says that on Spark 3.5+ the flag plans merge-on-read writes as the split plan. The fallback list in `iceberg-writes.md` says the same, and `the split flag keeps the split plan with iceberg-java's writer when the write flag is off` pins the combination.



##########
docs/source/contributor-guide/iceberg-spark-tests.md:
##########
@@ -31,13 +31,11 @@ Here is an overview of the changes that the diffs make to Iceberg:
   uses a native Iceberg scan, these classes fail to compile and must be removed.
 - Configure test base classes (`TestBase`, `ExtensionsTestBase`, `ScanTestBase`, etc.) to load the Comet Spark
   plugin and shuffle manager
-- Enable the Iceberg write split-operator plan (`spark.comet.write.iceberg.splitOperator.enabled`) alongside the
-  native scan in every Comet-configured session. The flag is off by default for users, so Iceberg's own suites
-  are the only place the split plan (`IcebergCommit -> IcebergWrite`) is exercised against Iceberg's write,
-  commit, and row-level-operation tests. See [#5259]
-- Enable Comet's native (iceberg-rust) Parquet writer (`spark.comet.write.iceberg.enabled`) in the same sessions.
-  The native writer is experimental and off by default for users, so this is where it runs against Iceberg's
-  write, commit, and row-level-operation tests.
+- Enable Comet's Iceberg write path, the split-operator plan (`IcebergCommit -> IcebergWrite`) and the native
+  (iceberg-rust) Parquet writer, alongside the native scan in every Comet-configured session, so that Iceberg's
+  write, commit, and row-level-operation tests run through it. `spark.comet.write.iceberg.enabled` switches both
+  and is on by default; the diffs set it explicitly, together with the testing-only
+  `spark.comet.write.iceberg.splitOperator.enabled`, which it makes redundant. See [#5259]

Review Comment:
   It is still needed for merge-on-read writes on Spark 3.5+, which the write flag alone leaves on Spark's plan. I changed the sentence in cb90a1b5f to say the flag is redundant for the normal write path but still plans `WriteDelta` as the split plan, so Iceberg's row-level-operation tests exercise it.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4252157905


##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,31 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,
+`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite`
+under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of
+each eligible write natively with iceberg-rust. A write that is not eligible still uses
+iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows,
+but natively written files differ from iceberg-java's in the ways listed under
+[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI
+show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator,
+as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the
+testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the
+new default unless it sets `spark.comet.write.iceberg.enabled=false`.
+`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to
+`false` does not turn the split operator off. See [Iceberg Writes](iceberg-writes.md).
+
+The native writer's buffers count against Comet's off-heap memory pool, where iceberg-java's buffers
+sit on the JVM heap. A fanout write keeps a data file open for every partition a task writes to, and
+each open file holds the row group it is writing, so a task that writes to many partitions can need
+more memory than the pool grants it. The task then fails with
+`Additional allocation failed for IcebergWriteExec`. Disabling the fanout writer

Review Comment:
   #6773 is merged, and this branch now includes it. I've rewritten this paragraph to match in 9e5a8be38: a fanout write that the pool can't hold now closes the partitions holding the most memory and finishes with more, smaller files instead of failing. Disabling the fanout writer or a larger `spark.memory.offHeap.size` avoids the extra files. Only a write that keeps one file open, unpartitioned or clustered, can still fail with `Additional allocation failed for IcebergWriteExec`, and only when that one file outgrows the pool. The paragraph links to Failure handling, which has the file counts to expect.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6024238882

   Thanks @comphead. Writing out the four combinations showed we didn't need two settings: the native flag did nothing without the split flag, and the split plan without the native writer gives users nothing over Spark's own operator. So instead of documenting the matrix, I've made `spark.comet.write.iceberg.enabled` the only setting. On, which is the default, Comet plans `IcebergCommit` over `IcebergWrite` and writes eligible data files natively, with iceberg-java writing the rest inside the same plan. Off, Spark plans its own write operator, as in 1.1.0. `spark.comet.write.iceberg.splitOperator.enabled` is back to being a testing-only setting, which the suites use to get the split plan with the native writer off.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable the Iceberg split-operator plan and native writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6028175937

   @sunchao  #5957 is now merged but we're blocked waiting for review on #6740 


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4219518071


##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -117,25 +120,114 @@ class CometIcebergWriteActionSuite
     }
   }
 
-  test("spark.comet.enabled=false keeps Spark's own write plan with the split flag on") {
+  // Turning off Comet, or only its native execution as an application that uses Comet just for
+  // scans or shuffle does, keeps Spark's own write operator.
+  Seq(CometConf.COMET_ENABLED, CometConf.COMET_EXEC_ENABLED).foreach { flag =>
+    test(s"${flag.key}=false keeps Spark's own write plan with the split flag on") {
+      assume(icebergAvailable, "Iceberg not available in classpath")
+      withIcebergCatalog { warehouseDir =>
+        val table = flag.key.replace('.', '_')
+        createTable(warehouseDir, table, partitionSpec = "")
+        // withSQLConf returns Unit before Spark 4.0, so the assertions run inside it.
+        withSQLConf(flag.key -> "false") {
+          val snapshot = captureWrite(table) {
+            spark.sql(s"INSERT INTO cat.db.$table 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,
+            s"expected Spark's own write plan with ${flag.key}=false. Plans:\n" +
+              snapshot.plans.mkString("\n--\n"))
+        }
+        assertRows(table, expectedIds = Seq(1, 2, 3))
+      }
+    }
+  }
+
+  // The suite pins both write flags, so this test unsets them to see what an application that
+  // sets neither gets. A Parquet scan is a native input, so the write is eligible.
+  // https://github.com/apache/datafusion-comet/issues/5644
+  test("an eligible Iceberg write runs natively under the split plan when no flag is set") {
     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)")
+      withTempPath { dir =>
+        spark
+          .range(10)
+          .selectExpr("CAST(id AS INT) AS id", "'eu' AS region", "CAST(id AS DOUBLE) AS amount")
+          .write
+          .parquet(dir.getCanonicalPath)
+        createTable(warehouseDir, "write_defaults", partitionSpec = "")
+        val snapshot = captureWrite("write_defaults") {
+          withSessionConf(
+            CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None,
+            CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) {
+            spark.read
+              .parquet(dir.getCanonicalPath)
+              .writeTo(s"$catalog.$ns.write_defaults")
+              .append()
+          }
         }
         assert(snapshot.snapshotDelta == 1L, s"expected 1 commit, got ${snapshot.snapshotDelta}")
-        val (commits, writes) = collectIcebergWriteOps(snapshot.plans)
+        val (commits, _) = collectIcebergWriteOps(snapshot.plans)
+        val nativeWrites = snapshot.plans.flatMap { plan =>
+          collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e }
+        }
         assert(
-          commits.isEmpty && writes.isEmpty,
-          "expected Spark's own write plan with Comet disabled. Plans:\n" +
+          commits.nonEmpty && nativeWrites.nonEmpty,
+          "expected an IcebergCommitExec over a CometIcebergWriteExec. Plans:\n" +
             snapshot.plans.mkString("\n--\n"))
+        assertRows("write_defaults", expectedIds = 0 until 10)
+      }
+    }
+  }
+
+  // With no flag set, an eligible write becomes a CometIcebergWriteExec, so reverting a
+  // transition-heavy stage has to put a JVM writer back rather than drop the write.
+  // https://github.com/apache/datafusion-comet/issues/5719
+  for (adaptive <- Seq(false, true)) {
+    test(s"transition-heavy fallback keeps the write when no flag is set with AQE=$adaptive") {
+      assume(icebergAvailable, "Iceberg not available in classpath")
+      withIcebergCatalog { warehouseDir =>
+        withTempPath { dir =>
+          spark
+            .range(3)
+            .selectExpr(
+              "CAST(id + 1 AS INT) AS id",
+              "'eu' AS region",
+              "CAST(id AS DOUBLE) AS amount")
+            .write
+            .parquet(dir.getCanonicalPath)
+          val table = s"transition_defaults_${if (adaptive) "aqe" else "no_aqe"}"
+          createTable(warehouseDir, table, partitionSpec = "")
+          // withSQLConf returns Unit before Spark 4.0, so the assertions run inside it.
+          withSQLConf(
+            SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+            CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "true",
+            CometConf.COMET_EXEC_TRANSITION_REVERT_MAX_TRANSITIONS.key -> "0") {
+            val snapshot = captureWrite(table) {
+              withSessionConf(
+                CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None,
+                CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) {
+                spark.read.parquet(dir.getCanonicalPath).writeTo(s"$catalog.$ns.$table").append()
+              }
+            }
+            assertExactlyOneCommit(snapshot)
+            // Also shows the stage was reverted: otherwise the native writer would still be here.
+            val nativeWrites = snapshot.plans.flatMap { plan =>
+              collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e }
+            }
+            assert(
+              nativeWrites.isEmpty,

Review Comment:
   Done in 70bb11db2. The test now runs the same write twice. The first run keeps the default threshold of two transitions, so its stage is not reverted, and it has to plan `IcebergCommit` over `CometIcebergWriteExec`. The second sets `maxTransitions=0` and has to plan `IcebergCommit` over a JVM `IcebergWriteExec` with no `CometIcebergWriteExec` left. Each run has to commit once, and the table ends with both runs' rows. With the Comet scan turned off inside the test, which makes the write ineligible, the first run now fails.
   



##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,31 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,
+`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite`
+under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of
+each eligible write natively with iceberg-rust. A write that is not eligible still uses
+iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows,
+but natively written files differ from iceberg-java's in the ways listed under
+[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI
+show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator,
+as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the
+testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the
+new default unless it sets `spark.comet.write.iceberg.enabled=false`.
+`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to
+`false` does not turn the split operator off. See [Iceberg Writes](iceberg-writes.md).
+
+The native writer's buffers count against Comet's off-heap memory pool, where iceberg-java's buffers
+sit on the JVM heap. A fanout write keeps a data file open for every partition a task writes to, and
+each open file holds the row group it is writing, so a task that writes to many partitions can need
+more memory than the pool grants it. The task then fails with
+`Additional allocation failed for IcebergWriteExec`. Disabling the fanout writer

Review Comment:
   I'd rather not keep fanout writes on iceberg-java. On Iceberg 1.5+ an unsorted partitioned table gets the fanout writer by default whatever its distribution mode, since `useFanoutWriter` is true whenever the write has no required ordering, so that would take most partitioned writes off the native path.
   
   #6773 takes your second suggestion instead. When the pool refuses the writer's reservation, it writes out and closes the partitions holding the most memory, counting both a partition's open file and the rows it has not handed to the file yet, until what is left fits. A closed partition's next rows open a new file, so the write finishes with more, smaller files rather than failing, and a `files closed early to free memory` metric on `CometIcebergWrite` counts them. More files is the same kind of difference as the roll points already listed under accepted divergences, and the PR adds it there.
   
   The suite's 64-partition fanout test now interleaves its partitions and passes under the same 4 MiB pool, writing 219 files where the full pool writes 64. A write that keeps one file open, unpartitioned or clustered, can still fail, but only when that one file's row group outgrows the pool. I've added #6773 to the blockers in the description.
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6132308135

   @viirya #6773 is merged and this branch now includes it, so the fanout blocker is resolved. The upgrade-guide paragraph from your thread now says a fanout write closes partitions instead of failing (9e5a8be38), and I replied there. `run-all-spark-profiles` is still on, so CI is running every profile on this head. Would you be able to take another look? Thanks.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove closed pull request #6664: feat: enable native Iceberg writes by default
URL: https://github.com/apache/datafusion-comet/pull/6664


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#issuecomment-6148134115

   @viirya @sunchao @comphead I think this is ready to go now. WDYT?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4262823808


##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,31 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,
+`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite`
+under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of
+each eligible write natively with iceberg-rust. A write that is not eligible still uses
+iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows,
+but natively written files differ from iceberg-java's in the ways listed under
+[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI
+show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator,
+as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the
+testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the
+new default unless it sets `spark.comet.write.iceberg.enabled=false`.
+`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to
+`false` does not turn the split operator off. See [Iceberg Writes](iceberg-writes.md).
+
+The native writer's buffers count against Comet's off-heap memory pool, where iceberg-java's buffers
+sit on the JVM heap. A fanout write keeps a data file open for every partition a task writes to, and
+each open file holds the row group it is writing, so a task that writes to many partitions can need
+more memory than the pool grants it. The task then fails with
+`Additional allocation failed for IcebergWriteExec`. Disabling the fanout writer

Review Comment:
   Thanks. With #6773 in, a fanout write that outgrows the pool finishes with more, smaller files instead of failing, and this paragraph now says that. Resolving.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat: enable native Iceberg writes by default [datafusion-comet]

Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #6664:
URL: https://github.com/apache/datafusion-comet/pull/6664#discussion_r4212967359


##########
docs/source/user-guide/latest/migration-guide.md:
##########
@@ -83,6 +83,31 @@ it. `spark.comet.convert.oneRowRelation.enabled` is on by default, so such a que
 without either setting, and logs no warning. The list remains the way to convert other leaf
 operators, such as the scan of a Data Source V2 connector.
 
+### Iceberg Writes
+
+`spark.comet.write.iceberg.enabled` now defaults to `true`. Comet plans an Iceberg `INSERT INTO`,
+`INSERT OVERWRITE`, and copy-on-write `DELETE`, `UPDATE` or `MERGE` as two operators, `IcebergWrite`
+under `IcebergCommit`, in place of Spark's single V2 write operator, and writes the data files of
+each eligible write natively with iceberg-rust. A write that is not eligible still uses
+iceberg-java's writer, and iceberg-java still commits every write. The table holds the same rows,
+but natively written files differ from iceberg-java's in the ways listed under
+[Accepted divergences](iceberg-writes.md#accepted-divergences), and explain output and the Spark UI
+show the new operators. Set `spark.comet.write.iceberg.enabled=false` to plan Spark's own operator,
+as in Comet 1.1.0. Comet 1.1.0 named this setting `spark.comet.iceberg.write.enabled`, in the
+testing category, and Comet now ignores that name, so a deployment that set it to `false` gets the
+new default unless it sets `spark.comet.write.iceberg.enabled=false`.
+`spark.comet.write.iceberg.splitOperator.enabled` is now a testing setting, and setting it to
+`false` does not turn the split operator off. See [Iceberg Writes](iceberg-writes.md).
+
+The native writer's buffers count against Comet's off-heap memory pool, where iceberg-java's buffers
+sit on the JVM heap. A fanout write keeps a data file open for every partition a task writes to, and
+each open file holds the row group it is writing, so a task that writes to many partitions can need
+more memory than the pool grants it. The task then fails with
+`Additional allocation failed for IcebergWriteExec`. Disabling the fanout writer

Review Comment:
   This paragraph is clear about the failure mode, but it leaves the user to discover it after a job fails. On Iceberg 1.5+, a partitioned table with `write.distribution-mode=none` and no sort order uses the fanout writer by default. That is a fairly common setup for people avoiding the shuffle. Under this PR those writes go native, their buffers come out of the off-heap pool, and the writer can't spill. A write that fit in the executor heap on 1.1.0 can now fail with `Additional allocation failed for IcebergWriteExec`. Spark's retry takes the same native path, so the job fails.
   
   Would it be reasonable to keep fanout writes on iceberg-java by default for now? Another option is for the native writer to close the largest open file when the pool refuses a reservation, then retry. Roll points are already an accepted divergence, so that wouldn't add a new kind of difference. If neither fits, could you share some numbers on how many partitions a fanout task can hold with a typical `spark.memory.offHeap.size`? Then we can judge whether the default is safe for that shape.



##########
spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala:
##########
@@ -117,25 +120,114 @@ class CometIcebergWriteActionSuite
     }
   }
 
-  test("spark.comet.enabled=false keeps Spark's own write plan with the split flag on") {
+  // Turning off Comet, or only its native execution as an application that uses Comet just for
+  // scans or shuffle does, keeps Spark's own write operator.
+  Seq(CometConf.COMET_ENABLED, CometConf.COMET_EXEC_ENABLED).foreach { flag =>
+    test(s"${flag.key}=false keeps Spark's own write plan with the split flag on") {
+      assume(icebergAvailable, "Iceberg not available in classpath")
+      withIcebergCatalog { warehouseDir =>
+        val table = flag.key.replace('.', '_')
+        createTable(warehouseDir, table, partitionSpec = "")
+        // withSQLConf returns Unit before Spark 4.0, so the assertions run inside it.
+        withSQLConf(flag.key -> "false") {
+          val snapshot = captureWrite(table) {
+            spark.sql(s"INSERT INTO cat.db.$table 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,
+            s"expected Spark's own write plan with ${flag.key}=false. Plans:\n" +
+              snapshot.plans.mkString("\n--\n"))
+        }
+        assertRows(table, expectedIds = Seq(1, 2, 3))
+      }
+    }
+  }
+
+  // The suite pins both write flags, so this test unsets them to see what an application that
+  // sets neither gets. A Parquet scan is a native input, so the write is eligible.
+  // https://github.com/apache/datafusion-comet/issues/5644
+  test("an eligible Iceberg write runs natively under the split plan when no flag is set") {
     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)")
+      withTempPath { dir =>
+        spark
+          .range(10)
+          .selectExpr("CAST(id AS INT) AS id", "'eu' AS region", "CAST(id AS DOUBLE) AS amount")
+          .write
+          .parquet(dir.getCanonicalPath)
+        createTable(warehouseDir, "write_defaults", partitionSpec = "")
+        val snapshot = captureWrite("write_defaults") {
+          withSessionConf(
+            CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None,
+            CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) {
+            spark.read
+              .parquet(dir.getCanonicalPath)
+              .writeTo(s"$catalog.$ns.write_defaults")
+              .append()
+          }
         }
         assert(snapshot.snapshotDelta == 1L, s"expected 1 commit, got ${snapshot.snapshotDelta}")
-        val (commits, writes) = collectIcebergWriteOps(snapshot.plans)
+        val (commits, _) = collectIcebergWriteOps(snapshot.plans)
+        val nativeWrites = snapshot.plans.flatMap { plan =>
+          collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e }
+        }
         assert(
-          commits.isEmpty && writes.isEmpty,
-          "expected Spark's own write plan with Comet disabled. Plans:\n" +
+          commits.nonEmpty && nativeWrites.nonEmpty,
+          "expected an IcebergCommitExec over a CometIcebergWriteExec. Plans:\n" +
             snapshot.plans.mkString("\n--\n"))
+        assertRows("write_defaults", expectedIds = 0 until 10)
+      }
+    }
+  }
+
+  // With no flag set, an eligible write becomes a CometIcebergWriteExec, so reverting a
+  // transition-heavy stage has to put a JVM writer back rather than drop the write.
+  // https://github.com/apache/datafusion-comet/issues/5719
+  for (adaptive <- Seq(false, true)) {
+    test(s"transition-heavy fallback keeps the write when no flag is set with AQE=$adaptive") {
+      assume(icebergAvailable, "Iceberg not available in classpath")
+      withIcebergCatalog { warehouseDir =>
+        withTempPath { dir =>
+          spark
+            .range(3)
+            .selectExpr(
+              "CAST(id + 1 AS INT) AS id",
+              "'eu' AS region",
+              "CAST(id AS DOUBLE) AS amount")
+            .write
+            .parquet(dir.getCanonicalPath)
+          val table = s"transition_defaults_${if (adaptive) "aqe" else "no_aqe"}"
+          createTable(warehouseDir, table, partitionSpec = "")
+          // withSQLConf returns Unit before Spark 4.0, so the assertions run inside it.
+          withSQLConf(
+            SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive.toString,
+            CometConf.COMET_EXEC_TRANSITION_REVERT_ENABLED.key -> "true",
+            CometConf.COMET_EXEC_TRANSITION_REVERT_MAX_TRANSITIONS.key -> "0") {
+            val snapshot = captureWrite(table) {
+              withSessionConf(
+                CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> None,
+                CometConf.COMET_ICEBERG_NATIVE_WRITE_ENABLED.key -> None) {
+                spark.read.parquet(dir.getCanonicalPath).writeTo(s"$catalog.$ns.$table").append()
+              }
+            }
+            assertExactlyOneCommit(snapshot)
+            // Also shows the stage was reverted: otherwise the native writer would still be here.
+            val nativeWrites = snapshot.plans.flatMap { plan =>
+              collectWithSubqueries(plan) { case e: CometIcebergWriteExec => e }
+            }
+            assert(
+              nativeWrites.isEmpty,

Review Comment:
   The only evidence here that the stage was reverted is that no `CometIcebergWriteExec` remains. That would also be true if this write stopped being eligible, for example after a gate change or a change to how the Parquet scan is planned. The test would then keep passing without exercising #5957. Could the test first run the same write without `maxTransitions=0` and assert that it plans a `CometIcebergWriteExec`? Then the reverted run would be compared against a write we know goes native, and the regression test can't pass vacuously.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org