| 1749 | } |
| 1750 | |
| 1751 | void tryMakeDirectJoinWithMergeTree(const JoinOperator & join_operator, |
| 1752 | QueryPlan & right_query_plan, |
| 1753 | PreparedJoinStorage & prepared_join, |
| 1754 | PlannerContextPtr & planner_context) |
| 1755 | { |
| 1756 | const auto & query_context = planner_context->getQueryContext(); |
| 1757 | const auto & settings = query_context->getSettingsRef(); |
| 1758 | |
| 1759 | /// In chooseJoinAlgorithm, direct has the highest priority (automatically used with dictionary or storage join). |
| 1760 | /// Use direct join with MergeTree only if 'direct' is explicitly specified as the single option. |
| 1761 | if (settings[Setting::join_algorithm].value != std::vector{JoinAlgorithm::DIRECT}) |
| 1762 | return; |
| 1763 | |
| 1764 | bool allow_strictness = join_operator.strictness == JoinStrictness::All || |
| 1765 | join_operator.strictness == JoinStrictness::Semi || |
| 1766 | join_operator.strictness == JoinStrictness::Anti; |
| 1767 | |
| 1768 | if (!allow_strictness || !isInnerOrLeft(join_operator.kind)) |
| 1769 | return; |
| 1770 | if (!join_operator.residual_filter.empty() || join_operator.expression.size() != 1) |
| 1771 | return; |
| 1772 | auto [predicate_type, lhs, rhs] = join_operator.expression[0].asBinaryPredicate(); |
| 1773 | if (predicate_type != JoinConditionOperator::Equals) |
| 1774 | return; |
| 1775 | |
| 1776 | /// Check that right plan is ReadFromMergeTree with ExpressionStep/FilterStep on the top |
| 1777 | auto * root_node = right_query_plan.getRootNode(); |
| 1778 | if (!root_node || !root_node->step) |
| 1779 | return; |
| 1780 | const auto * expr_step = root_node->step.get(); |
| 1781 | if (!typeid_cast<const ExpressionStep *>(expr_step) && !typeid_cast<const FilterStep *>(expr_step)) |
| 1782 | return; |
| 1783 | if (root_node->children.size() != 1 || !root_node->children.front()) |
| 1784 | return; |
| 1785 | |
| 1786 | const auto * children_step = root_node->children.front()->step.get(); |
| 1787 | bool is_allowed_storage = typeid_cast<const ReadFromMergeTree *>(children_step) |
| 1788 | || typeid_cast<const ReadNothingStep *>(children_step) |
| 1789 | || typeid_cast<const ReadFromPreparedSource *>(children_step); |
| 1790 | if (!is_allowed_storage) |
| 1791 | return; |
| 1792 | |
| 1793 | if (lhs.fromRight() && rhs.fromLeft()) |
| 1794 | std::swap(lhs, rhs); |
| 1795 | else if (!lhs.fromLeft() || !rhs.fromRight()) |
| 1796 | return; |
| 1797 | |
| 1798 | auto lookup_plan = right_query_plan.clone(); |
| 1799 | auto & lookup_read_step = lookup_plan.getRootNode()->children.front()->step; |
| 1800 | if (auto * lookup_reading_step = typeid_cast<ReadFromMergeTree *>(lookup_read_step.get())) |
| 1801 | { |
| 1802 | /// We need to analyze index again with new condition |
| 1803 | lookup_reading_step->setAnalyzedResult(nullptr); |
| 1804 | /// Hand-constructed filter dag has same hash key each time, so disable cache |
| 1805 | lookup_reading_step->disableQueryConditionCache(); |
| 1806 | /// initializePipeline is done multiple times concurrently, so not to remove parts snapshot |
| 1807 | lookup_reading_step->disableMergeTreePartsSnapshotRemoval(); |
| 1808 | } |
no test coverage detected