HPCC-21359 MySQL plugin error calling stored procedure

Also added ability to define and delete stored procedures, for ease of writing
test code.

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

plugins/mysql/mysqlembed.cpp

@@ -17,6 +17,7 @@
 
 #include "platform.h"
 #include "mysql.h"
+#include "mysqld_error.h"
 #include "jexcept.hpp"
 #include "jthread.hpp"
 #include "hqlplugins.hpp"
@@ -638,18 +639,17 @@ public:
     {
         // Create bindings for input parameters
         inputBindings.init(mysql_stmt_param_count(*stmt));
-        // And for results
-        res.setown(new MySQLResult(mysql_stmt_result_metadata(*stmt)));
-        if (*res)
-        {
-            resultBindings.bindResults(*res);
-            /* Bind the result buffers */
-            if (mysql_stmt_bind_result(*stmt, resultBindings.queryBindings()))
-                fail(mysql_stmt_error(*stmt));
-        }
-        else if (mysql_stmt_errno(*stmt))  // SQL actions don't return results...
+        // Bindings for results are created after the execute, as they are not always available until then (e.g. when calling a stored procedure)
+        if (mysql_stmt_errno(*stmt))
             fail(mysql_stmt_error(*stmt));
     }
+    MySQLPreparedStatement(MySQLConnection *_conn, const char *_query, unsigned _len)
+    : conn(_conn)
+    {
+        // Used for cases with no parameters or result, that are not supported in prepared query protocol - e.g. DROP PROCEDURE
+        query.set(_query, _len);
+        inputBindings.init(0);
+    }
     ~MySQLPreparedStatement()
     {
         stop();
@@ -673,11 +673,27 @@ public:
     }
     void execute()
     {
-        assertex(stmt && *stmt);
-        if (inputBindings.numColumns() && mysql_stmt_bind_param(*stmt, inputBindings.queryBindings()))
-            fail(mysql_stmt_error(*stmt));
-        if (mysql_stmt_execute(*stmt))
-            fail(mysql_stmt_error(*stmt));
+        if (stmt && *stmt)
+        {
+            if (inputBindings.numColumns() && mysql_stmt_bind_param(*stmt, inputBindings.queryBindings()))
+                fail(mysql_stmt_error(*stmt));
+            if (mysql_stmt_execute(*stmt))
+                fail(mysql_stmt_error(*stmt));
+            // NOTE - we ignore all but the first result from stored procedures
+            res.setown(new MySQLResult(mysql_stmt_result_metadata(*stmt)));
+            if (*res)
+            {
+                resultBindings.bindResults(*res);
+                /* Bind the result buffers */
+                if (mysql_stmt_bind_result(*stmt, resultBindings.queryBindings()))
+                    fail(mysql_stmt_error(*stmt));
+            }
+        }
+        else
+        {
+            if (mysql_query(*conn, query))
+                fail(mysql_error(*conn));
+        }
     }
     inline const MySQLBindingArray &queryResultBindings() const
     {
@@ -697,6 +713,7 @@ protected:
     Owned<MySQLResult> res;
     MySQLBindingArray inputBindings;
     MySQLBindingArray resultBindings;
+    StringAttr query;
 };
 
 // Conversions from MySQL values to ECL data
@@ -1385,8 +1402,8 @@ static void initializeMySqlThread()
 class MySQLEmbedFunctionContext : public CInterfaceOf<IEmbedFunctionContext>
 {
 public:
-    MySQLEmbedFunctionContext(const char *options)
-      : nextParam(0)
+    MySQLEmbedFunctionContext(unsigned _flags, const char *options)
+      : flags(_flags), nextParam(0)
     {
         initializeMySqlThread();
         conn.setown(MySQLConnection::findCachedConnection(options, false));
@@ -1450,9 +1467,9 @@ public:
     }
     virtual byte * getRowResult(IEngineRowAllocator * _resultAllocator)
     {
+        lazyExecute();
         if (!stmtInfo->hasResult())
             typeError("row", NULL);
-        lazyExecute();
         MySQLRowStream stream(NULL, stmtInfo, _resultAllocator);
         roxiemem::OwnedConstRoxieRow ret = stream.nextRow();
         roxiemem::OwnedConstRoxieRow ret2 = stream.nextRow();
@@ -1587,7 +1604,6 @@ public:
     {
     }
 
-
     virtual void importFunction(size32_t lenChars, const char *text)
     {
         throwUnexpected();
@@ -1602,6 +1618,18 @@ public:
                 fail("failed to create statement");
             if (mysql_stmt_prepare(*stmt, script, len))
             {
+                int rc = mysql_stmt_errno(*stmt);
+                if (rc == ER_UNSUPPORTED_PS)
+                {
+                    // Some functions are not supported in prepared statements, but are still handy to be able to call
+                    // So long as they have no bound vars and no return value, we can probably call them ok
+                    if ((flags & (EFnoreturn|EFnoparams)) == (EFnoreturn|EFnoparams))
+                    {
+                        stmtInfo.setown(new MySQLPreparedStatement(conn, script, len));
+                        break;
+                    }
+                    fail(mysql_stmt_error(*stmt));
+                }
                 // If we get an error, it could be that the cached connection is stale - retry
                 if (conn->wasCached())
                 {
@@ -1620,7 +1648,7 @@ public:
             failx("Not enough parameters supplied (%d parameters supplied, but statement has %d bound columns)", nextParam, stmtInfo->queryInputBindings().numColumns());
         // We actually do the execute later, when the result is fetched
         // Unless, there is no expected result, in that case execute query now
-        if (stmtInfo->queryResultBindings().numColumns() == 0)
+        if (flags & EFnoreturn)
             lazyExecute();
     }
 protected:
@@ -1633,9 +1661,9 @@ protected:
     }
     const MYSQL_BIND &getScalarResult()
     {
+        lazyExecute();
         if (!stmtInfo->hasResult() || stmtInfo->queryResultBindings().numColumns() != 1)
             typeError("scalar", NULL);
-        lazyExecute();
         if (!stmtInfo->next())
             typeError("scalar", NULL);
         return stmtInfo->queryResultBindings().queryColumn(0, NULL);
@@ -1655,6 +1683,7 @@ protected:
     Owned<MySQLConnection> conn;
     Owned<MySQLPreparedStatement> stmtInfo;
     Owned<MySQLDatasetBinder> inputStream;
+    unsigned flags;
     int nextParam;
 };
 
@@ -1670,7 +1699,7 @@ public:
         if (flags & EFimport)
             UNSUPPORTED("IMPORT");
         else
-            return new MySQLEmbedFunctionContext(options);
+            return new MySQLEmbedFunctionContext(flags, options);
     }
     virtual IEmbedServiceContext *createServiceContext(const char *service, unsigned flags, const char *options)
     {

testing/regress/ecl/key/mysqlembed.xml

@@ -60,3 +60,9 @@
  <Row><name>name1</name><bval>16737</bval><value>1</value><boolval>true</boolval><r8>1.2</r8><r4>3.400000095367432</r4><d>6161353561613535</d><ddd>1234567.89</ddd><u1>Straße</u1><u2>Straße  </u2><dt>1963-11-22 12:30:00</dt></Row>
  <Row><name>name2</name><bval>66</bval><value>2</value><boolval>false</boolval><r8>5.6</r8><r4>7.800000190734863</r4><d>3030</d><ddd>-1234567.89</ddd><u1>là</u1><u2>là      </u2><dt>2015-12-25 01:23:45</dt></Row>
 </Dataset>
+<Dataset name='Result 19'>
+ <Row><Result_19>Hello, World!</Result_19></Row>
+</Dataset>
+<Dataset name='Result 20'>
+ <Row><name>name1</name><bval>16737</bval><value>1</value><boolval>true</boolval><r8>1.2</r8><r4>3.400000095367432</r4><d>6161353561613535</d><ddd>1234567.89</ddd><u1>Straße</u1><u2>Straße  </u2><dt>1963-11-22 12:30:00</dt></Row>
+</Dataset>

testing/regress/ecl/mysqlembed.ecl

@@ -52,14 +52,38 @@ END;
 init := DATASET([{'name1', 0x4161, 1, true, 1.2, 3.4, D'aa55aa55', 1234567.89, U'Straße', U'Straße'},
                  {'name2', 66, 2, false, 5.6, 7.8, D'00', -1234567.89, U'là', U'là', '2015-12-25 01:23:45' }], childrec);
 
-drop() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+drop1() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
   DROP TABLE IF EXISTS tbl1;
 ENDEMBED;
 
-create() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+drop2() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+  DROP PROCEDURE IF EXISTS testSP;
+ENDEMBED;
+
+drop3() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+  DROP FUNCTION IF EXISTS hello;
+ENDEMBED;
+
+drop() := SEQUENTIAL(drop1(), drop2(), drop3());
+
+create1() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
   CREATE TABLE tbl1 ( name VARCHAR(20), bval BIT(15), value INT, boolval TINYINT, r8 DOUBLE, r4 FLOAT, d BLOB, ddd DECIMAL(10,2), u1 VARCHAR(10), u2 VARCHAR(10), dt DATETIME );
 ENDEMBED;
 
+create2() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+  CREATE FUNCTION hello(s CHAR(255)) RETURNS char(255) DETERMINISTIC
+     RETURN CONCAT('Hello, ',s,'!') 
+ENDEMBED;
+
+create3() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+  CREATE PROCEDURE testSP(IN in_name VARCHAR(255))
+  BEGIN
+     SELECT * FROM tbl1 where name=in_name;
+  END
+ENDEMBED;
+
+create() := SEQUENTIAL(create1(), create2(), create3());
+
 initialize(dataset(childrec) values) := EMBED(mysql : server(myServer),user(myUser),database(myDB))
   INSERT INTO tbl1 values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
 ENDEMBED;
@@ -110,6 +134,14 @@ string testMySQLString() := EMBED(mysql : server(myServer),user(myUser),database
   SELECT max(name) from tbl1;
 ENDEMBED;
 
+string testMySQLStoredProcedure() := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+  select Hello('World');
+ENDEMBED;
+
+streamed dataset(childrec) testMySQLStoredProcedure2(STRING lid) := EMBED(mysql : server(myServer),user(myUser),database(myDB))
+  CALL testSP(?);
+ENDEMBED;
+
 dataset(childrec) testMySQLStringParam(string filter) := EMBED(mysql : server(myServer),user(myUser),database(myDB))
   SELECT * from tbl1 where name = ?;
 ENDEMBED;
@@ -163,22 +195,25 @@ sequential (
   initializeNulls(),
   initializeUtf8(),
   PARALLEL (
-  OUTPUT(testMySQLDS()),
-  COUNT(testMySQLDS2()),
-  OUTPUT(testMySQLDS3(), {name}),
-  OUTPUT(testMySQLRow().name),
-  OUTPUT(testMySQLParms('name1', 1, true, 1.2, 3.4, D'aa55aa55', U'Straße', U'Straße')),
-  OUTPUT(testMySQLString()),
-  OUTPUT(testMySQLStringParam(testMySqlString())),
-  OUTPUT(testMySQLDSParam(PROJECT(init, extractName(LEFT)))),
-    OUTPUT(testMySQLInt()+testMySQLInt()),
-  OUTPUT(testMySQLBool()),
-  OUTPUT(testMySQLReal8()),
-  OUTPUT(testMySQLReal4()),
-  OUTPUT(testMySQLData()),
-  OUTPUT(testMySQLUtf8()),
-  OUTPUT(testMySQLUnicode()),
-  OUTPUT(testMySQLDateTime()),
-  OUTPUT(testMySQLTransform())
+      OUTPUT(testMySQLDS()),
+      COUNT(testMySQLDS2()),
+      OUTPUT(SORT(testMySQLDS2a(), name)),
+      OUTPUT(testMySQLDS3(), {name}),
+      OUTPUT(testMySQLRow().name),
+      OUTPUT(testMySQLParms('name1', 1, true, 1.2, 3.4, D'aa55aa55', U'Straße', U'Straße')),
+      OUTPUT(testMySQLString()),
+      OUTPUT(testMySQLStringParam(testMySqlString())),
+      OUTPUT(testMySQLDSParam(PROJECT(init, extractName(LEFT)))),
+      OUTPUT(testMySQLInt()+testMySQLInt()),
+      OUTPUT(testMySQLBool()),
+      OUTPUT(testMySQLReal8()),
+      OUTPUT(testMySQLReal4()),
+      OUTPUT(testMySQLData()),
+      OUTPUT(testMySQLUtf8()),
+      OUTPUT(testMySQLUnicode()),
+      OUTPUT(testMySQLDateTime()),
+      OUTPUT(testMySQLTransform()),
+      OUTPUT(testMySQLStoredProcedure()),
+      OUTPUT(testMySQLStoredProcedure2('name1'))
   )
 );