fn plan_extension(
        &self,
        planner: &dyn PhysicalPlanner,
        node: &dyn UserDefinedLogicalNode,
        logical_inputs: &[&LogicalPlan],
        physical_inputs: &[Arc<dyn ExecutionPlan>],
        ctx_state: &ExecutionContextState,
    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
        let any = node.as_any();
        let plan = if let Some(schema_pivot) = any.downcast_ref::<SchemaPivotNode>() {
            assert_eq!(physical_inputs.len(), 1, "Inconsistent number of inputs");
            Some(Arc::new(SchemaPivotExec::new(
                Arc::clone(&physical_inputs[0]),
                schema_pivot.schema().as_ref().clone().into(),
            )) as Arc<dyn ExecutionPlan>)
        } else if let Some(non_null_checker) = any.downcast_ref::<NonNullCheckerNode>() {
            assert_eq!(physical_inputs.len(), 1, "Inconsistent number of inputs");
            Some(Arc::new(NonNullCheckerExec::new(
                Arc::clone(&physical_inputs[0]),
                non_null_checker.schema().as_ref().clone().into(),
                non_null_checker.value(),
            )) as Arc<dyn ExecutionPlan>)
        } else if let Some(stream_split) = any.downcast_ref::<StreamSplitNode>() {
            assert_eq!(
                logical_inputs.len(),
                1,
                "Inconsistent number of logical inputs"
            );
            assert_eq!(
                physical_inputs.len(),
                1,
                "Inconsistent number of physical inputs"
            );

            let split_expr = planner.create_physical_expr(
                stream_split.split_expr(),
                logical_inputs[0].schema(),
                &physical_inputs[0].schema(),
                ctx_state,
            )?;

            Some(Arc::new(StreamSplitExec::new(
                Arc::clone(&physical_inputs[0]),
                split_expr,
            )) as Arc<dyn ExecutionPlan>)
        } else {
            None
        };
        Ok(plan)
    }
/// Create a physical plan for an extension node
Create a physical plan for an extension node
Create a physical plan for an extension node
fn plan_extension(
        &self,
        planner: &dyn PhysicalPlanner,
        node: &dyn UserDefinedLogicalNode,
        logical_inputs: &[&LogicalPlan],
        physical_inputs: &[Arc<dyn ExecutionPlan>],
        ctx_state: &ExecutionContextState,
    ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
        let any = node.as_any();
        let plan = if let Some(schema_pivot) = any.downcast_ref::<SchemaPivotNode>() {
            assert_eq!(physical_inputs.len(), 1, "Inconsistent number of inputs");
            Some(Arc::new(SchemaPivotExec::new(
                Arc::clone(&physical_inputs[0]),
                schema_pivot.schema().as_ref().clone().into(),
            )) as Arc<dyn ExecutionPlan>)
        } else if let Some(non_null_checker) = any.downcast_ref::<NonNullCheckerNode>() {
            assert_eq!(physical_inputs.len(), 1, "Inconsistent number of inputs");
            Some(Arc::new(NonNullCheckerExec::new(
                Arc::clone(&physical_inputs[0]),
                non_null_checker.schema().as_ref().clone().into(),
                non_null_checker.value(),
            )) as Arc<dyn ExecutionPlan>)
        } else if let Some(stream_split) = any.downcast_ref::<StreamSplitNode>() {
            assert_eq!(
                logical_inputs.len(),
                1,
                "Inconsistent number of logical inputs"
            );
            assert_eq!(
                physical_inputs.len(),
                1,
                "Inconsistent number of physical inputs"
            );
            let split_expr = planner.create_physical_expr(
                stream_split.split_expr(),
                logical_inputs[0].schema(),
                &physical_inputs[0].schema(),
                ctx_state,
            )?;
            Some(Arc::new(StreamSplitExec::new(
                Arc::clone(&physical_inputs[0]),
                split_expr,
            )) as Arc<dyn ExecutionPlan>)
        } else {
            None
        };
        Ok(plan)
    }
fn plan_extension ( & self , planner : & dyn PhysicalPlanner , node : & dyn UserDefinedLogicalNode , logical_inputs : & [ & LogicalPlan ] , physical_inputs : & [ Arc < dyn ExecutionPlan > ] , ctx_state : & ExecutionContextState , ) -> Result < Option < Arc < dyn ExecutionPlan > > > { let any = node . as_any ( ) ; let plan = if let Some ( schema_pivot ) = any . downcast_ref :: < SchemaPivotNode > ( ) { assert_eq ! ( physical_inputs len ( ) 1 " " ) ; Some ( Arc :: new ( SchemaPivotExec :: new ( Arc :: clone ( & physical_inputs [ 0 ] ) , schema_pivot . schema ( ) . as_ref ( ) . clone ( ) . into ( ) , ) ) as Arc < dyn ExecutionPlan > ) } else if let Some ( non_null_checker ) = any . downcast_ref :: < NonNullCheckerNode > ( ) { assert_eq ! ( physical_inputs len ( ) 1 " " ) ; Some ( Arc :: new ( NonNullCheckerExec :: new ( Arc :: clone ( & physical_inputs [ 0 ] ) , non_null_checker . schema ( ) . as_ref ( ) . clone ( ) . into ( ) , non_null_checker . value ( ) , ) ) as Arc < dyn ExecutionPlan > ) } else if let Some ( stream_split ) = any . downcast_ref :: < StreamSplitNode > ( ) { assert_eq ! ( logical_inputs len ( ) 1 " " ) ; assert_eq ! ( physical_inputs len ( ) 1 " " ) ; let split_expr = planner . create_physical_expr ( stream_split . split_expr ( ) , logical_inputs [ 0 ] . schema ( ) , & physical_inputs [ 0 ] . schema ( ) , ctx_state , ) ? ; Some ( Arc :: new ( StreamSplitExec :: new ( Arc :: clone ( & physical_inputs [ 0 ] ) , split_expr , ) ) as Arc < dyn ExecutionPlan > ) } else { None } ; Ok ( plan ) }
Create a physical plan for an extension node
Create a physical plan for an extension node
&self
planner
&dyn PhysicalPlanner
node
&dyn UserDefinedLogicalNode
logical_inputs
&[&LogicalPlan]
physical_inputs
&[Arc<dyn ExecutionPlan>]
ctx_state
&ExecutionContextState
&self
planner
&dyn PhysicalPlanner
node
&dyn UserDefinedLogicalNode
logical_inputs
&[&LogicalPlan]
physical_inputs
&[Arc<dyn ExecutionPlan>]
ctx_state
&ExecutionContextState