I have created a pipeline in ADF V2 that is copying data from an Odata connector source to a Azure SQL DB sink, the dataset source and sink are dynamic so it iterates through the multiple entities (tables) from the Odata source and copies them over to the corresponding tables in SQL. I've tested the copy activity independently and that works when iterating through the dynamic datasets but when I incorporate an until activity as a parent to the copy activity which is to used to iterate through multiple pages of data per entity(table) from the Odata source it gives me this error; "code":"BadRequest","message":null,"target":"pipeline//runid/XXXX","details":null,"error":null}
The only advice I can find from from similar threads on here is that there is an issue in the underlying JSON payload with a special character like a { or space corrupting the code and specific examples have been related to when the linked service has been made dynamic but I can not see any such issues in my case, I've pasted the code below;
Note: My linked service is not parameterised, just the datasets and I am able to preview the data within the datasets.
{
"name": "pl_AllEntities",
"properties": {
"activities": [
{
"name": "Lookup OdataMetadata",
"type": "Lookup",
"dependsOn": [],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "AzureSqlSource",
"sqlReaderQuery": {
"value": "SELECT OdataEntityName\nFROM [dbo].[Config_OdataEntitySync]",
"type": "Expression"
},
"queryTimeout": "02:00:00",
"isolationLevel": "ReadCommitted",
"partitionOption": "None"
},
"dataset": {
"referenceName": "ds_Config_OdataEntitySync",
"type": "DatasetReference"
},
"firstRowOnly": false
}
},
{
"name": "ForEach OdataEntity",
"type": "ForEach",
"dependsOn": [
{
"activity": "Lookup OdataMetadata",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"items": {
"value": "@activity('Lookup OdataMetadata').output.value",
"type": "Expression"
},
"activities": [
{
"name": "Lookup LastRowChangedDatetime",
"type": "Lookup",
"dependsOn": [],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "AzureSqlSource",
"sqlReaderStoredProcedureName": "[dbo].[usp_Config_GetHighWaterMark]",
"storedProcedureParameters": {
"inOdataEntityName": {
"type": "String",
"value": {
"value": "@item().OdataEntityName",
"type": "Expression"
}
}
},
"queryTimeout": "02:00:00",
"partitionOption": "None"
},
"dataset": {
"referenceName": "ds_GeenGBDBxxxxx",
"type": "DatasetReference"
}
}
},
{
"name": "Set LastRowChangeDatetime",
"type": "SetVariable",
"dependsOn": [
{
"activity": "Lookup LastRowChangedDatetime",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"variableName": "LastRowChangedDatetime",
"value": {
"value": "@string(activity('Lookup LastRowChangedDatetime').output.firstRow['LastRowChangedDatetime'])",
"type": "Expression"
}
}
},
{
"name": "Until AllRowsCopied",
"type": "Until",
"dependsOn": [
{
"activity": "Set LastRowChangeDatetime",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"expression": {
"value": "@equals(variables('RowRead'),true)",
"type": "Expression"
},
"activities": [
{
"name": "Copy Data",
"type": "Copy",
"dependsOn": [],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"source": {
"type": "ODataSource",
"query": {
"value": "@concat('$filter=RowChangedDatetime gt ',convertTimeZone(variables('LastRowChangedDatetime'),'UTC', 'UTC'),'&$top=',variables('RowCount'),'&$skip=',variables('RowCountAdder'))",
"type": "Expression"
},
"httpRequestTimeout": "00:05:00"
},
"sink": {
"type": "AzureSqlSink",
"writeBehavior": "insert",
"disableMetricsCollection": false
},
"enableStaging": false
},
"inputs": [
{
"referenceName": "ds_OData_DynamicSource",
"type": "DatasetReference",
"parameters": {
"EntityName": {
"value": "@item().OdataEntityName",
"type": "Expression"
}
}
}
],
"outputs": [
{
"referenceName": "ds_GeenSQL_DynamicSink",
"type": "DatasetReference",
"parameters": {
"SchemaName": "dbo",
"TableNamePrefix": "Staging_",
"TableEntityName": {
"value": "@item().OdataEntityName",
"type": "Expression"
}
}
}
]
},
{
"name": "Set RowCountAdderTemp",
"type": "SetVariable",
"dependsOn": [
{
"activity": "Copy Data",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"variableName": "RowCountAdderTemp",
"value": {
"value": "@string(add(int(variables('RowCountAdder')),10000))",
"type": "Expression"
}
}
},
{
"name": "Set RowCountAdder",
"type": "SetVariable",
"dependsOn": [
{
"activity": "Set RowCountAdderTemp",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"variableName": "RowCountAdder",
"value": {
"value": "@variables('RowCountAdderTemp')",
"type": "Expression"
}
}
},
{
"name": "If Condition1",
"type": "IfCondition",
"dependsOn": [
{
"activity": "Set RowCountAdder",
"dependencyConditions": [
"Succeeded"
]
}
],
"userProperties": [],
"typeProperties": {
"expression": {
"value": "@equals(activity('Copy Data').output.rowsRead,0)",
"type": "Expression"
},
"ifFalseActivities": [
{
"name": "Set variable1",
"type": "SetVariable",
"dependsOn": [],
"userProperties": [],
"typeProperties": {
"variableName": "RowRead",
"value": false
}
}
],
"ifTrueActivities": [
{
"name": "Set variable2",
"description": "",
"type": "SetVariable",
"dependsOn": [],
"userProperties": [],
"typeProperties": {
"variableName": "RowRead",
"value": true
}
}
]
}
}
],
"timeout": "7.00:00:00"
}
},
{
"name": "usp_Config_GetHighWaterMark",
"type": "SqlServerStoredProcedure",
"dependsOn": [
{
"activity": "Until AllRowsCopied",
"dependencyConditions": [
"Succeeded"
]
}
],
"policy": {
"timeout": "7.00:00:00",
"retry": 0,
"retryIntervalInSeconds": 30,
"secureOutput": false,
"secureInput": false
},
"userProperties": [],
"typeProperties": {
"storedProcedureName": "[dbo].[usp_Config_GetHighWaterMark]",
"storedProcedureParameters": {
"inOdataEntityName": {
"value": {
"value": "@item().OdataEntityName",
"type": "Expression"
},
"type": "String"
},
"inSyncEnabled": {
"value": {
"value": "@pipeline().parameters.inSyncEnabled",
"type": "Expression"
},
"type": "Boolean"
}
}
},
"linkedServiceName": {
"referenceName": "ls_AzureSqldb",
"type": "LinkedServiceReference"
}
}
]
}
}
],
"parameters": {
"OdataEntityName": {
"type": "string"
},
"inSyncEnabled": {
"type": "bool",
"defaultValue": true
}
},
"variables": {
"LastRowChangedDatetime": {
"type": "String"
},
"RowCount": {
"type": "String",
"defaultValue": "10000"
},
"RowCountAdder": {
"type": "String",
"defaultValue": "0"
},
"RowCountTemp": {
"type": "String"
},
"RowCountAdderTemp": {
"type": "String"
},
"RowRead": {
"type": "Boolean"
}
},
"folder": {
"name": "Odata"
},
"annotations": []
}
}
