CREATE STREAM¶

Syntaxe¶

La syntaxe de la commande diffère selon l’objet sur lequel le flux est créé :

-- table
CREATE [ OR REPLACE ] STREAM [IF NOT EXISTS]
  <name>
  [ COPY GRANTS ]
  ON TABLE <table_name>
  [ { AT | BEFORE } ( { TIMESTAMP => <timestamp> | OFFSET => <time_difference> | STATEMENT => <id> | STREAM => '<name>' } ) ]
  [ APPEND_ONLY = TRUE | FALSE ]
  [ SHOW_INITIAL_ROWS = TRUE | FALSE ]
  [ COMMENT = '<string_literal>' ]

-- External table
CREATE [ OR REPLACE ] STREAM [IF NOT EXISTS]
  <name>
  [ COPY GRANTS ]
  ON EXTERNAL TABLE <external_table_name>
  [ { AT | BEFORE } ( { TIMESTAMP => <timestamp> | OFFSET => <time_difference> | STATEMENT => <id> | STREAM => '<name>' } ) ]
  [ INSERT_ONLY = TRUE ]
  [ COMMENT = '<string_literal>' ]

-- Directory table
CREATE [ OR REPLACE ] STREAM [IF NOT EXISTS]
  <name>
  [ COPY GRANTS ]
  ON STAGE <stage_name>
  [ COMMENT = '<string_literal>' ]

-- View
CREATE [ OR REPLACE ] STREAM [IF NOT EXISTS]
  <name>
  [ COPY GRANTS ]
  ON VIEW <view_name>
  [ { AT | BEFORE } ( { TIMESTAMP => <timestamp> | OFFSET => <time_difference> | STATEMENT => <id> | STREAM => '<name>' } ) ]
  [ APPEND_ONLY = TRUE | FALSE ]
  [ SHOW_INITIAL_ROWS = TRUE | FALSE ]
  [ COMMENT = '<string_literal>' ]

Syntaxe des variantes¶

CREATE STREAM … CLONE

Crée un nouveau flux avec la même définition que le flux source. Le clone hérite du décalage actuel (c’est-à-dire de la version de table transactionnelle actuelle) du flux source.

CREATE [ OR REPLACE ] STREAM <name> CLONE <source_stream>
  [ COPY GRANTS ]
  [ ... ]

Copy

Pour plus d’informations sur le clonage, voir CREATE <objet> … CLONE.

Paramètres requis¶

name

Chaîne qui indique l’identificateur (c’est-à-dire le nom) du flux ; doit être unique pour le schéma dans lequel le flux est créé.

De plus, l’identificateur doit commencer par un caractère alphabétique et ne peut pas contenir d’espaces ou de caractères spéciaux à moins que toute la chaîne d’identificateur soit délimitée par des guillemets doubles (p. ex. "My object"). Les identificateurs entre guillemets doubles sont également sensibles à la casse.

Pour plus de détails, voir Exigences relatives à l’identificateur.

table_name

Chaîne spécifiant l’identificateur (nom) de la table dont les modifications sont suivies par le flux (c’est-à-dire la table source).

Contrôle d’accès:

Pour interroger un flux, un rôle doit disposer du privilège SELECT sur la table sous-jacente.

external_table_name

Chaîne spécifiant l’identificateur (c’est-à-dire le nom) de la table externe dont les modifications sont suivies par le flux (c’est-à-dire la table externe source).

Contrôle d’accès:

Pour interroger un flux, un rôle doit disposer du privilège SELECT sur la table externe sous-jacente.

stage_name

Chaîne spécifiant l’identificateur (nom) de la zone de préparation dont les modifications de table de répertoire sont suivies par le flux (c’est-à-dire la table du répertoire source).

Contrôle d’accès:

Pour interroger un flux, un rôle doit disposer du privilège USAGE (zone de préparation externe) ou READ (zone de préparation interne) sur la zone de préparation sous-jacente.

view_name

Chaîne spécifiant l’identificateur (c’est-à-dire le nom) de la vue source. Le flux suit les modifications DML apportées aux tables sous-jacentes de la vue.

Pour plus d’informations sur les flux dans les vues, voir Streams on Views.

Contrôle d’accès:

Pour interroger un flux, un rôle doit disposer du privilège SELECT sur la vue.

Paramètres facultatifs¶

COPY GRANTS

Spécifie de conserver les droits d’accès du flux d’origine lorsqu’un nouveau flux est créé à l’aide de l’une des variables CREATE STREAM suivantes :

• CREATE OR REPLACE STREAM

• CREATE STREAM … CLONE

Ce paramètre copie toutes les autorisations, excepté OWNERSHIP, du flux existant vers le nouveau flux. Par défaut, le rôle qui exécute la commande CREATE STREAM possède le nouveau flux.

Note

• Si l’instruction CREATE STREAM fait référence à plusieurs flux (p. ex. create or replace stream t1 clone t2;), la clause COPY GRANTS donne priorité au flux à remplacer.

• La sortie SHOW GRANTS pour le flux de remplacement liste le concessionnaire des privilèges copiés comme le rôle qui a exécuté l’instruction CREATE STREAM, avec l’horodatage courant lorsque l’instruction a été exécutée.

• L’opération de copie des accords s’effectue atomiquement dans la commande CREATE STREAM (c’est-à-dire dans la même transaction).

Note

Ce paramètre n’est pas pris en charge actuellement.

AT ( { TIMESTAMP => <timestamp> | OFFSET => <time_difference> | STATEMENT => <id> | STREAM => '<name>' ) | BEFORE ( TIMESTAMP => <timestamp> | OFFSET => <time_difference> | STATEMENT => <id> ) }

Crée un flux à un moment précis dans le passé (à l’aide de la fonction Time Travel). La clause AT | BEFORE détermine le point du passé à partir duquel des données historiques sont demandées :

• Le mot clé AT spécifie que la requête inclut tous les changements apportés par une instruction ou une transaction dont l’horodatage est égal au paramètre spécifié.

  La valeur STREAM => '<nom>' est spéciale. Lorsqu’elle est fournie, l’instruction CREATE STREAM crée le nouveau flux au même décalage que le flux spécifié. Vous pouvez également fournir cette valeur lors de la recréation d’un flux existant (en utilisant les mots-clés OR REPLACE) afin de conserver le décalage actuel du flux après sa recréation. '<nom>' est l’identificateur (c’est-à-dire le nom) du flux existant dont le décalage est copié vers le nouveau flux ou le flux recréé.

  Le flux nouveau ou recréé avance le décalage, comme d’habitude, lorsque le flux est utilisé dans une transaction DML.

• Le mot clé BEFORE spécifie que la requête se réfère à un point précédant immédiatement le paramètre spécifié.

Note

Si aucune donnée de suivi des modifications n’est disponible sur l’objet source à l’endroit spécifié dans le passé dans la clause AT | BEFORE, l’instruction CREATE STREAM échoue. Aucun flux ne peut être créé à un moment antérieur à l’enregistrement du suivi des modifications.

APPEND_ONLY = TRUE | FALSE

Uniquement pris en charge pour les flux sur les tables standard ou les flux sur les vues qui interrogent les tables standard.

Spécifie s’il s’agit d’un flux d’ajout uniquement. Les flux d’ajout uniquement suivent uniquement les insertions de ligne. Les opérations de mise à jour et de suppression (y compris les troncations de table) ne sont pas enregistrées. Par exemple, si 10 lignes sont insérées dans une table et que 5 de ces lignes sont supprimées avant que le décalage pour un flux d’ajout uniquement soit avancé, le flux enregistre 10 lignes.

Ce type de flux améliore les performances des requêtes par rapport aux flux standard et est très utile pour l’extraction, le chargement et la transformation (ELT), et pour des scénarios similaires qui dépendent exclusivement des insertions de ligne.

Un flux standard joint les lignes supprimées et insérées dans l’ensemble de modifications pour déterminer quelles lignes ont été supprimées et lesquelles ont été mises à jour. Un flux d’ajout uniquement renvoie les lignes ajoutées uniquement et peut donc être beaucoup plus performant qu’un flux standard. Par exemple, la table source peut être tronquée immédiatement après la consommation des lignes d’un flux d’ajout uniquement, et les suppressions d’enregistrement ne contribuent pas à la surcharge la prochaine fois que le flux est interrogé ou consommé.

Par défaut:

FALSE

INSERT_ONLY = TRUE | FALSE

Requis pour les flux sur des tables externes. Non pris en charge par les flux sur d’autres objets.

Spécifie s’il s’agit d’un flux à insertion uniquement. Les flux à insertion uniquement suivent uniquement les insertions de lignes ; ils n’enregistrent pas les opérations de suppression qui suppriment des lignes d’un ensemble inséré (c’est-à-dire sans opération). Par exemple, entre deux décalages, si File1 est supprimé de l’emplacement de stockage dans le Cloud référencé par la table externe et que File2 est ajouté, le flux renvoie des enregistrements pour les lignes de File2 uniquement. Contrairement au suivi des données de capture de données modifiées (CDC) pour les tables standard, Snowflake ne peut pas accéder aux enregistrements historiques des fichiers stockés dans le Cloud.

Par défaut:

FALSE

SHOW_INITIAL_ROWS = TRUE | FALSE

Spécifie les enregistrements à renvoyer la première fois que le flux est consommé.

TRUE

Le flux renvoie seulement les lignes qui existaient dans l’objet source au moment de la création du flux. La colonne METADATA$ISUPDATE indique une valeur FALSE dans ces lignes. Ensuite, le flux renvoie toutes les modifications DML apportées à l’objet source depuis le décalage le plus récent ; c’est le comportement normal du flux.

Ce paramètre permet d’initialiser tout processus en aval avec le contenu de l’objet source du flux.

FALSE

Le flux renvoie toutes les modifications DML apportées à l’objet source depuis le décalage le plus récent.

Par défaut:

FALSE

COMMENT = 'string_literal'

Chaîne (littéral) qui spécifie un commentaire pour le flux.

Par défaut : aucune valeur

Sortie¶

La sortie d’un flux comprend les mêmes colonnes que l’objet source, ainsi que les colonnes supplémentaires suivantes :

• METADATA$ACTION : spécifie l’action (INSERT ou DELETE).

• METADATA$ISUPDATE : spécifie si l’action enregistrée (INSERT ou DELETE) fait partie d’un UPDATE appliqué aux lignes de la table ou de la vue source.

  Notez que les flux enregistrent les différences entre deux décalages. Si une ligne est ajoutée puis mise à jour dans le décalage actuel, la modification delta est une nouvelle ligne. La ligne METADATA$ISUPDATE enregistre une valeur FALSE.

• METADATA$ROW_ID : spécifie l’ID unique et immuable pour la ligne, qui peut être utilisé pour suivre les modifications apportées à des lignes spécifiques au fil du temps.

Exigences en matière de contrôle d’accès¶

Un rôle utilisé pour exécuter cette commande SQL doit avoir les privilèges suivants définis au minimum ainsi :

Flux sur les tables standards :

Objet	Privilège	Notes
Schéma	CREATE STREAM
Table	SELECT	Si le suivi des modifications n’a pas été activé sur la table source (en utilisant ALTER TABLE … SET CHANGE_TRACKING = TRUE), seul le propriétaire de la table (c’est-à-dire le rôle qui possède le privilège OWNERSHIP sur la table) peut créer le flux initial sur la table. La création du flux initial active automatiquement le suivi des modifications sur la table.

Notez que l’exploitation d’un objet dans un schéma requiert également le privilège USAGE sur la base de données et le schéma parents.

Flux sur les vues :

Objet	Privilège	Notes
Schéma	CREATE STREAM
Vue	SELECT	Si le suivi des modifications n’a pas été activé sur la vue source et ses tables sous-jacentes, seuls un rôle disposant du privilège OWNERSHIP sur la vue et le propriétaire de ses tables sous-jacentes peuvent créer le flux initial sur la vue. La création du flux initial active automatiquement le suivi des modifications sur la table. Pour des instructions sur l’activation du suivi des modifications sur une vue et ses tables sous-jacentes, voir Activation du suivi des modifications sur les vues et les tables sous-jacentes. Notez que l’activation du suivi des modifications verrouille les tables sous-jacentes pendant l’activation du suivi des modifications. Les verrous sur les objets sous-jacents peuvent entraîner une latence dans les opérations DDL/DML avec ces objets. Pour plus d’informations, reportez-vous à Verrouillage des ressources.

Notez que l’exploitation d’un objet dans un schéma requiert également le privilège USAGE sur la base de données et le schéma parents.

Flux sur des tables de répertoire :

Objet	Privilège	Notes
Schéma	CREATE STREAM
Zone de préparation	USAGE (zone de préparation externe) ou READ (zone de préparation interne)

Notez que l’exploitation d’un objet dans un schéma requiert également le privilège USAGE sur la base de données et le schéma parents.

Flux sur des tables externes :

Objet	Privilège	Notes
Schéma	CREATE STREAM
Table externe	SELECT

Notez que l’exploitation d’un objet dans un schéma requiert également le privilège USAGE sur la base de données et le schéma parents.

Pour obtenir des instructions sur la création d’un rôle personnalisé avec un ensemble spécifique de privilèges, voir Création de rôles personnalisés.

Pour des informations générales sur les rôles et les privilèges accordés pour effectuer des actions SQL sur des objets sécurisables, voir Aperçu du contrôle d’accès.

Notes sur l’utilisation¶

• Un flux peut être interrogé plusieurs fois pour mettre à jour plusieurs objets dans la même transaction et il renverra les mêmes données.

• La position du flux (c’est-à-dire le décalage) est avancée lorsque le flux est utilisé dans une instruction DML. La position est mise à jour à la fin de la transaction avec l’horodatage de début de la transaction. Le flux décrit les enregistrements de modification commençant à la position actuelle du flux et se terminant à l’horodatage transactionnel actuel.

  Pour vous assurer que plusieurs instructions ont accès aux mêmes enregistrements de modification dans le flux, entourez-les d’une instruction de transaction explicite (BEGIN .. COMMIT). Une transaction explicite verrouille le flux, de sorte que les mises à jour DML de l’objet source ne soient pas signalées au flux tant que la transaction n’est pas validée.

• Les flux n’ont pas de période de conservation Fail-safe ou Time Travel. Les métadonnées de ces objets ne peuvent pas être récupérées si un flux est détruit.

• Flux sur les tables partagées :

  • La période de conservation d’une table source n’est pas prolongée automatiquement pour éviter que les flux de la table ne deviennent périmés.

• Les flux standards ne peuvent pas récupérer les données de changement pour les données géospatiales. Nous recommandons de créer des flux d’ajout uniquement sur les objets qui contiennent des données géospatiales.

• Flux sur les vues :

  • La création du premier flux sur une vue en utilisant le rôle de propriétaire de la vue (c’est-à-dire le rôle avec le privilège OWNERSHIP sur la vue) permet le suivi des modifications sur la vue. Si le même rôle est également propriétaire des tables sous-jacentes, le suivi des modifications est également activé sur les tables. Si le rôle n’a pas reçu le privilège OWNERSHIP à la fois sur la vue et sur ses tables sous-jacentes, le suivi des modifications doit être activé manuellement sur les objets concernés. Pour obtenir des instructions, voir Activation du suivi des modifications sur les vues et les tables sous-jacentes.

  • Selon le nombre de jointures dans une vue, une seule modification des tables sous-jacentes peut entraîner un grand nombre de modifications dans la sortie du flux.

  • Tout flux sur une vue donnée est interrompu si la vue source ou les tables sous-jacentes sont détruites ou recréées (en utilisant CREATE OR REPLACE VIEW).

  • Tous les flux sur une vue sécurisée adhèrent aux contraintes de la vue sécurisée.

    Si le propriétaire d’une vue non sécurisée (c’est-à-dire le rôle avec le privilège OWNERSHIP sur la vue) la change en vue sécurisée (en utilisant ALTER VIEW … SET SECURE), tout flux sur la vue applique automatiquement les contraintes de vue sécurisée.

    En outre, la période de conservation des tables sous-jacentes n’est pas prolongée automatiquement pour éviter que les flux de la vue sécurisée ne deviennent périmés.

  • Les flux basés sur des vues où la vue utilise des fonctions non déterministes peuvent renvoyer des résultats non déterministes.

    Par exemple, les résultats des fonctions contexte telles que CURRENT_DATE, et CURRENT_USER sont non déterministes. Les résultats des fonctions de génération de données telles que RANDOM sont également non déterministes. Si une vue contient une fonction non déterministe, tout flux sur cette vue ne sera pas un instantané constant de la sortie de la fonction. Au lieu de cela, la valeur dans le flux peut changer lorsqu’elle est interrogée.

    Nous vous recommandons de vous assurer que le non-déterminisme des résultats d’une vue n’affecte pas l’exactitude des résultats de la requête de flux.

    Pour un exemple, voir Flux sur une vue qui appelle une fonction SQL non-déterministe.

• Flux sur les tables du répertoire : les valeurs des colonnes METADATA$ROW_ID dans la sortie du flux sont vides.

• Concernant les métadonnées :

  Attention

  Les clients doivent s’assurer qu’aucune donnée personnelle (autre que pour un objet utilisateur), donnée sensible, donnée à exportation contrôlée ou autre donnée réglementée n’est saisie comme métadonnée lors de l’utilisation du service Snowflake. Pour plus d’informations, voir Champs de métadonnées dans Snowflake.

• Les instructions CREATE OR REPLACE <objet> sont atomiques. En d’autres termes, lorsqu’un objet est remplacé, l’ancien objet est supprimé et le nouvel objet est créé dans une seule transaction.

Exemples¶

CREATE STREAM mystream ON TABLE mytable;

CREATE STREAM mystream ON TABLE mytable BEFORE (TIMESTAMP => TO_TIMESTAMP(40*365*86400));

CREATE STREAM mystream ON TABLE mytable AT (TIMESTAMP => TO_TIMESTAMP_TZ('02/02/2019 01:02:03', 'mm/dd/yyyy hh24:mi:ss'));

CREATE STREAM mystream ON TABLE mytable AT(OFFSET => -60*5);

CREATE STREAM mystream ON TABLE mytable AT(STREAM => 'oldstream');

CREATE OR REPLACE STREAM mystream ON TABLE mytable AT(STREAM => 'mystream');

CREATE STREAM mystream ON TABLE mytable BEFORE(STATEMENT => '8e5d0ca9-005e-44e6-b858-a8f5b37c5726');

CREATE STREAM mystream ON VIEW myview;

-- Create an external table that points to the MY_EXT_STAGE stage.
-- The external table is partitioned by the date (in YYYY/MM/DD format) in the file path.
CREATE EXTERNAL TABLE my_ext_table (
  date_part date as to_date(substr(metadata$filename, 1, 10), 'YYYY/MM/DD'),
  ts timestamp AS (value:time::timestamp),
  user_id varchar AS (value:userId::varchar),
  color varchar AS (value:color::varchar)
) PARTITION BY (date_part)
  LOCATION=@my_ext_stage
  AUTO_REFRESH = false
  FILE_FORMAT=(TYPE=JSON);

-- Create a stream on the external table
CREATE STREAM my_ext_table_stream ON EXTERNAL TABLE my_ext_table INSERT_ONLY = TRUE;

-- Execute SHOW streams
-- The MODE column indicates that the new stream is an INSERT_ONLY stream
SHOW STREAMS;
+-------------------------------+------------------------+---------------+-------------+--------------+-----------+------------------------------------+-------+-------+-------------+
| created_on                    | name                   | database_name | schema_name | owner        | comment   | table_name                         | type  | stale | mode        |
|-------------------------------+------------------------+---------------+-------------+--------------+-----------+------------------------------------+-------+-------+-------------|
| 2020-08-02 05:13:20.174 -0800 | MY_EXT_TABLE_STREAM    | MYDB          | PUBLIC      | MYROLE       |           | MYDB.PUBLIC.EXTTABLE_S3_PART       | DELTA | false | INSERT_ONLY |
+-------------------------------+------------------------+---------------+-------------+--------------+-----------+------------------------------------+-------+-------+-------------+

-- Add a file named '2020/08/05/1408/log-08051409.json' to the stage using the appropriate tool for the cloud storage service.

-- Manually refresh the external table metadata.
ALTER EXTERNAL TABLE my_ext_table REFRESH;

-- Query the external table stream.
-- The stream indicates that the rows in the added JSON file were recorded in the external table metadata.
SELECT * FROM my_ext_table_stream;
+----------------------------------------+------------+-------------------------+---------+-------+-----------------+-------------------+-----------------+---------------------------------------------+
| VALUE                                  | DATE_PART  | TS                      | USER_ID | COLOR | METADATA$ACTION | METADATA$ISUPDATE | METADATA$ROW_ID | METADATA$FILENAME                           |
|----------------------------------------+------------+-------------------------+---------+-------+-----------------+-------------------+-----------------+---------------------------------------------|
| {                                      | 2020-08-05 | 2020-08-05 15:57:01.000 | user25  | green | INSERT          | False             |                 | test/logs/2020/08/05/1408/log-08051409.json |
|   "color": "green",                    |            |                         |         |       |                 |                   |                 |                                             |
|   "time": "2020-08-05 15:57:01-07:00", |            |                         |         |       |                 |                   |                 |                                             |
|   "userId": "user25"                   |            |                         |         |       |                 |                   |                 |                                             |
| }                                      |            |                         |         |       |                 |                   |                 |                                             |
| {                                      | 2020-08-05 | 2020-08-05 15:58:02.000 | user56  | brown | INSERT          | False             |                 | test/logs/2020/08/05/1408/log-08051409.json |
|   "color": "brown",                    |            |                         |         |       |                 |                   |                 |                                             |
|   "time": "2020-08-05 15:58:02-07:00", |            |                         |         |       |                 |                   |                 |                                             |
|   "userId": "user56"                   |            |                         |         |       |                 |                   |                 |                                             |
| }                                      |            |                         |         |       |                 |                   |                 |                                             |
+----------------------------------------+------------+-------------------------+---------+-------+-----------------+-------------------+-----------------+---------------------------------------------+

CREATE STREAM dirtable_mystage_s ON STAGE mystage;

ALTER STAGE mystage REFRESH;

SELECT * FROM dirtable_mystage_s;

+-------------------+--------+-------------------------------+----------------------------------+----------------------------------+-------------------------------------------------------------------------------------------+-----------------+-------------------+-----------------+
| RELATIVE_PATH     | SIZE   | LAST_MODIFIED                 | MD5                              | ETAG                             | FILE_URL                                                                                  | METADATA$ACTION | METADATA$ISUPDATE | METADATA$ROW_ID |
|-------------------+--------+-------------------------------+----------------------------------+----------------------------------+-------------------------------------------------------------------------------------------+-----------------+-------------------+-----------------|
| file1.csv.gz      |   1048 | 2021-05-14 06:09:08.000 -0700 | c98f600c492c39bef249e2fcc7a4b6fe | c98f600c492c39bef249e2fcc7a4b6fe | https://myaccount.snowflakecomputing.com/api/files/MYDB/MYSCHEMA/MYSTAGE/file1%2ecsv%2egz | INSERT          | False             |                 |
| file2.csv.gz      |   3495 | 2021-05-14 06:09:09.000 -0700 | 7f1a4f98ef4c7c42a2974504d11b0e20 | 7f1a4f98ef4c7c42a2974504d11b0e20 | https://myaccount.snowflakecomputing.com/api/files/MYDB/MYSCHEMA/MYSTAGE/file2%2ecsv%2egz | INSERT          | False             |                 |
+-------------------+--------+-------------------------------+----------------------------------+----------------------------------+-------------------------------------------------------------------------------------------+-----------------+-------------------+-----------------+