timsaucer commented on code in PR #18813:
URL: https://github.com/apache/datafusion/pull/18813#discussion_r2542340514
##########
datafusion/proto/src/physical_plan/from_proto.rs:
##########
@@ -385,54 +367,49 @@ pub fn parse_physical_expr(
like_expr.case_insensitive,
parse_required_physical_expr(
like_expr.expr.as_deref(),
- ctx,
+ parser,
"expr",
input_schema,
- codec,
)?,
parse_required_physical_expr(
like_expr.pattern.as_deref(),
- ctx,
+ parser,
"pattern",
input_schema,
- codec,
)?,
)),
ExprType::Extension(extension) => {
let inputs: Vec<Arc<dyn PhysicalExpr>> = extension
.inputs
.iter()
- .map(|e| parse_physical_expr(e, ctx, input_schema, codec))
+ .map(|e| parse_physical_expr(e, parser, input_schema))
.collect::<Result<_>>()?;
- (codec.try_decode_expr(extension.expr.as_slice(), &inputs)?) as _
+ parser.try_decode_expr(extension.expr.as_slice(), &inputs)? as _
Review Comment:
I'm trying to understand what additional hook you need. Here is an example
using the latest push. Of course my example doesn't really have an encoder so
it doesn't parse through to give a count on each one of the physical
expressions, so the output is :
```
Pre-encoding: 0 expressions.
post: 1 expressions
```
Minimal example:
```rust
#[cfg(test)]
mod tests {
use std::sync::Arc;
use datafusion::logical_expr::Operator;
use datafusion_common::{DataFusionError, ScalarValue};
use datafusion_expr::{lit, AggregateUDF, ScalarUDF, WindowUDF};
use datafusion_physical_expr::expressions::{BinaryExpr, Column, Literal};
use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
use datafusion_physical_plan::ExecutionPlan;
use crate::physical_plan::{PhysicalSerializer};
struct PrintOnSerialize {
num_exprs: u8,
}
impl PhysicalSerializer for PrintOnSerialize {
fn try_encode_execution_plan(&mut self, node: Arc<dyn
ExecutionPlan>, buf: &mut Vec<u8>) -> datafusion_common::Result<()> {
todo!()
}
fn try_encode_udf(&mut self, node: &ScalarUDF, buf: &mut Vec<u8>) ->
datafusion_common::Result<()> {
todo!()
}
fn try_encode_expr(&mut self, node: &Arc<dyn PhysicalExpr>, buf:
&mut Vec<u8>) -> datafusion_common::Result<()> {
println!("Pre-encoding: {} expressions.", self.num_exprs);
self.num_exprs += 1;
// Call to inner actual encoder happens here
buf.push(self.num_exprs);
println!(" post: {} expressions", self.num_exprs);
Ok(())
}
fn try_encode_udaf(&mut self, _node: &AggregateUDF, _buf: &mut
Vec<u8>) -> datafusion_common::Result<()> {
todo!()
}
fn try_encode_udwf(&mut self, _node: &WindowUDF, _buf: &mut Vec<u8>)
-> datafusion_common::Result<()> {
todo!()
}
}
#[test]
fn demonstrate_serialization_hooks() -> Result<(), DataFusionError> {
let literal_expr = Literal::new(ScalarValue::Boolean(Some(true)));
let column_expr = Column::new("a", 0);
let binary_expr = BinaryExpr::new(Arc::new(literal_expr),
Operator::And, Arc::new(column_expr));
let binary_expr = Arc::new(binary_expr) as Arc<dyn PhysicalExpr>;
let mut custom_serialize = PrintOnSerialize {
num_exprs: 0,
};
let mut bytes = Vec::new();
custom_serialize.try_encode_expr(&binary_expr, &mut bytes)?;
Ok(())
}
}
```
--
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]
