diff --git a/src/v0/destinations/snowpipe_streaming/transform.js b/src/v0/destinations/snowpipe_streaming/transform.js index 3b0a6e1a93..a71992949e 100644 --- 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 index 3c383d24bf..53b8738e1d 100644 --- 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 index c34a66a481..050b556a11 100644 --- 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 index 3491a257da..ea663c9b2f 100644 --- 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 index d1289bc674..95dd59b030 100644 --- 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); diff --git a/test/__tests__/data/warehouse/events.js b/test/__tests__/data/warehouse/events.js index bca6f776be..a839cf10e3 100644 --- a/test/__tests__/data/warehouse/events.js +++ b/test/__tests__/data/warehouse/events.js @@ -2193,6 +2193,13 @@ function output(eventType, provider) { if (provider === "snowflake") { return _.cloneDeep(sampleEvents[eventType].output.snowflake); } + if (provider === "snowpipe_streaming") { + if (eventType === 'identify') { + return _.cloneDeep([sampleEvents[eventType].output.snowflake[0]]); + } else { + return _.cloneDeep(sampleEvents[eventType].output.snowflake); + } + } if (provider === "s3_datalake") { return _.cloneDeep(sampleEvents[eventType].output.s3_datalake); } diff --git a/test/__tests__/data/warehouse/integration_options_events.js b/test/__tests__/data/warehouse/integration_options_events.js index 35eafb6a9b..d47efe1455 100644 --- a/test/__tests__/data/warehouse/integration_options_events.js +++ b/test/__tests__/data/warehouse/integration_options_events.js @@ -74,6 +74,13 @@ const sampleEvents = { jsonPaths: ["tMap"] } }, + SNOWPIPE_STREAMING: { + options: { + skipTracksTable: true, + useBlendoCasing: true, + jsonPaths: ["tMap"] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true @@ -1639,6 +1646,12 @@ function opOutput(eventType, provider) { switch (provider) { case "snowflake": return _.cloneDeep(sampleEvents[eventType].output.snowflake); + case "snowpipe_streaming": + if (eventType === 'users') { + return _.cloneDeep([sampleEvents[eventType].output.snowflake[0]]); + } else { + return _.cloneDeep(sampleEvents[eventType].output.snowflake); + } case "s3_datalake": return _.cloneDeep(sampleEvents[eventType].output.s3_datalake); case "rs": diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/aliases.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/aliases.js index 30cce51fb7..a57eb34179 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/aliases.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/aliases.js @@ -78,6 +78,14 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "testMap.nestedMap", + "ctestMap.cnestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/extract.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/extract.js index 9d38ecc292..bf863a1500 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/extract.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/extract.js @@ -59,6 +59,14 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "PMap.nestedMap", + "CMap.nestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/groups.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/groups.js index bbae993b27..6b4f97ca97 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/groups.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/groups.js @@ -78,6 +78,14 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "testMap.nestedMap", + "ctestMap.cnestedMap", + ] + } + }, GCS_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/identifies.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/identifies.js index 42f5cccf49..60853dbe10 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/identifies.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/identifies.js @@ -53,6 +53,16 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "UPMap.nestedMap", + "CTMap.nestedMap", + "TMap.nestedMap", + "CMap.nestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/pages.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/pages.js index dc2d3a3318..6e63293157 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/pages.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/pages.js @@ -67,6 +67,14 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "testMap.nestedMap", + "ctestMap.cnestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/screens.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/screens.js index 9b789d2747..12ecc2c820 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/screens.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/screens.js @@ -67,6 +67,14 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "testMap.nestedMap", + ".ctestMap.cnestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/tracks.js b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/tracks.js index 691ce57ca1..a62c4c274b 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/tracks.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/legacy/tracks.js @@ -78,6 +78,15 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "UPMap.nestedMap", + "PMap.nestedMap", + "CMap.nestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/aliases.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/aliases.js index d5d57f85b9..d2b3587a13 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/aliases.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/aliases.js @@ -66,6 +66,11 @@ module.exports = { jsonPaths: ["alias.traits.testMap.nestedMap", "alias.context.ctestMap.cnestedMap"] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: ["alias.traits.testMap.nestedMap", "alias.context.ctestMap.cnestedMap"] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/extract.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/extract.js index a950b7f526..12ddb2c491 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/extract.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/extract.js @@ -59,6 +59,14 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "extract.properties.PMap.nestedMap", + "extract.context.CMap.nestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/groups.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/groups.js index 6979aa1100..2dd867e330 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/groups.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/groups.js @@ -66,6 +66,11 @@ module.exports = { jsonPaths: ["group.traits.testMap.nestedMap", "group.context.ctestMap.cnestedMap"] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: ["group.traits.testMap.nestedMap", "group.context.ctestMap.cnestedMap"] + } + }, GCS_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/identifies.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/identifies.js index 89fcc23cd5..9ec648a921 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/identifies.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/identifies.js @@ -53,6 +53,16 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "identify.userProperties.UPMap.nestedMap", + "identify.context.traits.CTMap.nestedMap", + "identify.traits.TMap.nestedMap", + "identify.context.CMap.nestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/pages.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/pages.js index d5282c1cfd..c74cfac31f 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/pages.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/pages.js @@ -55,6 +55,11 @@ module.exports = { jsonPaths: ["page.properties.testMap.nestedMap", "page.context.ctestMap.cnestedMap"] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: ["page.properties.testMap.nestedMap", "page.context.ctestMap.cnestedMap"] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/screens.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/screens.js index 3eec47b89d..699ac3e812 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/screens.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/screens.js @@ -55,6 +55,11 @@ module.exports = { jsonPaths: ["screen.properties.testMap.nestedMap", "screen.context.ctestMap.cnestedMap"] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: ["screen.properties.testMap.nestedMap", "screen.context.ctestMap.cnestedMap"] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/data/warehouse/integrations/jsonpaths/new/tracks.js b/test/__tests__/data/warehouse/integrations/jsonpaths/new/tracks.js index 6b411835db..09d058b058 100644 --- a/test/__tests__/data/warehouse/integrations/jsonpaths/new/tracks.js +++ b/test/__tests__/data/warehouse/integrations/jsonpaths/new/tracks.js @@ -78,6 +78,15 @@ module.exports = { ] } }, + SNOWPIPE_STREAMING: { + options: { + jsonPaths: [ + "track.userProperties.UPMap.nestedMap", + "track.properties.PMap.nestedMap", + "track.context.CMap.nestedMap", + ] + } + }, S3_DATALAKE: { options: { skipReservedKeywordsEscaping: true diff --git a/test/__tests__/warehouse.test.js b/test/__tests__/warehouse.test.js index 7fdecbd7cf..4d564199e5 100644 --- a/test/__tests__/warehouse.test.js +++ b/test/__tests__/warehouse.test.js @@ -39,11 +39,7 @@ const integrations = [ ]; const integration = (index ) => { - const it = integrations[index]; - if (it === "snowflake" || it === "snowpipe_streaming") { - return "snowflake"; - } - return it; + return integrations[index]; } const transformers = integrations.map(integration => require(`../../src/${version}/destinations/${integration}/transform`) @@ -62,7 +58,7 @@ const propsKeyMap = { }; const integrationCasedString = (integration, str) => { - if (integration === "snowflake") { + if (integration === "snowflake" || integration === "snowpipe_streaming") { return str.toUpperCase(); } return str; @@ -219,7 +215,7 @@ describe("column & table names", () => { //KEY should be trimmed to 63 return; } - if (integration(index) === "snowflake") { + if (integration(index) === "snowflake" || integration(index) === "snowpipe_streaming") { expect(received[1].metadata).toHaveProperty( "table", "A_1_A_2_A_3_A_4_A_5_B_1_B_2_B_3_B_4_B_5_C_1_C_2_C_3_C_4_C_5_D_1_D_2_D_3_D_4_D_5_E_1_E_2_E_3_E_4_E_5_F_1_F_2_F_3_F_4_F_5_G_1_G_2" @@ -325,7 +321,7 @@ describe("conflict between rudder set props and user set props", () => { const propsKey = propsKeyMap[evType]; transformers.forEach((transformer, index) => { let sampleRudderPropKey = "id"; - if (integration(index) === "snowflake") { + if (integration(index) === "snowflake" || integration(index) === "snowpipe_streaming") { sampleRudderPropKey = "ID"; } @@ -373,7 +369,7 @@ describe("handle reserved words", () => { const received = transformer.process(i); const out = - evType === "track" || evType === "identify" + evType === "track" || (evType === "identify" && integration(index) !== 'snowpipe_streaming') ? received[1] : received[0]; @@ -386,7 +382,7 @@ describe("handle reserved words", () => { } else { k = snakeCasedKey; } - if (integration(index) === "snowflake") { + if (integration(index) === "snowflake" || integration(index) === "snowpipe_streaming") { expect(out.metadata.columns).toHaveProperty(k); } else { expect(out.metadata.columns).toHaveProperty(k.toLowerCase()); @@ -669,6 +665,9 @@ describe("id column datatype for users table", () => { integrationCasedString(integration(index), "user_id") ] ).toEqual("int"); + if (integration(index) === 'snowpipe_streaming') { + return + } expect( received[1].metadata.columns[ integrationCasedString(integration(index), "id") @@ -686,6 +685,9 @@ describe("id column datatype for users table", () => { integrationCasedString(integration(index), "user_id") ] ).toEqual("float"); + if (integration(index) === 'snowpipe_streaming') { + return + } expect( received[1].metadata.columns[ integrationCasedString(integration(index), "id") @@ -1054,6 +1056,12 @@ describe("Integration options", () => { return _.cloneDeep(config.output.postgres); case "snowflake": return _.cloneDeep(config.output.snowflake); + case "snowpipe_streaming": + if (eventType === 'identifies') { + return _.cloneDeep([config.output.snowflake[0]]); + } else { + return _.cloneDeep(config.output.snowflake); + } case "s3_datalake": case "gcs_datalake": case "azure_datalake": @@ -1242,7 +1250,11 @@ describe("Destination config", () => { transformers.forEach((transformer, index) => { const received = transformer.process(scenario.event); - expect(received).toHaveLength(scenario.expected.length); + if (integration(index) === 'snowpipe_streaming' && scenario.event.message.type == 'identify') { + expect(received).toHaveLength(1); + } else { + expect(received).toHaveLength(scenario.expected.length); + } for (const i in received) { const evt = received[i]; expect(evt.data.id ? evt.data.id : evt.data.ID).toEqual(scenario.expected[i].id); @@ -1285,7 +1297,12 @@ describe("Destination config", () => { } } const output = transformer.process(event); - const events = [output[0], output[1]]; // identifies and users event + let events = []; + if (integration(index) === 'snowpipe_streaming') { + events = [output[0]]; + } else { + events = [output[0], output[1]]; + } const traitsToCheck = { 'city': 'Disney', 'country': 'USA', @@ -1364,7 +1381,13 @@ describe("Destination config", () => { } } const received = transformer.process(event); - const events = [received[0], received[1]]; // identifies and users event + let events = []; + if (integration(index) === 'snowpipe_streaming') { + events = [received[0]]; + } else { + events = [received[0], received[1]]; + } + // identifies and users event const traitsToCheck = { 'city': 'Disney', 'country': 'USA',