Coverage Report

Created: 2024-11-21 20:39

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