Dans les environnements de traitement de données à haute performance, la gestion de la concurrence entre les requêtes analytiques lourdes et les requêtes interactives critiques est un défi majeur. Apache DataFusion, en tant que moteur de requête SQL extensible, propose des mécanismes pour structurer l'ordonnancement des tâches et permettre, via des implémentations personnalisées, la préemption de ressources par des processus prioritaires.

Structures de données pour la priorité : PriorityMap

Au cœur de la gestion de l'ordre d'exécution au sein de certains opérateurs (comme les agrégations TopK), DataFusion s'appuie sur une structure nommée PriorityMap. Ce composant est conçu pour maintenir un ensemble d'éléments triés selon des critères de priorité définis dans le schéma des données, souvent représentés par des types Int32 ou Utf8.

Dans l'implémentation des flux de données (streams), cette structure permet d'insérer des lignes de résultats et de ne conserver que les éléments les plus pertinents selon une hiérarchie spécifique. Par exemple, lors du traitement d'un flux de résultats, le moteur peut dynamiquement réévaluer l'importance d'un lot de données en fonction de métadonnées de priorité associées aux transactions.

Mécanismes d'interruption et de préemption

Bien qu'Apache DataFusion ne dispose pas d'une API native "stop-and-resume" universelle pour toutes les étapes de l'exécution, l'architecture basée sur le TaskContext et les ExecutionPlan permet d'injecter une logique de contrôle de flux. La préemption consiste ici à suspendre ou à céder les ressources CPU d'une requête de fond au profit d'une requête à haute priorité.

Voici un exemple conceptuel montrant comment un nœud d'exécution personnalisé peut surveiller l'état de priorité via le contexte de tâche pour décider d'une interruption volontaire :

fn supervise_execution(
    plan_node: Arc<dyn ExecutionPlan>, 
    task_ctx: Arc<TaskContext>
) -> Result<SendableRecordBatchStream> {
    let task_prio = task_ctx.session_config().get_priority_level();
    let global_scheduler = task_ctx.runtime_env().scheduler();

    // Vérification de la présence d'une requête urgente dans la file d'attente globale
    if global_scheduler.has_urgent_pending_tasks() && task_prio < PRIORITY_THRESHOLD {
        // Logique de mise en attente ou de réduction des ressources allouées
        task_ctx.yield_now(); 
        
        // Notification au gestionnaire de requêtes pour ré-ordonnancement
        global_scheduler.promote_urgent_tasks();
    }

    plan_node.execute(0, task_ctx)
}

Configuration des métadonnées de priorité

Pour que l'ordonnanceru puisse distinguer l'importance des requêtes, il est nécessaire d'injecter des métadonnées lors de la définition du schéma ou de la configuration de la session. Cela se traduit souvent par l'ajout de propriétés spécifiques dans les champs (Fields) du schéma Arrow.

Le code suivant illustre l'attribusion d'une valeur de priorité par défaut au niveau des métadonnées d'un champ de données :

let mut schema_meta = HashMap::new();
schema_meta.insert("x_engine_priority".to_string(), "high_availability".to_string());

let priority_col = Field::new("prio_level", DataType::Int32, false)
    .with_metadata(schema_meta);

let data_schema = Schema::new(vec![
    Field::new("query_id", DataType::Int64, false),
    priority_col,
]);

Cas d'usage critiques

L'implémentation d'une logique de préemption est particulièrement pertinente dans les scénarios suivants :

• Analyse en temps réel : Garantir que les calculs sur les flux de données récents ne soient pas ralentis par des requêtes historiques volumineuses.
• Garantie de SLA (Service Level Agreement) : Allouer dynamiquement plus de slots d'exécution aux utilisateurs "Premium" ou aux processus système critiques.
• Exploration interactive : Permettre aux analystes d'obtenir des résultats rapides sur des échantillons de données sans attendre la fin des batchs de nuit.

En exploitant les points d'extension du TaskContext, les développeurs peuvent créer des politiques d'ordonnancement sophistiquées qui transforment DataFusion en un moteur multi-tenant capable de gérer intelligemment la pression sur les ressources système.