[IGNITE-20383] Calcite engine. Convert one input of a join to the broadcast distribution

• Type: Improvement
• Status: In Progress
• Priority: Major
• Resolution: Unresolved
• Labels: calcite calcite2-required calcite3-required ise
• Ignite Flags: Release Notes Required
• Assignee: Aleksey Plekhanov
• Reporter: Aleksey Plekhanov
• Created: 08/Sep/23 07:09
• Updated: 15/Sep/23 06:45

Description

Sometimes, if join inputs are not collocated it's worth to broadcast one of the inputs, for example, query:

SELECT * FROM emps WHERE emps.salary = (SELECT AVG(emps.salary) FROM emps)

Currently has plan:

IgniteProject(ID=[$0], NAME=[$1], SALARY=[$2])
  IgniteNestedLoopJoin(condition=[=($2, $3)], joinType=[inner])
    IgniteExchange(distribution=[single])
      IgniteTableScan(table=[[PUBLIC, EMPS]])
    IgniteReduceHashAggregate(group=[{}], AVG(EMPS.SALARY)=[AVG($0)])
      IgniteExchange(distribution=[single])
        IgniteMapHashAggregate(group=[{}], AVG(EMPS.SALARY)=[AVG($0)])
          IgniteIndexScan(table=[[PUBLIC, EMPS]], index=[TST], requiredColumns=[{2}], collation=[[2 ASC-nulls-first]])

But this plan is not optimal, since we should send entire table EMP from all nodes to the single node. For such a query it's better to broadcast result of the aggregation, in this case plan will be something like:

IgniteExchange(distribution=[single])
  IgniteProject(...)
    IgniteCorrelatedNestedLoopJoin(...)
      IgniteExchange(distribution=[broadcast])
        IgniteReduceHashAggregate(group=[{}], AVG(EMPS.SALARY)=[AVG($0)])
          IgniteExchange(distribution=[single])
            IgniteMapHashAggregate(group=[{}], AVG(EMPS.SALARY)=[AVG($0)])
              IgniteIndexScan(table=[[PUBLIC, EMPS]], index=[SALARY_IDX])
      IgniteIndexScan(table=[[PUBLIC, EMPS]], index=[SALARY_IDX])

But currently we don't try to convert any of the join inputs to the broadcast distribution. We should try to do this.