CHUtil.cpp

/*
 * 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 "CHUtil.h"

#include <filesystem>
#include <memory>
#include <optional>
#include <unistd.h>
#include <AggregateFunctions/Combinators/AggregateFunctionCombinatorFactory.h>
#include <AggregateFunctions/registerAggregateFunctions.h>
#include <Columns/ColumnArray.h>
#include <Columns/ColumnConst.h>
#include <Columns/ColumnNullable.h>
#include <Columns/ColumnTuple.h>
#include <Columns/IColumn.h>
#include <Core/Block.h>
#include <Core/ColumnWithTypeAndName.h>
#include <Core/Defines.h>
#include <Core/NamesAndTypes.h>
#include <Core/ServerSettings.h>
#include <Core/Settings.h>
#include <DataTypes/DataTypeArray.h>
#include <DataTypes/DataTypeMap.h>
#include <DataTypes/DataTypeNullable.h>
#include <DataTypes/DataTypeString.h>
#include <DataTypes/DataTypeTuple.h>
#include <DataTypes/DataTypesDecimal.h>
#include <DataTypes/DataTypesNumber.h>
#include <DataTypes/NestedUtils.h>
#include <Disks/registerDisks.h>
#include <Disks/registerGlutenDisks.h>
#include <Formats/registerFormats.h>
#include <Functions/FunctionFactory.h>
#include <Functions/FunctionHelpers.h>
#include <Functions/registerFunctions.h>
#include <IO/SharedThreadPools.h>
#include <Interpreters/JIT/CompiledExpressionCache.h>
#include <Parser/RelParsers/RelParser.h>
#include <Planner/PlannerActionsVisitor.h>
#include <Processors/Chunk.h>
#include <Processors/QueryPlan/ExpressionStep.h>
#include <Processors/QueryPlan/QueryPlan.h>
#include <QueryPipeline/printPipeline.h>
#include <Storages/Cache/CacheManager.h>
#include <Storages/MergeTree/StorageMergeTreeFactory.h>
#include <Storages/Output/WriteBufferBuilder.h>
#include <Storages/SubstraitSource/ReadBufferBuilder.h>
#include <arrow/util/compression.h>
#include <boost/algorithm/string/case_conv.hpp>
#include <sys/resource.h>
#include <Poco/Logger.h>
#include <Poco/Util/MapConfiguration.h>
#include <Common/BitHelpers.h>
#include <Common/BlockTypeUtils.h>
#include <Common/Config/ConfigProcessor.h>
#include <Common/GlutenSignalHandler.h>
#include <Common/LoggerExtend.h>
#include <Common/QueryContext.h>
#include <Common/logger_useful.h>
#include <Common/typeid_cast.h>

namespace DB
{
namespace ServerSetting
{
extern const ServerSettingsString primary_index_cache_policy;
extern const ServerSettingsUInt64 primary_index_cache_size;
extern const ServerSettingsDouble primary_index_cache_size_ratio;
extern const ServerSettingsUInt64 max_prefixes_deserialization_thread_pool_size;
extern const ServerSettingsUInt64 max_prefixes_deserialization_thread_pool_free_size;
extern const ServerSettingsUInt64 prefixes_deserialization_thread_pool_thread_pool_queue_size;
extern const ServerSettingsUInt64 max_thread_pool_size;
extern const ServerSettingsUInt64 thread_pool_queue_size;
extern const ServerSettingsUInt64 max_io_thread_pool_size;
extern const ServerSettingsUInt64 io_thread_pool_queue_size;
extern const ServerSettingsString vector_similarity_index_cache_policy;
extern const ServerSettingsUInt64 vector_similarity_index_cache_size;
extern const ServerSettingsUInt64 vector_similarity_index_cache_max_entries;
extern const ServerSettingsDouble vector_similarity_index_cache_size_ratio;
}
namespace Setting
{
extern const SettingsMaxThreads max_threads;
extern const SettingsUInt64 prefer_external_sort_block_bytes;
extern const SettingsUInt64 max_bytes_before_external_sort;
extern const SettingsDouble max_bytes_ratio_before_external_sort;
extern const SettingsBool query_plan_merge_filters;
extern const SettingsBool compile_expressions;
extern const SettingsShortCircuitFunctionEvaluation short_circuit_function_evaluation;
extern const SettingsUInt64 output_format_compression_level;
extern const SettingsBool query_plan_optimize_lazy_materialization;
}
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
extern const int UNKNOWN_TYPE;
extern const int CANNOT_PARSE_PROTOBUF_SCHEMA;
}

extern void registerAggregateFunctionUniqHyperLogLogPlusPlus(AggregateFunctionFactory &);
}

namespace local_engine
{
namespace fs = std::filesystem;

DB::Block BlockUtil::buildRowCountHeader()
{
    return DB::Block{createColumn<UInt8>({}, VIRTUAL_ROW_COUNT_COLUMN)};
}

DB::Chunk BlockUtil::buildRowCountChunk(UInt64 rows)
{
    return DB::Chunk{{createColumnConst<UInt8>(rows, 0)}, rows};
}

DB::Block BlockUtil::buildRowCountBlock(UInt64 rows)
{
    return DB::Block{createColumnConst<UInt8>(rows, 0, VIRTUAL_ROW_COUNT_COLUMN)};
}

/// The column names may be different in two blocks.
/// and the nullability also could be different, with TPCDS-Q1 as an example.
DB::ColumnWithTypeAndName
BlockUtil::convertColumnAsNecessary(const DB::ColumnWithTypeAndName & column, const DB::ColumnWithTypeAndName & sample_column)
{
    if (sample_column.type->equals(*column.type))
        return {column.column, column.type, sample_column.name};
    else if (sample_column.type->isNullable() && !column.type->isNullable() && DB::removeNullable(sample_column.type)->equals(*column.type))
    {
        auto nullable_column = column;
        DB::JoinCommon::convertColumnToNullable(nullable_column);
        return {nullable_column.column, sample_column.type, sample_column.name};
    }
    else
        throw DB::Exception(
            DB::ErrorCodes::LOGICAL_ERROR,
            "Columns have different types. original:{} expected:{}",
            column.dumpStructure(),
            sample_column.dumpStructure());
}

/**
 * There is a special case with which we need be careful. In spark, struct/map/list are always
 * wrapped in Nullable, but this should not happen in clickhouse.
 */
DB::Block
BlockUtil::flattenBlock(const DB::Block & block, UInt64 flags, bool recursively, const std::unordered_set<size_t> & columns_to_skip_flatten)
{
    DB::Block res;

    for (size_t col_i = 0; col_i < block.columns(); ++col_i)
    {
        const auto & elem = block.getByPosition(col_i);

        if (columns_to_skip_flatten.contains(col_i))
        {
            res.insert(elem);
            continue;
        }

        DB::DataTypePtr nested_type = removeNullable(elem.type);
        DB::ColumnPtr nested_col = elem.column;
        DB::ColumnPtr null_map_col = nullptr;
        // A special case, const(Nullable(nothing))
        if (elem.type->isNullable() && typeid_cast<const DB::ColumnNullable *>(elem.column->getPtr().get()))
        {
            const auto * nullable_col = typeid_cast<const DB::ColumnNullable *>(elem.column->getPtr().get());
            nested_col = nullable_col->getNestedColumnPtr();
            null_map_col = nullable_col->getNullMapColumnPtr();
        }

        if (const DB::DataTypeArray * type_arr = typeid_cast<const DB::DataTypeArray *>(nested_type.get()))
        {
            const DB::DataTypeTuple * type_tuple = typeid_cast<const DB::DataTypeTuple *>(type_arr->getNestedType().get());
            if (type_tuple && type_tuple->hasExplicitNames() && (flags & FLAT_NESTED_TABLE))
            {
                const DB::DataTypes & element_types = type_tuple->getElements();
                const DB::Strings & names = type_tuple->getElementNames();
                size_t tuple_size = element_types.size();

                bool is_const = isColumnConst(*nested_col);
                const DB::ColumnArray * column_array;
                if (is_const)
                    column_array = typeid_cast<const DB::ColumnArray *>(&assert_cast<const DB::ColumnConst &>(*nested_col).getDataColumn());
                else
                    column_array = typeid_cast<const DB::ColumnArray *>(nested_col.get());

                const DB::ColumnPtr & column_offsets = column_array->getOffsetsPtr();

                const DB::ColumnTuple & column_tuple = typeid_cast<const DB::ColumnTuple &>(column_array->getData());
                const auto & element_columns = column_tuple.getColumns();

                for (size_t i = 0; i < tuple_size; ++i)
                {
                    String nested_name = DB::Nested::concatenateName(elem.name, names[i]);
                    DB::ColumnPtr column_array_of_element = DB::ColumnArray::create(element_columns[i], column_offsets);
                    auto named_column_array_of_element = DB::ColumnWithTypeAndName(
                        is_const ? DB::ColumnConst::create(std::move(column_array_of_element), block.rows()) : column_array_of_element,
                        std::make_shared<DB::DataTypeArray>(element_types[i]),
                        nested_name);

                    if (null_map_col)
                    {
                        // Should all field columns have the same null map ?
                        DB::DataTypePtr null_type = std::make_shared<DB::DataTypeNullable>(element_types[i]);
                        named_column_array_of_element.column
                            = DB::ColumnNullable::create(named_column_array_of_element.column, null_map_col);
                        named_column_array_of_element.type = null_type;
                    }

                    if (recursively)
                    {
                        auto flatten_one_col_block = flattenBlock({named_column_array_of_element}, flags, recursively);
                        for (const auto & named_col : flatten_one_col_block.getColumnsWithTypeAndName())
                            res.insert(named_col);
                    }
                    else
                        res.insert(named_column_array_of_element);
                }
            }
            else
                res.insert(elem);
        }
        else if (const DB::DataTypeTuple * type_tuple = typeid_cast<const DB::DataTypeTuple *>(nested_type.get()))
        {
            if ((flags & FLAT_STRUCT_FORCE) || (type_tuple->hasExplicitNames() && (flags & FLAT_STRUCT)))
            {
                const DB::DataTypes & element_types = type_tuple->getElements();
                DB::Strings element_names = type_tuple->getElementNames();
                if (element_names.empty())
                {
                    // This is a struct without named fields, we should flatten it.
                    // But we can't get the field names, so we use the field index as the field name.
                    for (size_t i = 0; i < element_types.size(); ++i)
                        element_names.push_back(elem.name + "_filed_" + std::to_string(i));
                }

                const DB::ColumnTuple * column_tuple;
                if (isColumnConst(*nested_col))
                    column_tuple = typeid_cast<const DB::ColumnTuple *>(&assert_cast<const DB::ColumnConst &>(*nested_col).getDataColumn());
                else
                    column_tuple = typeid_cast<const DB::ColumnTuple *>(nested_col.get());

                size_t tuple_size = column_tuple->tupleSize();
                for (size_t i = 0; i < tuple_size; ++i)
                {
                    const auto & element_column = column_tuple->getColumn(i);
                    String nested_name = DB::Nested::concatenateName(elem.name, element_names[i]);
                    auto new_element_col = DB::ColumnWithTypeAndName(element_column.getPtr(), element_types[i], nested_name);
                    if (null_map_col && !element_types[i]->isNullable())
                    {
                        // Should all field columns have the same null map ?
                        new_element_col.column = DB::ColumnNullable::create(new_element_col.column, null_map_col);
                        new_element_col.type = std::make_shared<DB::DataTypeNullable>(new_element_col.type);
                    }

                    if (recursively)
                    {
                        DB::Block one_col_block({new_element_col});
                        auto flatten_one_col_block = flattenBlock(one_col_block, flags, recursively);
                        for (const auto & named_col : flatten_one_col_block.getColumnsWithTypeAndName())
                            res.insert(named_col);
                    }
                    else
                        res.insert(std::move(new_element_col));
                }
            }
            else
                res.insert(elem);
        }
        else
            res.insert(elem);
    }

    return res;
}

DB::Block BlockUtil::concatenateBlocksMemoryEfficiently(std::vector<DB::Block> && blocks)
{
    if (blocks.empty())
        return {};

    size_t num_rows = 0;
    for (const auto & block : blocks)
        num_rows += block.rows();

    DB::Block out = blocks[0].cloneEmpty();
    DB::MutableColumns columns = out.mutateColumns();

    for (size_t i = 0; i < columns.size(); ++i)
    {
        columns[i]->reserve(num_rows);
        for (auto & block : blocks)
        {
            const auto & tmp_column = *block.getByPosition(0).column;
            columns[i]->insertRangeFrom(tmp_column, 0, block.rows());
            block.erase(0);
        }
    }
    blocks.clear();

    out.setColumns(std::move(columns));
    return out;
}

size_t PODArrayUtil::adjustMemoryEfficientSize(size_t n)
{
    /// According to definition of DEFUALT_BLOCK_SIZE
    size_t padding_n = 2 * DB::PADDING_FOR_SIMD - 1;
    size_t rounded_n = roundUpToPowerOfTwoOrZero(n);
    size_t padded_n = n;
    if (rounded_n > n + n / 2)
    {
        size_t smaller_rounded_n = rounded_n / 2;
        padded_n = smaller_rounded_n < padding_n ? n : smaller_rounded_n - padding_n;
    }
    else
    {
        padded_n = rounded_n - padding_n;
    }
    return padded_n;
}

NestedColumnExtractHelper::NestedColumnExtractHelper(const DB::Block & block_, bool case_insentive_)
    : block(block_), case_insentive(case_insentive_)
{
}

std::optional<DB::ColumnWithTypeAndName> NestedColumnExtractHelper::extractColumn(const String & column_name)
{
    if (const auto * col = findColumn(block, column_name))
        return {*col};

    auto nested_names = DB::Nested::splitName(column_name);
    if (case_insentive)
    {
        boost::to_lower(nested_names.first);
        boost::to_lower(nested_names.second);
    }
    if (!findColumn(block, nested_names.first))
        return {};

    if (!nested_tables.contains(nested_names.first))
    {
        DB::ColumnsWithTypeAndName columns = {*findColumn(block, nested_names.first)};
        nested_tables[nested_names.first] = std::make_shared<DB::Block>(BlockUtil::flattenBlock(columns));
    }

    return extractColumn(column_name, nested_names.first, nested_names.second);
}

std::optional<DB::ColumnWithTypeAndName> NestedColumnExtractHelper::extractColumn(
    const String & original_column_name, const String & column_name_prefix, const String & column_name_suffix)
{
    auto table_iter = nested_tables.find(column_name_prefix);
    if (table_iter == nested_tables.end())
        return {};

    auto & nested_table = table_iter->second;
    auto nested_names = DB::Nested::splitName(column_name_suffix);
    auto new_column_name_prefix = DB::Nested::concatenateName(column_name_prefix, nested_names.first);
    if (nested_names.second.empty())
    {
        if (const auto * column_ref = findColumn(*nested_table, new_column_name_prefix))
        {
            DB::ColumnWithTypeAndName column = *column_ref;
            if (case_insentive)
                column.name = original_column_name;
            return {std::move(column)};
        }
        else
        {
            return {};
        }
    }

    const auto * sub_col = findColumn(*nested_table, new_column_name_prefix);
    if (!sub_col)
        return {};

    DB::ColumnsWithTypeAndName columns = {*sub_col};
    DB::Block sub_block(columns);
    nested_tables[new_column_name_prefix] = std::make_shared<DB::Block>(BlockUtil::flattenBlock(sub_block));
    return extractColumn(original_column_name, new_column_name_prefix, nested_names.second);
}

const DB::ColumnWithTypeAndName * NestedColumnExtractHelper::findColumn(const DB::Block & in_block, const std::string & name) const
{
    return in_block.findByName(name, case_insentive);
}

const DB::ActionsDAG::Node * ActionsDAGUtil::convertNodeType(
    DB::ActionsDAG & actions_dag,
    const DB::ActionsDAG::Node * node,
    const DB::DataTypePtr & cast_to_type,
    DB::ContextPtr context,
    const std::string & result_name,
    DB::CastType cast_type)
{
    DB::ColumnWithTypeAndName type_name_col;
    type_name_col.name = cast_to_type->getName();
    type_name_col.column = DB::DataTypeString().createColumnConst(0, type_name_col.name);
    type_name_col.type = std::make_shared<DB::DataTypeString>();
    const auto * right_arg = &actions_dag.addColumn(std::move(type_name_col));
    const auto * left_arg = node;
    DB::CastDiagnostic diagnostic = {node->result_name, node->result_name};
    DB::ColumnWithTypeAndName left_column{nullptr, node->result_type, {}};
    DB::ActionsDAG::NodeRawConstPtrs children = {left_arg, right_arg};
    auto func_base_cast = createInternalCast(std::move(left_column), cast_to_type, cast_type, diagnostic, context);

    return &actions_dag.addFunction(func_base_cast, std::move(children), result_name);
}

const DB::ActionsDAG::Node * ActionsDAGUtil::convertNodeTypeIfNeeded(
    DB::ActionsDAG & actions_dag,
    const DB::ActionsDAG::Node * node,
    const DB::DataTypePtr & dst_type,
    DB::ContextPtr context,
    const std::string & result_name,
    DB::CastType cast_type)
{
    if (node->result_type->equals(*dst_type))
        return node;

    return convertNodeType(actions_dag, node, dst_type, context, result_name, cast_type);
}

String QueryPipelineUtil::explainPipeline(DB::QueryPipeline & pipeline)
{
    DB::WriteBufferFromOwnString buf;
    const auto & processors = pipeline.getProcessors();
    DB::printPipelineCompact(processors, buf, true);
    return buf.str();
}

using namespace DB;

std::vector<String> BackendInitializerUtil::wrapDiskPathConfig(
    const String & path_prefix, const String & path_suffix, Poco::Util::AbstractConfiguration & config)
{
    std::vector<String> changed_paths;
    if (path_prefix.empty() && path_suffix.empty())
        return changed_paths;

    auto change_func = [&](String key) -> void
    {
        if (const String value = config.getString(key, ""); value != "")
        {
            const String change_value = path_prefix + value + path_suffix;
            config.setString(key, change_value);
            changed_paths.emplace_back(change_value);
            LOG_INFO(getLogger("BackendInitializerUtil"), "Change config `{}` from '{}' to {}.", key, value, change_value);
        }
    };

    Poco::Util::AbstractConfiguration::Keys disks;
    std::unordered_set<String> disk_types = {GlutenObjectStorageConfig::S3_DISK_TYPE, GlutenObjectStorageConfig::HDFS_DISK_TYPE, "cache"};
    config.keys("storage_configuration.disks", disks);

    std::ranges::for_each(
        disks,
        [&](const auto & disk_name)
        {
            String disk_prefix = "storage_configuration.disks." + disk_name;
            String disk_type = config.getString(disk_prefix + ".type", "");
            if (!disk_types.contains(disk_type))
                return;
            if (disk_type == "cache")
                change_func(disk_prefix + ".path");
            else if (disk_type == GlutenObjectStorageConfig::S3_DISK_TYPE || disk_type == GlutenObjectStorageConfig::HDFS_DISK_TYPE)
                change_func(disk_prefix + ".metadata_path");
        });

    change_func("path");
    change_func(GlutenCacheConfig::PREFIX + ".path");

    return changed_paths;
}

DB::Context::ConfigurationPtr BackendInitializerUtil::initConfig(const SparkConfigs::ConfigMap & spark_conf_map)
{
    DB::Context::ConfigurationPtr config;

    if (const String config_file = tryGetConfigFile(spark_conf_map); !config_file.empty())
    {
        if (fs::exists(config_file) && fs::is_regular_file(config_file))
        {
            ConfigProcessor config_processor(config_file, false, true);
            DB::ConfigProcessor::setConfigPath(fs::path(config_file).parent_path());
            const auto loaded_config = config_processor.loadConfig(false);
            config = loaded_config.configuration;
        }
        else
            throw DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "{} is not a valid configure file.", config_file);
    }
    else
        config = Poco::AutoPtr(new Poco::Util::MapConfiguration());

    for (const auto & [key, value] : spark_conf_map)
    {
        if (key.starts_with(CH_RUNTIME_CONFIG_PREFIX) && key != CH_RUNTIME_CONFIG_FILE)
        {
            // Apply spark.gluten.sql.columnar.backend.ch.runtime_config.* to config
            const auto name = key.substr(CH_RUNTIME_CONFIG_PREFIX.size());
            if ((name == "storage_configuration.disks.s3.metadata_path" || name == "path") && !value.ends_with("/"))
                config->setString(name, value + "/");
            else
                config->setString(name, value);
        }
    }

    if (spark_conf_map.contains(GLUTEN_TASK_OFFHEAP))
        config->setString(MemoryConfig::CH_TASK_MEMORY, spark_conf_map.at(GLUTEN_TASK_OFFHEAP));

    const bool use_current_directory_as_tmp = config->getBool(PathConfig::USE_CURRENT_DIRECTORY_AS_TMP, false);
    char buffer[PATH_MAX];
    if (use_current_directory_as_tmp && getcwd(buffer, sizeof(buffer)) != nullptr)
        wrapDiskPathConfig(String(buffer), "", *config);

    const bool reuse_disk_cache = config->getBool("reuse_disk_cache", true);

    if (!reuse_disk_cache)
    {
        String pid = std::to_string(static_cast<Int64>(getpid()));
        auto path_need_clean = wrapDiskPathConfig("", "/" + pid, *config);
        std::lock_guard lock(BackendFinalizerUtil::paths_mutex);
        BackendFinalizerUtil::paths_need_to_clean.insert(
            BackendFinalizerUtil::paths_need_to_clean.end(), path_need_clean.begin(), path_need_clean.end());
    }

    // FIXMEX: workaround for https://github.com/ClickHouse/ClickHouse/pull/75452#pullrequestreview-2625467710
    // entry in DiskSelector::initialize
    // Bug in FileCacheSettings::loadFromConfig
    auto updateCacheDiskType = [](Poco::Util::AbstractConfiguration & config)
    {
        const std::string config_prefix = "storage_configuration.disks";
        Poco::Util::AbstractConfiguration::Keys keys;
        config.keys(config_prefix, keys);
        for (const auto & disk_name : keys)
        {
            const auto disk_config_prefix = config_prefix + "." + disk_name;
            const auto disk_type = config.getString(disk_config_prefix + ".type", "local");
            if (disk_type == "cache")
                config.setString(disk_config_prefix, "workaround");
        }
        config.setString(GlutenCacheConfig::PREFIX, "workaround");
    };

    updateCacheDiskType(*config);

    return config;
}

String BackendInitializerUtil::tryGetConfigFile(const SparkConfigs::ConfigMap & spark_conf_map)
{
    if (spark_conf_map.contains(CH_RUNTIME_CONFIG_FILE))
        return spark_conf_map.at(CH_RUNTIME_CONFIG_FILE);

    /// Try to get config path from environment variable
    if (const char * config_path = std::getenv("CLICKHOUSE_BACKEND_CONFIG"))
        return config_path;

    return String{};
}


void BackendInitializerUtil::initLoggers(DB::Context::ConfigurationPtr config)
{
    auto level = config->getString("logger.level", "warning");
    if (config->has("logger.log"))
        local_engine::LoggerExtend::initFileLogger(*config, "ClickHouseBackend");
    else
        local_engine::LoggerExtend::initConsoleLogger(level);

    logger = &Poco::Logger::get("ClickHouseBackend");
}

void BackendInitializerUtil::initEnvs(DB::Context::ConfigurationPtr config)
{
    /// Set environment variable TZ if possible
    if (config->has("timezone"))
    {
        const std::string config_timezone = config->getString("timezone");
        const String mapped_timezone = DateTimeUtil::convertTimeZone(config_timezone);
        if (0
            != setenv(
                "TZ", mapped_timezone.data(), 1)) // NOLINT(concurrency-mt-unsafe) // ok if not called concurrently with other setenv/getenv
            throw Poco::Exception("Cannot setenv TZ variable");

        tzset();
        DateLUT::setDefaultTimezone(mapped_timezone);
    }

    /// Set environment variable LIBHDFS3_CONF if possible
    if (config->has(LIBHDFS3_CONF_KEY))
    {
        std::string libhdfs3_conf = config->getString(LIBHDFS3_CONF_KEY, "");
        setenv("LIBHDFS3_CONF", libhdfs3_conf.c_str(), true); /// NOLINT
    }

    /// Enable logging in libhdfs3, logs will be written to stderr
    setenv("HDFS_ENABLE_LOGGING", "true", true); /// NOLINT

    /// Get environment varaible SPARK_USER if possible
    if (const char * spark_user_c_str = std::getenv("SPARK_USER"))
        spark_user = spark_user_c_str;
}

DB::Field BackendInitializerUtil::toField(const String & key, const String & value)
{
    if (BOOL_VALUE_SETTINGS.contains(key))
        return DB::Field(value == "true" || value == "1");
    else if (LONG_VALUE_SETTINGS.contains(key))
        return DB::Field(std::strtoll(value.c_str(), nullptr, 10));
    else
        return DB::Field(value);
}

void BackendInitializerUtil::initSettings(const SparkConfigs::ConfigMap & spark_conf_map, DB::Settings & settings)
{
    /// Initialize default setting.
    settings.set("date_time_input_format", "best_effort");
    settings.set(MERGETREE_MERGE_AFTER_INSERT, true);
    settings.set(MERGETREE_INSERT_WITHOUT_LOCAL_STORAGE, false);
    settings.set(DECIMAL_OPERATIONS_ALLOW_PREC_LOSS, true);
    settings.set("remote_filesystem_read_prefetch", false);
    settings.set("max_parsing_threads", 1);
    settings.set("max_download_threads", 1);
    settings.set("input_format_parquet_enable_row_group_prefetch", false);
    settings.set("output_format_parquet_use_custom_encoder", false);

    //1.
    // TODO: we need set Setting::max_threads to 1 by default, but now we can't get correct metrics for the some query if we set it to 1.
    // settings[Setting::max_threads] = 1;

    /// 2. After https://github.com/ClickHouse/ClickHouse/pull/71539
    /// Set false to query_plan_merge_filters.
    /// If true, we can't get correct metrics for the query
    settings[Setting::query_plan_merge_filters] = false;

    /// 3. After https://github.com/ClickHouse/ClickHouse/pull/70598.
    /// Set false to compile_expressions to avoid dead loop.
    /// TODO: FIXME set true again.
    /// We now set BuildQueryPipelineSettings according to config.
    settings[Setting::compile_expressions] = false;
    settings[Setting::short_circuit_function_evaluation] = ShortCircuitFunctionEvaluation::DISABLE;

    /// 4. After https://github.com/ClickHouse/ClickHouse/pull/73422
    /// Since we already set max_bytes_before_external_sort, set max_bytes_ratio_before_external_sort to 0
    settings[Setting::max_bytes_ratio_before_external_sort] = 0.;

    /// 5. After https://github.com/ClickHouse/ClickHouse/pull/73651.
    /// See following settings, we always use Snappy compression for Parquet, however after https://github.com/ClickHouse/ClickHouse/pull/73651,
    /// output_format_compression_level is set to 3, which is wrong, since snappy does not support it.
    settings[Setting::output_format_compression_level] = arrow::util::kUseDefaultCompressionLevel;

    /// 6. After https://github.com/ClickHouse/ClickHouse/pull/55518
    /// We currently do not support lazy materialization.
    /// "test 'order by' two keys" will failed if we enable it.
    settings[Setting::query_plan_optimize_lazy_materialization] = false;

    for (const auto & [key, value] : spark_conf_map)
    {
        // Firstly apply spark.gluten.sql.columnar.backend.ch.runtime_config.local_engine.settings.* to settings
        if (key.starts_with(CH_RUNTIME_CONFIG_PREFIX + SETTINGS_PATH + "."))
        {
            settings.set(key.substr((CH_RUNTIME_CONFIG_PREFIX + SETTINGS_PATH + ".").size()), value);
            LOG_DEBUG(&Poco::Logger::get("CHUtil"), "Set settings from config key:{} value:{}", key, value);
        }
        else if (key.starts_with(CH_RUNTIME_SETTINGS_PREFIX))
        {
            auto k = key.substr(CH_RUNTIME_SETTINGS_PREFIX.size());
            settings.set(k, toField(k, value));
            LOG_DEBUG(&Poco::Logger::get("CHUtil"), "Set settings key:{} value:{}", key, value);
        }
        else if (key.starts_with(SPARK_HADOOP_PREFIX + S3A_PREFIX))
        {
            // Apply general S3 configs, e.g. spark.hadoop.fs.s3a.access.key -> set in fs.s3a.access.key
            // deal with per bucket S3 configs, e.g. fs.s3a.bucket.bucket_name.assumed.role.arn
            // for gluten, we require first authenticate with AK/SK(or instance profile), then assume other roles with STS
            // so only the following per-bucket configs are supported:
            // 1. fs.s3a.bucket.bucket_name.assumed.role.arn
            // 2. fs.s3a.bucket.bucket_name.assumed.role.session.name
            // 3. fs.s3a.bucket.bucket_name.endpoint
            // 4. fs.s3a.bucket.bucket_name.assumed.role.externalId (non hadoop official)
            settings.set(key.substr(SPARK_HADOOP_PREFIX.length()), value);
        }
        else if (key.starts_with(SPARK_DELTA_PREFIX))
        {
            auto k = key.substr(SPARK_DELTA_PREFIX.size());
            settings.set(k, toField(k, value));
            LOG_DEBUG(&Poco::Logger::get("CHUtil"), "Set settings key:{} value:{}", key, value);
        }
        else if (key == SPARK_SESSION_TIME_ZONE)
        {
            String time_zone_val = DateTimeUtil::convertTimeZone(value);
            settings.set("session_timezone", time_zone_val);
            LOG_DEBUG(&Poco::Logger::get("CHUtil"), "Set settings key:{} value:{}", "session_timezone", time_zone_val);
        }
        else if (key == DECIMAL_OPERATIONS_ALLOW_PREC_LOSS)
        {
            settings.set(key, toField(key, value));
            LOG_DEBUG(&Poco::Logger::get("CHUtil"), "Set settings key:{} value:{}", key, value);
        }
        else if (key == TIMER_PARSER_POLICY)
        {
            settings.set(key, value);
            LOG_DEBUG(&Poco::Logger::get("CHUtil"), "Set settings key:{} value:{}", key, value);
        }
    }

    /// Finally apply some fixed kvs to settings.
    settings.set("join_use_nulls", true);
    settings.set("input_format_orc_allow_missing_columns", true);
    settings.set("input_format_orc_case_insensitive_column_matching", true);
    settings.set("input_format_orc_import_nested", true);
    settings.set("input_format_orc_skip_columns_with_unsupported_types_in_schema_inference", true);
    settings.set("input_format_parquet_allow_missing_columns", true);
    settings.set("input_format_parquet_case_insensitive_column_matching", true);
    settings.set("input_format_parquet_import_nested", true);
    settings.set("input_format_json_read_numbers_as_strings", true);
    settings.set("input_format_json_read_bools_as_numbers", false);
    settings.set("input_format_csv_trim_whitespaces", false);
    settings.set("input_format_csv_allow_cr_end_of_line", true);
    settings.set("output_format_orc_string_as_string", true);
    settings.set("output_format_parquet_version", "1.0");
    settings.set("output_format_parquet_compression_method", "snappy");
    settings.set("output_format_parquet_string_as_string", true);
    settings.set("output_format_parquet_fixed_string_as_fixed_byte_array", false);
    settings.set("output_format_json_quote_64bit_integers", false);
    settings.set("output_format_json_quote_denormals", true);
    settings.set("output_format_json_skip_null_value_in_named_tuples", true);
    settings.set("output_format_json_escape_forward_slashes", false);
    settings.set("function_json_value_return_type_allow_complex", true);
    settings.set("function_json_value_return_type_allow_nullable", true);
    settings.set("precise_float_parsing", true);
    settings.set("enable_named_columns_in_function_tuple", false);
    settings.set("date_time_64_output_format_cut_trailing_zeros_align_to_groups_of_thousands", true);
    settings.set("input_format_orc_dictionary_as_low_cardinality", false); //after https://github.com/ClickHouse/ClickHouse/pull/69481

    if (spark_conf_map.contains(GLUTEN_TASK_OFFHEAP))
    {
        auto task_memory = std::stoull(spark_conf_map.at(GLUTEN_TASK_OFFHEAP));
        if (!spark_conf_map.contains(CH_RUNTIME_SETTINGS_PREFIX + "max_bytes_before_external_sort"))
        {
            settings[Setting::max_bytes_before_external_sort] = static_cast<size_t>(0.8 * task_memory);
        }
        if (!spark_conf_map.contains(CH_RUNTIME_SETTINGS_PREFIX + "prefer_external_sort_block_bytes"))
        {
            auto mem_gb = task_memory / static_cast<double>(1_GiB);
            // 2.8x+5, Heuristics calculate the block size of external sort, [8,16]
            settings[Setting::prefer_external_sort_block_bytes]
                = std::max(std::min(static_cast<size_t>(2.8 * mem_gb + 5), 16ul), 8ul) * 1024 * 1024;
        }
    }
}

void BackendInitializerUtil::initContexts(DB::Context::ConfigurationPtr config)
{
    /// Make sure global_context and shared_context are constructed only once.
    if (auto global_context = QueryContext::globalMutableContext(); !global_context)
    {
        ServerSettings server_settings;
        server_settings.loadSettingsFromConfig(*config);

        auto log = getLogger("CHUtil");
        global_context = QueryContext::createGlobal();
        global_context->makeGlobalContext();
        global_context->setConfig(config);

        auto tmp_path = config->getString("tmp_path", PathConfig::DEFAULT_TEMP_FILE_PATH);
        if (config->getBool(PathConfig::USE_CURRENT_DIRECTORY_AS_TMP, false))
        {
            char buffer[PATH_MAX];
            if (getcwd(buffer, sizeof(buffer)) != nullptr)
                tmp_path = std::string(buffer) + tmp_path;
        };

        global_context->setTemporaryStoragePath(tmp_path, 0);
        if (!fs::exists(tmp_path))
            fs::create_directories(tmp_path);
        global_context->setPath(config->getString("path", "/"));

        String uncompressed_cache_policy = config->getString("uncompressed_cache_policy", DEFAULT_UNCOMPRESSED_CACHE_POLICY);
        size_t uncompressed_cache_size = config->getUInt64("uncompressed_cache_size", DEFAULT_UNCOMPRESSED_CACHE_MAX_SIZE);
        double uncompressed_cache_size_ratio = config->getDouble("uncompressed_cache_size_ratio", DEFAULT_UNCOMPRESSED_CACHE_SIZE_RATIO);
        global_context->setUncompressedCache(uncompressed_cache_policy, uncompressed_cache_size, uncompressed_cache_size_ratio);

        String mark_cache_policy = config->getString("mark_cache_policy", DEFAULT_MARK_CACHE_POLICY);
        size_t mark_cache_size = config->getUInt64("mark_cache_size", DEFAULT_MARK_CACHE_MAX_SIZE);
        double mark_cache_size_ratio = config->getDouble("mark_cache_size_ratio", DEFAULT_MARK_CACHE_SIZE_RATIO);
        if (!mark_cache_size)
            LOG_ERROR(log, "Mark cache is disabled, it will lead to severe performance degradation.");
        LOG_INFO(log, "mark cache size to {}.", formatReadableSizeWithBinarySuffix(mark_cache_size));
        global_context->setMarkCache(mark_cache_policy, mark_cache_size, mark_cache_size_ratio);

        String primary_index_cache_policy = server_settings[ServerSetting::primary_index_cache_policy];
        size_t primary_index_cache_size = server_settings[ServerSetting::primary_index_cache_size];
        double primary_index_cache_size_ratio = server_settings[ServerSetting::primary_index_cache_size_ratio];
        LOG_INFO(log, "Primary index cache size to {}.", formatReadableSizeWithBinarySuffix(primary_index_cache_size));
        global_context->setPrimaryIndexCache(primary_index_cache_policy, primary_index_cache_size, primary_index_cache_size_ratio);

        String index_uncompressed_cache_policy
            = config->getString("index_uncompressed_cache_policy", DEFAULT_INDEX_UNCOMPRESSED_CACHE_POLICY);
        size_t index_uncompressed_cache_size
            = config->getUInt64("index_uncompressed_cache_size", DEFAULT_INDEX_UNCOMPRESSED_CACHE_MAX_SIZE);
        double index_uncompressed_cache_size_ratio
            = config->getDouble("index_uncompressed_cache_size_ratio", DEFAULT_INDEX_UNCOMPRESSED_CACHE_SIZE_RATIO);
        global_context->setIndexUncompressedCache(
            index_uncompressed_cache_policy, index_uncompressed_cache_size, index_uncompressed_cache_size_ratio);

        String index_mark_cache_policy = config->getString("index_mark_cache_policy", DEFAULT_INDEX_MARK_CACHE_POLICY);
        size_t index_mark_cache_size = config->getUInt64("index_mark_cache_size", DEFAULT_INDEX_MARK_CACHE_MAX_SIZE);
        double index_mark_cache_size_ratio = config->getDouble("index_mark_cache_size_ratio", DEFAULT_INDEX_MARK_CACHE_SIZE_RATIO);
        global_context->setIndexMarkCache(index_mark_cache_policy, index_mark_cache_size, index_mark_cache_size_ratio);

        String vector_similarity_index_cache_policy = server_settings[ServerSetting::vector_similarity_index_cache_policy];
        size_t vector_similarity_index_cache_size = server_settings[ServerSetting::vector_similarity_index_cache_size];
        size_t vector_similarity_index_cache_max_count = server_settings[ServerSetting::vector_similarity_index_cache_max_entries];
        double vector_similarity_index_cache_size_ratio = server_settings[ServerSetting::vector_similarity_index_cache_size_ratio];
        LOG_INFO(log, "Lowered vector similarity index cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(vector_similarity_index_cache_size));

        global_context->setVectorSimilarityIndexCache(vector_similarity_index_cache_policy, vector_similarity_index_cache_size, vector_similarity_index_cache_max_count, vector_similarity_index_cache_size_ratio);

        getMergeTreePrefixesDeserializationThreadPool().initialize(
            server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_size],
            server_settings[ServerSetting::max_prefixes_deserialization_thread_pool_free_size],
            server_settings[ServerSetting::prefixes_deserialization_thread_pool_thread_pool_queue_size]);

        size_t mmap_cache_size = config->getUInt64("mmap_cache_size", DEFAULT_MMAP_CACHE_MAX_SIZE);
        global_context->setMMappedFileCache(mmap_cache_size);

        /// Initialize a dummy query result cache.
        global_context->setQueryResultCache(0, 0, 0, 0);

        /// Initialize a dummy query condition cache.
        global_context->setQueryConditionCache(DEFAULT_QUERY_CONDITION_CACHE_POLICY, 0, 0);

        // We must set the application type to CLIENT to avoid ServerUUID::get() throw exception
        global_context->setApplicationType(Context::ApplicationType::CLIENT);
    }
    else
    {
        // just for ut
        global_context->updateStorageConfiguration(*config);
    }
}

void BackendInitializerUtil::applyGlobalConfigAndSettings(const DB::Context::ConfigurationPtr & config, const DB::Settings & settings)
{
    const auto global_context = QueryContext::globalMutableContext();
    global_context->setConfig(config);
    global_context->setSettings(settings);
}

extern void registerAggregateFunctionCombinatorPartialMerge(AggregateFunctionCombinatorFactory &);
extern void registerAggregateFunctionsBloomFilter(AggregateFunctionFactory &);
extern void registerAggregateFunctionSparkAvg(AggregateFunctionFactory &);
extern void registerAggregateFunctionRowNumGroup(AggregateFunctionFactory &);
extern void registerAggregateFunctionDVRoaringBitmap(AggregateFunctionFactory &);


extern void registerFunctions(FunctionFactory &);

void registerAllFunctions()
{
    DB::registerFunctions();
    DB::registerAggregateFunctions();

    auto & agg_factory = AggregateFunctionFactory::instance();
    registerAggregateFunctionsBloomFilter(agg_factory);
    registerAggregateFunctionSparkAvg(agg_factory);
    registerAggregateFunctionRowNumGroup(agg_factory);
    DB::registerAggregateFunctionUniqHyperLogLogPlusPlus(agg_factory);
    registerAggregateFunctionDVRoaringBitmap(agg_factory);

    /// register aggregate function combinators from local_engine
    auto & combinator_factory = AggregateFunctionCombinatorFactory::instance();
    registerAggregateFunctionCombinatorPartialMerge(combinator_factory);
}

void registerGlutenDisks()
{
    registerDisks(true);
    registerGlutenDisks(true);
}

void BackendInitializerUtil::registerAllFactories()
{
    registerFormats();

    registerGlutenDisks();

    registerReadBufferBuilders();
    registerWriteBufferBuilders();

    LOG_INFO(logger, "Register read buffer builders.");

    registerRelParsers();
    LOG_INFO(logger, "Register relation parsers.");

    registerAllFunctions();
    LOG_INFO(logger, "Register all functions.");
}

void BackendInitializerUtil::initCompiledExpressionCache(DB::Context::ConfigurationPtr config)
{
#if USE_EMBEDDED_COMPILER
    /// 128 MB
    constexpr size_t compiled_expression_cache_size_default = 1024 * 1024 * 128;
    size_t compiled_expression_cache_size = config->getUInt64("compiled_expression_cache_size", compiled_expression_cache_size_default);

    constexpr size_t compiled_expression_cache_elements_size_default = 10000;
    size_t compiled_expression_cache_elements_size
        = config->getUInt64("compiled_expression_cache_elements_size", compiled_expression_cache_elements_size_default);

    CompiledExpressionCacheFactory::instance().init(compiled_expression_cache_size, compiled_expression_cache_elements_size);
#endif
}

void BackendInitializerUtil::initBackend(const SparkConfigs::ConfigMap & spark_conf_map)
{
    DB::Context::ConfigurationPtr config = initConfig(spark_conf_map);

    initLoggers(config);

    initEnvs(config);
    LOG_INFO(logger, "Init environment variables.");

    DB::Settings settings;
    initSettings(spark_conf_map, settings);
    LOG_INFO(logger, "Init settings.");

    initContexts(config);
    LOG_INFO(logger, "Init shared context and global context.");

    applyGlobalConfigAndSettings(config, settings);
    LOG_INFO(logger, "Apply configuration and setting for global context.");

    // clean static per_bucket_clients and shared_client before running local engine,
    // in case of running the multiple gluten ut in one process
    ReadBufferBuilderFactory::instance().clean();

    // Init the table metadata cache map
    StorageMergeTreeFactory::init_cache_map();

    JobScheduler::initialize(QueryContext::globalContext());
    CacheManager::initialize(QueryContext::globalMutableContext());

    std::call_once(
        init_flag,
        [&]
        {
            SignalHandler::instance().init();

            registerAllFactories();
            LOG_INFO(logger, "Register all factories.");

            initCompiledExpressionCache(config);
            LOG_INFO(logger, "Init compiled expressions cache factory.");

            ServerSettings server_settings;
            server_settings.loadSettingsFromConfig(*config);
            GlobalThreadPool::initialize(
                server_settings[ServerSetting::max_thread_pool_size], 0, server_settings[ServerSetting::thread_pool_queue_size]);
            getIOThreadPool().initialize(
                server_settings[ServerSetting::max_io_thread_pool_size], 0, server_settings[ServerSetting::io_thread_pool_queue_size]);

            const size_t active_parts_loading_threads = config->getUInt("max_active_parts_loading_thread_pool_size", 64);
            DB::getActivePartsLoadingThreadPool().initialize(
                active_parts_loading_threads,
                0, // We don't need any threads one all the parts will be loaded
                active_parts_loading_threads);

            const size_t cleanup_threads = config->getUInt("max_parts_cleaning_thread_pool_size", 128);
            getPartsCleaningThreadPool().initialize(
                cleanup_threads,
                0, // We don't need any threads one all the parts will be deleted
                cleanup_threads);

            // Avoid using LD_PRELOAD in child process
            unsetenv("LD_PRELOAD");
        });
}

void BackendFinalizerUtil::finalizeGlobally()
{
    // Make sure client caches release before ClientCacheRegistry
    ReadBufferBuilderFactory::instance().clean();
    StorageMergeTreeFactory::clear_cache_map();
    QueryContext::instance().reset();
    std::lock_guard lock(paths_mutex);
    std::ranges::for_each(
        paths_need_to_clean,
        [](const auto & path)
        {
            if (fs::exists(path))
                fs::remove_all(path);
        });
    paths_need_to_clean.clear();
}

void BackendFinalizerUtil::finalizeSessionally()
{
}

std::vector<String> BackendFinalizerUtil::paths_need_to_clean;

std::mutex BackendFinalizerUtil::paths_mutex;

Int64 DateTimeUtil::currentTimeMillis()
{
    return timeInMilliseconds(std::chrono::system_clock::now());
}

String DateTimeUtil::convertTimeZone(const String & time_zone)
{
    String res = time_zone;
    /// Convert timezone ID like '+08:00' to GMT+8:00
    if (time_zone.starts_with("+") || time_zone.starts_with("-"))
        res = "GMT" + time_zone;
    res = DateLUT::mappingForJavaTimezone(res);
    return res;
}

UInt64 MemoryUtil::getMemoryRSS()
{
    long rss = 0L;
    FILE * fp = NULL;
    char buf[4096];
    sprintf(buf, "/proc/%d/statm", getpid());
    if ((fp = fopen(buf, "r")) == NULL)
        return 0;
    fscanf(fp, "%*s%ld", &rss);
    fclose(fp);
    return rss * sysconf(_SC_PAGESIZE);
}

void JoinUtil::adjustJoinOutput(DB::QueryPlan & plan, DB::Names cols)
{
    auto header = plan.getCurrentHeader();
    std::unordered_map<String, const DB::ActionsDAG::Node *> name_to_node;
    ActionsDAG project;
    for (const auto & col : header->getColumnsWithTypeAndName())
    {
        const auto * node = &(project.addInput(col));
        name_to_node[col.name] = node;
    }
    for (const auto & col : cols)
    {
        const auto it = name_to_node.find(col);
        if (it == name_to_node.end())
        {
            throw DB::Exception(DB::ErrorCodes::LOGICAL_ERROR, "Column {} not found in header", col);
        }
        project.addOrReplaceInOutputs(*(it->second));
    }
    QueryPlanStepPtr project_step = std::make_unique<ExpressionStep>(plan.getCurrentHeader(), std::move(project));
    project_step->setStepDescription("Reorder Join Output");
    plan.addStep(std::move(project_step));
}

std::pair<DB::JoinKind, DB::JoinStrictness>
JoinUtil::getJoinKindAndStrictness(substrait::JoinRel_JoinType join_type, bool is_existence_join)
{
    switch (join_type)
    {
        case substrait::JoinRel_JoinType_JOIN_TYPE_INNER:
            return {DB::JoinKind::Inner, DB::JoinStrictness::All};
        case substrait::JoinRel_JoinType_JOIN_TYPE_LEFT_SEMI: {
            if (is_existence_join)
                return {DB::JoinKind::Left, DB::JoinStrictness::Any};
            return {DB::JoinKind::Left, DB::JoinStrictness::Semi};
        }
        case substrait::JoinRel_JoinType_JOIN_TYPE_RIGHT_SEMI:
            return {DB::JoinKind::Right, DB::JoinStrictness::Semi};
        case substrait::JoinRel_JoinType_JOIN_TYPE_LEFT_ANTI:
            return {DB::JoinKind::Left, DB::JoinStrictness::Anti};
        case substrait::JoinRel_JoinType_JOIN_TYPE_RIGHT_ANTI:
            return {DB::JoinKind::Right, DB::JoinStrictness::Anti};
        case substrait::JoinRel_JoinType_JOIN_TYPE_LEFT:
            return {DB::JoinKind::Left, DB::JoinStrictness::All};
        case substrait::JoinRel_JoinType_JOIN_TYPE_RIGHT:
            return {DB::JoinKind::Right, DB::JoinStrictness::All};
        case substrait::JoinRel_JoinType_JOIN_TYPE_OUTER:
            return {DB::JoinKind::Full, DB::JoinStrictness::All};
        default:
            throw Exception(ErrorCodes::UNKNOWN_TYPE, "unsupported join type {}.", magic_enum::enum_name(join_type));
    }
}

std::pair<DB::JoinKind, DB::JoinStrictness> JoinUtil::getCrossJoinKindAndStrictness(substrait::CrossRel_JoinType join_type)
{
    switch (join_type)
    {
        case substrait::CrossRel_JoinType_JOIN_TYPE_INNER:
            return {DB::JoinKind::Cross, DB::JoinStrictness::All};
        case substrait::CrossRel_JoinType_JOIN_TYPE_LEFT:
            return {DB::JoinKind::Left, DB::JoinStrictness::All};
        case substrait::CrossRel_JoinType_JOIN_TYPE_OUTER:
            return {DB::JoinKind::Full, DB::JoinStrictness::All};
        default:
            throw Exception(ErrorCodes::UNKNOWN_TYPE, "unsupported join type {}.", magic_enum::enum_name(join_type));
    }
}

}