diff --git a/src/common/settings.json b/src/common/settings.json index 9a4af33d5174..88dc8d8520c8 100644 --- a/src/common/settings.json +++ b/src/common/settings.json @@ -514,6 +514,14 @@ "default_scope": "global", "default_value": "false" }, + { + "name": "enable_pipeline_trace", + "description": "Enables pipeline trace logging: prints the pipeline graph and outputs a Chrome Trace JSON timing report to stderr on query completion", + "type": "BOOLEAN", + "scope": "local", + "struct": "EnablePipelineTraceSetting", + "custom_implementation": true + }, { "name": "enable_profiling", "description": "Enables profiling, and sets the output format (JSON, QUERY_TREE, QUERY_TREE_OPTIMIZER)", @@ -895,6 +903,22 @@ "scope": "global", "default_value": "auto" }, + { + "name": "pipeline_graph_output", + "description": "File path for pipeline graph output when enable_pipeline_trace is true (empty = stderr)", + "type": "VARCHAR", + "scope": "local", + "struct": "PipelineGraphOutputSetting", + "custom_implementation": true + }, + { + "name": "pipeline_trace_output", + "description": "File path for Chrome Trace JSON output when enable_pipeline_trace is true (empty = stderr)", + "type": "VARCHAR", + "scope": "local", + "struct": "PipelineTraceOutputSetting", + "custom_implementation": true + }, { "name": "pivot_filter_threshold", "description": "The threshold to switch from using filtered aggregates to LIST with a dedicated pivot operator", @@ -1119,4 +1143,4 @@ "scope": "global", "default_value": "4096" } -] \ No newline at end of file +] diff --git a/src/include/duckdb/execution/executor.hpp b/src/include/duckdb/execution/executor.hpp index 46315a989828..83486fc74df0 100644 --- a/src/include/duckdb/execution/executor.hpp +++ b/src/include/duckdb/execution/executor.hpp @@ -155,6 +155,10 @@ class Executor { vector> pipelines; //! The root pipelines of the query vector> root_pipelines; + //! Pipelines saved for trace output (persists after pipelines.clear()) + vector> traced_pipelines; + //! Steady-clock nanoseconds at query initialization (base for Chrome trace timestamps) + int64_t pipeline_trace_start_ns = 0; //! The recursive CTE's in this query plan vector> recursive_ctes; //! The pipeline executor for the root pipeline diff --git a/src/include/duckdb/main/client_config.hpp b/src/include/duckdb/main/client_config.hpp index 893ddcb318bf..2528af58021f 100644 --- a/src/include/duckdb/main/client_config.hpp +++ b/src/include/duckdb/main/client_config.hpp @@ -101,6 +101,14 @@ struct ClientConfig { //! If HTTP logging is enabled or not. bool enable_http_logging = true; + //! If pipeline trace logging is enabled (SET enable_pipeline_trace = true). + //! Prints the pipeline dependency graph and Chrome Trace JSON timing to stderr. + bool enable_pipeline_trace = false; + //! File path for pipeline graph output (empty = stderr). + string pipeline_graph_output; + //! File path for Chrome Trace JSON output (empty = stderr). + string pipeline_trace_output; + //! **DEPRECATED** The file to save query HTTP logging information to, instead of printing it to the console //! (empty = output to the DuckDB logger) string http_logging_output; diff --git a/src/include/duckdb/main/settings.hpp b/src/include/duckdb/main/settings.hpp index cb339df2ea24..114f93aa5b32 100644 --- a/src/include/duckdb/main/settings.hpp +++ b/src/include/duckdb/main/settings.hpp @@ -780,6 +780,40 @@ struct EnableObjectCacheSetting { static constexpr idx_t SettingIndex = 42; }; +struct EnablePipelineTraceSetting { + using RETURN_TYPE = bool; + static constexpr const char *Name = "enable_pipeline_trace"; + static constexpr const char *Description = + "Enables pipeline trace logging: prints the pipeline graph and outputs a Chrome Trace JSON " + "timing report to stderr on query completion"; + static constexpr const char *InputType = "BOOLEAN"; + static void SetLocal(ClientContext &context, const Value ¶meter); + static void ResetLocal(ClientContext &context); + static Value GetSetting(const ClientContext &context); +}; + +struct PipelineGraphOutputSetting { + using RETURN_TYPE = string; + static constexpr const char *Name = "pipeline_graph_output"; + static constexpr const char *Description = + "File path for pipeline graph output when enable_pipeline_trace is true (empty = stderr)"; + static constexpr const char *InputType = "VARCHAR"; + static void SetLocal(ClientContext &context, const Value ¶meter); + static void ResetLocal(ClientContext &context); + static Value GetSetting(const ClientContext &context); +}; + +struct PipelineTraceOutputSetting { + using RETURN_TYPE = string; + static constexpr const char *Name = "pipeline_trace_output"; + static constexpr const char *Description = + "File path for Chrome Trace JSON output when enable_pipeline_trace is true (empty = stderr)"; + static constexpr const char *InputType = "VARCHAR"; + static void SetLocal(ClientContext &context, const Value ¶meter); + static void ResetLocal(ClientContext &context); + static Value GetSetting(const ClientContext &context); +}; + struct EnableProfilingSetting { using RETURN_TYPE = string; static constexpr const char *Name = "enable_profiling"; diff --git a/src/include/duckdb/parallel/pipeline.hpp b/src/include/duckdb/parallel/pipeline.hpp index 0d8a665a5d65..d4a9218410fe 100644 --- a/src/include/duckdb/parallel/pipeline.hpp +++ b/src/include/duckdb/parallel/pipeline.hpp @@ -43,6 +43,15 @@ class PipelineTask : public ExecutorTask { public: TaskExecutionResult ExecuteTask(TaskExecutionMode mode) override; + +private: + //! Wall-clock start time of this task's first ExecuteTask invocation (-1 = not yet started). + //! Captured once even if the task yields and is rescheduled, so the recorded interval covers + //! the whole task lifetime including any blocked periods between Execute calls. + int64_t task_start_ns = -1; + //! std::hash of this->thread::id at the time of first ExecuteTask. Used as Chrome Trace tid + //! so each worker thread gets its own row in Perfetto. + uint64_t task_thread_hash = 0; }; class PipelineBuildState { @@ -76,6 +85,7 @@ class Pipeline : public enable_shared_from_this { friend class PipelineFinishEvent; friend class PipelineBuildState; friend class MetaPipeline; + friend class PipelineTracer; public: explicit Pipeline(Executor &execution_context); @@ -100,6 +110,22 @@ class Pipeline : public enable_shared_from_this { void Print() const; void PrintDependencies() const; + //! Per-PipelineTask timing record. One entry per parallel task that ran this pipeline, + //! capturing first-Execute-entry → TASK_FINISHED return wall-clock interval and the + //! worker thread's hash. + struct TaskTiming { + int64_t start_ns; + int64_t end_ns; + uint64_t thread_hash; + }; + + //! Append a task timing record. Thread-safe; called by PipelineTask on completion. + void RecordTaskTiming(int64_t start_ns, int64_t end_ns, uint64_t thread_hash); + + idx_t GetPipelineId() const { + return pipeline_id; + } + //! Returns query progress bool GetProgress(ProgressData &progress_data); @@ -132,6 +158,13 @@ class Pipeline : public enable_shared_from_this { atomic initialized; //! The source of this pipeline optional_ptr source; + + //! Pipeline ID assigned by PipelineTracer::AssignIds (0-based) + idx_t pipeline_id = 0; + //! Per-task timing records. Appended to from worker threads via RecordTaskTiming under + //! task_timings_lock; read once at query end by PipelineTracer::PrintChromeTrace. + mutable mutex task_timings_lock; + vector task_timings; //! The chain of intermediate operators vector> operators; //! The sink (i.e. destination) for data; this is e.g. a hash table to-be-built diff --git a/src/include/duckdb/parallel/pipeline_tracer.hpp b/src/include/duckdb/parallel/pipeline_tracer.hpp new file mode 100644 index 000000000000..96197cdeb01e --- /dev/null +++ b/src/include/duckdb/parallel/pipeline_tracer.hpp @@ -0,0 +1,50 @@ +//===----------------------------------------------------------------------===// +// DuckDB +// +// duckdb/parallel/pipeline_tracer.hpp +// +// +//===----------------------------------------------------------------------===// + +#pragma once + +#include "duckdb/common/common.hpp" + +namespace duckdb { + +class Pipeline; + +//! PipelineTracer collects and outputs pipeline structure and execution timing. +//! Enabled by SET enable_pipeline_trace = true. +//! +//! Output destinations (empty string = stderr): +//! SET pipeline_graph_output = '/path/graph.txt' -- pipeline dependency graph +//! SET pipeline_trace_output = '/path/trace.json' -- Chrome Trace JSON (Perfetto) +class PipelineTracer { +public: + //! Assign sequential IDs (0, 1, 2, ...) to all pipelines. + //! Must be called before PrintGraph or PrintChromeTrace. + static void AssignIds(vector> &pipelines); + + //! Print the static pipeline structure and dependency graph. + //! output_path: file path to write to, or empty string for stderr. + static void PrintGraph(const vector> &pipelines, const string &output_path); + + //! Print Chrome Trace JSON (loadable in https://ui.perfetto.dev/). + //! query_start_ns: steady_clock nanoseconds since epoch at query start. + //! output_path: file path to write to, or empty string for stderr. + static void PrintChromeTrace(const vector> &pipelines, int64_t query_start_ns, + const string &output_path); + +private: + //! Build a short human-readable description: "TableScan→HashJoinBuild→..." + static string Describe(const Pipeline &pipeline); + + //! Write content to output_path if non-empty, otherwise to stderr. + //! append=true opens the file in append mode (used for the graph, which accumulates + //! across queries); append=false truncates (used for the Chrome Trace JSON, which + //! must remain a single valid JSON object per file). + static void WriteOutput(const string &content, const string &output_path, bool append); +}; + +} // namespace duckdb diff --git a/src/main/config.cpp b/src/main/config.cpp index f5f56660bf65..b2e921cb2f0e 100644 --- a/src/main/config.cpp +++ b/src/main/config.cpp @@ -128,6 +128,7 @@ static const ConfigurationOption internal_options[] = { DUCKDB_GLOBAL(EnableLogging), DUCKDB_SETTING(EnableMacroDependenciesSetting), DUCKDB_SETTING(EnableObjectCacheSetting), + DUCKDB_LOCAL(EnablePipelineTraceSetting), DUCKDB_LOCAL(EnableProfilingSetting), DUCKDB_LOCAL(EnableProgressBarSetting), DUCKDB_LOCAL(EnableProgressBarPrintSetting), @@ -177,6 +178,8 @@ static const ConfigurationOption internal_options[] = { DUCKDB_SETTING(PasswordSetting), DUCKDB_SETTING_CALLBACK(PerfectHtThresholdSetting), DUCKDB_SETTING_CALLBACK(PinThreadsSetting), + DUCKDB_LOCAL(PipelineGraphOutputSetting), + DUCKDB_LOCAL(PipelineTraceOutputSetting), DUCKDB_SETTING(PivotFilterThresholdSetting), DUCKDB_SETTING(PivotLimitSetting), DUCKDB_SETTING(PreferRangeJoinsSetting), @@ -208,12 +211,12 @@ static const ConfigurationOption internal_options[] = { DUCKDB_SETTING(ZstdMinStringLengthSetting), FINAL_SETTING}; -static const ConfigurationAlias setting_aliases[] = {DUCKDB_SETTING_ALIAS("memory_limit", 99), +static const ConfigurationAlias setting_aliases[] = {DUCKDB_SETTING_ALIAS("memory_limit", 100), DUCKDB_SETTING_ALIAS("null_order", 43), - DUCKDB_SETTING_ALIAS("profiling_output", 118), - DUCKDB_SETTING_ALIAS("user", 133), + DUCKDB_SETTING_ALIAS("profiling_output", 121), + DUCKDB_SETTING_ALIAS("user", 136), DUCKDB_SETTING_ALIAS("wal_autocheckpoint", 25), - DUCKDB_SETTING_ALIAS("worker_threads", 132), + DUCKDB_SETTING_ALIAS("worker_threads", 135), FINAL_ALIAS}; vector DBConfig::GetOptions() { diff --git a/src/main/settings/custom_settings.cpp b/src/main/settings/custom_settings.cpp index ba82c71d5826..5e7f394f50da 100644 --- a/src/main/settings/custom_settings.cpp +++ b/src/main/settings/custom_settings.cpp @@ -1205,6 +1205,54 @@ Value EnableHTTPLoggingSetting::GetSetting(const ClientContext &context) { return Value::BOOLEAN(config.enable_http_logging); } +//===----------------------------------------------------------------------===// +// Enable Pipeline Trace +//===----------------------------------------------------------------------===// + +void EnablePipelineTraceSetting::SetLocal(ClientContext &context, const Value &input) { + ClientConfig::GetConfig(context).enable_pipeline_trace = input.GetValue(); +} + +void EnablePipelineTraceSetting::ResetLocal(ClientContext &context) { + ClientConfig::GetConfig(context).enable_pipeline_trace = ClientConfig().enable_pipeline_trace; +} + +Value EnablePipelineTraceSetting::GetSetting(const ClientContext &context) { + return Value::BOOLEAN(ClientConfig::GetConfig(context).enable_pipeline_trace); +} + +//===----------------------------------------------------------------------===// +// Pipeline Graph Output +//===----------------------------------------------------------------------===// + +void PipelineGraphOutputSetting::SetLocal(ClientContext &context, const Value &input) { + ClientConfig::GetConfig(context).pipeline_graph_output = input.GetValue(); +} + +void PipelineGraphOutputSetting::ResetLocal(ClientContext &context) { + ClientConfig::GetConfig(context).pipeline_graph_output = ClientConfig().pipeline_graph_output; +} + +Value PipelineGraphOutputSetting::GetSetting(const ClientContext &context) { + return Value(ClientConfig::GetConfig(context).pipeline_graph_output); +} + +//===----------------------------------------------------------------------===// +// Pipeline Trace Output +//===----------------------------------------------------------------------===// + +void PipelineTraceOutputSetting::SetLocal(ClientContext &context, const Value &input) { + ClientConfig::GetConfig(context).pipeline_trace_output = input.GetValue(); +} + +void PipelineTraceOutputSetting::ResetLocal(ClientContext &context) { + ClientConfig::GetConfig(context).pipeline_trace_output = ClientConfig().pipeline_trace_output; +} + +Value PipelineTraceOutputSetting::GetSetting(const ClientContext &context) { + return Value(ClientConfig::GetConfig(context).pipeline_trace_output); +} + //===----------------------------------------------------------------------===// // Enable Mbedtls //===----------------------------------------------------------------------===// diff --git a/src/parallel/CMakeLists.txt b/src/parallel/CMakeLists.txt index 307168ff49f8..ae366ef47727 100644 --- a/src/parallel/CMakeLists.txt +++ b/src/parallel/CMakeLists.txt @@ -10,6 +10,7 @@ add_library_unity( interrupt.cpp pipeline.cpp pipeline_complete_event.cpp + pipeline_tracer.cpp pipeline_event.cpp pipeline_executor.cpp pipeline_finish_event.cpp diff --git a/src/parallel/executor.cpp b/src/parallel/executor.cpp index bc8b75723359..b5a4e4d21638 100644 --- a/src/parallel/executor.cpp +++ b/src/parallel/executor.cpp @@ -1,5 +1,6 @@ #include "duckdb/execution/executor.hpp" +#include "duckdb/common/chrono.hpp" #include "duckdb/execution/execution_context.hpp" #include "duckdb/execution/operator/helper/physical_result_collector.hpp" #include "duckdb/execution/operator/scan/physical_table_scan.hpp" @@ -16,6 +17,7 @@ #include "duckdb/parallel/pipeline_finish_event.hpp" #include "duckdb/parallel/pipeline_initialize_event.hpp" #include "duckdb/parallel/pipeline_prepare_finish_event.hpp" +#include "duckdb/parallel/pipeline_tracer.hpp" #include "duckdb/parallel/task_scheduler.hpp" #include "duckdb/parallel/thread_context.hpp" @@ -424,6 +426,16 @@ void Executor::InitializeInternal(PhysicalOperator &plan) { // finally, verify and schedule VerifyPipelines(); ScheduleEvents(to_schedule); + + // pipeline trace: assign IDs, record query start time, and print the static graph + auto &trace_config = ClientConfig::GetConfig(context); + if (trace_config.enable_pipeline_trace) { + PipelineTracer::AssignIds(pipelines); + pipeline_trace_start_ns = static_cast( + duration_cast(steady_clock::now().time_since_epoch()).count()); + PipelineTracer::PrintGraph(pipelines, trace_config.pipeline_graph_output); + traced_pipelines = pipelines; + } } } @@ -622,6 +634,12 @@ PendingExecutionResult Executor::ExecuteTask(bool dry_run) { D_ASSERT(!task); lock_guard elock(executor_lock); + // emit Chrome Trace JSON before clearing pipelines (timing data lives in Pipeline objects) + if (!traced_pipelines.empty()) { + auto &trace_config = ClientConfig::GetConfig(context); + PipelineTracer::PrintChromeTrace(traced_pipelines, pipeline_trace_start_ns, trace_config.pipeline_trace_output); + traced_pipelines.clear(); + } pipelines.clear(); NextExecutor(); if (HasError()) { // LCOV_EXCL_START @@ -647,6 +665,8 @@ void Executor::Reset() { events.clear(); to_be_rescheduled_tasks.clear(); execution_result = PendingExecutionResult::RESULT_NOT_READY; + traced_pipelines.clear(); + pipeline_trace_start_ns = 0; } shared_ptr Executor::CreateChildPipeline(Pipeline ¤t, PhysicalOperator &op) { diff --git a/src/parallel/pipeline.cpp b/src/parallel/pipeline.cpp index d48cc1241cb7..8a94d2d517b0 100644 --- a/src/parallel/pipeline.cpp +++ b/src/parallel/pipeline.cpp @@ -1,6 +1,9 @@ #include "duckdb/parallel/pipeline.hpp" +#include + #include "duckdb/common/algorithm.hpp" +#include "duckdb/common/chrono.hpp" #include "duckdb/common/printer.hpp" #include "duckdb/common/tree_renderer/text_tree_renderer.hpp" #include "duckdb/execution/executor.hpp" @@ -30,7 +33,16 @@ const PipelineExecutor &PipelineTask::GetPipelineExecutor() const { return *pipeline_executor; } +static int64_t SteadyNowNs() { + return static_cast(duration_cast(steady_clock::now().time_since_epoch()).count()); +} + TaskExecutionResult PipelineTask::ExecuteTask(TaskExecutionMode mode) { + if (task_start_ns < 0) { + task_start_ns = SteadyNowNs(); + task_thread_hash = std::hash {}(std::this_thread::get_id()); + } + if (!pipeline_executor) { pipeline_executor = make_uniq(pipeline.GetClientContext(), pipeline); } @@ -60,6 +72,7 @@ TaskExecutionResult PipelineTask::ExecuteTask(TaskExecutionMode mode) { } } + pipeline.RecordTaskTiming(task_start_ns, SteadyNowNs(), task_thread_hash); event->FinishTask(); pipeline_executor.reset(); return TaskExecutionResult::TASK_FINISHED; @@ -73,6 +86,11 @@ ClientContext &Pipeline::GetClientContext() { return executor.context; } +void Pipeline::RecordTaskTiming(int64_t start_ns, int64_t end_ns, uint64_t thread_hash) { + lock_guard lock(task_timings_lock); + task_timings.push_back({start_ns, end_ns, thread_hash}); +} + bool Pipeline::GetProgress(ProgressData &progress) { D_ASSERT(source); idx_t source_cardinality = MinValue(source->estimated_cardinality, 1ULL << 48ULL); diff --git a/src/parallel/pipeline_tracer.cpp b/src/parallel/pipeline_tracer.cpp new file mode 100644 index 000000000000..3a0d206515c7 --- /dev/null +++ b/src/parallel/pipeline_tracer.cpp @@ -0,0 +1,105 @@ +#include "duckdb/parallel/pipeline_tracer.hpp" + +#include "duckdb/common/printer.hpp" +#include "duckdb/common/string_util.hpp" +#include "duckdb/execution/physical_operator.hpp" +#include "duckdb/parallel/pipeline.hpp" + +#include + +namespace duckdb { + +void PipelineTracer::AssignIds(vector> &pipelines) { + for (idx_t i = 0; i < pipelines.size(); i++) { + pipelines[i]->pipeline_id = i; + } +} + +string PipelineTracer::Describe(const Pipeline &pipeline) { + auto ops = pipeline.GetOperators(); + string result; + for (idx_t i = 0; i < ops.size(); i++) { + if (i > 0) { + result += "→"; + } + result += ops[i].get().GetName(); + } + return result; +} + +void PipelineTracer::WriteOutput(const string &content, const string &output_path, bool append) { + if (output_path.empty()) { + Printer::Print(OutputStream::STREAM_STDERR, content); + return; + } + auto mode = std::ios::out | (append ? std::ios::app : std::ios::trunc); + std::ofstream f(output_path, mode); + if (!f.is_open()) { + // Fall back to stderr if the file cannot be opened + Printer::Print(OutputStream::STREAM_STDERR, + "pipeline_tracer: could not open '" + output_path + "', falling back to stderr\n" + content); + return; + } + f << content; +} + +void PipelineTracer::PrintGraph(const vector> &pipelines, const string &output_path) { + string out = "\n=== Pipeline Graph ===\n"; + for (auto &p : pipelines) { + out += "Pipeline #" + to_string(p->pipeline_id) + ": " + Describe(*p) + "\n"; + auto deps = p->GetDependencies(); + if (!deps.empty()) { + out += " depends on:"; + for (auto &dep : deps) { + auto locked = dep.lock(); + if (locked) { + out += " #" + to_string(locked->pipeline_id); + } + } + out += "\n"; + } + } + out += "======================\n"; + // Append: the graph is plain text and the per-query "=== Pipeline Graph ===" header + // already separates entries, so multiple queries in the same session accumulate cleanly. + WriteOutput(out, output_path, /*append=*/true); +} + +void PipelineTracer::PrintChromeTrace(const vector> &pipelines, int64_t query_start_ns, + const string &output_path) { + string events; + bool first = true; + for (auto &p : pipelines) { + string desc = Describe(*p); + // Escape quotes for JSON safety; do it once per pipeline, not per task. + desc = StringUtil::Replace(desc, "\"", "\\\""); + // One Chrome Trace event per PipelineTask: each parallel slice of the pipeline gets + // its own bar. tid = thread hash so each worker thread occupies its own row in + // Perfetto, making real parallelism visible. + idx_t task_idx = 0; + for (auto &t : p->task_timings) { + int64_t start_us = (t.start_ns - query_start_ns) / 1000; + int64_t dur_us = (t.end_ns - t.start_ns) / 1000; + if (dur_us < 0) { + dur_us = 0; + } + // Mask to a 31-bit non-negative value for tools that treat tid as signed int32. + uint64_t tid = t.thread_hash & 0x7fffffffULL; + if (!first) { + events += ",\n "; + } else { + first = false; + } + events += "{\"name\":\"#" + to_string(p->pipeline_id) + "[" + to_string(task_idx) + "]: " + desc + + "\",\"ph\":\"X\",\"pid\":0,\"tid\":" + to_string(tid) + + ",\"ts\":" + to_string(start_us) + ",\"dur\":" + to_string(dur_us) + "}"; + task_idx++; + } + } + string json = "{\"traceEvents\":[\n " + events + "\n]}\n"; + // Truncate: Chrome Trace JSON must be a single valid JSON object; concatenating + // per-query traces would produce invalid JSON. Each query overwrites the file. + WriteOutput(json, output_path, /*append=*/false); +} + +} // namespace duckdb