make DistributedCodec not encode udfs

Author: gabotechs

src/protobuf/distributed_codec.rs

@@ -6,9 +6,10 @@ use crate::{NetworkShuffleExec, PartitionIsolatorExec};
 use bytes::Bytes;
 use datafusion::arrow::datatypes::Schema;
 use datafusion::arrow::datatypes::SchemaRef;
-use datafusion::common::internal_datafusion_err;
+use datafusion::common::{internal_datafusion_err, not_impl_err};
 use datafusion::error::DataFusionError;
 use datafusion::execution::FunctionRegistry;
+use datafusion::logical_expr::{AggregateUDF, ScalarUDF, WindowUDF};
 use datafusion::physical_expr::EquivalenceProperties;
 use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType};
 use datafusion::physical_plan::{ExecutionPlan, Partitioning, PlanProperties};
@@ -228,6 +229,18 @@ impl PhysicalExtensionCodec for DistributedCodec {
             Err(proto_error(format!("Unexpected plan {}", node.name())))
         }
     }
+
+    fn try_encode_udf(&self, _: &ScalarUDF, _: &mut Vec<u8>) -> datafusion::common::Result<()> {
+        not_impl_err!("DistributedCodec does not encode UDFs")
+    }
+
+    fn try_encode_udaf(&self, _: &AggregateUDF, _: &mut Vec<u8>) -> datafusion::common::Result<()> {
+        not_impl_err!("DistributedCodec does not encode UDAFs")
+    }
+
+    fn try_encode_udwf(&self, _: &WindowUDF, _: &mut Vec<u8>) -> datafusion::common::Result<()> {
+        not_impl_err!("DistributedCodec does not encode UDWFs")
+    }
 }
 
 /// A key that uniquely identifies a stage in a query