Merge remote-tracking branch 'origin/3.0' into feature/3.0_wxy
This commit is contained in:
Xiaoyu Wang
commit
a92fb4b788
|
|
@ -2319,7 +2319,7 @@ typedef struct {
|
|||
int64_t sliding;
|
||||
char* expr; // sma expression
|
||||
char* tagsFilter;
|
||||
SVgEpSet vgEpSet[];
|
||||
SVgEpSet* pVgEpSet;
|
||||
} STSma; // Time-range-wise SMA
|
||||
|
||||
typedef STSma SVCreateTSmaReq;
|
||||
|
|
|
|||
|
|
@ -1590,7 +1590,6 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
|
|||
for (int32_t k = 0; k < colNum; ++k) { // iterate by column
|
||||
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, k);
|
||||
STColumn* pCol = &pTSchema->columns[k];
|
||||
ASSERT(pCol->type == pColInfoData->info.type);
|
||||
void* var = POINTER_SHIFT(pColInfoData->pData, j * pColInfoData->info.bytes);
|
||||
switch (pColInfoData->info.type) {
|
||||
case TSDB_DATA_TYPE_TIMESTAMP:
|
||||
|
|
@ -1614,20 +1613,39 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
|
|||
case TSDB_DATA_TYPE_VARBINARY:
|
||||
case TSDB_DATA_TYPE_DECIMAL:
|
||||
case TSDB_DATA_TYPE_BLOB:
|
||||
case TSDB_DATA_TYPE_JSON:
|
||||
case TSDB_DATA_TYPE_MEDIUMBLOB:
|
||||
uError("the column type %" PRIi16 " is defined but not implemented yet", pColInfoData->info.type);
|
||||
TASSERT(0);
|
||||
break;
|
||||
default:
|
||||
if (pColInfoData->info.type < TSDB_DATA_TYPE_MAX && pColInfoData->info.type > TSDB_DATA_TYPE_NULL) {
|
||||
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pColInfoData->info.type, TD_VTYPE_NORM, var, true, offset, k);
|
||||
char tv[8] = {0};
|
||||
if (pColInfoData->info.type == TSDB_DATA_TYPE_FLOAT) {
|
||||
float v = 0;
|
||||
GET_TYPED_DATA(v, float, pColInfoData->info.type, var);
|
||||
SET_TYPED_DATA(&tv, pCol->type, v);
|
||||
} else if (pColInfoData->info.type == TSDB_DATA_TYPE_DOUBLE) {
|
||||
double v = 0;
|
||||
GET_TYPED_DATA(v, double, pColInfoData->info.type, var);
|
||||
SET_TYPED_DATA(&tv, pCol->type, v);
|
||||
} else if (IS_SIGNED_NUMERIC_TYPE(pColInfoData->info.type)) {
|
||||
int64_t v = 0;
|
||||
GET_TYPED_DATA(v, int64_t, pColInfoData->info.type, var);
|
||||
SET_TYPED_DATA(&tv, pCol->type, v);
|
||||
} else {
|
||||
uint64_t v = 0;
|
||||
GET_TYPED_DATA(v, uint64_t, pColInfoData->info.type, var);
|
||||
SET_TYPED_DATA(&tv, pCol->type, v);
|
||||
}
|
||||
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, pCol->type, TD_VTYPE_NORM, tv, true, offset, k);
|
||||
} else {
|
||||
uError("the column type %" PRIi16 " is undefined\n", pColInfoData->info.type);
|
||||
TASSERT(0);
|
||||
}
|
||||
break;
|
||||
}
|
||||
offset += TYPE_BYTES[pColInfoData->info.type];
|
||||
offset += TYPE_BYTES[pCol->type]; // sum/avg would convert to int64_t/uint64_t/double during aggregation
|
||||
}
|
||||
dataLen += TD_ROW_LEN(rb.pBuf);
|
||||
#ifdef TD_DEBUG_PRINT_ROW
|
||||
|
|
|
|||
|
|
@ -694,7 +694,6 @@ void tFreeSMAltertbReq(SMAlterStbReq *pReq) {
|
|||
pReq->pFields = NULL;
|
||||
}
|
||||
|
||||
|
||||
int32_t tSerializeSEpSet(void *buf, int32_t bufLen, const SEpSet *pEpset) {
|
||||
SEncoder encoder = {0};
|
||||
tEncoderInit(&encoder, buf, bufLen);
|
||||
|
|
@ -3674,12 +3673,12 @@ int32_t tEncodeTSma(SEncoder *pCoder, const STSma *pSma) {
|
|||
if (tEncodeCStr(pCoder, pSma->tagsFilter) < 0) return -1;
|
||||
}
|
||||
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
|
||||
if (tEncodeI32(pCoder, pSma->vgEpSet[v].vgId) < 0) return -1;
|
||||
if (tEncodeI8(pCoder, pSma->vgEpSet[v].epSet.inUse) < 0) return -1;
|
||||
int8_t numOfEps = pSma->vgEpSet[v].epSet.numOfEps;
|
||||
if (tEncodeI32(pCoder, pSma->pVgEpSet[v].vgId) < 0) return -1;
|
||||
if (tEncodeI8(pCoder, pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
|
||||
int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
|
||||
if (tEncodeI8(pCoder, numOfEps) < 0) return -1;
|
||||
for (int32_t n = 0; n < numOfEps; ++n) {
|
||||
const SEp *pEp = &pSma->vgEpSet[v].epSet.eps[n];
|
||||
const SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
|
||||
if (tEncodeCStr(pCoder, pEp->fqdn) < 0) return -1;
|
||||
if (tEncodeU16(pCoder, pEp->port) < 0) return -1;
|
||||
}
|
||||
|
|
@ -3712,17 +3711,27 @@ int32_t tDecodeTSma(SDecoder *pCoder, STSma *pSma) {
|
|||
} else {
|
||||
pSma->tagsFilter = NULL;
|
||||
}
|
||||
if (pSma->numOfVgroups > 0) {
|
||||
pSma->pVgEpSet = (SVgEpSet *)tDecoderMalloc(pCoder, pSma->numOfVgroups * sizeof(SVgEpSet));
|
||||
if (!pSma->pVgEpSet) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return -1;
|
||||
}
|
||||
|
||||
memset(pSma->pVgEpSet, 0, pSma->numOfVgroups * sizeof(SVgEpSet));
|
||||
|
||||
for (int32_t v = 0; v < pSma->numOfVgroups; ++v) {
|
||||
if (tDecodeI32(pCoder, &pSma->vgEpSet[v].vgId) < 0) return -1;
|
||||
if (tDecodeI8(pCoder, &pSma->vgEpSet[v].epSet.inUse) < 0) return -1;
|
||||
if (tDecodeI8(pCoder, &pSma->vgEpSet[v].epSet.numOfEps) < 0) return -1;
|
||||
int8_t numOfEps = pSma->vgEpSet[v].epSet.numOfEps;
|
||||
if (tDecodeI32(pCoder, &pSma->pVgEpSet[v].vgId) < 0) return -1;
|
||||
if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.inUse) < 0) return -1;
|
||||
if (tDecodeI8(pCoder, &pSma->pVgEpSet[v].epSet.numOfEps) < 0) return -1;
|
||||
int8_t numOfEps = pSma->pVgEpSet[v].epSet.numOfEps;
|
||||
for (int32_t n = 0; n < numOfEps; ++n) {
|
||||
SEp *pEp = &pSma->vgEpSet[v].epSet.eps[n];
|
||||
SEp *pEp = &pSma->pVgEpSet[v].epSet.eps[n];
|
||||
if (tDecodeCStrTo(pCoder, pEp->fqdn) < 0) return -1;
|
||||
if (tDecodeU16(pCoder, &pEp->port) < 0) return -1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
@ -4569,6 +4578,3 @@ void tFreeSMAlterStbRsp(SMAlterStbRsp* pRsp) {
|
|||
taosMemoryFree(pRsp->pMeta);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -316,10 +316,12 @@ typedef struct {
|
|||
int32_t tagsFilterLen;
|
||||
int32_t sqlLen;
|
||||
int32_t astLen;
|
||||
int32_t numOfVgroups;
|
||||
char* expr;
|
||||
char* tagsFilter;
|
||||
char* sql;
|
||||
char* ast;
|
||||
SVgEpSet* pVgEpSet;
|
||||
} SSmaObj;
|
||||
|
||||
typedef struct {
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ static SSdbRow *mndSmaActionDecode(SSdbRaw *pRaw);
|
|||
static int32_t mndSmaActionInsert(SSdb *pSdb, SSmaObj *pSma);
|
||||
static int32_t mndSmaActionDelete(SSdb *pSdb, SSmaObj *pSpSmatb);
|
||||
static int32_t mndSmaActionUpdate(SSdb *pSdb, SSmaObj *pOld, SSmaObj *pNew);
|
||||
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups);
|
||||
static int32_t mndProcessMCreateSmaReq(SRpcMsg *pReq);
|
||||
static int32_t mndProcessMDropSmaReq(SRpcMsg *pReq);
|
||||
static int32_t mndProcessGetSmaReq(SRpcMsg *pReq);
|
||||
|
|
@ -262,6 +263,8 @@ static void *mndBuildVCreateSmaReq(SMnode *pMnode, SVgObj *pVgroup, SSmaObj *pSm
|
|||
req.sliding = pSma->sliding;
|
||||
req.expr = pSma->expr;
|
||||
req.tagsFilter = pSma->tagsFilter;
|
||||
req.numOfVgroups = pSma->numOfVgroups;
|
||||
req.pVgEpSet = pSma->pVgEpSet;
|
||||
|
||||
// get length
|
||||
int32_t ret = 0;
|
||||
|
|
@ -420,6 +423,15 @@ static int32_t mndSetCreateSmaVgroupRedoActions(SMnode *pMnode, STrans *pTrans,
|
|||
mndReleaseDnode(pMnode, pDnode);
|
||||
|
||||
// todo add sma info here
|
||||
SVgEpSet *pVgEpSet = NULL;
|
||||
int32_t numOfVgroups = 0;
|
||||
if (mndSmaGetVgEpSet(pMnode, pDb, &pVgEpSet, &numOfVgroups) != 0) {
|
||||
return -1;
|
||||
}
|
||||
|
||||
pSma->pVgEpSet = pVgEpSet;
|
||||
pSma->numOfVgroups = numOfVgroups;
|
||||
|
||||
int32_t smaContLen = 0;
|
||||
void *pSmaReq = mndBuildVCreateSmaReq(pMnode, pVgroup, pSma, &smaContLen);
|
||||
if (pSmaReq == NULL) return -1;
|
||||
|
|
@ -964,3 +976,52 @@ static void mndCancelGetNextSma(SMnode *pMnode, void *pIter) {
|
|||
SSdb *pSdb = pMnode->pSdb;
|
||||
sdbCancelFetch(pSdb, pIter);
|
||||
}
|
||||
|
||||
static int32_t mndSmaGetVgEpSet(SMnode *pMnode, SDbObj *pDb, SVgEpSet **ppVgEpSet, int32_t *numOfVgroups) {
|
||||
SSdb *pSdb = pMnode->pSdb;
|
||||
SVgObj *pVgroup = NULL;
|
||||
void *pIter = NULL;
|
||||
SVgEpSet *pVgEpSet = NULL;
|
||||
int32_t nAllocVgs = 16;
|
||||
int32_t nVgs = 0;
|
||||
|
||||
pVgEpSet = taosMemoryCalloc(nAllocVgs, sizeof(SVgEpSet));
|
||||
if (!pVgEpSet) {
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return -1;
|
||||
}
|
||||
|
||||
while (1) {
|
||||
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
|
||||
if (pIter == NULL) break;
|
||||
if (pVgroup->dbUid != pDb->uid) {
|
||||
sdbRelease(pSdb, pVgroup);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (nVgs >= nAllocVgs) {
|
||||
void *p = taosMemoryRealloc(pVgEpSet, nAllocVgs * 2 * sizeof(SVgEpSet));
|
||||
if (!p) {
|
||||
taosMemoryFree(pVgEpSet);
|
||||
sdbCancelFetch(pSdb, pIter);
|
||||
sdbRelease(pSdb, pVgroup);
|
||||
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||
return -1;
|
||||
}
|
||||
pVgEpSet = (SVgEpSet *)p;
|
||||
nAllocVgs *= 2;
|
||||
}
|
||||
|
||||
(pVgEpSet + nVgs)->vgId = pVgroup->vgId;
|
||||
(pVgEpSet + nVgs)->epSet = mndGetVgroupEpset(pMnode, pVgroup);
|
||||
|
||||
++nVgs;
|
||||
|
||||
sdbRelease(pSdb, pVgroup);
|
||||
}
|
||||
|
||||
*ppVgEpSet = pVgEpSet;
|
||||
*numOfVgroups = nVgs;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -773,7 +773,9 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo*
|
|||
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SLimit* pLimit, SLimit* pSlimit, SNode* pCondition, SExecTaskInfo* pTaskInfo);
|
||||
SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExprInfo* pExprInfo, int32_t numOfCols,
|
||||
SArray* pIndexMap, SExecTaskInfo* pTaskInfo);
|
||||
|
||||
SOperatorInfo* createMultiwaySortMergeOperatorInfo(SOperatorInfo** downStreams, int32_t numStreams,
|
||||
SSDataBlock* pResBlock, SArray* pSortInfo, SArray* pColMatchColInfo,
|
||||
SExecTaskInfo* pTaskInfo);
|
||||
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
|
||||
|
||||
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
|
||||
|
|
|
|||
|
|
@ -4517,6 +4517,19 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
|
|||
extractColMatchInfo(pSortPhyNode->pTargets, pDescNode, &numOfOutputCols, pTaskInfo, COL_MATCH_FROM_SLOT_ID);
|
||||
|
||||
pOptr = createSortOperatorInfo(ops[0], pResBlock, info, pExprInfo, numOfCols, pColList, pTaskInfo);
|
||||
} else if (QUERY_NODE_PHYSICAL_PLAN_MERGE == type) {
|
||||
SMergePhysiNode* pMergePhyNode = (SMergePhysiNode*)pPhyNode;
|
||||
|
||||
SDataBlockDescNode* pDescNode = pPhyNode->pOutputDataBlockDesc;
|
||||
SSDataBlock* pResBlock = createResDataBlock(pDescNode);
|
||||
|
||||
SArray* sortInfo = createSortInfo(pMergePhyNode->pMergeKeys);
|
||||
int32_t numOfOutputCols = 0;
|
||||
SArray* pColList = NULL;
|
||||
//extractColMatchInfo(pMergePhyNode->pTargets, pDescNode, &numOfOutputCols, pTaskInfo, COL_MATCH_FROM_SLOT_ID);
|
||||
|
||||
|
||||
pOptr = createMultiwaySortMergeOperatorInfo(ops, size, pResBlock, sortInfo, pColList, pTaskInfo);
|
||||
} else if (QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW == type) {
|
||||
SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
|
||||
|
||||
|
|
|
|||
|
|
@ -13,8 +13,8 @@
|
|||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "tdatablock.h"
|
||||
#include "executorimpl.h"
|
||||
#include "tdatablock.h"
|
||||
|
||||
static SSDataBlock* doSort(SOperatorInfo* pOperator);
|
||||
static int32_t doOpenSortOperator(SOperatorInfo* pOperator);
|
||||
|
|
@ -22,8 +22,9 @@ static int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uin
|
|||
|
||||
static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput);
|
||||
|
||||
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExprInfo* pExprInfo, int32_t numOfCols,
|
||||
SArray* pColMatchColInfo, SExecTaskInfo* pTaskInfo) {
|
||||
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo,
|
||||
SExprInfo* pExprInfo, int32_t numOfCols, SArray* pColMatchColInfo,
|
||||
SExecTaskInfo* pTaskInfo) {
|
||||
SSortOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortOperatorInfo));
|
||||
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
||||
int32_t rowSize = pResBlock->info.rowSize;
|
||||
|
|
@ -48,12 +49,12 @@ SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pR
|
|||
pOperator->info = pInfo;
|
||||
|
||||
// lazy evaluation for the following parameter since the input datablock is not known till now.
|
||||
// pInfo->bufPageSize = rowSize < 1024 ? 1024 * 2 : rowSize * 2; // there are headers, so pageSize = rowSize + header
|
||||
// pInfo->sortBufSize = pInfo->bufPageSize * 16; // TODO dynamic set the available sort buffer
|
||||
// pInfo->bufPageSize = rowSize < 1024 ? 1024 * 2 : rowSize * 2; // there are headers, so pageSize = rowSize +
|
||||
// header pInfo->sortBufSize = pInfo->bufPageSize * 16; // TODO dynamic set the available sort buffer
|
||||
|
||||
pOperator->pTaskInfo = pTaskInfo;
|
||||
pOperator->fpSet =
|
||||
createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroyOrderOperatorInfo, NULL, NULL, getExplainExecInfo);
|
||||
pOperator->fpSet = createOperatorFpSet(doOpenSortOperator, doSort, NULL, NULL, destroyOrderOperatorInfo, NULL, NULL,
|
||||
getExplainExecInfo);
|
||||
|
||||
int32_t code = appendDownstream(pOperator, &downstream, 1);
|
||||
return pOperator;
|
||||
|
|
@ -80,7 +81,8 @@ void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle) {
|
|||
pBlock->info.rows += 1;
|
||||
}
|
||||
|
||||
SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity, SArray* pColMatchInfo) {
|
||||
SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity,
|
||||
SArray* pColMatchInfo) {
|
||||
blockDataCleanup(pDataBlock);
|
||||
ASSERT(taosArrayGetSize(pColMatchInfo) == pDataBlock->info.numOfCols);
|
||||
|
||||
|
|
@ -132,7 +134,8 @@ void applyScalarFunction(SSDataBlock* pBlock, void* param) {
|
|||
SOperatorInfo* pOperator = param;
|
||||
SSortOperatorInfo* pSort = pOperator->info;
|
||||
if (pOperator->pExpr != NULL) {
|
||||
int32_t code = projectApplyFunctions(pOperator->pExpr, pBlock, pBlock, pSort->binfo.pCtx, pOperator->numOfExprs, NULL);
|
||||
int32_t code =
|
||||
projectApplyFunctions(pOperator->pExpr, pBlock, pBlock, pSort->binfo.pCtx, pOperator->numOfExprs, NULL);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
longjmp(pOperator->pTaskInfo->env, code);
|
||||
}
|
||||
|
|
@ -150,8 +153,8 @@ int32_t doOpenSortOperator(SOperatorInfo* pOperator) {
|
|||
pInfo->startTs = taosGetTimestampUs();
|
||||
|
||||
// pInfo->binfo.pRes is not equalled to the input datablock.
|
||||
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_SINGLESOURCE_SORT,
|
||||
-1, -1, NULL, pTaskInfo->id.str);
|
||||
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_SINGLESOURCE_SORT, -1, -1,
|
||||
NULL, pTaskInfo->id.str);
|
||||
|
||||
tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, applyScalarFunction, pOperator);
|
||||
|
||||
|
|
@ -186,7 +189,8 @@ SSDataBlock* doSort(SOperatorInfo* pOperator) {
|
|||
longjmp(pTaskInfo->env, code);
|
||||
}
|
||||
|
||||
SSDataBlock* pBlock = getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pInfo->pColMatchInfo);
|
||||
SSDataBlock* pBlock =
|
||||
getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pInfo->pColMatchInfo);
|
||||
|
||||
if (pBlock != NULL) {
|
||||
pOperator->resultInfo.totalRows += pBlock->info.rows;
|
||||
|
|
@ -215,3 +219,140 @@ int32_t getExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t*
|
|||
*len = sizeof(SSortExecInfo);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
typedef struct SMultiwaySortMergeOperatorInfo {
|
||||
SOptrBasicInfo binfo;
|
||||
|
||||
int32_t bufPageSize;
|
||||
uint32_t sortBufSize; // max buffer size for in-memory sort
|
||||
|
||||
SArray* pSortInfo;
|
||||
SSortHandle* pSortHandle;
|
||||
SArray* pColMatchInfo; // for index map from table scan output
|
||||
|
||||
int64_t startTs; // sort start time
|
||||
} SMultiwaySortMergeOperatorInfo;
|
||||
|
||||
int32_t doOpenMultiwaySortMergeOperator(SOperatorInfo* pOperator) {
|
||||
SMultiwaySortMergeOperatorInfo* pInfo = pOperator->info;
|
||||
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
||||
|
||||
if (OPTR_IS_OPENED(pOperator)) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
pInfo->startTs = taosGetTimestampUs();
|
||||
|
||||
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
|
||||
|
||||
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, pInfo->pColMatchInfo, SORT_MULTISOURCE_MERGE,
|
||||
pInfo->bufPageSize, numOfBufPage, NULL, pTaskInfo->id.str);
|
||||
|
||||
tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock, NULL, NULL);
|
||||
|
||||
for (int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
|
||||
SSortSource ps = {0};
|
||||
ps.param = pOperator->pDownstream[i];
|
||||
tsortAddSource(pInfo->pSortHandle, &ps);
|
||||
}
|
||||
|
||||
int32_t code = tsortOpen(pInfo->pSortHandle);
|
||||
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
longjmp(pTaskInfo->env, terrno);
|
||||
}
|
||||
|
||||
pOperator->cost.openCost = (taosGetTimestampUs() - pInfo->startTs) / 1000.0;
|
||||
pOperator->status = OP_RES_TO_RETURN;
|
||||
|
||||
OPTR_SET_OPENED(pOperator);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
SSDataBlock* doMultiwaySortMerge(SOperatorInfo* pOperator) {
|
||||
if (pOperator->status == OP_EXEC_DONE) {
|
||||
return NULL;
|
||||
}
|
||||
|
||||
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
||||
SMultiwaySortMergeOperatorInfo* pInfo = pOperator->info;
|
||||
|
||||
int32_t code = pOperator->fpSet._openFn(pOperator);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
longjmp(pTaskInfo->env, code);
|
||||
}
|
||||
|
||||
SSDataBlock* pBlock =
|
||||
getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pOperator->resultInfo.capacity, pInfo->pColMatchInfo);
|
||||
|
||||
if (pBlock != NULL) {
|
||||
pOperator->resultInfo.totalRows += pBlock->info.rows;
|
||||
} else {
|
||||
doSetOperatorCompleted(pOperator);
|
||||
}
|
||||
return pBlock;
|
||||
}
|
||||
|
||||
void destroyMultiwaySortMergeOperatorInfo(void* param, int32_t numOfOutput) {
|
||||
SMultiwaySortMergeOperatorInfo * pInfo = (SMultiwaySortMergeOperatorInfo*)param;
|
||||
pInfo->binfo.pRes = blockDataDestroy(pInfo->binfo.pRes);
|
||||
|
||||
taosArrayDestroy(pInfo->pSortInfo);
|
||||
taosArrayDestroy(pInfo->pColMatchInfo);
|
||||
}
|
||||
|
||||
int32_t getMultiwaySortMergeExplainExecInfo(SOperatorInfo* pOptr, void** pOptrExplain, uint32_t* len) {
|
||||
ASSERT(pOptr != NULL);
|
||||
SSortExecInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortExecInfo));
|
||||
|
||||
SMultiwaySortMergeOperatorInfo* pOperatorInfo = (SMultiwaySortMergeOperatorInfo*)pOptr->info;
|
||||
|
||||
*pInfo = tsortGetSortExecInfo(pOperatorInfo->pSortHandle);
|
||||
*pOptrExplain = pInfo;
|
||||
*len = sizeof(SSortExecInfo);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
SOperatorInfo* createMultiwaySortMergeOperatorInfo(SOperatorInfo** downStreams, int32_t numStreams,
|
||||
SSDataBlock* pResBlock, SArray* pSortInfo, SArray* pColMatchColInfo,
|
||||
SExecTaskInfo* pTaskInfo) {
|
||||
SMultiwaySortMergeOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SMultiwaySortMergeOperatorInfo));
|
||||
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
||||
int32_t rowSize = pResBlock->info.rowSize;
|
||||
|
||||
if (pInfo == NULL || pOperator == NULL || rowSize > 100 * 1024 * 1024) {
|
||||
goto _error;
|
||||
}
|
||||
|
||||
pInfo->binfo.pRes = pResBlock;
|
||||
|
||||
initResultSizeInfo(pOperator, 1024);
|
||||
|
||||
pInfo->pSortInfo = pSortInfo;
|
||||
pInfo->pColMatchInfo = pColMatchColInfo;
|
||||
pOperator->name = "MultiwaySortMerge";
|
||||
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_MERGE;
|
||||
pOperator->blocking = true;
|
||||
pOperator->status = OP_NOT_OPENED;
|
||||
pOperator->info = pInfo;
|
||||
|
||||
pInfo->bufPageSize = rowSize < 1024 ? 1024 : rowSize * 2;
|
||||
pInfo->sortBufSize = pInfo->bufPageSize * 16;
|
||||
|
||||
pOperator->pTaskInfo = pTaskInfo;
|
||||
pOperator->fpSet =
|
||||
createOperatorFpSet(doOpenMultiwaySortMergeOperator, doMultiwaySortMerge, NULL, NULL,
|
||||
destroyMultiwaySortMergeOperatorInfo, NULL, NULL, getMultiwaySortMergeExplainExecInfo);
|
||||
|
||||
int32_t code = appendDownstream(pOperator, downStreams, numStreams);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
goto _error;
|
||||
}
|
||||
return pOperator;
|
||||
|
||||
_error:
|
||||
pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
|
||||
taosMemoryFree(pInfo);
|
||||
taosMemoryFree(pOperator);
|
||||
return NULL;
|
||||
}
|
||||
|
|
@ -171,8 +171,10 @@ static int32_t sifInitJsonParam(SNode *node, SIFParam *param, SIFCtx *ctx) {
|
|||
param->colId = l->colId;
|
||||
param->colValType = l->node.resType.type;
|
||||
memcpy(param->dbName, l->dbName, sizeof(l->dbName));
|
||||
|
||||
#pragma GCC diagnostic push
|
||||
#pragma GCC diagnostic ignored "-Wformat-overflow"
|
||||
sprintf(param->colName, "%s_%s", l->colName, r->literal);
|
||||
#pragma GCC diagnostic pop
|
||||
param->colValType = r->typeData;
|
||||
return 0;
|
||||
// memcpy(param->colName, l->colName, sizeof(l->colName));
|
||||
|
|
|
|||
|
|
@ -1006,7 +1006,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
|
|||
}
|
||||
|
||||
SSchema* pTagSchema = &pSchema[pCxt->tags.boundColumns[i]];
|
||||
char* tmpTokenBuf = taosMemoryCalloc(1, sToken.n); // this can be optimize with parse column
|
||||
char* tmpTokenBuf = taosMemoryCalloc(1, sToken.n); //todo this can be optimize with parse column
|
||||
code = checkAndTrimValue(&sToken, tmpTokenBuf, &pCxt->msg);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
taosMemoryFree(tmpTokenBuf);
|
||||
|
|
@ -1018,7 +1018,11 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
|
|||
taosMemoryFree(tmpTokenBuf);
|
||||
goto end;
|
||||
}
|
||||
if(isNullStr(&sToken)) {
|
||||
code = tTagNew(pTagVals, 1, true, &pTag);
|
||||
} else {
|
||||
code = parseJsontoTagData(sToken.z, pTagVals, &pTag, &pCxt->msg);
|
||||
}
|
||||
taosMemoryFree(tmpTokenBuf);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
goto end;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,113 @@
|
|||
###################################################################
|
||||
# Copyright (c) 2016 by TAOS Technologies, Inc.
|
||||
# All rights reserved.
|
||||
#
|
||||
# This file is proprietary and confidential to TAOS Technologies.
|
||||
# No part of this file may be reproduced, stored, transmitted,
|
||||
# disclosed or used in any form or by any means other than as
|
||||
# expressly provided by the written permission from Jianhui Tao
|
||||
#
|
||||
###################################################################
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import random
|
||||
import string
|
||||
from util.log import *
|
||||
from util.cases import *
|
||||
from util.sql import *
|
||||
|
||||
class TDTestCase:
|
||||
def init(self, conn, logSql):
|
||||
tdLog.debug("start to execute %s" % __file__)
|
||||
tdSql.init(conn.cursor())
|
||||
|
||||
def get_long_name(self, length, mode="mixed"):
|
||||
"""
|
||||
generate long name
|
||||
mode could be numbers/letters/letters_mixed/mixed
|
||||
"""
|
||||
if mode == "numbers":
|
||||
population = string.digits
|
||||
elif mode == "letters":
|
||||
population = string.ascii_letters.lower()
|
||||
elif mode == "letters_mixed":
|
||||
population = string.ascii_letters.upper() + string.ascii_letters.lower()
|
||||
else:
|
||||
population = string.ascii_letters.lower() + string.digits
|
||||
return "".join(random.choices(population, k=length))
|
||||
|
||||
def __create_tb(self,dbname,stbname,tbname,comment):
|
||||
tdSql.execute(f'create database if not exists {dbname}')
|
||||
tdSql.execute(f'use {dbname}')
|
||||
tdSql.execute(
|
||||
f'create table {stbname} (ts timestamp,c0 int) tags(t0 int) ')
|
||||
tdSql.execute(
|
||||
f'create table {tbname} using {stbname} tags(1) comment "{comment}"')
|
||||
def __create_normaltb(self,dbname,tbname,comment):
|
||||
tdSql.execute(f'create database if not exists {dbname}')
|
||||
tdSql.execute(f'use {dbname}')
|
||||
tdSql.execute(
|
||||
f'create table {tbname} (ts timestamp,c0 int) comment "{comment}"')
|
||||
|
||||
def check_comment(self):
|
||||
dbname = self.get_long_name(length=10, mode="letters")
|
||||
ntbname = self.get_long_name(length=5, mode="letters")
|
||||
|
||||
# create normal table with comment
|
||||
comment = self.get_long_name(length=10, mode="letters")
|
||||
self.__create_normaltb(dbname,ntbname,comment)
|
||||
ntb_kv_list = tdSql.getResult("show tables")
|
||||
print(ntb_kv_list)
|
||||
tdSql.checkEqual(ntb_kv_list[0][8], comment)
|
||||
tdSql.error('alter table {ntbname} comment "test1"')
|
||||
tdSql.execute(f'drop database {dbname}')
|
||||
|
||||
# max length(1024)
|
||||
comment = self.get_long_name(length=1024, mode="letters")
|
||||
self.__create_normaltb(dbname,ntbname,comment)
|
||||
ntb_kv_list = tdSql.getResult("show tables")
|
||||
tdSql.checkEqual(ntb_kv_list[0][8], comment)
|
||||
tdSql.execute(f'drop database {dbname}')
|
||||
|
||||
# error overlength
|
||||
comment = self.get_long_name(length=1025, mode="letters")
|
||||
tdSql.execute(f'create database if not exists {dbname}')
|
||||
tdSql.execute(f'use {dbname}')
|
||||
tdSql.error(f"create table ntb (ts timestamp,c0 int) comment '{comment}'")
|
||||
tdSql.execute(f'drop database {dbname}')
|
||||
|
||||
# create child table with comment
|
||||
comment = self.get_long_name(length=10, mode="letters")
|
||||
stbname = self.get_long_name(length=5, mode="letters")
|
||||
tbname = self.get_long_name(length=3, mode="letters")
|
||||
self.__create_tb(dbname,stbname,tbname,comment)
|
||||
ntb_kv_list = tdSql.getResult("show tables")
|
||||
tdSql.checkEqual(ntb_kv_list[0][8], comment)
|
||||
tdSql.error(f'alter table {tbname} comment "test1"')
|
||||
tdSql.execute(f'drop database {dbname}')
|
||||
|
||||
# max length 1024
|
||||
comment = self.get_long_name(length=1024, mode="letters")
|
||||
self.__create_tb(dbname,ntbname,comment)
|
||||
ntb_kv_list = tdSql.getResult("show tables")
|
||||
tdSql.checkEqual(ntb_kv_list[0][8], comment)
|
||||
tdSql.execute(f'drop database {dbname}')
|
||||
|
||||
# error overlength
|
||||
comment = self.get_long_name(length=1025, mode="letters")
|
||||
tdSql.execute(f'create database if not exists {dbname}')
|
||||
tdSql.execute(f'use {dbname}')
|
||||
tdSql.execute(f"create table stb (ts timestamp,c0 int) tags(t0 int)")
|
||||
tdSql.error(f'create table stb_1 us stb tags(1) comment "{comment}"')
|
||||
tdSql.execute(f'drop database {dbname}')
|
||||
|
||||
def run(self):
|
||||
self.check_comment()
|
||||
|
||||
def stop(self):
|
||||
tdSql.close()
|
||||
tdLog.success("%s successfully executed" % __file__)
|
||||
|
||||
tdCases.addWindows(__file__, TDTestCase())
|
||||
tdCases.addLinux(__file__, TDTestCase())
|
||||
|
|
@ -61,6 +61,8 @@ class TDTestCase:
|
|||
# test invalidate json
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('\"efwewf\"')")
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('3333')")
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags(76)")
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags(hell)")
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('33.33')")
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('false')")
|
||||
tdSql.error("CREATE TABLE if not exists jsons1_14 using jsons1 tags('[1,true]')")
|
||||
|
|
@ -128,7 +130,7 @@ class TDTestCase:
|
|||
tdSql.error("select ceil(jtag) from jsons1")
|
||||
#
|
||||
# # test select normal column
|
||||
tdSql.query("select dataint from jsons1")
|
||||
tdSql.query("select dataint from jsons1 order by dataint")
|
||||
tdSql.checkRows(9)
|
||||
tdSql.checkData(1, 0, 1)
|
||||
|
||||
|
|
@ -137,9 +139,9 @@ class TDTestCase:
|
|||
tdSql.checkRows(9)
|
||||
tdSql.query("select jtag from jsons1")
|
||||
tdSql.checkRows(13)
|
||||
tdSql.query("select jtag from jsons1 where jtag is null")
|
||||
# tdSql.query("select jtag from jsons1 where jtag is null")
|
||||
# tdSql.checkRows(5)
|
||||
tdSql.query("select jtag from jsons1 where jtag is not null")
|
||||
# tdSql.query("select jtag from jsons1 where jtag is not null")
|
||||
# tdSql.checkRows(8)
|
||||
|
||||
# test jtag is NULL
|
||||
|
|
@ -259,12 +261,6 @@ class TDTestCase:
|
|||
# tdSql.query("select * from jsons1 where jtag->'tag1'=null") # only json suport =null. This synatx will change later.
|
||||
# tdSql.checkRows(1)
|
||||
#
|
||||
# # where json is null
|
||||
tdSql.query("select * from jsons1 where jtag is null")
|
||||
# tdSql.checkRows(1)
|
||||
tdSql.query("select * from jsons1 where jtag is not null")
|
||||
# tdSql.checkRows(8)
|
||||
#
|
||||
# # where json key is null
|
||||
# tdSql.query("select * from jsons1 where jtag->'tag_no_exist'=3")
|
||||
# tdSql.checkRows(0)
|
||||
|
|
@ -358,25 +354,28 @@ class TDTestCase:
|
|||
# tdSql.checkRows(0)
|
||||
#
|
||||
# # test join
|
||||
# tdSql.execute("create table if not exists jsons2(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
|
||||
# tdSql.execute("insert into jsons2_1 using jsons2 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 2, false, 'json2', '你是2')")
|
||||
# tdSql.execute("insert into jsons2_2 using jsons2 tags('{\"tag1\":5,\"tag2\":null}') values (1591060628000, 2, true, 'json2', 'sss')")
|
||||
#
|
||||
# tdSql.execute("create table if not exists jsons3(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
|
||||
# tdSql.execute("insert into jsons3_1 using jsons3 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 3, false, 'json3', '你是3')")
|
||||
# tdSql.execute("insert into jsons3_2 using jsons3 tags('{\"tag1\":5,\"tag2\":\"beijing\"}') values (1591060638000, 2, true, 'json3', 'sss')")
|
||||
# tdSql.query("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'")
|
||||
# tdSql.checkData(0, 0, "sss")
|
||||
# tdSql.checkData(0, 2, "true")
|
||||
#
|
||||
# res = tdSql.getColNameList("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'")
|
||||
# cname_list = []
|
||||
# cname_list.append("sss")
|
||||
# cname_list.append("33")
|
||||
# cname_list.append("a.jtag->'tag3'")
|
||||
# tdSql.checkColNameList(res, cname_list)
|
||||
tdSql.execute("create table if not exists jsons2(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
|
||||
tdSql.execute("insert into jsons2_1 using jsons2 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 2, false, 'json2', '你是2')")
|
||||
tdSql.execute("insert into jsons2_2 using jsons2 tags('{\"tag1\":5,\"tag2\":null}') values (1591060628000, 2, true, 'json2', 'sss')")
|
||||
|
||||
tdSql.execute("create table if not exists jsons3(ts timestamp, dataInt int, dataBool bool, dataStr nchar(50), dataStrBin binary(150)) tags(jtag json)")
|
||||
tdSql.execute("insert into jsons3_1 using jsons3 tags('{\"tag1\":\"fff\",\"tag2\":5, \"tag3\":true}') values(1591060618000, 3, false, 'json3', '你是3')")
|
||||
tdSql.execute("insert into jsons3_2 using jsons3 tags('{\"tag1\":5,\"tag2\":\"beijing\"}') values (1591060638000, 2, true, 'json3', 'sss')")
|
||||
tdSql.query("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'")
|
||||
tdSql.checkData(0, 0, "sss")
|
||||
tdSql.checkData(0, 2, "true")
|
||||
|
||||
res = tdSql.getColNameList("select 'sss',33,a.jtag->'tag3' from jsons2 a,jsons3 b where a.ts=b.ts and a.jtag->'tag1'=b.jtag->'tag1'")
|
||||
cname_list = []
|
||||
cname_list.append("'sss'")
|
||||
cname_list.append("33")
|
||||
cname_list.append("a.jtag->'tag3'")
|
||||
tdSql.checkColNameList(res, cname_list)
|
||||
#
|
||||
# test group by & order by json tag
|
||||
# tdSql.error("select count(*) from jsons1 group by jtag")
|
||||
# tdSql.error("select count(*) from jsons1 partition by jtag")
|
||||
# tdSql.error("select count(*) from jsons1 group by jtag order by jtag")
|
||||
tdSql.error("select count(*) from jsons1 group by jtag->'tag1' order by jtag->'tag2'")
|
||||
tdSql.error("select count(*) from jsons1 group by jtag->'tag1' order by jtag")
|
||||
tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' order by jtag->'tag1' desc")
|
||||
|
|
@ -432,14 +431,14 @@ class TDTestCase:
|
|||
tdSql.checkData(10, 1, '"femail"')
|
||||
|
||||
# test having
|
||||
# tdSql.query("select stddev(dataint) from jsons1 group by jtag->'tag1' having stddev(dataint) > 0")
|
||||
# tdSql.checkRows(2)
|
||||
tdSql.query("select count(*),jtag->'tag1' from jsons1 group by jtag->'tag1' having count(*) > 1")
|
||||
tdSql.checkRows(3)
|
||||
|
||||
# subquery with json tag
|
||||
tdSql.query("select * from (select jtag, dataint from jsons1) order by dataint")
|
||||
tdSql.checkRows(11)
|
||||
tdSql.checkData(1, 1, 1)
|
||||
tdSql.checkData(2, 0, '{"tag1":5,"tag2":"beijing"}')
|
||||
tdSql.checkData(5, 0, '{"tag1":false,"tag2":"beijing"}')
|
||||
|
||||
# tdSql.query("select jtag->'tag1' from (select jtag->'tag1', dataint from jsons1)")
|
||||
# tdSql.checkRows(11)
|
||||
|
|
@ -457,16 +456,18 @@ class TDTestCase:
|
|||
# tdSql.checkData(1, 2, '"femail"')
|
||||
#
|
||||
# # union all
|
||||
# tdSql.error("select jtag->'tag1' from jsons1 union all select jtag->'tag2' from jsons2")
|
||||
# tdSql.error("select jtag->'tag1' from jsons1_1 union all select jtag->'tag2' from jsons2_1")
|
||||
#
|
||||
# tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag1' from jsons2_1")
|
||||
# tdSql.checkRows(2)
|
||||
# tdSql.query("select dataint,jtag->'tag1',tbname from jsons1 union all select dataint,jtag->'tag1',tbname from jsons2")
|
||||
# tdSql.checkRows(13)
|
||||
# tdSql.query("select dataint,jtag,tbname from jsons1 union all select dataint,jtag,tbname from jsons2")
|
||||
# tdSql.checkRows(13)
|
||||
#
|
||||
tdSql.query("select jtag->'tag1' from jsons1 union all select jtag->'tag2' from jsons2")
|
||||
tdSql.checkRows(17)
|
||||
tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag2' from jsons2_1")
|
||||
tdSql.checkRows(2)
|
||||
|
||||
tdSql.query("select jtag->'tag1' from jsons1_1 union all select jtag->'tag1' from jsons2_1")
|
||||
tdSql.checkRows(2)
|
||||
tdSql.query("select dataint,jtag->'tag1',tbname from jsons1 union all select dataint,jtag->'tag1',tbname from jsons2")
|
||||
tdSql.checkRows(13)
|
||||
tdSql.query("select dataint,jtag,tbname from jsons1 union all select dataint,jtag,tbname from jsons2")
|
||||
tdSql.checkRows(13)
|
||||
|
||||
# #show create table
|
||||
# tdSql.query("show create table jsons1")
|
||||
# tdSql.checkData(0, 1, 'CREATE TABLE `jsons1` (`ts` TIMESTAMP,`dataint` INT,`databool` BOOL,`datastr` NCHAR(50),`datastrbin` BINARY(150)) TAGS (`jtag` JSON)')
|
||||
|
|
@ -528,9 +529,9 @@ class TDTestCase:
|
|||
tdSql.execute("CREATE TABLE if not exists jsons1_20 using jsons1 tags(NULL)")
|
||||
tdSql.query("select jtag from jsons1_20")
|
||||
tdSql.checkData(0, 0, None)
|
||||
# tdSql.execute("insert into jsons1_21 using jsons1 tags(NULL) values(1591061628000, 11, false, '你就会','')")
|
||||
# tdSql.query("select jtag from jsons1_21")
|
||||
# tdSql.checkData(0, 0, None)
|
||||
tdSql.execute("insert into jsons1_21 using jsons1 tags(NULL) values(1591061628000, 11, false, '你就会','')")
|
||||
tdSql.query("select jtag from jsons1_21")
|
||||
tdSql.checkData(0, 0, None)
|
||||
#
|
||||
# #test TD-12389
|
||||
tdSql.query("describe jsons1")
|
||||
|
|
|
|||
|
|
@ -282,7 +282,7 @@ class TDTestCase:
|
|||
tdSql.execute("create topic %s as select ts, c1, c2, t1, t2 from %s.%s" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s" %(columnTopicFromNtb, parameterDict['dbName'], ntbName))
|
||||
|
||||
tsLog.info("======== super table test:")
|
||||
tdLog.info("======== super table test:")
|
||||
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
|
||||
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
|
|
@ -316,7 +316,7 @@ class TDTestCase:
|
|||
tdSql.query("alter table %s.%s add tag t3 int"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.query("alter table %s.%s add tag t4 float"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
|
||||
tsLog.info("======== normal table test:")
|
||||
tdLog.info("======== normal table test:")
|
||||
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
|
||||
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName))
|
||||
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName))
|
||||
|
|
@ -419,7 +419,7 @@ class TDTestCase:
|
|||
tdSql.execute("create topic %s as select ts, c1, c2, t1, t2 from %s.%s where c3 > 3 and c4 like 'abc' and t3 = 5 and t4 = 'beijing'" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.execute("create topic %s as select ts, c1, c2 from %s.%s where c3 > 3 and c4 like 'abc'" %(columnTopicFromNtb, parameterDict['dbName'], ntbName))
|
||||
|
||||
tsLog.info("======== super table test:")
|
||||
tdLog.info("======== super table test:")
|
||||
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
|
||||
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
|
|
@ -457,7 +457,7 @@ class TDTestCase:
|
|||
tdSql.query("alter table %s.%s add column c5 int"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.query("alter table %s.%s add tag t5 float"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
|
||||
tsLog.info("======== normal table test:")
|
||||
tdLog.info("======== normal table test:")
|
||||
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
|
||||
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName))
|
||||
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName))
|
||||
|
|
@ -566,7 +566,7 @@ class TDTestCase:
|
|||
tdSql.execute("create topic %s as select * from %s.%s" %(columnTopicFromStb, parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.execute("create topic %s as select * from %s.%s " %(columnTopicFromNtb, parameterDict['dbName'], ntbName))
|
||||
|
||||
tsLog.info("======== super table test:")
|
||||
tdLog.info("======== super table test:")
|
||||
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
|
||||
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
|
|
@ -601,7 +601,7 @@ class TDTestCase:
|
|||
tdSql.query("alter table %s.%s add column c6 int"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
tdSql.query("alter table %s.%s add tag t6 float"%(parameterDict['dbName'], parameterDict['stbName']))
|
||||
|
||||
tsLog.info("======== normal table test:")
|
||||
tdLog.info("======== normal table test:")
|
||||
# alter actions prohibited: drop column/tag, modify column/tag type, rename column/tag included in topic
|
||||
tdSql.error("alter table %s.%s drop column c1"%(parameterDict['dbName'], ntbName))
|
||||
tdSql.error("alter table %s.%s drop column c2"%(parameterDict['dbName'], ntbName))
|
||||
|
|
@ -687,8 +687,8 @@ class TDTestCase:
|
|||
tdLog.info("cfgPath: %s" % cfgPath)
|
||||
|
||||
self.tmqCase1(cfgPath, buildPath)
|
||||
self.tmqCase2(cfgPath, buildPath)
|
||||
self.tmqCase3(cfgPath, buildPath)
|
||||
# self.tmqCase2(cfgPath, buildPath)
|
||||
# self.tmqCase3(cfgPath, buildPath)
|
||||
|
||||
def stop(self):
|
||||
tdSql.close()
|
||||
|
|
|
|||
|
|
@ -21,6 +21,7 @@ python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
|
|||
#python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
|
||||
python3 ./test.py -f 1-insert/alter_stable.py
|
||||
python3 ./test.py -f 1-insert/alter_table.py
|
||||
# python3 ./test.py -f 1-inerst/create_table_comment.py
|
||||
python3 ./test.py -f 2-query/between.py
|
||||
python3 ./test.py -f 2-query/distinct.py
|
||||
python3 ./test.py -f 2-query/varchar.py
|
||||
|
|
@ -57,6 +58,7 @@ python3 ./test.py -f 2-query/To_unixtimestamp.py
|
|||
python3 ./test.py -f 2-query/timetruncate.py
|
||||
python3 ./test.py -f 2-query/diff.py
|
||||
python3 ./test.py -f 2-query/Timediff.py
|
||||
python3 ./test.py -f 2-query/json_tag.py
|
||||
|
||||
python3 ./test.py -f 2-query/top.py
|
||||
python3 ./test.py -f 2-query/bottom.py
|
||||
|
|
|
|||
Loading…
Reference in New Issue