pub trait ToStream {
// Required methods
fn logical_rewrite_for_stream(
&self,
ctx: &mut RewriteStreamContext,
) -> Result<(LogicalPlanRef, ColIndexMapping)>;
fn to_stream(&self, ctx: &mut ToStreamContext) -> Result<StreamPlanRef>;
// Provided methods
fn to_stream_with_dist_required(
&self,
required_dist: &RequiredDist,
ctx: &mut ToStreamContext,
) -> Result<StreamPlanRef> { ... }
fn try_better_locality(&self, _columns: &[usize]) -> Option<LogicalPlanRef> { ... }
}Expand description
ToStream converts a logical plan node to streaming physical node
with an optional required distribution.
when implement this trait you can choose the two ways
- Implement
to_streamand use the default implementation ofto_stream_with_dist_required - Or, if the required distribution is given, there will be a better plan. For example a hash
join with hash-key(a,b) and the plan is required hash-distributed by (a,b,c). you can
implement
to_stream_with_dist_required, and implementto_streamwithto_stream_with_dist_required(RequiredDist::Any). you can seeLogicalProjectas an example.
Required Methods§
Sourcefn logical_rewrite_for_stream(
&self,
ctx: &mut RewriteStreamContext,
) -> Result<(LogicalPlanRef, ColIndexMapping)>
fn logical_rewrite_for_stream( &self, ctx: &mut RewriteStreamContext, ) -> Result<(LogicalPlanRef, ColIndexMapping)>
logical_rewrite_for_stream will rewrite the logical node, and return (new_plan_node,
col_mapping), the col_mapping is for original columns have been changed into some other
position.
Now it is used to:
- ensure every plan node’s output having pk column
- add
row_count() in every Agg
Sourcefn to_stream(&self, ctx: &mut ToStreamContext) -> Result<StreamPlanRef>
fn to_stream(&self, ctx: &mut ToStreamContext) -> Result<StreamPlanRef>
to_stream is equivalent to to_stream_with_dist_required(RequiredDist::Any)
Provided Methods§
Sourcefn to_stream_with_dist_required(
&self,
required_dist: &RequiredDist,
ctx: &mut ToStreamContext,
) -> Result<StreamPlanRef>
fn to_stream_with_dist_required( &self, required_dist: &RequiredDist, ctx: &mut ToStreamContext, ) -> Result<StreamPlanRef>
convert the plan to streaming physical plan and satisfy the required distribution