Modifications de compatibilité dues aux changements dans les requêtes d'agrégation par groupe - Realtime Compute for Apache Flink

La compatibilité d'état détermine si un job Flink peut reprendre son exécution à partir des données d'état existantes après la modification d'une requête d'agrégation par groupe. Cette rubrique décrit les modifications compatibles et incompatibles, ainsi que la marche à suivre dans chaque cas.

Fonctionnement

Lorsque Flink exécute une requête d'agrégation par groupe, il stocke les résultats intermédiaires sous forme de données d'état. Cet état est indexé par les champs GROUP BY et organisé selon les champs agrégés. Lorsque vous modifiez la requête et redémarrez le job avec un état existant, Flink vérifie si la nouvelle requête peut lire cet état stocké et poursuivre l'accumulation des données.

La compatibilité comporte trois niveaux :

Niveau	Signification
Totalement compatible	La nouvelle requête lit et réutilise l'intégralité de l'état existant sans aucune perte de données.
Partiellement compatible	La nouvelle requête démarre à partir de l'état existant, mais les champs nouvellement ajoutés commencent leur accumulation au redémarrage du job (et non depuis le début du flux). L'état des champs supprimés est ignoré.
Incompatible	La nouvelle requête ne peut pas réutiliser l'état existant. Démarrer à partir de l'état existant produirait des résultats imprévisibles.

Pourquoi les champs DISTINCT obéissent-ils à des règles plus strictes ? Flink utilise l'ordre relatif des fonctions d'agrégation DISTINCT comme partie intégrante de la structure interne de la clé d'état. Modifier cet ordre réécrit le mappage des clés, rendant l'état existant illisible pour la nouvelle requête.

Modifications compatibles

Ajouter, supprimer ou modifier des champs agrégés non-DISTINCT

Modification	Compatibilité	Comportement
Ajout d'un champ agrégé	Partiellement compatible	Le champ ajouté commence son accumulation au redémarrage du job.
Suppression d'un champ agrégé	Totalement compatible	L'état du champ supprimé est ignoré.
Ajout et suppression simultanés de champs agrégés	Partiellement compatible	Le champ ajouté commence son accumulation au redémarrage du job. L'état du champ supprimé est ignoré.
Modification d'un champ agrégé existant	Partiellement compatible	Traitée comme une suppression suivie d'un ajout : l'état du champ d'origine est ignoré ; le nouveau champ démarre au redémarrage du job.

Les champs non modifiés ne sont pas affectés. Leurs résultats de calcul restent identiques, que vous utilisiez ou non les données d'état.

-- Original SQL statement:
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Partially compatible: Add COUNT(c).
-- SUM(b) and MAX(c) are unaffected. COUNT(c) starts from 0 at job restart.
SELECT a, SUM(b), MAX(c), COUNT(c) FROM MyTable GROUP BY a;

-- Fully compatible: Delete SUM(b).
-- MAX(c) is unaffected.
SELECT a, MAX(c) FROM MyTable GROUP BY a;

-- Partially compatible: Replace MAX(c) with MIN(c).
-- SUM(b) is unaffected. MAX(c) state is discarded. MIN(c) starts from job restart.
SELECT a, SUM(b), MIN(c) FROM MyTable GROUP BY a;

Réorganiser les champs agrégés non-DISTINCT

La réorganisation des champs agrégés qui ne contiennent pas le mot-clé DISTINCT est totalement compatible.

-- Original SQL statement:
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Fully compatible: Swap the order of SUM(b) and MAX(c).
-- Both fields are unaffected.
SELECT a, MAX(c), SUM(b) FROM MyTable GROUP BY a;

Modifier la logique de calcul d'un champ agrégé non-DISTINCT

La modification de la logique de calcul — par exemple, l'encapsulation d'un argument dans une fonction scalaire — est partiellement compatible. Le champ d'origine est traité comme supprimé et le nouveau champ démarre au redémarrage du job.

-- Original SQL statement:
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Partially compatible: Apply SUBSTRING before MAX.
-- SUM(b) is unaffected. MAX(c) state is discarded. MAX(SUBSTRING(c,1,5)) starts from job restart.
SELECT a, SUM(b), MAX(c) FROM (
  SELECT a, b, SUBSTRING(c, 1, 5) AS c FROM MyTable
) GROUP BY a;

Réorganiser toutes les fonctions d'agrégation tout en conservant l'ordre relatif des fonctions DISTINCT

Totalement compatible, à condition que l'ordre relatif des fonctions d'agrégation DISTINCT ne change pas.

-- Original SQL statement:
INSERT INTO MySink
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b) FROM MyTable GROUP BY a;

-- Fully compatible: MAX(b) is moved, but SUM(DISTINCT b) still precedes COUNT(DISTINCT b).
INSERT INTO MySink
SELECT a, SUM(DISTINCT b), COUNT(DISTINCT b), MAX(b) FROM MyTable GROUP BY a;

Autres modifications totalement compatibles

Absence de champs agrégés avant et après la modification — totalement compatible.
Suppression d'une fonction d'agrégation prenant en charge la rétraction — totalement compatible lorsque la suppression suit une opération de rétraction.

-- Original SQL statement:
SELECT c/2, AVG(avg_a) AS avg_avg_a, MAX(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

-- Fully compatible: Delete MAX(max_b).
SELECT c/2, AVG(avg_a) AS avg_avg_a FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

Modifications incompatibles

Les modifications suivantes ne permettent pas de réutiliser l'état existant.

Modifier les champs GROUP BY

L'ajout, la suppression ou la modification des champs dans la clause GROUP BY, ou encore le changement de leur logique de calcul, modifie la structure de la clé d'état et rend la modification incompatible.

-- Original SQL statement:
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Incompatible: Add d to GROUP BY.
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a, d;

-- Incompatible: Remove GROUP BY entirely.
SELECT SUM(b), MAX(c) FROM MyTable;

-- Incompatible: Change GROUP BY field from a to d.
SELECT d, SUM(b), MIN(c) FROM MyTable GROUP BY d;

-- Incompatible: Change GROUP BY field from a to a + 1.
SELECT a, SUM(b), MAX(c) FROM (
  SELECT a + 1 AS a, b, c FROM MyTable
) GROUP BY a;

Ajouter, supprimer ou modifier des champs agrégés DISTINCT

Toute modification apportée aux champs utilisant le mot-clé DISTINCT est incompatible.

-- Original SQL statement:
SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- Incompatible: Add COUNT(DISTINCT b).
SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- Incompatible: Delete SUM(DISTINCT b).
SELECT a, SUM(b), MAX(c), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- Incompatible: Change SUM(DISTINCT b) to AVG(DISTINCT b).
SELECT a, SUM(b), MAX(c), AVG(DISTINCT b), COUNT(DISTINCT c) FROM MyTable GROUP BY a;

-- Incompatible: Change computation logic of COUNT(DISTINCT c).
SELECT a, SUM(b), MAX(c), SUM(DISTINCT b), COUNT(DISTINCT c) FROM (
     SELECT a, b, AVG(c) AS c FROM MyTable GROUP BY a, b
) GROUP BY a;

Ajouter un champ agrégé dans une agrégation multiniveau

Dans le cadre d'une agrégation multiniveau, des opérations de rétraction peuvent survenir entre les niveaux d'agrégation. L'ajout d'un champ dans ce contexte produit des résultats imprévisibles et est donc incompatible.

-- Original SQL statement:
SELECT a/2, AVG(b), MIN(c) FROM (
    SELECT a, SUM(b) AS b, MAX(c) AS c FROM MyTable GROUP BY a
) GROUP BY a/2;

-- Incompatible: Add COUNT(c) in the outer aggregation.
SELECT a/2, AVG(b), MIN(c), COUNT(c) FROM (
    SELECT a, SUM(b) AS b, MAX(c) AS c FROM MyTable GROUP BY a
) GROUP BY a/2;

Supprimer tous les champs agrégés

La suppression de tous les champs agrégés entraîne l'abandon de toutes les données d'état sans aucune réutilisation.

-- Original SQL statement:
SELECT a, SUM(b), MAX(c) FROM MyTable GROUP BY a;

-- Incompatible: Delete both SUM(b) and MAX(c).
SELECT a FROM MyTable GROUP BY a;

Changer l'ordre relatif des fonctions d'agrégation DISTINCT

-- Original SQL statement:
INSERT INTO MySink
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b) FROM MyTable GROUP BY a;

-- Incompatible: COUNT(DISTINCT b) now precedes SUM(DISTINCT b).
INSERT INTO MySink
SELECT COUNT(DISTINCT b), a, MAX(b), SUM(DISTINCT b) FROM MyTable GROUP BY a;

Ajouter des champs agrégés à un job ne contenant aucun champ agrégé existant

-- Original SQL statement:
INSERT INTO MySink
SELECT a, b FROM MyTable GROUP BY a, b;

-- Incompatible: Add SUM(b).
INSERT INTO MySink
SELECT a, b, SUM(b) FROM MyTable GROUP BY a, b;

Conserver un seul champ agrégé et modifier sa logique de calcul

-- Original SQL statement:
INSERT INTO MySink
SELECT a, SUM(b), MAX(b), MAX(c) FROM MyTable GROUP BY a;

-- Incompatible: Keep only MAX(c) and change its input expression.
INSERT INTO MySink
SELECT a, MAX(c) FROM (SELECT a, b, c + 1 AS c, ts FROM MyTable) GROUP BY a;

Modifier tous les champs agrégés

-- Original SQL statement:
INSERT INTO MySink
SELECT a, b, MAX(c) FROM MyTable GROUP BY a, b;

-- Incompatible: Replace MAX(c) with MIN(c) (the only aggregated field).
INSERT INTO MySink
SELECT a, b, MIN(c) FROM MyTable GROUP BY a, b;

Ajouter ou modifier une fonction d'agrégation prenant en charge la rétraction

-- Original SQL statement:
SELECT c/2, AVG(avg_a) AS avg_avg_a, MAX(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

-- Incompatible: Add MIN(max_b), which supports retraction.
SELECT c/2, AVG(avg_a) AS avg_avg_a, MIN(max_b) min_max_b, MAX(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

-- Incompatible: Replace MAX(max_b) with MIN(max_b).
SELECT c/2, AVG(avg_a) AS avg_avg_a, MIN(max_b) max_max_b FROM
    (SELECT c, MAX(b) AS max_b, AVG(a) AS avg_a FROM MyTable GROUP BY c)
GROUP BY c/2;

Modifications dont la compatibilité est inconnue

Si une fonction d'agrégation définie par l'utilisateur (UDAF) Python est utilisée avant ou après la modification, le système ne peut pas déterminer la compatibilité.

-- Unknown compatibility: The query uses a Python UDAF (weighted_avg).
SELECT a, MAX(b), SUM(DISTINCT b), COUNT(DISTINCT b), weighted_avg(a, b)
FROM MyTable GROUP BY a, b;