在 StarRocks 中，SQL 查询的生命周期分为三个阶段：查询解析（Parsing）、查询规划（Planning）和查询执行（Execution）。查询计划由 Frontend (FE) 生成并拆分为多个 fragment，这些 fragment 被分发到多个 BE 节点并行执行。每个 BE 节点接收到的 fragment 包含具体的执行逻辑，例如扫描数据、执行算子（比如 JOIN、AGGREGATE）以及结果返回。本文主要分析在 BE 侧对 fragment的执行流程，基于StarRocks3.4版本。
ExecNode 和 DataSink
通过 explain sql 查看拆分成几个 fragment 和 fragment 的结构。比如，查询SQL：select dt, avg(imp_cnt) from xxx group by dt limit 3; 通过explain语句可以看到分成了3个fragment，每个fragment都至少会有一个 ExecNode 和 DataSink 节点。
每个plan_fragment 都会有 ExecNode节点和 DataSink节点。 ExecNode节点执行该fragment需要完成的动作， DataSink节点上报 ExecNode节点执行产生的结果。
ExecNode
- 初始化和准备节点以进行执行。
- 管理查询执行的生命周期，包括 open、get_next、reset 和 close 操作。
DataSink
DataSink
是一个抽象基类，为查询执行计划中的数据接收器提供通用接口和功能。子类（如
ResultSink
、Export
Sink
、TableFunctionTableSink
）实现特定类型的数据输出逻辑。
-
初始化和准备数据接收器。
-
管理数据发送、打开和关闭的生命周期。
调度
BE 在接收到 FE 发来的 fragment 信息以及做对应的处理主要由 internal_service， fragment_executor， pipeline_driver， pipeline_driver_exectuor 这几个组件做的处理，具体如下：
internal_service
接收 fragment 执行请求：FE 将查询计划的 fragment（ TExecPlanFragmentParams ）通过 gRPC 传递给 BE， InternalService 负责接收并触发执行。
fragment_executor
- 将 fragment 的执行计划分解为多个 pipeline，并为每个 pipeline 创建对应的 PipelineDriver 。
-
协调执行：通过
PipelineDriverExecutor
调度所有
PipelineDriver
，确保 fragment 的所有 pipeline 按依赖关系正确执行。
pipeline_driver
- 表示一个 pipeline 的执行实例。pipeline 是 StarRocks 中查询执行的基本单元，包含一系列算子（operators，如 ScanOperator、JoinOperator），这些算子以流式方式处理数据。
-
每个
PipelineDriver
负责执行一个 pipeline 的完整逻辑，包括从数据读取到结果输出。
pipeline_driver_exectuor
- 是一个全局的管理组件，负责调度和管理多个 PipelineDriver 的执行。
- 它维护一个线程池或工作队列，将 PipelineDriver 分配到线程中执行，并处理任务的并发、优先级和资源管理。
以一个查询hive表然后导出到hdfs上的sql为例，看看execnode和datasink 的创建
insert into files("path" = "hdfs://xxx/xx", "format" = "csv") select * from xxx limit 1;
通过如下堆栈：fragment_executor:: _prepare_exec_plan -> exec_node:: create_tree -> create_vectorized_node，在 create_vectorized_node方法里，创建对应的 exec_node 对象，如下：
Status ExecNode::create_vectorized_node(starrocks::RuntimeState* state, starrocks::ObjectPool* pool,
const starrocks::TPlanNode& tnode, const starrocks::DescriptorTbl& descs,
starrocks::ExecNode** node) {
switch (tnode.node_type) {
.............
case TPlanNodeType::EXCHANGE_NODE:
*node = pool->add(new ExchangeNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::HASH_JOIN_NODE:
*node = pool->add(new HashJoinNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::ANALYTIC_EVAL_NODE:
*node = pool->add(new AnalyticNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::SORT_NODE:
*node = pool->add(new TopNNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::CROSS_JOIN_NODE:
case TPlanNodeType::NESTLOOP_JOIN_NODE:
*node = pool->add(new CrossJoinNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::UNION_NODE:
*node = pool->add(new UnionNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::INTERSECT_NODE:
*node = pool->add(new IntersectNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::EXCEPT_NODE:
*node = pool->add(new ExceptNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::SELECT_NODE:
*node = pool->add(new SelectNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::FILE_SCAN_NODE: {
if (tnode.file_scan_node.__isset.enable_pipeline_load && tnode.file_scan_node.enable_pipeline_load) {
TPlanNode new_node = tnode;
TConnectorScanNode connector_scan_node;
connector_scan_node.connector_name = connector::Connector::FILE;
new_node.connector_scan_node = connector_scan_node;
*node = pool->add(new ConnectorScanNode(pool, new_node, descs));
} else {
*node = pool->add(new FileScanNode(pool, tnode, descs));
}
}
return Status::OK();
case TPlanNodeType::REPEAT_NODE:
*node = pool->add(new RepeatNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::ASSERT_NUM_ROWS_NODE:
*node = pool->add(new AssertNumRowsNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::PROJECT_NODE:
*node = pool->add(new ProjectNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::TABLE_FUNCTION_NODE:
*node = pool->add(new TableFunctionNode(pool, tnode, descs));
return Status::OK();
case TPlanNodeType::HDFS_SCAN_NODE:
case TPlanNodeType::KUDU_SCAN_NODE: {
TPlanNode new_node = tnode;
TConnectorScanNode connector_scan_node;
connector_scan_node.connector_name = connector::Connector::HIVE;
new_node.connector_scan_node = connector_scan_node;
*node = pool->add(new ConnectorScanNode(pool, new_node, descs));
return Status::OK();
}
............
}
在这个例子里，最终会创建 ConnectorScanNode，简单分析下堆栈：
get_next -> :: _start_scan_thread -> ::_submit_scanner -> ::_scanner_thread -> ::open --> HiveDataSource::open. --> _init_scanner 在这里会判断是生成哪种Scanner，走JNI还是不走JNI，然后读取数据源数据。
而data sink则是fragment_executor在方法 _prepare_pipeline_driver --> create_data_sink，具体代码如下，在这里会创建对应的 data sink 对象：
Status DataSink::create_data_sink(RuntimeState* state, const TDataSink& thrift_sink,
const std::vector<TExpr>& output_exprs, const TPlanFragmentExecParams& params,
int32_t sender_id, const RowDescriptor& row_desc, std::unique_ptr<DataSink>* sink) {
DCHECK(sink != nullptr);
switch (thrift_sink.type) {
case TDataSinkType::DATA_STREAM_SINK: {
if (!thrift_sink.__isset.stream_sink) {
return Status::InternalError("Missing data stream sink.");
}
*sink = create_data_stream_sink(state, thrift_sink.stream_sink, row_desc, params, sender_id,
params.destinations);
break;
}
case TDataSinkType::RESULT_SINK:
if (!thrift_sink.__isset.result_sink) {
return Status::InternalError("Missing data buffer sink.");
}
case TDataSinkType::EXPORT_SINK: {
if (!thrift_sink.__isset.export_sink) {
return Status::InternalError("Missing export sink sink.");
}
*sink = std::make_unique<ExportSink>(state->obj_pool(), row_desc, output_exprs);
break;
}
case TDataSinkType::OLAP_TABLE_SINK: {
Status status;
DCHECK(thrift_sink.__isset.olap_table_sink);
*sink = std::make_unique<OlapTableSink>(state->obj_pool(), output_exprs, &status, state);
RETURN_IF_ERROR(status);
break;
}
case TDataSinkType::MULTI_OLAP_TABLE_SINK: {
Status status;
DCHECK(thrift_sink.__isset.multi_olap_table_sinks);
*sink = std::make_unique<MultiOlapTableSink>(state->obj_pool(), output_exprs);
break;
}
case TDataSinkType::HIVE_TABLE_SINK: {
if (!thrift_sink.__isset.hive_table_sink) {
return Status::InternalError("Missing hive table sink");
}
*sink = std::make_unique<HiveTableSink>(state->obj_pool(), output_exprs);
break;
}
case TDataSinkType::TABLE_FUNCTION_TABLE_SINK: {
if (!thrift_sink.__isset.table_function_table_sink) {
return Status::InternalError("Missing table function table sink");
}
*sink = std::make_unique<TableFunctionTableSink>(state->obj_pool(), output_exprs);
break;
}
case TDataSinkType::BLACKHOLE_TABLE_SINK: {
*sink = std::make_unique<BlackHoleTableSink>(state->obj_pool());
break;
}
case TDataSinkType::DICTIONARY_CACHE_SINK: {
if (!thrift_sink.__isset.dictionary_cache_sink) {
return Status::InternalError("Missing dictionary cache sink");
}
if (!state->enable_pipeline_engine()) {
return Status::InternalError("dictionary cache only support pipeline engine");
}
*sink = std::make_unique<DictionaryCacheSink>();
break;
}
}
在这个例子里创建的是 TableFunctionTableSink
这里准备 sink context 上下文信息，包括 hdfs path地址和 hdfs conf的一些配置信息
到这里就找到了fs实例的创建，通过fs把数据写到hdfs上。
更多大数据干货，欢迎关注我的微信公众号—BigData共享
来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/70046653/viewspace-3091210/ ，如需转载，请注明出处，否则将追究法律责任。
