coordinator: display consumer dynamic filters after execution

Cargo.toml

@@ -18,6 +18,11 @@ license = "Apache-2.0"
 documentation = "https://datafusion-contrib.github.io/datafusion-distributed/"
 repository = "https://github.com/datafusion-contrib/datafusion-distributed"
 
+[[test]]
+name = "dynamic_filtering"
+path = "tests/dynamic_filtering/main.rs"
+required-features = ["integration"]
+
 [dependencies]
 chrono = { version = "0.4.44" }
 datafusion = { workspace = true, features = [

docs/source/user-guide/05-metrics.md

@@ -26,8 +26,11 @@ channel, so they are not lost even if the result stream is dropped early (for ex
 
 ## Rendering a plan with metrics
 
-Two functions, both exported from the crate root, do the work:
+These functions, all exported from the crate root, do the work:
 
+- `rewrite_distributed_plan_with_dynamic_filters(plan)` — folds the completed dynamic filters
+  reported by each worker task into an isolated copy of the plan. When displaying both dynamic
+  filters and metrics, apply the dynamic-filter rewrite first.
 - `rewrite_distributed_plan_with_metrics(plan, format)` — folds every task's metrics back into the
   coordinator's copy of the plan. It waits for all worker metrics to arrive, so the result is always
   complete. The `format` is a `DistributedMetricsFormat`:
@@ -54,11 +57,15 @@ execute_stream(plan.clone(), ctx.task_ctx())?
     .try_collect::<Vec<_>>()
     .await?;
 
-// 3. Fold the per-task metrics back into the plan...
+// 3. Fold the completed per-task dynamic filters back into the plan...
+let plan =
+    rewrite_distributed_plan_with_dynamic_filters(plan).await?;
+
+// 4. Fold the per-task metrics back into the plan...
 let plan =
     rewrite_distributed_plan_with_metrics(plan, DistributedMetricsFormat::Aggregated).await?;
 
-// 4. ...and render it.
+// 5. ...and render it.
 println!("{}", display_plan_ascii(plan.as_ref(), true));
 ```
 

src/codec/mod.rs

@@ -4,7 +4,8 @@ mod user_codec;
 
 pub use distributed_codec::DistributedCodec;
 pub(crate) use physical_plan::{
-    decode_execution_plan, decode_partitioning, encode_execution_plan, encode_partitioning,
+    decode_execution_plan, decode_partitioning, decode_physical_expr, encode_execution_plan,
+    encode_partitioning, encode_physical_expr,
 };
 pub(crate) use user_codec::{
     get_distributed_user_codecs, set_distributed_user_codec, set_distributed_user_codec_arc,

src/codec/physical_plan.rs

@@ -1,15 +1,17 @@
 use super::DistributedCodec;
-use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::arrow::datatypes::{Schema, SchemaRef};
 use datafusion::common::Result;
 use datafusion::execution::TaskContext;
-use datafusion::physical_expr::Partitioning;
+use datafusion::physical_expr::{Partitioning, PhysicalExpr};
 use datafusion::physical_plan::ExecutionPlan;
 use datafusion_proto::bytes::{
     physical_plan_from_bytes_with_proto_converter, physical_plan_to_bytes_with_proto_converter,
 };
 use datafusion_proto::physical_plan::from_proto::parse_protobuf_partitioning;
 use datafusion_proto::physical_plan::to_proto::serialize_partitioning;
-use datafusion_proto::physical_plan::{DeduplicatingProtoConverter, PhysicalPlanDecodeContext};
+use datafusion_proto::physical_plan::{
+    DeduplicatingProtoConverter, PhysicalPlanDecodeContext, PhysicalProtoConverterExtension,
+};
 use datafusion_proto::protobuf;
 use datafusion_proto::protobuf::proto_error;
 use prost::Message;
@@ -42,6 +44,26 @@ pub(crate) fn decode_execution_plan(
     physical_plan_from_bytes_with_proto_converter(encoded, task_ctx, &codec, &converter)
 }
 
+pub(crate) fn encode_physical_expr(
+    expression: &Arc<dyn PhysicalExpr>,
+    task_ctx: &TaskContext,
+) -> Result<protobuf::PhysicalExprNode> {
+    let codec = DistributedCodec::new_combined_with_user(task_ctx.session_config());
+    let converter = new_proto_converter();
+    converter.physical_expr_to_proto(expression, &codec)
+}
+
+pub(crate) fn decode_physical_expr(
+    proto: &protobuf::PhysicalExprNode,
+    input_schema: &Schema,
+    task_ctx: &TaskContext,
+) -> Result<Arc<dyn PhysicalExpr>> {
+    let codec = DistributedCodec::new_combined_with_user(task_ctx.session_config());
+    let decode_ctx = PhysicalPlanDecodeContext::new(task_ctx, &codec);
+    let converter = new_proto_converter();
+    converter.proto_to_physical_expr(proto, input_schema, &decode_ctx)
+}
+
 pub(crate) fn encode_partitioning(
     partitioning: &Partitioning,
     task_ctx: &TaskContext,

src/common/maybe_encoded.rs

@@ -1,17 +1,21 @@
 use crate::codec::{
-    decode_execution_plan, decode_partitioning, encode_execution_plan, encode_partitioning,
+    decode_execution_plan, decode_partitioning, decode_physical_expr, encode_execution_plan,
+    encode_partitioning, encode_physical_expr,
 };
-use datafusion::arrow::datatypes::SchemaRef;
+use datafusion::arrow::datatypes::{Schema, SchemaRef};
 use datafusion::common::{Result, internal_err};
 use datafusion::execution::TaskContext;
-use datafusion::physical_expr::Partitioning;
+use datafusion::physical_expr::{Partitioning, PhysicalExpr};
 use datafusion::physical_plan::ExecutionPlan;
+use datafusion_proto::protobuf::PhysicalExprNode;
+use datafusion_proto::protobuf::proto_error;
+use prost::Message;
 use std::sync::Arc;
 
 /// A value that a transport may either leave encoded or materialize in memory.
 /// Users are free to pass [MaybeEncoded::Encoded] or [MaybeEncoded::Decoded] at any
 /// moment and Distributed DataFusion's code will internally know how to handle it.
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub enum MaybeEncoded<T> {
     Encoded(Vec<u8>),
     Decoded(T),
@@ -82,6 +86,44 @@ impl MaybeEncoded<Partitioning> {
     }
 }
 
+impl MaybeEncoded<Arc<dyn PhysicalExpr>> {
+    /// Returns the encoded [`PhysicalExpr`] as protobuf bytes:
+    /// - If in `Decoded` state, it encodes it using the codecs registered in the [`TaskContext`].
+    /// - If in `Encoded` state, it passes through the existing bytes.
+    pub fn encode(self, ctx: &Arc<TaskContext>) -> Result<Vec<u8>> {
+        match self {
+            Self::Encoded(encoded) => Ok(encoded),
+            Self::Decoded(expression) => {
+                Ok(encode_physical_expr(&expression, ctx)?.encode_to_vec())
+            }
+        }
+    }
+
+    /// Returns the decoded [`PhysicalExpr`].
+    /// - If in `Decoded` state, it passes through the expression.
+    /// - If in `Encoded` state, it decodes it using the provided schema and task context.
+    pub fn decode(
+        self,
+        input_schema: &Schema,
+        task_ctx: &TaskContext,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        self.decode_with(|encoded| {
+            let proto = PhysicalExprNode::decode(encoded.as_slice())
+                .map_err(|error| proto_error(error.to_string()))?;
+            decode_physical_expr(&proto, input_schema, task_ctx)
+        })
+    }
+
+    /// Materializes the expression's protobuf representation without changing the stored form.
+    pub(crate) fn to_proto(&self, task_ctx: &TaskContext) -> Result<PhysicalExprNode> {
+        match self {
+            Self::Encoded(encoded) => PhysicalExprNode::decode(encoded.as_slice())
+                .map_err(|error| proto_error(error.to_string())),
+            Self::Decoded(expression) => encode_physical_expr(expression, task_ctx),
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;