/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
#include "JoinRelParser.h"
#include <optional>
#include <Core/Block.h>
#include <Core/Settings.h>
#include <Functions/FunctionFactory.h>
#include <Interpreters/CollectJoinOnKeysVisitor.h>
#include <Interpreters/ExpressionActions.h>
#include <Interpreters/FullSortingMergeJoin.h>
#include <Interpreters/GraceHashJoin.h>
#include <Interpreters/HashJoin/HashJoin.h>
#include <Interpreters/TableJoin.h>
#include <Join/BroadCastJoinBuilder.h>
#include <Join/StorageJoinFromReadBuffer.h>
#include <Operator/EarlyStopStep.h>
#include <Parser/AdvancedParametersParseUtil.h>
#include <Parser/ExpressionParser.h>
#include <Parser/SubstraitParserUtils.h>
#include <Parsers/ASTIdentifier.h>
#include <Processors/QueryPlan/ExpressionStep.h>
#include <Processors/QueryPlan/FilterStep.h>
#include <Processors/QueryPlan/JoinStep.h>
#include <google/protobuf/wrappers.pb.h>
#include <Common/CHUtil.h>
#include <Common/GlutenConfig.h>
#include <Common/logger_useful.h>

namespace DB
{
namespace Setting
{
extern const SettingsJoinAlgorithm join_algorithm;
extern const SettingsUInt64 max_block_size;
extern const SettingsUInt64 min_joined_block_size_rows;
extern const SettingsUInt64 min_joined_block_size_bytes;
extern const SettingsNonZeroUInt64 grace_hash_join_initial_buckets;
extern const SettingsNonZeroUInt64 grace_hash_join_max_buckets;
}
namespace ErrorCodes
{
extern const int LOGICAL_ERROR;
extern const int UNKNOWN_TYPE;
extern const int BAD_ARGUMENTS;
}
}
using namespace DB;

namespace local_engine
{
std::shared_ptr<DB::TableJoin> createDefaultTableJoin(substrait::JoinRel_JoinType join_type, const JoinOptimizationInfo & join_opt_info, ContextPtr & context)
{
    auto table_join
        = std::make_shared<TableJoin>(context->getSettingsRef(), context->getGlobalTemporaryVolume(), context->getTempDataOnDisk());

    std::pair<DB::JoinKind, DB::JoinStrictness> kind_and_strictness = JoinUtil::getJoinKindAndStrictness(join_type, join_opt_info.is_existence_join);
    table_join->setKind(kind_and_strictness.first);
    if (!join_opt_info.is_any_join)
        table_join->setStrictness(kind_and_strictness.second);
    else
        table_join->setStrictness(DB::JoinStrictness::Any);
    return table_join;
}

JoinRelParser::JoinRelParser(ParserContextPtr parser_context_) : RelParser(parser_context_), context(parser_context_->queryContext())
{
}

DB::QueryPlanPtr
JoinRelParser::parse(DB::QueryPlanPtr /*query_plan*/, const substrait::Rel & /*rel*/, std::list<const substrait::Rel *> & /*rel_stack_*/)
{
    throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call parse().");
}

std::vector<const substrait::Rel *> JoinRelParser::getInputs(const substrait::Rel & rel)
{
    const auto & join = rel.join();
    if (!join.has_left() || !join.has_right())
        throw Exception(ErrorCodes::BAD_ARGUMENTS, "left table or right table is missing.");

    return {&join.left(), &join.right()};
}
std::optional<const substrait::Rel *> JoinRelParser::getSingleInput(const substrait::Rel & /*rel*/)
{
    throw Exception(ErrorCodes::LOGICAL_ERROR, "join node has 2 inputs, can't call getSingleInput().");
}

DB::QueryPlanPtr JoinRelParser::parse(
    std::vector<DB::QueryPlanPtr> & input_plans_, const substrait::Rel & rel, std::list<const substrait::Rel *> & rel_stack_)
{
    assert(input_plans_.size() == 2);
    const auto & join = rel.join();
    return parseJoin(join, std::move(input_plans_[0]), std::move(input_plans_[1]));
}

std::unordered_set<DB::JoinTableSide> JoinRelParser::extractTableSidesFromExpression(
    const substrait::Expression & expr, const DB::Block & left_header, const DB::Block & right_header)
{
    std::unordered_set<DB::JoinTableSide> table_sides;
    if (expr.has_scalar_function())
    {
        for (const auto & arg : expr.scalar_function().arguments())
        {
            auto table_sides_from_arg = extractTableSidesFromExpression(arg.value(), left_header, right_header);
            table_sides.insert(table_sides_from_arg.begin(), table_sides_from_arg.end());
        }
    }
    else if (auto field = SubstraitParserUtils::getStructFieldIndex(expr))
    {
        if (*field < left_header.columns())
            table_sides.insert(DB::JoinTableSide::Left);
        else
            table_sides.insert(DB::JoinTableSide::Right);
    }
    else if (expr.has_singular_or_list())
    {
        auto child_table_sides = extractTableSidesFromExpression(expr.singular_or_list().value(), left_header, right_header);
        table_sides.insert(child_table_sides.begin(), child_table_sides.end());
        for (const auto & option : expr.singular_or_list().options())
        {
            child_table_sides = extractTableSidesFromExpression(option, left_header, right_header);
            table_sides.insert(child_table_sides.begin(), child_table_sides.end());
        }
    }
    else if (expr.has_cast())
    {
        auto child_table_sides = extractTableSidesFromExpression(expr.cast().input(), left_header, right_header);
        table_sides.insert(child_table_sides.begin(), child_table_sides.end());
    }
    else if (expr.has_if_then())
    {
        for (const auto & if_child : expr.if_then().ifs())
        {
            auto child_table_sides = extractTableSidesFromExpression(if_child.if_(), left_header, right_header);
            table_sides.insert(child_table_sides.begin(), child_table_sides.end());
            child_table_sides = extractTableSidesFromExpression(if_child.then(), left_header, right_header);
            table_sides.insert(child_table_sides.begin(), child_table_sides.end());
        }
        auto child_table_sides = extractTableSidesFromExpression(expr.if_then().else_(), left_header, right_header);
        table_sides.insert(child_table_sides.begin(), child_table_sides.end());
    }
    else if (expr.has_literal())
    {
        // nothing
    }
    else
    {
        throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Illegal expression '{}'", expr.DebugString());
    }
    return table_sides;
}


void JoinRelParser::renamePlanColumns(DB::QueryPlan & left, DB::QueryPlan & right, const StorageJoinFromReadBuffer & storage_join)
{
    /// To support mixed join conditions, we must make sure that the column names in the right be the same as
    /// storage_join's right sample block.
    ActionsDAG right_project = ActionsDAG::makeConvertingActions(
        right.getCurrentHeader()->getColumnsWithTypeAndName(),
        storage_join.getRightSampleBlock().getColumnsWithTypeAndName(),
        ActionsDAG::MatchColumnsMode::Position);

    QueryPlanStepPtr right_project_step = std::make_unique<ExpressionStep>(right.getCurrentHeader(), std::move(right_project));
    right_project_step->setStepDescription("Rename Broadcast Table Name");
    steps.emplace_back(right_project_step.get());
    right.addStep(std::move(right_project_step));

    /// If the columns name in right table is duplicated with left table, we need to rename the left table's columns,
    /// avoid the columns name in the right table be changed in `addConvertStep`.
    /// This could happen in tpc-ds q44.
    DB::ColumnsWithTypeAndName new_left_cols;
    const auto & right_header = *right.getCurrentHeader();
    auto left_prefix = getUniqueName("left");
    for (const auto & col : *left.getCurrentHeader())
        if (right_header.has(col.name))
            new_left_cols.emplace_back(col.column, col.type, left_prefix + col.name);
        else
            new_left_cols.emplace_back(col.column, col.type, col.name);
    ActionsDAG left_project = ActionsDAG::makeConvertingActions(
        left.getCurrentHeader()->getColumnsWithTypeAndName(), new_left_cols, ActionsDAG::MatchColumnsMode::Position);

    QueryPlanStepPtr left_project_step = std::make_unique<ExpressionStep>(left.getCurrentHeader(), std::move(left_project));
    left_project_step->setStepDescription("Rename Left Table Name for broadcast join");
    steps.emplace_back(left_project_step.get());
    left.addStep(std::move(left_project_step));
}

DB::QueryPlanPtr JoinRelParser::parseJoin(const substrait::JoinRel & join, DB::QueryPlanPtr left, DB::QueryPlanPtr right)
{
    auto join_config = JoinConfig::loadFromContext(getContext());
    google::protobuf::StringValue optimization_info;
    optimization_info.ParseFromString(join.advanced_extension().optimization().value());
    auto join_opt_info = JoinOptimizationInfo::parse(optimization_info.value());
    LOG_DEBUG(getLogger("JoinRelParser"), "optimization info:{}", optimization_info.value());
    auto storage_join = join_opt_info.is_broadcast ? BroadCastJoinBuilder::getJoin(join_opt_info.storage_join_key) : nullptr;
    if (storage_join)
        renamePlanColumns(*left, *right, *storage_join);

    auto table_join = createDefaultTableJoin(join.type(), join_opt_info, context);
    DB::Block right_header_before_convert_step{*right->getCurrentHeader()};
    addConvertStep(*table_join, *left, *right);

    // Add a check to find error easily.
    if (storage_join)
    {
        bool is_col_names_changed = false;
        const auto & current_right_header = *right->getCurrentHeader();
        if (right_header_before_convert_step.columns() != current_right_header.columns())
            is_col_names_changed = true;
        if (!is_col_names_changed)
        {
            for (size_t i = 0; i < right_header_before_convert_step.columns(); i++)
            {
                if (right_header_before_convert_step.getByPosition(i).name != current_right_header.getByPosition(i).name)
                {
                    is_col_names_changed = true;
                    break;
                }
            }
        }
        if (is_col_names_changed)
        {
            throw DB::Exception(
                DB::ErrorCodes::LOGICAL_ERROR,
                "For broadcast join, we must not change the columns name in the right table.\nleft header:{},\nright header: {} -> {}",
                left->getCurrentHeader()->dumpStructure(),
                right_header_before_convert_step.dumpStructure(),
                right->getCurrentHeader()->dumpStructure());
        }
    }

    Names after_join_names;
    auto left_names = left->getCurrentHeader()->getNames();
    after_join_names.insert(after_join_names.end(), left_names.begin(), left_names.end());
    auto right_name = table_join->columnsFromJoinedTable().getNames();
    after_join_names.insert(after_join_names.end(), right_name.begin(), right_name.end());

    const auto & left_header = *left->getCurrentHeader();
    const auto & right_header = *right->getCurrentHeader();

    QueryPlanPtr query_plan;

    /// some examples to explain when the post_join_filter is not empty
    /// - on t1.key = t2.key and t1.v1 > 1 and t2.v1 > 1, 't1.v1> 1' is in the  post filter. but 't2.v1 > 1'
    ///   will be pushed down into right table by spark and is not in the post filter. 't1.key = t2.key ' is
    ///   in JoinRel::expression.
    /// - on t1.key = t2. key and t1.v1 > t2.v2, 't1.v1 > t2.v2' is in the post filter.
    collectJoinKeys(*table_join, join, left_header, right_header);

    if (storage_join)
    {
        if (join_opt_info.is_null_aware_anti_join && join.type() == substrait::JoinRel_JoinType_JOIN_TYPE_LEFT_ANTI)
        {
            if (storage_join->has_null_key_value)
            {
                // if there is a null key value on the build side, it will return the empty result
                auto empty_step = std::make_unique<EarlyStopStep>(left->getCurrentHeader());
                left->addStep(std::move(empty_step));
            }
            else if (!storage_join->is_empty_hash_table)
            {
                auto input_header = *left->getCurrentHeader();
                DB::ActionsDAG filter_is_not_null_dag{input_header.getColumnsWithTypeAndName()};
                // when is_null_aware_anti_join is true, there is only one join key
                auto field_index = SubstraitParserUtils::getStructFieldIndex(join.expression().scalar_function().arguments(0).value());
                if (!field_index)
                    throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "The join key is not found in the expression.");
                const auto * key_field = filter_is_not_null_dag.getInputs()[*field_index];

                auto result_node = filter_is_not_null_dag.tryFindInOutputs(key_field->result_name);
                // add a function isNotNull to filter the null key on the left side
                const auto * cond_node = buildFunctionNode(filter_is_not_null_dag, "isNotNull", {result_node});
                filter_is_not_null_dag.addOrReplaceInOutputs(*cond_node);
                auto filter_step = std::make_unique<FilterStep>(
                    left->getCurrentHeader(), std::move(filter_is_not_null_dag), cond_node->result_name, true);
                left->addStep(std::move(filter_step));
            }
            // other case: is_empty_hash_table, don't need to handle
        }
        applyJoinFilter(*table_join, join, *left, *right, true);
        auto broadcast_hash_join = storage_join->getJoinLocked(table_join, context);

        QueryPlanStepPtr join_step = std::make_unique<FilledJoinStep>(left->getCurrentHeader(), broadcast_hash_join, 8192);

        join_step->setStepDescription("STORAGE_JOIN");
        steps.emplace_back(join_step.get());
        left->addStep(std::move(join_step));
        query_plan = std::move(left);
        /// hold right plan for profile
        extra_plan_holder.emplace_back(std::move(right));
    }
    else if (join_opt_info.is_smj)
    {
        bool need_post_filter = !applyJoinFilter(*table_join, join, *left, *right, false);

        /// If applyJoinFilter returns false, it means there are mixed conditions in the post_join_filter.
        /// It should be a inner join.
        /// TODO: make smj support mixed conditions
        if (need_post_filter && table_join->kind() != DB::JoinKind::Inner)
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Sort merge join doesn't support mixed join conditions, except inner join.");

        SharedHeader rigth_sample_block = right->getCurrentHeader();
        JoinPtr smj_join = std::make_shared<FullSortingMergeJoin>(table_join, rigth_sample_block, -1);
        MultiEnum<DB::JoinAlgorithm> join_algorithm = context->getSettingsRef()[Setting::join_algorithm];
        QueryPlanStepPtr join_step = std::make_unique<DB::JoinStep>(
            left->getCurrentHeader(),
            right->getCurrentHeader(),
            smj_join,
            context->getSettingsRef()[Setting::max_block_size],
            context->getSettingsRef()[Setting::min_joined_block_size_rows],
            context->getSettingsRef()[Setting::min_joined_block_size_bytes],
            1,
            /* required_output_ = */ NameSet{},
            false,
            /* use_new_analyzer_ = */ false);

        join_step->setStepDescription("SORT_MERGE_JOIN");
        steps.emplace_back(join_step.get());
        std::vector<QueryPlanPtr> plans;
        plans.emplace_back(std::move(left));
        plans.emplace_back(std::move(right));

        query_plan = std::make_unique<QueryPlan>();
        query_plan->unitePlans(std::move(join_step), {std::move(plans)});
        if (need_post_filter)
            addPostFilter(*query_plan, join);
    }
    else
    {
        std::vector<DB::TableJoin::JoinOnClause> join_on_clauses;
        if (table_join->getClauses().empty())
            table_join->addDisjunct();
        bool is_multi_join_on_clauses
            = couldRewriteToMultiJoinOnClauses(table_join->getOnlyClause(), join_on_clauses, join, left_header, right_header);
        if (is_multi_join_on_clauses && join_config.prefer_multi_join_on_clauses && join_opt_info.right_table_rows > 0
            && join_opt_info.partitions_num > 0
            && join_opt_info.right_table_rows / join_opt_info.partitions_num < join_config.multi_join_on_clauses_build_side_rows_limit)
        {
            query_plan = buildMultiOnClauseHashJoin(table_join, std::move(left), std::move(right), join_on_clauses);
        }
        else
        {
            query_plan = buildSingleOnClauseHashJoin(join, table_join, std::move(left), std::move(right));
        }
    }

    JoinUtil::adjustJoinOutput(*query_plan, after_join_names);
    /// Need to project the right table column into boolean type
    if (join_opt_info.is_existence_join)
        existenceJoinPostProject(*query_plan, left_names);

    return query_plan;
}


/// We use left any join to implement ExistenceJoin.
/// The result columns of ExistenceJoin are left table columns + one flag column.
/// The flag column indicates whether a left row is matched or not. We build the flag column here.
/// The input plan's header is left table columns + right table columns. If one row in the right row is null,
/// we mark the flag 0, otherwise mark it 1.
void JoinRelParser::existenceJoinPostProject(DB::QueryPlan & plan, const DB::Names & left_input_cols)
{
    DB::ActionsDAG actions_dag{plan.getCurrentHeader()->getColumnsWithTypeAndName()};
    const auto * right_col_node = actions_dag.getInputs().back();
    auto function_builder = DB::FunctionFactory::instance().get("isNotNull", getContext());
    const auto * not_null_node = &actions_dag.addFunction(function_builder, {right_col_node}, right_col_node->result_name);
    actions_dag.addOrReplaceInOutputs(*not_null_node);
    DB::Names required_cols = left_input_cols;
    required_cols.emplace_back(not_null_node->result_name);
    actions_dag.removeUnusedActions(required_cols);
    auto project_step = std::make_unique<DB::ExpressionStep>(plan.getCurrentHeader(), std::move(actions_dag));
    project_step->setStepDescription("ExistenceJoin Post Project");
    steps.emplace_back(project_step.get());
    plan.addStep(std::move(project_step));
}

void JoinRelParser::addConvertStep(TableJoin & table_join, DB::QueryPlan & left, DB::QueryPlan & right)
{
    /// If the columns name in right table is duplicated with left table, we need to rename the right table's columns.
    NameSet left_columns_set;
    for (const auto & col : left.getCurrentHeader()->getNames())
        left_columns_set.emplace(col);
    table_join.setColumnsFromJoinedTable(
        right.getCurrentHeader()->getNamesAndTypesList(),
        left_columns_set,
        getUniqueName("right") + ".",
        left.getCurrentHeader()->getNamesAndTypesList());

    // fix right table key duplicate
    NamesWithAliases right_table_alias;
    for (size_t idx = 0; idx < table_join.columnsFromJoinedTable().size(); idx++)
    {
        auto origin_name = right.getCurrentHeader()->getByPosition(idx).name;
        auto dedup_name = table_join.columnsFromJoinedTable().getNames().at(idx);
        if (origin_name != dedup_name)
            right_table_alias.emplace_back(NameWithAlias(origin_name, dedup_name));
    }
    if (!right_table_alias.empty())
    {
        ActionsDAG rename_dag{right.getCurrentHeader()->getNamesAndTypesList()};
        const auto & original_right_columns = *right.getCurrentHeader();
        for (const auto & column_alias : right_table_alias)
        {
            if (original_right_columns.has(column_alias.first))
            {
                auto pos = original_right_columns.getPositionByName(column_alias.first);
                const auto & alias = rename_dag.addAlias(*rename_dag.getInputs()[pos], column_alias.second);
                rename_dag.getOutputs()[pos] = &alias;
            }
        }

        QueryPlanStepPtr project_step = std::make_unique<ExpressionStep>(right.getCurrentHeader(), std::move(rename_dag));
        project_step->setStepDescription("Right Table Rename");
        steps.emplace_back(project_step.get());
        right.addStep(std::move(project_step));
    }

    for (const auto & column : table_join.columnsFromJoinedTable())
        table_join.addJoinedColumn(column);
    std::optional<ActionsDAG> left_convert_actions;
    std::optional<ActionsDAG> right_convert_actions;
    std::tie(left_convert_actions, right_convert_actions) = table_join.createConvertingActions(
        left.getCurrentHeader()->getColumnsWithTypeAndName(), right.getCurrentHeader()->getColumnsWithTypeAndName());

    if (right_convert_actions)
    {
        auto converting_step = std::make_unique<ExpressionStep>(right.getCurrentHeader(), std::move(*right_convert_actions));
        converting_step->setStepDescription("Convert joined columns");
        steps.emplace_back(converting_step.get());
        right.addStep(std::move(converting_step));
    }

    if (left_convert_actions)
    {
        auto converting_step = std::make_unique<ExpressionStep>(left.getCurrentHeader(), std::move(*left_convert_actions));
        converting_step->setStepDescription("Convert joined columns");
        steps.emplace_back(converting_step.get());
        left.addStep(std::move(converting_step));
    }
}

/// Join keys are collected from substrait::JoinRel::expression() which only contains the equal join conditions.
void JoinRelParser::collectJoinKeys(
    TableJoin & table_join, const substrait::JoinRel & join_rel, const DB::Block & left_header, const DB::Block & right_header)
{
    if (!join_rel.has_expression())
        return;
    /// Support only one join clause.
    table_join.addDisjunct();
    const auto & expr = join_rel.expression();
    auto & join_clause = table_join.getClauses().back();
    std::list<const substrait::Expression *> expressions_stack;
    expressions_stack.push_back(&expr);
    while (!expressions_stack.empty())
    {
        /// Must handle the expressions in DF order. It matters in sort merge join.
        const auto * current_expr = expressions_stack.back();
        expressions_stack.pop_back();
        if (!current_expr->has_scalar_function())
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Function expression is expected");
        auto function_name = parseFunctionName(current_expr->scalar_function());
        if (!function_name)
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Invalid function expression");
        if (*function_name == "equals")
        {
            String left_key, right_key;
            size_t left_pos = 0, right_pos = 0;
            for (const auto & arg : current_expr->scalar_function().arguments())
            {
                auto field_index = SubstraitParserUtils::getStructFieldIndex(arg.value());
                if (!field_index)
                {
                    throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "A column reference is expected");
                }
                auto col_pos_ref = *field_index;
                if (col_pos_ref < left_header.columns())
                {
                    left_pos = col_pos_ref;
                    left_key = left_header.getByPosition(col_pos_ref).name;
                }
                else
                {
                    right_pos = col_pos_ref - left_header.columns();
                    right_key = right_header.getByPosition(col_pos_ref - left_header.columns()).name;
                }
            }
            if (left_key.empty() || right_key.empty())
                throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Invalid key equal join condition");
            join_clause.addKey(left_key, right_key, false);
        }
        else if (*function_name == "and")
        {
            expressions_stack.push_back(&current_expr->scalar_function().arguments().at(1).value());
            expressions_stack.push_back(&current_expr->scalar_function().arguments().at(0).value());
        }
        else
        {
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Unknow function: {}", *function_name);
        }
    }
}

bool JoinRelParser::applyJoinFilter(
    DB::TableJoin & table_join,
    const substrait::JoinRel & join_rel,
    DB::QueryPlan & left,
    DB::QueryPlan & right,
    bool allow_mixed_condition)
{
    if (!join_rel.has_post_join_filter())
        return true;
    const auto & expr = join_rel.post_join_filter();

    const auto & left_header = *left.getCurrentHeader();
    const auto & right_header = *right.getCurrentHeader();
    ColumnsWithTypeAndName mixed_columns;
    std::unordered_set<String> added_column_name;
    for (const auto & col : left_header.getColumnsWithTypeAndName())
    {
        mixed_columns.emplace_back(col);
        added_column_name.insert(col.name);
    }
    for (const auto & col : right_header.getColumnsWithTypeAndName())
    {
        const auto & renamed_col_name = table_join.renamedRightColumnNameWithAlias(col.name);
        if (added_column_name.find(col.name) != added_column_name.end())
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Right column's name conflict with left column: {}", col.name);
        mixed_columns.emplace_back(col);
        added_column_name.insert(col.name);
    }
    DB::Block mixed_header(mixed_columns);

    auto table_sides = extractTableSidesFromExpression(expr, left_header, right_header);

    auto get_input_expressions = [](const DB::Block & header)
    {
        std::vector<substrait::Expression> exprs;
        for (size_t i = 0; i < header.columns(); ++i)
        {
            substrait::Expression expr = SubstraitParserUtils::buildStructFieldExpression(i);
            exprs.emplace_back(expr);
        }
        return exprs;
    };

    /// If the columns in the expression are all from one table, use analyzer_left_filter_condition_column_name
    /// and analyzer_left_filter_condition_column_name to filt the join result data. It requires to build the filter
    /// column at first.
    /// If the columns in the expression are from both tables, use mixed_join_expression to filt the join result data.
    /// the filter columns will be built inner the join step.
    if (table_sides.size() == 1)
    {
        auto table_side = *table_sides.begin();
        if (table_side == DB::JoinTableSide::Left)
        {
            auto input_exprs = get_input_expressions(left_header);
            input_exprs.push_back(expr);
            auto actions_dag = expressionsToActionsDAG(input_exprs, left_header);
            table_join.getClauses().back().analyzer_left_filter_condition_column_name = actions_dag.getOutputs().back()->result_name;
            QueryPlanStepPtr before_join_step = std::make_unique<ExpressionStep>(left.getCurrentHeader(), std::move(actions_dag));
            before_join_step->setStepDescription("Before JOIN LEFT");
            steps.emplace_back(before_join_step.get());
            left.addStep(std::move(before_join_step));
        }
        else
        {
            /// since the field reference in expr is the index of left_header ++ right_header, so we use
            /// mixed_header to build the actions_dag
            auto input_exprs = get_input_expressions(mixed_header);
            input_exprs.push_back(expr);
            auto actions_dag = expressionsToActionsDAG(input_exprs, mixed_header);

            /// clear unused columns in actions_dag
            for (const auto & col : left_header.getColumnsWithTypeAndName())
                actions_dag.removeUnusedResult(col.name);
            actions_dag.removeUnusedActions();

            table_join.getClauses().back().analyzer_right_filter_condition_column_name = actions_dag.getOutputs().back()->result_name;
            QueryPlanStepPtr before_join_step = std::make_unique<ExpressionStep>(right.getCurrentHeader(), std::move(actions_dag));
            before_join_step->setStepDescription("Before JOIN RIGHT");
            steps.emplace_back(before_join_step.get());
            right.addStep(std::move(before_join_step));
        }
    }
    else if (table_sides.size() == 2)
    {
        if (!allow_mixed_condition)
            return false;
        auto mixed_join_expressions_actions = expressionsToActionsDAG({expr}, mixed_header);
        mixed_join_expressions_actions.removeUnusedActions();
        table_join.getMixedJoinExpression()
            = std::make_shared<DB::ExpressionActions>(std::move(mixed_join_expressions_actions), ExpressionActionsSettings(context));
    }
    else
    {
        throw DB::Exception(
            DB::ErrorCodes::LOGICAL_ERROR, "Not any table column is used in the join condition.\n{}", join_rel.DebugString());
    }
    return true;
}

void JoinRelParser::addPostFilter(DB::QueryPlan & query_plan, const substrait::JoinRel & join)
{
    std::string filter_name;
    ActionsDAG actions_dag{query_plan.getCurrentHeader()->getColumnsWithTypeAndName()};
    if (!join.post_join_filter().has_scalar_function())
    {
        // It may be singular_or_list
        const auto * in_node = expression_parser->parseExpression(actions_dag, join.post_join_filter());
        filter_name = in_node->result_name;
    }
    else
    {
        const auto * func_node = expression_parser->parseFunction(join.post_join_filter().scalar_function(), actions_dag, true);
        filter_name = func_node->result_name;
    }
    auto filter_step = std::make_unique<FilterStep>(query_plan.getCurrentHeader(), std::move(actions_dag), filter_name, true);
    filter_step->setStepDescription("Post Join Filter");
    steps.emplace_back(filter_step.get());
    query_plan.addStep(std::move(filter_step));
}

/// Only support following pattern: a1 = b1 or a2 = b2 or (a3 = b3 and a4 = b4)
bool JoinRelParser::couldRewriteToMultiJoinOnClauses(
    const DB::TableJoin::JoinOnClause & prefix_clause,
    std::vector<DB::TableJoin::JoinOnClause> & clauses,
    const substrait::JoinRel & join_rel,
    const DB::Block & left_header,
    const DB::Block & right_header)
{
    if (!join_rel.has_post_join_filter())
        return false;
    const auto & filter_expr = join_rel.post_join_filter();

    auto check_function = [&](const String function_name_, const substrait::Expression & e)
    {
        if (!e.has_scalar_function())
            return false;
        auto function_name = parseFunctionName(e.scalar_function());
        return function_name.has_value() && *function_name == function_name_;
    };

    std::function<void(std::vector<const substrait::Expression *> &, const substrait::Expression &)> dfs_visit_or_expr
        = [&](std::vector<const substrait::Expression *> & or_exprs, const substrait::Expression & e) -> void
    {
        if (!check_function("or", e))
        {
            or_exprs.push_back(&e);
            return;
        }
        const auto & args = e.scalar_function().arguments();
        dfs_visit_or_expr(or_exprs, args[0].value());
        dfs_visit_or_expr(or_exprs, args[1].value());
    };

    std::function<void(std::vector<const substrait::Expression *> &, const substrait::Expression &)> dfs_visit_and_expr
        = [&](std::vector<const substrait::Expression *> & and_exprs, const substrait::Expression & e) -> void
    {
        if (!check_function("and", e))
        {
            and_exprs.push_back(&e);
            return;
        }
        const auto & args = e.scalar_function().arguments();
        dfs_visit_and_expr(and_exprs, args[0].value());
        dfs_visit_and_expr(and_exprs, args[1].value());
    };

    auto visit_equal_expr = [&](const substrait::Expression & e) -> std::optional<std::pair<String, String>>
    {
        if (!check_function("equals", e))
            return {};
        const auto & args = e.scalar_function().arguments();
        auto l_field_ref = SubstraitParserUtils::getStructFieldIndex(args[0].value());
        auto r_field_ref = SubstraitParserUtils::getStructFieldIndex(args[1].value());
        if (!l_field_ref.has_value() || !r_field_ref.has_value())
            return {};
        size_t l_pos = *l_field_ref;
        size_t r_pos = *r_field_ref;
        size_t l_cols = left_header.columns();
        size_t total_cols = l_cols + right_header.columns();

        if (l_pos < l_cols && r_pos >= l_cols && r_pos < total_cols)
            return std::make_pair(left_header.getByPosition(l_pos).name, right_header.getByPosition(r_pos - l_cols).name);
        else if (r_pos < l_cols && l_pos >= l_cols && l_pos < total_cols)
            return std::make_pair(left_header.getByPosition(r_pos).name, right_header.getByPosition(l_pos - l_cols).name);
        return {};
    };


    std::vector<const substrait::Expression *> or_exprs;
    dfs_visit_or_expr(or_exprs, filter_expr);
    if (or_exprs.empty())
        return false;
    for (const auto * or_expr : or_exprs)
    {
        DB::TableJoin::JoinOnClause new_clause = prefix_clause;
        clauses.push_back(new_clause);
        auto & current_clause = clauses.back();
        std::vector<const substrait::Expression *> and_exprs;
        dfs_visit_and_expr(and_exprs, *or_expr);
        for (const auto * and_expr : and_exprs)
        {
            auto join_keys = visit_equal_expr(*and_expr);
            if (!join_keys)
                return false;
            current_clause.addKey(join_keys->first, join_keys->second, false);
        }
    }
    return true;
}

DB::QueryPlanPtr JoinRelParser::buildMultiOnClauseHashJoin(
    std::shared_ptr<DB::TableJoin> table_join,
    DB::QueryPlanPtr left_plan,
    DB::QueryPlanPtr right_plan,
    const std::vector<DB::TableJoin::JoinOnClause> & join_on_clauses)
{
    DB::TableJoin::JoinOnClause & base_join_on_clause = table_join->getOnlyClause();
    base_join_on_clause = join_on_clauses[0];
    for (size_t i = 1; i < join_on_clauses.size(); ++i)
    {
        table_join->addDisjunct();
        auto & join_on_clause = table_join->getClauses().back();
        join_on_clause = join_on_clauses[i];
    }

    LOG_INFO(getLogger("JoinRelParser"), "multi join on clauses:\n{}", DB::TableJoin::formatClauses(table_join->getClauses()));

    JoinPtr hash_join = std::make_shared<HashJoin>(table_join, right_plan->getCurrentHeader());
    QueryPlanStepPtr join_step = std::make_unique<DB::JoinStep>(
        left_plan->getCurrentHeader(),
        right_plan->getCurrentHeader(),
        hash_join,
        context->getSettingsRef()[Setting::max_block_size],
        context->getSettingsRef()[Setting::min_joined_block_size_rows],
        context->getSettingsRef()[Setting::min_joined_block_size_bytes],
        1,
        /* required_output_ = */ NameSet{},
        false,
        /* use_new_analyzer_ = */ false);
    join_step->setStepDescription("Multi join on clause hash join");
    steps.emplace_back(join_step.get());
    std::vector<QueryPlanPtr> plans;
    plans.emplace_back(std::move(left_plan));
    plans.emplace_back(std::move(right_plan));
    auto query_plan = std::make_unique<QueryPlan>();
    query_plan->unitePlans(std::move(join_step), {std::move(plans)});
    return query_plan;
}

DB::QueryPlanPtr JoinRelParser::buildSingleOnClauseHashJoin(
    const substrait::JoinRel & join_rel, std::shared_ptr<DB::TableJoin> table_join, DB::QueryPlanPtr left_plan, DB::QueryPlanPtr right_plan)
{
    applyJoinFilter(*table_join, join_rel, *left_plan, *right_plan, true);
    /// Following is some configurations for grace hash join.
    /// - spark.gluten.sql.columnar.backend.ch.runtime_settings.join_algorithm=grace_hash. This will
    ///   enable grace hash join.
    /// - spark.gluten.sql.columnar.backend.ch.runtime_settings.max_bytes_in_join=3145728. This setup
    ///   the memory limitation fro grace hash join. If the memory consumption exceeds the limitation,
    ///   data will be spilled to disk. Don't set the limitation too small, otherwise the buckets number
    ///   will be too large and the performance will be bad.
    JoinPtr hash_join = nullptr;
    MultiEnum<DB::JoinAlgorithm> join_algorithm = context->getSettingsRef()[Setting::join_algorithm];
    if (join_algorithm.isSet(DB::JoinAlgorithm::GRACE_HASH))
    {
        hash_join = std::make_shared<GraceHashJoin>(
            context->getSettingsRef()[Setting::grace_hash_join_initial_buckets],
            context->getSettingsRef()[Setting::grace_hash_join_max_buckets],
            table_join, left_plan->getCurrentHeader(), right_plan->getCurrentHeader(), context->getTempDataOnDisk());
    }
    else
    {
        hash_join = std::make_shared<HashJoin>(table_join, right_plan->getCurrentHeader());
    }
    QueryPlanStepPtr join_step = std::make_unique<DB::JoinStep>(
        left_plan->getCurrentHeader(),
        right_plan->getCurrentHeader(),
        hash_join,
        context->getSettingsRef()[Setting::max_block_size],
        context->getSettingsRef()[Setting::min_joined_block_size_rows],
        context->getSettingsRef()[Setting::min_joined_block_size_bytes],
        1,
        /* required_output_ = */ NameSet{},
        false,
        /* use_new_analyzer_ = */ false);

    join_step->setStepDescription("HASH_JOIN");
    steps.emplace_back(join_step.get());
    std::vector<QueryPlanPtr> plans;
    plans.emplace_back(std::move(left_plan));
    plans.emplace_back(std::move(right_plan));

    auto query_plan = std::make_unique<QueryPlan>();
    query_plan->unitePlans(std::move(join_step), {std::move(plans)});
    return query_plan;
}

void registerJoinRelParser(RelParserFactory & factory)
{
    auto builder = [](ParserContextPtr parser_context) { return std::make_shared<JoinRelParser>(parser_context); };
    factory.registerBuilder(substrait::Rel::RelTypeCase::kJoin, builder);
}

}