algeon-datafusion 0.1.2

//! Final-plan coverage projections and candidate-selection diagnostics.

use datafusion::physical_plan::ExecutionPlan;

use super::{
    model::{
        QueryGpuCandidateOutcome, QueryGpuCandidateReasonKind, QueryGpuCugraphCall, QueryGpuRuntimeCaveat,
        QueryGpuValidatedUnder,
    },
    remedy::cost_remedy,
};
#[cfg(feature = "cugraph")]
use crate::{cugraph_sql::CugraphAlgoExec, final_plan_observation::cugraph_input_runs_natively};
use crate::{
    native_report::candidate_shape_family,
    planner::FinalPlanCoverage,
    report::{CandidateOutcome, PlanningReport},
};

pub(super) fn cugraph_calls(plan: &dyn ExecutionPlan) -> Vec<QueryGpuCugraphCall> {
    let mut calls = Vec::new();
    collect_cugraph_calls(plan, &mut calls);
    calls
}

#[cfg(feature = "cugraph")]
fn collect_cugraph_calls(plan: &dyn ExecutionPlan, calls: &mut Vec<QueryGpuCugraphCall>) {
    if let Some(cugraph) = plan.downcast_ref::<CugraphAlgoExec>() {
        let inputs_native = cugraph.children().iter().all(|input| cugraph_input_runs_natively(input.as_ref()));
        let mut notes = vec![
            "physical cuGraph plan constructed".to_string(),
            "runtime seed, personalization, and device-side validation are not decided at plan time".to_string(),
        ];
        if !inputs_native {
            notes.push(format!(
                "an input relation runs on DataFusion and is staged through host Arrow; the candidate rows labelled {}.<relation> give the reason",
                cugraph.coverage_algorithm_name()
            ));
        }
        calls.push(QueryGpuCugraphCall {
            algorithm: cugraph.coverage_algorithm_name().to_string(),
            statically_valid: true,
            inputs_native,
            notes,
        });
    }
    for child in plan.children() {
        collect_cugraph_calls(child.as_ref(), calls);
    }
}

#[cfg(not(feature = "cugraph"))]
fn collect_cugraph_calls(_plan: &dyn ExecutionPlan, _calls: &mut Vec<QueryGpuCugraphCall>) {}

pub(super) fn candidate_outcomes_from_report(report: &PlanningReport) -> Vec<QueryGpuCandidateOutcome> {
    report
        .outcomes()
        .iter()
        .map(|outcome| match outcome {
            CandidateOutcome::Selected { candidate } => QueryGpuCandidateOutcome {
                selection: outcome.kind(),
                candidate_family: candidate_shape_family(candidate),
                reason_code: None,
                reason_kind: None,
                detail: candidate.shape.to_string(),
                category: None,
                remedy: None,
                blocking: None,
            },
            CandidateOutcome::NotSupported { candidate, capability_reason, detail } => QueryGpuCandidateOutcome {
                selection: outcome.kind(),
                candidate_family: candidate_shape_family(candidate),
                reason_code: Some(capability_reason.code().to_string()),
                reason_kind: Some(capability_reason.facts().kind),
                detail: candidate.shape.to_string(),
                category: Some(capability_reason.facts().triage.as_str().to_string()),
                remedy: Some(capability_reason.remedy(detail)),
                blocking: detail.blocking(),
            },
            CandidateOutcome::NotSelectedByCost { candidate, cost_reason } => QueryGpuCandidateOutcome {
                selection: outcome.kind(),
                candidate_family: candidate_shape_family(candidate),
                reason_code: Some(cost_reason.code().to_string()),
                reason_kind: Some(QueryGpuCandidateReasonKind::Cost),
                detail: candidate.shape.to_string(),
                category: Some("cost".to_string()),
                remedy: Some(cost_remedy(*cost_reason)),
                blocking: None,
            },
        })
        .collect()
}

pub(super) fn runtime_caveats(
    observation: FinalPlanCoverage,
    cugraph_calls: &[QueryGpuCugraphCall],
    validated_under: &QueryGpuValidatedUnder,
) -> Vec<QueryGpuRuntimeCaveat> {
    if !observation.has_gpu_runtime() {
        return Vec::new();
    }
    let mut caveats = vec![
        QueryGpuRuntimeCaveat::new(
            "runtime_gpu_admission_unknown",
            "query-service grants are acquired at execution time and can wait or fail under concurrency; admitted grants are immutable",
        ),
        QueryGpuRuntimeCaveat::new(
            "device_memory_runtime_unknown",
            "device allocation, reservation, and retry outcomes are data- and concurrency-dependent",
        ),
        QueryGpuRuntimeCaveat::new(
            "native_runtime_failure_has_no_cpu_rescue",
            "once native GPU execution starts, native failures are surfaced as failures rather than CPU fallback",
        ),
    ];
    if observation.host_materialization_count() > 0 {
        caveats.push(QueryGpuRuntimeCaveat::new(
            "host_output_materialization",
            "DataFusion-facing query output is materialized back to host Arrow outside the GPU-execution promise",
        ));
    }
    if !cugraph_calls.is_empty() {
        caveats.push(QueryGpuRuntimeCaveat::new(
            "cugraph_runtime_validation_unknown",
            "cuGraph seed, personalization, weight, and device-side checks can still fail at runtime",
        ));
    }
    if let Some(execution_mode) = &validated_under.execution_mode {
        caveats.push(QueryGpuRuntimeCaveat::new(
            "execution_mode_plan_time_only",
            format!(
                "validated under native pipeline execution mode {execution_mode}; bounded runtime validation still executes later"
            ),
        ));
    }
    caveats
}