Tracking: better plan for projection expression by using output indices

[st1page]

Background

in our project operator, there is 2 kind of expressions

1. calculation expressions, like Add(input_ref(0), input_ref(1)
2. projection expression, those only one input_ref expression.
   and this issue will talk about how to reduce the projection expression in project operators.

for 2 reason

1. they are really different during optimizing, the projection expression might be able to reduce the column of the input operators. so we want to split them into two project operators. for example, we'd like to do calculations distributedly, but want to push down the column projection
2. and during optimizing, such as when unnesting or join reordering, we will construct lots of projection operators with the 2nd expression which is really trouble.

Solution

To eliminate these logical projection nodes we introduce output_index.

col_prune v2

To preserve the information of column order, we introduce column prune v2. We should change the definition fn prune_col(&self, required_cols: &FixedBitSet) -> PlanRef; to fn prune_col(&self, required_cols: &[usize]) -> PlanRef; . Therefore we can prune columns while reordering the columns.

Output index

A field output_index: Vec<usize> will be added on LogicalAgg, LogicalJoin, LogicalHopWindow, and any logical plan node which needs extra projection before. It will represent the operator's output column index based on our current schema of PlanNode.
for example

SELECT t2.a, t1.a from t1 join t2 on t1.b = t2.b

in our current implementation, we will get a join plan node with schema [t1.a, t1.b, t2.a, t2.b], and a project on the join with expressions [input_ref(2), input_ref(0)].
And with the output_index there will only be a join plan node with output_index [2, 0]

Implementation considerations

When adding this field to the plan nodes, we should carefully re-think those already existing functions.

• the schema of the plan node(output index will change the schema of the plan node)
• the property(dist, order, pk...) derive in the new()
• the o2i_mapping and other similar functions
• rewrite_with_input, rewrite_for_stream.

The output_index is natural in our chunk-based vectorized query execution, so we can add the field on the proto and implement it in executors in the future. but now we can just add a stream/batch project node when converting the logical plan to a stream/batch plan.

[Enter-tainer]

Tracking

Column prune

• Change the function signature to fn prune_col(&self, required_cols: &[usize]) -> PlanRef; refactor(optimizer): column prune refactor #2095
• Ensure every logical node correctly handle the reorder of columns when performing column prune. test(optimizer): ensure column prune correctly handle reorder #2603 refactor(optimizer): make LogicalHopWindow column order aware #2667

Output Indices

• Add output indices for LogicalHopWindow feat(optimizer): add output_indices for LogicalHopWindow #2769

• Add output indices for LogicalJoin feat(optimizer): Add output indices to LogicalJoin #2748

• Add output indices for LogicalAgg

• Add output indices for LogicalFilter

• Add output indices to hop_window executor feat(executor): add output indices to HopWindow executor #2922

• Add output indices to join executor

[jon-chuang]

Hmm, my guess is that it would be better to keep both the original schema and the new schema if we have output_indices not_none, so that we can keep prune_col's idempotency.

[Enter-tainer]

Could I ask what would be the schema of a logical node with output_index? Would it be restricted and reordered as well? I believe may actually not be better to not do so, as it will make prune_col lose its idempotency.

IMO the schema will be changed as well. This is based on the idea of "hiding" projections inside these nodes. And add these projections back, when converting logical nodes to stream/batch nodes. I'm currently working on LogicalHopWindow and still exploring where to put output_indices. I put output_indices and the original schema in LogicalHopWindow itself. The reason why I do not put it in PlanBase is because only logical nodes need output_indices, stream/batch nodes do not have it.

BTW, I'm not sure if we are on the same page, but I think column prune does not have idempotency. For example, if we call col_column([1, 3, 5]) on some logical node twice, the second time will fail because we simply do not have column 3 and column 5 --- they have been pruned in the first call and we only have 3 columns now. hmm I'm not sure what idempotency is in column prune. Can you explain more about this?

Anyway, I think it is worth a discussion on whether to keep the schema or not. cc @st1page what's your idea?

[st1page]

And add these projections back, when converting logical nodes to stream/batch nodes

and for some operators such as hash_join, we can even implement the behaviour in the executor

[st1page]

when the prune_col of input is called, it is expected to change the schema as the prune_col requiring