(
&self,
num_input_columns: usize,
physical_expr: PlannedExprResult,
schema: &Schema,
)
| 3025 | } |
| 3026 | |
| 3027 | fn try_plan_async_exprs( |
| 3028 | &self, |
| 3029 | num_input_columns: usize, |
| 3030 | physical_expr: PlannedExprResult, |
| 3031 | schema: &Schema, |
| 3032 | ) -> Result<PlanAsyncExpr> { |
| 3033 | let mut async_map = AsyncMapper::new(num_input_columns); |
| 3034 | match &physical_expr { |
| 3035 | PlannedExprResult::ExprWithName(exprs) => { |
| 3036 | exprs |
| 3037 | .iter() |
| 3038 | .try_for_each(|(expr, _)| async_map.find_references(expr, schema))?; |
| 3039 | } |
| 3040 | PlannedExprResult::Expr(exprs) => { |
| 3041 | exprs |
| 3042 | .iter() |
| 3043 | .try_for_each(|expr| async_map.find_references(expr, schema))?; |
| 3044 | } |
| 3045 | } |
| 3046 | |
| 3047 | if async_map.is_empty() { |
| 3048 | return Ok(PlanAsyncExpr::Sync(physical_expr)); |
| 3049 | } |
| 3050 | |
| 3051 | let new_exprs = match physical_expr { |
| 3052 | PlannedExprResult::ExprWithName(exprs) => PlannedExprResult::ExprWithName( |
| 3053 | exprs |
| 3054 | .iter() |
| 3055 | .map(|(expr, column_name)| { |
| 3056 | let new_expr = Arc::clone(expr) |
| 3057 | .transform_up(|e| Ok(async_map.map_expr(e)))?; |
| 3058 | Ok((new_expr.data, column_name.to_string())) |
| 3059 | }) |
| 3060 | .collect::<Result<_>>()?, |
| 3061 | ), |
| 3062 | PlannedExprResult::Expr(exprs) => PlannedExprResult::Expr( |
| 3063 | exprs |
| 3064 | .iter() |
| 3065 | .map(|expr| { |
| 3066 | let new_expr = Arc::clone(expr) |
| 3067 | .transform_up(|e| Ok(async_map.map_expr(e)))?; |
| 3068 | Ok(new_expr.data) |
| 3069 | }) |
| 3070 | .collect::<Result<_>>()?, |
| 3071 | ), |
| 3072 | }; |
| 3073 | // rewrite the projection's expressions in terms of the columns with the result of async evaluation |
| 3074 | Ok(PlanAsyncExpr::Async(async_map, new_exprs)) |
| 3075 | } |
| 3076 | } |
| 3077 | |
| 3078 | #[derive(Debug)] |
no test coverage detected