Merge branch '3.0' into feat/agg_client_api

include/common/tcommon.h

@@ -268,26 +268,6 @@ typedef struct SSortExecInfo {
   int32_t readBytes;   // read io bytes
 } SSortExecInfo;
 
-//======================================================================================================================
-// for grant 
-typedef enum {
-  TSDB_GRANT_ALL,
-  TSDB_GRANT_TIME,
-  TSDB_GRANT_USER,
-  TSDB_GRANT_DB,
-  TSDB_GRANT_TIMESERIES,
-  TSDB_GRANT_DNODE,
-  TSDB_GRANT_ACCT,
-  TSDB_GRANT_STORAGE,
-  TSDB_GRANT_SPEED,
-  TSDB_GRANT_QUERY_TIME,
-  TSDB_GRANT_CONNS,
-  TSDB_GRANT_STREAMS,
-  TSDB_GRANT_CPU_CORES,
-} EGrantType;
-
-int32_t grantCheck(EGrantType grant);
-
 #ifdef __cplusplus
 }
 #endif

include/common/tgrant.h

+/*
+ * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
+ *
+ * This program is free software: you can use, redistribute, and/or modify
+ * it under the terms of the GNU Affero General Public License, version 3
+ * or later ("AGPL"), as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.
+ *
+ * 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/>.
+ */
+
+#ifndef _TD_COMMON_GRANT_H_
+#define _TD_COMMON_GRANT_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "os.h"
+
+typedef enum {
+  TSDB_GRANT_ALL,
+  TSDB_GRANT_TIME,
+  TSDB_GRANT_USER,
+  TSDB_GRANT_DB,
+  TSDB_GRANT_TIMESERIES,
+  TSDB_GRANT_DNODE,
+  TSDB_GRANT_ACCT,
+  TSDB_GRANT_STORAGE,
+  TSDB_GRANT_SPEED,
+  TSDB_GRANT_QUERY_TIME,
+  TSDB_GRANT_CONNS,
+  TSDB_GRANT_STREAMS,
+  TSDB_GRANT_CPU_CORES,
+} EGrantType;
+
+int32_t grantCheck(EGrantType grant);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_COMMON_GRANT_H_*/
\ No newline at end of file

source/dnode/mgmt/mgmt_vnode/src/vmWorker.c

@@ -153,9 +153,15 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
 
   switch (qtype) {
     case QUERY_QUEUE:
-      vnodePreprocessQueryMsg(pVnode->pImpl, pMsg);
-      dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pQueryQ, pMsg);
+      if ((pMsg->msgType == TDMT_SCH_QUERY) && (grantCheck(TSDB_GRANT_TIME) != TSDB_CODE_SUCCESS)) {
+        terrno = TSDB_CODE_GRANT_EXPIRED;
+        code = terrno;
+        dDebug("vgId:%d, msg:%p put into vnode-query queue failed since %s", pVnode->vgId, pMsg, terrstr());
+      } else {
+        vnodePreprocessQueryMsg(pVnode->pImpl, pMsg);
+        dGTrace("vgId:%d, msg:%p put into vnode-query queue", pVnode->vgId, pMsg);
+        taosWriteQitem(pVnode->pQueryQ, pMsg);
+      }
       break;
     case STREAM_QUEUE:
       dGTrace("vgId:%d, msg:%p put into vnode-stream queue", pVnode->vgId, pMsg);
@@ -166,9 +172,14 @@ static int32_t vmPutMsgToQueue(SVnodeMgmt *pMgmt, SRpcMsg *pMsg, EQueueType qtyp
       taosWriteQitem(pVnode->pFetchQ, pMsg);
       break;
     case WRITE_QUEUE:
-
-      dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);
-      taosWriteQitem(pVnode->pWriteQ, pMsg);
+      if ((pMsg->msgType == TDMT_VND_SUBMIT) && (grantCheck(TSDB_GRANT_STORAGE) != TSDB_CODE_SUCCESS)) {
+        terrno = TSDB_CODE_VND_NO_WRITE_AUTH;
+        code = terrno;
+        dDebug("vgId:%d, msg:%p put into vnode-write queue failed since %s", pVnode->vgId, pMsg, terrstr());
+      } else {
+        dGTrace("vgId:%d, msg:%p put into vnode-write queue", pVnode->vgId, pMsg);
+        taosWriteQitem(pVnode->pWriteQ, pMsg);
+      }
       break;
     case SYNC_QUEUE:
       dGTrace("vgId:%d, msg:%p put into vnode-sync queue", pVnode->vgId, pMsg);

source/dnode/mnode/impl/inc/mndInt.h

@@ -24,6 +24,7 @@
 #include "tcache.h"
 #include "tdatablock.h"
 #include "tglobal.h"
+#include "tgrant.h"
 #include "tqueue.h"
 #include "ttime.h"
 #include "version.h"

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

@@ -509,11 +509,10 @@ static int32_t mndProcessCreateDbReq(SRpcMsg *pReq) {
   SUserObj    *pUser = NULL;
   SCreateDbReq createReq = {0};
 
-  // code = grantCheck(TSDB_GRANT_DB);
-  // if (code != 0) {
-  //   terrno = code;
-  //   goto _OVER;
-  // }
+  if ((terrno = grantCheck(TSDB_GRANT_DB)) != 0) {
+    code = terrno;
+    goto _OVER;
+  }
 
   if (tDeserializeSCreateDbReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;

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

@@ -621,11 +621,10 @@ static int32_t mndProcessCreateDnodeReq(SRpcMsg *pReq) {
   SDnodeObj      *pDnode = NULL;
   SCreateDnodeReq createReq = {0};
 
-  // code = grantCheck(TSDB_GRANT_DNODE);
-  // if (code != TSDB_CODE_SUCCESS) {
-  //   terrno = code;
-  //   goto _OVER;
-  // }
+  if ((terrno = grantCheck(TSDB_GRANT_DNODE)) != 0) {
+    code = terrno;
+    goto _OVER;
+  }
 
   if (tDeserializeSCreateDnodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;

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

@@ -363,11 +363,10 @@ static int32_t mndProcessCreateUserReq(SRpcMsg *pReq) {
     goto _OVER;
   }
 
-  // code = grantCheck(TSDB_GRANT_USER);
-  // if (code != TSDB_CODE_SUCCESS) {
-  //   terrno = code;
-  //   goto _OVER;
-  // }
+  if ((terrno = grantCheck(TSDB_GRANT_USER)) != 0) {
+    code = terrno;
+    goto _OVER;
+  }
 
   code = mndCreateUser(pMnode, pOperUser->acct, &createReq, pReq);
   if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;

source/dnode/vnode/inc/vnode.h

@@ -27,6 +27,7 @@
 #include "wal.h"
 
 #include "tcommon.h"
+#include "tgrant.h"
 #include "tfs.h"
 #include "tmsg.h"
 #include "trow.h"

source/dnode/vnode/src/meta/metaQuery.c

@@ -205,7 +205,7 @@ _query:
   }
 
   tDecoderInit(&dc, pData, nData);
-  tDecodeSSchemaWrapper(&dc, &schema);
+  tDecodeSSchemaWrapperEx(&dc, &schema);
   pSchema = tCloneSSchemaWrapper(&schema);
   tDecoderClear(&dc);
 
@@ -470,9 +470,9 @@ int64_t metaGetTbNum(SMeta *pMeta) {
 }
 
 // N.B. Called by statusReq per second
-int64_t metaGetTimeSeriesNum(SMeta *pMeta) { 
+int64_t metaGetTimeSeriesNum(SMeta *pMeta) {
   // TODO
-  return 400; 
+  return 400;
 }
 
 typedef struct {

source/dnode/vnode/src/tq/tqExec.c

@@ -64,9 +64,16 @@ int64_t tqScan(STQ* pTq, const STqHandle* pHandle, SMqDataRsp* pRsp, STqOffsetVa
   qTaskInfo_t          task = pExec->execCol.task[0];
 
   if (qStreamPrepareScan(task, pOffset) < 0) {
-    ASSERT(pOffset->type == TMQ_OFFSET__LOG);
-    pRsp->rspOffset = *pOffset;
-    return 0;
+    if (pOffset->type == TMQ_OFFSET__LOG) {
+      pRsp->rspOffset = *pOffset;
+      return 0;
+    } else {
+      tqOffsetResetToLog(pOffset, pHandle->snapshotVer + 1);
+      if (qStreamPrepareScan(task, pOffset) < 0) {
+        pRsp->rspOffset = *pOffset;
+        return 0;
+      }
+    }
   }
 
   int32_t rowCnt = 0;

source/dnode/vnode/src/vnd/vnodeSvr.c

@@ -630,6 +630,9 @@ _exit:
   tEncoderInit(&ec, pRsp->pCont, pRsp->contLen);
   tEncodeSVAlterTbRsp(&ec, &vAlterTbRsp);
   tEncoderClear(&ec);
+  if (vMetaRsp.pSchemas) {
+    taosMemoryFree(vMetaRsp.pSchemas);
+  }
   return 0;
 }
 

source/libs/executor/inc/executil.h

@@ -96,9 +96,9 @@ static FORCE_INLINE SResultRow* getResultRowByPos(SDiskbasedBuf* pBuf, SResultRo
 }
 
 void initGroupedResultInfo(SGroupResInfo* pGroupResInfo, SHashObj* pHashmap, int32_t order);
-void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);
-
 void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo);
+
+void initMultiResInfoFromArrayList(SGroupResInfo* pGroupResInfo, SArray* pArrayList);
 bool hasDataInGroupInfo(SGroupResInfo* pGroupResInfo);
 
 int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);

source/libs/executor/src/executil.c

@@ -72,8 +72,12 @@ size_t getResultRowSize(SqlFunctionCtx* pCtx, int32_t numOfOutput) {
 void cleanupGroupResInfo(SGroupResInfo* pGroupResInfo) {
   assert(pGroupResInfo != NULL);
 
-  taosArrayDestroy(pGroupResInfo->pRows);
-  pGroupResInfo->pRows = NULL;
+  for(int32_t i = 0; i < taosArrayGetSize(pGroupResInfo->pRows); ++i) {
+    SResKeyPos* pRes = taosArrayGetP(pGroupResInfo->pRows, i);
+    taosMemoryFree(pRes);
+  }
+
+  pGroupResInfo->pRows = taosArrayDestroy(pGroupResInfo->pRows);
   pGroupResInfo->index = 0;
 }
 

source/libs/executor/src/executorMain.c

@@ -332,6 +332,8 @@ int32_t qStreamPrepareScan(qTaskInfo_t tinfo, const STqOffsetVal* pOffset) {
               STableKeyInfo* pTableInfo = taosArrayGet(pTaskInfo->tableqinfoList.pTableList, 0);
               uid = pTableInfo->uid;
               ts = INT64_MIN;
+            } else {
+              return -1;
             }
           }
           /*if (pTaskInfo->streamInfo.lastStatus.type != TMQ_OFFSET__SNAPSHOT_DATA ||*/

source/libs/executor/src/executorimpl.c

@@ -377,6 +377,10 @@ void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow
   colDataAppendInt64(pColData, 4, &pQueryWindow->ekey);
 }
 
+void cleanupExecTimeWindowInfo(SColumnInfoData* pColData) {
+  colDataDestroy(pColData);
+}
+
 void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, STimeWindow* pWin,
                       SColumnInfoData* pTimeWindowData, int32_t offset, int32_t forwardStep, TSKEY* tsCol,
                       int32_t numOfTotal, int32_t numOfOutput, int32_t order) {
@@ -3737,7 +3741,7 @@ void destroyAggOperatorInfo(void* param, int32_t numOfOutput) {
   cleanupBasicInfo(&pInfo->binfo);
 
   cleanupAggSup(&pInfo->aggSup);
-  taosArrayDestroyEx(pInfo->groupResInfo.pRows, freeItem);
+  cleanupGroupResInfo(&pInfo->groupResInfo);
   taosMemoryFreeClear(param);
 }
 

source/libs/executor/src/timewindowoperator.c

@@ -1513,12 +1513,25 @@ static void destroyStateWindowOperatorInfo(void* param, int32_t numOfOutput) {
   taosMemoryFreeClear(param);
 }
 
+static void freeItem(void* param) {
+  SGroupKeys *pKey = (SGroupKeys*) param;
+  taosMemoryFree(pKey->pData);
+}
+
 void destroyIntervalOperatorInfo(void* param, int32_t numOfOutput) {
   SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)param;
   cleanupBasicInfo(&pInfo->binfo);
   cleanupAggSup(&pInfo->aggSup);
-  taosArrayDestroy(pInfo->pRecycledPages);
+  pInfo->pRecycledPages = taosArrayDestroy(pInfo->pRecycledPages);
+  pInfo->pInterpCols = taosArrayDestroy(pInfo->pInterpCols);
+  taosArrayDestroyEx(pInfo->pPrevValues, freeItem);
 
+  pInfo->pPrevValues = NULL;
+  pInfo->pDelWins = taosArrayDestroy(pInfo->pDelWins);
+  pInfo->pDelRes = blockDataDestroy(pInfo->pDelRes);
+
+  cleanupGroupResInfo(&pInfo->groupResInfo);
+  colDataDestroy(&pInfo->twAggSup.timeWindowData);
   taosMemoryFreeClear(param);
 }