/root/doris/be/src/exec/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 "exec/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 "exec/schema_scanner/schema_active_queries_scanner.h"
#include "exec/schema_scanner/schema_backend_active_tasks.h"
#include "exec/schema_scanner/schema_backend_configuration_scanner.h"
#include "exec/schema_scanner/schema_backend_kerberos_ticket_cache.h"
#include "exec/schema_scanner/schema_catalog_meta_cache_stats_scanner.h"
#include "exec/schema_scanner/schema_charsets_scanner.h"
#include "exec/schema_scanner/schema_collations_scanner.h"
#include "exec/schema_scanner/schema_columns_scanner.h"
#include "exec/schema_scanner/schema_dummy_scanner.h"
#include "exec/schema_scanner/schema_encryption_keys_scanner.h"
#include "exec/schema_scanner/schema_file_cache_statistics.h"
#include "exec/schema_scanner/schema_files_scanner.h"
#include "exec/schema_scanner/schema_metadata_name_ids_scanner.h"
#include "exec/schema_scanner/schema_partitions_scanner.h"
#include "exec/schema_scanner/schema_processlist_scanner.h"
#include "exec/schema_scanner/schema_profiling_scanner.h"
#include "exec/schema_scanner/schema_routine_load_job_scanner.h"
#include "exec/schema_scanner/schema_routine_scanner.h"
#include "exec/schema_scanner/schema_rowsets_scanner.h"
#include "exec/schema_scanner/schema_schema_privileges_scanner.h"
#include "exec/schema_scanner/schema_schemata_scanner.h"
#include "exec/schema_scanner/schema_table_options_scanner.h"
#include "exec/schema_scanner/schema_table_privileges_scanner.h"
#include "exec/schema_scanner/schema_table_properties_scanner.h"
#include "exec/schema_scanner/schema_tables_scanner.h"
#include "exec/schema_scanner/schema_tablets_scanner.h"
#include "exec/schema_scanner/schema_user_privileges_scanner.h"
#include "exec/schema_scanner/schema_user_scanner.h"
#include "exec/schema_scanner/schema_variables_scanner.h"
#include "exec/schema_scanner/schema_view_dependency_scanner.h"
#include "exec/schema_scanner/schema_views_scanner.h"
#include "exec/schema_scanner/schema_workload_group_privileges.h"
#include "exec/schema_scanner/schema_workload_group_resource_usage_scanner.h"
#include "exec/schema_scanner/schema_workload_groups_scanner.h"
#include "exec/schema_scanner/schema_workload_sched_policy_scanner.h"
#include "olap/hll.h"
#include "pipeline/dependency.h"
#include "runtime/define_primitive_type.h"
#include "runtime/fragment_mgr.h"
#include "runtime/types.h"
#include "util/string_util.h"
#include "util/types.h"
#include "vec/columns/column.h"
#include "vec/columns/column_complex.h"
#include "vec/columns/column_nullable.h"
#include "vec/columns/column_string.h"
#include "vec/columns/column_vector.h"
#include "vec/common/string_ref.h"
#include "vec/core/block.h"
#include "vec/core/column_with_type_and_name.h"
#include "vec/core/types.h"
#include "vec/data_types/data_type.h"
#include "vec/data_types/data_type_factory.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, vectorized::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 = vectorized::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_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_PROCEDURES:
        return SchemaRoutinesScanner::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_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_BACKEND_TABLETS:
        return SchemaTabletsScanner::create_unique();
    case TSchemaTableType::SCH_VIEW_DEPENDENCY:
        return SchemaViewDependencyScanner::create_unique();
    case TSchemaTableType::SCH_ENCRYPTION_KEYS:
        return SchemaEncryptionKeysScanner::create_unique();
    default:
        return SchemaDummyScanner::create_unique();
        break;
    }
}

void SchemaScanner::_init_block(vectorized::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 = vectorized::DataTypeFactory::instance().create_data_type(
                columns_desc[i].type, true);
        src_block->insert(vectorized::ColumnWithTypeAndName(data_type->create_column(), data_type,
                                                            columns_desc[i].name));
    }
}

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

    auto* nullable_column = reinterpret_cast<vectorized::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<vectorized::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<vectorized::ColumnString*>(col_ptr)->insert_data(str_slot->data,
                                                                         str_slot->size);
            break;
        }

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

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

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

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

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

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

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

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

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

        case TYPE_DATEV2: {
            uint32_t num = *reinterpret_cast<uint32_t*>(data);
            assert_cast<vectorized::ColumnDateV2*>(col_ptr)->insert_value(num);
            break;
        }

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

        case TYPE_DATETIMEV2: {
            uint64_t num = *reinterpret_cast<uint64_t*>(data);
            assert_cast<vectorized::ColumnDateTimeV2*>(col_ptr)->insert_value(num);
            break;
        }

        case TYPE_DECIMALV2: {
            const vectorized::Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
            assert_cast<vectorized::ColumnDecimal128V2*>(col_ptr)->insert_data(
                    reinterpret_cast<const char*>(&num), 0);
            break;
        }
        case TYPE_DECIMAL128I: {
            const vectorized::Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
            assert_cast<vectorized::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<vectorized::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<vectorized::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, vectorized::Block* block,
                                          PrimitiveType type) {
    vectorized::MutableColumnPtr mutable_col_ptr;
    mutable_col_ptr = std::move(*block->get_by_position(col_index).column).assume_mutable();
    auto* nullable_column = reinterpret_cast<vectorized::ColumnNullable*>(mutable_col_ptr.get());
    vectorized::IColumn* col_ptr = &nullable_column->get_nested_column();

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

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

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

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

    case TYPE_DATETIME: {
        std::vector<void*> datas(1);
        VecDateTimeValue src[1];
        src[0].from_date_str(cell.stringVal.data(), cell.stringVal.size());
        datas[0] = src;
        auto data = datas[0];
        reinterpret_cast<vectorized::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: 2025-09-11 18:52

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 "exec/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 "exec/schema_scanner/schema_active_queries_scanner.h"
30		#include "exec/schema_scanner/schema_backend_active_tasks.h"
31		#include "exec/schema_scanner/schema_backend_configuration_scanner.h"
32		#include "exec/schema_scanner/schema_backend_kerberos_ticket_cache.h"
33		#include "exec/schema_scanner/schema_catalog_meta_cache_stats_scanner.h"
34		#include "exec/schema_scanner/schema_charsets_scanner.h"
35		#include "exec/schema_scanner/schema_collations_scanner.h"
36		#include "exec/schema_scanner/schema_columns_scanner.h"
37		#include "exec/schema_scanner/schema_dummy_scanner.h"
38		#include "exec/schema_scanner/schema_encryption_keys_scanner.h"
39		#include "exec/schema_scanner/schema_file_cache_statistics.h"
40		#include "exec/schema_scanner/schema_files_scanner.h"
41		#include "exec/schema_scanner/schema_metadata_name_ids_scanner.h"
42		#include "exec/schema_scanner/schema_partitions_scanner.h"
43		#include "exec/schema_scanner/schema_processlist_scanner.h"
44		#include "exec/schema_scanner/schema_profiling_scanner.h"
45		#include "exec/schema_scanner/schema_routine_load_job_scanner.h"
46		#include "exec/schema_scanner/schema_routine_scanner.h"
47		#include "exec/schema_scanner/schema_rowsets_scanner.h"
48		#include "exec/schema_scanner/schema_schema_privileges_scanner.h"
49		#include "exec/schema_scanner/schema_schemata_scanner.h"
50		#include "exec/schema_scanner/schema_table_options_scanner.h"
51		#include "exec/schema_scanner/schema_table_privileges_scanner.h"
52		#include "exec/schema_scanner/schema_table_properties_scanner.h"
53		#include "exec/schema_scanner/schema_tables_scanner.h"
54		#include "exec/schema_scanner/schema_tablets_scanner.h"
55		#include "exec/schema_scanner/schema_user_privileges_scanner.h"
56		#include "exec/schema_scanner/schema_user_scanner.h"
57		#include "exec/schema_scanner/schema_variables_scanner.h"
58		#include "exec/schema_scanner/schema_view_dependency_scanner.h"
59		#include "exec/schema_scanner/schema_views_scanner.h"
60		#include "exec/schema_scanner/schema_workload_group_privileges.h"
61		#include "exec/schema_scanner/schema_workload_group_resource_usage_scanner.h"
62		#include "exec/schema_scanner/schema_workload_groups_scanner.h"
63		#include "exec/schema_scanner/schema_workload_sched_policy_scanner.h"
64		#include "olap/hll.h"
65		#include "pipeline/dependency.h"
66		#include "runtime/define_primitive_type.h"
67		#include "runtime/fragment_mgr.h"
68		#include "runtime/types.h"
69		#include "util/string_util.h"
70		#include "util/types.h"
71		#include "vec/columns/column.h"
72		#include "vec/columns/column_complex.h"
73		#include "vec/columns/column_nullable.h"
74		#include "vec/columns/column_string.h"
75		#include "vec/columns/column_vector.h"
76		#include "vec/common/string_ref.h"
77		#include "vec/core/block.h"
78		#include "vec/core/column_with_type_and_name.h"
79		#include "vec/core/types.h"
80		#include "vec/data_types/data_type.h"
81		#include "vec/data_types/data_type_factory.hpp"
82
83		namespace doris {
84		class ObjectPool;
85
86		SchemaScanner::SchemaScanner(const std::vector<ColumnDesc>& columns, TSchemaTableType::type type)
87	1	: _is_init(false), _columns(columns), _schema_table_type(type) {}
88
89	1	SchemaScanner::~SchemaScanner() = default;
90
91	0	Status SchemaScanner::start(RuntimeState* state) {
92	0	if (!_is_init) {
93	0	return Status::InternalError("call Start before Init.");
94	0	}
95
96	0	return Status::OK();
97	0	}
98
99	0	Status SchemaScanner::get_next_block(RuntimeState* state, vectorized::Block* block, bool* eos) {
100	0	if (_data_block == nullptr) {
101	0	return Status::InternalError("No data left!");
102	0	}
103	0	DCHECK(_async_thread_running == false);
104	0	RETURN_IF_ERROR(_scanner_status.status());
105	0	for (size_t i = 0; i < block->columns(); i++) {
106	0	std::move(*block->get_by_position(i).column)
107	0	.mutate()
108	0	->insert_range_from(*_data_block->get_by_position(i).column, 0,
109	0	_data_block->rows());
110	0	}
111	0	_data_block->clear_column_data();
112	0	*eos = _eos;
113	0	if (!*eos) {
114	0	RETURN_IF_ERROR(get_next_block_async(state));
115	0	}
116	0	return Status::OK();
117	0	}
118
119	0	Status SchemaScanner::get_next_block_async(RuntimeState* state) {
120	0	_dependency->block();
121	0	auto task_ctx = state->get_task_execution_context();
122	0	RETURN_IF_ERROR(ExecEnv::GetInstance()->fragment_mgr()->get_thread_pool()->submit_func(
123	0	[this, task_ctx, state]() {
124	0	auto task_lock = task_ctx.lock();
125	0	if (task_lock == nullptr) {
126	0	return;
127	0	}
128	0	DCHECK(_async_thread_running == false);
129	0	SCOPED_ATTACH_TASK(state);
130	0	_async_thread_running = true;
131	0	if (!_opened) {
132	0	_data_block = vectorized::Block::create_unique();
133	0	_init_block(_data_block.get());
134	0	_scanner_status.update(start(state));
135	0	_opened = true;
136	0	}
137	0	bool eos = false;
138	0	auto call_next_block_internal = [&]() -> Status {
139	0	RETURN_IF_CATCH_EXCEPTION(
140	0	{ return get_next_block_internal(_data_block.get(), &eos); });
141	0	};
142	0	_scanner_status.update(call_next_block_internal());
143	0	_eos = eos;
144	0	_async_thread_running = false;
145	0	_dependency->set_ready();
146	0	}));
147	0	return Status::OK();
148	0	}
149
150	0	Status SchemaScanner::init(RuntimeState* state, SchemaScannerParam* param, ObjectPool* pool) {
151	0	if (_is_init) {
152	0	return Status::OK();
153	0	}
154	0	if (nullptr == param \|\| nullptr == pool) {
155	0	return Status::InternalError("invalid parameter");
156	0	}
157
158	0	_param = param;
159	0	_timezone_obj = state->timezone_obj();
160	0	_is_init = true;
161
162	0	if (_param->profile) {
163	0	_get_db_timer = ADD_TIMER(_param->profile, "GetDbTime");
164	0	_get_table_timer = ADD_TIMER(_param->profile, "GetTableTime");
165	0	_get_describe_timer = ADD_TIMER(_param->profile, "GetDescribeTime");
166	0	_fill_block_timer = ADD_TIMER(_param->profile, "FillBlockTime");
167	0	}
168
169	0	return Status::OK();
170	0	}
171
172	0	std::unique_ptr<SchemaScanner> SchemaScanner::create(TSchemaTableType::type type) {
173	0	switch (type) {
174	0	case TSchemaTableType::SCH_TABLES:
175	0	return SchemaTablesScanner::create_unique();
176	0	case TSchemaTableType::SCH_SCHEMATA:
177	0	return SchemaSchemataScanner::create_unique();
178	0	case TSchemaTableType::SCH_COLUMNS:
179	0	return SchemaColumnsScanner::create_unique();
180	0	case TSchemaTableType::SCH_CHARSETS:
181	0	return SchemaCharsetsScanner::create_unique();
182	0	case TSchemaTableType::SCH_COLLATIONS:
183	0	return SchemaCollationsScanner::create_unique();
184	0	case TSchemaTableType::SCH_GLOBAL_VARIABLES:
185	0	return SchemaVariablesScanner::create_unique(TVarType::GLOBAL);
186	0	case TSchemaTableType::SCH_SESSION_VARIABLES:
187	0	case TSchemaTableType::SCH_VARIABLES:
188	0	return SchemaVariablesScanner::create_unique(TVarType::SESSION);
189	0	case TSchemaTableType::SCH_VIEWS:
190	0	return SchemaViewsScanner::create_unique();
191	0	case TSchemaTableType::SCH_TABLE_PRIVILEGES:
192	0	return SchemaTablePrivilegesScanner::create_unique();
193	0	case TSchemaTableType::SCH_SCHEMA_PRIVILEGES:
194	0	return SchemaSchemaPrivilegesScanner::create_unique();
195	0	case TSchemaTableType::SCH_USER_PRIVILEGES:
196	0	return SchemaUserPrivilegesScanner::create_unique();
197	0	case TSchemaTableType::SCH_FILES:
198	0	return SchemaFilesScanner::create_unique();
199	0	case TSchemaTableType::SCH_PARTITIONS:
200	0	return SchemaPartitionsScanner::create_unique();
201	0	case TSchemaTableType::SCH_BACKEND_CONFIGURATION:
202	0	return SchemaBackendConfigurationScanner::create_unique();
203	0	case TSchemaTableType::SCH_ROWSETS:
204	0	return SchemaRowsetsScanner::create_unique();
205	0	case TSchemaTableType::SCH_METADATA_NAME_IDS:
206	0	return SchemaMetadataNameIdsScanner::create_unique();
207	0	case TSchemaTableType::SCH_PROFILING:
208	0	return SchemaProfilingScanner::create_unique();
209	0	case TSchemaTableType::SCH_BACKEND_ACTIVE_TASKS:
210	0	return SchemaBackendActiveTasksScanner::create_unique();
211	0	case TSchemaTableType::SCH_ACTIVE_QUERIES:
212	0	return SchemaActiveQueriesScanner::create_unique();
213	0	case TSchemaTableType::SCH_WORKLOAD_GROUPS:
214	0	return SchemaWorkloadGroupsScanner::create_unique();
215	0	case TSchemaTableType::SCH_PROCESSLIST:
216	0	return SchemaProcessListScanner::create_unique();
217	0	case TSchemaTableType::SCH_PROCEDURES:
218	0	return SchemaRoutinesScanner::create_unique();
219	0	case TSchemaTableType::SCH_USER:
220	0	return SchemaUserScanner::create_unique();
221	0	case TSchemaTableType::SCH_WORKLOAD_POLICY:
222	0	return SchemaWorkloadSchedulePolicyScanner::create_unique();
223	0	case TSchemaTableType::SCH_TABLE_OPTIONS:
224	0	return SchemaTableOptionsScanner::create_unique();
225	0	case TSchemaTableType::SCH_WORKLOAD_GROUP_PRIVILEGES:
226	0	return SchemaWorkloadGroupPrivilegesScanner::create_unique();
227	0	case TSchemaTableType::SCH_WORKLOAD_GROUP_RESOURCE_USAGE:
228	0	return SchemaBackendWorkloadGroupResourceUsage::create_unique();
229	0	case TSchemaTableType::SCH_TABLE_PROPERTIES:
230	0	return SchemaTablePropertiesScanner::create_unique();
231	0	case TSchemaTableType::SCH_FILE_CACHE_STATISTICS:
232	0	return SchemaFileCacheStatisticsScanner::create_unique();
233	0	case TSchemaTableType::SCH_CATALOG_META_CACHE_STATISTICS:
234	0	return SchemaCatalogMetaCacheStatsScanner::create_unique();
235	0	case TSchemaTableType::SCH_BACKEND_KERBEROS_TICKET_CACHE:
236	0	return SchemaBackendKerberosTicketCacheScanner::create_unique();
237	0	case TSchemaTableType::SCH_ROUTINE_LOAD_JOBS:
238	0	return SchemaRoutineLoadJobScanner::create_unique();
239	0	case TSchemaTableType::SCH_BACKEND_TABLETS:
240	0	return SchemaTabletsScanner::create_unique();
241	0	case TSchemaTableType::SCH_VIEW_DEPENDENCY:
242	0	return SchemaViewDependencyScanner::create_unique();
243	0	case TSchemaTableType::SCH_ENCRYPTION_KEYS:
244	0	return SchemaEncryptionKeysScanner::create_unique();
245	0	default:
246	0	return SchemaDummyScanner::create_unique();
247	0	break;
248	0	}
249	0	}
250
251	1	void SchemaScanner::_init_block(vectorized::Block* src_block) {
252	1	const std::vector<SchemaScanner::ColumnDesc>& columns_desc(get_column_desc());
253	12	for (int i = 0; i < columns_desc.size(); ++i) {
254	11	auto data_type = vectorized::DataTypeFactory::instance().create_data_type(
255	11	columns_desc[i].type, true);
256	11	src_block->insert(vectorized::ColumnWithTypeAndName(data_type->create_column(), data_type,
257	11	columns_desc[i].name));
258	11	}
259	1	}
260
261		Status SchemaScanner::fill_dest_column_for_range(vectorized::Block* block, size_t pos,
262	11	const std::vector<void*>& datas) {
263	11	const ColumnDesc& col_desc = _columns[pos];
264	11	vectorized::MutableColumnPtr column_ptr;
265	11	column_ptr = std::move(*block->get_by_position(pos).column).assume_mutable();
266	11	vectorized::IColumn* col_ptr = column_ptr.get();
267
268	11	auto* nullable_column = reinterpret_cast<vectorized::ColumnNullable*>(col_ptr);
269
270		// Resize in advance to improve insertion efficiency.
271	11	size_t fill_num = datas.size();
272	11	col_ptr = &nullable_column->get_nested_column();
273	22	for (int i = 0; i < fill_num; ++i) {
274	11	auto* data = datas[i];
275	11	if (data == nullptr) {
276		// For nested column need not insert default.
277	0	nullable_column->insert_data(nullptr, 0);
278	0	continue;
279	11	} else {
280	11	nullable_column->push_false_to_nullmap(1);
281	11	}
282	11	switch (col_desc.type) {
283	0	case TYPE_HLL: {
284	0	auto* hll_slot = reinterpret_cast<HyperLogLog*>(data);
285	0	assert_cast<vectorized::ColumnHLL>(col_ptr)->get_data().emplace_back(hll_slot);
286	0	break;
287	0	}
288	0	case TYPE_VARCHAR:
289	0	case TYPE_CHAR:
290	6	case TYPE_STRING: {
291	6	auto* str_slot = reinterpret_cast<StringRef*>(data);
292	6	assert_cast<vectorized::ColumnString*>(col_ptr)->insert_data(str_slot->data,
293	6	str_slot->size);
294	6	break;
295	0	}
296
297	0	case TYPE_BOOLEAN: {
298	0	uint8_t num = reinterpret_cast<bool>(data);
299	0	assert_cast<vectorized::ColumnBool*>(col_ptr)->insert_value(num);
300	0	break;
301	0	}
302
303	0	case TYPE_TINYINT: {
304	0	int8_t num = reinterpret_cast<int8_t>(data);
305	0	assert_cast<vectorized::ColumnInt8*>(col_ptr)->insert_value(num);
306	0	break;
307	0	}
308
309	0	case TYPE_SMALLINT: {
310	0	int16_t num = reinterpret_cast<int16_t>(data);
311	0	assert_cast<vectorized::ColumnInt16*>(col_ptr)->insert_value(num);
312	0	break;
313	0	}
314
315	2	case TYPE_INT: {
316	2	int32_t num = reinterpret_cast<int32_t>(data);
317	2	assert_cast<vectorized::ColumnInt32*>(col_ptr)->insert_value(num);
318	2	break;
319	0	}
320
321	1	case TYPE_BIGINT: {
322	1	int64_t num = reinterpret_cast<int64_t>(data);
323	1	assert_cast<vectorized::ColumnInt64*>(col_ptr)->insert_value(num);
324	1	break;
325	0	}
326
327	0	case TYPE_LARGEINT: {
328	0	__int128 num;
329	0	memcpy(&num, data, sizeof(__int128));
330	0	assert_cast<vectorized::ColumnInt128*>(col_ptr)->insert_value(num);
331	0	break;
332	0	}
333
334	0	case TYPE_FLOAT: {
335	0	float num = reinterpret_cast<float>(data);
336	0	assert_cast<vectorized::ColumnFloat32*>(col_ptr)->insert_value(num);
337	0	break;
338	0	}
339
340	0	case TYPE_DOUBLE: {
341	0	double num = reinterpret_cast<double>(data);
342	0	assert_cast<vectorized::ColumnFloat64*>(col_ptr)->insert_value(num);
343	0	break;
344	0	}
345
346	0	case TYPE_DATE: {
347	0	assert_cast<vectorized::ColumnDate*>(col_ptr)->insert_data(
348	0	reinterpret_cast<char*>(data), 0);
349	0	break;
350	0	}
351
352	0	case TYPE_DATEV2: {
353	0	uint32_t num = reinterpret_cast<uint32_t>(data);
354	0	assert_cast<vectorized::ColumnDateV2*>(col_ptr)->insert_value(num);
355	0	break;
356	0	}
357
358	0	case TYPE_DATETIME: {
359	0	assert_cast<vectorized::ColumnDateTime*>(col_ptr)->insert_data(
360	0	reinterpret_cast<char*>(data), 0);
361	0	break;
362	0	}
363
364	2	case TYPE_DATETIMEV2: {
365	2	uint64_t num = reinterpret_cast<uint64_t>(data);
366	2	assert_cast<vectorized::ColumnDateTimeV2*>(col_ptr)->insert_value(num);
367	2	break;
368	0	}
369
370	0	case TYPE_DECIMALV2: {
371	0	const vectorized::Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
372	0	assert_cast<vectorized::ColumnDecimal128V2*>(col_ptr)->insert_data(
373	0	reinterpret_cast<const char*>(&num), 0);
374	0	break;
375	0	}
376	0	case TYPE_DECIMAL128I: {
377	0	const vectorized::Int128 num = (reinterpret_cast<PackedInt128*>(data))->value;
378	0	assert_cast<vectorized::ColumnDecimal128V3*>(col_ptr)->insert_data(
379	0	reinterpret_cast<const char*>(&num), 0);
380	0	break;
381	0	}
382
383	0	case TYPE_DECIMAL32: {
384	0	const int32_t num = reinterpret_cast<int32_t>(data);
385	0	assert_cast<vectorized::ColumnDecimal32*>(col_ptr)->insert_data(
386	0	reinterpret_cast<const char*>(&num), 0);
387	0	break;
388	0	}
389
390	0	case TYPE_DECIMAL64: {
391	0	const int64_t num = reinterpret_cast<int64_t>(data);
392	0	assert_cast<vectorized::ColumnDecimal64*>(col_ptr)->insert_data(
393	0	reinterpret_cast<const char*>(&num), 0);
394	0	break;
395	0	}
396
397	0	default: {
398	0	DCHECK(false) << "bad slot type: " << col_desc.type;
399	0	std::stringstream ss;
400	0	ss << "Fail to convert schema type:'" << col_desc.type << " on column:`"
401	0	<< std::string(col_desc.name) + "`";
402	0	return Status::InternalError(ss.str());
403	0	}
404	11	}
405	11	}
406	11	return Status::OK();
407	11	}
408
409	0	std::string SchemaScanner::get_db_from_full_name(const std::string& full_name) {
410	0	std::vector<std::string> part = split(full_name, ".");
411	0	if (part.size() == 2) {
412	0	return part[1];
413	0	}
414	0	return full_name;
415	0	}
416
417		Status SchemaScanner::insert_block_column(TCell cell, int col_index, vectorized::Block* block,
418	0	PrimitiveType type) {
419	0	vectorized::MutableColumnPtr mutable_col_ptr;
420	0	mutable_col_ptr = std::move(*block->get_by_position(col_index).column).assume_mutable();
421	0	auto* nullable_column = reinterpret_cast<vectorized::ColumnNullable*>(mutable_col_ptr.get());
422	0	vectorized::IColumn* col_ptr = &nullable_column->get_nested_column();
423
424	0	switch (type) {
425	0	case TYPE_BIGINT: {
426	0	reinterpret_cast<vectorized::ColumnInt64*>(col_ptr)->insert_value(cell.longVal);
427	0	break;
428	0	}
429
430	0	case TYPE_INT: {
431	0	reinterpret_cast<vectorized::ColumnInt32*>(col_ptr)->insert_value(cell.intVal);
432	0	break;
433	0	}
434
435	0	case TYPE_BOOLEAN: {
436	0	reinterpret_cast<vectorized::ColumnUInt8*>(col_ptr)->insert_value(cell.boolVal);
437	0	break;
438	0	}
439
440	0	case TYPE_STRING:
441	0	case TYPE_VARCHAR:
442	0	case TYPE_CHAR: {
443	0	reinterpret_cast<vectorized::ColumnString*>(col_ptr)->insert_data(cell.stringVal.data(),
444	0	cell.stringVal.size());
445	0	break;
446	0	}
447
448	0	case TYPE_DATETIME: {
449	0	std::vector<void*> datas(1);
450	0	VecDateTimeValue src[1];
451	0	src[0].from_date_str(cell.stringVal.data(), cell.stringVal.size());
452	0	datas[0] = src;
453	0	auto data = datas[0];
454	0	reinterpret_cast<vectorized::ColumnDateTime*>(col_ptr)->insert_data(
455	0	reinterpret_cast<char*>(data), 0);
456	0	break;
457	0	}
458	0	default: {
459	0	std::stringstream ss;
460	0	ss << "unsupported column type:" << type;
461	0	return Status::InternalError(ss.str());
462	0	}
463	0	}
464	0	nullable_column->push_false_to_nullmap(1);
465	0	return Status::OK();
466	0	}
467
468		} // namespace doris