Merge pull request #10050 from taosdata/feature/3.0_liaohj

[td-11818] Add the implementation of api in taos.h

include/client/taos.h

@@ -92,14 +92,16 @@ typedef struct taosField {
 
 typedef void (*__taos_async_fn_t)(void *param, TAOS_RES *, int code);
 
-DLL_EXPORT void  taos_cleanup(void);
-DLL_EXPORT int   taos_options(TSDB_OPTION option, const void *arg, ...);
-DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen, const char *db, int dbLen, uint16_t port);
-DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
-DLL_EXPORT void  taos_close(TAOS *taos);
+typedef struct tmq_t tmq_t;
+typedef struct tmq_conf_t tmq_conf_t;
+typedef struct tmq_list_t tmq_list_t;
 
-const char *taos_data_type(int type);
+typedef struct tmq_message_t       tmq_message_t;
+typedef struct tmq_message_topic_t tmq_message_topic_t;
+typedef struct tmq_message_tb_t    tmq_message_tb_t;
+typedef struct tmq_tb_iter_t       tmq_tb_iter_t;
+typedef struct tmq_message_col_t   tmq_message_col_t;
+typedef struct tmq_col_iter_t      tmq_col_iter_t;
 
 typedef struct TAOS_BIND {
   int            buffer_type;
@@ -134,6 +136,15 @@ typedef struct TAOS_MULTI_BIND {
   int            num;
 } TAOS_MULTI_BIND;
 
+DLL_EXPORT void  taos_cleanup(void);
+DLL_EXPORT int   taos_options(TSDB_OPTION option, const void *arg, ...);
+DLL_EXPORT TAOS *taos_connect(const char *ip, const char *user, const char *pass, const char *db, uint16_t port);
+DLL_EXPORT TAOS *taos_connect_l(const char *ip, int ipLen, const char *user, int userLen, const char *pass, int passLen, const char *db, int dbLen, uint16_t port);
+DLL_EXPORT TAOS *taos_connect_auth(const char *ip, const char *user, const char *auth, const char *db, uint16_t port);
+DLL_EXPORT void  taos_close(TAOS *taos);
+
+const char *taos_data_type(int type);
+
 DLL_EXPORT TAOS_STMT *taos_stmt_init(TAOS *taos);
 DLL_EXPORT int        taos_stmt_prepare(TAOS_STMT *stmt, const char *sql, unsigned long length);
 DLL_EXPORT int        taos_stmt_set_tbname_tags(TAOS_STMT* stmt, const char* name, TAOS_BIND* tags);
@@ -192,16 +203,6 @@ DLL_EXPORT void taos_close_stream(TAOS_STREAM *tstr);
 
 DLL_EXPORT int taos_load_table_info(TAOS *taos, const char* tableNameList);
 DLL_EXPORT TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int protocol, int precision);
-typedef struct tmq_t tmq_t;
-typedef struct tmq_conf_t tmq_conf_t;
-typedef struct tmq_list_t tmq_list_t;
-
-typedef struct tmq_message_t       tmq_message_t;
-typedef struct tmq_message_topic_t tmq_message_topic_t;
-typedef struct tmq_message_tb_t    tmq_message_tb_t;
-typedef struct tmq_tb_iter_t       tmq_tb_iter_t;
-typedef struct tmq_message_col_t   tmq_message_col_t;
-typedef struct tmq_col_iter_t      tmq_col_iter_t;
 
 DLL_EXPORT tmq_list_t* tmq_list_new();
 DLL_EXPORT int32_t tmq_list_append(tmq_list_t*, char*);

include/common/common.h

@@ -38,6 +38,12 @@
 //  int16_t bytes;
 //} SSchema;
 
+typedef struct {
+  uint32_t  numOfTables;
+  SArray   *pGroupList;
+  SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
+} STableGroupInfo;
+
 typedef struct SColumnDataAgg {
   int16_t colId;
   int64_t sum;
@@ -57,17 +63,12 @@ typedef struct SDataBlockInfo {
 
 typedef struct SConstantItem {
   SColumnInfo info;
-  int32_t     startIndex;    // run-length-encoding to save the space for multiple rows
-  int32_t     endIndex;
+  int32_t     startRow;    // run-length-encoding to save the space for multiple rows
+  int32_t     endRow;
   SVariant    value;
 } SConstantItem;
 
-typedef struct {
-  uint32_t  numOfTables;
-  SArray   *pGroupList;
-  SHashObj *map;  // speedup acquire the tableQueryInfo by table uid
-} STableGroupInfo;
-
+// info.numOfCols = taosArrayGetSize(pDataBlock) + taosArrayGetSize(pConstantList);
 typedef struct SSDataBlock {
   SColumnDataAgg *pBlockAgg;
   SArray         *pDataBlock;    // SArray<SColumnInfoData>
@@ -75,9 +76,12 @@ typedef struct SSDataBlock {
   SDataBlockInfo  info;
 } SSDataBlock;
 
+// pBlockAgg->numOfNull == info.rows, all data are null
+// pBlockAgg->numOfNull == 0, no data are null.
 typedef struct SColumnInfoData {
-  SColumnInfo info;     // TODO filter info needs to be removed
-  char       *pData;    // the corresponding block data in memory
+  SColumnInfo info;      // TODO filter info needs to be removed
+  char       *nullbitmap;//
+  char       *pData;     // the corresponding block data in memory
 } SColumnInfoData;
 
 static FORCE_INLINE int32_t tEncodeDataBlock(void** buf, const SSDataBlock* pBlock) {

include/common/tmsg.h

@@ -289,6 +289,7 @@ static FORCE_INLINE void* taosDecodeSEpSet(void* buf, SEpSet* pEp) {
   }
   return buf;
 }
+
 typedef struct {
   int32_t acctId;
   int64_t clusterId;
@@ -296,6 +297,7 @@ typedef struct {
   int8_t  superUser;
   int8_t  align[3];
   SEpSet  epSet;
+  char    sVersion[128];
 } SConnectRsp;
 
 typedef struct {

source/client/inc/clientInt.h

@@ -114,6 +114,7 @@ typedef struct STscObj {
   char             user[TSDB_USER_LEN];
   char             pass[TSDB_PASSWORD_LEN];
   char             db[TSDB_DB_FNAME_LEN];
+  char             ver[128];
   int32_t          acctId;
   uint32_t         connId;
   int32_t          connType;

source/client/src/clientImpl.c

@@ -218,12 +218,10 @@ int32_t getPlan(SRequestObj* pRequest, SQueryNode* pQueryNode, SQueryDag** pDag,
 
   if (pQueryNode->type == TSDB_SQL_SELECT) {
     setResSchemaInfo(&pRequest->body.resInfo, pSchema, numOfCols);
-    tfree(pSchema);
     pRequest->type = TDMT_VND_QUERY;
-  } else {
-    tfree(pSchema);
   }
 
+  tfree(pSchema);
   return code;
 }
 

source/client/src/clientMain.c

@@ -275,3 +275,70 @@ int taos_affected_rows(TAOS_RES *res) {
 }
 
 int taos_result_precision(TAOS_RES *res) { return TSDB_TIME_PRECISION_MILLI; }
+
+int taos_select_db(TAOS *taos, const char *db) {
+  STscObj *pObj = (STscObj *)taos;
+  if (pObj == NULL) {
+    terrno = TSDB_CODE_TSC_DISCONNECTED;
+    return TSDB_CODE_TSC_DISCONNECTED;
+  }
+
+  if (db == NULL || strlen(db) == 0) {
+    terrno = TSDB_CODE_TSC_INVALID_INPUT;
+    return terrno;
+  }
+
+  char sql[256] = {0};
+  snprintf(sql, tListLen(sql), "use %s", db);
+
+  TAOS_RES* pRequest = taos_query(taos, sql);
+  int32_t code = taos_errno(pRequest);
+
+  taos_free_result(pRequest);
+  return code;
+}
+
+void taos_stop_query(TAOS_RES *res) {
+  if (res == NULL) {
+    return;
+  }
+
+  SRequestObj* pRequest = (SRequestObj*) res;
+  int32_t numOfFields = taos_num_fields(pRequest);
+
+  // It is not a query, no need to stop.
+  if (numOfFields == 0) {
+    return;
+  }
+
+//  scheduleCancelJob(pRequest->body.pQueryJob);
+}
+
+bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
+  return false;
+}
+
+int  taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows) {
+  return 0;
+}
+
+int  taos_validate_sql(TAOS *taos, const char *sql) {
+  return true;
+}
+
+const char *taos_get_server_info(TAOS *taos) {
+  if (taos == NULL) {
+    return NULL;
+  }
+
+  STscObj* pTscObj = (STscObj*) taos;
+  return pTscObj->ver;
+}
+
+void taos_query_a(TAOS *taos, const char *sql, __taos_async_fn_t fp, void *param) {
+  // TODO
+}
+
+void taos_fetch_rows_a(TAOS_RES *res, __taos_async_fn_t fp, void *param) {
+  // TODO
+}
\ No newline at end of file

source/client/src/clientMsgHandler.c

@@ -67,6 +67,7 @@ int processConnectRsp(void* param, const SDataBuf* pMsg, int32_t code) {
 
   pTscObj->connId = pConnect->connId;
   pTscObj->acctId = pConnect->acctId;
+  tstrncpy(pTscObj->ver, pConnect->sVersion, tListLen(pTscObj->ver));
 
   // update the appInstInfo
   pTscObj->pAppInfo->clusterId = pConnect->clusterId;

source/common/src/tep.c

 #include "tep.h"
+#include "common.h"
 #include "tglobal.h"
 #include "tlockfree.h"
 
@@ -59,3 +60,99 @@ SEpSet getEpSet_s(SCorEpSet *pEpSet) {
   return ep;
 }
 
+bool colDataIsNull(const SColumnInfoData* pColumnInfoData, uint32_t totalRows, uint32_t row, SColumnDataAgg* pColAgg) {
+  if (pColAgg != NULL) {
+    if (pColAgg->numOfNull == totalRows) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return true;
+    } else if (pColAgg->numOfNull == 0) {
+      ASSERT(pColumnInfoData->nullbitmap == NULL);
+      return false;
+    }
+  }
+
+  if (pColumnInfoData->nullbitmap == NULL) {
+    return false;
+  }
+
+  uint8_t v = (pColumnInfoData->nullbitmap[row>>3] & (1<<(8 - (row&0x07))));
+  return (v == 1);
+}
+
+bool colDataIsNull_f(const char* bitmap, uint32_t row) {
+  return (bitmap[row>>3] & (1<<(8 - (row&0x07))));
+}
+
+void colDataSetNull_f(char* bitmap, uint32_t row) { // TODO
+  return;
+}
+
+void* colDataGet(const SColumnInfoData* pColumnInfoData, uint32_t row) {
+  if (IS_VAR_DATA_TYPE(pColumnInfoData->info.type)) {
+    uint32_t offset = ((uint32_t*)pColumnInfoData->pData)[row];
+    return (char*)(pColumnInfoData->pData) + offset; // the first part is the pointer to the true binary data
+  } else {
+    return (char*)(pColumnInfoData->pData) + (row * pColumnInfoData->info.bytes);
+  }
+}
+
+int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull) {
+  ASSERT(pColumnInfoData != NULL);
+
+  if (isNull) {
+    // TODO set null value in the nullbitmap
+    return 0;
+  }
+
+  int32_t type = pColumnInfoData->info.type;
+  if (IS_VAR_DATA_TYPE(type)) {
+    // TODO continue append var_type
+  } else {
+    char* p = pColumnInfoData->pData + pColumnInfoData->info.bytes * currentRow;
+    switch(type) {
+      case TSDB_DATA_TYPE_TINYINT:
+      case TSDB_DATA_TYPE_UTINYINT: {*(int8_t*) p = *(int8_t*) pData;break;}
+      default:
+        assert(0);
+    }
+
+  }
+
+  return 0;
+}
+
+size_t colDataGetCols(const SSDataBlock* pBlock) {
+  ASSERT(pBlock);
+
+  size_t constantCols = (pBlock->pConstantList != NULL)? taosArrayGetSize(pBlock->pConstantList):0;
+  ASSERT( pBlock->info.numOfCols == taosArrayGetSize(pBlock->pDataBlock) + constantCols);
+  return pBlock->info.numOfCols;
+}
+
+size_t colDataGetRows(const SSDataBlock* pBlock) {
+  return pBlock->info.rows;
+}
+
+int32_t colDataUpdateTsWindow(SSDataBlock* pDataBlock) {
+  if (pDataBlock == NULL || pDataBlock->info.rows <= 0) {
+    return 0;
+  }
+
+  if (pDataBlock->info.numOfCols <= 0) {
+    return -1;
+  }
+
+  SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, 0);
+  if (pColInfoData->info.type != TSDB_DATA_TYPE_TIMESTAMP) {
+    return 0;
+  }
+
+  ASSERT(pColInfoData->nullbitmap == NULL);
+  pDataBlock->info.window.skey = *(TSKEY*) colDataGet(pColInfoData, 0);
+  pDataBlock->info.window.ekey = *(TSKEY*) colDataGet(pColInfoData, (pDataBlock->info.rows - 1));
+  return 0;
+}
+
+
+
+

source/dnode/mnode/impl/src/mndMnode.c

@@ -237,13 +237,11 @@ void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
     if (pIter == NULL) break;
     if (pObj->pDnode == NULL) break;
 
-    pEpSet->eps[pEpSet->numOfEps].port = htons(pObj->pDnode->port);
-    memcpy(pEpSet->eps[pEpSet->numOfEps].fqdn, pObj->pDnode->fqdn, TSDB_FQDN_LEN);
     if (pObj->role == TAOS_SYNC_STATE_LEADER) {
       pEpSet->inUse = pEpSet->numOfEps;
     }
 
-    pEpSet->numOfEps++;
+    addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, htons(pObj->pDnode->port));
     sdbRelease(pSdb, pObj);
   }
 }

source/dnode/mnode/impl/src/mndProfile.c

@@ -14,14 +14,15 @@
  */
 
 #define _DEFAULT_SOURCE
+#include "tglobal.h"
 #include "mndProfile.h"
-#include "mndConsumer.h"
+//#include "mndConsumer.h"
 #include "mndDb.h"
 #include "mndMnode.h"
 #include "mndShow.h"
-#include "mndTopic.h"
+//#include "mndTopic.h"
 #include "mndUser.h"
-#include "mndVgroup.h"
+//#include "mndVgroup.h"
 
 #define QUERY_ID_SIZE 20
 #define QUERY_OBJ_ID_SIZE 18
@@ -230,10 +231,12 @@ static int32_t mndProcessConnectReq(SMnodeMsg *pReq) {
     goto CONN_OVER;
   }
 
-  pRsp->acctId = htonl(pUser->acctId);
+  pRsp->acctId    = htonl(pUser->acctId);
   pRsp->superUser = pUser->superUser;
   pRsp->clusterId = htobe64(pMnode->clusterId);
-  pRsp->connId = htonl(pConn->id);
+  pRsp->connId    = htonl(pConn->id);
+
+  snprintf(pRsp->sVersion, tListLen(pRsp->sVersion), "ver:%s\nbuild:%s\ngitinfo:%s", version, buildinfo, gitinfo);
   mndGetMnodeEpSet(pMnode, &pRsp->epSet);
 
   pReq->contLen = sizeof(SConnectRsp);

source/libs/planner/src/physicalPlanJson.c

@@ -1125,8 +1125,6 @@ int32_t subPlanToString(const SSubplan* subplan, char** str, int32_t* len) {
   *str = cJSON_Print(json);
   cJSON_Delete(json);
 
-//  printf("====Physical plan:====\n");
-//  printf("%s\n", *str);
   *len = strlen(*str) + 1;
   return TSDB_CODE_SUCCESS;
 }

source/libs/planner/src/planner.c

@@ -40,7 +40,8 @@ void qDestroyQueryDag(struct SQueryDag* pDag) {
   tfree(pDag);
 }
 
-int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pResSchema, int32_t* numOfCols, SArray* pNodeList, uint64_t requestId) {
+int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag, SSchema** pResSchema, int32_t* numOfCols, SArray* pNodeList,
+    uint64_t requestId) {
   SQueryPlanNode* pLogicPlan;
   int32_t code = createQueryPlan(pNode, &pLogicPlan);
   if (TSDB_CODE_SUCCESS != code) {
@@ -49,9 +50,10 @@ int32_t qCreateQueryDag(const struct SQueryNode* pNode, struct SQueryDag** pDag,
   }
 
   if (pLogicPlan->info.type != QNODE_MODIFY) {
-//   char* str = NULL;
-//   queryPlanToString(pLogicPlan, &str);
-//   printf("%s\n", str);
+    char* str = NULL;
+    queryPlanToString(pLogicPlan, &str);
+    qDebug("reqId:0x%"PRIx64": %s", requestId, str);
+    tfree(str);
   }
 
   code = optimizeQueryPlan(pLogicPlan);

source/libs/scheduler/src/scheduler.c

@@ -1194,17 +1194,18 @@ int32_t schLaunchTask(SSchJob *pJob, SSchTask *pTask) {
     
     code = atomic_load_32(&pJob->errCode);
     SCH_ERR_RET(code);
-    
     SCH_RET(TSDB_CODE_SCH_STATUS_ERROR);
   }
   
   SSubplan *plan = pTask->plan;
 
-  if (NULL == pTask->msg) {
+  if (NULL == pTask->msg) { // TODO add more detailed reason for failure
     code = qSubPlanToString(plan, &pTask->msg, &pTask->msgLen);
     if (TSDB_CODE_SUCCESS != code || NULL == pTask->msg || pTask->msgLen <= 0) {
-      SCH_TASK_ELOG("subplanToString error, code:%x, msg:%p, len:%d", code, pTask->msg, pTask->msgLen);
+      SCH_TASK_ELOG("failed to create physical plan, code:%s, msg:%p, len:%d", tstrerror(code), pTask->msg, pTask->msgLen);
       SCH_ERR_JRET(code);
+    } else {
+      SCH_TASK_DLOG(" ===physical plan=== len:%d, %s", pTask->msgLen, pTask->msg);
     }
   }
   
@@ -1218,13 +1219,10 @@ int32_t schLaunchTask(SSchJob *pJob, SSchTask *pTask) {
   }
 
   SCH_ERR_JRET(schBuildAndSendMsg(pJob, pTask, NULL, plan->msgType));
-  
   return TSDB_CODE_SUCCESS;
 
 _return:
-
   SCH_ERR_RET(schProcessOnTaskFailure(pJob, pTask, code));
-  
   SCH_RET(code);
 }
 

tools/shell/src/shellEngine.c

@@ -360,6 +360,7 @@ void shellRunCommandOnServer(TAOS *con, char command[]) {
     }
   } else {
     int num_rows_affacted = taos_affected_rows(pSql);
+    taos_free_result(pSql);
     et = taosGetTimestampUs();
     printf("Query OK, %d of %d row(s) in database (%.6fs)\n", num_rows_affacted, num_rows_affacted, (et - st) / 1E6);
   }