Merge pull request #3360 from taosdata/feature/query

Feature/query

src/client/inc/tsclient.h

@@ -221,20 +221,18 @@ typedef struct STableDataBlocks {
   SParamInfo *params;
 } STableDataBlocks;
 
-//typedef struct SDataBlockList {  // todo remove
-//  uint32_t           nSize;
-//  uint32_t           nAlloc;
-//  STableDataBlocks **pData;
-//} SDataBlockList;
-
 typedef struct SQueryInfo {
   int16_t          command;       // the command may be different for each subclause, so keep it seperately.
+  uint32_t         type;          // query/insert type
+  // TODO refactor
   char             intervalTimeUnit;
   char             slidingTimeUnit;
-  uint32_t         type;          // query/insert type
   STimeWindow      window;        // query time window
-  int64_t          intervalTime;  // aggregation time interval
+  int64_t          intervalTime;  // aggregation time window range
   int64_t          slidingTime;   // sliding window in mseconds
+  int64_t          intervalOffset;// start offset of each time window
+  int32_t          tz;            // query client timezone
+
   SSqlGroupbyExpr  groupbyExpr;   // group by tags info
   SArray *         colList;       // SArray<SColumn*>
   SFieldInfo       fieldsInfo;

src/query/inc/tsqlfunction.h

@@ -69,6 +69,15 @@ extern "C" {
 #define TSDB_FUNC_AVG_IRATE    33
 
 #define TSDB_FUNC_TID_TAG      34
+#define TSDB_FUNC_HISTOGRAM    35
+#define TSDB_FUNC_HLL          36
+#define TSDB_FUNC_MODE         37
+#define TSDB_FUNC_SAMPLE       38
+#define TSDB_FUNC_CEIL         39
+#define TSDB_FUNC_FLOOR        40
+#define TSDB_FUNC_ROUND        41
+#define TSDB_FUNC_MAVG         42
+#define TSDB_FUNC_CSUM         43
 
 #define TSDB_FUNCSTATE_SO           0x1u    // single output
 #define TSDB_FUNCSTATE_MO           0x2u    // dynamic number of output, not multinumber of output e.g., TOP/BOTTOM

src/query/src/qExecutor.c

@@ -35,8 +35,6 @@
  * forced to load primary column explicitly.
  */
 #define Q_STATUS_EQUAL(p, s)  (((p) & (s)) != 0)
-
-
 #define QUERY_IS_ASC_QUERY(q) (GET_FORWARD_DIRECTION_FACTOR((q)->order.order) == QUERY_ASC_FORWARD_STEP)
 
 #define IS_MASTER_SCAN(runtime)        ((runtime)->scanFlag == MASTER_SCAN)
@@ -1602,11 +1600,11 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int16_t order
     SColIndex* pIndex = &pSqlFuncMsg->colInfo;
 
     if (TSDB_COL_REQ_NULL(pIndex->flag)) {
-      pCtx->requireNull = true; 
-      pIndex->flag &= ~(TSDB_COL_NULL); 
+      pCtx->requireNull = true;
+      pIndex->flag &= ~(TSDB_COL_NULL);
     } else {
-      pCtx->requireNull = false; 
-    } 
+      pCtx->requireNull = false;
+    }
 
     int32_t index = pSqlFuncMsg->colInfo.colIndex;
     if (TSDB_COL_IS_TAG(pIndex->flag)) {
@@ -1927,24 +1925,24 @@ static bool onlyFirstQuery(SQuery *pQuery) { return onlyOneQueryType(pQuery, TSD
 static bool onlyLastQuery(SQuery *pQuery) { return onlyOneQueryType(pQuery, TSDB_FUNC_LAST, TSDB_FUNC_LAST_DST); }
 
 // todo refactor, add iterator
-static void doExchangeTimeWindow(SQInfo* pQInfo) {
-  size_t t = GET_NUM_OF_TABLEGROUP(pQInfo);
+static void doExchangeTimeWindow(SQInfo* pQInfo, STimeWindow* win) {
+  size_t t = taosArrayGetSize(pQInfo->tableGroupInfo.pGroupList);
   for(int32_t i = 0; i < t; ++i) {
-    SArray* p1 = GET_TABLEGROUP(pQInfo, i);
+    SArray* p1 = taosArrayGetP(pQInfo->tableGroupInfo.pGroupList, i);
 
-    SArray* tableKeyGroup = taosArrayGetP(pQInfo->tableGroupInfo.pGroupList, i);
     size_t len = taosArrayGetSize(p1);
     for(int32_t j = 0; j < len; ++j) {
-      STableQueryInfo* pTableQueryInfo = (STableQueryInfo*) taosArrayGetP(p1, j);
-      SWAP(pTableQueryInfo->win.skey, pTableQueryInfo->win.ekey, TSKEY);
+      STableKeyInfo* pInfo = taosArrayGet(p1, j);
 
-      STableKeyInfo* pInfo = taosArrayGet(tableKeyGroup, j);
-      pInfo->lastKey = pTableQueryInfo->win.skey;
+      // update the new lastkey if it is equalled to the value of the old skey
+      if (pInfo->lastKey == win->ekey) {
+        pInfo->lastKey = win->skey;
+      }
     }
   }
 }
 
-static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
+static void changeExecuteScanOrder(SQInfo *pQInfo, SQueryTableMsg* pQueryMsg, bool stableQuery) {
   SQuery* pQuery = pQInfo->runtimeEnv.pQuery;
 
   // in case of point-interpolation query, use asc order scan
@@ -1961,15 +1959,17 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
     if (pQuery->window.skey > pQuery->window.ekey) {
       SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
     }
+
     return;
   }
 
-  if (isGroupbyNormalCol(pQuery->pGroupbyExpr)) {
+  if (isGroupbyNormalCol(pQuery->pGroupbyExpr) && pQuery->order.order == TSDB_ORDER_DESC) {
     pQuery->order.order = TSDB_ORDER_ASC;
     if (pQuery->window.skey > pQuery->window.ekey) {
       SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
     }
 
+    doExchangeTimeWindow(pQInfo, &pQuery->window);
     return;
   }
 
@@ -1991,7 +1991,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
                pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
 
         SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
-        doExchangeTimeWindow(pQInfo);
+        doExchangeTimeWindow(pQInfo, &pQuery->window);
       }
 
       pQuery->order.order = TSDB_ORDER_ASC;
@@ -2001,7 +2001,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
                pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
 
         SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
-        doExchangeTimeWindow(pQInfo);
+        doExchangeTimeWindow(pQInfo, &pQuery->window);
       }
 
       pQuery->order.order = TSDB_ORDER_DESC;
@@ -2015,6 +2015,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
                  pQuery->window.skey, pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
 
           SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
+          doExchangeTimeWindow(pQInfo, &pQuery->window);
         }
 
         pQuery->order.order = TSDB_ORDER_ASC;
@@ -2024,6 +2025,7 @@ static void changeExecuteScanOrder(SQInfo *pQInfo, bool stableQuery) {
                  pQuery->window.skey, pQuery->window.ekey, pQuery->window.ekey, pQuery->window.skey);
 
           SWAP(pQuery->window.skey, pQuery->window.ekey, TSKEY);
+          doExchangeTimeWindow(pQInfo, &pQuery->window);
         }
 
         pQuery->order.order = TSDB_ORDER_DESC;
@@ -4449,10 +4451,6 @@ int32_t doInitQInfo(SQInfo *pQInfo, STSBuf *pTsBuf, void *tsdb, int32_t vgId, bo
 
   setScanLimitationByResultBuffer(pQuery);
 
-  // NOTE: pTableCheckInfo need to update the query time range and the lastKey info
-  // TODO fixme
-  changeExecuteScanOrder(pQInfo, isSTableQuery);
-
   code = setupQueryHandle(tsdb, pQInfo, isSTableQuery);
   if (code != TSDB_CODE_SUCCESS) {
     return code;
@@ -6022,14 +6020,6 @@ static void doUpdateExprColumnIndex(SQuery *pQuery) {
   }
 }
 
-static int compareTableIdInfo(const void* a, const void* b) {
-  const STableIdInfo* x = (const STableIdInfo*)a;
-  const STableIdInfo* y = (const STableIdInfo*)b;
-  if (x->uid > y->uid) return 1;
-  if (x->uid < y->uid) return -1;
-  return 0;
-}
-
 static void freeQInfo(SQInfo *pQInfo);
 
 static void calResultBufSize(SQuery* pQuery) {
@@ -6051,8 +6041,8 @@ static void calResultBufSize(SQuery* pQuery) {
   }
 }
 
-static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
-                               STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols) {
+static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SSqlGroupbyExpr *pGroupbyExpr, SExprInfo *pExprs,
+                               STableGroupInfo *pTableGroupInfo, SColumnInfo* pTagCols, bool stableQuery) {
   int16_t numOfCols = pQueryMsg->numOfCols;
   int16_t numOfOutput = pQueryMsg->numOfOutput;
 
@@ -6151,8 +6141,6 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
   }
 
   int tableIndex = 0;
-  STimeWindow window = pQueryMsg->window;
-  taosArraySort(pTableIdList, compareTableIdInfo);
 
   pQInfo->runtimeEnv.interBufSize = getOutputInterResultBufSize(pQuery);
   pQInfo->pBuf = calloc(pTableGroupInfo->numOfTables, sizeof(STableQueryInfo));
@@ -6161,12 +6149,20 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
   }
 
   // NOTE: pTableCheckInfo need to update the query time range and the lastKey info
-//  changeExecuteScanOrder(pQInfo, stableQuery);
+  pQInfo->arrTableIdInfo = taosArrayInit(tableIndex, sizeof(STableIdInfo));
+  pQInfo->dataReady = QUERY_RESULT_NOT_READY;
+  pthread_mutex_init(&pQInfo->lock, NULL);
+
+  pQuery->pos = -1;
+  pQuery->window = pQueryMsg->window;
+  changeExecuteScanOrder(pQInfo, pQueryMsg, stableQuery);
+
+  STimeWindow window = pQuery->window;
 
   int32_t index = 0;
 
   for(int32_t i = 0; i < numOfGroups; ++i) {
-    SArray* pa = taosArrayGetP(pTableGroupInfo->pGroupList, i);
+    SArray* pa = taosArrayGetP(pQInfo->tableGroupInfo.pGroupList, i);
 
     size_t s = taosArrayGetSize(pa);
     SArray* p1 = taosArrayInit(s, POINTER_BYTES);
@@ -6179,12 +6175,9 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
     for(int32_t j = 0; j < s; ++j) {
       STableKeyInfo* info = taosArrayGet(pa, j);
 
-      STableId* id = TSDB_TABLEID(info->pTable);
-      STableIdInfo* pTableId = taosArraySearch(pTableIdList, id, compareTableIdInfo);
-
-      window.skey = (pTableId != NULL)? pTableId->key:pQueryMsg->window.skey;
       void* buf = (char*)pQInfo->pBuf + index * sizeof(STableQueryInfo);
 
+      window.skey = info->lastKey;
       STableQueryInfo* item = createTableQueryInfo(&pQInfo->runtimeEnv, info->pTable, window, buf);
       if (item == NULL) {
         goto _cleanup;
@@ -6192,17 +6185,13 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
 
       item->groupIndex = i;
       taosArrayPush(p1, &item);
+
+      STableId* id = TSDB_TABLEID(info->pTable);
       taosHashPut(pQInfo->tableqinfoGroupInfo.map, &id->tid, sizeof(id->tid), &item, POINTER_BYTES);
       index += 1;
     }
   }
 
-  pQInfo->arrTableIdInfo = taosArrayInit(tableIndex, sizeof(STableIdInfo));
-  pQInfo->dataReady = QUERY_RESULT_NOT_READY;
-  pthread_mutex_init(&pQInfo->lock, NULL);
-
-  pQuery->pos = -1;
-  pQuery->window = pQueryMsg->window;
   colIdCheck(pQuery);
 
   qDebug("qmsg:%p QInfo:%p created", pQueryMsg, pQInfo);
@@ -6558,7 +6547,7 @@ int32_t qCreateQueryInfo(void* tsdb, int32_t vgId, SQueryTableMsg* pQueryMsg, qi
     assert(0);
   }
 
-  (*pQInfo) = createQInfoImpl(pQueryMsg, pTableIdList, pGroupbyExpr, pExprs, &tableGroupInfo, pTagColumnInfo);
+  (*pQInfo) = createQInfoImpl(pQueryMsg, pGroupbyExpr, pExprs, &tableGroupInfo, pTagColumnInfo, isSTableQuery);
   pExprs = NULL;
   pGroupbyExpr = NULL;
   pTagColumnInfo = NULL;

src/util/src/tcache.c

@@ -378,40 +378,43 @@ void taosCacheRelease(SCacheObj *pCacheObj, void **data, bool _remove) {
     } else {
       // NOTE: remove it from hash in the first place, otherwise, the pNode may have been released by other thread
       // when reaches here.
-      SCacheDataNode* p = NULL;
-      int32_t ret = taosHashRemoveWithData(pCacheObj->pHashTable, pNode->key, pNode->keySize, &p, sizeof(void*));
+      SCacheDataNode *p = NULL;
+      int32_t ret = taosHashRemoveWithData(pCacheObj->pHashTable, pNode->key, pNode->keySize, &p, sizeof(void *));
       ref = T_REF_DEC(pNode);
 
       // successfully remove from hash table, if failed, this node must have been move to trash already, do nothing.
       // note that the remove operation can be executed only once.
       if (ret == 0) {
-	if (p != pNode) {
-	uDebug("cache:%s, key:%p, successfully removed a new entry:%p, refcnt:%d, prev entry:%p has been removed by others already", pCacheObj->name, pNode->key, p->data, T_REF_VAL_GET(p), pNode->data);
-	assert(p->pTNodeHeader == NULL);
-	taosAddToTrash(pCacheObj, p);
-	} else {
+        if (p != pNode) {
+          uDebug( "cache:%s, key:%p, successfully removed a new entry:%p, refcnt:%d, prev entry:%p has been removed by "
+              "others already", pCacheObj->name, pNode->key, p->data, T_REF_VAL_GET(p), pNode->data);
+
+          assert(p->pTNodeHeader == NULL);
+          taosAddToTrash(pCacheObj, p);
+        } else {
+          uDebug("cache:%s, key:%p, %p successfully removed from hash table, refcnt:%d", pCacheObj->name, pNode->key,
+                 pNode->data, ref);
+          if (ref > 0) {
+            assert(pNode->pTNodeHeader == NULL);
 
-        uDebug("cache:%s, key:%p, %p successfully removed from hash table, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, ref);
-        if (ref > 0) {
-          assert(pNode->pTNodeHeader == NULL);
+            taosAddToTrash(pCacheObj, pNode);
+          } else {  // ref == 0
+            atomic_sub_fetch_64(&pCacheObj->totalSize, pNode->size);
 
-          taosAddToTrash(pCacheObj, pNode);
-        } else {  // ref == 0
-          atomic_sub_fetch_64(&pCacheObj->totalSize, pNode->size);
+            int32_t size = (int32_t)taosHashGetSize(pCacheObj->pHashTable);
+            uDebug("cache:%s, key:%p, %p is destroyed from cache, size:%dbytes, num:%d size:%" PRId64 "bytes",
+                   pCacheObj->name, pNode->key, pNode->data, pNode->size, size, pCacheObj->totalSize);
 
-          int32_t size = (int32_t)taosHashGetSize(pCacheObj->pHashTable);
-          uDebug("cache:%s, key:%p, %p is destroyed from cache, size:%dbytes, num:%d size:%" PRId64 "bytes",
-                 pCacheObj->name, pNode->key, pNode->data, pNode->size, size, pCacheObj->totalSize);
+            if (pCacheObj->freeFp) {
+              pCacheObj->freeFp(pNode->data);
+            }
 
-          if (pCacheObj->freeFp) {
-            pCacheObj->freeFp(pNode->data);
+            free(pNode);
           }
-
-          free(pNode);
         }
-	}
       } else {
-        uDebug("cache:%s, key:%p, %p has been removed from hash table by other thread already, refcnt:%d", pCacheObj->name, pNode->key, pNode->data, ref);
+        uDebug("cache:%s, key:%p, %p has been removed from hash table by other thread already, refcnt:%d",
+               pCacheObj->name, pNode->key, pNode->data, ref);
       }
     }
 
@@ -513,7 +516,7 @@ void taosAddToTrash(SCacheObj *pCacheObj, SCacheDataNode *pNode) {
   pCacheObj->numOfElemsInTrash++;
   __cache_unlock(pCacheObj);
 
-  uDebug("%s key:%p, %p move to trash, numOfElem in trash:%d", pCacheObj->name, pNode->key, pNode->data,
+  uDebug("cache:%s key:%p, %p move to trash, numOfElem in trash:%d", pCacheObj->name, pNode->key, pNode->data,
       pCacheObj->numOfElemsInTrash);
 }
 

src/util/tests/CMakeLists.txt

@@ -10,6 +10,6 @@ IF (HEADER_GTEST_INCLUDE_DIR AND LIB_GTEST_STATIC_DIR)
     INCLUDE_DIRECTORIES(${HEADER_GTEST_INCLUDE_DIR})
     AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SOURCE_LIST)
 
-    ADD_EXECUTABLE(utilTest ./cacheTest.cpp ./hashTest.cpp)
+    ADD_EXECUTABLE(utilTest ${SOURCE_LIST})
     TARGET_LINK_LIBRARIES(utilTest tutil common osdetail gtest pthread gcov)
 ENDIF()

src/util/tests/cacheTest.cpp

 #include "os.h"
 #include <iostream>
 #include <gtest/gtest.h>
-#include <sys/time.h>
 
 #include "taos.h"
-//#include "tsdb.h"
-
-//#include "testCommon.h"
-#include "tstoken.h"
-#include "tutil.h"
 #include "tcache.h"
-#include "ttimer.h"
 
 namespace {
 int32_t tsMaxMgmtConnections = 10000;

tests/script/general/parser/constCol.sim

@@ -347,6 +347,8 @@ if $rows != 3 then
   return -1
 endi
 
+print ======================udc with normal column group by
+
 sql_error select from t1
 sql_error select abc from t1
 sql_error select abc as tu from t1

tests/script/general/parser/lastrow_query.sim

@@ -152,3 +152,5 @@ sql select t1,t1,count(*),t1,t1 from lr_stb0 where ts>'2018-09-24 00:00:00.000'
 if $rows != 46 then
   return -1
 endi
+
+

tests/script/general/parser/testSuite.sim

@@ -99,6 +99,8 @@ run general/parser/union.sim
 sleep 2000
 run general/parser/constCol.sim
 sleep 2000
+run general/parser/timestamp.sim
+sleep 2000
 run general/parser/sliding.sim
 
 #sleep 2000

tests/script/general/parser/topbot.sim

@@ -118,4 +118,23 @@ if $data21 != 2.10000 then
   return -1
 endi
 
+print =====================td-1302 case
+sql create database t1 keep 36500;
+sql use t1;
+sql create table test(ts timestamp, k int);
+sql insert into test values(29999, 1)(70000, 2)(80000, 3)
+
+print ================== restart server to commit data into disk
+system sh/exec.sh -n dnode1 -s stop -x SIGINT
+sleep 5000
+system sh/exec.sh -n dnode1 -s start
+print ================== server restart completed
+sql connect
+sleep 3000
+
+sql select count(*) from t1.test where ts>10000 and ts<90000 interval(5000a)
+if $rows != 3 then
+ return -1
+endi
+
 system sh/exec.sh -n dnode1 -s stop -x SIGINT
\ No newline at end of file