test(query): add trace_debug OTLP dump utilities for direct and HTTP query tracing

Author: forsaken628

Cargo.lock

@@ -521,12 +521,12 @@ vergen-gix = { version = "1", features = ["build", "cargo", "rustc"] }
 
 # Observability
 env_logger = "0.11"
-fastrace = { version = "0.7.14", features = ["enable"] }
-fastrace-opentelemetry = "0.10"
+fastrace = { version = "0.7.17", features = ["enable"] }
+fastrace-opentelemetry = "0.16"
 log = { version = "0.4.27", features = ["serde", "kv_serde", "kv_unstable_std"] }
 logcall = "0.1.9"
-opentelemetry = { version = "0.29", features = ["trace", "logs"] }
-opentelemetry-otlp = { version = "0.29", default-features = false, features = [
+opentelemetry = { version = "0.31", features = ["trace", "logs"] }
+opentelemetry-otlp = { version = "0.31", default-features = false, features = [
     "trace",
     "logs",
     "metrics",
@@ -536,7 +536,8 @@ opentelemetry-otlp = { version = "0.29", default-features = false, features = [
     "http-proto",
     "reqwest-client",
 ] }
-opentelemetry_sdk = { version = "0.29", features = ["trace", "logs", "rt-tokio"] }
+opentelemetry-proto = { version = "0.31", features = ["gen-tonic-messages", "trace", "with-serde"] }
+opentelemetry_sdk = { version = "0.31", features = ["trace", "logs", "rt-tokio"] }
 prometheus-client = "0.22"
 prometheus-parse = "0.2.3"
 tracing = "0.1.40"
@@ -548,6 +549,9 @@ sqllogictest = "0.28.4"
 
 [workspace.lints.rust]
 async_fn_in_trait = "allow"
+# Temporary workaround for false positives from async_backtrace::framed and
+# fastrace::trace macro expansion under `cargo clippy --workspace --all-targets -- -D warnings`.
+unused_braces = "allow"
 
 [workspace.lints.clippy]
 useless_format = "allow"

Cargo.toml

@@ -2,6 +2,7 @@
 
 - For Rust changes that will remain in the branch, use `cargo clippy` to confirm there are no compilation or lint errors.
 - Start with partial verification when a full workspace pass is too expensive, but move toward stronger coverage before handoff when the resulting changes will remain in the branch.
+- Use the trace-debug utilities under [`src/query/service/tests/it/trace_debug/README.md`](../src/query/service/tests/it/trace_debug/README.md) when the question is about query lifecycle or stage transitions, such as parser/planner/interpreter/pipeline flow, HTTP query state and pagination behavior, or which execution spans and attached events appear for a specific SQL path.
 - For temporary investigation outputs that will not be submitted, keep validation minimal and purpose-driven. Run only the checks needed to establish the conclusion or unblock the investigation.
 - If investigation work begins producing code, tests, or docs that should remain in the branch, raise the validation bar before handoff.
 

agents/debug-and-validation.md

@@ -198,7 +198,6 @@ pub fn init_logging(
             let reporter = rt.block_on(async {
                 fastrace_opentelemetry::OpenTelemetryReporter::new(
                     exporter,
-                    opentelemetry::trace::SpanKind::Server,
                     Cow::Owned(resource),
                     opentelemetry::InstrumentationScope::builder(trace_name).build(),
                 )

src/common/tracing/src/init.rs

@@ -125,6 +125,13 @@ unsafe impl Send for ProcessorPtr {}
 unsafe impl Sync for ProcessorPtr {}
 
 impl ProcessorPtr {
+    fn current_thread_name() -> String {
+        std::thread::current()
+            .name()
+            .unwrap_or("unnamed")
+            .to_string()
+    }
+
     #[allow(clippy::arc_with_non_send_sync)]
     pub fn create(inner: Box<dyn Processor>) -> ProcessorPtr {
         ProcessorPtr {
@@ -174,8 +181,10 @@ impl ProcessorPtr {
     /// # Safety
     pub unsafe fn process(&self) -> Result<()> {
         unsafe {
+            let thread_name = Self::current_thread_name();
             let span = LocalSpan::enter_with_local_parent(format!("{}::process", self.name()))
-                .with_property(|| ("graph-node-id", self.id().index().to_string()));
+                .with_property(|| ("graph-node-id", self.id().index().to_string()))
+                .with_property(|| ("thread_name", thread_name.clone()));
 
             match (*self.inner.get()).process() {
                 Ok(_) => Ok(()),
@@ -201,6 +210,7 @@ impl ProcessorPtr {
             let id = self.id();
             let mut name = self.name();
             name.push_str("::async_process");
+            let thread_name = Self::current_thread_name();
 
             let task = (*self.inner.get()).async_process();
 
@@ -216,7 +226,8 @@ impl ProcessorPtr {
             let inner = self.inner.clone();
             async move {
                 let span = Span::enter_with_local_parent(name)
-                    .with_property(|| ("graph-node-id", id.index().to_string()));
+                    .with_property(|| ("graph-node-id", id.index().to_string()))
+                    .with_property(|| ("thread_name", thread_name.clone()));
 
                 match task.await {
                     Ok(_) => {

src/query/pipeline/src/core/processor.rs

@@ -182,13 +182,15 @@ zip = { workspace = true }
 arrow-cast = { workspace = true }
 databend-common-sql-test-support = { workspace = true }
 databend-storages-common-pruner = { workspace = true }
+fastrace-opentelemetry = { workspace = true }
 geo = { workspace = true }
 goldenfile = { workspace = true }
 hex = { workspace = true }
 hyper-util = { workspace = true }
 jwt-simple = { workspace = true }
 maplit = { workspace = true }
 mysql_async = { workspace = true }
+opentelemetry-proto = { workspace = true }
 p256 = { workspace = true }
 pretty_assertions = { workspace = true }
 proptest = { workspace = true }

src/query/service/Cargo.toml

@@ -41,6 +41,7 @@ use databend_common_sql::plans::Plan;
 use databend_storages_common_cache::CacheManager;
 use derive_visitor::DriveMut;
 use derive_visitor::VisitorMut;
+use fastrace::collector::SpanContext;
 use md5::Digest;
 use md5::Md5;
 
@@ -161,7 +162,11 @@ pub trait Interpreter: Sync + Send {
             complete_executor.execute()?;
             self.inject_result()
         } else {
-            let pulling_executor = PipelinePullingExecutor::from_pipelines(build_res, settings)?;
+            let thread_span_parent = ctx
+                .get_executor_tracing_context()
+                .or_else(SpanContext::current_local_parent);
+            let pulling_executor =
+                PipelinePullingExecutor::from_pipelines(build_res, settings, thread_span_parent)?;
 
             ctx.set_executor(pulling_executor.get_inner())?;
             Ok(Box::pin(ProgressStream::try_create(

src/query/service/src/interpreters/interpreter.rs

@@ -43,6 +43,7 @@ use databend_common_storages_basic::gen_result_cache_key;
 use databend_common_storages_fuse::FuseLazyPartInfo;
 use databend_common_storages_fuse::FuseTable;
 use databend_common_users::UserApiProvider;
+use fastrace::collector::SpanContext;
 use serde::Serialize;
 use serde_json;
 
@@ -552,7 +553,15 @@ impl ExplainInterpreter {
                 executor.execute()?;
             }
             false => {
-                let mut executor = PipelinePullingExecutor::from_pipelines(build_res, settings)?;
+                let thread_span_parent = self
+                    .ctx
+                    .get_executor_tracing_context()
+                    .or_else(SpanContext::current_local_parent);
+                let mut executor = PipelinePullingExecutor::from_pipelines(
+                    build_res,
+                    settings,
+                    thread_span_parent,
+                )?;
                 executor.start();
                 while (executor.pull_data()?).is_some() {}
             }

src/query/service/src/interpreters/interpreter_explain.rs

@@ -34,6 +34,7 @@ use databend_common_pipeline::core::always_callback;
 use databend_common_sql::Planner;
 use databend_meta_plugin_semaphore::acquirer::Permit;
 use databend_meta_runtime::DatabendRuntime;
+use fastrace::collector::SpanContext;
 
 use crate::interpreters::Interpreter;
 use crate::interpreters::InterpreterFactory;
@@ -174,7 +175,11 @@ impl ExplainPerfInterpreter {
             ctx.set_executor(executor.get_inner())?;
             executor.execute()?;
         } else {
-            let mut executor = PipelinePullingExecutor::from_pipelines(build_res, settings)?;
+            let thread_span_parent = ctx
+                .get_executor_tracing_context()
+                .or_else(SpanContext::current_local_parent);
+            let mut executor =
+                PipelinePullingExecutor::from_pipelines(build_res, settings, thread_span_parent)?;
             ctx.set_executor(executor.get_inner())?;
             executor.start();
             while (executor.pull_data()?).is_some() {}

src/query/service/src/interpreters/interpreter_explain_perf.rs

@@ -21,7 +21,6 @@ use databend_common_base::runtime::drop_guard;
 use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_pipeline::core::Pipeline;
-use fastrace::func_path;
 use fastrace::prelude::*;
 
 use crate::pipelines::executor::ExecutorSettings;
@@ -101,11 +100,21 @@ impl PipelineCompleteExecutor {
         .flatten()
     }
 
-    fn thread_function(&self) -> impl Fn() -> Result<()> + use<> {
-        let span = Span::enter_with_local_parent(func_path!());
+    fn thread_function(&self) -> impl FnOnce() -> Result<()> + use<> {
+        let parent = SpanContext::current_local_parent();
         let executor = self.executor.clone();
 
         move || {
+            let span = if let Some(parent) = parent {
+                let thread_name = std::thread::current()
+                    .name()
+                    .unwrap_or("unnamed")
+                    .to_string();
+                Span::root("PipelineCompleteExecutor::thread_function", parent)
+                    .with_property(|| ("thread_name", thread_name))
+            } else {
+                Span::noop()
+            };
             let _g = span.set_local_parent();
             executor.execute()
         }