Autres limites pour les modifications SQL - Realtime Compute for Apache Flink

Flink SQL est déclaratif : le planificateur de tables détermine automatiquement la topologie des opérateurs sous-jacents et la disposition de l'état. Toute modification du code SQL, même mineure, peut générer un plan d'exécution différent et empêcher le job de restaurer l'état à partir d'un point de contrôle (checkpoint) ou d'un point de sauvegarde (savepoint). Cette rubrique présente les limites relatives aux changements autres que les modifications apportées aux requêtes, aux tables source et aux tables de destination (sink).

Référence de compatibilité

Le tableau ci-dessous résume tous les types de changement couverts et leurs résultats.

Type de changement	Résultat de compatibilité
Version différente d'Apache Flink	Inconnu
Connecteurs personnalisés ou dépendances UDF modifiés	Inconnu (vérification manuelle requise)
Plus d'un élément modifié avant une vérification de compatibilité	Inconnu
Ajout d'une requête introduisant un nouvel état	Incompatible
Suppression d'une table de destination et modification ou suppression d'une instruction TEMPORARY TABLE	Inconnu
Suppression d'une table de destination sans toucher à aucune instruction TEMPORARY TABLE	Compatible

Limites

La version d'Apache Flink doit correspondre

Le job redémarré doit utiliser la même version d'Apache Flink que le job ayant généré le point de contrôle ou le point de sauvegarde. Flink ne peut pas déterminer la compatibilité entre différentes versions.

Les dépendances doivent être compatibles

Les dépendances du job redémarré doivent être compatibles avec celles du job d'origine. Si vous modifiez des connecteurs personnalisés ou les dépendances des fonctions définies par l'utilisateur (UDF), vérifiez la compatibilité manuellement : Flink ne détecte pas ces problèmes automatiquement.

La modification de plusieurs éléments entraîne une compatibilité inconnue

Si vous modifiez plus d'un élément avant une vérification de compatibilité, Flink ne peut pas déterminer celle-ci. Un élément désigne une fonction d'agrégation, une table de destination ou une clause WHERE qui affecte le calcul avec état.

Les exemples suivants entraînent une compatibilité inconnue :

-- Original SQL statements:
CREATE TABLE MyTable (
  a int,
  b bigint,
  c varchar
) WITH (
  'connector' = 'datagen'
);

CREATE TABLE MySink (
  a int,
  b bigint,
  c varchar
) WITH (
  'connector' = 'print'
);

INSERT INTO MySink SELECT a, sum(b), max(c) FROM MyTable GROUP BY a;

-- Example 1: Rename the sink table to MySink2 and change the aggregate function from max(c) to min(c).
-- Result: unknown compatibility (two items changed: sink table + aggregate function)
CREATE TABLE MySink2 (
  a int,
  b bigint,
  c varchar
) WITH (
  'connector' = 'print'
);

INSERT INTO MySink2 SELECT a, sum(b), min(c) FROM MyTable GROUP BY a;

-- Example 2: Add a WHERE clause that contains the a > 10 condition, set
-- table.optimizer.state-compatibility.ignore-filter to true, and change
-- the aggregate function from max(c) to min(c).
-- Result: unknown compatibility (two items changed: WHERE clause + aggregate function)
INSERT INTO MySink
SELECT a, sum(b), min(c) FROM (
  SELECT * FROM MyTable WHERE a > 10
) GROUP BY a;

L'ajout d'une requête générant un nouvel état provoque un changement incompatible

Si vous ajoutez une requête introduisant un nouveau calcul avec état, le job ne peut pas restaurer l'état à partir du point de contrôle ou du point de sauvegarde existant.

-- Original SQL statements:
CREATE TABLE MyTable (
  a int,
  b bigint,
  c varchar
) WITH (
  'connector' = 'datagen'
);

CREATE TABLE MySink (
  a int,
  b bigint,
  c varchar
) WITH (
  'connector' = 'print'
);

INSERT INTO MySink SELECT a, b, c FROM MyTable;

-- Adding a group aggregate query generates new state data — incompatible change.
INSERT INTO MySink SELECT a, sum(b), min(c) FROM MyTable GROUP BY a;

La suppression d'une table de destination avec des modifications TEMPORARY TABLE entraîne une compatibilité inconnue

Si vous supprimez une table de destination et modifiez ou supprimez également l'instruction TEMPORARY TABLE associée à cette table de destination ou à sa table source correspondante, Flink ne peut pas déterminer la compatibilité. En revanche, si vous supprimez une table de destination sans modifier aucune instruction TEMPORARY TABLE, la compatibilité n'est pas affectée.

-- Original SQL statements

-- Source table 1
CREATE TEMPORARY TABLE MyTable (
  a int,
  b bigint,
  c bigint,
  ts timestamp(3),
  proctime as proctime(),
  WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen');

-- Source table 2
CREATE TEMPORARY TABLE MyTable2 (
  a int,
  b bigint,
  c bigint,
  ts timestamp(3),
  proctime as proctime(),
  WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen');

-- Sink table 1
CREATE TEMPORARY TABLE MySink (a int, b bigint) WITH ('connector' = 'print');
-- Sink table 2
CREATE TEMPORARY TABLE MySink2 (a int, b bigint) WITH ('connector' = 'print');

-- Query
BEGIN STATEMENT SET;
INSERT INTO MySink SELECT a, sum(b) FROM MyTable GROUP BY a;
INSERT INTO MySink2 SELECT a, b FROM MyTable2 WHERE a > 10;
END;

-- Example 1: Delete MySink2 and modify the TEMPORARY TABLE for MyTable (add column d bigint).
-- Result: unknown compatibility

-- Source table 1 (modified: column d bigint added)
CREATE TEMPORARY TABLE MyTable (
  a int,
  b bigint,
  c bigint,
  d bigint,
  ts timestamp(3),
  proctime as proctime(),
  WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen');

-- Sink table 1
CREATE TEMPORARY TABLE MySink (a int, b bigint) WITH ('connector' = 'print');

-- Query
INSERT INTO MySink SELECT a, sum(b) FROM MyTable GROUP BY a;

-- Example 2: Delete MySink2 without modifying any TEMPORARY TABLE statement.
-- Result: fully compatible

-- Source table 1
CREATE TEMPORARY TABLE MyTable (
  a int,
  b bigint,
  c bigint,
  ts timestamp(3),
  proctime as proctime(),
  WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen');

-- Source table 2
CREATE TEMPORARY TABLE MyTable2 (
  a int,
  b bigint,
  c bigint,
  ts timestamp(3),
  proctime as proctime(),
  WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen');

-- Sink table 1
CREATE TEMPORARY TABLE MySink (a int, b bigint) WITH ('connector' = 'print');
-- Sink table 2
CREATE TEMPORARY TABLE MySink2 (a int, b bigint) WITH ('connector' = 'print');

-- Query
INSERT INTO MySink SELECT a, sum(b) FROM MyTable GROUP BY a;