ORCA: allow not enforce distribution key in 3-stage aggregate

[jiaqizho]

Fixes#659

Change log

In the case where the group-by key is a distribution key, ORCA disables 3-stage aggregate by default.

Add a new GUC(optimizer_enable_use_distribution_in_dqa) to allow not enforce distribution key in 3-stage aggregate in ORCA. which enables local deduplication before the 2-stage agg.

Type of Change

• Bug fix (non-breaking change)
• New feature (non-breaking change)
• Breaking change (fix or feature with breaking changes)
• Documentation update

Breaking Changes

Test Plan

• Unit tests added/updated
• Integration tests added/updated
• Passed make installcheck
• Passed make -C src/test installcheck-cbdb-parallel

Impact

Performance:

User-facing changes:

Dependencies:

Checklist

• Followed contribution guide
• Added/updated documentation
• Reviewed code for security implications
• Requested review from cloudberry committers

Additional Context

CI Skip Instructions

---

[jiaqizho]

The context about this PR:

ORCA can create the 3-stage agg in agg(distinct) path, but prohibit the distribution key in distinct. Because ORCA think 2-stage agg is enough.

So u can see the 3-stage agg path but no cost generated in the memo. like

ROOT
Group 9 (#GExprs: 8):
0: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] [ 0 8 ]
1: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Three Stage Scalar DQA ] [ 12 15 ]
2: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 16 15 ]
3: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 20 25 ]
4: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 20 25 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 484.101386
5: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 16 15 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 679.103561
6: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Three Stage Scalar DQA ] [ 12 15 ]
Cost Ctxts:
7: CPhysicalScalarAgg( Global ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] [ 0 8 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 527.493765
Grp OptCtxts:
0 (stage 0): (req CTEs: [], req order: [<empty> match: satisfy ], req dist: [SINGLETON (master) match: satisfy], req rewind: [], req rewind: [NONE NO-MOTION match: satisfy], req partition propagation: [<empty> match: satisfy ]) => Best Expr:4

The GroupExpression#6 is the Three Stage Scalar DQA.

Details of why this path have no cost:

1. ORCA will create a CPhysicalMotionHashDistribute HASHED GroupExpression(Redistribute Motion) below the GroupExpression#6. If current physical agg enforce the distribution.
2. If current required distribution key(hashagg) is same as distribution key, then Redistribute Motion won't get the cost(in CPhysicalMotion::FValidContext)
3. So the root group(GroupExpression#6) won't be selected.

[my-ship-it]

Just tried, but cannot produce 3-stage aggregate still, the cost is still higher than 2-stage aggregate.

Group 9 (#GExprs: 8):
0: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] [ 0 8 ]
1: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Three Stage Scalar DQA ] [ 12 15 ]
2: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 16 15 ]
3: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 20 25 ]
4: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 20 25 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 566.388669
5: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 16 15 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 1090.678248
6: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Three Stage Scalar DQA ] [ 12 15 ]
Cost Ctxts:
7: CPhysicalScalarAgg( Global ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] [ 0 8 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 765.687983
Grp OptCtxts:
0 (stage 0): (req CTEs: [], req order: [<empty> match: satisfy ], req dist: [SINGLETON (master) match: satisfy], req rewind: [], req rewind: [NONE NO-MOTION match: satisfy], req partition propagation: [<empty> match: satisfy ]) => Best Expr:4

[jiaqizho]

Just tried, but cannot produce 3-stage aggregate still, the cost is still higher than 2-stage aggregate.

Group 9 (#GExprs: 8):
0: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] [ 0 8 ]
1: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Three Stage Scalar DQA ] [ 12 15 ]
2: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 16 15 ]
3: CLogicalGbAgg( Global ) Grp Cols: [][Global], Minimal Grp Cols: [], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 20 25 ]
4: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 20 25 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 566.388669
5: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Two Stage Scalar DQA ] [ 16 15 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 1090.678248
6: CPhysicalScalarAgg( Global, multi-stage ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] , m_aggStage :[ Three Stage Scalar DQA ] [ 12 15 ]
Cost Ctxts:
7: CPhysicalScalarAgg( Global ) Grp Cols: [], Minimal Grp Cols:[], Generates Duplicates :[ 0 ] [ 0 8 ]
Cost Ctxts:
main ctxt (stage 0)0.1, child ctxts:[0], rows:1.000000 (group), cost: 765.687983
Grp OptCtxts:
0 (stage 0): (req CTEs: [], req order: [<empty> match: satisfy ], req dist: [SINGLETON (master) match: satisfy], req rewind: [], req rewind: [NONE NO-MOTION match: satisfy], req partition propagation: [<empty> match: satisfy ]) => Best Expr:4

also need running with GUC optimizer_force_three_stage_scalar_dqa/optimizer_force_multistage_agg.

updated the test cases...

[my-ship-it]

Good work. LGTM