Group Window Aggregate change compatibility - Realtime Compute for Apache Flink

Lorsque vous modifiez l'instruction SQL d'un déploiement Realtime Compute for Apache Flink en cours d'exécution, Flink évalue si les données d'état existantes peuvent être réutilisées. Comprendre quelles modifications préservent la compatibilité des états vous aide à éviter toute perte de données inattendue ou un retraitement complet.

Cette rubrique présente les résultats de compatibilité des états pour les instructions SQL qui utilisent des fonctions d'agrégation par fenêtre avec une clause GROUP BY.

Une modification peut avoir trois issues :

Entièrement compatible : le déploiement se poursuit en conservant l'intégralité des données d'état historiques.
Partiellement compatible : les agrégations non modifiées reprennent à partir de l'état historique ; les agrégations nouvellement ajoutées redémarrent à zéro.
Incompatible : le déploiement ne peut réutiliser aucune donnée d'état et doit redémarrer depuis le début.

Comment Flink évalue la compatibilité des états

Flink évalue la réutilisation des états en fonction des composants de l'instruction SQL modifiés :

Composant	Flexibilité	Exemples
Agrégations non distinctes	Flexible — compatibilité partielle possible	`SUM(b)`, `MAX(c)`, `COUNT(c)`
Champs d'attribut de fenêtre	Flexible — les modifications sont entièrement compatibles	`tumble_start`, `tumble_end`
Position de la fonction de fenêtre dans GROUP BY	Flexible — le réordonnancement de la clé de fenêtre est entièrement compatible	Position de `TUMBLE(ts, ...)` dans `GROUP BY`
Définition de la fenêtre	Fixe — toute modification rompt la compatibilité	Type de fenêtre, taille de fenêtre, attribut temporel
Clés GROUP BY (dimensions statistiques)	Fixe — toute modification rompt la compatibilité	`GROUP BY a, d, ...`
Agrégations distinctes	Fixe — toute modification rompt la compatibilité	`COUNT(DISTINCT c)`
Paramètres early-fire et late-fire	Fixe — toute modification rompt la compatibilité	`table.exec.emit.early-fire.enabled`

Ce cadre explique pourquoi certaines modifications peuvent être partiellement prises en charge (Flink peut supprimer ou ajouter des agrégations individuelles sans reconstruire l'état complet), tandis que d'autres nécessitent un redémarrage complet de l'état (la structure fondamentale du calcul change).

Référence rapide

Utilisez ce tableau pour déterminer l'impact sur la compatibilité d'une modification spécifique avant de l'appliquer.

Modification	Compatibilité
Ajout d'une agrégation non distincte	Partielle
Suppression d'une agrégation non distincte	Totale
Ajout et suppression d'agrégations non distinctes en une seule modification	Partielle
Modification d'une agrégation non distincte	Partielle
Réordonnancement d'agrégations non distinctes	Totale
Modification de la logique de calcul d'un champ dans une agrégation non distincte	Partielle
Ajout ou suppression d'un champ d'attribut de fenêtre (`tumble_start`, `tumble_end`)	Totale
Réordonnancement des clés GROUP BY — seule la position de la clé de fonction de fenêtre change	Totale
Aucune agrégation avant ou après la modification	Totale
Modification du type de fenêtre (par exemple, TUMBLE vers HOP)	Incompatible
Modification de la taille de la fenêtre	Incompatible
Modification de l'attribut temporel (`rowtime` vers `proctime`, ou inversement)	Incompatible
Ajout, suppression ou modification d'une clé GROUP BY (dimension statistique)	Incompatible
Ajout, suppression ou modification d'une agrégation distincte	Incompatible
Suppression de toutes les agrégations	Incompatible
Ajout ou suppression de early-fire ou late-fire	Incompatible
Réordonnancement des clés GROUP BY non liées aux fenêtres	Incompatible
Ajout d'une agrégation à un déploiement qui n'en contenait aucune	Incompatible
Une seule agrégation existe et sa logique de calcul change	Incompatible
L'ensemble des agrégations change entièrement (aucun chevauchement)	Incompatible
L'instruction SQL contient `'table.exec.emit.early-fire.enabled' = 'true'` ou `'table.exec.emit.late-fire.enabled' = 'true'` avant ou après la modification	Incompatible
L'instruction SQL contient une fonction d'agrégation définie par l'utilisateur Python (UDAF) avant ou après la modification	Inconnue

Modifications qui préservent ou préservent partiellement l'état

Ajouter, supprimer ou modifier une agrégation non distincte

Les modifications suivantes entraînent une compatibilité partielle ou totale :

Ajout : partiellement compatible. La nouvelle agrégation commence à compter à partir de zéro au redémarrage du déploiement.
Suppression : entièrement compatible. Les données d'état de l'agrégation supprimée sont ignorées ; toutes les autres agrégations restent inchangées.
Ajout et suppression en une seule modification : partiellement compatible. L'agrégation ajoutée démarre à zéro ; l'état de l'agrégation supprimée est ignoré.
Modification (remplacement d'une agrégation par une autre) : partiellement compatible. L'agrégation d'origine est traitée comme supprimée et son état est ignoré ; la nouvelle agrégation est traitée comme ajoutée et démarre à zéro.

Pour les agrégations que vous ne modifiez pas, les résultats après réutilisation de l'état sont identiques aux résultats calculés à partir de l'historique complet des données.

-- Original SQL statement
SELECT a, SUM(b), MAX(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Add COUNT(c): partially compatible.
-- SUM(b) and MAX(c) are unaffected. COUNT(c) starts from 0.
SELECT a, SUM(b), MAX(c), COUNT(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Delete SUM(b): fully compatible.
-- MAX(c) is unaffected.
SELECT a, MAX(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Change MAX(c) to MIN(c): partially compatible.
-- SUM(b) is unaffected. MAX(c) state is discarded; MIN(c) starts from 0.
SELECT a, SUM(b), MIN(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

Réordonner des agrégations non distinctes

Le réordonnancement des agrégations dans la liste SELECT n'affecte pas la compatibilité. Tous les résultats restent corrects.

-- Original SQL statement
SELECT a, SUM(b), MAX(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Swap SUM(b) and MAX(c): fully compatible.
SELECT a, MAX(c), SUM(b)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

Modifier la logique de calcul d'un champ dans une agrégation non distincte

La modification de la manière dont un champ d'entrée est calculé compte comme une modification de l'agrégation. L'agrégation d'origine est ignorée et la nouvelle démarre à zéro.

-- Original SQL statement
SELECT a, SUM(b), MAX(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Change MAX(c) to MAX(SUBSTRING(c, 1, 5)): partially compatible.
-- SUM(b) is unaffected. MAX(c) state is discarded; MAX(SUBSTRING(c, 1, 5)) starts from 0.
SELECT a, SUM(b), MAX(c)
FROM (SELECT a, b, SUBSTRING(c, 1, 5) AS c FROM MyTable)
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

Ajouter ou supprimer un champ d'attribut de fenêtre

L'ajout ou la suppression de tumble_start, tumble_end ou de champs d'attribut de fenêtre similaires est entièrement compatible.

-- Original SQL statement
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Add window_end: fully compatible.
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start,
  tumble_end(ts, INTERVAL '1' MINUTE) AS window_end
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Delete window_start: fully compatible.
SELECT a,
  SUM(b),
  MAX(c)
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

Réordonner les clés GROUP BY — uniquement la clé de fonction de fenêtre

Si seule la position de la clé de fonction de fenêtre dans la clause GROUP BY change et que toutes les autres clés conservent le même ordre, la compatibilité est préservée.

-- Original SQL statement
SELECT a, SUM(DISTINCT b), MAX(DISTINCT c), COUNT(c)
FROM MyTable
GROUP BY a, b, TUMBLE(rowtime, INTERVAL '15' MINUTE);

-- Move the window function key: fully compatible.
-- The positions of a and b are unchanged.
SELECT a, SUM(DISTINCT b), MAX(DISTINCT c), COUNT(c)
FROM MyTable
GROUP BY a, TUMBLE(rowtime, INTERVAL '15' MINUTE), b;

Modifications qui rompent la compatibilité des états

Les modifications suivantes rendent un déploiement incompatible avec ses données d'état existantes. Le déploiement doit ignorer tout l'état et recalculer les résultats depuis le début de ses entrées.

Modifier les attributs de fenêtre

La modification du type de fenêtre, de la taille de la fenêtre ou de l'attribut temporel rompt la compatibilité.

-- Original SQL statement
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Change window type from TUMBLE to HOP: incompatible.
SELECT a,
  SUM(b),
  MAX(c),
  hop_start(ts, INTERVAL '1' MINUTE, INTERVAL '2' MINUTE) AS window_start
FROM MyTable
GROUP BY a, HOP(ts, INTERVAL '1' MINUTE, INTERVAL '2' MINUTE);

-- Change window size from 1 minute to 2 minutes: incompatible.
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '2' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '2' MINUTE);

-- Change time attribute from ts (rowtime) to proctime: incompatible.
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(proctime, INTERVAL '1' MINUTE);

Ajouter, supprimer ou modifier une clé GROUP BY

Les clés GROUP BY définissent les dimensions statistiques. Toute modification de ces clés, y compris la modification de la logique de calcul d'un champ utilisé comme clé, rompt la compatibilité.

-- Original SQL statement
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Add GROUP BY key d: incompatible.
SELECT a,
  SUM(b),
  MAX(c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, d, TUMBLE(ts, INTERVAL '1' MINUTE);

Ajouter, supprimer ou modifier une agrégation distincte

Toute modification des agrégations utilisant DISTINCT rompt la compatibilité, même si toutes les autres agrégations restent inchangées.

-- Original SQL statement
SELECT a,
  SUM(b),
  MAX(c),
  COUNT(DISTINCT c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Add COUNT(DISTINCT b): incompatible.
SELECT a,
  SUM(b),
  COUNT(DISTINCT b),
  MAX(c),
  COUNT(DISTINCT c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

Supprimer toutes les agrégations

La suppression de chaque agrégation de l'instruction SQL rompt la compatibilité. Tout l'état est ignoré.

-- Original SQL statement
SELECT a,
  SUM(b),
  COUNT(DISTINCT b),
  MAX(c),
  COUNT(DISTINCT c),
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

-- Delete all aggregates: incompatible.
SELECT a,
  tumble_start(ts, INTERVAL '1' MINUTE) AS window_start
FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' MINUTE);

Ajouter ou supprimer early-fire ou late-fire

Toute modification des paramètres d'émission early-fire ou late-fire rompt la compatibilité.

Réordonner les clés GROUP BY non liées aux fenêtres

La modification de l'ordre des clés GROUP BY autres que la clé de fonction de fenêtre rompt la compatibilité.

-- Original SQL statement
SELECT a, SUM(DISTINCT b), MAX(DISTINCT c), COUNT(c)
FROM MyTable
GROUP BY a, b, TUMBLE(rowtime, INTERVAL '15' MINUTE);

-- Swap a and b: incompatible.
SELECT a, SUM(DISTINCT b), MAX(DISTINCT c), COUNT(c)
FROM MyTable
GROUP BY b, a, TUMBLE(rowtime, INTERVAL '15' MINUTE);

Ajouter une agrégation à un déploiement sans agrégations

Si l'instruction SQL d'origine ne contient aucune agrégation et que vous en ajoutez une, la compatibilité est rompue.

-- Original SQL statement (no aggregates)
SELECT a, b, c
FROM MyTable
GROUP BY a, b, c, TUMBLE(rowtime, INTERVAL '15' MINUTE);

-- Add COUNT(c): incompatible.
SELECT a, b, c, COUNT(c)
FROM MyTable
GROUP BY a, b, c, TUMBLE(rowtime, INTERVAL '15' MINUTE);

Une seule agrégation existe et sa logique de calcul change

Si le déploiement comporte exactement une agrégation et que vous modifiez la manière dont elle est calculée, la compatibilité est rompue.

-- Original SQL statement
SELECT a, SUM(b), MAX(b), MAX(c)
FROM MyTable
GROUP BY b, a, TUMBLE(ts, INTERVAL '15' MINUTE);

-- Modify the only remaining aggregate: incompatible.
SELECT a, MAX(c)
FROM (SELECT a, b, c + 1 AS c, ts FROM MyTable)
GROUP BY b, a, TUMBLE(ts, INTERVAL '15' MINUTE);

Les métriques statistiques avant et après la modification sont entièrement différentes

Si aucune agrégation d'avant la modification ne subsiste après celle-ci, la compatibilité est rompue.

-- Original SQL statement
SELECT a, SUM(b), MAX(b), MAX(c) FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' SECOND);

-- Replace with a completely different set of aggregates: incompatible.
SELECT a, MIN(b), AVG(b) FROM MyTable
GROUP BY a, TUMBLE(ts, INTERVAL '1' SECOND);

L'instruction SQL contient une configuration early-fire ou late-fire

Si l'instruction SQL inclut 'table.exec.emit.early-fire.enabled' = 'true' ou 'table.exec.emit.late-fire.enabled' = 'true' soit avant, soit après la modification, la compatibilité est rompue.

-- Original SQL statement
SELECT a, MAX(c)
FROM (SELECT a, b, c + 1 AS c, ts FROM MyTable)
GROUP BY b, a, TUMBLE(ts, INTERVAL '15' MINUTE);

-- Add early-fire or late-fire configuration: incompatible.
SET 'table.exec.emit.early-fire.enabled' = 'true';
SET 'table.exec.emit.early-fire.delay' = '500ms';
-- or
SET 'table.exec.emit.late-fire.enabled' = 'true';
SET 'table.exec.emit.late-fire.delay' = '1s';
SET 'table.exec.emit.allow-lateness' = '5s';

SELECT a, MAX(c)
FROM (SELECT a, b, c + 1 AS c, ts FROM MyTable)
GROUP BY b, a, TUMBLE(ts, INTERVAL '15' MINUTE);

L'instruction SQL contient une UDAF Python

Si l'instruction SQL inclut une fonction d'agrégation définie par l'utilisateur Python (UDAF) soit avant, soit après la modification, la compatibilité entre le déploiement et ses données d'état est inconnue.

-- SQL with a Python UDAF (weighted_avg) before or after modification — compatibility is unknown.
SELECT COUNT(DISTINCT b), a, SUM(DISTINCT b), weighted_avg(a, b)
FROM MyTable
GROUP BY a, c;