Duckdb：Push-Based Execution Model

// Executor::SchedulePipeline => 创建 pipeline events
// create events/stack for the base pipeline
auto base_pipeline = meta_pipeline->GetBasePipeline();
auto base_initialize_event = make_shared<PipelineInitializeEvent>(base_pipeline);
auto base_event = make_shared<PipelineEvent>(base_pipeline);
auto base_finish_event = make_shared<PipelineFinishEvent>(base_pipeline);
auto base_complete_event = make_shared<PipelineCompleteEvent>(base_pipeline->executor, event_data.initial_schedule);
PipelineEventStack base_stack(*base_initialize_event, *base_event, *base_finish_event, *base_complete_event);
events.push_back(std::move(base_initialize_event));
events.push_back(std::move(base_event));
events.push_back(std::move(base_finish_event));
events.push_back(std::move(base_complete_event));

// dependencies: initialize -> event -> finish -> complete
base_stack.pipeline_event.AddDependency(base_stack.pipeline_initialize_event);
base_stack.pipeline_finish_event.AddDependency(base_stack.pipeline_event);
base_stack.pipeline_complete_event.AddDependency(base_stack.pipeline_finish_event);

// 设置同个meta_pipeline 内pipeline 间event 依赖
PipelineEventStack pipeline_stack(base_stack.pipeline_initialize_event, *pipeline_event,
                                  *pipeline_finish_event_ptr, base_stack.pipeline_complete_event);
events.push_back(std::move(pipeline_event));

// dependencies: base_initialize -> pipeline_event -> base_finish
pipeline_stack.pipeline_event.AddDependency(base_stack.pipeline_initialize_event);
pipeline_stack.pipeline_finish_event.AddDependency(pipeline_stack.pipeline_event);
...


// Executor::ScheduleEventsInternal
auto &event_map = event_data.event_map;
for (auto &entry : event_map) {
  auto &pipeline = entry.first.get();
  for (auto &dependency : pipeline.dependencies) {
    auto dep = dependency.lock();
    D_ASSERT(dep);
    auto event_map_entry = event_map.find(*dep);
    D_ASSERT(event_map_entry != event_map.end());
    auto &dep_entry = event_map_entry->second;
    // 设置跨 meta_pipeline 之间的event 依赖
    entry.second.pipeline_event.AddDependency(dep_entry.pipeline_complete_event);
  }
}

// duckdb::Event
void Event::CompleteDependency() {
	idx_t current_finished = ++finished_dependencies;
	D_ASSERT(current_finished <= total_dependencies);
	if (current_finished == total_dependencies) {
		// all dependencies have been completed: schedule the event
		D_ASSERT(total_tasks == 0);
		Schedule();
		if (total_tasks == 0) {		// 若当前 event 不生成 task，直接执行 finish
			Finish();
		}
	}
}

void Event::Finish() {
	D_ASSERT(!finished);
	FinishEvent();
	finished = true;
	// finished processing the pipeline, now we can schedule pipelines that depend on this pipeline
	for (auto &parent_entry : parents) {
		auto parent = parent_entry.lock();
		if (!parent) { // LCOV_EXCL_START
			continue;
		} // LCOV_EXCL_STOP
		// mark a dependency as completed for each of the parents
		parent->CompleteDependency();
	}
	FinalizeFinish();
}

// PipelineEvent::Schedule
// -> Pipeline::Schedule -> ScheduleParallel -> LaunchScanTasks 
bool Pipeline::LaunchScanTasks(shared_ptr<Event> &event, idx_t max_threads) {
	// split the scan up into parts and schedule the parts
	auto &scheduler = TaskScheduler::GetScheduler(executor.context);
	idx_t active_threads = scheduler.NumberOfThreads();
	if (max_threads > active_threads) {
		max_threads = active_threads;
	}
	if (max_threads <= 1) {
		// too small to parallelize
		return false;
	}

	// launch a task for every thread
  // 并行任务 PipelineTask，以pipeline 为单位
	vector<shared_ptr<Task>> tasks;
	for (idx_t i = 0; i < max_threads; i++) {
		tasks.push_back(make_uniq<PipelineTask>(*this, event));
	}
  // 任务写入TaskScheduler 队列，等待调度
	event->SetTasks(std::move(tasks));
	return true;
}  

// PipelineFinishEvent::FinishEvent
void PipelineFinishEvent::FinishEvent() {
	pipeline->Finalize(*this);
}
// Pipeline::Finalize
void Pipeline::Finalize(Event &event) {
  ...
  auto sink_state = sink->Finalize(*this, event, executor.context, *sink->sink_state);
	sink->sink_state->state = sink_state;
  ...
}

// PipelineCompleteEvent::FinalizeFinish
void PipelineCompleteEvent::FinalizeFinish() {
	if (complete_pipeline) {
		executor.CompletePipeline();
	}
}
// Executor::CompletePipeline
void CompletePipeline() {
  completed_pipelines++;
}