Method rewriteDynamicFilter

src/QueryPlan/TableScanStep.cpp:965–1147

}

void TableScanStep::rewriteDynamicFilter(SelectQueryInfo & select_query, const BuildQueryPipelineSettings & build_settings, bool use_expand_pipe)
{
    if (select_query.partition_filter)
    {
        Stopwatch watch;

        auto [rf_filters, where_predicates] = RuntimeFilterUtils::extractRuntimeFilters(select_query.partition_filter);
        std::vector<RuntimeFilterDescription> descriptions;
        descriptions.reserve(rf_filters.size());
        for (const auto & predicate : rf_filters)
            descriptions.emplace_back(RuntimeFilterUtils::extractDescription(predicate).value());

        const auto & query_id = build_settings.distributed_settings.query_id;
        const auto & setting = build_settings.context->getSettingsRef();
        size_t wait_ms = use_expand_pipe ? 0 : setting.wait_runtime_filter_timeout;
        bool enable_bf_in_prewhere = setting.enable_rewrite_bf_into_prewhere;
        bool enable_range_cover = setting.enable_range_cover;
        for (auto & description : descriptions)
        {
            bool is_range_or_set = false;
            bool has_bf = false;
            auto runtime_filters = RuntimeFilterUtils::createRuntimeFilterForTableScan(
                description, query_id, wait_ms, enable_bf_in_prewhere, enable_range_cover, is_range_or_set, has_bf);
            where_predicates.insert(where_predicates.end(), runtime_filters.begin(), runtime_filters.end());
        }
        auto where_dicates = PredicateUtils::combineConjuncts(where_predicates);
        select_query.partition_filter = !PredicateUtils::isTruePredicate(where_dicates) ? std::move(where_dicates) : nullptr;

        LOG_DEBUG(
            log,
            "rewrite partition runtime filter done, cost:{} ms, query:{}",
            watch.elapsedMilliseconds(),
            select_query.partition_filter ? queryToString(*select_query.partition_filter) : "null");
    }

    auto * query = select_query.query->as<ASTSelectQuery>();
    auto where = query->getWhere();
    auto prehwere = query->getPrewhere();
    if (where || prehwere)
    {
        auto [rf_filters, where_predicates] = RuntimeFilterUtils::extractRuntimeFilters(where);
        auto [prewher_rf_filters, prewhere_predicates] = RuntimeFilterUtils::extractRuntimeFilters(prehwere);
        if (rf_filters.empty() && prewher_rf_filters.empty())
            return;
        rf_filters.insert(rf_filters.end(), prewher_rf_filters.begin(), prewher_rf_filters.end());

        std::vector<ConstASTPtr> tmp_prewhere_predicates;
        std::vector<RuntimeFilterDescription> descriptions;
        descriptions.reserve(rf_filters.size());
        for (const auto & predicate : rf_filters)
            descriptions.emplace_back(RuntimeFilterUtils::extractDescription(predicate).value());

        if (descriptions.empty())
            return ;

        std::sort(descriptions.begin(), descriptions.end(), [](const auto & lhs, const auto & rhs) {
            return lhs.filter_factor > rhs.filter_factor;
        });