[TD-489]

src/client/inc/tsclient.h

@@ -317,6 +317,7 @@ typedef struct SSqlObj {
   SRpcIpSet        ipList;
   char             freed : 4;
   char             listed : 4;
+  uint32_t         insertType;
   tsem_t           rspSem;
   SSqlCmd          cmd;
   SSqlRes          res;
@@ -402,6 +403,7 @@ void tscCloseTscObj(STscObj *pObj);
 
 TAOS *taos_connect_a(char *ip, char *user, char *pass, char *db, uint16_t port, void (*fp)(void *, TAOS_RES *, int),
                      void *param, void **taos);
+void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ;
 
 void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen);
 

src/client/src/tscAsync.c

@@ -482,6 +482,14 @@ void tscTableMetaCallBack(void *param, TAOS_RES *res, int code) {
         }
       } else {
         code = tsParseSql(pSql, false);
+        if ((pQueryInfo->type & TSDB_QUERY_TYPE_STMT_INSERT) == TSDB_QUERY_TYPE_STMT_INSERT) {
+          STableMetaInfo* pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0);
+          code = tscGetTableMeta(pSql, pTableMetaInfo);
+          assert(code == TSDB_CODE_SUCCESS && pTableMetaInfo->pTableMeta != NULL);
+          (*pSql->fp)(pSql->param, NULL, code);
+          return;
+        }
+        
         if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
       }
     }

src/client/src/tscParseInsert.c

@@ -1312,6 +1312,7 @@ int tsParseInsertSql(SSqlObj *pSql) {
   tscGetQueryInfoDetailSafely(pCmd, pCmd->clauseIndex, &pQueryInfo);
 
   TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT);
+  TSDB_QUERY_SET_TYPE(pQueryInfo->type, pSql->insertType);
 
   sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL);
   if (sToken.type != TK_INTO) {
@@ -1339,7 +1340,7 @@ int tsParseSql(SSqlObj *pSql, bool initialParse) {
      * Set the fp before parse the sql string, in case of getTableMeta failed, in which
      * the error handle callback function can rightfully restore the user-defined callback function (fp).
      */
-    if (initialParse) {
+    if (initialParse && (pSql->insertType != TSDB_QUERY_TYPE_STMT_INSERT)) {
       pSql->fetchFp = pSql->fp;
       pSql->fp = (void(*)())tscHandleMultivnodeInsert;
     }

src/client/src/tscPrepare.c

@@ -12,6 +12,7 @@
  * You should have received a copy of the GNU Affero General Public License
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
+#include "os.h"
 
 #include "taos.h"
 #include "tsclient.h"
@@ -20,6 +21,7 @@
 #include "taosmsg.h"
 #include "tstrbuild.h"
 #include "tscLog.h"
+#include "tscSubquery.h"
 
 int tsParseInsertSql(SSqlObj *pSql);
 int taos_query_imp(STscObj* pObj, SSqlObj* pSql);
@@ -262,7 +264,11 @@ static char* normalStmtBuildSql(STscStmt* stmt) {
 
 static int doBindParam(char* data, SParamInfo* param, TAOS_BIND* bind) {
   if (bind->is_null != NULL && *(bind->is_null)) {
-    setNull(data, param->type, param->bytes);
+    if (param->type == TSDB_DATA_TYPE_BINARY || param->type == TSDB_DATA_TYPE_NCHAR) {
+      setVardataNull(data + param->offset, param->type);
+    } else {
+      setNull(data + param->offset, param->type, param->bytes);
+    }
     return TSDB_CODE_SUCCESS;
   }
 
@@ -297,14 +303,17 @@ static int doBindParam(char* data, SParamInfo* param, TAOS_BIND* bind) {
         return TSDB_CODE_INVALID_VALUE;
       }
       size = (short)*bind->length;
-      break;
+      STR_WITH_SIZE_TO_VARSTR(data + param->offset, bind->buffer, size);
+      return TSDB_CODE_SUCCESS;
     
-    case TSDB_DATA_TYPE_NCHAR:
-      if (!taosMbsToUcs4(bind->buffer, *bind->length, data + param->offset, param->bytes, NULL)) {
+    case TSDB_DATA_TYPE_NCHAR: {
+      size_t output = 0;
+      if (!taosMbsToUcs4(bind->buffer, *bind->length, varDataVal(data + param->offset), param->bytes - VARSTR_HEADER_SIZE, &output)) {
         return TSDB_CODE_INVALID_VALUE;
-      }
+      }      
+      varDataSetLen(data + param->offset, output);
       return TSDB_CODE_SUCCESS;
-
+    }
     default:
       assert(false);
       return TSDB_CODE_INVALID_VALUE;
@@ -383,14 +392,6 @@ static int insertStmtAddBatch(STscStmt* stmt) {
   return TSDB_CODE_SUCCESS;
 }
 
-static int insertStmtPrepare(STscStmt* stmt) {
-  SSqlObj *pSql = stmt->pSql;
-  pSql->cmd.numOfParams = 0;
-  pSql->cmd.batchSize = 0;
-
-  return tsParseInsertSql(pSql);
-}
-
 static int insertStmtReset(STscStmt* pStmt) {
   SSqlCmd* pCmd = &pStmt->pSql->cmd;
   if (pCmd->batchSize > 2) {
@@ -451,14 +452,16 @@ static int insertStmtExecute(STscStmt* stmt) {
   
   pRes->qhandle = 0;
 
+  pSql->insertType = 0;
+  pSql->fetchFp    = waitForQueryRsp;
+  pSql->fp         = (void(*)())tscHandleMultivnodeInsert;
+
   tscDoQuery(pSql);
 
-  // tscTrace("%p SQL result:%d, %s pObj:%p", pSql, pRes->code, taos_errstr(taos), pObj);
-  if (pRes->code != TSDB_CODE_SUCCESS) {
-    tscPartiallyFreeSqlObj(pSql);
-  }
+  // wait for the callback function to post the semaphore
+  tsem_wait(&pSql->rspSem);
+  return pSql->res.code;
 
-  return pRes->code;
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -478,6 +481,7 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) {
     tscError("failed to allocate memory for statement");
     return NULL;
   }
+  pStmt->taos = pObj;
 
   SSqlObj* pSql = calloc(1, sizeof(SSqlObj));
   if (pSql == NULL) {
@@ -488,8 +492,10 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) {
   }
 
   tsem_init(&pSql->rspSem, 0, 0);
-  pSql->signature = pSql;
-  pSql->pTscObj = pObj;
+  pSql->signature     = pSql;
+  pSql->pTscObj       = pObj;
+  pSql->pTscObj->pSql = pSql;
+  pSql->maxRetry      = TSDB_MAX_REPLICA_NUM;
 
   pStmt->pSql = pSql;
   return pStmt;
@@ -497,22 +503,55 @@ TAOS_STMT* taos_stmt_init(TAOS* taos) {
 
 int taos_stmt_prepare(TAOS_STMT* stmt, const char* sql, unsigned long length) {
   STscStmt* pStmt = (STscStmt*)stmt;
-  if (length == 0) {
-    length = strlen(sql);
+
+  if (stmt == NULL || pStmt->taos == NULL || pStmt->pSql == NULL) {
+    terrno = TSDB_CODE_DISCONNECTED;
+    return TSDB_CODE_DISCONNECTED;
+  }
+
+  SSqlObj* pSql = pStmt->pSql;
+  size_t   sqlLen = strlen(sql);
+  
+  //doAsyncQuery(pObj, pSql, waitForQueryRsp, taos, sqlstr, sqlLen);
+  SSqlCmd *pCmd    = &pSql->cmd;
+  SSqlRes *pRes    = &pSql->res;
+  pSql->param      = (void*)pStmt->taos;
+  pSql->fp         = waitForQueryRsp;
+  pSql->insertType = TSDB_QUERY_TYPE_STMT_INSERT;
+  
+  if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_DEFAULT_PAYLOAD_SIZE)) {
+    tscError("%p failed to malloc payload buffer", pSql);
+    return TSDB_CODE_CLI_OUT_OF_MEMORY;
   }
-  char* sqlstr = (char*)malloc(length + 1);
-  if (sqlstr == NULL) {
-    tscError("failed to malloc sql string buffer");
+  
+  pSql->sqlstr = realloc(pSql->sqlstr, sqlLen + 1);
+  
+  if (pSql->sqlstr == NULL) {
+    tscError("%p failed to malloc sql string buffer", pSql);
+    free(pCmd->payload);
     return TSDB_CODE_CLI_OUT_OF_MEMORY;
   }
-  memcpy(sqlstr, sql, length);
-  sqlstr[length] = 0;
-  strtolower(sqlstr, sqlstr);
+  
+  pRes->qhandle = 0;
+  pRes->numOfRows = 1;
+  
+  strtolower(pSql->sqlstr, sql);
+  tscDump("%p SQL: %s", pSql, pSql->sqlstr);
 
-  pStmt->pSql->sqlstr = sqlstr;
-  if (tscIsInsertData(sqlstr)) {
+  if (tscIsInsertData(pSql->sqlstr)) {  
     pStmt->isInsert = true;
-    return insertStmtPrepare(pStmt);
+    
+    pSql->cmd.numOfParams = 0;
+    pSql->cmd.batchSize   = 0;
+    
+    int32_t code = tsParseSql(pSql, true);
+    if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
+      // wait for the callback function to post the semaphore
+      tsem_wait(&pSql->rspSem);
+      return pSql->res.code;
+    }
+    
+    return code;
   }
 
   pStmt->isInsert = false;
@@ -574,7 +613,7 @@ int taos_stmt_execute(TAOS_STMT* stmt) {
     } else {
       tfree(pStmt->pSql->sqlstr);
       pStmt->pSql->sqlstr = sql;
-      ret = taos_query_imp(pStmt->taos, pStmt->pSql);
+      ret = taos_query(pStmt->taos, pStmt->pSql->sqlstr);
     }
   }
   return ret;

src/client/src/tscSql.c

@@ -264,7 +264,7 @@ int taos_query_imp(STscObj *pObj, SSqlObj *pSql) {
   return pRes->code;
 }
 
-static void waitForQueryRsp(void *param, TAOS_RES *tres, int code) {
+void waitForQueryRsp(void *param, TAOS_RES *tres, int code) {
   assert(param != NULL);
   SSqlObj *pSql = ((STscObj *)param)->pSql;
   

src/inc/taosdef.h

@@ -331,6 +331,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
 #define TSDB_QUERY_TYPE_TAG_FILTER_QUERY              0x400u
 #define TSDB_QUERY_TYPE_INSERT                        0x100u    // insert type
 #define TSDB_QUERY_TYPE_MULTITABLE_QUERY              0x200u
+#define TSDB_QUERY_TYPE_STMT_INSERT                   0x800u    // stmt insert type
 
 #define TSDB_QUERY_HAS_TYPE(x, _type)         (((x) & (_type)) != 0)
 #define TSDB_QUERY_SET_TYPE(x, _type)         ((x) |= (_type))