MySQL node not returning anything on database error

[CreativeWarlock]

Hi guys,

I have a for loop here which basically iterates over an array of insert strings.

As you can see, one of the inserts resulted in a DUPLICATE entry error.

Ofc I could check beforehand if the data already exists in the DB, but I would rather avoid that extra query. As you can see in the flow, I was expecting that this mySql node would return something every time, so that the error can be handled, e.g. remove that insert statement from the array.
It would just be perfectly simple.

Has anyone a good idea? @dceejay or @knolleary maybe?

Cheers,
Marcel

P.S.: Here's the content of that 'increase iterator' node, just to demonstrate what I would wanna do if the mysql node were returning an error object:

if (msg.payload !== undefined && msg.payload.affectedRows == 1) {
    node.error("1 affected row.");

    var iter = flow.get("temp_iterator");

    if (iter !== undefined ) {
        flow.set("temp_iterator", iter+1);
    }
    
    // TODO: remove insert statement from 'temp_signal_inserts' array

    return [msg, null];
}

node.error("Error! Failure to insert ...");
//node.error(msg.topic);

// TODO: remove insert statement from 'temp_signal_inserts' array

return [null, msg];

[Steve-Mcl]

Have you tried a catch node?

Also, looping isnt really recommended (if you make a mistake you could cause a stack crash) - the better node-red pattern is to use the split node. If you export your flow I will show you how.

[CreativeWarlock]

After coming back to my flow, I had the very same idea.

Sometimes you just don't see the forest if you're among hundreds of trees.

[CreativeWarlock]

Sure, happy to learn something new @Steve-Mcl

Here's my flow exported:

[{"id":"c55fdfbe.ac712","type":"group","z":"9f0ab3cb.f5606","name":"Write historic signals to DB","style":{"fill":"#bfdbef","label":true},"nodes":["fb6bfd79.6e458","c1e1c4b9.745ab8","f9c80301.0d122","160c6cfd.f69063","99bdab1f.d5a4a8","dd74071.180c5f8","b6cdac5f.a7c36","8c5d833d.a13dd"],"x":434,"y":759,"w":1152,"h":162,"info":"### DB Schema\n\n## Measurement\n\n    <Symbol>\n    \n    // values: Usa500, Silver, ...\n    \n## Tags\n\n    New | C1TB | C1IT | C4BO | Event | UA\n    \n    // New = All signals are new until approved by GPA and MFI -> Then this flow will move this data point to the appropriate signal type\n    \n    // UA = User Annotation\n    \n## Field-Keys and Field-Values\n    signalType=<string> // signal type (C1TB | C1IT | C4BO | ...)\n    entry=<float>       // entry price\n    vol=<float>         // volume aka contract size\n    sl=<float>          // recommended stop loss\n    tp=<float>          // recommended take profit\n    expiresIn=<integer> // Expiry of the TLI signal (in days)\n    gpaAnno=<string>    // Annotation by GPA\n    mfiAnno=<string>    // Annotation by MFI\n    gpaGo=<boolean>     // Signal approved by GPA\n    mfiGo=<boolean>     // Signal approved by MFI\n    \n    // TLI related\n    age1=<integer>      // age1 of related TL\n    age2=<integer>      // age2 of related TL\n\n## Inserts\n\nINSERT Usa500,Signal=NEW signalType=\"C1TB\",entry=3450.0,volume=0.01,sl=3400.0,tp=3550.0,expiresIn=6i,gpaGo=false,mfiGo=true,age1=16i,age2=4i"},{"id":"fb6bfd79.6e458","type":"mysql","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","mydb":"dda14633.c67378","name":"MySQL DB","x":890,"y":800,"wires":[["dd74071.180c5f8"]]},{"id":"c1e1c4b9.745ab8","type":"debug","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"Log insert","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":1280,"y":800,"wires":[]},{"id":"f9c80301.0d122","type":"function","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"For Each","func":"var signalInserts = flow.get(\"temp_signal_inserts\");\n\nif (signalInserts == undefined) {\n    return [msg, null];\n}\n\nvar iter = flow.get(\"temp_iterator\");\n\nif (iter == undefined ) {\n    flow.set(\"temp_iterator\", 0);\n    iter = 0;\n}\n\n//node.error(\"iter: \" + iter);\n\nvar newMsg = {}\nnewMsg.payload = {};\n\nif (iter < signalInserts.length) {\n    \n    msg.topic = \"INSERT INTO `registry` (`key`, `value`) VALUES ('\" + msg.topic + \"', '\" + msg.payload + \"') ON DUPLICATE KEY UPDATE `value` = '\" + msg.payload + \"' \";\n    \n    newMsg.topic = signalInserts[iter];\n    return [newMsg, null];\n}\n\n// Pseudo loop finished. Cleanup temp variables:\nflow.set(\"temp_iterator\", undefined);\nflow.set(\"signalInserts\", undefined);\n\nreturn [null, msg];","outputs":2,"noerr":0,"initialize":"","finalize":"","x":720,"y":800,"wires":[["fb6bfd79.6e458"],[]]},{"id":"160c6cfd.f69063","type":"inject","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"","payloadType":"date","x":540,"y":800,"wires":[["f9c80301.0d122"]]},{"id":"99bdab1f.d5a4a8","type":"function","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"increase iterator","func":"if (msg.payload !== undefined && msg.payload.affectedRows == 1) {\n    node.error(\"1 affected row.\");\n\n    var iter = flow.get(\"temp_iterator\");\n\n    if (iter !== undefined ) {\n        flow.set(\"temp_iterator\", iter+1);\n    }\n    \n\n    return [msg, null];\n}\n\nnode.error(\"Error! Failure to insert ...\");\n//node.error(msg.topic);\n\nreturn [null, msg];","outputs":2,"noerr":0,"initialize":"","finalize":"","x":1280,"y":880,"wires":[["f9c80301.0d122"],["b6cdac5f.a7c36"]]},{"id":"dd74071.180c5f8","type":"delay","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"","pauseType":"delay","timeout":"100","timeoutUnits":"milliseconds","rate":"1","nbRateUnits":"1","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"x":1070,"y":800,"wires":[["99bdab1f.d5a4a8","c1e1c4b9.745ab8"]]},{"id":"b6cdac5f.a7c36","type":"debug","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"Failed insert","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":1470,"y":880,"wires":[]},{"id":"8c5d833d.a13dd","type":"catch","z":"9f0ab3cb.f5606","g":"c55fdfbe.ac712","name":"MySQL Error","scope":["fb6bfd79.6e458"],"uncaught":false,"x":1070,"y":880,"wires":[["99bdab1f.d5a4a8"]]},{"id":"dda14633.c67378","type":"MySQLdatabase","name":"TradeData-DB","host":"mySQL","port":"3306","db":"TRADE_DATA","tz":"","charset":"UTF8"}]

And this "box" is responsible for reading in CSV data into that flow context variable:

[{"id":"22cea03a.4f19e","type":"group","z":"9f0ab3cb.f5606","name":"Read incoming historic signals","style":{"fill":"#bfc7d7","label":true},"nodes":["dbfe48d2.fa8638","f47fba1a.a9dcf8","aee57303.1a46f","2e8c9531.fecb9a","208d48b2.2dcd18","42335099.22e0d","f3a1379c.16db18"],"x":54,"y":559,"w":1052,"h":149.5},{"id":"dbfe48d2.fa8638","type":"file in","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","name":"Read CSV","filename":"","format":"utf8","chunk":false,"sendError":false,"encoding":"utf8","x":210,"y":660,"wires":[["f47fba1a.a9dcf8"]]},{"id":"f47fba1a.a9dcf8","type":"csv","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","name":"","sep":";","hdrin":true,"hdrout":"none","multi":"one","ret":"\\r\\n","temp":"","skip":"0","strings":true,"include_empty_strings":false,"include_null_values":true,"x":370,"y":660,"wires":[["2e8c9531.fecb9a"]]},{"id":"aee57303.1a46f","type":"watch-directory","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","folder":"/signal_data_historic","recursive":"3","typeEvent":"create","ignoreInitial":true,"ignoredFiles":"log$|jpg$|png$","ignoredFilesType":"re","name":"","x":170,"y":600,"wires":[["dbfe48d2.fa8638"]]},{"id":"2e8c9531.fecb9a","type":"switch","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","name":"","property":"payload.SIGNAL_TYPE","propertyType":"msg","rules":[{"t":"eq","v":"C1TB","vt":"str"},{"t":"eq","v":"C1IT","vt":"str"},{"t":"else"}],"checkall":"false","repair":false,"outputs":3,"x":510,"y":660,"wires":[["208d48b2.2dcd18"],["42335099.22e0d"],[]]},{"id":"208d48b2.2dcd18","type":"function","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","name":"INSERT C1TB signal(s)","func":"if (msg.payload == undefined || msg.payload.broker == \"\")\n    return [null, msg];\n\nmsg.topic = `INSERT INTO H_SIGNALS_C1TB_A14 (BROKER, SYMBOL, TIME, TIMESTAMP, DIRECTION, VOL, PRICE, SL, TP) \n    VALUES (\n        '${msg.payload.BROKER}',\n        '${msg.payload.SYMBOL}',\n        FROM_UNIXTIME(${msg.payload.TIMESTAMP}),\n        ${msg.payload.TIMESTAMP},\n        '${msg.payload.DIRECTION}',\n        ${msg.payload.VOL},\n        ${msg.payload.PRICE},\n        ${msg.payload.SL},\n        ${msg.payload.TP}\n    );`\n\nreturn [msg, null];","outputs":2,"noerr":0,"initialize":"","finalize":"","x":710,"y":600,"wires":[["f3a1379c.16db18"],[]]},{"id":"42335099.22e0d","type":"function","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","name":"INSERT C1IT signal(s)","func":"if (msg.payload == undefined || msg.payload.broker == \"\")\n    return [null, msg];\n\nmsg.topic = `INSERT INTO H_SIGNALS_C1IT_A14 (BROKER, SYMBOL, TIME, TIMESTAMP, DIRECTION, VOL, PRICE, SL, TP) \n    VALUES (\n        '${msg.payload.BROKER}',\n        '${msg.payload.SYMBOL}',\n        FROM_UNIXTIME(${msg.payload.TIMESTAMP}),\n        ${msg.payload.TIMESTAMP},\n        '${msg.payload.DIRECTION}',\n        ${msg.payload.VOL},\n        ${msg.payload.PRICE},\n        ${msg.payload.SL},\n        ${msg.payload.TP}\n    );`\n\nreturn [msg, null];","outputs":2,"noerr":0,"initialize":"","finalize":"","x":700,"y":660,"wires":[["f3a1379c.16db18"],[]]},{"id":"f3a1379c.16db18","type":"function","z":"9f0ab3cb.f5606","g":"22cea03a.4f19e","name":"Add to 'signal_inserts' array","func":"if (msg.topic !== undefined ) {\n    var signalInserts = flow.get(\"temp_signal_inserts\");\n    \n    if (signalInserts === undefined ) {\n        node.error(\"No signalInserts array found. Inintializing new one.\");\n        signalInserts = [];\n    }\n    \n    signalInserts.push(msg.topic);\n        \n    flow.set(\"temp_signal_inserts\", signalInserts);\n} else {\n    node.error(\"Nothing to insert to temp_signal_inserts array.\");\n}\n\nvar iter = flow.get(\"temp_iterator\");\n\nreturn msg;","outputs":1,"noerr":0,"initialize":"","finalize":"","x":960,"y":640,"wires":[[]]}]

[Ashfaak]

Could try "insert ignore" rather than "insert" alone for this purpose.

[Steve-Mcl]

@CreativeWarlock

Sure, happy to learn something new @Steve-Mcl

So here it is without a loop & code is somewhat simplified.

Also, I changed from dynamic SQL to prepared statements to avoid any chance of someone causing a SQL injection hack (someone could easily write some bad string in your CSV to wipe out your database)

take a look, see what you think.

[{"id":"63387117.11e7f","type":"inject","z":"6fe62fab.dc09d","name":"","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"","payloadType":"date","x":840,"y":140,"wires":[["e57058a5.a0a1a8"]]},{"id":"e57058a5.a0a1a8","type":"change","z":"6fe62fab.dc09d","name":"","rules":[{"t":"set","p":"temp_signal_inserts","pt":"flow","to":"[{\"key\":\"sensor1\",\"val\":11.98},{\"key\":\"sensor1\",\"val\":11.78},{\"key\":\"sensor2\",\"val\":12.02},{\"key\":\"sensor2\",\"val\":11.99},{\"key\":\"sensor1\",\"val\":11.69},{\"key\":\"sensor2\",\"val\":11.68},{\"key\":\"sensor2\",\"val\":11.52},{\"key\":\"sensor1\",\"val\":10.91}]","tot":"json"}],"action":"","property":"","from":"","to":"","reg":false,"x":1040,"y":140,"wires":[[]]},{"id":"5053fe36.791c","type":"comment","z":"6fe62fab.dc09d","name":"Generate some fake CSV data","info":"","x":890,"y":100,"wires":[]},{"id":"c55fdfbe.ac712","type":"group","z":"6fe62fab.dc09d","name":"Write historic signals to DB","style":{"fill":"#bfdbef","label":true},"nodes":["fb6bfd79.6e458","c1e1c4b9.745ab8","f9c80301.0d122","160c6cfd.f69063","39ba4502.661bda","57c98031.a8b76","62142ed5.2d303"],"x":754,"y":179,"w":872,"h":122,"info":"### DB Schema\n\n## Measurement\n\n    <Symbol>\n    \n    // values: Usa500, Silver, ...\n    \n## Tags\n\n    New | C1TB | C1IT | C4BO | Event | UA\n    \n    // New = All signals are new until approved by GPA and MFI -> Then this flow will move this data point to the appropriate signal type\n    \n    // UA = User Annotation\n    \n## Field-Keys and Field-Values\n    signalType=<string> // signal type (C1TB | C1IT | C4BO | ...)\n    entry=<float>       // entry price\n    vol=<float>         // volume aka contract size\n    sl=<float>          // recommended stop loss\n    tp=<float>          // recommended take profit\n    expiresIn=<integer> // Expiry of the TLI signal (in days)\n    gpaAnno=<string>    // Annotation by GPA\n    mfiAnno=<string>    // Annotation by MFI\n    gpaGo=<boolean>     // Signal approved by GPA\n    mfiGo=<boolean>     // Signal approved by MFI\n    \n    // TLI related\n    age1=<integer>      // age1 of related TL\n    age2=<integer>      // age2 of related TL\n\n## Inserts\n\nINSERT Usa500,Signal=NEW signalType=\"C1TB\",entry=3450.0,volume=0.01,sl=3400.0,tp=3550.0,expiresIn=6i,gpaGo=false,mfiGo=true,age1=16i,age2=4i"},{"id":"fb6bfd79.6e458","type":"mysql","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","mydb":"9721b43b.2be718","name":"MySQL DB","x":1470,"y":260,"wires":[["c1e1c4b9.745ab8"]]},{"id":"c1e1c4b9.745ab8","type":"debug","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","name":"Log insert","active":true,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":1520,"y":220,"wires":[]},{"id":"f9c80301.0d122","type":"function","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","name":"make SQL","func":"  \nmsg.topic = \"INSERT INTO `registry` (`key`, `value`) VALUES (?,?);\";\nmsg.payload = [\n    msg.payload.key,\n    msg.payload.val,\n]   \nreturn msg;","outputs":1,"noerr":0,"initialize":"","finalize":"","libs":[],"x":1050,"y":260,"wires":[["39ba4502.661bda"]]},{"id":"160c6cfd.f69063","type":"inject","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","name":"flow.temp_signal_inserts","props":[{"p":"payload"},{"p":"topic","vt":"str"}],"repeat":"","crontab":"","once":false,"onceDelay":0.1,"topic":"","payload":"temp_signal_inserts","payloadType":"flow","x":910,"y":220,"wires":[["57c98031.a8b76"]]},{"id":"39ba4502.661bda","type":"delay","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","name":"rate limit 100ms apart","pauseType":"rate","timeout":"100","timeoutUnits":"milliseconds","rate":"10","nbRateUnits":"1","rateUnits":"second","randomFirst":"1","randomLast":"5","randomUnits":"seconds","drop":false,"x":1240,"y":260,"wires":[["62142ed5.2d303","fb6bfd79.6e458"]]},{"id":"57c98031.a8b76","type":"split","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","name":"","splt":"\\n","spltType":"str","arraySplt":1,"arraySpltType":"len","stream":false,"addname":"","x":910,"y":260,"wires":[["f9c80301.0d122"]]},{"id":"62142ed5.2d303","type":"debug","z":"6fe62fab.dc09d","g":"c55fdfbe.ac712","name":"Log insert","active":false,"tosidebar":true,"console":false,"tostatus":false,"complete":"true","targetType":"full","statusVal":"","statusType":"auto","x":1300,"y":220,"wires":[]},{"id":"9721b43b.2be718","type":"MySQLdatabase","name":"","host":"192.168.1.38","port":"3306","db":"test","tz":"","charset":"UTF8"}]

[CreativeWarlock]

Didn't know of this option. Thanks a lot, @Ashfaak!

[CreativeWarlock]

Thanks, Steve - I'll have a closer look tonight.

[CreativeWarlock]

This is just fantastic! I love the simplicity & beauty of it. No more silly loops in a flow!

You guys rock! Thanks a lot for all your help!

Cheers,
Marcel