fix: snowpipe streaming users table skipping

rudderlabs · Nov 25, 2024 · 19988fc · 19988fc
1 parent ed9fe55
commit 19988fc
Show file tree

Hide file tree

Showing 23 changed files with 2,123 additions and 140 deletions.
diff --git a/src/v0/destinations/snowpipe_streaming/transform.js b/src/v0/destinations/snowpipe_streaming/transform.js
@@ -1 +1,28 @@
-module.exports = require('../snowflake/transform');
+const transform = require('../snowflake/transform');
+const { processWarehouseMessage } = require('../../../warehouse');
+
+const provider = 'snowpipe_streaming';
+
+function process(event) {
+  const whSchemaVersion = event.request.query.whSchemaVersion || 'v1';
+  const whIDResolve = event.request.query.whIDResolve === 'true' || false;
+  const whStoreEvent = event.destination.Config.storeFullEvent === true;
+  const destJsonPaths = event.destination?.Config?.jsonPaths || '';
+  return processWarehouseMessage(event.message, {
+    metadata: event.metadata,
+    whSchemaVersion,
+    whStoreEvent,
+    whIDResolve,
+    getDataTypeOverride: transform.getDataTypeOverride,
+    provider,
+    sourceCategory: event.metadata ? event.metadata.sourceCategory : null,
+    destJsonPaths,
+    destConfig: event.destination?.Config,
+  });
+}
+
+module.exports = {
+  provider,
+  process,
+  getDataTypeOverride: transform.getDataTypeOverride,
+};
diff --git a/src/warehouse/config/ReservedKeywords.json b/src/warehouse/config/ReservedKeywords.json
@@ -1519,6 +1519,99 @@
     "WHERE": true,
     "WITH": true
   },
+  "SNOWPIPE_STREAMING": {
+    "ACCOUNT": true,
+    "ALL": true,
+    "ALTER": true,
+    "AND": true,
+    "ANY": true,
+    "AS": true,
+    "BETWEEN": true,
+    "BY": true,
+    "CASE": true,
+    "CAST": true,
+    "CHECK": true,
+    "COLUMN": true,
+    "CONNECT": true,
+    "CONNECTION": true,
+    "CONSTRAINT": true,
+    "CREATE": true,
+    "CROSS": true,
+    "CURRENT": true,
+    "CURRENT_DATE": true,
+    "CURRENT_TIME": true,
+    "CURRENT_TIMESTAMP": true,
+    "CURRENT_USER": true,
+    "DATABASE": true,
+    "DELETE": true,
+    "DISTINCT": true,
+    "DROP": true,
+    "ELSE": true,
+    "EXISTS": true,
+    "FALSE": true,
+    "FOLLOWING": true,
+    "FOR": true,
+    "FROM": true,
+    "FULL": true,
+    "GRANT": true,
+    "GROUP": true,
+    "GSCLUSTER": true,
+    "HAVING": true,
+    "ILIKE": true,
+    "IN": true,
+    "INCREMENT": true,
+    "INNER": true,
+    "INSERT": true,
+    "INTERSECT": true,
+    "INTO": true,
+    "IS": true,
+    "ISSUE": true,
+    "JOIN": true,
+    "LATERAL": true,
+    "LEFT": true,
+    "LIKE": true,
+    "LOCALTIME": true,
+    "LOCALTIMESTAMP": true,
+    "MINUS": true,
+    "NATURAL": true,
+    "NOT": true,
+    "NULL": true,
+    "OF": true,
+    "ON": true,
+    "OR": true,
+    "ORDER": true,
+    "ORGANIZATION": true,
+    "QUALIFY": true,
+    "REGEXP": true,
+    "REVOKE": true,
+    "RIGHT": true,
+    "RLIKE": true,
+    "ROW": true,
+    "ROWS": true,
+    "SAMPLE": true,
+    "SCHEMA": true,
+    "SELECT": true,
+    "SET": true,
+    "SOME": true,
+    "START": true,
+    "TABLE": true,
+    "TABLESAMPLE": true,
+    "THEN": true,
+    "TO": true,
+    "TRIGGER": true,
+    "TRUE": true,
+    "TRY_CAST": true,
+    "UNION": true,
+    "UNIQUE": true,
+    "UPDATE": true,
+    "USING": true,
+    "VALUES": true,
+    "VIEW": true,
+    "WHEN": true,
+    "WHENEVER": true,
+    "WHERE": true,
+    "WITH": true
+  },
   "CLICKHOUSE": {},
   "S3_DATALAKE": {
     "ALL": true,

diff --git a/src/warehouse/identity.js b/src/warehouse/identity.js
@@ -2,7 +2,7 @@ const _ = require('lodash');
 const { InstrumentationError } = require('@rudderstack/integrations-lib');
 const { getVersionedUtils } = require('./util');
 
-const identityEnabledWarehouses = ['snowflake', 'bq'];
+const identityEnabledWarehouses = ['snowflake', 'snowpipe_streaming', 'bq'];
 const versionedMergePropColumns = {};
 const versionedMergeRuleTableNames = {};
 

diff --git a/src/warehouse/index.js b/src/warehouse/index.js
@@ -349,7 +349,10 @@ function stringLikeObjectToString(obj) {
  */
 function getColumns(options, event, columnTypes) {
   const columns = {};
-  const uuidTS = options.provider === 'snowflake' ? 'UUID_TS' : 'uuid_ts';
+  const uuidTS =
+    options.provider === 'snowflake' || options.provider === 'snowpipe_streaming'
+      ? 'UUID_TS'
+      : 'uuid_ts';
   columns[uuidTS] = 'datetime';
   // add loaded_at for bq to be segment compatible
   if (options.provider === 'bq') {
@@ -377,6 +380,7 @@ function getColumns(options, event, columnTypes) {
 
 const fullEventColumnTypeByProvider = {
   snowflake: 'json',
+  snowpipe_streaming: 'json',
   rs: 'text',
   bq: 'string',
   postgres: 'json',
@@ -613,6 +617,15 @@ function enhanceContextWithSourceDestInfo(message, metadata) {
   message.context = context;
 }
 
+function shouldSkipUsersTable(options) {
+  return (
+    options.provider === 'snowpipe_streaming' ||
+    options.destConfig?.skipUsersTable ||
+    options.integrationOptions?.skipUsersTable ||
+    false
+  );
+}
+
 function processWarehouseMessage(message, options) {
   const utils = getVersionedUtils(options.whSchemaVersion);
   options.utils = utils;
@@ -638,8 +651,7 @@ function processWarehouseMessage(message, options) {
   const eventType = message.type?.toLowerCase();
   const skipTracksTable =
     options.destConfig?.skipTracksTable || options.integrationOptions.skipTracksTable || false;
-  const skipUsersTable =
-    options.destConfig?.skipUsersTable || options.integrationOptions.skipUsersTable || false;
+  const skipUsersTable = shouldSkipUsersTable(options);
   const skipReservedKeywordsEscaping =
     options.integrationOptions.skipReservedKeywordsEscaping || false;
 

diff --git a/src/warehouse/v1/util.js b/src/warehouse/v1/util.js
@@ -11,7 +11,7 @@ function safeTableName(options, name = '') {
   if (tableName === '') {
     throw new TransformationError('Table name cannot be empty.');
   }
-  if (provider === 'snowflake') {
+  if (provider === 'snowflake' || provider === 'snowpipe_streaming') {
     tableName = tableName.toUpperCase();
   } else if (provider === 'postgres') {
     tableName = tableName.substr(0, 63);
@@ -41,7 +41,7 @@ function safeColumnName(options, name = '') {
   if (columnName === '') {
     throw new TransformationError('Column name cannot be empty.');
   }
-  if (provider === 'snowflake') {
+  if (provider === 'snowflake' || provider === 'snowpipe_streaming') {
     columnName = columnName.toUpperCase();
   } else if (provider === 'postgres') {
     columnName = columnName.substr(0, 63);