You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudberry.apache.org by ma...@apache.org on 2025/02/18 05:40:07 UTC
(cloudberry) 28/31: Improve DPv2 algorithm to include distribution spec information with partition selectors
This is an automated email from the ASF dual-hosted git repository.
maxyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
commit 06215e3713b86521f66ba2470ca4584092b1beb0
Author: Chris Hajas <ch...@vmware.com>
AuthorDate: Mon Nov 27 17:34:51 2023 -0800
Improve DPv2 algorithm to include distribution spec information with partition selectors
DPv2 produces a few different join order alternatives. One of these
alternatives, stored in m_top_k_part_expressions, attempts to produce a
plan while considering the effects of partition selection. This was
intially added in 9c44532. However, when considering a partition
selector's effect, we also need to consider the distribution spec.
Otherwise we may choose a particular join order that may appear to be a
valid join order with a partition selector, but it actually isn't due to
a mismatch of distribution specs. In many cases, the DPv2 algorithm's
partition selection logic wasn't producing a valid join order.
Consider the tables P1, P2, T1, T2 all inner joined together. We
currently produce the join order:
```
PS2
PT2
PT1
PS1
```
This occurs in the hope that PS1 is a partition selector for PT1, and PS2 is a partition selector for PT2. That is, the join tree ideally would look like:
```
HJ3
HJ2 PS2
PT2 HJ1
PT1 PS1
```
However, if PT2 has a distribution spec that is not in HJ2's condition,
it will have a redistribute. At this point, PS2 is in a different slice
and a partition selector here isn't valid.
Now, we check that the distribution cols of PT2 is included in the hash
condition of HJ2. More generally, when checking whether a partition
selector will have any effect, we check if the outer child's (HJ2) join
condition contains the distribution columns of the partition table we're
impacting.
---
.../libgpopt/include/gpopt/operators/CLogical.h | 10 +-
.../libgpopt/include/gpopt/xforms/CJoinOrderDPv2.h | 4 +
.../gporca/libgpopt/src/operators/CLogical.cpp | 6 +
.../gporca/libgpopt/src/xforms/CJoinOrderDPv2.cpp | 207 +++++++++++++--------
4 files changed, 146 insertions(+), 81 deletions(-)
diff --git a/src/backend/gporca/libgpopt/include/gpopt/operators/CLogical.h b/src/backend/gporca/libgpopt/include/gpopt/operators/CLogical.h
index 0c23032186..1fcf5f5d14 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/operators/CLogical.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/operators/CLogical.h
@@ -64,11 +64,6 @@ public:
EspHigh // operator has high priority for stat derivation
};
-private:
- // private copy ctor
- CLogical(const CLogical &);
-
-
protected:
// set of locally used columns
CColRefSet *m_pcrsLocalUsed;
@@ -176,6 +171,8 @@ protected:
CColRefSet *pcrsUsedAdditional);
public:
+ CLogical(const CLogical &) = delete;
+
// ctor
explicit CLogical(CMemoryPool *mp);
@@ -330,9 +327,6 @@ public:
// returns the table descriptor for (Dynamic)(BitmapTable)Get operators
static CTableDescriptor *PtabdescFromTableGet(COperator *pop);
- // returns the output columns for selected operator
- static CColRefArray *PoutputColsFromTableGet(COperator *pop);
-
// extract the output columns descriptor from a logical get or dynamic get operator
static CColRefArray *PdrgpcrOutputFromLogicalGet(CLogical *pop);
diff --git a/src/backend/gporca/libgpopt/include/gpopt/xforms/CJoinOrderDPv2.h b/src/backend/gporca/libgpopt/include/gpopt/xforms/CJoinOrderDPv2.h
index 6ff3d342d8..0550fa5b84 100644
--- a/src/backend/gporca/libgpopt/include/gpopt/xforms/CJoinOrderDPv2.h
+++ b/src/backend/gporca/libgpopt/include/gpopt/xforms/CJoinOrderDPv2.h
@@ -516,6 +516,10 @@ private:
void FinalizeDPLevel(ULONG level);
+ CDouble CostJoinWithPartitionSelection(
+ SExpressionInfo *join_expr_info, SExpressionInfo *atom_ps,
+ SGroupInfo *pt_atom, SGroupInfo *part_selector_group_info);
+
SGroupInfoArray *
GetGroupsForLevel(ULONG level) const
{
diff --git a/src/backend/gporca/libgpopt/src/operators/CLogical.cpp b/src/backend/gporca/libgpopt/src/operators/CLogical.cpp
index 196e3862d6..bbed275569 100644
--- a/src/backend/gporca/libgpopt/src/operators/CLogical.cpp
+++ b/src/backend/gporca/libgpopt/src/operators/CLogical.cpp
@@ -27,7 +27,9 @@
#include "gpopt/operators/CLogicalApply.h"
#include "gpopt/operators/CLogicalBitmapTableGet.h"
#include "gpopt/operators/CLogicalDynamicBitmapTableGet.h"
+#include "gpopt/operators/CLogicalDynamicForeignGet.h"
#include "gpopt/operators/CLogicalDynamicGet.h"
+#include "gpopt/operators/CLogicalForeignGet.h"
#include "gpopt/operators/CLogicalGet.h"
#include "gpopt/operators/CLogicalNAryJoin.h"
#include "gpopt/operators/CLogicalSelect.h"
@@ -1263,6 +1265,10 @@ CLogical::PtabdescFromTableGet(COperator *pop)
return CLogicalDynamicGet::PopConvert(pop)->Ptabdesc();
case CLogical::EopLogicalBitmapTableGet:
return CLogicalBitmapTableGet::PopConvert(pop)->Ptabdesc();
+ case CLogical::EopLogicalForeignGet:
+ return CLogicalForeignGet::PopConvert(pop)->Ptabdesc();
+ case CLogical::EopLogicalDynamicForeignGet:
+ return CLogicalDynamicForeignGet::PopConvert(pop)->Ptabdesc();
case CLogical::EopLogicalDynamicBitmapTableGet:
return CLogicalDynamicBitmapTableGet::PopConvert(pop)->Ptabdesc();
case CLogical::EopLogicalSelect:
diff --git a/src/backend/gporca/libgpopt/src/xforms/CJoinOrderDPv2.cpp b/src/backend/gporca/libgpopt/src/xforms/CJoinOrderDPv2.cpp
index f2a506e37b..661dba3886 100644
--- a/src/backend/gporca/libgpopt/src/xforms/CJoinOrderDPv2.cpp
+++ b/src/backend/gporca/libgpopt/src/xforms/CJoinOrderDPv2.cpp
@@ -840,7 +840,7 @@ CJoinOrderDPv2::RecursivelyMarkEdgesAsUsed(CExpression *expr)
// - table scan (or tree)
// - predicate
//
-// In this case, we can put the PS over the TS
+// In this case, we can put the PS over the table scan
// We make a few assumptions:
// 1. The benefits of a PS are from the selectivity of a single table, rather than the join result between two tables. We find this table by looking for logical selects.
// 2. The selectivty of this single table is equal to the selectivity of the PS
@@ -849,6 +849,12 @@ CJoinOrderDPv2::RecursivelyMarkEdgesAsUsed(CExpression *expr)
// If it is, we mark this SExpressionInfo as containing a PS
// We only consider linear trees here, since bushy trees would increase the search space and increase the
// chance of motions between the PS and PT, which then would fail requirements during optimization
+//
+// This function includes 4 main steps:
+// 1. Exit early if group contains a partition table or if partition table already has a partition selector
+// 2. Exit if distribution specs are incompitible and motion would be added, which makes partition selector invalid
+// 3. Check whether join expression condition contains the partition key
+// 4. DPE cost adjustment
void
CJoinOrderDPv2::PopulateDPEInfo(SExpressionInfo *join_expr_info,
SGroupInfo *part_table_group_info,
@@ -856,101 +862,156 @@ CJoinOrderDPv2::PopulateDPEInfo(SExpressionInfo *join_expr_info,
{
SGroupInfoArray *atom_groups = GetGroupsForLevel(1);
-
- COptimizerConfig *optimizer_config =
- COptCtxt::PoctxtFromTLS()->GetOptimizerConfig();
- ULONG broadcast_threshold =
- optimizer_config->GetHint()->UlBroadcastThreshold();
-
CBitSetIter iter_pt(*part_table_group_info->m_atoms);
- SGroupInfo *pt_atom = nullptr;
+ CExpression *pt_atom_expr = nullptr;
CPartKeysArray *partition_keys = nullptr;
+ SGroupInfo *pt_atom = nullptr;
+ // iterate through each atom of the "outer" (left) group, and look for a partition table
+ // that does not yet have a partition selector.
+ // Once a partition table is found, check if it has a valid partition selector. I
+ // if no valid partition selectors are found, continue to the next partition table.
while (iter_pt.Advance())
{
pt_atom = (*atom_groups)[iter_pt.Bit()];
+ pt_atom_expr = (*pt_atom->m_best_expr_info_array)[0]->m_expr;
partition_keys =
(*pt_atom->m_best_expr_info_array)[0]->m_atom_part_keys_array;
- if (partition_keys != nullptr && partition_keys->Size() > 0)
+ // continue if there are no partition keys or if the partition table already has an
+ // associated partition selector
+ if (partition_keys == nullptr || partition_keys->Size() == 0 ||
+ join_expr_info->m_contain_PS->Get(iter_pt.Bit()))
{
- break;
+ continue;
}
- }
- if (nullptr != partition_keys)
- {
- GPOS_ASSERT(nullptr != pt_atom);
- GPOS_ASSERT(nullptr != partition_keys && partition_keys->Size() > 0);
+
+ // check if this join order will produce a valid partition selector
+ // this can only occur if the distribution spec matches.
+ // consider the below join tree:
+ // HJ3
+ // HJ2 PS2
+ // PT2 HJ1
+ // PT1 PS1
+ // If there is a motion between PT2 and PS2, then this is not a valid plan for partition selection
+ // Therefore, when considering a table that may produce a partition selector, we need to check the join expression
+ // below the current join.
+
+ // columns of child join condition (HJ2 in example above)
+ CColRefSet *left_child_join_condition_cols = nullptr;
+
+ // only check for a valid partition selector for joins with more than 1 level.
+ // For single level joins, eg: A HJ B, there is no join below A and we can skip this logic
+ // and go straight ahead to check if the join expr overlaps the partition key
+ if (!part_table_group_info->IsAnAtom())
+ {
+ CExpression *left_child_join_expr =
+ join_expr_info->m_left_child_expr.GetExprInfo()->m_expr;
+ left_child_join_condition_cols =
+ (*left_child_join_expr)[left_child_join_expr->Arity() - 1]
+ ->DeriveUsedColumns();
+ // retrieve the table descriptor in order to get the distribution columns of the partition table
+ CLogical *popLogical = CLogical::PopConvert(pt_atom_expr->Pop());
+ CTableDescriptor *atom_table_descriptor =
+ CLogical::PtabdescFromTableGet(popLogical);
+ if (nullptr != atom_table_descriptor &&
+ nullptr != left_child_join_condition_cols)
+ {
+ CColRefArray *atomOutputColArray =
+ pt_atom_expr->DeriveOutputColumns()->Pdrgpcr(m_mp);
+ CColRefSet *pt_atom_distribution_cols = CLogical::PcrsDist(
+ m_mp, atom_table_descriptor, atomOutputColArray);
+ atomOutputColArray->Release();
+ // check that the child join condition contains the partition table's distribution columns
+ // if it doesn't, it won't be a valid partition selector due to the added redistribute
+ if (!left_child_join_condition_cols->ContainsAll(
+ pt_atom_distribution_cols))
+ {
+ pt_atom_distribution_cols->Release();
+ continue;
+ }
+ pt_atom_distribution_cols->Release();
+ }
+ }
+
CExpression *join_expr = join_expr_info->m_expr;
CExpression *scalar_expr = (*join_expr)[join_expr->Arity() - 1];
CColRefSet *join_expr_cols = scalar_expr->DeriveUsedColumns();
+
+ // iterate through each partition key of the partition table (typically of size 1)
for (ULONG i = 0; i < partition_keys->Size(); i++)
{
CPartKeys *part_keys = (*partition_keys)[i];
- // If the join expr overlaps the partition key, then we consider the expression as having a possible PS for that PT
- if (part_keys->FOverlap(join_expr_cols))
+ // If the join expr overlaps the partition key, then we consider
+ // the expression as having a possible PS for that PT, otherwise continue
+ if (!part_keys->FOverlap(join_expr_cols))
{
- CBitSetIter iter_ps(*part_selector_group_info->m_atoms);
+ continue;
+ }
+ CBitSetIter iter_ps(*part_selector_group_info->m_atoms);
+ GPOS_ASSERT(part_selector_group_info->m_atoms->Size() == 1);
+ iter_ps.Advance();
+ // a partition selector atom is a potential partition selector
+ // only if it is a logical select (ie: it has a filter)
+ if ((*(*atom_groups)[iter_ps.Bit()]->m_best_expr_info_array)[0]
+ ->m_expr->Pop()
+ ->Eopid() == COperator::EopLogicalSelect)
+ {
+ // mark join expr as containing a partition selector
join_expr_info->m_contain_PS->ExchangeSet(iter_pt.Bit());
- while (iter_ps.Advance())
- {
- // if the part selector group has a potential PS, ensure that one of the group's atoms is a logical select
- if ((*(*atom_groups)[iter_ps.Bit()]
- ->m_best_expr_info_array)[0]
- ->m_expr->Pop()
- ->Eopid() == COperator::EopLogicalSelect)
- {
- SExpressionInfo *atom_ps =
- (*(*atom_groups)[iter_ps.Bit()]
- ->m_best_expr_info_array)[0];
- // This is a bit simplistic. We calculate how much we are reducing the cardinality of the atom,
- // but also take into account the cost of broadcasting the inner rows. If the number of rows
- // broadcasted is much larger than the savings, then PS will likely not benefit in this case
- // The numbers are from the cost model used during optimization
-
- // for a select(some_non_get_node()) ==> 0.9
- // for a non-select node (won't even come here) ==> 0.0, in effect
- // for a select(get) ==> 1 - (row count of select / row count of get)
-
- CDouble percent_reduction =
- .9; // an arbitary default if the logical operator is not a simple select
- ICostModel *cost_model =
- COptCtxt::PoctxtFromTLS()->GetCostModel();
- CDouble num_segments = cost_model->UlHosts();
- CDouble distribution_cost_factor =
- (num_segments * BCAST_RECV_COST + BCAST_SEND_COST) /
- SEQ_SCAN_COST;
- CDouble broadcast_penalty =
- part_selector_group_info->m_cardinality *
- distribution_cost_factor;
- // penalize broadcast if it exceeds broadcast threshold (specified via GUC), just like
- // in the cost model during optimization
- if (part_selector_group_info->m_cardinality >
- broadcast_threshold)
- {
- broadcast_penalty =
- broadcast_penalty * BROACAST_PENALTY;
- }
-
- if (atom_ps->m_atom_base_table_rows.Get() > 0)
- {
- percent_reduction =
- (1 - (atom_ps->m_cost.Get() /
- atom_ps->m_atom_base_table_rows.Get()));
- }
-
- // Adjust the cost of the expression for each partition selector
- join_expr_info->m_cost_adj_PS =
- join_expr_info->m_cost_adj_PS -
- (percent_reduction * pt_atom->m_cardinality) +
- broadcast_penalty;
- }
- }
- break;
+ SExpressionInfo *atom_ps =
+ (*(*atom_groups)[iter_ps.Bit()]->m_best_expr_info_array)[0];
+
+ // Adjust the cost of the expression for each partition selector
+ join_expr_info->m_cost_adj_PS = CostJoinWithPartitionSelection(
+ join_expr_info, atom_ps, pt_atom, part_selector_group_info);
+ return;
}
}
}
}
+CDouble
+CJoinOrderDPv2::CostJoinWithPartitionSelection(
+ SExpressionInfo *join_expr_info, SExpressionInfo *atom_ps,
+ SGroupInfo *pt_atom, SGroupInfo *part_selector_group_info)
+{
+ COptimizerConfig *optimizer_config =
+ COptCtxt::PoctxtFromTLS()->GetOptimizerConfig();
+ ULONG broadcast_threshold =
+ optimizer_config->GetHint()->UlBroadcastThreshold();
+
+ // This is a bit simplistic. We calculate how much we are reducing the cardinality of the atom,
+ // but also take into account the cost of broadcasting the inner rows. If the number of rows
+ // broadcasted is much larger than the savings, then PS will likely not benefit in this case
+ // The numbers are from the cost model used during optimization
+
+ // for a select(some_non_get_node()) ==> 0.9
+ // for a non-select node (won't even come here) ==> 0.0, in effect
+ // for a select(get) ==> 1 - (row count of select / row count of get)
+ CDouble percent_reduction =
+ .9; // an arbitary default if the logical operator is not a simple select
+ ICostModel *cost_model = COptCtxt::PoctxtFromTLS()->GetCostModel();
+ CDouble num_segments = cost_model->UlHosts();
+ CDouble distribution_cost_factor =
+ (num_segments * BCAST_RECV_COST + BCAST_SEND_COST) / SEQ_SCAN_COST;
+ CDouble broadcast_cost =
+ part_selector_group_info->m_cardinality * distribution_cost_factor;
+
+ // penalize broadcast if it exceeds broadcast threshold (specified via GUC), just like
+ // in the cost model during optimization
+ if (part_selector_group_info->m_cardinality > broadcast_threshold)
+ {
+ broadcast_cost = broadcast_cost * BROACAST_PENALTY;
+ }
+
+ if (atom_ps->m_atom_base_table_rows.Get() > 0)
+ {
+ percent_reduction = (1 - (atom_ps->m_cost.Get() /
+ atom_ps->m_atom_base_table_rows.Get()));
+ }
+ return join_expr_info->m_cost_adj_PS -
+ (percent_reduction * pt_atom->m_cardinality) + broadcast_cost;
+}
//---------------------------------------------------------------------------
// @function:
// CJoinOrderDPv2::SearchJoinOrders
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org