adriangb opened a new pull request, #26273:
URL: https://github.com/apache/datafusion/pull/26273
## Which issue does this PR close?
- Closes apache/datafusion#26231.
Stacked on apache/datafusion#25580 and on #26269 (apache/datafusion#26238).
Only the last commit is new.
## Rationale for this change
A plan that has a materialized CTE cannot be sent to another process.
```rust
let df = ctx
.sql("WITH c AS MATERIALIZED (SELECT * FROM (VALUES (1), (2)) AS t(v)) \
SELECT * FROM c AS a JOIN c AS b ON a.v = b.v")
.await?;
logical_plan_to_bytes(df.logical_plan())?;
physical_plan_to_bytes(df.create_physical_plan().await?)?;
```
Before this PR, the two calls return an error. After this PR, the two calls
succeed. The decoded physical plan runs the body of the CTE one time, and all
the scans read its buffer.
## What changes are included in this PR?
- New protobuf messages `MaterializedCteNode`, `MaterializedCteScanNode`,
`MaterializedCteExecNode` and `MaterializedCteScanExecNode`.
- Logical plans: the encoder and the decoder keep the id of the CTE. A
logical id is used only inside one plan tree, and the physical planner gives
each CTE a new id.
- Physical plans: the encoding has only the id and the name of the CTE, not
the buffer.
- `PhysicalPlanDecodeContext` holds a `DecodedMaterializedCteBuffers`
registry. Each decoded node asks the registry for the buffer of its id. Thus a
scan and its `MaterializedCteExec` share one buffer in any decode order, also
when the scan is in a scalar subquery and is not below the
`MaterializedCteExec`.
- Decoded buffers get new ids from `MaterializedCteId::next()`, because an
id from another process can be equal to a local id.
- A plan fragment that has a scan but not its `MaterializedCteExec` can be
decoded. It fails when the scan is executed, with an error that says that the
body and the scans must be in one plan and run in one process.
Known limits:
- An extension codec that decodes embedded plan bytes with a new
`PhysicalPlanDecodeContext` has a separate registry. A CTE that is split across
such a boundary fails with the error above.
- The branches for apache/datafusion#26240 (`projection` on the scan) and
apache/datafusion#26239 (`pushed_filter`) add fields that this PR does not
encode. The PR that merges second must add them.
## Are these changes tested?
Yes. Physical round trip of a CTE with two scans and of a CTE with a scan in
a scalar subquery; each test runs the decoded plan and checks the result and
that the body ran one time. A test for a scan without its
`MaterializedCteExec`. Logical round trip of the same two queries, before and
after the optimizer. The error for two decoded `MaterializedCteExec` nodes with
one id has no test.
## Are there any user-facing changes?
Yes. Plans with `WITH ... AS MATERIALIZED` can be serialized with
`datafusion-proto`. API additions: `MaterializedCteId::from_u64`,
`MaterializedCteBuffer::name`, `DecodedMaterializedCteBuffers`,
`try_from_proto` on the two physical nodes, and
`ExecutionPlanDecodeCtx::materialized_cte_buffer`. The `#[doc(hidden)]` trait
`ExecutionPlanDecode` has a new required method.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
