Modifications de compatibilité dues aux modifications des requêtes de déduplication - Realtime Compute for Apache Flink

Cette rubrique décrit la compatibilité entre un job Flink et les données d'état utilisées pour le démarrer après modification des requêtes de déduplication. La déduplication s'appuie sur ROW_NUMBER() OVER (PARTITION BY ... ORDER BY ...) AS rk avec un filtre WHERE rk = 1. Lorsque vous modifiez une requête de déduplication et reprenez l'exécution d'un job, certaines modifications préservent la structure d'état existante, tandis que d'autres ne la préservent pas.

Les règles de compatibilité varient selon le mode de déduplication utilisé par votre requête : proctime ASC , proctime DESC ou rowtime . Identifiez votre mode avant d'appliquer les règles ci-dessous.

Référence rapide

Modification	proctime ASC	proctime DESC	rowtime
Ajouter, supprimer ou transformer un champ sélectionné qui n'est pas une clé de partition	Compatible	Incompatible (modifications du schéma)	Incompatible (modifications du schéma)
Modifier un champ sélectionné sans changer le schéma de sortie	Compatible	Compatible	Compatible
Réorganiser les clés de partition	Compatible	Compatible	Compatible
Ajouter, supprimer ou modifier les clés de partition	Incompatible	Incompatible	Incompatible
Changer le champ ORDER BY ou sa direction	Incompatible	Incompatible	Incompatible

Modifications compatibles

Déduplication basée sur proctime ASC

Ajoutez, supprimez ou transformez n'importe quel champ sélectionné qui n'est pas une clé de partition. Les modifications apportées aux colonnes de sortie n'affectent pas la structure de l'état.

-- Base query
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Fully compatible: Add d to selected fields
SELECT a, b, c, d FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Fully compatible: Remove b from selected fields
SELECT a, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Fully compatible: Transform c to SUBSTRING(c, 1, 5)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) AS rk
    FROM (SELECT a, b, SUBSTRING(c,1,5) as c, proctime FROM MyTable))
WHERE rk = 1;

Tous les modes de déduplication

La réorganisation des clés de partition est entièrement compatible. L'ensemble des clés — et par conséquent la structure de l'état indexé — reste inchangé.

-- Base query
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a, b, c ORDER BY proctime ASC) as rk
    FROM MyTable)
WHERE rk = 1;

-- Fully compatible: Reorder partition keys
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY c, a, b ORDER BY proctime ASC) as rk
    FROM MyTable)
WHERE rk = 1;

Déduplication basée sur rowtime ou proctime DESC

Modifiez un champ sélectionné sans changer le schéma de sortie. Les modifications qui n'altèrent pas la structure du schéma sont sûres.

-- Base query
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM MyTable)
WHERE rk = 1 AND c > 10;

-- Fully compatible: Remove b without changing the schema
SELECT a, b FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM MyTable)
WHERE rk = 1 AND c > 10;

Modifications incompatibles

Ajouter, supprimer ou modifier les clés de partition (tous les modes)

La clause PARTITION BY définit les clés utilisées pour rechercher l'état indexé. Toute modification de l'ensemble des clés de partition — ou du calcul qui produit ces clés — modifie l'espace de clés de l'état existant, ce qui le rend illisible par le nouvel opérateur.

-- Base query
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Add d as a partition key (incompatible)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a, d ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Remove a from the partition keys (incompatible)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Change partition key computation: a to a + 1 (incompatible)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (ORDER BY proctime ASC) AS rk
    FROM (SELECT a + 1 AS a, b, c, proctime FROM MyTable))
WHERE rk = 1;

Modifier la clause ORDER BY (tous les modes)

Le changement du champ de tri ou de l'ordre de tri modifie la sémantique de l'enregistrement conservé comme « premier » ou « dernier ». L'état existant a été construit selon l'ordre initial et ne peut pas être mappé vers un ordre différent.

-- Base query
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Change sort field from proctime to ts (incompatible)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY ts ASC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Change sort order from ASC to DESC (incompatible)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM MyTable)
WHERE rk = 1;

Ajouter, supprimer ou transformer des champs qui modifient le schéma (rowtime ou proctime DESC)

Dans les modes rowtime et proctime DESC, toute modification qui altère le schéma de sortie — même l'ajout ou la suppression d'un seul champ — rend l'état stocké structurellement incompatible avec le nouvel opérateur.

-- Base query
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Add d to selected fields (incompatible — schema changes)
SELECT a, b, c, d FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Remove c from selected fields (incompatible — schema changes)
SELECT a, b FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM MyTable)
WHERE rk = 1;

-- Transform c to substring(c, 1, 5) (incompatible — schema changes)
SELECT a, b, c FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY a ORDER BY proctime DESC) AS rk
    FROM (select a, b, substring(c, 1, 5) as c, ts from MyTable))
WHERE rk = 1;