Matt/recursive cte eliminate distribution and coalesce in recursive children

[matthewgapp]

This a WIP. It attempts to eliminate unnecessary (and costly) physical plan steps within a recursive query, speeding recursive queries up by ~30x but perhaps more in some cases.

Specifically, it prevents RepartitionExec and CoalesceExec when they'd otherwise be fsfsdecedents of a RecursiveQueryExec. It does, however, allow them to be applied to parts of the plan that aren't under a RecursiveQueryExec.

The PR makes changes to DistributionContext and introduces the context pattern into the Coalesce physical optimizer to determine if a given plan is within a RecursiveQuery.

Example:

  EXPLAIN WITH RECURSIVE nodes(id, region, account, increase) AS (

    SELECT CAST(id as BIGINT) as id, users.region as region, account, 0 as increase
    FROM users

    UNION ALL

    SELECT id + 1 as id, growth.region as region, account + growth.increase as account, growth.increase as increase
    FROM nodes
    JOIN growth ON nodes.region = growth.region

    WHERE id < 10000
)
SELECT id, account, region FROM nodes
WHERE id > 9950

Before

cargo run -r time: 6.5 seconds

+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                                           |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Projection: users.id, users.account, region                                                                                                                    |
|               |   Filter: CAST(users.id AS Int64) > Int64(99500)                                                                                                               |
|               |     Projection: users.id, region, users.account                                                                                                                |
|               |       RecursiveQuery: is_distinct=false                                                                                                                        |
|               |         Projection: users.id, users.region AS region, users.account, Int64(0) AS increase                                                                      |
|               |           TableScan: users projection=[id, account, region]                                                                                                    |
|               |         Projection: CAST(users.id AS Int64) + Int64(1) AS id, growth.region AS region, users.account + growth.increase AS account, growth.increase AS increase |
|               |           Inner Join: nodes.region = growth.region                                                                                                             |
|               |             Filter: CAST(users.id AS Int64) < Int64(100000)                                                                                                    |
|               |               Projection: users.id, nodes.region, users.account                                                                                                |
|               |                 NamedRelation: nodes                                                                                                                           |
|               |             TableScan: growth projection=[increase, region]                                                                                                    |
| physical_plan | ProjectionExec: expr=[id@0 as id, account@2 as account, region@1 as region]                                                                                    |
|               |   CoalesceBatchesExec: target_batch_size=8192                                                                                                                  |
|               |     FilterExec: CAST(id@0 AS Int64) > 99500                                                                                                                    |
|               |       RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1                                                                                    |
|               |         ProjectionExec: expr=[id@0 as id, region@1 as region, account@2 as account]                                                                            |
|               |           RecursiveQueryExec: is_distinct=false                                                                                                                |
|               |             ProjectionExec: expr=[id@0 as id, region@2 as region, account@1 as account, 0 as increase]                                                         |
|               |               MemoryExec: partitions=1, partition_sizes=[1]                                                                                                    |
|               |             CoalescePartitionsExec                                                                                                                             |
|               |               ProjectionExec: expr=[CAST(id@0 AS Int64) + 1 as id, region@4 as region, account@2 + increase@3 as account, increase@3 as increase]              |
|               |                 CoalesceBatchesExec: target_batch_size=8192                                                                                                    |
|               |                   HashJoinExec: mode=Partitioned, join_type=Inner, on=[(region@1, region@1)]                                                                   |
|               |                     CoalesceBatchesExec: target_batch_size=8192                                                                                                |
|               |                       RepartitionExec: partitioning=Hash([region@1], 12), input_partitions=12                                                                  |
|               |                         CoalesceBatchesExec: target_batch_size=8192                                                                                            |
|               |                           FilterExec: CAST(id@0 AS Int64) < 100000                                                                                             |
|               |                             RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1                                                              |
|               |                               ProjectionExec: expr=[id@0 as id, region@1 as region, account@2 as account]                                                      |
|               |                                 ContinuanceExec: name=nodes                                                                                                    |
|               |                     CoalesceBatchesExec: target_batch_size=8192                                                                                                |
|               |                       RepartitionExec: partitioning=Hash([region@1], 12), input_partitions=12                                                                  |
|               |                         RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1                                                                  |
|               |                           MemoryExec: partitions=1, partition_sizes=[1]                                                                                        |
|               |                                                                                                                                                                |
+---------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------+

After

cargo run -r time: 220ms

+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type     | plan                                                                                                                                            |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan  | Projection: id, users.account, region                                                                                                           |
|               |   Filter: id > Int64(9950)                                                                                                                      |
|               |     Projection: id, region, users.account                                                                                                       |
|               |       RecursiveQuery: is_distinct=false                                                                                                         |
|               |         Projection: CAST(users.id AS Int64) AS id, users.region AS region, users.account, Int64(0) AS increase                                  |
|               |           TableScan: users projection=[id, account, region]                                                                                     |
|               |         Projection: nodes.id + Int64(1) AS id, growth.region AS region, users.account + growth.increase AS account, growth.increase AS increase |
|               |           Inner Join: nodes.region = growth.region                                                                                              |
|               |             Filter: nodes.id < Int64(10000)                                                                                                     |
|               |               Projection: nodes.id, nodes.region, users.account                                                                                 |
|               |                 NamedRelation: nodes                                                                                                            |
|               |             TableScan: growth projection=[increase, region]                                                                                     |
| physical_plan | ProjectionExec: expr=[id@0 as id, account@2 as account, region@1 as region]                                                                     |
|               |   CoalesceBatchesExec: target_batch_size=8192                                                                                                   |
|               |     FilterExec: id@0 > 9950                                                                                                                     |
|               |       RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1                                                                     |
|               |         ProjectionExec: expr=[id@0 as id, region@1 as region, account@2 as account]                                                             |
|               |           RecursiveQueryExec: is_distinct=false                                                                                                 |
|               |             ProjectionExec: expr=[CAST(id@0 AS Int64) as id, region@2 as region, account@1 as account, 0 as increase]                           |
|               |               MemoryExec: partitions=1, partition_sizes=[1]                                                                                     |
|               |             ProjectionExec: expr=[id@0 + 1 as id, region@4 as region, account@2 + increase@3 as account, increase@3 as increase]                |
|               |               HashJoinExec: mode=Partitioned, join_type=Inner, on=[(region@1, region@1)]                                                        |
|               |                 FilterExec: id@0 < 10000                                                                                                        |
|               |                   ProjectionExec: expr=[id@0 as id, region@1 as region, account@2 as account]                                                   |
|               |                     ContinuanceExec: name=nodes                                                                                                 |
|               |                 MemoryExec: partitions=1, partition_sizes=[1]                                                                                   |
|               |                                                                                                                                                 |
+---------------+-------------------------------------------------------------------------------------------------------------------------------------------------+

Which issue does this PR close?

Closes #.

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

[matthewgapp]

rebased onto updated (rebased) cte PR. Plan to ready this PR by tomorrow.

[matthewgapp]

hey this is a test