be/src/information_schema/schema_scanner.cpp

Source
// 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 "information_schema/schema_scanner.h"

#include <gen_cpp/Descriptors_types.h>
#include <gen_cpp/Types_types.h>
#include <glog/logging.h>
#include <string.h>

#include <new>
#include <ostream>
#include <utility>

#include "core/block/block.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column.h"
#include "core/column/column_complex.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/define_primitive_type.h"
#include "core/data_type/primitive_type.h"
#include "core/packed_int128.h"
#include "core/string_ref.h"
#include "core/types.h"
#include "core/value/hll.h"
#include "exec/pipeline/dependency.h"
#include "information_schema/schema_active_queries_scanner.h"
#include "information_schema/schema_authentication_integrations_scanner.h"
#include "information_schema/schema_backend_active_tasks.h"
#include "information_schema/schema_backend_configuration_scanner.h"
#include "information_schema/schema_backend_kerberos_ticket_cache.h"
#include "information_schema/schema_catalog_meta_cache_stats_scanner.h"
#include "information_schema/schema_charsets_scanner.h"
#include "information_schema/schema_cluster_snapshot_properties_scanner.h"
#include "information_schema/schema_cluster_snapshots_scanner.h"
#include "information_schema/schema_collations_scanner.h"
#include "information_schema/schema_column_data_sizes_scanner.h"
#include "information_schema/schema_columns_scanner.h"
#include "information_schema/schema_database_properties_scanner.h"
#include "information_schema/schema_dummy_scanner.h"
#include "information_schema/schema_encryption_keys_scanner.h"
#include "information_schema/schema_file_cache_info_scanner.h"
#include "information_schema/schema_file_cache_statistics.h"
#include "information_schema/schema_files_scanner.h"
#include "information_schema/schema_load_job_scanner.h"
#include "information_schema/schema_metadata_name_ids_scanner.h"
#include "information_schema/schema_partitions_scanner.h"
#include "information_schema/schema_processlist_scanner.h"
#include "information_schema/schema_profiling_scanner.h"
#include "information_schema/schema_routine_load_job_scanner.h"
#include "information_schema/schema_rowsets_scanner.h"
#include "information_schema/schema_schema_privileges_scanner.h"
#include "information_schema/schema_schemata_scanner.h"
#include "information_schema/schema_sql_block_rule_status_scanner.h"
#include "information_schema/schema_table_options_scanner.h"
#include "information_schema/schema_table_privileges_scanner.h"
#include "information_schema/schema_table_properties_scanner.h"
#include "information_schema/schema_tables_scanner.h"
#include "information_schema/schema_tablets_scanner.h"
#include "information_schema/schema_user_privileges_scanner.h"
#include "information_schema/schema_user_scanner.h"
#include "information_schema/schema_variables_scanner.h"
#include "information_schema/schema_view_dependency_scanner.h"
#include "information_schema/schema_views_scanner.h"
#include "information_schema/schema_workload_group_privileges.h"
#include "information_schema/schema_workload_group_resource_usage_scanner.h"
#include "information_schema/schema_workload_groups_scanner.h"
#include "information_schema/schema_workload_sched_policy_scanner.h"
#include "runtime/fragment_mgr.h"
#include "util/string_util.h"
#include "exprs/function/cast/cast_to_date_or_datetime_impl.hpp"

namespace doris {
class ObjectPool;

SchemaScanner::SchemaScanner(const std::vector<ColumnDesc>& columns, TSchemaTableType::type type)
        : _is_init(false), _columns(columns), _schema_table_type(type) {}

SchemaScanner::~SchemaScanner() = default;

Status SchemaScanner::start(RuntimeState* state) {
    if (!_is_init) {
        return Status::InternalError("call Start before Init.");
    }

    return Status::OK();
}

Status SchemaScanner::get_next_block(RuntimeState* state, Block* block, bool* eos) {
    if (_data_block == nullptr) {
        return Status::InternalError("No data left!");
    }
    DCHECK(_async_thread_running == false);
    RETURN_IF_ERROR(_scanner_status.status());
    for (size_t i = 0; i < block->columns(); i++) {
        std::move(*block->get_by_position(i).column)
                .mutate()
                ->insert_range_from(*_data_block->get_by_position(i).column, 0,
                                    _data_block->rows());
    }
    _data_block->clear_column_data();
    *eos = _eos;
    if (!*eos) {
        RETURN_IF_ERROR(get_next_block_async(state));
    }
    return Status::OK();
}

Status SchemaScanner::get_next_block_async(RuntimeState* state) {
    _dependency->block();
    auto task_ctx = state->get_task_execution_context();
    RETURN_IF_ERROR(ExecEnv::GetInstance()->fragment_mgr()->get_thread_pool()->submit_func(
            [this, task_ctx, state]() {
                auto task_lock = task_ctx.lock();
                if (task_lock == nullptr) {
                    return;
                }
                DCHECK(_async_thread_running == false);
                SCOPED_ATTACH_TASK(state);
                _async_thread_running = true;
                if (!_opened) {
                    _data_block = Block::create_unique();
                    _init_block(_data_block.get());
                    _scanner_status.update(start(state));
                    _opened = true;
                }
                bool eos = false;
                auto call_next_block_internal = [&]() -> Status {
                    RETURN_IF_CATCH_EXCEPTION(
                            { return get_next_block_internal(_data_block.get(), &eos); });
                };
                _scanner_status.update(call_next_block_internal());
                _eos = eos;
                _async_thread_running = false;
                _dependency->set_ready();
            }));
    return Status::OK();
}

Status SchemaScanner::init(RuntimeState* state, SchemaScannerParam* param, ObjectPool* pool) {
    if (_is_init) {
        return Status::OK();
    }
    if (nullptr == param || nullptr == pool) {
        return Status::InternalError("invalid parameter");
    }

    _param = param;
    _timezone = state->timezone();
    _timezone_obj = state->timezone_obj();
    _is_init = true;

    if (_param->profile) {
        _get_db_timer = ADD_TIMER(_param->profile, "GetDbTime");
        _get_table_timer = ADD_TIMER(_param->profile, "GetTableTime");
        _get_describe_timer = ADD_TIMER(_param->profile, "GetDescribeTime");
        _fill_block_timer = ADD_TIMER(_param->profile, "FillBlockTime");
    }

    return Status::OK();
}

std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type) {
    switch (type) {
    case TSchemaTableType::SCH_TABLES:
        return SchemaTablesScanner::create_unique();
    case TSchemaTableType::SCH_SCHEMATA:
        return SchemaSchemataScanner::create_unique();
    case TSchemaTableType::SCH_COLUMNS:
        return SchemaColumnsScanner::create_unique();
    case TSchemaTableType::SCH_CHARSETS:
        return SchemaCharsetsScanner::create_unique();
    case TSchemaTableType::SCH_COLLATIONS:
        return SchemaCollationsScanner::create_unique();
    case TSchemaTableType::SCH_GLOBAL_VARIABLES:
        return SchemaVariablesScanner::create_unique(TVarType::GLOBAL);
    case TSchemaTableType::SCH_SESSION_VARIABLES:
    case TSchemaTableType::SCH_VARIABLES:
        return SchemaVariablesScanner::create_unique(TVarType::SESSION);
    case TSchemaTableType::SCH_VIEWS:
        return SchemaViewsScanner::create_unique();
    case TSchemaTableType::SCH_TABLE_PRIVILEGES:
        return SchemaTablePrivilegesScanner::create_unique();
    case TSchemaTableType::SCH_SCHEMA_PRIVILEGES:
        return SchemaSchemaPrivilegesScanner::create_unique();
    case TSchemaTableType::SCH_USER_PRIVILEGES:
        return SchemaUserPrivilegesScanner::create_unique();
    case TSchemaTableType::SCH_FILES:
        return SchemaFilesScanner::create_unique();
    case TSchemaTableType::SCH_PARTITIONS:
        return SchemaPartitionsScanner::create_unique();
    case TSchemaTableType::SCH_BACKEND_CONFIGURATION:
        return SchemaBackendConfigurationScanner::create_unique();
    case TSchemaTableType::SCH_ROWSETS:
        return SchemaRowsetsScanner::create_unique();
    case TSchemaTableType::SCH_METADATA_NAME_IDS:
        return SchemaMetadataNameIdsScanner::create_unique();
    case TSchemaTableType::SCH_PROFILING:
        return SchemaProfilingScanner::create_unique();
    case TSchemaTableType::SCH_BACKEND_ACTIVE_TASKS:
        return SchemaBackendActiveTasksScanner::create_unique();
    case TSchemaTableType::SCH_ACTIVE_QUERIES:
        return SchemaActiveQueriesScanner::create_unique();
    case TSchemaTableType::SCH_WORKLOAD_GROUPS:
        return SchemaWorkloadGroupsScanner::create_unique();
    case TSchemaTableType::SCH_PROCESSLIST:
        return SchemaProcessListScanner::create_unique();
    case TSchemaTableType::SCH_USER:
        return SchemaUserScanner::create_unique();
    case TSchemaTableType::SCH_WORKLOAD_POLICY:
        return SchemaWorkloadSchedulePolicyScanner::create_unique();
    case TSchemaTableType::SCH_TABLE_OPTIONS:
        return SchemaTableOptionsScanner::create_unique();
    case TSchemaTableType::SCH_WORKLOAD_GROUP_PRIVILEGES:
        return SchemaWorkloadGroupPrivilegesScanner::create_unique();
    case TSchemaTableType::SCH_WORKLOAD_GROUP_RESOURCE_USAGE:
        return SchemaBackendWorkloadGroupResourceUsage::create_unique();
    case TSchemaTableType::SCH_TABLE_PROPERTIES:
        return SchemaTablePropertiesScanner::create_unique();
    case TSchemaTableType::SCH_DATABASE_PROPERTIES:
        return SchemaDatabasePropertiesScanner::create_unique();
    case TSchemaTableType::SCH_FILE_CACHE_STATISTICS:
        return SchemaFileCacheStatisticsScanner::create_unique();
    case TSchemaTableType::SCH_CATALOG_META_CACHE_STATISTICS:
        return SchemaCatalogMetaCacheStatsScanner::create_unique();
    case TSchemaTableType::SCH_BACKEND_KERBEROS_TICKET_CACHE:
        return SchemaBackendKerberosTicketCacheScanner::create_unique();
    case TSchemaTableType::SCH_ROUTINE_LOAD_JOBS:
        return SchemaRoutineLoadJobScanner::create_unique();
    case TSchemaTableType::SCH_LOAD_JOBS:
        return SchemaLoadJobScanner::create_unique();
    case TSchemaTableType::SCH_BACKEND_TABLETS:
        return SchemaTabletsScanner::create_unique();
    case TSchemaTableType::SCH_VIEW_DEPENDENCY:
        return SchemaViewDependencyScanner::create_unique();
    case TSchemaTableType::SCH_SQL_BLOCK_RULE_STATUS:
        return SchemaSqlBlockRuleStatusScanner::create_unique();
    case TSchemaTableType::SCH_ENCRYPTION_KEYS:
        return SchemaEncryptionKeysScanner::create_unique();
    case TSchemaTableType::SCH_CLUSTER_SNAPSHOTS:
        return SchemaClusterSnapshotsScanner::create_unique();
    case TSchemaTableType::SCH_CLUSTER_SNAPSHOT_PROPERTIES:
        return SchemaClusterSnapshotPropertiesScanner::create_unique();
    case TSchemaTableType::SCH_COLUMN_DATA_SIZES:
        return SchemaColumnDataSizesScanner::create_unique();
    case TSchemaTableType::SCH_FILE_CACHE_INFO:
        return SchemaFileCacheInfoScanner::create_unique();
    case TSchemaTableType::SCH_AUTHENTICATION_INTEGRATIONS:
        return SchemaAuthenticationIntegrationsScanner::create_unique();
    default:
        return SchemaDummyScanner::create_unique();
        break;
    }
}

void SchemaScanner::_init_block(Block* src_block) {
    const std::vector<SchemaScanner::ColumnDesc>& columns_desc(get_column_desc());
    for (int i = 0; i < columns_desc.size(); ++i) {
        auto data_type = DataTypeFactory::instance().create_data_type(columns_desc[i].type, true);
        src_block->insert(
                ColumnWithTypeAndName(data_type->create_column(), data_type, columns_desc[i].name));
    }
}

Status SchemaScanner::fill_dest_column_for_range(Block* block, size_t pos,
                                                 const std::vector<void*>& datas) {
    const ColumnDesc& col_desc = _columns[pos];
    MutableColumnPtr column_ptr;
    column_ptr = std::move(*block->get_by_position(pos).column).assume_mutable();
    IColumn* col_ptr = column_ptr.get();

    auto* nullable_column = reinterpret_cast<ColumnNullable*>(col_ptr);

    // Resize in advance to improve insertion efficiency.
    size_t fill_num = datas.size();
    col_ptr = &nullable_column->get_nested_column();
    for (int i = 0; i < fill_num; ++i) {
        auto* data = datas[i];
        if (data == nullptr) {
            // For nested column need not insert default.
            nullable_column->insert_data(nullptr, 0);
            continue;
        } else {
            nullable_column->push_false_to_nullmap(1);
        }
        switch (col_desc.type) {
        case TYPE_HLL: {
            auto* hll_slot = reinterpret_cast<HyperLogLog*>(data);
            assert_cast<ColumnHLL*>(col_ptr)->get_data().emplace_back(*hll_slot);
            break;
        }
        case TYPE_VARCHAR:
        case TYPE_CHAR:
        case TYPE_STRING: {
            auto* str_slot = reinterpret_cast<StringRef*>(data);
            assert_cast<ColumnString*>(col_ptr)->insert_data(str_slot->data, str_slot->size);
            break;
        }

        case TYPE_BOOLEAN: {
            uint8_t num = *reinterpret_cast<bool*>(data);
            assert_cast<ColumnBool*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_TINYINT: {
            int8_t num = *reinterpret_cast<int8_t*>(data);
            assert_cast<ColumnInt8*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_SMALLINT: {
            int16_t num = *reinterpret_cast<int16_t*>(data);
            assert_cast<ColumnInt16*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_INT: {
            int32_t num = *reinterpret_cast<int32_t*>(data);
            assert_cast<ColumnInt32*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_BIGINT: {
            int64_t num = *reinterpret_cast<int64_t*>(data);
            assert_cast<ColumnInt64*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_LARGEINT: {
            __int128 num;
            memcpy(&num, data, sizeof(__int128));
            assert_cast<ColumnInt128*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_FLOAT: {
            float num = *reinterpret_cast<float*>(data);
            assert_cast<ColumnFloat32*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_DOUBLE: {
            double num = *reinterpret_cast<double*>(data);
            assert_cast<ColumnFloat64*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_DATE: {
            assert_cast<ColumnDate*>(col_ptr)->insert_data(reinterpret_cast<char*>(data), 0);
            break;
        }

        case TYPE_DATEV2: {
            assert_cast<ColumnDateV2*>(col_ptr)->insert_value(
                    *reinterpret_cast<DateV2Value<DateV2ValueType>*>(data));
            break;
        }

        case TYPE_DATETIME: {
            assert_cast<ColumnDateTime*>(col_ptr)->insert_data(reinterpret_cast<char*>(data), 0);
            break;
        }

        case TYPE_DATETIMEV2: {
            assert_cast<ColumnDateTimeV2*>(col_ptr)->insert_value(
                    *reinterpret_cast<DateV2Value<DateTimeV2ValueType>*>(data));
            break;
        }

        case TYPE_TIMESTAMPTZ: {
            assert_cast<ColumnTimeStampTz*>(col_ptr)->insert_value(
                    *reinterpret_cast<TimestampTzValue*>(data));
            break;
        }

        case TYPE_DECIMALV2: {
            const Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
            assert_cast<ColumnDecimal128V2*>(col_ptr)->insert_data(
                    reinterpret_cast<const char*>(&num), 0);
            break;
        }
        case TYPE_DECIMAL128I: {
            const Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
            assert_cast<ColumnDecimal128V3*>(col_ptr)->insert_data(
                    reinterpret_cast<const char*>(&num), 0);
            break;
        }

        case TYPE_DECIMAL32: {
            const int32_t num = *reinterpret_cast<int32_t*>(data);
            assert_cast<ColumnDecimal32*>(col_ptr)->insert_data(reinterpret_cast<const char*>(&num),
                                                                0);
            break;
        }

        case TYPE_DECIMAL64: {
            const int64_t num = *reinterpret_cast<int64_t*>(data);
            assert_cast<ColumnDecimal64*>(col_ptr)->insert_data(reinterpret_cast<const char*>(&num),
                                                                0);
            break;
        }

        default: {
            DCHECK(false) << "bad slot type: " << col_desc.type;
            std::stringstream ss;
            ss << "Fail to convert schema type:'" << col_desc.type << " on column:`"
               << std::string(col_desc.name) + "`";
            return Status::InternalError(ss.str());
        }
        }
    }
    return Status::OK();
}

std::string SchemaScanner::get_db_from_full_name(const std::string& full_name) {
    std::vector<std::string> part = split(full_name, ".");
    if (part.size() == 2) {
        return part[1];
    }
    return full_name;
}

Status SchemaScanner::insert_block_column(TCell cell, int col_index, Block* block,
                                          PrimitiveType type) {
    MutableColumnPtr mutable_col_ptr;
    mutable_col_ptr = std::move(*block->get_by_position(col_index).column).assume_mutable();
    auto* nullable_column = reinterpret_cast<ColumnNullable*>(mutable_col_ptr.get());
    IColumn* col_ptr = &nullable_column->get_nested_column();

    switch (type) {
    case TYPE_BIGINT: {
        reinterpret_cast<ColumnInt64*>(col_ptr)->insert_value(cell.longVal);
        break;
    }

    case TYPE_INT: {
        reinterpret_cast<ColumnInt32*>(col_ptr)->insert_value(cell.intVal);
        break;
    }

    case TYPE_FLOAT: {
        assert_cast<ColumnFloat32*>(col_ptr)->insert_value(cell.doubleVal);
        break;
    }

    case TYPE_DOUBLE: {
        assert_cast<ColumnFloat64*>(col_ptr)->insert_value(cell.doubleVal);
        break;
    }

    case TYPE_BOOLEAN: {
        reinterpret_cast<ColumnUInt8*>(col_ptr)->insert_value(cell.boolVal);
        break;
    }

    case TYPE_STRING:
    case TYPE_VARCHAR:
    case TYPE_CHAR: {
        reinterpret_cast<ColumnString*>(col_ptr)->insert_data(cell.stringVal.data(),
                                                              cell.stringVal.size());
        break;
    }

    case TYPE_DATETIME: {
        std::vector<void*> datas(1);
        VecDateTimeValue src[1];
        CastParameters params;
        CastToDateOrDatetime::from_string_non_strict_mode<true>(
                {cell.stringVal.data(), cell.stringVal.size()}, src[0], nullptr, params);
        datas[0] = src;
        auto data = datas[0];
        reinterpret_cast<ColumnDateTime*>(col_ptr)->insert_data(reinterpret_cast<char*>(data), 0);
        break;
    }
    default: {
        std::stringstream ss;
        ss << "unsupported column type:" << type;
        return Status::InternalError(ss.str());
    }
    }
    nullable_column->push_false_to_nullmap(1);
    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-03-24 16:29

Line	Count	Source
1		// Licensed to the Apache Software Foundation (ASF) under one
2		// or more contributor license agreements. See the NOTICE file
3		// distributed with this work for additional information
4		// regarding copyright ownership. The ASF licenses this file
5		// to you under the Apache License, Version 2.0 (the
6		// "License"); you may not use this file except in compliance
7		// with the License. You may obtain a copy of the License at
8		//
9		// http://www.apache.org/licenses/LICENSE-2.0
10		//
11		// Unless required by applicable law or agreed to in writing,
12		// software distributed under the License is distributed on an
13		// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14		// KIND, either express or implied. See the License for the
15		// specific language governing permissions and limitations
16		// under the License.
17
18		#include "information_schema/schema_scanner.h"
19
20		#include <gen_cpp/Descriptors_types.h>
21		#include <gen_cpp/Types_types.h>
22		#include <glog/logging.h>
23		#include <string.h>
24
25		#include <new>
26		#include <ostream>
27		#include <utility>
28
29		#include "core/block/block.h"
30		#include "core/block/column_with_type_and_name.h"
31		#include "core/column/column.h"
32		#include "core/column/column_complex.h"
33		#include "core/column/column_nullable.h"
34		#include "core/column/column_string.h"
35		#include "core/column/column_vector.h"
36		#include "core/data_type/data_type.h"
37		#include "core/data_type/data_type_factory.hpp"
38		#include "core/data_type/define_primitive_type.h"
39		#include "core/data_type/primitive_type.h"
40		#include "core/packed_int128.h"
41		#include "core/string_ref.h"
42		#include "core/types.h"
43		#include "core/value/hll.h"
44		#include "exec/pipeline/dependency.h"
45		#include "information_schema/schema_active_queries_scanner.h"
46		#include "information_schema/schema_authentication_integrations_scanner.h"
47		#include "information_schema/schema_backend_active_tasks.h"
48		#include "information_schema/schema_backend_configuration_scanner.h"
49		#include "information_schema/schema_backend_kerberos_ticket_cache.h"
50		#include "information_schema/schema_catalog_meta_cache_stats_scanner.h"
51		#include "information_schema/schema_charsets_scanner.h"
52		#include "information_schema/schema_cluster_snapshot_properties_scanner.h"
53		#include "information_schema/schema_cluster_snapshots_scanner.h"
54		#include "information_schema/schema_collations_scanner.h"
55		#include "information_schema/schema_column_data_sizes_scanner.h"
56		#include "information_schema/schema_columns_scanner.h"
57		#include "information_schema/schema_database_properties_scanner.h"
58		#include "information_schema/schema_dummy_scanner.h"
59		#include "information_schema/schema_encryption_keys_scanner.h"
60		#include "information_schema/schema_file_cache_info_scanner.h"
61		#include "information_schema/schema_file_cache_statistics.h"
62		#include "information_schema/schema_files_scanner.h"
63		#include "information_schema/schema_load_job_scanner.h"
64		#include "information_schema/schema_metadata_name_ids_scanner.h"
65		#include "information_schema/schema_partitions_scanner.h"
66		#include "information_schema/schema_processlist_scanner.h"
67		#include "information_schema/schema_profiling_scanner.h"
68		#include "information_schema/schema_routine_load_job_scanner.h"
69		#include "information_schema/schema_rowsets_scanner.h"
70		#include "information_schema/schema_schema_privileges_scanner.h"
71		#include "information_schema/schema_schemata_scanner.h"
72		#include "information_schema/schema_sql_block_rule_status_scanner.h"
73		#include "information_schema/schema_table_options_scanner.h"
74		#include "information_schema/schema_table_privileges_scanner.h"
75		#include "information_schema/schema_table_properties_scanner.h"
76		#include "information_schema/schema_tables_scanner.h"
77		#include "information_schema/schema_tablets_scanner.h"
78		#include "information_schema/schema_user_privileges_scanner.h"
79		#include "information_schema/schema_user_scanner.h"
80		#include "information_schema/schema_variables_scanner.h"
81		#include "information_schema/schema_view_dependency_scanner.h"
82		#include "information_schema/schema_views_scanner.h"
83		#include "information_schema/schema_workload_group_privileges.h"
84		#include "information_schema/schema_workload_group_resource_usage_scanner.h"
85		#include "information_schema/schema_workload_groups_scanner.h"
86		#include "information_schema/schema_workload_sched_policy_scanner.h"
87		#include "runtime/fragment_mgr.h"
88		#include "util/string_util.h"
89		#include "exprs/function/cast/cast_to_date_or_datetime_impl.hpp"
90
91		namespace doris {
92		class ObjectPool;
93
94		SchemaScanner::SchemaScanner(const std::vector<ColumnDesc>& columns, TSchemaTableType::type type)
95	4.68k	: _is_init(false), _columns(columns), _schema_table_type(type) {}
96
97	4.68k	SchemaScanner::~SchemaScanner() = default;
98
99	6	Status SchemaScanner::start(RuntimeState* state) {
100	6	if (!_is_init) {
101	0	return Status::InternalError("call Start before Init.");
102	0	}
103
104	6	return Status::OK();
105	6	}
106
107	12.4k	Status SchemaScanner::get_next_block(RuntimeState* state, Block* block, bool* eos) {
108	12.4k	if (_data_block == nullptr) {
109	0	return Status::InternalError("No data left!");
110	0	}
111	12.4k	DCHECK(_async_thread_running == false);
112	12.4k	RETURN_IF_ERROR(_scanner_status.status());
113	211k	for (size_t i = 0; i < block->columns(); i++) {
114	199k	std::move(*block->get_by_position(i).column)
115	199k	.mutate()
116	199k	->insert_range_from(*_data_block->get_by_position(i).column, 0,
117	199k	_data_block->rows());
118	199k	}
119	12.4k	_data_block->clear_column_data();
120	12.4k	*eos = _eos;
121	12.4k	if (!*eos) {
122	10.1k	RETURN_IF_ERROR(get_next_block_async(state));
123	10.1k	}
124	12.4k	return Status::OK();
125	12.4k	}
126
127	12.4k	Status SchemaScanner::get_next_block_async(RuntimeState* state) {
128	12.4k	_dependency->block();
129	12.4k	auto task_ctx = state->get_task_execution_context();
130	12.4k	RETURN_IF_ERROR(ExecEnv::GetInstance()->fragment_mgr()->get_thread_pool()->submit_func(
131	12.4k	[this, task_ctx, state]() {
132	12.4k	auto task_lock = task_ctx.lock();
133	12.4k	if (task_lock == nullptr) {
134	12.4k	return;
135	12.4k	}
136	12.4k	DCHECK(_async_thread_running == false);
137	12.4k	SCOPED_ATTACH_TASK(state);
138	12.4k	_async_thread_running = true;
139	12.4k	if (!_opened) {
140	12.4k	_data_block = Block::create_unique();
141	12.4k	_init_block(_data_block.get());
142	12.4k	_scanner_status.update(start(state));
143	12.4k	_opened = true;
144	12.4k	}
145	12.4k	bool eos = false;
146	12.4k	auto call_next_block_internal = [&]() -> Status {
147	12.4k	RETURN_IF_CATCH_EXCEPTION(
148	12.4k	{ return get_next_block_internal(_data_block.get(), &eos); });
149	12.4k	};
150	12.4k	_scanner_status.update(call_next_block_internal());
151	12.4k	_eos = eos;
152	12.4k	_async_thread_running = false;
153	12.4k	_dependency->set_ready();
154	12.4k	}));
155	12.4k	return Status::OK();
156	12.4k	}
157
158	2.33k	Status SchemaScanner::init(RuntimeState* state, SchemaScannerParam* param, ObjectPool* pool) {
159	2.33k	if (_is_init) {
160	0	return Status::OK();
161	0	}
162	2.33k	if (nullptr == param \|\| nullptr == pool) {
163	0	return Status::InternalError("invalid parameter");
164	0	}
165
166	2.33k	_param = param;
167	2.33k	_timezone = state->timezone();
168	2.33k	_timezone_obj = state->timezone_obj();
169	2.33k	_is_init = true;
170
171	2.33k	if (_param->profile) {
172	2.33k	_get_db_timer = ADD_TIMER(_param->profile, "GetDbTime");
173	2.33k	_get_table_timer = ADD_TIMER(_param->profile, "GetTableTime");
174	2.33k	_get_describe_timer = ADD_TIMER(_param->profile, "GetDescribeTime");
175	2.33k	_fill_block_timer = ADD_TIMER(_param->profile, "FillBlockTime");
176	2.33k	}
177
178	2.33k	return Status::OK();
179	2.33k	}
180
181	4.67k	std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type) {
182	4.67k	switch (type) {
183	824	case TSchemaTableType::SCH_TABLES:
184	824	return SchemaTablesScanner::create_unique();
185	54	case TSchemaTableType::SCH_SCHEMATA:
186	54	return SchemaSchemataScanner::create_unique();
187	162	case TSchemaTableType::SCH_COLUMNS:
188	162	return SchemaColumnsScanner::create_unique();
189	6	case TSchemaTableType::SCH_CHARSETS:
190	6	return SchemaCharsetsScanner::create_unique();
191	6	case TSchemaTableType::SCH_COLLATIONS:
192	6	return SchemaCollationsScanner::create_unique();
193	6	case TSchemaTableType::SCH_GLOBAL_VARIABLES:
194	6	return SchemaVariablesScanner::create_unique(TVarType::GLOBAL);
195	70	case TSchemaTableType::SCH_SESSION_VARIABLES:
196	70	case TSchemaTableType::SCH_VARIABLES:
197	70	return SchemaVariablesScanner::create_unique(TVarType::SESSION);
198	16	case TSchemaTableType::SCH_VIEWS:
199	16	return SchemaViewsScanner::create_unique();
200	12	case TSchemaTableType::SCH_TABLE_PRIVILEGES:
201	12	return SchemaTablePrivilegesScanner::create_unique();
202	8	case TSchemaTableType::SCH_SCHEMA_PRIVILEGES:
203	8	return SchemaSchemaPrivilegesScanner::create_unique();
204	6	case TSchemaTableType::SCH_USER_PRIVILEGES:
205	6	return SchemaUserPrivilegesScanner::create_unique();
206	4	case TSchemaTableType::SCH_FILES:
207	4	return SchemaFilesScanner::create_unique();
208	42	case TSchemaTableType::SCH_PARTITIONS:
209	42	return SchemaPartitionsScanner::create_unique();
210	2	case TSchemaTableType::SCH_BACKEND_CONFIGURATION:
211	2	return SchemaBackendConfigurationScanner::create_unique();
212	18	case TSchemaTableType::SCH_ROWSETS:
213	18	return SchemaRowsetsScanner::create_unique();
214	14	case TSchemaTableType::SCH_METADATA_NAME_IDS:
215	14	return SchemaMetadataNameIdsScanner::create_unique();
216	4	case TSchemaTableType::SCH_PROFILING:
217	4	return SchemaProfilingScanner::create_unique();
218	1.21k	case TSchemaTableType::SCH_BACKEND_ACTIVE_TASKS:
219	1.21k	return SchemaBackendActiveTasksScanner::create_unique();
220	1.22k	case TSchemaTableType::SCH_ACTIVE_QUERIES:
221	1.22k	return SchemaActiveQueriesScanner::create_unique();
222	52	case TSchemaTableType::SCH_WORKLOAD_GROUPS:
223	52	return SchemaWorkloadGroupsScanner::create_unique();
224	20	case TSchemaTableType::SCH_PROCESSLIST:
225	20	return SchemaProcessListScanner::create_unique();
226	2	case TSchemaTableType::SCH_USER:
227	2	return SchemaUserScanner::create_unique();
228	16	case TSchemaTableType::SCH_WORKLOAD_POLICY:
229	16	return SchemaWorkloadSchedulePolicyScanner::create_unique();
230	12	case TSchemaTableType::SCH_TABLE_OPTIONS:
231	12	return SchemaTableOptionsScanner::create_unique();
232	26	case TSchemaTableType::SCH_WORKLOAD_GROUP_PRIVILEGES:
233	26	return SchemaWorkloadGroupPrivilegesScanner::create_unique();
234	6	case TSchemaTableType::SCH_WORKLOAD_GROUP_RESOURCE_USAGE:
235	6	return SchemaBackendWorkloadGroupResourceUsage::create_unique();
236	16	case TSchemaTableType::SCH_TABLE_PROPERTIES:
237	16	return SchemaTablePropertiesScanner::create_unique();
238	12	case TSchemaTableType::SCH_DATABASE_PROPERTIES:
239	12	return SchemaDatabasePropertiesScanner::create_unique();
240	118	case TSchemaTableType::SCH_FILE_CACHE_STATISTICS:
241	118	return SchemaFileCacheStatisticsScanner::create_unique();
242	26	case TSchemaTableType::SCH_CATALOG_META_CACHE_STATISTICS:
243	26	return SchemaCatalogMetaCacheStatsScanner::create_unique();
244	2	case TSchemaTableType::SCH_BACKEND_KERBEROS_TICKET_CACHE:
245	2	return SchemaBackendKerberosTicketCacheScanner::create_unique();
246	6	case TSchemaTableType::SCH_ROUTINE_LOAD_JOBS:
247	6	return SchemaRoutineLoadJobScanner::create_unique();
248	4	case TSchemaTableType::SCH_LOAD_JOBS:
249	4	return SchemaLoadJobScanner::create_unique();
250	2	case TSchemaTableType::SCH_BACKEND_TABLETS:
251	2	return SchemaTabletsScanner::create_unique();
252	16	case TSchemaTableType::SCH_VIEW_DEPENDENCY:
253	16	return SchemaViewDependencyScanner::create_unique();
254	6	case TSchemaTableType::SCH_SQL_BLOCK_RULE_STATUS:
255	6	return SchemaSqlBlockRuleStatusScanner::create_unique();
256	0	case TSchemaTableType::SCH_ENCRYPTION_KEYS:
257	0	return SchemaEncryptionKeysScanner::create_unique();
258	4	case TSchemaTableType::SCH_CLUSTER_SNAPSHOTS:
259	4	return SchemaClusterSnapshotsScanner::create_unique();
260	2	case TSchemaTableType::SCH_CLUSTER_SNAPSHOT_PROPERTIES:
261	2	return SchemaClusterSnapshotPropertiesScanner::create_unique();
262	6	case TSchemaTableType::SCH_COLUMN_DATA_SIZES:
263	6	return SchemaColumnDataSizesScanner::create_unique();
264	2	case TSchemaTableType::SCH_FILE_CACHE_INFO:
265	2	return SchemaFileCacheInfoScanner::create_unique();
266	4	case TSchemaTableType::SCH_AUTHENTICATION_INTEGRATIONS:
267	4	return SchemaAuthenticationIntegrationsScanner::create_unique();
268	620	default:
269	620	return SchemaDummyScanner::create_unique();
270	0	break;
271	4.67k	}
272	4.67k	}
273
274	2.34k	void SchemaScanner::_init_block(Block* src_block) {
275	2.34k	const std::vector<SchemaScanner::ColumnDesc>& columns_desc(get_column_desc());
276	32.2k	for (int i = 0; i < columns_desc.size(); ++i) {
277	29.9k	auto data_type = DataTypeFactory::instance().create_data_type(columns_desc[i].type, true);
278	29.9k	src_block->insert(
279	29.9k	ColumnWithTypeAndName(data_type->create_column(), data_type, columns_desc[i].name));
280	29.9k	}
281	2.34k	}
282
283		Status SchemaScanner::fill_dest_column_for_range(Block* block, size_t pos,
284	92.2k	const std::vector<void*>& datas) {
285	92.2k	const ColumnDesc& col_desc = _columns[pos];
286	92.2k	MutableColumnPtr column_ptr;
287	92.2k	column_ptr = std::move(*block->get_by_position(pos).column).assume_mutable();
288	92.2k	IColumn* col_ptr = column_ptr.get();
289
290	92.2k	auto* nullable_column = reinterpret_cast<ColumnNullable*>(col_ptr);
291
292		// Resize in advance to improve insertion efficiency.
293	92.2k	size_t fill_num = datas.size();
294	92.2k	col_ptr = &nullable_column->get_nested_column();
295	6.13M	for (int i = 0; i < fill_num; ++i) {
296	6.03M	auto* data = datas[i];
297	6.03M	if (data == nullptr) {
298		// For nested column need not insert default.
299	684k	nullable_column->insert_data(nullptr, 0);
300	684k	continue;
301	5.35M	} else {
302	5.35M	nullable_column->push_false_to_nullmap(1);
303	5.35M	}
304	5.35M	switch (col_desc.type) {
305	0	case TYPE_HLL: {
306	0	auto* hll_slot = reinterpret_cast<HyperLogLog*>(data);
307	0	assert_cast<ColumnHLL>(col_ptr)->get_data().emplace_back(hll_slot);
308	0	break;
309	0	}
310	1.62M	case TYPE_VARCHAR:
311	1.62M	case TYPE_CHAR:
312	1.69M	case TYPE_STRING: {
313	1.69M	auto* str_slot = reinterpret_cast<StringRef*>(data);
314	1.69M	assert_cast<ColumnString*>(col_ptr)->insert_data(str_slot->data, str_slot->size);
315	1.69M	break;
316	1.62M	}
317
318	874	case TYPE_BOOLEAN: {
319	874	uint8_t num = reinterpret_cast<bool>(data);
320	874	assert_cast<ColumnBool*>(col_ptr)->insert_value(num);
321	874	break;
322	1.62M	}
323
324	0	case TYPE_TINYINT: {
325	0	int8_t num = reinterpret_cast<int8_t>(data);
326	0	assert_cast<ColumnInt8*>(col_ptr)->insert_value(num);
327	0	break;
328	1.62M	}
329
330	0	case TYPE_SMALLINT: {
331	0	int16_t num = reinterpret_cast<int16_t>(data);
332	0	assert_cast<ColumnInt16*>(col_ptr)->insert_value(num);
333	0	break;
334	1.62M	}
335
336	325k	case TYPE_INT: {
337	325k	int32_t num = reinterpret_cast<int32_t>(data);
338	325k	assert_cast<ColumnInt32*>(col_ptr)->insert_value(num);
339	325k	break;
340	1.62M	}
341
342	3.06M	case TYPE_BIGINT: {
343	3.06M	int64_t num = reinterpret_cast<int64_t>(data);
344	3.06M	assert_cast<ColumnInt64*>(col_ptr)->insert_value(num);
345	3.06M	break;
346	1.62M	}
347
348	167	case TYPE_LARGEINT: {
349	167	__int128 num;
350	167	memcpy(&num, data, sizeof(__int128));
351	167	assert_cast<ColumnInt128*>(col_ptr)->insert_value(num);
352	167	break;
353	1.62M	}
354
355	0	case TYPE_FLOAT: {
356	0	float num = reinterpret_cast<float>(data);
357	0	assert_cast<ColumnFloat32*>(col_ptr)->insert_value(num);
358	0	break;
359	1.62M	}
360
361	0	case TYPE_DOUBLE: {
362	0	double num = reinterpret_cast<double>(data);
363	0	assert_cast<ColumnFloat64*>(col_ptr)->insert_value(num);
364	0	break;
365	1.62M	}
366
367	0	case TYPE_DATE: {
368	0	assert_cast<ColumnDate>(col_ptr)->insert_data(reinterpret_cast<char>(data), 0);
369	0	break;
370	1.62M	}
371
372	0	case TYPE_DATEV2: {
373	0	assert_cast<ColumnDateV2*>(col_ptr)->insert_value(
374	0	reinterpret_cast<DateV2Value<DateV2ValueType>>(data));
375	0	break;
376	1.62M	}
377
378	285k	case TYPE_DATETIME: {
379	285k	assert_cast<ColumnDateTime>(col_ptr)->insert_data(reinterpret_cast<char>(data), 0);
380	285k	break;
381	1.62M	}
382
383	171	case TYPE_DATETIMEV2: {
384	171	assert_cast<ColumnDateTimeV2*>(col_ptr)->insert_value(
385	171	reinterpret_cast<DateV2Value<DateTimeV2ValueType>>(data));
386	171	break;
387	1.62M	}
388
389	0	case TYPE_TIMESTAMPTZ: {
390	0	assert_cast<ColumnTimeStampTz*>(col_ptr)->insert_value(
391	0	reinterpret_cast<TimestampTzValue>(data));
392	0	break;
393	1.62M	}
394
395	0	case TYPE_DECIMALV2: {
396	0	const Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
397	0	assert_cast<ColumnDecimal128V2*>(col_ptr)->insert_data(
398	0	reinterpret_cast<const char*>(&num), 0);
399	0	break;
400	1.62M	}
401	0	case TYPE_DECIMAL128I: {
402	0	const Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
403	0	assert_cast<ColumnDecimal128V3*>(col_ptr)->insert_data(
404	0	reinterpret_cast<const char*>(&num), 0);
405	0	break;
406	1.62M	}
407
408	0	case TYPE_DECIMAL32: {
409	0	const int32_t num = reinterpret_cast<int32_t>(data);
410	0	assert_cast<ColumnDecimal32>(col_ptr)->insert_data(reinterpret_cast<const char>(&num),
411	0	0);
412	0	break;
413	1.62M	}
414
415	0	case TYPE_DECIMAL64: {
416	0	const int64_t num = reinterpret_cast<int64_t>(data);
417	0	assert_cast<ColumnDecimal64>(col_ptr)->insert_data(reinterpret_cast<const char>(&num),
418	0	0);
419	0	break;
420	1.62M	}
421
422	0	default: {
423	0	DCHECK(false) << "bad slot type: " << col_desc.type;
424	0	std::stringstream ss;
425	0	ss << "Fail to convert schema type:'" << col_desc.type << " on column:`"
426	0	<< std::string(col_desc.name) + "`";
427	0	return Status::InternalError(ss.str());
428	1.62M	}
429	5.35M	}
430	5.35M	}
431	96.7k	return Status::OK();
432	92.2k	}
433
434	0	std::string SchemaScanner::get_db_from_full_name(const std::string& full_name) {
435	0	std::vector<std::string> part = split(full_name, ".");
436	0	if (part.size() == 2) {
437	0	return part[1];
438	0	}
439	0	return full_name;
440	0	}
441
442		Status SchemaScanner::insert_block_column(TCell cell, int col_index, Block* block,
443	1.71M	PrimitiveType type) {
444	1.71M	MutableColumnPtr mutable_col_ptr;
445	1.71M	mutable_col_ptr = std::move(*block->get_by_position(col_index).column).assume_mutable();
446	1.71M	auto* nullable_column = reinterpret_cast<ColumnNullable*>(mutable_col_ptr.get());
447	1.71M	IColumn* col_ptr = &nullable_column->get_nested_column();
448
449	1.71M	switch (type) {
450	223k	case TYPE_BIGINT: {
451	223k	reinterpret_cast<ColumnInt64*>(col_ptr)->insert_value(cell.longVal);
452	223k	break;
453	0	}
454
455	70.1k	case TYPE_INT: {
456	70.1k	reinterpret_cast<ColumnInt32*>(col_ptr)->insert_value(cell.intVal);
457	70.1k	break;
458	0	}
459
460	0	case TYPE_FLOAT: {
461	0	assert_cast<ColumnFloat32*>(col_ptr)->insert_value(cell.doubleVal);
462	0	break;
463	0	}
464
465	11.6k	case TYPE_DOUBLE: {
466	11.6k	assert_cast<ColumnFloat64*>(col_ptr)->insert_value(cell.doubleVal);
467	11.6k	break;
468	0	}
469
470	17.4k	case TYPE_BOOLEAN: {
471	17.4k	reinterpret_cast<ColumnUInt8*>(col_ptr)->insert_value(cell.boolVal);
472	17.4k	break;
473	0	}
474
475	614k	case TYPE_STRING:
476	1.37M	case TYPE_VARCHAR:
477	1.37M	case TYPE_CHAR: {
478	1.37M	reinterpret_cast<ColumnString*>(col_ptr)->insert_data(cell.stringVal.data(),
479	1.37M	cell.stringVal.size());
480	1.37M	break;
481	1.37M	}
482
483	26.3k	case TYPE_DATETIME: {
484	26.3k	std::vector<void*> datas(1);
485	26.3k	VecDateTimeValue src[1];
486	26.3k	CastParameters params;
487	26.3k	CastToDateOrDatetime::from_string_non_strict_mode<true>(
488	26.3k	{cell.stringVal.data(), cell.stringVal.size()}, src[0], nullptr, params);
489	26.3k	datas[0] = src;
490	26.3k	auto data = datas[0];
491	26.3k	reinterpret_cast<ColumnDateTime>(col_ptr)->insert_data(reinterpret_cast<char>(data), 0);
492	26.3k	break;
493	1.37M	}
494	0	default: {
495	0	std::stringstream ss;
496	0	ss << "unsupported column type:" << type;
497	0	return Status::InternalError(ss.str());
498	1.37M	}
499	1.71M	}
500	1.71M	nullable_column->push_false_to_nullmap(1);
501	1.71M	return Status::OK();
502	1.71M	}
503
504		} // namespace doris