Merge pull request #2572 from taosdata/feature/query

Feature/query

src/client/src/tscLocalMerge.c

@@ -364,7 +364,7 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
   
   TSKEY stime = MIN(pQueryInfo->window.skey, pQueryInfo->window.ekey);
   int64_t revisedSTime =
-      taosGetIntervalStartTimestamp(stime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
+      taosGetIntervalStartTimestamp(stime, pQueryInfo->slidingTime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
   
   if (pQueryInfo->fillType != TSDB_FILL_NONE) {
     SFillColInfo* pFillCol = createFillColInfo(pQueryInfo);
@@ -831,7 +831,7 @@ void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SQueryInfo
   if (pFillInfo != NULL) {
     int64_t stime = (pQueryInfo->window.skey < pQueryInfo->window.ekey) ? pQueryInfo->window.skey : pQueryInfo->window.ekey;
     int64_t revisedSTime =
-        taosGetIntervalStartTimestamp(stime, pQueryInfo->slidingTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
+        taosGetIntervalStartTimestamp(stime, pQueryInfo->slidingTime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, tinfo.precision);
   
     taosResetFillInfo(pFillInfo, revisedSTime);
   }
@@ -1301,9 +1301,7 @@ static void resetEnvForNewResultset(SSqlRes *pRes, SSqlCmd *pCmd, SLocalReducer
   if (pQueryInfo->fillType != TSDB_FILL_NONE) {
     TSKEY skey = MIN(pQueryInfo->window.skey, pQueryInfo->window.ekey);
     int64_t newTime =
-        taosGetIntervalStartTimestamp(skey, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, precision);
-//    taosResetFillInfo(pLocalReducer->pFillInfo, pQueryInfo->order.order, newTime,
-//                        pQueryInfo->groupbyExpr.numOfGroupCols, 4096, 0, NULL, pLocalReducer->rowSize);
+        taosGetIntervalStartTimestamp(skey, pQueryInfo->slidingTime, pQueryInfo->intervalTime, pQueryInfo->slidingTimeUnit, precision);
     taosResetFillInfo(pLocalReducer->pFillInfo, newTime);
   }
 }

src/client/src/tscSQLParser.c

@@ -4487,10 +4487,12 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
 
     SUpdateTableTagValMsg* pUpdateMsg = (SUpdateTableTagValMsg*) pCmd->payload;
     pUpdateMsg->head.vgId = htonl(pTableMeta->vgroupInfo.vgId);
-    pUpdateMsg->tid = htonl(pTableMeta->sid);
-    pUpdateMsg->uid = htobe64(pTableMeta->uid);
-    pUpdateMsg->colId = htons(pTagsSchema->colId);
-    pUpdateMsg->tversion = htons(pTableMeta->tversion);
+    pUpdateMsg->tid       = htonl(pTableMeta->sid);
+    pUpdateMsg->uid       = htobe64(pTableMeta->uid);
+    pUpdateMsg->colId     = htons(pTagsSchema->colId);
+    pUpdateMsg->type      = htons(pTagsSchema->type);
+    pUpdateMsg->bytes     = htons(pTagsSchema->bytes);
+    pUpdateMsg->tversion  = htons(pTableMeta->tversion);
     pUpdateMsg->numOfTags = htons(numOfTags);
     pUpdateMsg->schemaLen = htonl(schemaLen);
 

src/common/inc/tname.h

@@ -29,4 +29,6 @@ bool tscValidateTableNameLength(size_t len);
 
 SColumnFilterInfo* tscFilterInfoClone(const SColumnFilterInfo* src, int32_t numOfFilters);
 
+int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, int64_t intervalTime, char timeUnit, int16_t precision);
+
 #endif  // TDENGINE_NAME_H

src/common/src/tname.c

@@ -75,3 +75,33 @@ SColumnFilterInfo* tscFilterInfoClone(const SColumnFilterInfo* src, int32_t numO
 
   return pFilter;
 }
+
+int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, int64_t intervalTime, char timeUnit, int16_t precision) {
+  if (slidingTime == 0) {
+    return startTime;
+  }
+
+  int64_t start = ((startTime - intervalTime) / slidingTime + 1) * slidingTime;
+  if (!(timeUnit == 'a' || timeUnit == 'm' || timeUnit == 's' || timeUnit == 'h')) {
+    /*
+     * here we revised the start time of day according to the local time zone,
+     * but in case of DST, the start time of one day need to be dynamically decided.
+     */
+    // todo refactor to extract function that is available for Linux/Windows/Mac platform
+#if defined(WINDOWS) && _MSC_VER >= 1900
+    // see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
+    int64_t timezone = _timezone;
+    int32_t daylight = _daylight;
+    char**  tzname = _tzname;
+#endif
+
+    int64_t t = (precision == TSDB_TIME_PRECISION_MILLI) ? MILLISECOND_PER_SECOND : MILLISECOND_PER_SECOND * 1000L;
+    start += timezone * t;
+  }
+
+  int64_t end = start + intervalTime - 1;
+  if (end < startTime) {
+    start += slidingTime;
+  }
+  return start;
+}

src/inc/taosmsg.h

@@ -285,6 +285,8 @@ typedef struct {
   int32_t   tid;
   int16_t   tversion;
   int16_t   colId;
+  int16_t   type;
+  int16_t   bytes;
   int32_t   tagValLen;
   int16_t   numOfTags;
   int32_t   schemaLen;

src/inc/tsdb.h

@@ -108,7 +108,9 @@ void tsdbClearTableCfg(STableCfg *config);
 
 void* tsdbGetTableTagVal(const void* pTable, int32_t colId, int16_t type, int16_t bytes);
 char* tsdbGetTableName(void *pTable);
-STableId tsdbGetTableId(void *pTable);
+
+#define TSDB_TABLEID(_table) ((STableId*) (_table))
+
 STableCfg *tsdbCreateTableCfgFromMsg(SMDCreateTableMsg *pMsg);
 
 int   tsdbCreateTable(TSDB_REPO_T *repo, STableCfg *pCfg);

src/query/inc/qExecutor.h

@@ -154,6 +154,7 @@ typedef struct SQuery {
 } SQuery;
 
 typedef struct SQueryRuntimeEnv {
+  jmp_buf              env;
   SResultInfo*         resultInfo;       // todo refactor to merge with SWindowResInfo
   SQuery*              pQuery;
   SQLFunctionCtx*      pCtx;
@@ -169,6 +170,8 @@ typedef struct SQueryRuntimeEnv {
   void*                pSecQueryHandle;  // another thread for
   bool                 stableQuery;      // super table query or not
   bool                 topBotQuery;      // false
+  bool                 groupbyNormalCol; // denote if this is a groupby normal column query
+  bool                 hasTagResults;    // if there are tag values in final result or not
   int32_t              prevGroupId;      // previous executed group id
   SDiskbasedResultBuf* pResultBuf;       // query result buffer based on blocked-wised disk file
 } SQueryRuntimeEnv;
@@ -197,8 +200,10 @@ typedef struct SQInfo {
    */
   int32_t          tableIndex;
   int32_t          numOfGroupResultPages;
-  _qinfo_free_fn_t freeFn;
-  jmp_buf          env;
+  _qinfo_free_fn_t freeFn;  //todo remove it
+
+  void*     pBuf; // allocated buffer for STableQueryInfo, sizeof(STableQueryInfo)*numOfTables;
+
 } SQInfo;
 
 #endif  // TDENGINE_QUERYEXECUTOR_H

src/query/inc/qfill.h

@@ -60,8 +60,6 @@ typedef struct SPoint {
   void *  val;
 } SPoint;
 
-int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, char timeUnit, int16_t precision);
-
 SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
                             int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
                             SFillColInfo* pFillCol);

src/query/src/qUtil.c

@@ -32,7 +32,6 @@ int32_t initWindowResInfo(SWindowResInfo *pWindowResInfo, SQueryRuntimeEnv *pRun
   pWindowResInfo->threshold = threshold;
   
   pWindowResInfo->type = type;
-  
   _hash_fn_t fn = taosGetDefaultHashFunction(type);
   pWindowResInfo->hashList = taosHashInit(threshold, fn, false);
   
@@ -54,7 +53,8 @@ void destroyTimeWindowRes(SWindowResult *pWindowRes, int32_t nOutputCols) {
   if (pWindowRes == NULL) {
     return;
   }
-  
+
+  // TODO opt malloc strategy
   for (int32_t i = 0; i < nOutputCols; ++i) {
     free(pWindowRes->resultInfo[i].interResultBuf);
   }
@@ -180,19 +180,34 @@ void closeAllTimeWindow(SWindowResInfo *pWindowResInfo) {
 
 /*
  * remove the results that are not the FIRST time window that spreads beyond the
- * the last qualified time stamp in case of sliding query, which the sliding time is not equalled to the interval time
+ * the last qualified time stamp in case of sliding query, which the sliding time is not equalled to the interval time.
+ * NOTE: remove redundant, only when the result set order equals to traverse order
  */
 void removeRedundantWindow(SWindowResInfo *pWindowResInfo, TSKEY lastKey, int32_t order) {
   assert(pWindowResInfo->size >= 0 && pWindowResInfo->capacity >= pWindowResInfo->size);
-  
+  if (pWindowResInfo->size <= 1) {
+    return;
+  }
+
+  // get the result order
+  int32_t resultOrder = (pWindowResInfo->pResult[0].window.skey < pWindowResInfo->pResult[1].window.skey)?
+                         TSDB_ORDER_ASC:TSDB_ORDER_DESC;
+
+  if (order != resultOrder) {
+    return;
+  }
+
   int32_t i = 0;
-  while (i < pWindowResInfo->size &&
-      ((pWindowResInfo->pResult[i].window.ekey < lastKey && order == QUERY_ASC_FORWARD_STEP) ||
-          (pWindowResInfo->pResult[i].window.skey > lastKey && order == QUERY_DESC_FORWARD_STEP))) {
-    ++i;
+  if (order == QUERY_ASC_FORWARD_STEP) {
+    while (i < pWindowResInfo->size && (pWindowResInfo->pResult[i].window.ekey < lastKey)) {
+      ++i;
+    }
+  } else if (order == QUERY_DESC_FORWARD_STEP) {
+    while (i < pWindowResInfo->size && (pWindowResInfo->pResult[i].window.skey > lastKey)) {
+      ++i;
+    }
   }
-  
-  //  assert(i < pWindowResInfo->size);
+
   if (i < pWindowResInfo->size) {
     pWindowResInfo->size = (i + 1);
   }

src/query/src/qextbuffer.c

@@ -118,7 +118,7 @@ static bool tExtMemBufferAlloc(tExtMemBuffer *pMemBuffer) {
    * To flush data to disk to accommodate more data
    */
   if (pMemBuffer->numOfInMemPages > 0 && pMemBuffer->numOfInMemPages == pMemBuffer->inMemCapacity) {
-    if (!tExtMemBufferFlush(pMemBuffer)) {
+    if (tExtMemBufferFlush(pMemBuffer) != 0) {
       return false;
     }
   }
@@ -268,6 +268,7 @@ int32_t tExtMemBufferFlush(tExtMemBuffer *pMemBuffer) {
     size_t retVal = fwrite((char *)&(first->item), pMemBuffer->pageSize, 1, pMemBuffer->file);
     if (retVal <= 0) {  // failed to write to buffer, may be not enough space
       ret = TAOS_SYSTEM_ERROR(errno);
+      return ret;
     }
 
     pMemBuffer->fileMeta.numOfElemsInFile += first->item.num;

src/query/src/qfill.c

@@ -22,41 +22,6 @@
 
 #define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
 
-int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, char timeUnit, int16_t precision) {
-  if (slidingTime == 0) {
-    return startTime;
-  }
-
-  if (timeUnit == 'a' || timeUnit == 'm' || timeUnit == 's' || timeUnit == 'h') {
-    return (startTime / slidingTime) * slidingTime;
-  } else {
-    /*
-     * here we revised the start time of day according to the local time zone,
-     * but in case of DST, the start time of one day need to be dynamically decided.
-     *
-     * TODO dynamically decide the start time of a day, move to common module
-     */
-
-    // todo refactor to extract function that is available for Linux/Windows/Mac platform
-#if defined(WINDOWS) && _MSC_VER >= 1900
-    // see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
-    int64_t timezone = _timezone;
-    int32_t daylight = _daylight;
-    char**  tzname = _tzname;
-#endif
-
-    int64_t t = (precision == TSDB_TIME_PRECISION_MILLI) ? MILLISECOND_PER_SECOND : MILLISECOND_PER_SECOND * 1000L;
-
-    int64_t revStartime = (startTime / slidingTime) * slidingTime + timezone * t;
-    int64_t revEndtime = revStartime + slidingTime - 1;
-    if (revEndtime < startTime) {
-      revStartime += slidingTime;
-    }
-
-    return revStartime;
-  }
-}
-
 SFillInfo* taosInitFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
     int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType, SFillColInfo* pFillCol) {
   if (fillType == TSDB_FILL_NONE) {
@@ -128,7 +93,7 @@ static TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int64_t timeInterva
   if (order == TSDB_ORDER_ASC) {
     return ekey;
   } else {
-    return taosGetIntervalStartTimestamp(ekey, timeInterval, slidingTimeUnit, precision);
+    return taosGetIntervalStartTimestamp(ekey, timeInterval, timeInterval, slidingTimeUnit, precision);
   }
 }
 

src/tsdb/inc/tsdbMain.h

@@ -47,9 +47,9 @@ extern int tsdbDebugFlag;
 // Definitions
 // ------------------ tsdbMeta.c
 typedef struct STable {
+  STableId       tableId;
   ETableType     type;
   tstr*          name;  // NOTE: there a flexible string here
-  STableId       tableId;
   uint64_t       suid;
   struct STable* pSuper;  // super table pointer
   uint8_t        numOfSchemas;
@@ -294,14 +294,36 @@ typedef struct {
 #define TABLE_SUID(t) (t)->suid
 #define TABLE_LASTKEY(t) (t)->lastKey
 
+static FORCE_INLINE STSchema *tsdbGetTableSchema(STable *pTable) {
+  if (pTable->type == TSDB_CHILD_TABLE) {  // check child table first
+    STable *pSuper = pTable->pSuper;
+    if (pSuper == NULL) return NULL;
+    return pSuper->schema[pSuper->numOfSchemas - 1];
+  } else if (pTable->type == TSDB_NORMAL_TABLE || pTable->type == TSDB_SUPER_TABLE || pTable->type == TSDB_STREAM_TABLE) {
+    return pTable->schema[pTable->numOfSchemas - 1];
+  } else {
+    return NULL;
+  }
+}
+
+static FORCE_INLINE STSchema *tsdbGetTableTagSchema(STable *pTable) {
+  if (pTable->type == TSDB_CHILD_TABLE) {  // check child table first
+    STable *pSuper = pTable->pSuper;
+    if (pSuper == NULL) return NULL;
+    return pSuper->tagSchema;
+  } else if (pTable->type == TSDB_SUPER_TABLE) {
+    return pTable->tagSchema;
+  } else {
+    return NULL;
+  }
+}
+
 STsdbMeta* tsdbNewMeta(STsdbCfg* pCfg);
 void       tsdbFreeMeta(STsdbMeta* pMeta);
 int        tsdbOpenMeta(STsdbRepo* pRepo);
 int        tsdbCloseMeta(STsdbRepo* pRepo);
-STSchema*  tsdbGetTableSchema(STable* pTable);
 STable*    tsdbGetTableByUid(STsdbMeta* pMeta, uint64_t uid);
 STSchema*  tsdbGetTableSchemaByVersion(STable* pTable, int16_t version);
-STSchema*  tsdbGetTableTagSchema(STable* pTable);
 int        tsdbUpdateTable(STsdbRepo* pRepo, STable* pTable, STableCfg* pCfg);
 int        tsdbWLockRepoMeta(STsdbRepo* pRepo);
 int        tsdbRLockRepoMeta(STsdbRepo* pRepo);

src/tsdb/src/tsdbMeta.c

@@ -274,6 +274,8 @@ int tsdbUpdateTagValue(TSDB_REPO_T *repo, SUpdateTableTagValMsg *pMsg) {
   pMsg->tid = htonl(pMsg->tid);
   pMsg->tversion  = htons(pMsg->tversion);
   pMsg->colId     = htons(pMsg->colId);
+  pMsg->type      = htons(pMsg->type);
+  pMsg->bytes     = htons(pMsg->bytes);
   pMsg->tagValLen = htonl(pMsg->tagValLen);
   pMsg->numOfTags = htons(pMsg->numOfTags);
   pMsg->schemaLen = htonl(pMsg->schemaLen);
@@ -449,18 +451,6 @@ int tsdbCloseMeta(STsdbRepo *pRepo) {
   return 0;
 }
 
-STSchema *tsdbGetTableSchema(STable *pTable) {
-  if (pTable->type == TSDB_NORMAL_TABLE || pTable->type == TSDB_SUPER_TABLE || pTable->type == TSDB_STREAM_TABLE) {
-    return pTable->schema[pTable->numOfSchemas - 1];
-  } else if (pTable->type == TSDB_CHILD_TABLE) {
-    STable *pSuper = pTable->pSuper;
-    if (pSuper == NULL) return NULL;
-    return pSuper->schema[pSuper->numOfSchemas - 1];
-  } else {
-    return NULL;
-  }
-}
-
 STable *tsdbGetTableByUid(STsdbMeta *pMeta, uint64_t uid) {
   void *ptr = taosHashGet(pMeta->uidMap, (char *)(&uid), sizeof(uid));
 
@@ -480,18 +470,6 @@ STSchema *tsdbGetTableSchemaByVersion(STable *pTable, int16_t version) {
   return *(STSchema **)ptr;
 }
 
-STSchema *tsdbGetTableTagSchema(STable *pTable) {
-  if (pTable->type == TSDB_SUPER_TABLE) {
-    return pTable->tagSchema;
-  } else if (pTable->type == TSDB_CHILD_TABLE) {
-    STable *pSuper = pTable->pSuper;
-    if (pSuper == NULL) return NULL;
-    return pSuper->tagSchema;
-  } else {
-    return NULL;
-  }
-}
-
 int tsdbUpdateTable(STsdbRepo *pRepo, STable *pTable, STableCfg *pCfg) {
   // TODO: this function can only be called when there is no query and commit on this table
   ASSERT(TABLE_TYPE(pTable) != TSDB_CHILD_TABLE);

src/vnode/src/vnodeRead.c

@@ -110,13 +110,13 @@ static int32_t vnodeProcessQueryMsg(SVnodeObj *pVnode, SReadMsg *pReadMsg) {
 
     // current connect is broken
     if (code == TSDB_CODE_SUCCESS) {
-      // add lock here
       handle = qRegisterQInfo(pVnode->qMgmt, pQInfo);
       if (handle == NULL) {  // failed to register qhandle
         pRsp->code = TSDB_CODE_QRY_INVALID_QHANDLE;
 
         qKillQuery(pQInfo);
         qKillQuery(pQInfo);
+        pQInfo = NULL;
       } else {
         assert(*handle == pQInfo);
         pRsp->qhandle = htobe64((uint64_t) (handle));