Coverage Report

Created: 2025-06-06 13:46

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