feat: support stateless physical plans

This patch introduces the stateless physical plan feature. Currently, the physical-plan crate
is fully supported. This feature allows for the reuse of physical plans and their concurrent
execution.

The feature is implemented by adding a separate Cargo feature named "stateless_plan".
The implementation consists of several parts:

* State tree.

With the "stateless_plan" feature enabled, the plans themselves do not store state. The state
is stored in a separate tree composed of PlanStateNodes, which is built lazily during plan execution.
Each node of the tree stores not only the shared state of the plan but also its metrics. The shape
of the state tree matches the shape of the execution plan tree.

* Metrics

Metrics are stored in the nodes of the state tree and can be accessed after plan execution. Support
is provided for performing EXPLAIN using the state.

* Dynamic Filters

In the case of stateless plans, dynamic filters cannot simply be stored inside the plans, as the same
plan can be executed concurrently. To overcome this, a dynamic filter is split into two parts: a planning-time
version and an execution-time version. The plans contain the planning-time version, which is transformed into
the execution version during the execution phase and then passed from parent nodes to child nodes using the
state tree.

* WorkTable

Instead of explicitly injecting the WorkTable into nodes, RecursiveExec exposes the WorkTable in the state stored
within the State Tree. Then, a node interested in obtaining the WorkTable traverses up the State Tree and thus
retrieves the current WorkTable.

Planned following work:

- Support stateless plan for all other DataFusion crates.
- Enable running tests with this feature in CI.
- Deprecate stateful plans to eventually transition completely to the stateless version.
- Add `fmt_as_with_state` to allow plans to include state-specific details in the EXPLAIN output, such as dynamic filters.

Closes apache#19351

Author: askalt

datafusion/datasource-parquet/src/opener.rs

@@ -1303,10 +1303,8 @@ mod test {
     }
 
     fn make_dynamic_expr(expr: Arc<dyn PhysicalExpr>) -> Arc<dyn PhysicalExpr> {
-        Arc::new(DynamicFilterPhysicalExpr::new(
-            expr.children().into_iter().map(Arc::clone).collect(),
-            expr,
-        ))
+        let children = expr.children().into_iter().map(Arc::clone).collect();
+        Arc::new(DynamicFilterPhysicalExpr::new(expr, children))
     }
 
     #[tokio::test]

datafusion/execution/src/metrics/mod.rs

@@ -89,9 +89,9 @@ pub struct Metric {
 ///   will be shown.
 /// - When set to `summary`, only metrics with type `MetricType::Summary` are shown.
 ///
-/// # Difference from `EXPLAIN ANALYZE VERBOSE`:  
-/// The `VERBOSE` keyword controls whether per-partition metrics are shown (when specified),  
-/// or aggregated metrics are displayed (when omitted).  
+/// # Difference from `EXPLAIN ANALYZE VERBOSE`:
+/// The `VERBOSE` keyword controls whether per-partition metrics are shown (when specified),
+/// or aggregated metrics are displayed (when omitted).
 /// In contrast, the `analyze_level` configuration determines which categories or
 /// levels of metrics are displayed.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -213,6 +213,16 @@ impl MetricsSet {
         Default::default()
     }
 
+    /// Return a number of metrics.
+    pub fn len(&self) -> usize {
+        self.metrics.len()
+    }
+
+    /// Check if the set is empty.
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
     /// Add the specified metric
     pub fn push(&mut self, metric: Arc<Metric>) {
         self.metrics.push(metric)