alamb commented on code in PR #24018:
URL: https://github.com/apache/datafusion/pull/24018#discussion_r3738643098
##########
datafusion-examples/examples/relation_planner/table_sample.rs:
##########
@@ -749,6 +749,22 @@ impl ExecutionPlan for SampleExec {
Ok(Arc::new(stats))
}
+
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(
+ &Arc<dyn datafusion::physical_plan::PhysicalExpr>,
+ ) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ datafusion::physical_plan::apply_expression_roots(
+ self.cache
+ .output_ordering()
+ .into_iter()
+ .flatten()
+ .map(|sort_expr| &sort_expr.expr),
+ f,
+ )
Review Comment:
How about this?
```rust
properties.apply_expressions(f)
```
##########
docs/source/library-user-guide/upgrading/55.0.0.md:
##########
@@ -986,205 +894,58 @@ let plan = deserialize_bytes(&proto_bytes)?;
See [PR #23827](https://github.com/apache/datafusion/pull/23827) for details.
-### `WindowExpr::evaluate_stateful` now takes a `WindowEvalContext`
-
-`WindowExpr::evaluate_stateful` (and the provided
-`AggregateWindowExpr::aggregate_evaluate_stateful` method) take a new
-`WindowEvalContext` argument carrying stream-level information that is shared
-by all partitions:
-
-```rust,ignore
-// Before
-fn evaluate_stateful(
- &self,
- partition_batches: &PartitionBatches,
- window_agg_state: &mut PartitionWindowAggStates,
-) -> Result<()>
-
-// After
-fn evaluate_stateful(
- &self,
- partition_batches: &PartitionBatches,
- window_agg_state: &mut PartitionWindowAggStates,
- eval_ctx: &WindowEvalContext<'_>,
-) -> Result<()>
-```
-
-`WindowEvalContext` currently carries the most recent input row, which
-previously lived in each partition's `PartitionBatchState` (see the next
-section). The struct is `#[non_exhaustive]` so that fields can be added
-without further signature changes: construct it with
-`WindowEvalContext::default()` and set fields through its builder methods.
-
-**Who is affected:**
-
-- Implementations of the `WindowExpr` trait that override `evaluate_stateful`
- must add the new parameter.
-- Callers of `evaluate_stateful` or `aggregate_evaluate_stateful` must pass a
- context.
-
-**Migration guide:**
-
-```rust,ignore
-use datafusion_physical_expr::window::WindowEvalContext;
-
-// Before
-window_expr.evaluate_stateful(&partition_batches, &mut window_agg_state)?;
-
-// After
-let eval_ctx = WindowEvalContext::default()
- .with_most_recent_row(most_recent_row.as_ref());
-window_expr.evaluate_stateful(
- &partition_batches,
- &mut window_agg_state,
- &eval_ctx,
-)?;
-```
-
-Pass `WindowEvalContext::default()` when no most-recent-row watermark is
-available (for example, when the input is sorted by the partition keys and
-partition ends are detected directly).
-
-### `PartitionBatchState::most_recent_row` removed
+### `ExecutionPlan::apply_expressions` is now a required method
-The `most_recent_row` field and the `set_most_recent_row` method have been
-removed from `datafusion_expr::window_state::PartitionBatchState`. The most
-recent input row is a property of the whole input stream rather than
-per-partition state: every partition observed the same value. It is now
-tracked once by the operator driving the evaluation and passed to window
-expressions through the new `WindowEvalContext` argument of
-`WindowExpr::evaluate_stateful` described above.
+`apply_expressions` has been added as a **required** method on the
`ExecutionPlan`, `FileSource`, and `DataSource` traits. Any custom
implementation of
+these traits must now implement `apply_expressions`.
**Who is affected:**
-- Code that read `PartitionBatchState::most_recent_row` or called
- `set_most_recent_row`, such as custom streaming window operators.
+- Users who implement custom `ExecutionPlan` nodes
+- Users who implement custom `FileSource` or `DataSource` sources
**Migration guide:**
-Track the most recent input row once per stream (for example, a one-row
-slice of the last non-empty input batch) and pass it to window expressions
-via `WindowEvalContext::with_most_recent_row` instead of copying it into
-each partition's state.
+Add `apply_expressions` to your implementation. Call `f` on each top-level
Review Comment:
This is fine, though it is technically redundant with the (very nice docs)
on apply_expressions
##########
datafusion-examples/examples/relation_planner/table_sample.rs:
##########
@@ -749,6 +749,22 @@ impl ExecutionPlan for SampleExec {
Ok(Arc::new(stats))
}
+
+ fn apply_expressions(
+ &self,
+ f: &mut dyn FnMut(
+ &Arc<dyn datafusion::physical_plan::PhysicalExpr>,
+ ) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ datafusion::physical_plan::apply_expression_roots(
+ self.cache
+ .output_ordering()
+ .into_iter()
+ .flatten()
+ .map(|sort_expr| &sort_expr.expr),
+ f,
+ )
Review Comment:
(can be a follow on PR)
--
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]
