From 7ffa86657d4e745b16098194129c091f38a345a4 Mon Sep 17 00:00:00 2001 From: obdev Date: Wed, 26 Oct 2022 11:40:48 +0000 Subject: [PATCH] fix partial partition wise merge join bug --- src/sql/optimizer/ob_join_order.cpp | 18 +++++++-------- src/sql/optimizer/ob_join_order.h | 8 +++---- src/sql/optimizer/ob_logical_operator.cpp | 27 ++++++++++++++--------- src/sql/optimizer/ob_logical_operator.h | 8 +++---- src/sql/optimizer/ob_select_log_plan.cpp | 7 ++---- 5 files changed, 34 insertions(+), 34 deletions(-) diff --git a/src/sql/optimizer/ob_join_order.cpp b/src/sql/optimizer/ob_join_order.cpp index 4b03dd027..a03b8e5f7 100644 --- a/src/sql/optimizer/ob_join_order.cpp +++ b/src/sql/optimizer/ob_join_order.cpp @@ -3719,7 +3719,7 @@ int oceanbase::sql::Path::assign(const Path &other, common::ObIAllocator *alloca phy_plan_type_ = other.phy_plan_type_; location_type_ = other.location_type_; contain_fake_cte_ = other.contain_fake_cte_; - contain_merge_op_ = other.contain_merge_op_; + contain_pw_merge_op_ = other.contain_pw_merge_op_; contain_das_op_ = other.contain_das_op_; parallel_ = other.parallel_; server_cnt_ = other.server_cnt_; @@ -4826,8 +4826,9 @@ int JoinPath::compute_join_path_info() LOG_WARN("get unexpected null", K(left_path_), K(right_path_), K(ret)); } else { contain_fake_cte_ = left_path_->contain_fake_cte_ || right_path_->contain_fake_cte_; - contain_merge_op_ = left_path_->contain_merge_op_ || right_path_->contain_merge_op_ || - (join_algo_ == JoinAlgo::MERGE_JOIN); + contain_pw_merge_op_ = (left_path_->contain_pw_merge_op_ && !is_left_need_exchange()) || + (right_path_->contain_pw_merge_op_ && !is_right_need_exchange()) || + (join_algo_ == JoinAlgo::MERGE_JOIN && is_partition_wise()); contain_das_op_ = left_path_->contain_das_op_ || right_path_->contain_das_op_; } return ret; @@ -5367,7 +5368,7 @@ void JoinPath::reuse() phy_plan_type_ = ObPhyPlanType::OB_PHY_PLAN_UNINITIALIZED; location_type_ = ObPhyPlanType::OB_PHY_PLAN_UNINITIALIZED; contain_fake_cte_ = false; - contain_merge_op_ = false; + contain_pw_merge_op_ = false; contain_das_op_ = false; parallel_ = 1; server_cnt_ = 1; @@ -6075,7 +6076,7 @@ int ObJoinOrder::compute_subquery_path_property(const uint64_t table_id, path->phy_plan_type_ = root->get_phy_plan_type(); path->location_type_ = root->get_location_type(); path->contain_fake_cte_ = root->get_contains_fake_cte(); - path->contain_merge_op_ = root->get_contains_merge_op(); + path->contain_pw_merge_op_ = root->get_contains_pw_merge_op(); path->contain_das_op_ = root->get_contains_das_op(); path->parallel_ = root->get_parallel(); path->server_cnt_ = root->get_server_cnt(); @@ -7309,11 +7310,8 @@ bool ObJoinOrder::is_partition_wise_valid(const Path &left_path, const Path &right_path) { bool is_valid = true; - if (left_path.exchange_allocated_ && !right_path.exchange_allocated_ && - right_path.contain_merge_op()) { - is_valid = false; - } else if (!left_path.exchange_allocated_ && right_path.exchange_allocated_ && - left_path.contain_merge_op()) { + if ((left_path.exchange_allocated_ || right_path.exchange_allocated_) && + (left_path.contain_pw_merge_op() || right_path.contain_pw_merge_op())) { is_valid = false; } else { is_valid = true; diff --git a/src/sql/optimizer/ob_join_order.h b/src/sql/optimizer/ob_join_order.h index 3d0d99ec9..11931021c 100644 --- a/src/sql/optimizer/ob_join_order.h +++ b/src/sql/optimizer/ob_join_order.h @@ -310,7 +310,7 @@ struct EstimateCostInfo { phy_plan_type_(ObPhyPlanType::OB_PHY_PLAN_UNINITIALIZED), location_type_(ObPhyPlanType::OB_PHY_PLAN_UNINITIALIZED), contain_fake_cte_(false), - contain_merge_op_(false), + contain_pw_merge_op_(false), contain_das_op_(false), parallel_(1), server_cnt_(1) @@ -335,7 +335,7 @@ struct EstimateCostInfo { phy_plan_type_(ObPhyPlanType::OB_PHY_PLAN_UNINITIALIZED), location_type_(ObPhyPlanType::OB_PHY_PLAN_UNINITIALIZED), contain_fake_cte_(false), - contain_merge_op_(false), + contain_pw_merge_op_(false), contain_das_op_(false), parallel_(1), server_cnt_(1), @@ -409,7 +409,7 @@ struct EstimateCostInfo { virtual int re_estimate_cost(EstimateCostInfo &info, double &card, double &cost); double get_path_output_rows() const; bool contain_fake_cte() const { return contain_fake_cte_; } - bool contain_merge_op() const { return contain_merge_op_; } + bool contain_pw_merge_op() const { return contain_pw_merge_op_; } bool is_pipelined_path() const { return is_pipelined_path_; } bool is_nl_style_pipelined_path() const { return is_nl_style_pipelined_path_; } virtual int compute_pipeline_info(); @@ -480,7 +480,7 @@ struct EstimateCostInfo { ObPhyPlanType phy_plan_type_; ObPhyPlanType location_type_; bool contain_fake_cte_; - bool contain_merge_op_; + bool contain_pw_merge_op_; bool contain_das_op_; // remember the parallel info to get this sharding int64_t parallel_; diff --git a/src/sql/optimizer/ob_logical_operator.cpp b/src/sql/optimizer/ob_logical_operator.cpp index e943d8422..218ec1ba7 100644 --- a/src/sql/optimizer/ob_logical_operator.cpp +++ b/src/sql/optimizer/ob_logical_operator.cpp @@ -344,7 +344,7 @@ ObLogicalOperator::ObLogicalOperator(ObLogPlan &plan) dblink_id_(0), // 0 represent local cluster. plan_depth_(0), contain_fake_cte_(false), - contain_merge_op_(false), + contain_pw_merge_op_(false), contain_das_op_(false), dup_table_pos_(), strong_sharding_(NULL), @@ -802,27 +802,32 @@ int ObLogicalOperator::compute_op_other_info() // compute contains merge style op if (OB_SUCC(ret)) { - for (int64_t i = 0; OB_SUCC(ret) && !contain_merge_op_ && i < get_num_of_child(); i++) { + for (int64_t i = 0; OB_SUCC(ret) && !contain_pw_merge_op_ && i < get_num_of_child(); i++) { if (OB_ISNULL(get_child(i))) { ret = OB_ERR_UNEXPECTED; LOG_WARN("get unexpected null", K(ret)); } else { - contain_merge_op_ |= get_child(i)->get_contains_merge_op(); + contain_pw_merge_op_ |= get_child(i)->get_contains_pw_merge_op() && + log_op_def::LOG_EXCHANGE != get_child(i)->get_type(); } } - if (OB_SUCC(ret) && !contain_merge_op_) { + if (OB_SUCC(ret) && !contain_pw_merge_op_) { if (log_op_def::LOG_GROUP_BY == get_type()) { ObLogGroupBy *group_by = static_cast(this); - contain_merge_op_ = !group_by->get_group_by_exprs().empty() && - (AggregateAlgo::MERGE_AGGREGATE == group_by->get_algo()); + contain_pw_merge_op_ = !group_by->get_group_by_exprs().empty() && + (AggregateAlgo::MERGE_AGGREGATE == group_by->get_algo()) && + is_partition_wise(); } else if (log_op_def::LOG_DISTINCT == get_type()) { ObLogDistinct *distinct = static_cast(this); - contain_merge_op_ = AggregateAlgo::MERGE_AGGREGATE == distinct->get_algo(); + contain_pw_merge_op_ = AggregateAlgo::MERGE_AGGREGATE == distinct->get_algo() && + is_partition_wise(); } else if (log_op_def::LOG_SET == get_type()) { ObLogSet *set = static_cast(this); - contain_merge_op_ = set->is_set_distinct() && SetAlgo::MERGE_SET == set->get_algo(); + contain_pw_merge_op_ = set->is_set_distinct() && SetAlgo::MERGE_SET == set->get_algo() && + is_partition_wise(); } else if (log_op_def::LOG_WINDOW_FUNCTION == get_type()) { - contain_merge_op_ = is_block_op(); + contain_pw_merge_op_ = is_block_op() && + is_partition_wise(); } else { /*do nothing*/ } } } @@ -885,7 +890,7 @@ int ObLogicalOperator::compute_property(Path *path) set_phy_plan_type(path->phy_plan_type_); set_location_type(path->location_type_); set_contains_fake_cte(path->contain_fake_cte_); - set_contains_merge_op(path->contain_merge_op_); + set_contains_pw_merge_op(path->contain_pw_merge_op_); set_contains_das_op(path->contain_das_op_); is_pipelined_plan_ = path->is_pipelined_path(); is_nl_style_pipelined_plan_ = path->is_nl_style_pipelined_path(); @@ -1013,7 +1018,7 @@ int ObLogicalOperator::compute_property() K(phy_plan_type_), K(location_type_), K(contain_fake_cte_), - K(contain_merge_op_), + K(contain_pw_merge_op_), K(contain_das_op_), K(width_)); } diff --git a/src/sql/optimizer/ob_logical_operator.h b/src/sql/optimizer/ob_logical_operator.h index 05d283aaa..25ca8b7f6 100644 --- a/src/sql/optimizer/ob_logical_operator.h +++ b/src/sql/optimizer/ob_logical_operator.h @@ -1497,10 +1497,10 @@ public: { contain_fake_cte_ = contain_fake_cte; } - inline bool get_contains_merge_op() const { return contain_merge_op_; } - inline void set_contains_merge_op(bool contain_merge_op) + inline bool get_contains_pw_merge_op() const { return contain_pw_merge_op_; } + inline void set_contains_pw_merge_op(bool contain_pw_merge_op) { - contain_merge_op_ = contain_merge_op; + contain_pw_merge_op_ = contain_pw_merge_op; } inline bool get_contains_das_op() const { return contain_das_op_; } inline void set_contains_das_op(bool contain_das_op) @@ -1820,7 +1820,7 @@ protected: uint64_t dblink_id_; int64_t plan_depth_; bool contain_fake_cte_; - bool contain_merge_op_; + bool contain_pw_merge_op_; bool contain_das_op_; common::ObSEArray dup_table_pos_; // for duplicated table ObShardingInfo *strong_sharding_; diff --git a/src/sql/optimizer/ob_select_log_plan.cpp b/src/sql/optimizer/ob_select_log_plan.cpp index 0956bbe6b..d619a1185 100644 --- a/src/sql/optimizer/ob_select_log_plan.cpp +++ b/src/sql/optimizer/ob_select_log_plan.cpp @@ -2622,11 +2622,8 @@ bool ObSelectLogPlan::is_set_partition_wise_valid(const ObLogicalOperator &left_ const ObLogicalOperator &right_plan) { bool is_valid = true; - if (left_plan.is_exchange_allocated() && !right_plan.is_exchange_allocated() && - right_plan.get_contains_merge_op()) { - is_valid = false; - } else if (!left_plan.is_exchange_allocated() && - right_plan.is_exchange_allocated() && left_plan.get_contains_merge_op()) { + if ((left_plan.is_exchange_allocated() || right_plan.is_exchange_allocated()) && + (left_plan.get_contains_pw_merge_op() || right_plan.get_contains_pw_merge_op())) { is_valid = false; } else { /*do nothing*/ } return is_valid;