be/src/exec/scan/meta_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/scan/meta_scanner.h"

#include <fmt/format.h>
#include <gen_cpp/FrontendService.h>
#include <gen_cpp/FrontendService_types.h>
#include <gen_cpp/HeartbeatService_types.h>
#include <gen_cpp/PaloInternalService_types.h>
#include <gen_cpp/PlanNodes_types.h>

#include <ostream>
#include <string>
#include <unordered_map>

#include "common/cast_set.h"
#include "common/logging.h"
#include "core/block/block.h"
#include "core/column/column.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/define_primitive_type.h"
#include "core/types.h"
#include "format/table/parquet_metadata_reader.h"
#include "runtime/descriptors.h"
#include "runtime/exec_env.h"
#include "runtime/runtime_state.h"
#include "util/client_cache.h"
#include "util/thrift_rpc_helper.h"

namespace doris {
class RuntimeProfile;
class VExprContext;
} // namespace doris

namespace doris {

MetaScanner::MetaScanner(RuntimeState* state, ScanLocalStateBase* local_state, TupleId tuple_id,
                         const TScanRangeParams& scan_range, int64_t limit, RuntimeProfile* profile,
                         TUserIdentity user_identity)
        : Scanner(state, local_state, limit, profile),
          _meta_eos(false),
          _tuple_id(tuple_id),
          _user_identity(user_identity),
          _scan_range(scan_range.scan_range) {}

Status MetaScanner::_open_impl(RuntimeState* state) {
    VLOG_CRITICAL << "MetaScanner::open";
    RETURN_IF_ERROR(Scanner::_open_impl(state));
    if (_scan_range.meta_scan_range.metadata_type == TMetadataType::PARQUET) {
        auto reader = ParquetMetadataReader::create_unique(_tuple_desc->slots(), state, _profile,
                                                           _scan_range.meta_scan_range);
        RETURN_IF_ERROR(reader->init_reader());
        _reader = std::move(reader);
    } else {
        RETURN_IF_ERROR(_fetch_metadata(_scan_range.meta_scan_range));
    }
    return Status::OK();
}

Status MetaScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
    VLOG_CRITICAL << "MetaScanner::init";
    RETURN_IF_ERROR(Scanner::init(_state, conjuncts));
    _tuple_desc = state->desc_tbl().get_tuple_descriptor(_tuple_id);
    return Status::OK();
}

Status MetaScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eof) {
    VLOG_CRITICAL << "MetaScanner::_get_block_impl";
    if (nullptr == state || nullptr == block || nullptr == eof) {
        return Status::InternalError("input is NULL pointer");
    }

    // Build name to index map only once on first call
    if (_src_block_name_to_idx.empty()) {
        _src_block_name_to_idx = block->get_name_to_pos_map();
    }

    if (_reader) {
        // TODO: This is a temporary workaround; the code is planned to be refactored later.
        size_t read_rows = 0;
        return _reader->get_next_block(block, &read_rows, eof);
    }

    if (_meta_eos == true) {
        *eof = true;
        return Status::OK();
    }

    auto column_size = _tuple_desc->slots().size();
    std::vector<MutableColumnPtr> columns(column_size);
    bool mem_reuse = block->mem_reuse();
    do {
        RETURN_IF_CANCELLED(state);

        columns.resize(column_size);
        for (auto i = 0; i < column_size; i++) {
            if (mem_reuse) {
                columns[i] = block->get_by_position(i).column->assume_mutable();
            } else {
                columns[i] = _tuple_desc->slots()[i]->get_empty_mutable_column();
            }
        }
        // fill block
        RETURN_IF_ERROR(_fill_block_with_remote_data(columns));
        if (_meta_eos == true) {
            if (block->rows() == 0) {
                *eof = true;
            }
            break;
        }
        // Before really use the Block, must clear other ptr of column in block
        // So here need do std::move and clear in `columns`
        if (!mem_reuse) {
            int column_index = 0;
            for (const auto slot_desc : _tuple_desc->slots()) {
                block->insert(ColumnWithTypeAndName(std::move(columns[column_index++]),
                                                    slot_desc->get_data_type_ptr(),
                                                    slot_desc->col_name()));
            }
        } else {
            columns.clear();
        }
        VLOG_ROW << "VMetaScanNode output rows: " << block->rows();
    } while (block->rows() == 0 && !(*eof));
    return Status::OK();
}

Status MetaScanner::_fill_block_with_remote_data(const std::vector<MutableColumnPtr>& columns) {
    VLOG_CRITICAL << "MetaScanner::_fill_block_with_remote_data";
    for (int col_idx = 0; col_idx < columns.size(); col_idx++) {
        auto slot_desc = _tuple_desc->slots()[col_idx];

        for (int _row_idx = 0; _row_idx < _batch_data.size(); _row_idx++) {
            IColumn* col_ptr = columns[col_idx].get();
            TCell& cell = _batch_data[_row_idx].column_value[col_idx];
            if (cell.__isset.isNull && cell.isNull) {
                DCHECK(slot_desc->is_nullable())
                        << "cell is null but column is not nullable: " << slot_desc->col_name();
                auto& null_col = reinterpret_cast<ColumnNullable&>(*col_ptr);
                null_col.get_nested_column().insert_default();
                null_col.get_null_map_data().push_back(1);
            } else {
                if (slot_desc->is_nullable()) {
                    auto& null_col = reinterpret_cast<ColumnNullable&>(*col_ptr);
                    null_col.get_null_map_data().push_back(0);
                    col_ptr = null_col.get_nested_column_ptr().get();
                }
                switch (slot_desc->type()->get_primitive_type()) {
                case TYPE_BOOLEAN: {
                    bool data = cell.boolVal;
                    assert_cast<ColumnBool*>(col_ptr)->insert_value((uint8_t)data);
                    break;
                }
                case TYPE_TINYINT: {
                    int8_t data = (int8_t)cell.intVal;
                    assert_cast<ColumnInt8*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_SMALLINT: {
                    int16_t data = (int16_t)cell.intVal;
                    assert_cast<ColumnInt16*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_INT: {
                    int32_t data = cell.intVal;
                    assert_cast<ColumnInt32*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_BIGINT: {
                    int64_t data = cell.longVal;
                    assert_cast<ColumnInt64*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_FLOAT: {
                    auto data = static_cast<float>(cell.doubleVal);
                    assert_cast<ColumnFloat32*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_DOUBLE: {
                    double data = cell.doubleVal;
                    assert_cast<ColumnFloat64*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_DATEV2: {
                    uint32_t data = (uint32_t)cell.longVal;
                    assert_cast<ColumnDateV2*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_DATETIMEV2: {
                    uint64_t data = cell.longVal;
                    assert_cast<ColumnDateTimeV2*>(col_ptr)->insert_value(data);
                    break;
                }
                case TYPE_STRING:
                case TYPE_CHAR:
                case TYPE_VARCHAR: {
                    std::string data = cell.stringVal;
                    assert_cast<ColumnString*>(col_ptr)->insert_data(data.c_str(), data.length());
                    break;
                }
                default: {
                    std::string error_msg =
                            fmt::format("Invalid column type {} on column: {}.",
                                        slot_desc->type()->get_name(), slot_desc->col_name());
                    return Status::InternalError(std::string(error_msg));
                }
                }
            }
        }
    }
    _meta_eos = true;
    return Status::OK();
}

Status MetaScanner::_fetch_metadata(const TMetaScanRange& meta_scan_range) {
    VLOG_CRITICAL << "MetaScanner::_fetch_metadata";
    TFetchSchemaTableDataRequest request;
    switch (meta_scan_range.metadata_type) {
    case TMetadataType::HUDI:
        RETURN_IF_ERROR(_build_hudi_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::BACKENDS:
        RETURN_IF_ERROR(_build_backends_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::FRONTENDS:
        RETURN_IF_ERROR(_build_frontends_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::FRONTENDS_DISKS:
        RETURN_IF_ERROR(_build_frontends_disks_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::WORKLOAD_SCHED_POLICY:
        RETURN_IF_ERROR(_build_workload_sched_policy_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::CATALOGS:
        RETURN_IF_ERROR(_build_catalogs_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::MATERIALIZED_VIEWS:
        RETURN_IF_ERROR(_build_materialized_views_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::PARTITIONS:
        RETURN_IF_ERROR(_build_partitions_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::JOBS:
        RETURN_IF_ERROR(_build_jobs_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::TASKS:
        RETURN_IF_ERROR(_build_tasks_metadata_request(meta_scan_range, &request));
        break;
    case TMetadataType::PARTITION_VALUES:
        RETURN_IF_ERROR(_build_partition_values_metadata_request(meta_scan_range, &request));
        break;
    default:
        _meta_eos = true;
        return Status::OK();
    }

    // set filter columns
    std::vector<std::string> filter_columns;
    for (const auto& slot : _tuple_desc->slots()) {
        filter_columns.emplace_back(slot->col_name_lower_case());
    }
    request.metada_table_params.__set_columns_name(filter_columns);

    // _state->execution_timeout() is seconds, change to milliseconds
    int time_out = _state->execution_timeout() * 1000;
    TNetworkAddress master_addr = ExecEnv::GetInstance()->cluster_info()->master_fe_addr;
    TFetchSchemaTableDataResult result;
    RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
            master_addr.hostname, master_addr.port,
            [&request, &result](FrontendServiceConnection& client) {
                client->fetchSchemaTableData(result, request);
            },
            time_out));

    Status status(Status::create(result.status));
    if (!status.ok()) {
        LOG(WARNING) << "fetch schema table data from master failed, errmsg=" << status;
        return status;
    }
    _batch_data = std::move(result.data_batch);
    return Status::OK();
}

Status MetaScanner::_build_hudi_metadata_request(const TMetaScanRange& meta_scan_range,
                                                 TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_hudi_metadata_request";
    if (!meta_scan_range.__isset.hudi_params) {
        return Status::InternalError("Can not find THudiMetadataParams from meta_scan_range.");
    }

    // create request
    request->__set_cluster_name("");
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::HUDI);
    metadata_table_params.__set_hudi_metadata_params(meta_scan_range.hudi_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_backends_metadata_request(const TMetaScanRange& meta_scan_range,
                                                     TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_backends_metadata_request";
    if (!meta_scan_range.__isset.backends_params) {
        return Status::InternalError("Can not find TBackendsMetadataParams from meta_scan_range.");
    }
    // create request
    request->__set_cluster_name("");
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::BACKENDS);
    metadata_table_params.__set_backends_metadata_params(meta_scan_range.backends_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_frontends_metadata_request(const TMetaScanRange& meta_scan_range,
                                                      TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_frontends_metadata_request";
    if (!meta_scan_range.__isset.frontends_params) {
        return Status::InternalError("Can not find TFrontendsMetadataParams from meta_scan_range.");
    }
    // create request
    request->__set_cluster_name("");
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::FRONTENDS);
    metadata_table_params.__set_frontends_metadata_params(meta_scan_range.frontends_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_frontends_disks_metadata_request(const TMetaScanRange& meta_scan_range,
                                                            TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_frontends_metadata_request";
    if (!meta_scan_range.__isset.frontends_params) {
        return Status::InternalError("Can not find TFrontendsMetadataParams from meta_scan_range.");
    }
    // create request
    request->__set_cluster_name("");
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::FRONTENDS_DISKS);
    metadata_table_params.__set_frontends_metadata_params(meta_scan_range.frontends_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_workload_sched_policy_metadata_request(
        const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_workload_sched_policy_metadata_request";

    // create request
    request->__set_cluster_name("");
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::WORKLOAD_SCHED_POLICY);
    metadata_table_params.__set_current_user_ident(_user_identity);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_catalogs_metadata_request(const TMetaScanRange& meta_scan_range,
                                                     TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_catalogs_metadata_request";

    // create request
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::CATALOGS);
    metadata_table_params.__set_current_user_ident(_user_identity);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_materialized_views_metadata_request(
        const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_materialized_views_metadata_request";
    if (!meta_scan_range.__isset.materialized_views_params) {
        return Status::InternalError(
                "Can not find TMaterializedViewsMetadataParams from meta_scan_range.");
    }

    // create request
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::MATERIALIZED_VIEWS);
    metadata_table_params.__set_materialized_views_metadata_params(
            meta_scan_range.materialized_views_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_partitions_metadata_request(const TMetaScanRange& meta_scan_range,
                                                       TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_partitions_metadata_request";
    if (!meta_scan_range.__isset.partitions_params) {
        return Status::InternalError(
                "Can not find TPartitionsMetadataParams from meta_scan_range.");
    }

    // create request
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::PARTITIONS);
    metadata_table_params.__set_partitions_metadata_params(meta_scan_range.partitions_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_jobs_metadata_request(const TMetaScanRange& meta_scan_range,
                                                 TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_jobs_metadata_request";
    if (!meta_scan_range.__isset.jobs_params) {
        return Status::InternalError("Can not find TJobsMetadataParams from meta_scan_range.");
    }

    // create request
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::JOBS);
    metadata_table_params.__set_jobs_metadata_params(meta_scan_range.jobs_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_tasks_metadata_request(const TMetaScanRange& meta_scan_range,
                                                  TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_tasks_metadata_request";
    if (!meta_scan_range.__isset.tasks_params) {
        return Status::InternalError("Can not find TTasksMetadataParams from meta_scan_range.");
    }

    // create request
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::TASKS);
    metadata_table_params.__set_tasks_metadata_params(meta_scan_range.tasks_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::_build_partition_values_metadata_request(
        const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
    VLOG_CRITICAL << "MetaScanner::_build_partition_values_metadata_request";
    if (!meta_scan_range.__isset.partition_values_params) {
        return Status::InternalError(
                "Can not find TPartitionValuesMetadataParams from meta_scan_range.");
    }

    // create request
    request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);

    // create TMetadataTableRequestParams
    TMetadataTableRequestParams metadata_table_params;
    metadata_table_params.__set_metadata_type(TMetadataType::PARTITION_VALUES);
    metadata_table_params.__set_partition_values_metadata_params(
            meta_scan_range.partition_values_params);

    request->__set_metada_table_params(metadata_table_params);
    return Status::OK();
}

Status MetaScanner::close(RuntimeState* state) {
    VLOG_CRITICAL << "MetaScanner::close";
    if (!_try_close()) {
        return Status::OK();
    }
    if (_reader) {
        RETURN_IF_ERROR(_reader->close());
    }
    RETURN_IF_ERROR(Scanner::close(state));
    return Status::OK();
}

} // namespace doris

Coverage Report

Created: 2026-04-11 00:05

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/scan/meta_scanner.h"
19
20		#include <fmt/format.h>
21		#include <gen_cpp/FrontendService.h>
22		#include <gen_cpp/FrontendService_types.h>
23		#include <gen_cpp/HeartbeatService_types.h>
24		#include <gen_cpp/PaloInternalService_types.h>
25		#include <gen_cpp/PlanNodes_types.h>
26
27		#include <ostream>
28		#include <string>
29		#include <unordered_map>
30
31		#include "common/cast_set.h"
32		#include "common/logging.h"
33		#include "core/block/block.h"
34		#include "core/column/column.h"
35		#include "core/column/column_nullable.h"
36		#include "core/column/column_string.h"
37		#include "core/column/column_vector.h"
38		#include "core/data_type/define_primitive_type.h"
39		#include "core/types.h"
40		#include "format/table/parquet_metadata_reader.h"
41		#include "runtime/descriptors.h"
42		#include "runtime/exec_env.h"
43		#include "runtime/runtime_state.h"
44		#include "util/client_cache.h"
45		#include "util/thrift_rpc_helper.h"
46
47		namespace doris {
48		class RuntimeProfile;
49		class VExprContext;
50		} // namespace doris
51
52		namespace doris {
53
54		MetaScanner::MetaScanner(RuntimeState* state, ScanLocalStateBase* local_state, TupleId tuple_id,
55		const TScanRangeParams& scan_range, int64_t limit, RuntimeProfile* profile,
56		TUserIdentity user_identity)
57	0	: Scanner(state, local_state, limit, profile),
58	0	_meta_eos(false),
59	0	_tuple_id(tuple_id),
60	0	_user_identity(user_identity),
61	0	_scan_range(scan_range.scan_range) {}
62
63	0	Status MetaScanner::_open_impl(RuntimeState* state) {
64	0	VLOG_CRITICAL << "MetaScanner::open";
65	0	RETURN_IF_ERROR(Scanner::_open_impl(state));
66	0	if (_scan_range.meta_scan_range.metadata_type == TMetadataType::PARQUET) {
67	0	auto reader = ParquetMetadataReader::create_unique(_tuple_desc->slots(), state, _profile,
68	0	_scan_range.meta_scan_range);
69	0	RETURN_IF_ERROR(reader->init_reader());
70	0	_reader = std::move(reader);
71	0	} else {
72	0	RETURN_IF_ERROR(_fetch_metadata(_scan_range.meta_scan_range));
73	0	}
74	0	return Status::OK();
75	0	}
76
77	0	Status MetaScanner::init(RuntimeState* state, const VExprContextSPtrs& conjuncts) {
78	0	VLOG_CRITICAL << "MetaScanner::init";
79	0	RETURN_IF_ERROR(Scanner::init(_state, conjuncts));
80	0	_tuple_desc = state->desc_tbl().get_tuple_descriptor(_tuple_id);
81	0	return Status::OK();
82	0	}
83
84	0	Status MetaScanner::_get_block_impl(RuntimeState* state, Block* block, bool* eof) {
85	0	VLOG_CRITICAL << "MetaScanner::_get_block_impl";
86	0	if (nullptr == state \|\| nullptr == block \|\| nullptr == eof) {
87	0	return Status::InternalError("input is NULL pointer");
88	0	}
89
90		// Build name to index map only once on first call
91	0	if (_src_block_name_to_idx.empty()) {
92	0	_src_block_name_to_idx = block->get_name_to_pos_map();
93	0	}
94
95	0	if (_reader) {
96		// TODO: This is a temporary workaround; the code is planned to be refactored later.
97	0	size_t read_rows = 0;
98	0	return _reader->get_next_block(block, &read_rows, eof);
99	0	}
100
101	0	if (_meta_eos == true) {
102	0	*eof = true;
103	0	return Status::OK();
104	0	}
105
106	0	auto column_size = _tuple_desc->slots().size();
107	0	std::vector<MutableColumnPtr> columns(column_size);
108	0	bool mem_reuse = block->mem_reuse();
109	0	do {
110	0	RETURN_IF_CANCELLED(state);
111
112	0	columns.resize(column_size);
113	0	for (auto i = 0; i < column_size; i++) {
114	0	if (mem_reuse) {
115	0	columns[i] = block->get_by_position(i).column->assume_mutable();
116	0	} else {
117	0	columns[i] = _tuple_desc->slots()[i]->get_empty_mutable_column();
118	0	}
119	0	}
120		// fill block
121	0	RETURN_IF_ERROR(_fill_block_with_remote_data(columns));
122	0	if (_meta_eos == true) {
123	0	if (block->rows() == 0) {
124	0	*eof = true;
125	0	}
126	0	break;
127	0	}
128		// Before really use the Block, must clear other ptr of column in block
129		// So here need do std::move and clear in `columns`
130	0	if (!mem_reuse) {
131	0	int column_index = 0;
132	0	for (const auto slot_desc : _tuple_desc->slots()) {
133	0	block->insert(ColumnWithTypeAndName(std::move(columns[column_index++]),
134	0	slot_desc->get_data_type_ptr(),
135	0	slot_desc->col_name()));
136	0	}
137	0	} else {
138	0	columns.clear();
139	0	}
140	0	VLOG_ROW << "VMetaScanNode output rows: " << block->rows();
141	0	} while (block->rows() == 0 && !(*eof));
142	0	return Status::OK();
143	0	}
144
145	0	Status MetaScanner::_fill_block_with_remote_data(const std::vector<MutableColumnPtr>& columns) {
146	0	VLOG_CRITICAL << "MetaScanner::_fill_block_with_remote_data";
147	0	for (int col_idx = 0; col_idx < columns.size(); col_idx++) {
148	0	auto slot_desc = _tuple_desc->slots()[col_idx];
149
150	0	for (int _row_idx = 0; _row_idx < _batch_data.size(); _row_idx++) {
151	0	IColumn* col_ptr = columns[col_idx].get();
152	0	TCell& cell = _batch_data[_row_idx].column_value[col_idx];
153	0	if (cell.__isset.isNull && cell.isNull) {
154	0	DCHECK(slot_desc->is_nullable())
155	0	<< "cell is null but column is not nullable: " << slot_desc->col_name();
156	0	auto& null_col = reinterpret_cast<ColumnNullable&>(*col_ptr);
157	0	null_col.get_nested_column().insert_default();
158	0	null_col.get_null_map_data().push_back(1);
159	0	} else {
160	0	if (slot_desc->is_nullable()) {
161	0	auto& null_col = reinterpret_cast<ColumnNullable&>(*col_ptr);
162	0	null_col.get_null_map_data().push_back(0);
163	0	col_ptr = null_col.get_nested_column_ptr().get();
164	0	}
165	0	switch (slot_desc->type()->get_primitive_type()) {
166	0	case TYPE_BOOLEAN: {
167	0	bool data = cell.boolVal;
168	0	assert_cast<ColumnBool*>(col_ptr)->insert_value((uint8_t)data);
169	0	break;
170	0	}
171	0	case TYPE_TINYINT: {
172	0	int8_t data = (int8_t)cell.intVal;
173	0	assert_cast<ColumnInt8*>(col_ptr)->insert_value(data);
174	0	break;
175	0	}
176	0	case TYPE_SMALLINT: {
177	0	int16_t data = (int16_t)cell.intVal;
178	0	assert_cast<ColumnInt16*>(col_ptr)->insert_value(data);
179	0	break;
180	0	}
181	0	case TYPE_INT: {
182	0	int32_t data = cell.intVal;
183	0	assert_cast<ColumnInt32*>(col_ptr)->insert_value(data);
184	0	break;
185	0	}
186	0	case TYPE_BIGINT: {
187	0	int64_t data = cell.longVal;
188	0	assert_cast<ColumnInt64*>(col_ptr)->insert_value(data);
189	0	break;
190	0	}
191	0	case TYPE_FLOAT: {
192	0	auto data = static_cast<float>(cell.doubleVal);
193	0	assert_cast<ColumnFloat32*>(col_ptr)->insert_value(data);
194	0	break;
195	0	}
196	0	case TYPE_DOUBLE: {
197	0	double data = cell.doubleVal;
198	0	assert_cast<ColumnFloat64*>(col_ptr)->insert_value(data);
199	0	break;
200	0	}
201	0	case TYPE_DATEV2: {
202	0	uint32_t data = (uint32_t)cell.longVal;
203	0	assert_cast<ColumnDateV2*>(col_ptr)->insert_value(data);
204	0	break;
205	0	}
206	0	case TYPE_DATETIMEV2: {
207	0	uint64_t data = cell.longVal;
208	0	assert_cast<ColumnDateTimeV2*>(col_ptr)->insert_value(data);
209	0	break;
210	0	}
211	0	case TYPE_STRING:
212	0	case TYPE_CHAR:
213	0	case TYPE_VARCHAR: {
214	0	std::string data = cell.stringVal;
215	0	assert_cast<ColumnString*>(col_ptr)->insert_data(data.c_str(), data.length());
216	0	break;
217	0	}
218	0	default: {
219	0	std::string error_msg =
220	0	fmt::format("Invalid column type {} on column: {}.",
221	0	slot_desc->type()->get_name(), slot_desc->col_name());
222	0	return Status::InternalError(std::string(error_msg));
223	0	}
224	0	}
225	0	}
226	0	}
227	0	}
228	0	_meta_eos = true;
229	0	return Status::OK();
230	0	}
231
232	0	Status MetaScanner::_fetch_metadata(const TMetaScanRange& meta_scan_range) {
233	0	VLOG_CRITICAL << "MetaScanner::_fetch_metadata";
234	0	TFetchSchemaTableDataRequest request;
235	0	switch (meta_scan_range.metadata_type) {
236	0	case TMetadataType::HUDI:
237	0	RETURN_IF_ERROR(_build_hudi_metadata_request(meta_scan_range, &request));
238	0	break;
239	0	case TMetadataType::BACKENDS:
240	0	RETURN_IF_ERROR(_build_backends_metadata_request(meta_scan_range, &request));
241	0	break;
242	0	case TMetadataType::FRONTENDS:
243	0	RETURN_IF_ERROR(_build_frontends_metadata_request(meta_scan_range, &request));
244	0	break;
245	0	case TMetadataType::FRONTENDS_DISKS:
246	0	RETURN_IF_ERROR(_build_frontends_disks_metadata_request(meta_scan_range, &request));
247	0	break;
248	0	case TMetadataType::WORKLOAD_SCHED_POLICY:
249	0	RETURN_IF_ERROR(_build_workload_sched_policy_metadata_request(meta_scan_range, &request));
250	0	break;
251	0	case TMetadataType::CATALOGS:
252	0	RETURN_IF_ERROR(_build_catalogs_metadata_request(meta_scan_range, &request));
253	0	break;
254	0	case TMetadataType::MATERIALIZED_VIEWS:
255	0	RETURN_IF_ERROR(_build_materialized_views_metadata_request(meta_scan_range, &request));
256	0	break;
257	0	case TMetadataType::PARTITIONS:
258	0	RETURN_IF_ERROR(_build_partitions_metadata_request(meta_scan_range, &request));
259	0	break;
260	0	case TMetadataType::JOBS:
261	0	RETURN_IF_ERROR(_build_jobs_metadata_request(meta_scan_range, &request));
262	0	break;
263	0	case TMetadataType::TASKS:
264	0	RETURN_IF_ERROR(_build_tasks_metadata_request(meta_scan_range, &request));
265	0	break;
266	0	case TMetadataType::PARTITION_VALUES:
267	0	RETURN_IF_ERROR(_build_partition_values_metadata_request(meta_scan_range, &request));
268	0	break;
269	0	default:
270	0	_meta_eos = true;
271	0	return Status::OK();
272	0	}
273
274		// set filter columns
275	0	std::vector<std::string> filter_columns;
276	0	for (const auto& slot : _tuple_desc->slots()) {
277	0	filter_columns.emplace_back(slot->col_name_lower_case());
278	0	}
279	0	request.metada_table_params.__set_columns_name(filter_columns);
280
281		// _state->execution_timeout() is seconds, change to milliseconds
282	0	int time_out = _state->execution_timeout() * 1000;
283	0	TNetworkAddress master_addr = ExecEnv::GetInstance()->cluster_info()->master_fe_addr;
284	0	TFetchSchemaTableDataResult result;
285	0	RETURN_IF_ERROR(ThriftRpcHelper::rpc<FrontendServiceClient>(
286	0	master_addr.hostname, master_addr.port,
287	0	[&request, &result](FrontendServiceConnection& client) {
288	0	client->fetchSchemaTableData(result, request);
289	0	},
290	0	time_out));
291
292	0	Status status(Status::create(result.status));
293	0	if (!status.ok()) {
294	0	LOG(WARNING) << "fetch schema table data from master failed, errmsg=" << status;
295	0	return status;
296	0	}
297	0	_batch_data = std::move(result.data_batch);
298	0	return Status::OK();
299	0	}
300
301		Status MetaScanner::_build_hudi_metadata_request(const TMetaScanRange& meta_scan_range,
302	0	TFetchSchemaTableDataRequest* request) {
303	0	VLOG_CRITICAL << "MetaScanner::_build_hudi_metadata_request";
304	0	if (!meta_scan_range.__isset.hudi_params) {
305	0	return Status::InternalError("Can not find THudiMetadataParams from meta_scan_range.");
306	0	}
307
308		// create request
309	0	request->__set_cluster_name("");
310	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
311
312		// create TMetadataTableRequestParams
313	0	TMetadataTableRequestParams metadata_table_params;
314	0	metadata_table_params.__set_metadata_type(TMetadataType::HUDI);
315	0	metadata_table_params.__set_hudi_metadata_params(meta_scan_range.hudi_params);
316
317	0	request->__set_metada_table_params(metadata_table_params);
318	0	return Status::OK();
319	0	}
320
321		Status MetaScanner::_build_backends_metadata_request(const TMetaScanRange& meta_scan_range,
322	0	TFetchSchemaTableDataRequest* request) {
323	0	VLOG_CRITICAL << "MetaScanner::_build_backends_metadata_request";
324	0	if (!meta_scan_range.__isset.backends_params) {
325	0	return Status::InternalError("Can not find TBackendsMetadataParams from meta_scan_range.");
326	0	}
327		// create request
328	0	request->__set_cluster_name("");
329	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
330
331		// create TMetadataTableRequestParams
332	0	TMetadataTableRequestParams metadata_table_params;
333	0	metadata_table_params.__set_metadata_type(TMetadataType::BACKENDS);
334	0	metadata_table_params.__set_backends_metadata_params(meta_scan_range.backends_params);
335
336	0	request->__set_metada_table_params(metadata_table_params);
337	0	return Status::OK();
338	0	}
339
340		Status MetaScanner::_build_frontends_metadata_request(const TMetaScanRange& meta_scan_range,
341	0	TFetchSchemaTableDataRequest* request) {
342	0	VLOG_CRITICAL << "MetaScanner::_build_frontends_metadata_request";
343	0	if (!meta_scan_range.__isset.frontends_params) {
344	0	return Status::InternalError("Can not find TFrontendsMetadataParams from meta_scan_range.");
345	0	}
346		// create request
347	0	request->__set_cluster_name("");
348	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
349
350		// create TMetadataTableRequestParams
351	0	TMetadataTableRequestParams metadata_table_params;
352	0	metadata_table_params.__set_metadata_type(TMetadataType::FRONTENDS);
353	0	metadata_table_params.__set_frontends_metadata_params(meta_scan_range.frontends_params);
354
355	0	request->__set_metada_table_params(metadata_table_params);
356	0	return Status::OK();
357	0	}
358
359		Status MetaScanner::_build_frontends_disks_metadata_request(const TMetaScanRange& meta_scan_range,
360	0	TFetchSchemaTableDataRequest* request) {
361	0	VLOG_CRITICAL << "MetaScanner::_build_frontends_metadata_request";
362	0	if (!meta_scan_range.__isset.frontends_params) {
363	0	return Status::InternalError("Can not find TFrontendsMetadataParams from meta_scan_range.");
364	0	}
365		// create request
366	0	request->__set_cluster_name("");
367	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
368
369		// create TMetadataTableRequestParams
370	0	TMetadataTableRequestParams metadata_table_params;
371	0	metadata_table_params.__set_metadata_type(TMetadataType::FRONTENDS_DISKS);
372	0	metadata_table_params.__set_frontends_metadata_params(meta_scan_range.frontends_params);
373
374	0	request->__set_metada_table_params(metadata_table_params);
375	0	return Status::OK();
376	0	}
377
378		Status MetaScanner::_build_workload_sched_policy_metadata_request(
379	0	const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
380	0	VLOG_CRITICAL << "MetaScanner::_build_workload_sched_policy_metadata_request";
381
382		// create request
383	0	request->__set_cluster_name("");
384	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
385
386		// create TMetadataTableRequestParams
387	0	TMetadataTableRequestParams metadata_table_params;
388	0	metadata_table_params.__set_metadata_type(TMetadataType::WORKLOAD_SCHED_POLICY);
389	0	metadata_table_params.__set_current_user_ident(_user_identity);
390
391	0	request->__set_metada_table_params(metadata_table_params);
392	0	return Status::OK();
393	0	}
394
395		Status MetaScanner::_build_catalogs_metadata_request(const TMetaScanRange& meta_scan_range,
396	0	TFetchSchemaTableDataRequest* request) {
397	0	VLOG_CRITICAL << "MetaScanner::_build_catalogs_metadata_request";
398
399		// create request
400	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
401
402		// create TMetadataTableRequestParams
403	0	TMetadataTableRequestParams metadata_table_params;
404	0	metadata_table_params.__set_metadata_type(TMetadataType::CATALOGS);
405	0	metadata_table_params.__set_current_user_ident(_user_identity);
406
407	0	request->__set_metada_table_params(metadata_table_params);
408	0	return Status::OK();
409	0	}
410
411		Status MetaScanner::_build_materialized_views_metadata_request(
412	0	const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
413	0	VLOG_CRITICAL << "MetaScanner::_build_materialized_views_metadata_request";
414	0	if (!meta_scan_range.__isset.materialized_views_params) {
415	0	return Status::InternalError(
416	0	"Can not find TMaterializedViewsMetadataParams from meta_scan_range.");
417	0	}
418
419		// create request
420	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
421
422		// create TMetadataTableRequestParams
423	0	TMetadataTableRequestParams metadata_table_params;
424	0	metadata_table_params.__set_metadata_type(TMetadataType::MATERIALIZED_VIEWS);
425	0	metadata_table_params.__set_materialized_views_metadata_params(
426	0	meta_scan_range.materialized_views_params);
427
428	0	request->__set_metada_table_params(metadata_table_params);
429	0	return Status::OK();
430	0	}
431
432		Status MetaScanner::_build_partitions_metadata_request(const TMetaScanRange& meta_scan_range,
433	0	TFetchSchemaTableDataRequest* request) {
434	0	VLOG_CRITICAL << "MetaScanner::_build_partitions_metadata_request";
435	0	if (!meta_scan_range.__isset.partitions_params) {
436	0	return Status::InternalError(
437	0	"Can not find TPartitionsMetadataParams from meta_scan_range.");
438	0	}
439
440		// create request
441	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
442
443		// create TMetadataTableRequestParams
444	0	TMetadataTableRequestParams metadata_table_params;
445	0	metadata_table_params.__set_metadata_type(TMetadataType::PARTITIONS);
446	0	metadata_table_params.__set_partitions_metadata_params(meta_scan_range.partitions_params);
447
448	0	request->__set_metada_table_params(metadata_table_params);
449	0	return Status::OK();
450	0	}
451
452		Status MetaScanner::_build_jobs_metadata_request(const TMetaScanRange& meta_scan_range,
453	0	TFetchSchemaTableDataRequest* request) {
454	0	VLOG_CRITICAL << "MetaScanner::_build_jobs_metadata_request";
455	0	if (!meta_scan_range.__isset.jobs_params) {
456	0	return Status::InternalError("Can not find TJobsMetadataParams from meta_scan_range.");
457	0	}
458
459		// create request
460	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
461
462		// create TMetadataTableRequestParams
463	0	TMetadataTableRequestParams metadata_table_params;
464	0	metadata_table_params.__set_metadata_type(TMetadataType::JOBS);
465	0	metadata_table_params.__set_jobs_metadata_params(meta_scan_range.jobs_params);
466
467	0	request->__set_metada_table_params(metadata_table_params);
468	0	return Status::OK();
469	0	}
470
471		Status MetaScanner::_build_tasks_metadata_request(const TMetaScanRange& meta_scan_range,
472	0	TFetchSchemaTableDataRequest* request) {
473	0	VLOG_CRITICAL << "MetaScanner::_build_tasks_metadata_request";
474	0	if (!meta_scan_range.__isset.tasks_params) {
475	0	return Status::InternalError("Can not find TTasksMetadataParams from meta_scan_range.");
476	0	}
477
478		// create request
479	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
480
481		// create TMetadataTableRequestParams
482	0	TMetadataTableRequestParams metadata_table_params;
483	0	metadata_table_params.__set_metadata_type(TMetadataType::TASKS);
484	0	metadata_table_params.__set_tasks_metadata_params(meta_scan_range.tasks_params);
485
486	0	request->__set_metada_table_params(metadata_table_params);
487	0	return Status::OK();
488	0	}
489
490		Status MetaScanner::_build_partition_values_metadata_request(
491	0	const TMetaScanRange& meta_scan_range, TFetchSchemaTableDataRequest* request) {
492	0	VLOG_CRITICAL << "MetaScanner::_build_partition_values_metadata_request";
493	0	if (!meta_scan_range.__isset.partition_values_params) {
494	0	return Status::InternalError(
495	0	"Can not find TPartitionValuesMetadataParams from meta_scan_range.");
496	0	}
497
498		// create request
499	0	request->__set_schema_table_name(TSchemaTableName::METADATA_TABLE);
500
501		// create TMetadataTableRequestParams
502	0	TMetadataTableRequestParams metadata_table_params;
503	0	metadata_table_params.__set_metadata_type(TMetadataType::PARTITION_VALUES);
504	0	metadata_table_params.__set_partition_values_metadata_params(
505	0	meta_scan_range.partition_values_params);
506
507	0	request->__set_metada_table_params(metadata_table_params);
508	0	return Status::OK();
509	0	}
510
511	0	Status MetaScanner::close(RuntimeState* state) {
512	0	VLOG_CRITICAL << "MetaScanner::close";
513	0	if (!_try_close()) {
514	0	return Status::OK();
515	0	}
516	0	if (_reader) {
517	0	RETURN_IF_ERROR(_reader->close());
518	0	}
519	0	RETURN_IF_ERROR(Scanner::close(state));
520	0	return Status::OK();
521	0	}
522
523		} // namespace doris