Use MERGE INTO to combine INSERT, UPDATE, and DELETE commands - MaxCompute

Lorsque vous devez appliquer des opérations INSERT, UPDATE et DELETE à une table transactionnelle ou à une table Delta, l'exécution de ces opérations sous forme d'instructions distinctes nécessite plusieurs analyses complètes de la table. MERGE INTO combine ces trois opérations en une seule instruction et n'analyse la table cible qu'une seule fois, ce qui réduit le temps d'exécution et élimine le risque d'échec partiel impossible à annuler.

Les cas d'utilisation courants incluent l'insertion ou la mise à jour (upsert) des données source dans une table cible, l'application des événements de capture des données modifiées (CDC) et la synchronisation des partitions.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Des autorisations Select et Update sur la table transactionnelle cible

Pour la configuration des autorisations, consultez Autorisations MaxCompute.

Fonctionnement

MERGE INTO joint une table source (ou une sous-requête) à la table cible à l'aide de la condition ON, puis applique chaque clause WHEN au résultat de la jointure :

WHEN MATCHED : lignes de la table cible qui correspondent à une ligne source — appliquez UPDATE ou DELETE.
WHEN NOT MATCHED : lignes de la table source sans ligne correspondante dans la table cible — appliquez INSERT.

Toutes les opérations s'exécutent de manière atomique : si une opération échoue, l'intégralité de l'instruction est annulée et aucune modification n'est validée. Cela garantit la cohérence d'une manière que des instructions INSERT, UPDATE et DELETE séparées ne permettent pas : un échec au milieu d'instructions distinctes laisse les modifications déjà validées sans possibilité d'annulation.

Limitations

Une seule instruction MERGE INTO ne peut pas effectuer plusieurs opérations INSERT ou UPDATE sur les mêmes lignes.

Syntaxe

MERGE INTO <target_table> [AS <alias_name_t>]
USING <source_expression | table_name> [AS <alias_name_s>]
ON <boolean_expression1>
[ matchedClause [ matchedClause ] ]
[ notMatchedClause ]

matchedClause :

WHEN MATCHED [AND <boolean_expression>]
  THEN UPDATE SET <set_clause_list>
  | DELETE

notMatchedClause :

WHEN NOT MATCHED [AND <boolean_expression>]
  THEN INSERT VALUES <value_list>

Paramètres

Paramètres principaux

Paramètre	Obligatoire	Description
`target_table`	Oui	Nom d'une table cible existante
`alias_name_t`	Non	Alias de la table cible
`source_expression	table_name`	Oui	Table source, vue ou sous-requête à joindre avec la table cible
`alias_name_s`	Non	Alias de la table source, de la vue ou de la sous-requête
`boolean_expression1`	Oui	Condition de jointure qui renvoie une valeur booléenne (True ou False)

Paramètres de la clause WHEN

Paramètre	Obligatoire	Description
`boolean_expression` (matched)	Non	Filtre supplémentaire appliqué à la branche UPDATE ou DELETE
`boolean_expression` (not matched)	Non	Filtre supplémentaire appliqué à la branche INSERT
`set_clause_list`	Obligatoire pour UPDATE	Colonnes et valeurs à mettre à jour. Consultez la section « UPDATE » de UPDATE et DELETE
`value_list`	Obligatoire pour INSERT	Valeurs à insérer. Consultez VALUES

Notes d'utilisation pour les clauses WHEN :

Une instruction peut inclure au maximum une clause UPDATE, une clause DELETE et une clause INSERT.
Si les clauses UPDATE et DELETE sont toutes deux présentes, ajoutez une condition AND <boolean_expression> à la branche qui doit être évaluée en premier afin d'éviter toute ambiguïté dans la correspondance des lignes.
WHEN NOT MATCHED doit être la dernière clause WHEN et ne prend en charge que l'opération INSERT.

Exemples

Les exemples suivants couvrent les modèles les plus courants : upsert, fusion limitée à une partition et fusion de table Delta avec des opérations UPDATE et DELETE conditionnelles.

Exemple 1 : Upsert de lignes à l'aide d'une colonne de type d'événement

Cet exemple met à jour les lignes correspondantes et insère les lignes non correspondantes lorsque _event_type_ est égal à I (événement d'insertion). Ce modèle est typique pour l'application des événements CDC à partir d'une table intermédiaire.

-- Create a target transactional table partitioned by date and hour.
CREATE TABLE IF NOT EXISTS acid_address_book_base1
(id BIGINT, first_name STRING, last_name STRING, phone STRING)
PARTITIONED BY (year STRING, month STRING, day STRING, hour STRING)
tblproperties ("transactional"="true");

-- Create a source table with an event-type column.
CREATE TABLE IF NOT EXISTS tmp_table1
(id BIGINT, first_name STRING, last_name STRING, phone STRING, _event_type_ STRING);

-- Load test data into the target table.
INSERT OVERWRITE TABLE acid_address_book_base1
PARTITION(year='2020', month='08', day='20', hour='16')
VALUES (4, 'nihaho', 'li', '222'), (5, 'tahao', 'ha', '333'),
(7, 'djh', 'hahh', '555');

-- Verify the target table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM acid_address_book_base1;
-- Return results
+------------+------------+------------+------------+------------+------------+------------+------------+
| id         | first_name | last_name  | phone      | year       | month      | day        | hour       |
+------------+------------+------------+------------+------------+------------+------------+------------+
| 4          | nihaho     | li         | 222        | 2020       | 08         | 20         | 16         |
| 5          | tahao      | ha         | 333        | 2020       | 08         | 20         | 16         |
| 7          | djh        | hahh       | 555        | 2020       | 08         | 20         | 16         |
+------------+------------+------------+------------+------------+------------+------------+------------+

-- Load test data into the source table.
INSERT OVERWRITE TABLE tmp_table1 VALUES
(1, 'hh', 'liu', '999', 'I'), (2, 'cc', 'zhang', '888', 'I'),
(3, 'cy', 'zhang', '666', 'I'), (4, 'hh', 'liu', '999', 'U'),
(5, 'cc', 'zhang', '888', 'U'), (6, 'cy', 'zhang', '666', 'U');

-- Verify the source table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM tmp_table1;
-- Return results
+------------+------------+------------+------------+--------------+
| id         | first_name | last_name  | phone      | _event_type_ |
+------------+------------+------------+------------+--------------+
| 1          | hh         | liu        | 999        | I            |
| 2          | cc         | zhang      | 888        | I            |
| 3          | cy         | zhang      | 666        | I            |
| 4          | hh         | liu        | 999        | U            |
| 5          | cc         | zhang      | 888        | U            |
| 6          | cy         | zhang      | 666        | U            |
+------------+------------+------------+------------+--------------+

-- Run MERGE INTO: update matched rows, insert unmatched rows with event type 'I'.
MERGE INTO acid_address_book_base1 AS t USING tmp_table1 AS s
ON s.id = t.id AND t.year='2020' AND t.month='08' AND t.day='20' AND t.hour='16'
WHEN MATCHED THEN UPDATE SET t.first_name = s.first_name, t.last_name = s.last_name, t.phone = s.phone
WHEN NOT MATCHED AND (s._event_type_='I') THEN INSERT VALUES(s.id, s.first_name, s.last_name, s.phone, '2020', '08', '20', '16');

-- Verify the result.
SET odps.sql.allow.fullscan=true;
SELECT * FROM acid_address_book_base1;
-- Return results
+------------+------------+------------+------------+------------+------------+------------+------------+
| id         | first_name | last_name  | phone      | year       | month      | day        | hour       |
+------------+------------+------------+------------+------------+------------+------------+------------+
| 4          | hh         | liu        | 999        | 2020       | 08         | 20         | 16         |
| 5          | cc         | zhang      | 888        | 2020       | 08         | 20         | 16         |
| 7          | djh        | hahh       | 555        | 2020       | 08         | 20         | 16         |
| 1          | hh         | liu        | 999        | 2020       | 08         | 20         | 16         |
| 2          | cc         | zhang      | 888        | 2020       | 08         | 20         | 16         |
| 3          | cy         | zhang      | 666        | 2020       | 08         | 20         | 16         |
+------------+------------+------------+------------+------------+------------+------------+------------+

Remarque

La condition WHEN NOT MATCHED AND (s._event_type_='I') empêche l'insertion des événements de type mise à jour (U). Les lignes dont _event_type_='U' et qui n'ont aucune ligne cible correspondante sont ignorées silencieusement. Si votre flux CDC contient des types d'événements autres que I et U, ajoutez des filtres explicites à chaque branche pour les gérer correctement.

Exemple 2 : Fusion sur toutes les partitions

Sans contraintes de partition dans la clause ON, MERGE INTO s'applique à toutes les partitions de la table cible.

-- Create the target table.
CREATE TABLE IF NOT EXISTS merge_acid_dp(c1 BIGINT NOT NULL, c2 BIGINT NOT NULL)
PARTITIONED BY (dd STRING, hh STRING) tblproperties ("transactional" = "true");

-- Create the source table.
CREATE TABLE IF NOT EXISTS merge_acid_source(c1 BIGINT NOT NULL, c2 BIGINT NOT NULL,
  c3 STRING, c4 STRING) lifecycle 30;

-- Load test data into the target table.
INSERT OVERWRITE TABLE merge_acid_dp PARTITION (dd='01', hh='01')
VALUES (1, 1), (2, 2);
INSERT OVERWRITE TABLE merge_acid_dp PARTITION (dd='02', hh='02')
VALUES (4, 1), (3, 2);

-- Verify the target table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM merge_acid_dp;
-- Return results
+------------+------------+----+----+
| c1         | c2         | dd | hh |
+------------+------------+----+----+
| 1          | 1          | 01 | 01 |
| 2          | 2          | 01 | 01 |
| 4          | 1          | 02 | 02 |
| 3          | 2          | 02 | 02 |
+------------+------------+----+----+

-- Load test data into the source table.
INSERT OVERWRITE TABLE merge_acid_source VALUES(8, 2, '03', '03'),
(5, 5, '05', '05'), (6, 6, '02', '02');

-- Verify the source table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM merge_acid_source;
-- Return results
+------------+------------+----+----+
| c1         | c2         | c3 | c4 |
+------------+------------+----+----+
| 8          | 2          | 03 | 03 |
| 5          | 5          | 05 | 05 |
| 6          | 6          | 02 | 02 |
+------------+------------+----+----+

-- Run MERGE INTO across all partitions.
SET odps.sql.allow.fullscan=true;
MERGE INTO merge_acid_dp tar USING merge_acid_source src
ON tar.c2 = src.c2
WHEN MATCHED THEN
UPDATE SET tar.c1 = src.c1
WHEN NOT MATCHED THEN
INSERT VALUES(src.c1, src.c2, src.c3, src.c4);

-- Verify the result.
SET odps.sql.allow.fullscan=true;
SELECT * FROM merge_acid_dp;
-- Return results
+------------+------------+----+----+
| c1         | c2         | dd | hh |
+------------+------------+----+----+
| 6          | 6          | 02 | 02 |
| 5          | 5          | 05 | 05 |
| 8          | 2          | 02 | 02 |
| 8          | 2          | 01 | 01 |
| 1          | 1          | 01 | 01 |
| 4          | 1          | 02 | 02 |
+------------+------------+----+----+

Remarque

La clause ON effectue la jointure uniquement sur c2, sans filtre de partition. Cela déclenche une analyse complète de la table sur les deux partitions. Pour les tables volumineuses, ajoutez les colonnes de partition à la clause ON (voir l'exemple 3) pour limiter la portée de l'analyse et améliorer les performances.

Exemple 3 : Fusion dans une partition spécifique

Ajoutez les colonnes de partition à la clause ON pour limiter la fusion à une partition spécifique. Les lignes en dehors de la partition spécifiée ne sont pas affectées.

-- Create the target table.
CREATE TABLE IF NOT EXISTS merge_acid_sp(c1 BIGINT NOT NULL, c2 BIGINT NOT NULL)
PARTITIONED BY (dd STRING, hh STRING) tblproperties ("transactional" = "true");

-- Create the source table.
CREATE TABLE IF NOT EXISTS merge_acid_source(c1 BIGINT NOT NULL, c2 BIGINT NOT NULL,
  c3 STRING, c4 STRING) lifecycle 30;

-- Load test data into the target table.
INSERT OVERWRITE TABLE merge_acid_sp PARTITION (dd='01', hh='01')
VALUES (1, 1), (2, 2);
INSERT OVERWRITE TABLE merge_acid_sp PARTITION (dd='02', hh='02')
VALUES (4, 1), (3, 2);

-- Verify the target table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM merge_acid_sp;
-- Return results
+------------+------------+----+----+
| c1         | c2         | dd | hh |
+------------+------------+----+----+
| 1          | 1          | 01 | 01 |
| 2          | 2          | 01 | 01 |
| 4          | 1          | 02 | 02 |
| 3          | 2          | 02 | 02 |
+------------+------------+----+----+

-- Load test data into the source table.
INSERT OVERWRITE TABLE merge_acid_source VALUES(8, 2, '03', '03'),
(5, 5, '05', '05'), (6, 6, '02', '02');

-- Verify the source table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM merge_acid_source;
-- Return results
+------------+------------+----+----+
| c1         | c2         | c3 | c4 |
+------------+------------+----+----+
| 8          | 2          | 03 | 03 |
| 5          | 5          | 05 | 05 |
| 6          | 6          | 02 | 02 |
+------------+------------+----+----+

-- Run MERGE INTO scoped to partition dd='01', hh='01'.
SET odps.sql.allow.fullscan=true;
MERGE INTO merge_acid_sp tar USING merge_acid_source src
ON tar.c2 = src.c2 AND tar.dd = '01' AND tar.hh = '01'
WHEN MATCHED THEN
UPDATE SET tar.c1 = src.c1
WHEN NOT MATCHED THEN
INSERT VALUES(src.c1, src.c2, src.c3, src.c4);

-- Verify the result.
SET odps.sql.allow.fullscan=true;
SELECT * FROM merge_acid_sp;
+------------+------------+----+----+
| c1         | c2         | dd | hh |
+------------+------------+----+----+
| 5          | 5          | 05 | 05 |
| 6          | 6          | 02 | 02 |
| 8          | 2          | 01 | 01 |
| 1          | 1          | 01 | 01 |
| 4          | 1          | 02 | 02 |
| 3          | 2          | 02 | 02 |
+------------+------------+----+----+

Remarque

Le filtre de partition dans la clause ON (tar.dd = '01' AND tar.hh = '01') limite la portée de la correspondance : les lignes de la partition dd='02', hh='02' ne sont pas mises en correspondance. Toutefois, les lignes source non correspondantes (celles qui n'ont aucun équivalent dans dd='01', hh='01') sont toujours insérées, et leur partition de destination est déterminée par les valeurs de value_list. Dans cet exemple, les nouvelles lignes atterrissent dans de nouvelles partitions (05/05 et 02/02) en fonction des valeurs source.

Exemple 4 : Opérations UPDATE, DELETE et INSERT conditionnelles sur une table Delta

Cet exemple utilise une table Delta comme cible et applique les trois opérations en une seule instruction. Une condition supplémentaire sur la branche de correspondance détermine si une ligne correspondante est mise à jour ou supprimée.

-- Create a Delta table with a primary key.
CREATE TABLE IF NOT EXISTS mf_tt6 (pk BIGINT NOT NULL PRIMARY KEY,
                  val BIGINT NOT NULL)
                  PARTITIONED BY (dd STRING, hh STRING)
                  tblproperties ("transactional"="true");

-- Load test data into the target table.
INSERT OVERWRITE TABLE mf_tt6 PARTITION (dd='01', hh='02') VALUES (1, 1), (2, 2), (3, 3);
INSERT OVERWRITE TABLE mf_tt6 PARTITION (dd='01', hh='01') VALUES (1, 10), (2, 20), (3, 30);

-- Verify the target table.
SET odps.sql.allow.fullscan=true;
SELECT * FROM mf_tt6;
-- Return results
+------------+------------+----+----+
| pk         | val        | dd | hh |
+------------+------------+----+----+
| 1          | 10         | 01 | 01 |
| 3          | 30         | 01 | 01 |
| 2          | 20         | 01 | 01 |
| 1          | 1          | 01 | 02 |
| 3          | 3          | 01 | 02 |
| 2          | 2          | 01 | 02 |
+------------+------------+----+----+

-- Create the source table.
CREATE TABLE IF NOT EXISTS mf_delta AS SELECT pk, val FROM VALUES (1, 10), (2, 20), (6, 60) t (pk, val);

-- Verify the source table.
SELECT * FROM mf_delta;
-- Return results
+------+------+
| pk   | val  |
+------+------+
| 1    | 10   |
| 2    | 20   |
| 6    | 60   |
+------+------+

-- Run MERGE INTO scoped to partition dd='01', hh='02'.
-- Matched rows where pk > 1 are updated; matched rows where pk = 1 are deleted.
-- Unmatched rows are inserted into the partition.
MERGE INTO mf_tt6 USING mf_delta
ON mf_tt6.pk = mf_delta.pk AND mf_tt6.dd='01' AND mf_tt6.hh='02'
WHEN MATCHED AND (mf_tt6.pk > 1) THEN
UPDATE SET mf_tt6.val = mf_delta.val
WHEN MATCHED THEN DELETE
WHEN NOT MATCHED THEN
INSERT VALUES (mf_delta.pk, mf_delta.val, '01', '02');

-- Verify the result.
SET odps.sql.allow.fullscan=true;
SELECT * FROM mf_tt6;
-- Return results
+------------+------------+----+----+
| pk         | val        | dd | hh |
+------------+------------+----+----+
| 1          | 10         | 01 | 01 |
| 3          | 30         | 01 | 01 |
| 2          | 20         | 01 | 01 |
| 3          | 3          | 01 | 02 |
| 6          | 60         | 01 | 02 |
| 2          | 20         | 01 | 02 |
+------------+------------+----+----+

Remarque

Lorsque deux clauses WHEN MATCHED sont présentes, elles sont évaluées dans l'ordre. La première clause correspondante l'emporte. Dans cet exemple, les lignes avec pk > 1 correspondent à la première clause et sont mises à jour ; la ligne correspondante restante (pk = 1) passe à la deuxième clause et est supprimée. Placez la condition la plus spécifique en premier.