Merge pull request #8589 from taosdata/feature/query

Feature/query

src/client/src/tscSql.c

@@ -781,6 +781,16 @@ bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col) {
   return isNull(((char*) pSql->res.urow[col]) + row * pInfo->field.bytes, pInfo->field.type);
 }
 
+bool taos_is_update_query(TAOS_RES *res) {
+  SSqlObj *pSql = (SSqlObj *)res;
+  if (pSql == NULL || pSql->signature != pSql) {
+    return false;
+  }
+
+  SSqlCmd* pCmd = &pSql->cmd;
+  return ((pCmd->command >= TSDB_SQL_INSERT && pCmd->command <= TSDB_SQL_DROP_DNODE) || TSDB_SQL_RESET_CACHE == pCmd->command || TSDB_SQL_USE_DB == pCmd->command);
+}
+
 int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields) {
   int len = 0;
 
@@ -909,7 +919,6 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
 
   strtolower(pSql->sqlstr, sql);
 
-//  pCmd->curSql = NULL;
   if (NULL != pCmd->insertParam.pTableBlockHashList) {
     taosHashCleanup(pCmd->insertParam.pTableBlockHashList);
     pCmd->insertParam.pTableBlockHashList = NULL;
@@ -934,6 +943,17 @@ int taos_validate_sql(TAOS *taos, const char *sql) {
   return code;
 }
 
+void taos_reset_current_db(TAOS *taos) {
+  STscObj* pObj = (STscObj*) taos;
+  if (pObj == NULL || pObj->signature != pObj) {
+    return;
+  }
+
+  pthread_mutex_lock(&pObj->mutex);
+  memset(pObj->db, 0, tListLen(pObj->db));
+  pthread_mutex_unlock(&pObj->mutex);
+}
+
 void loadMultiTableMetaCallback(void *param, TAOS_RES *res, int code) {
   SSqlObj* pSql = (SSqlObj*)taosAcquireRef(tscObjRef, (int64_t)param);
   if (pSql == NULL) {

src/inc/taos.h

@@ -175,11 +175,13 @@ DLL_EXPORT int taos_select_db(TAOS *taos, const char *db);
 DLL_EXPORT int taos_print_row(char *str, TAOS_ROW row, TAOS_FIELD *fields, int num_fields);
 DLL_EXPORT void taos_stop_query(TAOS_RES *res);
 DLL_EXPORT bool taos_is_null(TAOS_RES *res, int32_t row, int32_t col);
+DLL_EXPORT bool taos_is_update_query(TAOS_RES *res);
 DLL_EXPORT int taos_fetch_block(TAOS_RES *res, TAOS_ROW *rows);
-DLL_EXPORT int taos_validate_sql(TAOS *taos, const char *sql);
-
 DLL_EXPORT int* taos_fetch_lengths(TAOS_RES *res);
 
+DLL_EXPORT int taos_validate_sql(TAOS *taos, const char *sql);
+DLL_EXPORT void taos_reset_current_db(TAOS *taos);
+
 // TAOS_RES   *taos_list_tables(TAOS *mysql, const char *wild);
 // TAOS_RES   *taos_list_dbs(TAOS *mysql, const char *wild);
 
@@ -192,7 +194,6 @@ DLL_EXPORT int taos_errno(TAOS_RES *tres);
 
 DLL_EXPORT void taos_query_a(TAOS *taos, const char *sql, void (*fp)(void *param, TAOS_RES *, int code), void *param);
 DLL_EXPORT void taos_fetch_rows_a(TAOS_RES *res, void (*fp)(void *param, TAOS_RES *, int numOfRows), void *param);
-//DLL_EXPORT void taos_fetch_row_a(TAOS_RES *res, void (*fp)(void *param, TAOS_RES *, TAOS_ROW row), void *param);
 
 typedef void (*TAOS_SUBSCRIBE_CALLBACK)(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code);
 DLL_EXPORT TAOS_SUB *taos_subscribe(TAOS* taos, int restart, const char* topic, const char *sql, TAOS_SUBSCRIBE_CALLBACK fp, void *param, int interval);

src/query/inc/qExecutor.h

@@ -355,16 +355,16 @@ enum OPERATOR_TYPE_E {
 
 typedef struct SOperatorInfo {
   uint8_t               operatorType;
-  bool                  blockingOptr;  // block operator or not
-  uint8_t               status;        // denote if current operator is completed
-  int32_t               numOfOutput;   // number of columns of the current operator results
-  char                 *name;          // name, used to show the query execution plan
-  void                 *info;          // extension attribution
+  bool                  blockingOptr;   // block operator or not
+  uint8_t               status;         // denote if current operator is completed
+  int32_t               numOfOutput;    // number of columns of the current operator results
+  char                 *name;           // name, used to show the query execution plan
+  void                 *info;           // extension attribution
   SExprInfo            *pExpr;
   SQueryRuntimeEnv     *pRuntimeEnv;
 
-  struct SOperatorInfo **upstream;     // upstream pointer list
-  int32_t               numOfUpstream; // number of upstream. The value is always ONE expect for join operator
+  struct SOperatorInfo **upstream;      // upstream pointer list
+  int32_t               numOfUpstream;  // number of upstream. The value is always ONE expect for join operator
   __operator_fn_t       exec;
   __optr_cleanup_fn_t   cleanup;
 } SOperatorInfo;

src/query/src/qExecutor.c

@@ -4231,7 +4231,7 @@ static int32_t doCopyToSDataBlock(SQueryRuntimeEnv* pRuntimeEnv, SGroupResInfo*
   int32_t start = 0;
   int32_t step = -1;
 
-  qDebug("QInfo:0x%"PRIx64" start to copy data from windowResInfo to output buf", GET_QID(pRuntimeEnv));
+  qDebug("QInfo:0x%"PRIx64" start to copy data from resultrowInfo to output buf", GET_QID(pRuntimeEnv));
   assert(orderType == TSDB_ORDER_ASC || orderType == TSDB_ORDER_DESC);
 
   if (orderType == TSDB_ORDER_ASC) {
@@ -5220,7 +5220,6 @@ SOperatorInfo* createTableScanOperator(void* pTsdbQueryHandle, SQueryRuntimeEnv*
   pInfo->reverseTimes = 0;
   pInfo->order        = pRuntimeEnv->pQueryAttr->order.order;
   pInfo->current      = 0;
-//  pInfo->prevGroupId  = -1;
 
   SOperatorInfo* pOperator = calloc(1, sizeof(SOperatorInfo));
   pOperator->name         = "TableScanOperator";

src/query/src/qUtil.c

@@ -448,7 +448,9 @@ int32_t tsDescOrder(const void* p1, const void* p2) {
   }
 }
 
-void orderTheResultRows(SQueryRuntimeEnv* pRuntimeEnv) {
+void
+
+orderTheResultRows(SQueryRuntimeEnv* pRuntimeEnv) {
   __compar_fn_t  fn = NULL;
   if (pRuntimeEnv->pQueryAttr->order.order == TSDB_ORDER_ASC) {
     fn = tsAscOrder;

src/tsdb/src/tsdbRead.c

@@ -3452,9 +3452,12 @@ void filterPrepare(void* expr, void* param) {
      int dummy = -1;
      SHashObj *pObj = NULL;
      if (pInfo->sch.colId == TSDB_TBNAME_COLUMN_INDEX) {
-        pObj = taosHashInit(256, taosGetDefaultHashFunction(pInfo->sch.type), true, false);
         SArray *arr = (SArray *)(pCond->arr);
-        for (size_t i = 0; i < taosArrayGetSize(arr); i++) {
+
+       size_t size = taosArrayGetSize(arr);
+       pObj = taosHashInit(size * 2, taosGetDefaultHashFunction(pInfo->sch.type), true, false);
+
+        for (size_t i = 0; i < size; i++) {
           char* p = taosArrayGetP(arr, i);
           strntolower_s(varDataVal(p), varDataVal(p), varDataLen(p));
           taosHashPut(pObj, varDataVal(p), varDataLen(p), &dummy, sizeof(dummy));
@@ -3462,12 +3465,14 @@ void filterPrepare(void* expr, void* param) {
      } else {
        buildFilterSetFromBinary((void **)&pObj, pCond->pz, pCond->nLen);
      }
+
      pInfo->q = (char *)pObj;
   } else if (pCond != NULL) {
     uint32_t size = pCond->nLen * TSDB_NCHAR_SIZE;
     if (size < (uint32_t)pSchema->bytes) {
       size = pSchema->bytes;
     }
+
     // to make sure tonchar does not cause invalid write, since the '\0' needs at least sizeof(wchar_t) space.
     pInfo->q = calloc(1, size + TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
     tVariantDump(pCond, pInfo->q, pSchema->type, true);
@@ -3615,7 +3620,7 @@ SArray* createTableGroup(SArray* pTableList, STSchema* pTagSchema, SColIndex* pC
   return pTableGroup;
 }
 
-int32_t tsdbQuerySTableByTagCond(STsdbRepo* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len, 
+int32_t tsdbQuerySTableByTagCond(STsdbRepo* tsdb, uint64_t uid, TSKEY skey, const char* pTagCond, size_t len,
                                  STableGroupInfo* pGroupInfo, SColIndex* pColIndex, int32_t numOfCols) {
   if (tsdbRLockRepoMeta(tsdb) < 0) goto _error;
 
@@ -3677,19 +3682,19 @@ int32_t tsdbQuerySTableByTagCond(STsdbRepo* tsdb, uint64_t uid, TSKEY skey, cons
   } END_TRY
 
   void *filterInfo = NULL;
-  
+
   ret = filterInitFromTree(expr, &filterInfo, 0);
   if (ret != TSDB_CODE_SUCCESS) {
     terrno = ret;
     goto _error;
   }
-  
+
   tsdbQueryTableList(pTable, res, filterInfo);
 
   filterFreeInfo(filterInfo);
 
   tExprTreeDestroy(expr, NULL);
-  
+
   pGroupInfo->numOfTables = (uint32_t)taosArrayGetSize(res);
   pGroupInfo->pGroupList  = createTableGroup(res, pTagSchema, pColIndex, numOfCols, skey);
 
@@ -3876,7 +3881,7 @@ void tsdbDestroyTableGroup(STableGroupInfo *pGroupList) {
 
 static FORCE_INLINE int32_t tsdbGetTagDataFromId(void *param, int32_t id, void **data) {
   STable* pTable = (STable*)(SL_GET_NODE_DATA((SSkipListNode *)param));
-  
+
   if (id == TSDB_TBNAME_COLUMN_INDEX) {
     *data = TABLE_NAME(pTable);
   } else {
@@ -3909,7 +3914,7 @@ static void queryIndexedColumn(SSkipList* pSkipList, void* filterInfo, SArray* r
       iter = tSkipListCreateIterFromVal(pSkipList, startVal, pSkipList->type, TSDB_ORDER_DESC);
       FILTER_CLR_FLAG(order, TSDB_ORDER_DESC);
     }
-    
+
     while (tSkipListIterNext(iter)) {
       SSkipListNode *pNode = tSkipListIterGet(iter);
 
@@ -3918,7 +3923,7 @@ static void queryIndexedColumn(SSkipList* pSkipList, void* filterInfo, SArray* r
         filterSetColFieldData(filterInfo, pNode, tsdbGetTagDataFromId);
         all = filterExecute(filterInfo, 1, &addToResult, NULL, 0);
       }
-      
+
       char *pData = SL_GET_NODE_DATA(pNode);
 
       tsdbDebug("filter index column, table:%s, result:%d", ((STable *)pData)->name->data, all);
@@ -3950,7 +3955,7 @@ static void queryIndexlessColumn(SSkipList* pSkipList, void* filterInfo, SArray*
     SSkipListNode *pNode = tSkipListIterGet(iter);
 
     filterSetColFieldData(filterInfo, pNode, tsdbGetTagDataFromId);
-    
+
     char *pData = SL_GET_NODE_DATA(pNode);
 
     bool all = filterExecute(filterInfo, 1, &addToResult, NULL, 0);
@@ -3958,7 +3963,7 @@ static void queryIndexlessColumn(SSkipList* pSkipList, void* filterInfo, SArray*
     if (all || (addToResult && *addToResult)) {
       STableKeyInfo info = {.pTable = (void*)pData, .lastKey = TSKEY_INITIAL_VAL};
       taosArrayPush(res, &info);
-    }    
+    }
   }
 
   tfree(addToResult);
@@ -3971,9 +3976,9 @@ static int32_t tsdbQueryTableList(STable* pTable, SArray* pRes, void* filterInfo
   STSchema*   pTSSchema = pTable->tagSchema;
   bool indexQuery = false;
   SSkipList *pSkipList = pTable->pIndex;
-  
+
   filterIsIndexedColumnQuery(filterInfo, pTSSchema->columns->colId, &indexQuery);
-  
+
   if (indexQuery) {
     queryIndexedColumn(pSkipList, filterInfo, pRes);
   } else {