I have an ADF Copy Activity (Snowflake source, Azure Storage sink) which passes in the schema and table name using variables. Below is the dynamic query...
It is failing with the error 'The template function 'iem' is not defined or not valid'. The input generated by the query is this...
Not sure what I'm doing wrong. The schema and table needs the double quotes or else the Snowflake instance will reject it. When make the query for a hard coded table (i.e. SELECT * FROM "schema"."table") it works fine. Any suggestions would be much appreciated.
Updated with JSON. I've changed it to a simple select statement with no parameters. Still getting the same error. However, it works when I preview the data.
I've only included JSON for the For Each loop as the full pipeline was too large.
{
"name": "ForEachINCTable",
"type": "ForEach",
"dependsOn": [
{
"activity": "LookupINCTable",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"items": {
"value": "@activity('LookupINCTable').output.value",
"type": "Expression"
},
"batchCount": 10,
"activities": [
{
"name": "WriteCopyLogInc",
"type": "SqlServerStoredProcedure",
"dependsOn": [
{
"activity": "aCopyINC",
"dependencyConditions": [
"Completed"
]
}
],
"policy": {
"timeout": "7.00:00:00",
"retry": 1,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"storedProcedureName": "Log.InsertADFActivityLog",
"storedProcedureParameters": {
"01ID": {
"value": {
"value": "@activity('aCopyINC').ActivityRunId",
"type": "Expression"
},
"type": "String"
},
"02ProcessVariablesJSON": {
"value": {
"value": "@{item()}",
"type": "Expression"
},
"type": "String"
},
"03ActivityOutputJSON": {
"value": {
"value": "@{activity('aCopyINC')}",
"type": "Expression"
},
"type": "String"
},
"04PipelineVariablesJSON": {
"value": {
"value": "@{pipeline()}",
"type": "Expression"
},
"type": "String"
}
}
},
"linkedServiceName": {
"referenceName": "dnaMetadataLS",
"type": "LinkedServiceReference"
}
},
{
"name": "AddFileDescriptionINC",
"type": "SqlServerStoredProcedure",
"dependsOn": [
{
"activity": "aCopyINC",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "7.00:00:00",
"retry": 1,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"storedProcedureName": "[Metadata].InsertFileDescription",
"storedProcedureParameters": {
"01TTID": {
"value": {
"value": "@item().SourceEntityId",
"type": "Expression"
},
"type": "Int64"
},
"02ADFRunId": {
"value": {
"value": "@pipeline().TriggerId",
"type": "Expression"
},
"type": "Guid"
},
"03Container": {
"value": {
"value": "@item().TargetSubSystem",
"type": "Expression"
},
"type": "String"
},
"04Directory": {
"value": {
"value": "@item().TargetDirectory",
"type": "Expression"
},
"type": "String"
},
"05File": {
"value": {
"value": "@item().TargetFile",
"type": "Expression"
},
"type": "String"
},
"06CreatedDate": {
"value": {
"value": "@formatDateTime(utcnow(),'yyyy-MM-dd HH:mm:ss')",
"type": "Expression"
},
"type": "DateTime"
},
"07LoadType": {
"value": {
"value": "@item().LoadType",
"type": "Expression"
},
"type": "String"
},
"08IncrementalFrom": {
"value": {
"value": "@item().IncrementalFrom",
"type": "Expression"
},
"type": "String"
},
"09IncrementalTo": {
"value": {
"value": "@item().IncrementalTo",
"type": "Expression"
},
"type": "String"
},
"10LoadStatusID": {
"value": "2",
"type": "Int32"
}
}
},
"linkedServiceName": {
"referenceName": "dnaMetadataLS",
"type": "LinkedServiceReference"
}
},
{
"name": "aCopyINC",
"type": "Copy",
"dependsOn": [
{
"activity": "Script Get Trigger Time",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "7.00:00:00",
"retry": 2,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [
{
"name": "SourceSystem",
"value": "@{item().ConnectionID}"
},
{
"name": "Table",
"value": "@{item().SourceEntityName}"
},
{
"name": "TTID",
"value": "@{item().SourceEntityId}"
}
],
"typeProperties": {
"source": {
"type": "SnowflakeV2Source",
"query": {
"value": "@concat('SELECT * FROM \"', item().SourceSchemaName, '\".\"', item().SourceEntityName, \n '\" WHERE ', item().IncrementalColumn, ' >= ''', formatDateTime(replace(item().IncrementalFrom, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), \n ''' AND ', item().IncrementalColumn, ' < ''', formatDateTime(replace(item().IncrementalTo, 'T', ' '), 'yyyy-MM-ddTHH:mm:ss.fff'), '''')",
"type": "Expression"
},
"exportSettings": {
"type": "SnowflakeExportCopyCommand"
}
},
"sink": {
"type": "DelimitedTextSink",
"storeSettings": {
"type": "AzureBlobStorageWriteSettings"
},
"formatSettings": {
"type": "DelimitedTextWriteSettings",
"quoteAllText": true,
"fileExtension": ".txt"
}
},
"enableStaging": false
},
"inputs": [
{
"referenceName": "DSRedzone",
"type": "DatasetReference",
"parameters": {
"tableName": {
"value": "@item().SourceEntityName",
"type": "Expression"
},
"schemaName": {
"value": "@item().SourceSchemaName",
"type": "Expression"
}
}
}
],
"outputs": [
{
"referenceName": "DSBlobStorage_SAS_csv",
"type": "DatasetReference",
"parameters": {
"Container": {
"value": "@item().TargetSubSystem",
"type": "Expression"
},
"Directory": {
"value": "@{item().TargetDirectory}",
"type": "Expression"
},
"File": {
"value": "@{item().TargetFile}",
"type": "Expression"
},
"Separator": ",",
"HeaderRow": "true",
"Escape": "\"",
"Quote": "\""
}
}
]
},
{
"name": "Update Incremental Value",
"type": "Script",
"dependsOn": [
{
"activity": "aCopyINC",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"linkedServiceName": {
"referenceName": "dnaMetadataLS",
"type": "LinkedServiceReference"
},
"typeProperties": {
"scripts": [
{
"type": "NonQuery",
"text": {
"value": "UPDATE [Metadata].[IncrementalState]\nSET [LastRunIncrementalFrom] = CONVERT(DATETIME2(0),'@{item().IncrementalFrom}',120),\n [LastRunIncrementalTo] = CONVERT(DATETIME2(0), SYSDATETIME(),120)\nWHERE SourceEntityID = @{item().SourceEntityId} ;\n\nUPDATE [Metadata].[IncrementalState]\nSET [IncrementalFrom] = CONVERT(DATETIME2(0), SYSDATETIME(), 120),\n [IncrementalTo] = '9999-12-31 00:00:00'\nWHERE SourceEntityID = @{item().SourceEntityId} ;",
"type": "Expression"
}
}
],
"scriptBlockExecutionTimeout": "02:00:00"
}
},
{
"name": "Script Get Trigger Time",
"type": "Script",
"dependsOn": [],
"policy": {
"timeout": "0.12:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"linkedServiceName": {
"referenceName": "dnaMetadataLS",
"type": "LinkedServiceReference"
},
"typeProperties": {
"scripts": [
{
"type": "Query",
"text": {
"value": "DECLARE @@TriggerTime AS Datetime2(0);\n\nSELECT @@TriggerTime = TriggerTime \nFROM [Metadata].[PipelineExecutionLastSuccessfull]\nWHERE pipeline='@{pipeline().Pipeline}';\n\n",
"type": "Expression"
}
}
],
"scriptBlockExecutionTimeout": "02:00:00"
}
}
]
}
}
