use std::ffi::CString;
use std::sync::atomic::AtomicU64;
use perfetto_sdk::track_event::{
EventContext, TrackEventDebugArg, TrackEventTrack, TrackEventType,
};
use perfetto_sdk::{track_event, track_event_begin, track_event_end, track_event_instant};
use tracing::{
Event, Subscriber,
field::{Field, Visit},
span::{Attributes, Id, Record},
};
use tracing_subscriber::{Layer, layer::Context, registry::LookupSpan};
use super::{
SemanticTrack, delta_scan_output_track, diagnostics_track, operation_track, owner_track,
perfetto_te_ns, phase_track, planning_track, query_track, worker_track,
};
use crate::profiling::{
OBJECT_STORE_TRANSPORT_CONTEXT_NAME, OBJECT_STORE_TRANSPORT_DISPLAY_NAME, allocate_id,
};
pub const PROFILE_TARGET: &str = "delta_funnel::profile";
const TIBERIUS_PROFILE_TARGET: &str = "tiberius_raw_bulk::protocol";
const BULK_FINALIZE_PREPARE: &str = "protocol.bulk_load.finalize.prepare";
const BULK_FINALIZE_WRITE: &str = "protocol.bulk_load.finalize.write";
const BULK_FINALIZE_FLUSH: &str = "protocol.bulk_load.finalize.flush";
const BULK_FINALIZE_RESULT: &str = "protocol.bulk_load.finalize.result";
static NEXT_PROFILE_CONTEXT_ID: AtomicU64 = AtomicU64::new(1);
pub fn is_profile_target(target: &str) -> bool {
matches!(target, PROFILE_TARGET | TIBERIUS_PROFILE_TARGET)
}
#[derive(Debug, Default)]
pub struct PerfettoProfileLayer;
impl<S> Layer<S> for PerfettoProfileLayer
where
S: Subscriber + for<'lookup> LookupSpan<'lookup>,
{
fn on_new_span(&self, attributes: &Attributes<'_>, id: &Id, context: Context<'_, S>) {
let Some(span) = context.span(id) else {
return;
};
let metadata = attributes.metadata();
let active = if metadata.target() == PROFILE_TARGET {
let mut fields = ProfileFields::default();
attributes.record(&mut fields);
ActiveProfileSpan::from_fields(metadata.name(), fields)
} else if metadata.target() == TIBERIUS_PROFILE_TARGET {
let Some(name) = dependency_span_label(metadata.name()) else {
return;
};
let mut ancestor = span.parent();
loop {
let Some(parent) = ancestor else {
break None;
};
let inherited = parent
.extensions()
.get::<ActiveProfileSpan>()
.and_then(|active| active.inherit(name));
if inherited.is_some() {
break inherited;
}
ancestor = parent.parent();
}
} else {
None
};
let Some(active) = active else {
return;
};
active.emit_begin();
span.extensions_mut().insert(active);
}
fn on_record(&self, id: &Id, values: &Record<'_>, context: Context<'_, S>) {
let Some(span) = context.span(id) else {
return;
};
let mut fields = ProfileFields::default();
values.record(&mut fields);
if let Some(active) = span.extensions_mut().get_mut::<ActiveProfileSpan>() {
active.record(fields);
}
}
fn on_enter(&self, id: &Id, context: Context<'_, S>) {
let Some(span) = context.span(id) else {
return;
};
if let Some(active) = span.extensions_mut().get_mut::<ActiveProfileSpan>() {
active.emit_context_enter();
}
}
fn on_exit(&self, id: &Id, context: Context<'_, S>) {
let Some(span) = context.span(id) else {
return;
};
if span
.extensions()
.get::<ActiveProfileSpan>()
.is_some_and(ActiveProfileSpan::context_event_enabled)
{
track_event_end!("delta_funnel.profile.context");
}
}
fn on_close(&self, id: Id, context: Context<'_, S>) {
let Some(span) = context.span(&id) else {
return;
};
if let Some(active) = span.extensions_mut().remove::<ActiveProfileSpan>() {
active.emit_end();
}
}
fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) {
if event.metadata().name() != "Operator activity trace truncated" {
return;
}
let mut fields = ProfileFields::default();
event.record(&mut fields);
let (Some(operation_id), Some(maximum_spans)) = (fields.operation_id, fields.maximum_spans)
else {
return;
};
let diagnostics = diagnostics_track(TrackEventTrack::process_track_uuid());
let operation = operation_track(operation_id, diagnostics.uuid);
track_event_instant!(
"delta_funnel.profile",
"Operator activity trace truncated",
|context: &mut EventContext| {
operation.set_on(context);
context.add_debug_arg("maximum_spans", TrackEventDebugArg::Uint64(maximum_spans));
}
);
}
}
fn dependency_span_label(name: &str) -> Option<&'static str> {
match name {
BULK_FINALIZE_PREPARE => Some("Prepare final bulk packet"),
BULK_FINALIZE_WRITE => Some("Write final bulk packet"),
BULK_FINALIZE_FLUSH => Some("Flush SQL Server connection"),
BULK_FINALIZE_RESULT => Some("Await SQL Server result"),
_ => None,
}
}
#[derive(Clone, Debug, Default)]
struct ProfileFields {
operation_id: Option<u64>,
capture_scope_id: Option<u64>,
query_execution_id: Option<u64>,
query_scope: Option<String>,
query_owner: Option<String>,
worker_lane_id: Option<u64>,
worker_kind: Option<String>,
node_id: Option<u64>,
parent_node_id: Option<u64>,
operator_partition: Option<u64>,
execution_stream_id: Option<u64>,
activity: Option<String>,
planning_activity_name: Option<String>,
execution_activity_name: Option<String>,
maximum_spans: Option<u64>,
operator_name: Option<String>,
phase: Option<String>,
operation_kind: Option<String>,
stage_name: Option<String>,
stage_category: Option<String>,
stage_owner_id: Option<u64>,
result: Option<String>,
time_semantics: Option<String>,
}
impl Visit for ProfileFields {
fn record_u64(&mut self, field: &Field, value: u64) {
match field.name() {
"operation_id" => self.operation_id = Some(value),
"capture_scope_id" => self.capture_scope_id = (value != 0).then_some(value),
"query_execution_id" => self.query_execution_id = Some(value),
"worker_lane_id" => self.worker_lane_id = Some(value),
"node_id" => self.node_id = Some(value),
"parent_node_id" => self.parent_node_id = Some(value),
"operator_partition" => self.operator_partition = Some(value),
"execution_stream_id" => self.execution_stream_id = Some(value),
"maximum_spans" => self.maximum_spans = Some(value),
"stage_owner_id" => self.stage_owner_id = Some(value),
_ => {}
}
}
fn record_str(&mut self, field: &Field, value: &str) {
match field.name() {
"query_scope" => self.query_scope = Some(value.to_owned()),
"query_owner" => self.query_owner = Some(value.to_owned()),
"worker_kind" => self.worker_kind = Some(value.to_owned()),
"activity" => self.activity = Some(value.to_owned()),
"planning_activity_name" => self.planning_activity_name = Some(value.to_owned()),
"execution_activity_name" => self.execution_activity_name = Some(value.to_owned()),
"operator_name" => self.operator_name = Some(value.to_owned()),
"phase" => self.phase = Some(value.to_owned()),
"operation_kind" => self.operation_kind = Some(value.to_owned()),
"stage_name" => self.stage_name = Some(value.to_owned()),
"stage_category" => self.stage_category = Some(value.to_owned()),
"result" => self.result = Some(value.to_owned()),
"time_semantics" => self.time_semantics = Some(value.to_owned()),
_ => {}
}
}
fn record_debug(&mut self, _field: &Field, _value: &dyn std::fmt::Debug) {}
}
#[derive(Debug)]
struct ActiveProfileSpan {
event: ProfileEvent,
fields: ProfileFields,
profile_context_id: Option<u64>,
profile_context_identity_dirty: bool,
}
impl ActiveProfileSpan {
fn from_fields(name: &str, fields: ProfileFields) -> Option<Self> {
let operation_id = fields.operation_id?;
let diagnostics = diagnostics_track(TrackEventTrack::process_track_uuid());
let operation = operation_track(operation_id, diagnostics.uuid);
let event = match name {
"Delta Funnel preview" => ProfileEvent::Operation {
kind: OperationKind::Preview,
diagnostics,
operation,
},
"Delta Funnel SQL Server write" => ProfileEvent::Operation {
kind: OperationKind::MssqlWrite,
diagnostics,
operation,
},
"Delta Funnel SQL Server write_all" => ProfileEvent::Operation {
kind: OperationKind::WriteAll,
diagnostics,
operation,
},
"DataFusion query planning" => {
let query_execution_id = fields.query_execution_id?;
let phases = phase_track(operation_id, operation.uuid);
ProfileEvent::Planning {
planning: planning_track(operation_id, query_execution_id, phases.uuid),
query: query_track(operation_id, query_execution_id, operation.uuid),
}
}
"DataFusion planning activity" => {
let query_execution_id = fields.query_execution_id?;
let name = fields.planning_activity_name.clone()?;
if fields.activity.as_deref()?.is_empty() {
return None;
}
let phases = phase_track(operation_id, operation.uuid);
ProfileEvent::PlanningActivity {
name,
planning: planning_track(operation_id, query_execution_id, phases.uuid),
}
}
"DataFusion execution activity" => {
let query_execution_id = fields.query_execution_id?;
let execution_stream_id = fields.execution_stream_id?;
let name = fields.execution_activity_name.clone()?;
if fields.activity.as_deref()?.is_empty() {
return None;
}
let query = query_track(operation_id, query_execution_id, operation.uuid);
ProfileEvent::ExecutionActivity {
name,
output: delta_scan_output_track(
operation_id,
query_execution_id,
execution_stream_id,
query.uuid,
),
}
}
"Delta Funnel operation phase" => ProfileEvent::Phase {
kind: OperationPhaseKind::from_str(fields.phase.as_deref()?)?,
phases: phase_track(operation_id, operation.uuid),
},
"Delta Funnel operation stage" => {
OperationKind::from_str(fields.operation_kind.as_deref()?)?;
let name = fields.stage_name.clone()?;
if fields.stage_category.as_deref()?.is_empty() {
return None;
}
let track = fields
.stage_owner_id
.filter(|owner_id| *owner_id != 0)
.map_or_else(
|| phase_track(operation_id, operation.uuid),
|owner_id| owner_track(operation_id, owner_id, operation.uuid),
);
ProfileEvent::Stage { name, track }
}
"DataFusion operator activity" => {
let query_execution_id = fields.query_execution_id?;
let worker_lane_id = fields.worker_lane_id?;
let query = query_track(operation_id, query_execution_id, operation.uuid);
ProfileEvent::Operator {
name: fields.operator_name.clone()?,
worker: worker_track(
operation_id,
query_execution_id,
worker_lane_id,
query.uuid,
worker_lane_id,
),
}
}
"DataFusion task context" => ProfileEvent::TaskContext {
name: fields.operator_name.clone()?,
},
name if name == OBJECT_STORE_TRANSPORT_CONTEXT_NAME => {
fields.query_execution_id?;
fields.execution_stream_id?;
ProfileEvent::TaskContext {
name: OBJECT_STORE_TRANSPORT_DISPLAY_NAME.to_owned(),
}
}
_ => return None,
};
let profile_context_identity_dirty = matches!(event, ProfileEvent::TaskContext { .. });
Some(Self {
event,
fields,
profile_context_id: None,
profile_context_identity_dirty,
})
}
fn inherit(&self, name: &str) -> Option<Self> {
let track = match &self.event {
ProfileEvent::TaskContext { .. } => self.task_track()?,
_ => self.track().clone(),
};
Some(Self {
event: ProfileEvent::Detail {
name: name.to_owned(),
track,
},
fields: self.fields.clone(),
profile_context_id: None,
profile_context_identity_dirty: false,
})
}
fn record(&mut self, fields: ProfileFields) {
let previous_worker_lane_id = self.fields.worker_lane_id;
let previous_parent_node_id = self.fields.parent_node_id;
if matches!(&self.event, ProfileEvent::TaskContext { .. }) {
if fields.worker_lane_id.is_some() {
self.fields.worker_lane_id = fields.worker_lane_id;
}
if fields.worker_kind.is_some() {
self.fields.worker_kind = fields.worker_kind;
}
}
if !matches!(&self.event, ProfileEvent::PlanningActivity { .. })
&& fields.query_owner.is_some()
{
self.fields.query_owner = fields.query_owner;
}
if fields.parent_node_id.is_some() {
self.fields.parent_node_id = fields.parent_node_id;
}
if fields.result.is_some() {
self.fields.result = fields.result;
}
if matches!(&self.event, ProfileEvent::TaskContext { .. })
&& (self.fields.worker_lane_id != previous_worker_lane_id
|| self.fields.parent_node_id != previous_parent_node_id)
{
self.profile_context_identity_dirty = true;
}
}
fn emit_begin(&self) {
match &self.event {
ProfileEvent::Operation {
kind,
diagnostics,
operation,
} => {
track_event_instant!(
"delta_funnel.profile",
"Delta Funnel diagnostic group",
|context: &mut EventContext| diagnostics.set_on(context)
);
match kind {
OperationKind::Preview => track_event_begin!(
"delta_funnel.profile",
"Delta Funnel preview",
|context: &mut EventContext| self.set_operation_on(context, operation)
),
OperationKind::MssqlWrite => track_event_begin!(
"delta_funnel.profile",
"Delta Funnel SQL Server write",
|context: &mut EventContext| self.set_operation_on(context, operation)
),
OperationKind::WriteAll => track_event_begin!(
"delta_funnel.profile",
"Delta Funnel SQL Server write_all",
|context: &mut EventContext| self.set_operation_on(context, operation)
),
}
}
ProfileEvent::Planning { planning, query } => {
track_event_instant!(
"delta_funnel.profile",
"DataFusion query",
|context: &mut EventContext| {
query.set_on(context);
self.add_profile_args(context);
}
);
track_event_begin!(
"delta_funnel.profile",
"DataFusion query planning",
|context: &mut EventContext| {
planning.set_on(context);
self.add_profile_args(context);
}
);
}
ProfileEvent::PlanningActivity { name, planning } => {
if let Ok(name) = CString::new(name.as_str()) {
track_event!(
"delta_funnel.profile",
TrackEventType::SliceBegin(name.as_ptr()),
|context: &mut EventContext| {
planning.set_on(context);
self.add_profile_args(context);
}
);
} else {
track_event_begin!(
"delta_funnel.profile",
"DataFusion planning activity",
|context: &mut EventContext| {
planning.set_on(context);
self.add_profile_args(context);
}
);
}
}
ProfileEvent::ExecutionActivity { name, output } => {
if let Ok(name) = CString::new(name.as_str()) {
track_event!(
"delta_funnel.profile",
TrackEventType::SliceBegin(name.as_ptr()),
|context: &mut EventContext| {
output.set_on(context);
self.add_profile_args(context);
}
);
} else {
track_event_begin!(
"delta_funnel.profile",
"DataFusion execution activity",
|context: &mut EventContext| {
output.set_on(context);
self.add_profile_args(context);
}
);
}
}
ProfileEvent::Phase { kind, phases } => match kind {
OperationPhaseKind::Planning => track_event_begin!(
"delta_funnel.profile",
"Planning",
|context: &mut EventContext| {
phases.set_on(context);
self.add_profile_args(context);
}
),
OperationPhaseKind::Execution => track_event_begin!(
"delta_funnel.profile",
"Execution",
|context: &mut EventContext| {
phases.set_on(context);
self.add_profile_args(context);
}
),
OperationPhaseKind::Finalization => track_event_begin!(
"delta_funnel.profile",
"Finalization",
|context: &mut EventContext| {
phases.set_on(context);
self.add_profile_args(context);
}
),
},
ProfileEvent::Stage { name, track } => {
if let Ok(name) = CString::new(name.as_str()) {
track_event!(
"delta_funnel.profile",
TrackEventType::SliceBegin(name.as_ptr()),
|context: &mut EventContext| {
track.set_on(context);
self.add_profile_args(context);
}
);
} else {
track_event_begin!(
"delta_funnel.profile",
"Delta Funnel operation stage",
|context: &mut EventContext| {
track.set_on(context);
self.add_profile_args(context);
}
);
}
}
ProfileEvent::Detail { name, track } => {
if let Ok(name) = CString::new(name.as_str()) {
track_event!(
"delta_funnel.profile",
TrackEventType::SliceBegin(name.as_ptr()),
|context: &mut EventContext| {
track.set_on(context);
self.add_profile_args(context);
}
);
} else {
track_event_begin!(
"delta_funnel.profile",
"Instrumented detail",
|context: &mut EventContext| {
track.set_on(context);
self.add_profile_args(context);
}
);
}
}
ProfileEvent::Operator { name, worker } => {
if let Ok(name) = CString::new(name.as_str()) {
track_event!(
"delta_funnel.profile",
TrackEventType::SliceBegin(name.as_ptr()),
|context: &mut EventContext| {
worker.set_on(context);
self.add_profile_args(context);
}
);
} else {
track_event_begin!(
"delta_funnel.profile",
"DataFusion operator",
|context: &mut EventContext| {
worker.set_on(context);
self.add_profile_args(context);
}
);
}
}
ProfileEvent::TaskContext { .. } => {}
}
}
fn emit_end(self) {
if matches!(&self.event, ProfileEvent::TaskContext { .. }) {
return;
}
let flush = matches!(&self.event, ProfileEvent::Operation { .. });
let track = self.track();
track_event_end!("delta_funnel.profile", |context: &mut EventContext| {
track.set_on(context);
self.add_completion_args(context);
if flush {
context.set_flush();
}
});
}
fn emit_context_enter(&mut self) {
if matches!(&self.event, ProfileEvent::TaskContext { .. }) {
let identity_changed = self.refresh_profile_context_identity();
if let Some(profile_context_id) = self.profile_context_id {
if identity_changed {
track_event_begin!(
"delta_funnel.profile.context",
"Delta Funnel execution context",
|context: &mut EventContext| {
context.add_debug_arg(
"profile_context_id",
TrackEventDebugArg::Uint64(profile_context_id),
);
context.add_debug_arg(
"context_name",
TrackEventDebugArg::String(self.event.name()),
);
self.add_profile_args(context);
}
);
} else {
track_event_begin!(
"delta_funnel.profile.context",
"Delta Funnel execution context",
|context: &mut EventContext| {
context.add_debug_arg(
"profile_context_id",
TrackEventDebugArg::Uint64(profile_context_id),
);
}
);
}
}
return;
}
if let Ok(name) = CString::new(self.event.name()) {
track_event!(
"delta_funnel.profile.context",
TrackEventType::SliceBegin(name.as_ptr()),
|context: &mut EventContext| self.add_profile_args(context)
);
} else {
track_event_begin!(
"delta_funnel.profile.context",
"Delta Funnel execution context",
|context: &mut EventContext| self.add_profile_args(context)
);
}
}
fn refresh_profile_context_identity(&mut self) -> bool {
if !self.profile_context_identity_dirty {
return false;
}
self.profile_context_id = allocate_id(&NEXT_PROFILE_CONTEXT_ID);
self.profile_context_identity_dirty = false;
true
}
fn context_event_enabled(&self) -> bool {
!matches!(&self.event, ProfileEvent::TaskContext { .. })
|| self.profile_context_id.is_some()
}
fn task_track(&self) -> Option<SemanticTrack> {
let operation_id = self.fields.operation_id?;
let query_execution_id = self.fields.query_execution_id?;
let worker_lane_id = self.fields.worker_lane_id?;
let diagnostics = diagnostics_track(TrackEventTrack::process_track_uuid());
let operation = operation_track(operation_id, diagnostics.uuid);
let query = query_track(operation_id, query_execution_id, operation.uuid);
Some(worker_track(
operation_id,
query_execution_id,
worker_lane_id,
query.uuid,
worker_lane_id,
))
}
fn set_operation_on(&self, context: &mut EventContext, operation: &SemanticTrack) {
operation.set_on(context);
self.add_profile_args(context);
}
fn add_profile_args(&self, context: &mut EventContext) {
if let Some(operation_id) = self.fields.operation_id {
context.add_debug_arg("operation_id", TrackEventDebugArg::Uint64(operation_id));
}
if let Some(capture_scope_id) = self.fields.capture_scope_id {
context.add_debug_arg(
"capture_scope_id",
TrackEventDebugArg::Uint64(capture_scope_id),
);
}
if let Some(query_execution_id) = self.fields.query_execution_id {
context.add_debug_arg(
"query_execution_id",
TrackEventDebugArg::Uint64(query_execution_id),
);
}
if let Some(query_scope) = &self.fields.query_scope {
context.add_debug_arg("query_scope", TrackEventDebugArg::String(query_scope));
}
if let Some(query_owner) = &self.fields.query_owner {
context.add_debug_arg("query_owner", TrackEventDebugArg::String(query_owner));
}
if let Some(worker_lane_id) = self.fields.worker_lane_id {
context.add_debug_arg("worker_lane_id", TrackEventDebugArg::Uint64(worker_lane_id));
}
if let Some(worker_kind) = &self.fields.worker_kind {
context.add_debug_arg("worker_kind", TrackEventDebugArg::String(worker_kind));
}
if let Some(node_id) = self.fields.node_id {
context.add_debug_arg("node_id", TrackEventDebugArg::Uint64(node_id));
}
if let Some(parent_node_id) = self.fields.parent_node_id {
context.add_debug_arg("parent_node_id", TrackEventDebugArg::Uint64(parent_node_id));
}
if let Some(operator_partition) = self.fields.operator_partition {
context.add_debug_arg(
"operator_partition",
TrackEventDebugArg::Uint64(operator_partition),
);
}
if let Some(execution_stream_id) = self.fields.execution_stream_id {
context.add_debug_arg(
"execution_stream_id",
TrackEventDebugArg::Uint64(execution_stream_id),
);
}
if let Some(activity) = &self.fields.activity {
context.add_debug_arg("activity", TrackEventDebugArg::String(activity));
}
if let Some(name) = &self.fields.planning_activity_name {
context.add_debug_arg("planning_activity_name", TrackEventDebugArg::String(name));
}
if let Some(name) = &self.fields.execution_activity_name {
context.add_debug_arg("execution_activity_name", TrackEventDebugArg::String(name));
}
if let Some(operation_kind) = &self.fields.operation_kind {
context.add_debug_arg("operation_kind", TrackEventDebugArg::String(operation_kind));
}
if let Some(stage_name) = &self.fields.stage_name {
context.add_debug_arg("stage_name", TrackEventDebugArg::String(stage_name));
}
if let Some(stage_category) = &self.fields.stage_category {
context.add_debug_arg("stage_category", TrackEventDebugArg::String(stage_category));
}
if let Some(stage_owner_id) = self.fields.stage_owner_id {
context.add_debug_arg("stage_owner_id", TrackEventDebugArg::Uint64(stage_owner_id));
}
if let Some(time_semantics) = &self.fields.time_semantics {
context.add_debug_arg("time_semantics", TrackEventDebugArg::String(time_semantics));
}
}
fn add_completion_args(&self, context: &mut EventContext) {
if let Some(query_owner) = &self.fields.query_owner {
context.add_debug_arg("query_owner", TrackEventDebugArg::String(query_owner));
}
if let Some(parent_node_id) = self.fields.parent_node_id {
context.add_debug_arg("parent_node_id", TrackEventDebugArg::Uint64(parent_node_id));
}
if let Some(result) = &self.fields.result {
context.add_debug_arg("result", TrackEventDebugArg::String(result));
}
}
fn track(&self) -> &SemanticTrack {
match &self.event {
ProfileEvent::Operation { operation, .. } => operation,
ProfileEvent::Planning { planning, .. }
| ProfileEvent::PlanningActivity { planning, .. } => planning,
ProfileEvent::ExecutionActivity { output, .. } => output,
ProfileEvent::Phase { phases, .. } => phases,
ProfileEvent::Stage { track, .. } | ProfileEvent::Detail { track, .. } => track,
ProfileEvent::Operator { worker, .. } => worker,
ProfileEvent::TaskContext { .. } => {
unreachable!("a task context does not own a semantic track")
}
}
}
}
#[derive(Debug)]
enum ProfileEvent {
Operation {
kind: OperationKind,
diagnostics: SemanticTrack,
operation: SemanticTrack,
},
Planning {
planning: SemanticTrack,
query: SemanticTrack,
},
PlanningActivity {
name: String,
planning: SemanticTrack,
},
ExecutionActivity {
name: String,
output: SemanticTrack,
},
Phase {
kind: OperationPhaseKind,
phases: SemanticTrack,
},
Stage {
name: String,
track: SemanticTrack,
},
Detail {
name: String,
track: SemanticTrack,
},
Operator {
name: String,
worker: SemanticTrack,
},
TaskContext {
name: String,
},
}
impl ProfileEvent {
fn name(&self) -> &str {
match self {
Self::Operation { kind, .. } => kind.name(),
Self::Planning { .. } => "DataFusion query planning",
Self::PlanningActivity { name, .. }
| Self::ExecutionActivity { name, .. }
| Self::Stage { name, .. }
| Self::Detail { name, .. }
| Self::Operator { name, .. }
| Self::TaskContext { name } => name,
Self::Phase { kind, .. } => kind.name(),
}
}
}
#[derive(Debug)]
enum OperationKind {
Preview,
MssqlWrite,
WriteAll,
}
impl OperationKind {
const fn name(&self) -> &'static str {
match self {
Self::Preview => "Delta Funnel preview",
Self::MssqlWrite => "Delta Funnel SQL Server write",
Self::WriteAll => "Delta Funnel SQL Server write_all",
}
}
fn from_str(value: &str) -> Option<Self> {
match value {
"preview" => Some(Self::Preview),
"mssql_write" => Some(Self::MssqlWrite),
"write_all" => Some(Self::WriteAll),
_ => None,
}
}
}
#[derive(Debug)]
enum OperationPhaseKind {
Planning,
Execution,
Finalization,
}
impl OperationPhaseKind {
const fn name(&self) -> &'static str {
match self {
Self::Planning => "Planning",
Self::Execution => "Execution",
Self::Finalization => "Finalization",
}
}
fn from_str(value: &str) -> Option<Self> {
match value {
"planning" => Some(Self::Planning),
"execution" => Some(Self::Execution),
"finalization" => Some(Self::Finalization),
_ => None,
}
}
}
#[cfg(test)]
mod tests {
use std::sync::{
Arc,
atomic::{AtomicUsize, Ordering},
};
use tracing_subscriber::{filter::filter_fn, prelude::*};
use super::*;
fn fields(operation_id: u64, query_execution_id: u64, worker_lane_id: u64) -> ProfileFields {
ProfileFields {
operation_id: Some(operation_id),
query_execution_id: Some(query_execution_id),
query_scope: Some("preview".to_owned()),
query_owner: Some("orders".to_owned()),
worker_lane_id: Some(worker_lane_id),
worker_kind: Some("runtime".to_owned()),
node_id: Some(7),
parent_node_id: Some(3),
operator_partition: Some(2),
execution_stream_id: Some(11),
activity: Some("poll_next".to_owned()),
operator_name: Some("FilterExec".to_owned()),
time_semantics: Some("active".to_owned()),
..ProfileFields::default()
}
}
#[test]
fn canonical_fields_map_to_deterministic_exact_worker_tracks() {
let first = ActiveProfileSpan::from_fields("DataFusion operator activity", fields(1, 1, 1))
.expect("complete operator identity should map");
let duplicate =
ActiveProfileSpan::from_fields("DataFusion operator activity", fields(1, 1, 1))
.expect("the same operator identity should map");
let worker_10 =
ActiveProfileSpan::from_fields("DataFusion operator activity", fields(1, 1, 10))
.expect("a second worker should map");
assert_eq!(first.track(), duplicate.track());
assert_ne!(first.track().uuid, worker_10.track().uuid);
assert!(
first
.track()
.name
.contains("worker [w-00000000000000000001]")
);
assert!(
!worker_10
.track()
.name
.contains("worker [w-00000000000000000001]")
);
assert_eq!(first.fields.query_scope.as_deref(), Some("preview"));
assert_eq!(first.fields.query_owner.as_deref(), Some("orders"));
assert_eq!(first.fields.worker_kind.as_deref(), Some("runtime"));
assert_eq!(first.fields.node_id, Some(7));
assert_eq!(first.fields.parent_node_id, Some(3));
assert_eq!(first.fields.operator_partition, Some(2));
assert_eq!(first.fields.execution_stream_id, Some(11));
assert_eq!(first.fields.activity.as_deref(), Some("poll_next"));
assert_eq!(first.fields.time_semantics.as_deref(), Some("active"));
}
#[test]
fn task_context_accepts_worker_identity_recorded_before_each_poll() {
let mut initial = fields(1, 1, 1);
initial.worker_lane_id = None;
initial.worker_kind = None;
initial.activity = Some("spawned_task".to_owned());
let mut active = ActiveProfileSpan::from_fields("DataFusion task context", initial)
.expect("task creation should not require a worker assignment");
assert!(matches!(active.event, ProfileEvent::TaskContext { .. }));
active.record(ProfileFields {
worker_lane_id: Some(4),
worker_kind: Some("runtime".to_owned()),
..ProfileFields::default()
});
assert!(active.refresh_profile_context_identity());
let worker_4_context_id = active
.profile_context_id
.expect("the first complete task identity should allocate an id");
active.record(ProfileFields {
worker_lane_id: Some(4),
worker_kind: Some("runtime".to_owned()),
..ProfileFields::default()
});
assert!(!active.refresh_profile_context_identity());
assert_eq!(active.profile_context_id, Some(worker_4_context_id));
active.record(ProfileFields {
worker_lane_id: Some(5),
worker_kind: Some("runtime".to_owned()),
..ProfileFields::default()
});
assert!(active.refresh_profile_context_identity());
assert_ne!(active.profile_context_id, Some(worker_4_context_id));
let detail = active
.inherit("DataFusion dependency")
.expect("an entered task context should propagate its worker assignment");
assert_eq!(active.fields.worker_lane_id, Some(5));
assert_eq!(active.fields.worker_kind.as_deref(), Some("runtime"));
assert!(
detail
.track()
.name
.contains("worker [w-00000000000000000005]")
);
}
#[test]
fn object_store_transport_context_uses_stream_identity_without_a_worker() {
let mut transport = fields(1, 2, 3);
transport.worker_lane_id = None;
transport.worker_kind = None;
let active = ActiveProfileSpan::from_fields(OBJECT_STORE_TRANSPORT_CONTEXT_NAME, transport)
.expect("complete transport identity should map");
assert!(matches!(
active.event,
ProfileEvent::TaskContext { ref name } if name == OBJECT_STORE_TRANSPORT_DISPLAY_NAME
));
assert_eq!(active.fields.query_execution_id, Some(2));
assert_eq!(active.fields.execution_stream_id, Some(11));
assert_eq!(active.fields.worker_lane_id, None);
let mut incomplete = fields(1, 2, 3);
incomplete.execution_stream_id = None;
assert!(
ActiveProfileSpan::from_fields(OBJECT_STORE_TRANSPORT_CONTEXT_NAME, incomplete)
.is_none()
);
}
#[test]
fn completion_records_preserve_begin_identity_and_update_completion_fields() {
let mut initial = fields(1, 1, 1);
initial.capture_scope_id = Some(42);
initial.query_owner = None;
initial.parent_node_id = None;
let mut active = ActiveProfileSpan::from_fields("DataFusion operator activity", initial)
.expect("complete operator identity should map");
active.record(ProfileFields {
operation_id: Some(99),
capture_scope_id: Some(99),
query_execution_id: Some(99),
query_owner: Some("orders".to_owned()),
worker_lane_id: Some(99),
parent_node_id: Some(3),
result: Some("batch".to_owned()),
..ProfileFields::default()
});
assert_eq!(active.fields.operation_id, Some(1));
assert_eq!(active.fields.capture_scope_id, Some(42));
assert_eq!(active.fields.query_execution_id, Some(1));
assert_eq!(active.fields.worker_lane_id, Some(1));
assert_eq!(active.fields.query_owner.as_deref(), Some("orders"));
assert_eq!(active.fields.parent_node_id, Some(3));
assert_eq!(active.fields.result.as_deref(), Some("batch"));
}
#[test]
fn incomplete_or_unknown_spans_are_ignored() {
assert!(
ActiveProfileSpan::from_fields(
"DataFusion operator activity",
ProfileFields::default()
)
.is_none()
);
assert!(ActiveProfileSpan::from_fields("application span", fields(1, 1, 1)).is_none());
}
#[test]
fn bulk_finalize_dependency_spans_use_friendly_labels_on_the_parent_track() {
let parent = ActiveProfileSpan::from_fields(
"Delta Funnel operation stage",
ProfileFields {
operation_id: Some(7),
operation_kind: Some("mssql_write".to_owned()),
stage_name: Some("Finalize SQL Server writer".to_owned()),
stage_category: Some("delta_funnel.write.sql_server".to_owned()),
stage_owner_id: Some(3),
..ProfileFields::default()
},
)
.expect("the semantic parent should map");
let child = parent
.inherit(
dependency_span_label(BULK_FINALIZE_RESULT)
.expect("the stable dependency span should map"),
)
.expect("the complete parent identity should propagate");
assert_eq!(child.track(), parent.track());
assert_eq!(child.fields.operation_id, Some(7));
assert_eq!(child.fields.stage_owner_id, Some(3));
assert!(matches!(
child.event,
ProfileEvent::Detail { ref name, .. } if name == "Await SQL Server result"
));
assert!(is_profile_target(PROFILE_TARGET));
assert!(is_profile_target(TIBERIUS_PROFILE_TARGET));
assert!(!is_profile_target("application"));
for (name, label) in [
(BULK_FINALIZE_PREPARE, "Prepare final bulk packet"),
(BULK_FINALIZE_WRITE, "Write final bulk packet"),
(BULK_FINALIZE_FLUSH, "Flush SQL Server connection"),
(BULK_FINALIZE_RESULT, "Await SQL Server result"),
] {
assert_eq!(dependency_span_label(name), Some(label));
}
assert_eq!(dependency_span_label("protocol.bulk_load.request"), None);
}
#[test]
fn operation_phases_share_one_deterministic_track() {
let phase = |value: &str| ProfileFields {
operation_id: Some(7),
phase: Some(value.to_owned()),
..ProfileFields::default()
};
let planning =
ActiveProfileSpan::from_fields("Delta Funnel operation phase", phase("planning"))
.expect("a known phase should map");
let execution =
ActiveProfileSpan::from_fields("Delta Funnel operation phase", phase("execution"))
.expect("a known phase should map");
let finalization =
ActiveProfileSpan::from_fields("Delta Funnel operation phase", phase("finalization"))
.expect("a known phase should map");
assert_eq!(planning.track(), execution.track());
assert_eq!(execution.track(), finalization.track());
assert!(
ActiveProfileSpan::from_fields("Delta Funnel operation phase", phase("unknown"))
.is_none()
);
assert!(
ActiveProfileSpan::from_fields(
"Delta Funnel operation phase",
ProfileFields {
operation_id: Some(7),
..ProfileFields::default()
}
)
.is_none()
);
}
#[test]
fn operation_stages_use_bounded_fields_and_deterministic_owner_tracks() {
let stage = |owner_id| ProfileFields {
operation_id: Some(7),
operation_kind: Some("write_all".to_owned()),
stage_name: Some("Execute output workflow".to_owned()),
stage_category: Some("delta_funnel.write_all.execution".to_owned()),
stage_owner_id: owner_id,
time_semantics: Some("wall_clock".to_owned()),
..ProfileFields::default()
};
let operation_stage =
ActiveProfileSpan::from_fields("Delta Funnel operation stage", stage(None))
.expect("an operation stage should map");
let operation_phase = ActiveProfileSpan::from_fields(
"Delta Funnel operation phase",
ProfileFields {
operation_id: Some(7),
phase: Some("execution".to_owned()),
..ProfileFields::default()
},
)
.expect("an operation phase should map");
let owner = ActiveProfileSpan::from_fields("Delta Funnel operation stage", stage(Some(4)))
.expect("an owner stage should map");
let duplicate =
ActiveProfileSpan::from_fields("Delta Funnel operation stage", stage(Some(4)))
.expect("the same owner should map");
let other_owner =
ActiveProfileSpan::from_fields("Delta Funnel operation stage", stage(Some(5)))
.expect("another owner should map");
assert_eq!(operation_stage.track(), operation_phase.track());
assert_eq!(owner.track(), duplicate.track());
assert_ne!(owner.track().uuid, other_owner.track().uuid);
assert!(
owner
.track()
.name
.contains("owner [o-00000000000000000004]")
);
assert_eq!(
owner.track().parent_uuid,
operation_phase.track().parent_uuid
);
assert!(
ActiveProfileSpan::from_fields(
"Delta Funnel operation stage",
ProfileFields {
operation_id: Some(7),
operation_kind: Some("unknown".to_owned()),
stage_name: Some("Unknown".to_owned()),
stage_category: Some("delta_funnel.test".to_owned()),
..ProfileFields::default()
},
)
.is_none()
);
assert!(
ActiveProfileSpan::from_fields(
"Delta Funnel operation stage",
ProfileFields {
operation_id: Some(7),
operation_kind: Some("preview".to_owned()),
stage_name: Some("Missing category".to_owned()),
stage_category: Some(String::new()),
..ProfileFields::default()
},
)
.is_none()
);
}
#[test]
fn operation_stage_begin_fields_are_immutable_and_result_is_terminal() {
let mut active = ActiveProfileSpan::from_fields(
"Delta Funnel operation stage",
ProfileFields {
operation_id: Some(7),
operation_kind: Some("preview".to_owned()),
stage_name: Some("Physical planning".to_owned()),
stage_category: Some("delta_funnel.preview.phase".to_owned()),
stage_owner_id: Some(2),
time_semantics: Some("wall_clock".to_owned()),
..ProfileFields::default()
},
)
.expect("the stage should map");
active.record(ProfileFields {
operation_id: Some(99),
operation_kind: Some("write_all".to_owned()),
stage_name: Some("Changed".to_owned()),
stage_category: Some("changed".to_owned()),
stage_owner_id: Some(99),
result: Some("ok".to_owned()),
..ProfileFields::default()
});
assert_eq!(active.fields.operation_id, Some(7));
assert_eq!(active.fields.operation_kind.as_deref(), Some("preview"));
assert_eq!(
active.fields.stage_name.as_deref(),
Some("Physical planning")
);
assert_eq!(
active.fields.stage_category.as_deref(),
Some("delta_funnel.preview.phase")
);
assert_eq!(active.fields.stage_owner_id, Some(2));
assert_eq!(active.fields.result.as_deref(), Some("ok"));
}
#[test]
fn planning_activities_share_their_query_planning_track() {
let planning_fields = |query_execution_id, name: &str| ProfileFields {
operation_id: Some(7),
query_execution_id: Some(query_execution_id),
query_scope: Some("mssql_output".to_owned()),
query_owner: Some("orders".to_owned()),
planning_activity_name: Some(name.to_owned()),
activity: Some("delta_scan_planning".to_owned()),
time_semantics: Some("wall_clock".to_owned()),
..ProfileFields::default()
};
let query_planning = ActiveProfileSpan::from_fields(
"DataFusion query planning",
planning_fields(3, "unused"),
)
.expect("query planning should map");
let activity = ActiveProfileSpan::from_fields(
"DataFusion planning activity",
planning_fields(3, "Delta scan planning"),
)
.expect("the planning activity should map");
let duplicate = ActiveProfileSpan::from_fields(
"DataFusion planning activity",
planning_fields(3, "Delta scan planning"),
)
.expect("the same planning activity should map");
let other_query = ActiveProfileSpan::from_fields(
"DataFusion planning activity",
planning_fields(4, "Delta scan planning"),
)
.expect("another query should map");
assert_eq!(query_planning.track(), activity.track());
assert_eq!(activity.track(), duplicate.track());
assert_ne!(activity.track().uuid, other_query.track().uuid);
assert_eq!(
activity.track().parent_uuid,
phase_track(7, operation_track(7, diagnostics_track(0).uuid).uuid).uuid
);
assert!(
activity
.track()
.name
.contains("query [q-00000000000000000003] / planning")
);
assert_eq!(
activity.fields.planning_activity_name.as_deref(),
Some("Delta scan planning")
);
assert_eq!(
activity.fields.activity.as_deref(),
Some("delta_scan_planning")
);
assert_eq!(activity.fields.query_scope.as_deref(), Some("mssql_output"));
assert_eq!(activity.fields.query_owner.as_deref(), Some("orders"));
assert_eq!(
activity.fields.time_semantics.as_deref(),
Some("wall_clock")
);
}
#[test]
fn planning_activity_identity_is_begin_only_and_result_is_terminal() {
let mut active = ActiveProfileSpan::from_fields(
"DataFusion planning activity",
ProfileFields {
operation_id: Some(7),
query_execution_id: Some(3),
query_scope: Some("preview".to_owned()),
planning_activity_name: Some("Delta scan planning".to_owned()),
activity: Some("delta_scan_planning".to_owned()),
time_semantics: Some("wall_clock".to_owned()),
..ProfileFields::default()
},
)
.expect("the planning activity should map");
active.record(ProfileFields {
operation_id: Some(99),
query_execution_id: Some(99),
query_scope: Some("mssql_output".to_owned()),
query_owner: Some("changed".to_owned()),
planning_activity_name: Some("Changed".to_owned()),
activity: Some("changed".to_owned()),
time_semantics: Some("active".to_owned()),
result: Some("ok".to_owned()),
..ProfileFields::default()
});
assert_eq!(active.fields.operation_id, Some(7));
assert_eq!(active.fields.query_execution_id, Some(3));
assert_eq!(active.fields.query_scope.as_deref(), Some("preview"));
assert_eq!(active.fields.query_owner, None);
assert_eq!(
active.fields.planning_activity_name.as_deref(),
Some("Delta scan planning")
);
assert_eq!(
active.fields.activity.as_deref(),
Some("delta_scan_planning")
);
assert_eq!(active.fields.time_semantics.as_deref(), Some("wall_clock"));
assert_eq!(active.fields.result.as_deref(), Some("ok"));
}
#[test]
fn incomplete_planning_activities_are_ignored() {
let complete = || ProfileFields {
operation_id: Some(7),
query_execution_id: Some(3),
planning_activity_name: Some("Delta scan planning".to_owned()),
activity: Some("delta_scan_planning".to_owned()),
..ProfileFields::default()
};
assert!(
ActiveProfileSpan::from_fields(
"DataFusion planning activity",
ProfileFields {
query_execution_id: None,
..complete()
}
)
.is_none()
);
assert!(
ActiveProfileSpan::from_fields(
"DataFusion planning activity",
ProfileFields {
planning_activity_name: None,
..complete()
}
)
.is_none()
);
assert!(
ActiveProfileSpan::from_fields(
"DataFusion planning activity",
ProfileFields {
activity: Some(String::new()),
..complete()
}
)
.is_none()
);
}
#[test]
fn execution_activities_use_query_specific_output_tracks() {
let execution_fields = |query_execution_id, execution_stream_id| ProfileFields {
operation_id: Some(7),
query_execution_id: Some(query_execution_id),
query_scope: Some("preview".to_owned()),
execution_stream_id: Some(execution_stream_id),
operator_partition: Some(2),
execution_activity_name: Some("Await Delta scan output".to_owned()),
activity: Some("await_output".to_owned()),
time_semantics: Some("wall_clock".to_owned()),
..ProfileFields::default()
};
let activity =
ActiveProfileSpan::from_fields("DataFusion execution activity", execution_fields(3, 1))
.expect("the execution activity should map");
let duplicate =
ActiveProfileSpan::from_fields("DataFusion execution activity", execution_fields(3, 1))
.expect("the same execution activity should map");
let other_stream = ActiveProfileSpan::from_fields(
"DataFusion execution activity",
execution_fields(3, 10),
)
.expect("another output stream should map");
let other_query =
ActiveProfileSpan::from_fields("DataFusion execution activity", execution_fields(4, 1))
.expect("another query should map");
assert_eq!(activity.track(), duplicate.track());
assert_ne!(activity.track().uuid, other_stream.track().uuid);
assert_ne!(activity.track().uuid, other_query.track().uuid);
let query = query_track(7, 3, operation_track(7, diagnostics_track(0).uuid).uuid);
let same_id_worker = worker_track(7, 3, 10, query.uuid, 10);
assert!(same_id_worker.sibling_order_rank < other_stream.track().sibling_order_rank);
assert_eq!(activity.track().parent_uuid, query.uuid);
assert!(
activity
.track()
.name
.contains("Delta scan output [s-00000000000000000001]")
);
assert!(
!other_stream
.track()
.name
.contains("Delta scan output [s-00000000000000000001]")
);
assert_eq!(
activity.fields.execution_activity_name.as_deref(),
Some("Await Delta scan output")
);
assert_eq!(activity.fields.activity.as_deref(), Some("await_output"));
assert_eq!(activity.fields.execution_stream_id, Some(1));
assert_eq!(activity.fields.operator_partition, Some(2));
assert_eq!(
activity.fields.time_semantics.as_deref(),
Some("wall_clock")
);
}
#[test]
fn execution_activity_requires_bounded_identity_and_only_updates_terminal_fields() {
let complete = || ProfileFields {
operation_id: Some(7),
query_execution_id: Some(3),
execution_stream_id: Some(1),
execution_activity_name: Some("Await Delta scan output".to_owned()),
activity: Some("await_output".to_owned()),
time_semantics: Some("wall_clock".to_owned()),
..ProfileFields::default()
};
for incomplete in [
ProfileFields {
query_execution_id: None,
..complete()
},
ProfileFields {
execution_stream_id: None,
..complete()
},
ProfileFields {
execution_activity_name: None,
..complete()
},
ProfileFields {
activity: Some(String::new()),
..complete()
},
] {
assert!(
ActiveProfileSpan::from_fields("DataFusion execution activity", incomplete)
.is_none()
);
}
let mut active =
ActiveProfileSpan::from_fields("DataFusion execution activity", complete())
.expect("the complete execution activity should map");
active.record(ProfileFields {
operation_id: Some(99),
query_execution_id: Some(99),
execution_stream_id: Some(99),
execution_activity_name: Some("Changed".to_owned()),
activity: Some("changed".to_owned()),
time_semantics: Some("active".to_owned()),
result: Some("ok".to_owned()),
..ProfileFields::default()
});
assert_eq!(active.fields.operation_id, Some(7));
assert_eq!(active.fields.query_execution_id, Some(3));
assert_eq!(active.fields.execution_stream_id, Some(1));
assert_eq!(
active.fields.execution_activity_name.as_deref(),
Some("Await Delta scan output")
);
assert_eq!(active.fields.activity.as_deref(), Some("await_output"));
assert_eq!(active.fields.time_semantics.as_deref(), Some("wall_clock"));
assert_eq!(active.fields.result.as_deref(), Some("ok"));
}
#[derive(Clone)]
struct EventCounter(Arc<AtomicUsize>);
impl<S: Subscriber> Layer<S> for EventCounter {
fn on_event(&self, _event: &Event<'_>, _context: Context<'_, S>) {
self.0.fetch_add(1, Ordering::Relaxed);
}
}
#[test]
fn profile_filter_does_not_hide_events_from_other_layers() {
let count = Arc::new(AtomicUsize::new(0));
let subscriber = tracing_subscriber::registry()
.with(EventCounter(Arc::clone(&count)))
.with(
PerfettoProfileLayer
.with_filter(filter_fn(|metadata| is_profile_target(metadata.target()))),
);
tracing::subscriber::with_default(subscriber, || {
tracing::info!(target: "application", "visible to the application layer");
});
assert_eq!(count.load(Ordering::Relaxed), 1);
}
}