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::allocate_id;
pub const PROFILE_TARGET: &str = "delta_funnel::profile";
const TIBERIUS_PROFILE_TARGET: &str = "tiberius_raw_bulk::protocol";
const DELTA_ARROW_READER_PROFILE_TARGET: &str = "delta_arrow_reader::profile";
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 | DELTA_ARROW_READER_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 {
let Some(name) = dependency_span_label(metadata.target(), 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();
}
};
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_follows_from(&self, id: &Id, follows: &Id, context: Context<'_, S>) {
let Some(span) = context.span(id) else {
return;
};
let metadata = span.metadata();
let Some(name) = dependency_span_label(metadata.target(), metadata.name()) else {
return;
};
let Some(active) = context.span(follows).and_then(|parent| {
parent
.extensions()
.get::<ActiveProfileSpan>()?
.inherit(name)
}) else {
return;
};
active.emit_begin();
span.extensions_mut().insert(active);
}
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(target: &str, name: &'static str) -> Option<&'static str> {
match (target, name) {
(TIBERIUS_PROFILE_TARGET, BULK_FINALIZE_PREPARE) => Some("Prepare final bulk packet"),
(TIBERIUS_PROFILE_TARGET, BULK_FINALIZE_WRITE) => Some("Write final bulk packet"),
(TIBERIUS_PROFILE_TARGET, BULK_FINALIZE_FLUSH) => Some("Flush SQL Server connection"),
(TIBERIUS_PROFILE_TARGET, BULK_FINALIZE_RESULT) => Some("Await SQL Server result"),
(
DELTA_ARROW_READER_PROFILE_TARGET,
name @ ("Delta scan planning"
| "Delta projection planning"
| "Delta filter planning"
| "Delta Kernel scan construction"
| "Delta scan metadata expansion"
| "Delta file task partitioning"
| "Delta partition target selection"
| "Delta scan execution setup"
| "Delta partition task"
| "Object store transport"),
) => Some(name),
_ => 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()?,
},
_ => 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 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 dependency_spans_use_stable_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(TIBERIUS_PROFILE_TARGET, 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(DELTA_ARROW_READER_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(TIBERIUS_PROFILE_TARGET, name),
Some(label)
);
}
assert_eq!(
dependency_span_label(TIBERIUS_PROFILE_TARGET, "protocol.bulk_load.request"),
None
);
for name in [
"Delta scan planning",
"Delta projection planning",
"Delta filter planning",
"Delta Kernel scan construction",
"Delta scan metadata expansion",
"Delta file task partitioning",
"Delta partition target selection",
"Delta scan execution setup",
"Delta partition task",
"Object store transport",
] {
assert_eq!(
dependency_span_label(DELTA_ARROW_READER_PROFILE_TARGET, name),
Some(name)
);
}
assert_eq!(
dependency_span_label(DELTA_ARROW_READER_PROFILE_TARGET, "unrelated"),
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);
}
}