|
DAGQueryBlock::DAGQueryBlock(UInt32 id_, const ::google::protobuf::RepeatedPtrField<tipb::Executor> & executors) |
|
: id(id_) |
|
, root(nullptr) |
|
, qb_column_prefix("__QB_" + std::to_string(id_) + "_") |
|
{ |
|
for (int i = executors.size() - 1; i >= 0; i--) |
|
{ |
|
switch (executors[i].tp()) |
|
{ |
|
case tipb::ExecType::TypeTableScan: |
|
GET_METRIC(tiflash_coprocessor_executor_count, type_ts).Increment(); |
|
assignOrThrowException(&source, &executors[i], SOURCE_NAME); |
|
/// use index as the prefix for executor name so when we sort by |
|
/// the executor name, it will result in the same order as it is |
|
/// in the dag_request, this is needed when filling execution_summary |
|
/// in DAGDriver |
|
if (executors[i].has_executor_id()) |
|
source_name = executors[i].executor_id(); |
|
else |
|
source_name = std::to_string(i) + "_tablescan"; |
|
break; |
|
case tipb::ExecType::TypeSelection: |
|
GET_METRIC(tiflash_coprocessor_executor_count, type_sel).Increment(); |
|
assignOrThrowException(&selection, &executors[i], SEL_NAME); |
|
if (executors[i].has_executor_id()) |
|
selection_name = executors[i].executor_id(); |
|
else |
|
selection_name = std::to_string(i) + "_selection"; |
|
break; |
|
case tipb::ExecType::TypeStreamAgg: |
|
case tipb::ExecType::TypeAggregation: |
|
GET_METRIC(tiflash_coprocessor_executor_count, type_agg).Increment(); |
|
assignOrThrowException(&aggregation, &executors[i], AGG_NAME); |
|
if (executors[i].has_executor_id()) |
|
aggregation_name = executors[i].executor_id(); |
|
else |
|
aggregation_name = std::to_string(i) + "_aggregation"; |
|
break; |
|
case tipb::ExecType::TypeTopN: |
|
GET_METRIC(tiflash_coprocessor_executor_count, type_topn).Increment(); |
|
assignOrThrowException(&limit_or_topn, &executors[i], TOPN_NAME); |
|
if (executors[i].has_executor_id()) |
|
limit_or_topn_name = executors[i].executor_id(); |
|
else |
|
limit_or_topn_name = std::to_string(i) + "_limitOrTopN"; |
|
break; |
|
case tipb::ExecType::TypeLimit: |
|
GET_METRIC(tiflash_coprocessor_executor_count, type_limit).Increment(); |
|
assignOrThrowException(&limit_or_topn, &executors[i], LIMIT_NAME); |
|
if (executors[i].has_executor_id()) |
|
limit_or_topn_name = executors[i].executor_id(); |
|
else |
|
limit_or_topn_name = std::to_string(i) + "_limitOrTopN"; |
|
break; |
|
default: |
|
throw TiFlashException( |
|
"Unsupported executor in DAG request: " + executors[i].DebugString(), |
|
Errors::Coprocessor::Unimplemented); |
|
} |
|
} |
|
} |