/// This function translates data in DataFusion `MetricSets` into IOx
/// [`Span`]s. It records a snapshot of the current state of the
/// DataFusion metrics, so it should only be invoked *after* a plan is
/// fully `collect`ed.
///
/// Each `ExecutionPlan` in the plan gets its own new [`Span`] that covers
/// the time spent executing its partitions and its children
///
/// Each `ExecutionPlan` also has a new [`Span`] for each of its
/// partitions that collected metrics
///
/// The start and end time of the span are taken from the
/// ExecutionPlan's metrics, falling back to the parent span's
/// timestamps if there are no metrics
///
/// Span metadata is used to record:
/// 1. If the ExecutionPlan had no metrics
/// 2. The total number of rows produced by the ExecutionPlan (if available)
/// 3. The elapsed compute time taken by the ExecutionPlan
This function translates data in DataFusion `MetricSets` into IOx
[`Span`]s. It records a snapshot of the current state of the
DataFusion metrics, so it should only be invoked *after* a plan is
fully `collect`ed.

Each `ExecutionPlan` in the plan gets its own new [`Span`] that covers
the time spent executing its partitions and its children

Each `ExecutionPlan` also has a new [`Span`] for each of its
partitions that collected metrics

The start and end time of the span are taken from the
ExecutionPlan's metrics, falling back to the parent span's
timestamps if there are no metrics

Span metadata is used to record:
1. If the ExecutionPlan had no metrics
2. The total number of rows produced by the ExecutionPlan (if available)
3. The elapsed compute time taken by the ExecutionPlan
This function translates data in DataFusion ` MetricSets ` into IOx [ ` Span ` ] s . It records a snapshot of the current state of the DataFusion metrics so it should only be invoked * after * a plan is fully ` collect ` ed . Each ` ExecutionPlan ` in the plan gets its own new [ ` Span ` ] that covers the time spent executing its partitions and its children Each ` ExecutionPlan ` also has a new [ ` Span ` ] for each of its partitions that collected metrics The start and end time of the span are taken from the ExecutionPlan ' s metrics falling back to the parent span ' s timestamps if there are no metrics Span metadata is used to record : 1 . If the ExecutionPlan had no metrics 2 . The total number of rows produced by the ExecutionPlan ( if available ) 3 . The elapsed compute time taken by the ExecutionPlan
fn send_metrics_to_tracing(
    default_end_time: DateTime<Utc>,
    parent_span: &Span,
    physical_plan: &dyn ExecutionPlan,
    per_partition_tracing: bool,
) {
    // Something like this when one_line is contributed back upstream
    //let plan_name = physical_plan.displayable().one_line().to_string();
    let desc = one_line(physical_plan).to_string();
    let operator_name: String = desc.chars().take_while(|x| *x != ':').collect();
    // Get the timings of the parent operator
    let parent_start_time = parent_span.start.unwrap_or(default_end_time);
    let parent_end_time = parent_span.end.unwrap_or(default_end_time);
    // A span for the operation, this is the aggregate of all the partition spans
    let mut operator_span = parent_span.child(operator_name.clone());
    operator_span.metadata.insert("desc".into(), desc.into());
    let mut operator_metrics = SpanMetrics {
        output_rows: None,
        elapsed_compute_nanos: None,
    };
    // The total duration for this span and all its children and partitions
    let mut operator_start_time = chrono::MAX_DATETIME;
    let mut operator_end_time = chrono::MIN_DATETIME;
    match physical_plan.metrics() {
        None => {
            // this DataFusion node had no metrics, so record that in
            // metadata and use the start/stop time of the parent span
            operator_span
                .metadata
                .insert("missing_statistics".into(), "true".into());
        }
        Some(metrics) => {
            // Create a separate span for each partition in the operator
            for (partition, metrics) in partition_metrics(metrics) {
                let (start_ts, end_ts) = get_timestamps(&metrics);
                let partition_start_time = start_ts.unwrap_or(parent_start_time);
                let partition_end_time = end_ts.unwrap_or(parent_end_time);
                let partition_metrics = SpanMetrics {
                    output_rows: metrics.output_rows(),
                    elapsed_compute_nanos: metrics.elapsed_compute(),
                };
                operator_start_time = operator_start_time.min(partition_start_time);
                operator_end_time = operator_end_time.max(partition_end_time);
                // Update the aggregate totals in the operator span
                operator_metrics.aggregate_child(&partition_metrics);
                // Generate a span for the partition if
                // - these metrics correspond to a partition
                // - per partition tracing is enabled
                if per_partition_tracing {
                    if let Some(partition) = partition {
                        let mut partition_span =
                            operator_span.child(format!("{} ({})", operator_name, partition));
                        partition_span.start = Some(partition_start_time);
                        partition_span.end = Some(partition_end_time);
                        partition_metrics.add_to_span(&mut partition_span);
                        partition_span.export();
                    }
                }
            }
        }
    }
    // If we've not encountered any metrics to determine the operator's start
    // and end time, use those of the parent
    if operator_start_time == chrono::MAX_DATETIME {
        operator_start_time = parent_span.start.unwrap_or(default_end_time);
    }
    if operator_end_time == chrono::MIN_DATETIME {
        operator_end_time = parent_span.end.unwrap_or(default_end_time);
    }
    operator_span.start = Some(operator_start_time);
    operator_span.end = Some(operator_end_time);
    // recurse
    for child in physical_plan.children() {
        send_metrics_to_tracing(
            operator_end_time,
            &operator_span,
            child.as_ref(),
            per_partition_tracing,
        );
    }
    operator_metrics.add_to_span(&mut operator_span);
    operator_span.export();
}
fn send_metrics_to_tracing ( default_end_time : DateTime < Utc > , parent_span : & Span , physical_plan : & dyn ExecutionPlan , per_partition_tracing : bool , ) { // Something like this when one_line is contributed back upstream //let plan_name = physical_plan.displayable().one_line().to_string(); let desc = one_line ( physical_plan ) . to_string ( ) ; let operator_name : String = desc . chars ( ) . take_while ( | x | * x != ':' ) . collect ( ) ; // Get the timings of the parent operator let parent_start_time = parent_span . start . unwrap_or ( default_end_time ) ; let parent_end_time = parent_span . end . unwrap_or ( default_end_time ) ; // A span for the operation, this is the aggregate of all the partition spans let mut operator_span = parent_span . child ( operator_name . clone ( ) ) ; operator_span . metadata . insert ( " " . into ( ) , desc . into ( ) ) ; let mut operator_metrics = SpanMetrics { output_rows : None , elapsed_compute_nanos : None , } ; // The total duration for this span and all its children and partitions let mut operator_start_time = chrono :: MAX_DATETIME ; let mut operator_end_time = chrono :: MIN_DATETIME ; match physical_plan . metrics ( ) { None => { // this DataFusion node had no metrics, so record that in // metadata and use the start/stop time of the parent span operator_span . metadata . insert ( " " . into ( ) , " " . into ( ) ) ; } Some ( metrics ) => { // Create a separate span for each partition in the operator for ( partition , metrics ) in partition_metrics ( metrics ) { let ( start_ts , end_ts ) = get_timestamps ( & metrics ) ; let partition_start_time = start_ts . unwrap_or ( parent_start_time ) ; let partition_end_time = end_ts . unwrap_or ( parent_end_time ) ; let partition_metrics = SpanMetrics { output_rows : metrics . output_rows ( ) , elapsed_compute_nanos : metrics . elapsed_compute ( ) , } ; operator_start_time = operator_start_time . min ( partition_start_time ) ; operator_end_time = operator_end_time . max ( partition_end_time ) ; // Update the aggregate totals in the operator span operator_metrics . aggregate_child ( & partition_metrics ) ; // Generate a span for the partition if // - these metrics correspond to a partition // - per partition tracing is enabled if per_partition_tracing { if let Some ( partition ) = partition { let mut partition_span = operator_span . child ( format ! ( " " operator_name partition ) ) ; partition_span . start = Some ( partition_start_time ) ; partition_span . end = Some ( partition_end_time ) ; partition_metrics . add_to_span ( & mut partition_span ) ; partition_span . export ( ) ; } } } } } // If we've not encountered any metrics to determine the operator's start // and end time, use those of the parent if operator_start_time == chrono :: MAX_DATETIME { operator_start_time = parent_span . start . unwrap_or ( default_end_time ) ; } if operator_end_time == chrono :: MIN_DATETIME { operator_end_time = parent_span . end . unwrap_or ( default_end_time ) ; } operator_span . start = Some ( operator_start_time ) ; operator_span . end = Some ( operator_end_time ) ; // recurse for child in physical_plan . children ( ) { send_metrics_to_tracing ( operator_end_time , & operator_span , child . as_ref ( ) , per_partition_tracing , ) ; } operator_metrics . add_to_span ( & mut operator_span ) ; operator_span . export ( ) ; }
This function translates data in DataFusion `MetricSets` into IOx
[`Span`]s. It records a snapshot of the current state of the
DataFusion metrics, so it should only be invoked *after* a plan is
fully `collect`ed.
This function translates data in DataFusion ` MetricSets ` into IOx [ ` Span ` ] s . It records a snapshot of the current state of the DataFusion metrics so it should only be invoked * after * a plan is fully ` collect ` ed .
default_end_time
DateTime<Utc>
parent_span
&Span
physical_plan
&dyn ExecutionPlan
per_partition_tracing
bool