You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/06 23:55:09 UTC
(datafusion-comet) branch asf-site updated: Publish built docs triggered by 46d91ef02d80f2a42cd89034699550f45fa13394
This is an automated email from the ASF dual-hosted git repository.
github-actions[bot] pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/asf-site by this push:
new 12ca068bbe Publish built docs triggered by 46d91ef02d80f2a42cd89034699550f45fa13394
12ca068bbe is described below
commit 12ca068bbeed43762b92b5dee9e95a07a0ff3675
Author: github-actions[bot] <gi...@users.noreply.github.com>
AuthorDate: Tue Oct 6 23:55:04 2026 +0000
Publish built docs triggered by 46d91ef02d80f2a42cd89034699550f45fa13394
---
...id-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg | 2 +-
.../contributor-guide/adding_a_new_operator.md.txt | 28 ++++++++++++++++++++++
_sources/contributor-guide/iceberg-writes.md.txt | 13 ++++++----
_sources/contributor-guide/native_shuffle.md.txt | 6 +++++
contributor-guide/adding_a_new_operator.html | 25 +++++++++++++++++++
contributor-guide/iceberg-writes.html | 13 ++++++----
contributor-guide/native_shuffle.html | 5 ++++
searchindex.js | 2 +-
8 files changed, 82 insertions(+), 12 deletions(-)
diff --git a/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg b/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg
index bfdf703a9f..6f7540d38f 100644
--- a/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg
+++ b/_images/mermaid-f5b40459e5fe9b892b845c9198fe1c9420e05d14.svg
@@ -1 +1 @@
-<svg id="my-svg" width="100%" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" class="flowchart" style="max-width: 511.73px; background-color: transparent;" viewBox="0 0 511.7299499511719 966" role="graphics-document document" aria-roledescription="flowchart-v2"><style>#my-svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#my-svg [...]
\ No newline at end of file
+<svg id="my-svg" width="100%" xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" class="flowchart" style="max-width: 511.73px; background-color: transparent;" viewBox="0 0 511.7299499511719 966" role="graphics-document document" aria-roledescription="flowchart-v2"><style>#my-svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#my-svg [...]
\ No newline at end of file
diff --git a/_sources/contributor-guide/adding_a_new_operator.md.txt b/_sources/contributor-guide/adding_a_new_operator.md.txt
index 15c4e9169b..09cac3a1c7 100644
--- a/_sources/contributor-guide/adding_a_new_operator.md.txt
+++ b/_sources/contributor-guide/adding_a_new_operator.md.txt
@@ -731,6 +731,34 @@ Use `QueryPlanSerde.exprToProto` to convert Spark expressions to protobuf:
val protoExpr = exprToProto(sparkExpr, inputSchema)
```
+### Restoring the Spark operator (`sparkFallback`)
+
+`CometExec.originalPlan` is the Spark operator this node replaced. `CometExecRule` copies
+`originalPlan.logicalLink` onto the Comet node, which is how AQE finds the node again when it
+re-plans a stage. `RevertNativeForTransitionHeavyStages` calls `sparkFallback(newChildren)` to
+rebuild that Spark operator with the children of the reverted stage.
+
+The default implementation is `originalPlan.withNewChildren(newChildren)`. It refuses a null
+`originalPlan`, an `originalPlan` that is one of the node's own children, or a different number of
+children than the Spark operator has.
+
+Override `sparkFallback` when conversion changes the plan shape, so the restored node is not that
+Spark operator with the same children. `CometNativeWriteExec` replaces a `DataWritingCommandExec`
+and drops the `WriteFilesExec` under it; its override puts that wrapper back around the restored
+input. `CometIcebergWriteExec` keeps the same shape as `IcebergWriteExec`, so the default is
+enough.
+
+Also override `sparkFallback` when the operator's live state differs from `originalPlan`.
+`CometNativeScanExec` restores its current partition and data filters, and
+`CometIcebergNativeScanExec` restores its current runtime filters, so AQE's executable DPP
+subqueries survive reversion. `CometLocalTopKExec` returns the restored child directly: Comet
+inserted that local candidate selection, and only the outer TopK restores Spark's offset and
+projection. Rebuilding the original TopK at both nodes would apply it twice.
+
+Do not point `originalPlan` at a child. If that child is a shuffle or query stage, the copied
+logical link puts this node inside the stage's `LogicalQueryStage`. AQE then re-plans a second
+copy of the operator around the one that is already there.
+
### Handling Fallback
Use `withInfo` to tag operators with fallback reasons:
diff --git a/_sources/contributor-guide/iceberg-writes.md.txt b/_sources/contributor-guide/iceberg-writes.md.txt
index a90cff0807..e254776282 100644
--- a/_sources/contributor-guide/iceberg-writes.md.txt
+++ b/_sources/contributor-guide/iceberg-writes.md.txt
@@ -107,9 +107,9 @@ serializer requires that tag on Spark 4.1+, leaving stock writers on Spark throu
## From `IcebergWrite` to `CometIcebergWrite`
`CometExecRule` converts an `IcebergWriteExec` with the `CometIcebergNativeWrite` operator serde
-when `spark.comet.write.iceberg.enabled` is on. Two arms in `CometExecRule` handle it: one unwraps
-the double conversion AQE can produce when it re-fires write planning over a sub-tree that already
-contains a `CometIcebergWriteExec`, and the other calls `convertToComet`.
+when `spark.comet.write.iceberg.enabled` is on. A single arm in `CometExecRule` handles the
+conversion by calling `convertToComet`. The converted node keeps the `IcebergWriteExec` as its
+`originalPlan`, so AQE re-plans the write from that node.
`CometIcebergNativeWrite.requiresNativeChildren` is `true`. The native writer consumes Arrow
batches from its child over FFI, so the conversion is declined unless the child is already a Comet
@@ -479,8 +479,11 @@ Each of these has caused a bug on this path:
([#5691](https://github.com/apache/datafusion-comet/issues/5691),
[#5693](https://github.com/apache/datafusion-comet/issues/5693),
[#6141](https://github.com/apache/datafusion-comet/issues/6141)).
-- **Plan rewrites must keep the write node.** Rules that restore Spark operators from a Comet
- node's `originalPlan` have to handle the write execs, whose `originalPlan` today is their child
+- **Plan rewrites must keep the write node.** `CometIcebergWriteExec.originalPlan` is the
+ `IcebergWriteExec` it replaced. Restoring Spark execution goes through
+ [`CometExec.sparkFallback`](adding_a_new_operator.md#restoring-the-spark-operator-sparkfallback),
+ which rebuilds that node around the reverted children. Pointing `originalPlan` at the child
+ drops the write, and AQE then treats the write as part of that child's stage
([#5719](https://github.com/apache/datafusion-comet/issues/5719)).
- **The kill switch must still work.** Code that runs for Iceberg writes has to respect
`spark.comet.enabled`, so that disabling Comet restores Spark's own plan
diff --git a/_sources/contributor-guide/native_shuffle.md.txt b/_sources/contributor-guide/native_shuffle.md.txt
index 6108844788..e9cf777441 100644
--- a/_sources/contributor-guide/native_shuffle.md.txt
+++ b/_sources/contributor-guide/native_shuffle.md.txt
@@ -39,6 +39,12 @@ Compare this to JVM shuffle's data path:
Comet Native (columnar) → ColumnarToRowExec → rows → JVM Shuffle → Arrow IPC → columnar
```
+When `RevertNativeForTransitionHeavyStages` restores the map stage to Spark execution, the
+native exchange stays in place. Its input still needs Arrow-backed Comet vectors, even if the
+restored Spark operator supports columnar output. The rule adds `CometSparkToColumnarExec` to
+convert either Spark rows or Spark columnar batches to Arrow before the native shuffle consumes
+them. Spark's `RowToColumnarExec` alone does not satisfy this input contract.
+
## When Native Shuffle is Used
Native shuffle (`CometExchange`) is selected when all of the following conditions are met:
diff --git a/contributor-guide/adding_a_new_operator.html b/contributor-guide/adding_a_new_operator.html
index 01fadd1779..5d3ad55758 100644
--- a/contributor-guide/adding_a_new_operator.html
+++ b/contributor-guide/adding_a_new_operator.html
@@ -1231,6 +1231,30 @@ query-stage wrappers when checking the native operator and reused exchange.</p>
</pre></div>
</div>
</section>
+<section id="restoring-the-spark-operator-sparkfallback">
+<h3>Restoring the Spark operator (<code class="docutils literal notranslate"><span class="pre">sparkFallback</span></code>)<a class="headerlink" href="#restoring-the-spark-operator-sparkfallback" title="Link to this heading">#</a></h3>
+<p><code class="docutils literal notranslate"><span class="pre">CometExec.originalPlan</span></code> is the Spark operator this node replaced. <code class="docutils literal notranslate"><span class="pre">CometExecRule</span></code> copies
+<code class="docutils literal notranslate"><span class="pre">originalPlan.logicalLink</span></code> onto the Comet node, which is how AQE finds the node again when it
+re-plans a stage. <code class="docutils literal notranslate"><span class="pre">RevertNativeForTransitionHeavyStages</span></code> calls <code class="docutils literal notranslate"><span class="pre">sparkFallback(newChildren)</span></code> to
+rebuild that Spark operator with the children of the reverted stage.</p>
+<p>The default implementation is <code class="docutils literal notranslate"><span class="pre">originalPlan.withNewChildren(newChildren)</span></code>. It refuses a null
+<code class="docutils literal notranslate"><span class="pre">originalPlan</span></code>, an <code class="docutils literal notranslate"><span class="pre">originalPlan</span></code> that is one of the node’s own children, or a different number of
+children than the Spark operator has.</p>
+<p>Override <code class="docutils literal notranslate"><span class="pre">sparkFallback</span></code> when conversion changes the plan shape, so the restored node is not that
+Spark operator with the same children. <code class="docutils literal notranslate"><span class="pre">CometNativeWriteExec</span></code> replaces a <code class="docutils literal notranslate"><span class="pre">DataWritingCommandExec</span></code>
+and drops the <code class="docutils literal notranslate"><span class="pre">WriteFilesExec</span></code> under it; its override puts that wrapper back around the restored
+input. <code class="docutils literal notranslate"><span class="pre">CometIcebergWriteExec</span></code> keeps the same shape as <code class="docutils literal notranslate"><span class="pre">IcebergWriteExec</span></code>, so the default is
+enough.</p>
+<p>Also override <code class="docutils literal notranslate"><span class="pre">sparkFallback</span></code> when the operator’s live state differs from <code class="docutils literal notranslate"><span class="pre">originalPlan</span></code>.
+<code class="docutils literal notranslate"><span class="pre">CometNativeScanExec</span></code> restores its current partition and data filters, and
+<code class="docutils literal notranslate"><span class="pre">CometIcebergNativeScanExec</span></code> restores its current runtime filters, so AQE’s executable DPP
+subqueries survive reversion. <code class="docutils literal notranslate"><span class="pre">CometLocalTopKExec</span></code> returns the restored child directly: Comet
+inserted that local candidate selection, and only the outer TopK restores Spark’s offset and
+projection. Rebuilding the original TopK at both nodes would apply it twice.</p>
+<p>Do not point <code class="docutils literal notranslate"><span class="pre">originalPlan</span></code> at a child. If that child is a shuffle or query stage, the copied
+logical link puts this node inside the stage’s <code class="docutils literal notranslate"><span class="pre">LogicalQueryStage</span></code>. AQE then re-plans a second
+copy of the operator around the one that is already there.</p>
+</section>
<section id="handling-fallback">
<h3>Handling Fallback<a class="headerlink" href="#handling-fallback" title="Link to this heading">#</a></h3>
<p>Use <code class="docutils literal notranslate"><span class="pre">withInfo</span></code> to tag operators with fallback reasons:</p>
@@ -1365,6 +1389,7 @@ query-stage wrappers when checking the native operator and reused exchange.</p>
<li class="toc-h2 nav-item toc-entry"><a class="reference internal nav-link" href="#implementing-a-jvm-operator">Implementing a JVM Operator</a></li>
<li class="toc-h2 nav-item toc-entry"><a class="reference internal nav-link" href="#common-patterns-and-helpers">Common Patterns and Helpers</a><ul class="nav section-nav flex-column">
<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#expression-conversion">Expression Conversion</a></li>
+<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#restoring-the-spark-operator-sparkfallback">Restoring the Spark operator (<code class="docutils literal notranslate"><span class="pre">sparkFallback</span></code>)</a></li>
<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#handling-fallback">Handling Fallback</a></li>
<li class="toc-h3 nav-item toc-entry"><a class="reference internal nav-link" href="#child-operator-validation">Child Operator Validation</a></li>
</ul>
diff --git a/contributor-guide/iceberg-writes.html b/contributor-guide/iceberg-writes.html
index 53e225332a..331648782f 100644
--- a/contributor-guide/iceberg-writes.html
+++ b/contributor-guide/iceberg-writes.html
@@ -666,9 +666,9 @@ serializer requires that tag on Spark 4.1+, leaving stock writers on Spark throu
<section id="from-icebergwrite-to-cometicebergwrite">
<h2>From <code class="docutils literal notranslate"><span class="pre">IcebergWrite</span></code> to <code class="docutils literal notranslate"><span class="pre">CometIcebergWrite</span></code><a class="headerlink" href="#from-icebergwrite-to-cometicebergwrite" title="Link to this heading">#</a></h2>
<p><code class="docutils literal notranslate"><span class="pre">CometExecRule</span></code> converts an <code class="docutils literal notranslate"><span class="pre">IcebergWriteExec</span></code> with the <code class="docutils literal notranslate"><span class="pre">CometIcebergNativeWrite</span></code> operator serde
-when <code class="docutils literal notranslate"><span class="pre">spark.comet.write.iceberg.enabled</span></code> is on. Two arms in <code class="docutils literal notranslate"><span class="pre">CometExecRule</span></code> handle it: one unwraps
-the double conversion AQE can produce when it re-fires write planning over a sub-tree that already
-contains a <code class="docutils literal notranslate"><span class="pre">CometIcebergWriteExec</span></code>, and the other calls <code class="docutils literal notranslate"><span class="pre">convertToComet</span></code>.</p>
+when <code class="docutils literal notranslate"><span class="pre">spark.comet.write.iceberg.enabled</span></code> is on. A single arm in <code class="docutils literal notranslate"><span class="pre">CometExecRule</span></code> handles the
+conversion by calling <code class="docutils literal notranslate"><span class="pre">convertToComet</span></code>. The converted node keeps the <code class="docutils literal notranslate"><span class="pre">IcebergWriteExec</span></code> as its
+<code class="docutils literal notranslate"><span class="pre">originalPlan</span></code>, so AQE re-plans the write from that node.</p>
<p><code class="docutils literal notranslate"><span class="pre">CometIcebergNativeWrite.requiresNativeChildren</span></code> is <code class="docutils literal notranslate"><span class="pre">true</span></code>. The native writer consumes Arrow
batches from its child over FFI, so the conversion is declined unless the child is already a Comet
native operator. This is why a write fed by a <code class="docutils literal notranslate"><span class="pre">LocalTableScanExec</span></code> (<code class="docutils literal notranslate"><span class="pre">INSERT</span> <span class="pre">...</span> <span class="pre">VALUES</span></code>, a local
@@ -1072,8 +1072,11 @@ such a field with a live one is declined by the gate, since iceberg-java cannot
(<a class="reference external" href="https://github.com/apache/datafusion-comet/issues/5691">#5691</a>,
<a class="reference external" href="https://github.com/apache/datafusion-comet/issues/5693">#5693</a>,
<a class="reference external" href="https://github.com/apache/datafusion-comet/issues/6141">#6141</a>).</p></li>
-<li><p><strong>Plan rewrites must keep the write node.</strong> Rules that restore Spark operators from a Comet
-node’s <code class="docutils literal notranslate"><span class="pre">originalPlan</span></code> have to handle the write execs, whose <code class="docutils literal notranslate"><span class="pre">originalPlan</span></code> today is their child
+<li><p><strong>Plan rewrites must keep the write node.</strong> <code class="docutils literal notranslate"><span class="pre">CometIcebergWriteExec.originalPlan</span></code> is the
+<code class="docutils literal notranslate"><span class="pre">IcebergWriteExec</span></code> it replaced. Restoring Spark execution goes through
+<a class="reference internal" href="adding_a_new_operator.html#restoring-the-spark-operator-sparkfallback"><span class="std std-ref"><code class="docutils literal notranslate"><span class="pre">CometExec.sparkFallback</span></code></span></a>,
+which rebuilds that node around the reverted children. Pointing <code class="docutils literal notranslate"><span class="pre">originalPlan</span></code> at the child
+drops the write, and AQE then treats the write as part of that child’s stage
(<a class="reference external" href="https://github.com/apache/datafusion-comet/issues/5719">#5719</a>).</p></li>
<li><p><strong>The kill switch must still work.</strong> Code that runs for Iceberg writes has to respect
<code class="docutils literal notranslate"><span class="pre">spark.comet.enabled</span></code>, so that disabling Comet restores Spark’s own plan
diff --git a/contributor-guide/native_shuffle.html b/contributor-guide/native_shuffle.html
index f064d7f3e0..fbf10662dc 100644
--- a/contributor-guide/native_shuffle.html
+++ b/contributor-guide/native_shuffle.html
@@ -566,6 +566,11 @@ overhead that JVM shuffle incurs.</p>
<div class="highlight-default notranslate"><div class="highlight"><pre><span></span>Comet Native (columnar) → ColumnarToRowExec → rows → JVM Shuffle → Arrow IPC → columnar
</pre></div>
</div>
+<p>When <code class="docutils literal notranslate"><span class="pre">RevertNativeForTransitionHeavyStages</span></code> restores the map stage to Spark execution, the
+native exchange stays in place. Its input still needs Arrow-backed Comet vectors, even if the
+restored Spark operator supports columnar output. The rule adds <code class="docutils literal notranslate"><span class="pre">CometSparkToColumnarExec</span></code> to
+convert either Spark rows or Spark columnar batches to Arrow before the native shuffle consumes
+them. Spark’s <code class="docutils literal notranslate"><span class="pre">RowToColumnarExec</span></code> alone does not satisfy this input contract.</p>
</section>
<section id="when-native-shuffle-is-used">
<h2>When Native Shuffle is Used<a class="headerlink" href="#when-native-shuffle-is-used" title="Link to this heading">#</a></h2>
diff --git a/searchindex.js b/searchindex.js
index 01cd985b74..1834ffec24 100644
--- a/searchindex.js
+++ b/searchindex.js
@@ -1 +1 @@
-Search.setIndex({"alltitles": {"!": [[58, "id1"]], "%": [[56, "id1"]], "&": [[46, "id1"]], "*": [[56, "id2"]], "+": [[56, "id3"]], "-": [[56, "id4"]], "/": [[56, "id5"]], "1. Format Your Code": [[43, "format-your-code"]], "1. Install Comet": [[64, "install-comet"], [76, "install-comet"]], "1. Native Operators (nativeExecs map)": [[29, "native-operators-nativeexecs-map"]], "2. Build and Verify": [[43, "build-and-verify"]], "2. Clone Iceberg and Apply Diff": [[64, "clone-iceberg-and-apply- [...]
\ No newline at end of file
+Search.setIndex({"alltitles": {"!": [[58, "id1"]], "%": [[56, "id1"]], "&": [[46, "id1"]], "*": [[56, "id2"]], "+": [[56, "id3"]], "-": [[56, "id4"]], "/": [[56, "id5"]], "1. Format Your Code": [[43, "format-your-code"]], "1. Install Comet": [[64, "install-comet"], [76, "install-comet"]], "1. Native Operators (nativeExecs map)": [[29, "native-operators-nativeexecs-map"]], "2. Build and Verify": [[43, "build-and-verify"]], "2. Clone Iceberg and Apply Diff": [[64, "clone-iceberg-and-apply- [...]
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org