Skip to content

T
TDengine

taosdata / TDengine
大约 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
提交 a2136fd3 编写于 作者: W wenzhouwww@live.cn
浏览文件
操作

Merge branch '3.0' of github.com:taosdata/TDengine into 3.0

上级 ef659e47 8bd24a57
展开全部 隐藏空白更改
内联 并排
Showing with 1201 addition and 151 deletion
include/libs/scalar/scalar.h
浏览文件 @ a2136fd3
...@@ -103,6 +103,7 @@ int32_t minScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam * ...@@ -103,6 +103,7 @@ int32_t minScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *
int32_t maxScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t maxScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t avgScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t avgScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput); int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
int32_t leastSQRScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput);
#ifdef __cplusplus #ifdef __cplusplus
} }
......
source/dnode/vnode/src/tsdb/tsdbCacheRead.c
浏览文件 @ a2136fd3
...@@ -177,7 +177,6 @@ int32_t tsdbRetrieveLastRow(void* pReader, SSDataBlock* pResBlock, const int32_t ...@@ -177,7 +177,6 @@ int32_t tsdbRetrieveLastRow(void* pReader, SSDataBlock* pResBlock, const int32_t
saveOneRow(pRow, pResBlock, pr, slotIds); saveOneRow(pRow, pResBlock, pr, slotIds);
taosArrayPush(pTableUidList, &pKeyInfo->uid); taosArrayPush(pTableUidList, &pKeyInfo->uid);
// taosMemoryFree(pRow);
tsdbCacheRelease(lruCache, h); tsdbCacheRelease(lruCache, h);
pr->tableIndex += 1; pr->tableIndex += 1;
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ a2136fd3
...@@ -830,9 +830,8 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI ...@@ -830,9 +830,8 @@ static int32_t doLoadFileBlockData(STsdbReader* pReader, SDataBlockIter* pBlockI
SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo; SBlockLoadSuppInfo* pSupInfo = &pReader->suppInfo;
SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo; SFileBlockDumpInfo* pDumpInfo = &pReader->status.fBlockDumpInfo;
uint8_t *pb = NULL, *pb1 = NULL;
int32_t code = tsdbReadColData(pReader->pFileReader, &pBlockScanInfo->blockIdx, pBlock, pSupInfo->colIds, numOfCols, int32_t code = tsdbReadColData(pReader->pFileReader, &pBlockScanInfo->blockIdx, pBlock, pSupInfo->colIds, numOfCols,
pBlockData, &pb, &pb1); pBlockData, NULL, NULL);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
...@@ -3007,11 +3006,14 @@ SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) { ...@@ -3007,11 +3006,14 @@ SArray* tsdbRetrieveDataBlock(STsdbReader* pReader, SArray* pIdList) {
code = doLoadFileBlockData(pReader, &pStatus->blockIter, pBlockScanInfo, &pStatus->fileBlockData); code = doLoadFileBlockData(pReader, &pStatus->blockIter, pBlockScanInfo, &pStatus->fileBlockData);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
tBlockDataClear(&pStatus->fileBlockData);
terrno = code; terrno = code;
return NULL; return NULL;
} }
copyBlockDataToSDataBlock(pReader, pBlockScanInfo); copyBlockDataToSDataBlock(pReader, pBlockScanInfo);
tBlockDataClear(&pStatus->fileBlockData);
return pReader->pResBlock->pDataBlock; return pReader->pResBlock->pDataBlock;
} }
......
source/libs/executor/inc/executorimpl.h
浏览文件 @ a2136fd3
...@@ -316,12 +316,16 @@ typedef struct STagScanInfo { ...@@ -316,12 +316,16 @@ typedef struct STagScanInfo {
typedef struct SLastrowScanInfo { typedef struct SLastrowScanInfo {
SSDataBlock *pRes; SSDataBlock *pRes;
SArray *pTableList;
SReadHandle readHandle; SReadHandle readHandle;
void *pLastrowReader; void *pLastrowReader;
SArray *pColMatchInfo; SArray *pColMatchInfo;
int32_t *pSlotIds; int32_t *pSlotIds;
SExprSupp pseudoExprSup; SExprSupp pseudoExprSup;
int32_t retrieveType;
int32_t currentGroupIndex;
SSDataBlock *pBufferredRes;
SArray *pUidList;
int32_t indexOfBufferedRes;
} SLastrowScanInfo; } SLastrowScanInfo;
typedef enum EStreamScanMode { typedef enum EStreamScanMode {
...@@ -825,8 +829,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys ...@@ -825,8 +829,7 @@ SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SProjectPhys
SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo); SOperatorInfo* createSortOperatorInfo(SOperatorInfo* downstream, SSortPhysiNode* pSortPhyNode, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo); SOperatorInfo* createMultiwayMergeOperatorInfo(SOperatorInfo** dowStreams, size_t numStreams, SMergePhysiNode* pMergePhysiNode, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo); SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pTableScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo);
SArray* pTableList, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols,
SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId, SSDataBlock* pResBlock, SInterval* pInterval, int32_t primaryTsSlotId,
...@@ -944,8 +947,9 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi ...@@ -944,8 +947,9 @@ int32_t finalizeResultRowIntoResultDataBlock(SDiskbasedBuf* pBuf, SResultRowPosi
SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset, SqlFunctionCtx* pCtx, SExprInfo* pExprInfo, int32_t numOfExprs, const int32_t* rowCellOffset,
SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo); SSDataBlock* pBlock, SExecTaskInfo* pTaskInfo);
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, int32_t createScanTableListInfo(SScanPhysiNode *pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId); STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId);
SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode, SOperatorInfo* createGroupSortOperatorInfo(SOperatorInfo* downstream, SGroupSortPhysiNode* pSortPhyNode,
SExecTaskInfo* pTaskInfo); SExecTaskInfo* pTaskInfo);
SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo *pTableListInfo, SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanNode, STableListInfo *pTableListInfo,
......
source/libs/executor/src/cachescanoperator.c
浏览文件 @ a2136fd3
...@@ -30,15 +30,13 @@ static SSDataBlock* doScanLastrow(SOperatorInfo* pOperator); ...@@ -30,15 +30,13 @@ static SSDataBlock* doScanLastrow(SOperatorInfo* pOperator);
static void destroyLastrowScanOperator(void* param, int32_t numOfOutput); static void destroyLastrowScanOperator(void* param, int32_t numOfOutput);
static int32_t extractTargetSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds); static int32_t extractTargetSlotId(const SArray* pColMatchInfo, SExecTaskInfo* pTaskInfo, int32_t** pSlotIds);
SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SReadHandle* readHandle, SArray* pTableList, SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SReadHandle* readHandle, SExecTaskInfo* pTaskInfo) {
SExecTaskInfo* pTaskInfo) {
SLastrowScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SLastrowScanInfo)); SLastrowScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SLastrowScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
goto _error; goto _error;
} }
pInfo->pTableList = pTableList;
pInfo->readHandle = *readHandle; pInfo->readHandle = *readHandle;
pInfo->pRes = createResDataBlock(pScanNode->scan.node.pOutputDataBlockDesc); pInfo->pRes = createResDataBlock(pScanNode->scan.node.pOutputDataBlockDesc);
...@@ -50,8 +48,22 @@ SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SRead ...@@ -50,8 +48,22 @@ SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SRead
goto _error; goto _error;
} }
tsdbLastRowReaderOpen(readHandle->vnode, LASTROW_RETRIEVE_TYPE_SINGLE, pTableList, taosArrayGetSize(pInfo->pColMatchInfo), STableListInfo* pTableList = &pTaskInfo->tableqinfoList;
&pInfo->pLastrowReader);
initResultSizeInfo(pOperator, 1024);
blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
pInfo->pUidList = taosArrayInit(4, sizeof(int64_t));
// partition by tbname
if (taosArrayGetSize(pTableList->pGroupList) == taosArrayGetSize(pTableList->pTableList)) {
pInfo->retrieveType = LASTROW_RETRIEVE_TYPE_ALL;
tsdbLastRowReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pTableList->pTableList,
taosArrayGetSize(pInfo->pColMatchInfo), &pInfo->pLastrowReader);
pInfo->pBufferredRes = createOneDataBlock(pInfo->pRes, false);
blockDataEnsureCapacity(pInfo->pBufferredRes, pOperator->resultInfo.capacity);
} else { // by tags
pInfo->retrieveType = LASTROW_RETRIEVE_TYPE_SINGLE;
}
if (pScanNode->scan.pScanPseudoCols != NULL) { if (pScanNode->scan.pScanPseudoCols != NULL) {
SExprSupp* pPseudoExpr = &pInfo->pseudoExprSup; SExprSupp* pPseudoExpr = &pInfo->pseudoExprSup;
...@@ -60,19 +72,17 @@ SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SRead ...@@ -60,19 +72,17 @@ SOperatorInfo* createLastrowScanOperator(SLastRowScanPhysiNode* pScanNode, SRead
pPseudoExpr->pCtx = createSqlFunctionCtx(pPseudoExpr->pExprInfo, pPseudoExpr->numOfExprs, &pPseudoExpr->rowEntryInfoOffset); pPseudoExpr->pCtx = createSqlFunctionCtx(pPseudoExpr->pExprInfo, pPseudoExpr->numOfExprs, &pPseudoExpr->rowEntryInfoOffset);
} }
pOperator->name = "LastrowScanOperator"; pOperator->name = "LastrowScanOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN; pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN;
pOperator->blocking = false; pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock); pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
initResultSizeInfo(pOperator, 1024);
blockDataEnsureCapacity(pInfo->pRes, pOperator->resultInfo.capacity);
pOperator->fpSet = pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doScanLastrow, NULL, NULL, destroyLastrowScanOperator, NULL, NULL, NULL); createOperatorFpSet(operatorDummyOpenFn, doScanLastrow, NULL, NULL, destroyLastrowScanOperator, NULL, NULL, NULL);
pOperator->cost.openCost = 0; pOperator->cost.openCost = 0;
return pOperator; return pOperator;
...@@ -90,43 +100,105 @@ SSDataBlock* doScanLastrow(SOperatorInfo* pOperator) { ...@@ -90,43 +100,105 @@ SSDataBlock* doScanLastrow(SOperatorInfo* pOperator) {
SLastrowScanInfo* pInfo = pOperator->info; SLastrowScanInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
STableListInfo* pTableList = &pTaskInfo->tableqinfoList;
int32_t size = taosArrayGetSize(pInfo->pTableList); int32_t size = taosArrayGetSize(pTableList->pTableList);
if (size == 0) { if (size == 0) {
setTaskStatus(pTaskInfo, TASK_COMPLETED); doSetOperatorCompleted(pOperator);
return NULL; return NULL;
} }
blockDataCleanup(pInfo->pRes);
// check if it is a group by tbname // check if it is a group by tbname
if (size == taosArrayGetSize(pInfo->pTableList)) { if (pInfo->retrieveType == LASTROW_RETRIEVE_TYPE_ALL) {
blockDataCleanup(pInfo->pRes); if (pInfo->indexOfBufferedRes >= pInfo->pBufferredRes->info.rows) {
SArray* pUidList = taosArrayInit(1, sizeof(tb_uid_t)); blockDataCleanup(pInfo->pBufferredRes);
int32_t code = tsdbRetrieveLastRow(pInfo->pLastrowReader, pInfo->pRes, pInfo->pSlotIds, pUidList); taosArrayClear(pInfo->pUidList);
if (code != TSDB_CODE_SUCCESS) {
longjmp(pTaskInfo->env, code); int32_t code = tsdbRetrieveLastRow(pInfo->pLastrowReader, pInfo->pBufferredRes, pInfo->pSlotIds, pInfo->pUidList);
if (code != TSDB_CODE_SUCCESS) {
longjmp(pTaskInfo->env, code);
}
// check for tag values
int32_t resultRows = pInfo->pBufferredRes->info.rows;
ASSERT(resultRows == taosArrayGetSize(pInfo->pUidList));
pInfo->indexOfBufferedRes = 0;
}
if (pInfo->indexOfBufferedRes < pInfo->pBufferredRes->info.rows) {
for(int32_t i = 0; i < taosArrayGetSize(pInfo->pColMatchInfo); ++i) {
SColMatchInfo* pMatchInfo = taosArrayGet(pInfo->pColMatchInfo, i);
int32_t slotId = pMatchInfo->targetSlotId;
SColumnInfoData* pSrc = taosArrayGet(pInfo->pBufferredRes->pDataBlock, slotId);
SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, slotId);
char* p = colDataGetData(pSrc, pInfo->indexOfBufferedRes);
bool isNull = colDataIsNull_s(pSrc, pInfo->indexOfBufferedRes);
colDataAppend(pDst, 0, p, isNull);
}
if (pInfo->pseudoExprSup.numOfExprs > 0) {
SExprSupp* pSup = &pInfo->pseudoExprSup;
addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes,
GET_TASKID(pTaskInfo));
}
pInfo->pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, pInfo->indexOfBufferedRes);
int64_t* groupId = taosHashGet(pTableList->map, &pInfo->pRes->info.uid, sizeof(int64_t));
pInfo->pRes->info.groupId = *groupId;
pInfo->indexOfBufferedRes += 1;
pInfo->pRes->info.rows = 1;
return pInfo->pRes;
} else {
doSetOperatorCompleted(pOperator);
return NULL;
} }
} else {
size_t totalGroups = taosArrayGetSize(pTableList->pGroupList);
while (pInfo->currentGroupIndex < totalGroups) {
SArray* pGroupTableList = taosArrayGetP(pTableList->pGroupList, pInfo->currentGroupIndex);
tsdbLastRowReaderOpen(pInfo->readHandle.vnode, pInfo->retrieveType, pGroupTableList,
taosArrayGetSize(pInfo->pColMatchInfo), &pInfo->pLastrowReader);
taosArrayClear(pInfo->pUidList);
int32_t code = tsdbRetrieveLastRow(pInfo->pLastrowReader, pInfo->pRes, pInfo->pSlotIds, pInfo->pUidList);
if (code != TSDB_CODE_SUCCESS) {
longjmp(pTaskInfo->env, code);
}
// check for tag values pInfo->currentGroupIndex += 1;
if (pInfo->pRes->info.rows > 0 && pInfo->pseudoExprSup.numOfExprs > 0) {
SExprSupp* pSup = &pInfo->pseudoExprSup; // check for tag values
pInfo->pRes->info.uid = *(tb_uid_t*) taosArrayGet(pUidList, 0); if (pInfo->pRes->info.rows > 0) {
addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes, GET_TASKID(pTaskInfo)); if (pInfo->pseudoExprSup.numOfExprs > 0) {
SExprSupp* pSup = &pInfo->pseudoExprSup;
pInfo->pRes->info.uid = *(tb_uid_t*)taosArrayGet(pInfo->pUidList, 0);
STableKeyInfo* pKeyInfo = taosArrayGet(pGroupTableList, 0);
pInfo->pRes->info.groupId = pKeyInfo->groupId;
addTagPseudoColumnData(&pInfo->readHandle, pSup->pExprInfo, pSup->numOfExprs, pInfo->pRes,
GET_TASKID(pTaskInfo));
}
tsdbLastrowReaderClose(pInfo->pLastrowReader);
return pInfo->pRes;
}
} }
doSetOperatorCompleted(pOperator); doSetOperatorCompleted(pOperator);
return (pInfo->pRes->info.rows == 0) ? NULL : pInfo->pRes; return NULL;
} else {
// todo fetch the result for each group
} }
return pInfo->pRes->info.rows == 0 ? NULL : pInfo->pRes;
} }
void destroyLastrowScanOperator(void* param, int32_t numOfOutput) { void destroyLastrowScanOperator(void* param, int32_t numOfOutput) {
SLastrowScanInfo* pInfo = (SLastrowScanInfo*)param; SLastrowScanInfo* pInfo = (SLastrowScanInfo*)param;
blockDataDestroy(pInfo->pRes); blockDataDestroy(pInfo->pRes);
tsdbLastrowReaderClose(pInfo->pLastrowReader);
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }
......
source/libs/executor/src/executorimpl.c
浏览文件 @ a2136fd3
...@@ -514,8 +514,10 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt ...@@ -514,8 +514,10 @@ static int32_t doSetInputDataBlock(SOperatorInfo* pOperator, SqlFunctionCtx* pCt
pInput->startRowIndex = 0; pInput->startRowIndex = 0;
// NOTE: the last parameter is the primary timestamp column // NOTE: the last parameter is the primary timestamp column
// todo: refactor this
if (fmIsTimelineFunc(pCtx[i].functionId) && (j == pOneExpr->base.numOfParams - 1)) { if (fmIsTimelineFunc(pCtx[i].functionId) && (j == pOneExpr->base.numOfParams - 1)) {
pInput->pPTS = pInput->pData[j]; pInput->pPTS = pInput->pData[j]; // in case of merge function, this is not always the ts column data.
// ASSERT(pInput->pPTS->info.type == TSDB_DATA_TYPE_TIMESTAMP);
} }
ASSERT(pInput->pData[j] != NULL); ASSERT(pInput->pData[j] != NULL);
} else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) { } else if (pFuncParam->type == FUNC_PARAM_TYPE_VALUE) {
...@@ -4291,6 +4293,7 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, ...@@ -4291,6 +4293,7 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
} }
} }
} }
int32_t len = (int32_t)(pStart - (char*)keyBuf); int32_t len = (int32_t)(pStart - (char*)keyBuf);
uint64_t groupId = calcGroupId(keyBuf, len); uint64_t groupId = calcGroupId(keyBuf, len);
taosHashPut(pTableListInfo->map, &(info->uid), sizeof(uint64_t), &groupId, sizeof(uint64_t)); taosHashPut(pTableListInfo->map, &(info->uid), sizeof(uint64_t), &groupId, sizeof(uint64_t));
...@@ -4309,6 +4312,30 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle, ...@@ -4309,6 +4312,30 @@ int32_t generateGroupIdMap(STableListInfo* pTableListInfo, SReadHandle* pHandle,
return TDB_CODE_SUCCESS; return TDB_CODE_SUCCESS;
} }
static int32_t initTableblockDistQueryCond(uint64_t uid, SQueryTableDataCond* pCond) {
memset(pCond, 0, sizeof(SQueryTableDataCond));
pCond->order = TSDB_ORDER_ASC;
pCond->numOfCols = 1;
pCond->colList = taosMemoryCalloc(1, sizeof(SColumnInfo));
if (pCond->colList == NULL) {
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return terrno;
}
pCond->colList->colId = 1;
pCond->colList->type = TSDB_DATA_TYPE_TIMESTAMP;
pCond->colList->bytes = sizeof(TSKEY);
pCond->twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
pCond->suid = uid;
pCond->type = BLOCK_LOAD_OFFSET_ORDER;
pCond->startVersion = -1;
pCond->endVersion = -1;
return TSDB_CODE_SUCCESS;
}
SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle,
uint64_t queryId, uint64_t taskId, STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId, STableListInfo* pTableListInfo,
const char* pUser) { const char* pUser) {
...@@ -4318,7 +4345,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo ...@@ -4318,7 +4345,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) { if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == type) {
STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode; STableScanPhysiNode* pTableScanNode = (STableScanPhysiNode*)pPhyNode;
int32_t code = createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId); int32_t code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags,
pTableScanNode->groupSort, pHandle, pTableListInfo, queryId, taskId);
if (code) { if (code) {
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
...@@ -4337,7 +4365,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo ...@@ -4337,7 +4365,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
} else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_TABLE_MERGE_SCAN == type) {
STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode; STableMergeScanPhysiNode* pTableScanNode = (STableMergeScanPhysiNode*)pPhyNode;
int32_t code = createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId); int32_t code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags,
pTableScanNode->groupSort, pHandle, pTableListInfo, queryId, taskId);
if (code) { if (code) {
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
...@@ -4366,7 +4395,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo ...@@ -4366,7 +4395,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
.maxTs = INT64_MIN, .maxTs = INT64_MIN,
}; };
if (pHandle) { if (pHandle) {
int32_t code = createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId); int32_t code = createScanTableListInfo(&pTableScanNode->scan, pTableScanNode->pGroupTags,
pTableScanNode->groupSort, pHandle, pTableListInfo, queryId, taskId);
if (code) { if (code) {
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
...@@ -4406,25 +4436,9 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo ...@@ -4406,25 +4436,9 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
} }
SQueryTableDataCond cond = {0}; SQueryTableDataCond cond = {0};
int32_t code = initTableblockDistQueryCond(pBlockNode->suid, &cond);
{ if (code != TSDB_CODE_SUCCESS) {
cond.order = TSDB_ORDER_ASC; return NULL;
cond.numOfCols = 1;
cond.colList = taosMemoryCalloc(1, sizeof(SColumnInfo));
if (cond.colList == NULL) {
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
return NULL;
}
cond.colList->colId = 1;
cond.colList->type = TSDB_DATA_TYPE_TIMESTAMP;
cond.colList->bytes = sizeof(TSKEY);
cond.twindows = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
cond.suid = pBlockNode->suid;
cond.type = BLOCK_LOAD_OFFSET_ORDER;
cond.startVersion = -1;
cond.endVersion = -1;
} }
STsdbReader* pReader = NULL; STsdbReader* pReader = NULL;
...@@ -4435,31 +4449,20 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo ...@@ -4435,31 +4449,20 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
} else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_LAST_ROW_SCAN == type) {
SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode; SLastRowScanPhysiNode* pScanNode = (SLastRowScanPhysiNode*)pPhyNode;
// int32_t code = createScanTableListInfo(pTableScanNode, pHandle, pTableListInfo, queryId, taskId); int32_t code = createScanTableListInfo(&pScanNode->scan, pScanNode->pGroupTags, true, pHandle, pTableListInfo,
// if (code) { queryId, taskId);
// pTaskInfo->code = code;
// return NULL;
// }
int32_t code = extractTableSchemaInfo(pHandle, pScanNode->scan.uid, pTaskInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
} }
pTableListInfo->pTableList = taosArrayInit(4, sizeof(STableKeyInfo)); code = extractTableSchemaInfo(pHandle, pScanNode->scan.uid, pTaskInfo);
if (pScanNode->scan.tableType == TSDB_SUPER_TABLE) { if (code != TSDB_CODE_SUCCESS) {
code = vnodeGetAllTableList(pHandle->vnode, pScanNode->scan.uid, pTableListInfo->pTableList); pTaskInfo->code = code;
if (code != TSDB_CODE_SUCCESS) { return NULL;
pTaskInfo->code = terrno;
return NULL;
}
} else { // Create one table group.
STableKeyInfo info = {.lastKey = 0, .uid = pScanNode->scan.uid, .groupId = 0};
taosArrayPush(pTableListInfo->pTableList, &info);
} }
return createLastrowScanOperator(pScanNode, pHandle, pTableListInfo->pTableList, pTaskInfo); return createLastrowScanOperator(pScanNode, pHandle, pTaskInfo);
} else { } else {
ASSERT(0); ASSERT(0);
} }
...@@ -4928,6 +4931,9 @@ static void doDestroyTableList(STableListInfo* pTableqinfoList) { ...@@ -4928,6 +4931,9 @@ static void doDestroyTableList(STableListInfo* pTableqinfoList) {
if (pTableqinfoList->needSortTableByGroupId) { if (pTableqinfoList->needSortTableByGroupId) {
for (int32_t i = 0; i < taosArrayGetSize(pTableqinfoList->pGroupList); i++) { for (int32_t i = 0; i < taosArrayGetSize(pTableqinfoList->pGroupList); i++) {
SArray* tmp = taosArrayGetP(pTableqinfoList->pGroupList, i); SArray* tmp = taosArrayGetP(pTableqinfoList->pGroupList, i);
if (tmp == pTableqinfoList->pTableList) {
continue;
}
taosArrayDestroy(tmp); taosArrayDestroy(tmp);
} }
} }
......
source/libs/executor/src/scanoperator.c
浏览文件 @ a2136fd3
...@@ -2402,9 +2402,9 @@ typedef struct STableMergeScanInfo { ...@@ -2402,9 +2402,9 @@ typedef struct STableMergeScanInfo {
SSampleExecInfo sample; // sample execution info SSampleExecInfo sample; // sample execution info
} STableMergeScanInfo; } STableMergeScanInfo;
int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle* pHandle, int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags, bool groupSort, SReadHandle* pHandle,
STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId) { STableListInfo* pTableListInfo, uint64_t queryId, uint64_t taskId) {
int32_t code = getTableList(pHandle->meta, pHandle->vnode, &pTableScanNode->scan, pTableListInfo); int32_t code = getTableList(pHandle->meta, pHandle->vnode, pScanNode, pTableListInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }
...@@ -2414,8 +2414,8 @@ int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle ...@@ -2414,8 +2414,8 @@ int32_t createScanTableListInfo(STableScanPhysiNode* pTableScanNode, SReadHandle
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
pTableListInfo->needSortTableByGroupId = pTableScanNode->groupSort; pTableListInfo->needSortTableByGroupId = groupSort;
code = generateGroupIdMap(pTableListInfo, pHandle, pTableScanNode->pGroupTags); code = generateGroupIdMap(pTableListInfo, pHandle, pGroupTags);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
return code; return code;
} }
......
source/libs/function/src/builtins.c
浏览文件 @ a2136fd3
...@@ -1981,6 +1981,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { ...@@ -1981,6 +1981,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.getEnvFunc = getLeastSQRFuncEnv, .getEnvFunc = getLeastSQRFuncEnv,
.initFunc = leastSQRFunctionSetup, .initFunc = leastSQRFunctionSetup,
.processFunc = leastSQRFunction, .processFunc = leastSQRFunction,
.sprocessFunc = leastSQRScalarFunction,
.finalizeFunc = leastSQRFinalize, .finalizeFunc = leastSQRFinalize,
.invertFunc = NULL, .invertFunc = NULL,
.combineFunc = leastSQRCombine, .combineFunc = leastSQRCombine,
...@@ -2228,7 +2229,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = { ...@@ -2228,7 +2229,7 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
{ {
.name = "_cache_last_row", .name = "_cache_last_row",
.type = FUNCTION_TYPE_CACHE_LAST_ROW, .type = FUNCTION_TYPE_CACHE_LAST_ROW,
.classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC, .classification = FUNC_MGT_AGG_FUNC | FUNC_MGT_MULTI_RES_FUNC | FUNC_MGT_SELECT_FUNC | FUNC_MGT_TIMELINE_FUNC | FUNC_MGT_IMPLICIT_TS_FUNC,
.translateFunc = translateFirstLast, .translateFunc = translateFirstLast,
.getEnvFunc = getFirstLastFuncEnv, .getEnvFunc = getFirstLastFuncEnv,
.initFunc = functionSetup, .initFunc = functionSetup,
......
source/libs/scalar/src/sclfunc.c
浏览文件 @ a2136fd3
...@@ -2139,3 +2139,171 @@ int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara ...@@ -2139,3 +2139,171 @@ int32_t stddevScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarPara
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
#define LEASTSQR_CAL(p, x, y, index, step) \
do { \
(p)[0][0] += (double)(x) * (x); \
(p)[0][1] += (double)(x); \
(p)[0][2] += (double)(x) * (y)[index]; \
(p)[1][2] += (y)[index]; \
(x) += step; \
} while (0)
int32_t leastSQRScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutput) {
SColumnInfoData *pInputData = pInput->columnData;
SColumnInfoData *pOutputData = pOutput->columnData;
double startVal, stepVal;
double matrix[2][3] = {0};
GET_TYPED_DATA(startVal, double, GET_PARAM_TYPE(&pInput[1]), pInput[1].columnData->pData);
GET_TYPED_DATA(stepVal, double, GET_PARAM_TYPE(&pInput[2]), pInput[2].columnData->pData);
int32_t type = GET_PARAM_TYPE(pInput);
int64_t count = 0;
switch(type) {
case TSDB_DATA_TYPE_TINYINT: {
int8_t *in = (int8_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_SMALLINT: {
int16_t *in = (int16_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_INT: {
int32_t *in = (int32_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_BIGINT: {
int64_t *in = (int64_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_UTINYINT: {
uint8_t *in = (uint8_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_USMALLINT: {
uint16_t *in = (uint16_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_UINT: {
uint32_t *in = (uint32_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_UBIGINT: {
uint64_t *in = (uint64_t *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_FLOAT: {
float *in = (float *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
double *in = (double *)pInputData->pData;
for (int32_t i = 0; i < pInput->numOfRows; ++i) {
if (colDataIsNull_s(pInputData, i)) {
continue;
}
count++;
LEASTSQR_CAL(matrix, startVal, in, i, stepVal);
}
break;
}
}
if (count == 0) {
colDataAppendNULL(pOutputData, 0);
} else {
matrix[1][1] = (double)count;
matrix[1][0] = matrix[0][1];
double matrix00 = matrix[0][0] - matrix[1][0] * (matrix[0][1] / matrix[1][1]);
double matrix02 = matrix[0][2] - matrix[1][2] * (matrix[0][1] / matrix[1][1]);
double matrix12 = matrix[1][2] - matrix02 * (matrix[1][0] / matrix00);
matrix02 /= matrix00;
matrix12 /= matrix[1][1];
char buf[64] = {0};
size_t len =
snprintf(varDataVal(buf), sizeof(buf) - VARSTR_HEADER_SIZE, "{slop:%.6lf, intercept:%.6lf}", matrix02, matrix12);
varDataSetLen(buf, len);
colDataAppend(pOutputData, 0, buf, false);
}
pOutput->numOfRows = 1;
return TSDB_CODE_SUCCESS;
}
tests/system-test/2-query/csum.py
浏览文件 @ a2136fd3
...@@ -425,8 +425,8 @@ class TDTestCase: ...@@ -425,8 +425,8 @@ class TDTestCase:
tdSql.checkRows(70) tdSql.checkRows(70)
tdSql.query("select csum(c1) from stb1 partition by tbname ") tdSql.query("select csum(c1) from stb1 partition by tbname ")
tdSql.checkRows(40) tdSql.checkRows(40)
# tdSql.query("select csum(st1) from stb1 partition by tbname") tdSql.query("select csum(st1) from stb1 partition by tbname")
# tdSql.checkRows(70) tdSql.checkRows(70)
tdSql.query("select csum(st1+c1) from stb1 partition by tbname") tdSql.query("select csum(st1+c1) from stb1 partition by tbname")
tdSql.checkRows(40) tdSql.checkRows(40)
tdSql.query("select csum(st1+c1) from stb1 partition by tbname") tdSql.query("select csum(st1+c1) from stb1 partition by tbname")
...@@ -445,22 +445,22 @@ class TDTestCase: ...@@ -445,22 +445,22 @@ class TDTestCase:
tdSql.checkRows(40) tdSql.checkRows(40)
# bug need fix # bug need fix
# tdSql.query("select tbname , csum(c1) from stb1 partition by tbname") tdSql.query("select tbname , csum(c1) from stb1 partition by tbname")
# tdSql.checkRows(40) tdSql.checkRows(40)
# tdSql.query("select tbname , csum(st1) from stb1 partition by tbname") tdSql.query("select tbname , csum(st1) from stb1 partition by tbname")
# tdSql.checkRows(70) tdSql.checkRows(70)
# tdSql.query("select tbname , csum(st1) from stb1 partition by tbname slimit 1") tdSql.query("select tbname , csum(st1) from stb1 partition by tbname slimit 1")
# tdSql.checkRows(7) tdSql.checkRows(7)
# partition by tags # partition by tags
# tdSql.query("select st1 , csum(c1) from stb1 partition by st1") tdSql.query("select st1 , csum(c1) from stb1 partition by st1")
# tdSql.checkRows(40) tdSql.checkRows(40)
# tdSql.query("select csum(c1) from stb1 partition by st1") tdSql.query("select csum(c1) from stb1 partition by st1")
# tdSql.checkRows(40) tdSql.checkRows(40)
# tdSql.query("select st1 , csum(c1) from stb1 partition by st1 slimit 1") tdSql.query("select st1 , csum(c1) from stb1 partition by st1 slimit 1")
# tdSql.checkRows(4) tdSql.checkRows(4)
# tdSql.query("select csum(c1) from stb1 partition by st1 slimit 1") tdSql.query("select csum(c1) from stb1 partition by st1 slimit 1")
# tdSql.checkRows(4) tdSql.checkRows(4)
# partition by col # partition by col
# tdSql.query("select c1 , csum(c1) from stb1 partition by c1") # tdSql.query("select c1 , csum(c1) from stb1 partition by c1")
......
tests/system-test/2-query/last_row.py 0 → 100644
浏览文件 @ a2136fd3
此差异已折叠。
tests/system-test/2-query/max_partition.py
浏览文件 @ a2136fd3
...@@ -45,8 +45,8 @@ class TDTestCase: ...@@ -45,8 +45,8 @@ class TDTestCase:
tdSql.query(" select max(c1) from stb group by tbname order by tbname ") tdSql.query(" select max(c1) from stb group by tbname order by tbname ")
tdSql.checkRows(self.tb_nums) tdSql.checkRows(self.tb_nums)
# bug need fix # bug need fix
# tdSql.query(" select max(t1) from stb group by t2 order by t2 ") tdSql.query(" select max(t2) from stb group by t2 order by t2 ")
# tdSql.checkRows(self.tb_nums) tdSql.checkRows(self.tb_nums)
tdSql.query(" select max(c1) from stb group by c1 order by c1 ") tdSql.query(" select max(c1) from stb group by c1 order by c1 ")
tdSql.checkRows(self.row_nums+1) tdSql.checkRows(self.row_nums+1)
...@@ -90,8 +90,8 @@ class TDTestCase: ...@@ -90,8 +90,8 @@ class TDTestCase:
tdSql.query("select tbname , max(t2) from stb partition by t2 order by t2") tdSql.query("select tbname , max(t2) from stb partition by t2 order by t2")
# # bug need fix # # bug need fix
# tdSql.query("select t2 , max(t2) from stb partition by t2 order by t2") tdSql.query("select t2 , max(t2) from stb partition by t2 order by t2")
# tdSql.checkRows(self.tb_nums) tdSql.checkRows(self.tb_nums)
tdSql.query("select tbname , max(c1) from stb partition by tbname order by tbname") tdSql.query("select tbname , max(c1) from stb partition by tbname order by tbname")
tdSql.checkRows(self.tb_nums) tdSql.checkRows(self.tb_nums)
...@@ -126,8 +126,8 @@ class TDTestCase: ...@@ -126,8 +126,8 @@ class TDTestCase:
tdSql.checkData(0,0,self.row_nums) tdSql.checkData(0,0,self.row_nums)
# bug need fix # bug need fix
# tdSql.query("select count(c1) , max(t1) ,abs(c1) from stb partition by abs(c1) order by abs(c1)") tdSql.query("select count(c1) , max(t2) ,abs(c1) from stb partition by abs(c1) order by abs(c1)")
# tdSql.checkRows(self.row_nums+1) tdSql.checkRows(self.row_nums+1)
tdSql.query("select max(ceil(c2)) , max(floor(t2)) ,max(floor(c2)) from stb partition by abs(c2) order by abs(c2)") tdSql.query("select max(ceil(c2)) , max(floor(t2)) ,max(floor(c2)) from stb partition by abs(c2) order by abs(c2)")
...@@ -136,6 +136,18 @@ class TDTestCase: ...@@ -136,6 +136,18 @@ class TDTestCase:
tdSql.query("select max(ceil(c1-2)) , max(floor(t2+1)) ,max(c2-c1) from stb partition by abs(floor(c1)) order by abs(floor(c1))") tdSql.query("select max(ceil(c1-2)) , max(floor(t2+1)) ,max(c2-c1) from stb partition by abs(floor(c1)) order by abs(floor(c1))")
tdSql.checkRows(self.row_nums+1) tdSql.checkRows(self.row_nums+1)
tdSql.query("select tbname , max(c1) ,c1 from stb partition by tbname order by tbname")
tdSql.checkRows(self.tb_nums)
tdSql.checkData(0,0,'sub_stb_0')
tdSql.checkData(0,1,9)
tdSql.checkData(0,2,9)
tdSql.query("select tbname ,top(c1,1) ,c1 from stb partition by tbname order by tbname")
tdSql.checkRows(self.tb_nums)
tdSql.query(" select c1 , sample(c1,2) from stb partition by tbname order by tbname ")
tdSql.checkRows(self.tb_nums*2)
# interval # interval
......
tests/system-test/2-query/sample.py
浏览文件 @ a2136fd3
...@@ -31,6 +31,7 @@ class TDTestCase: ...@@ -31,6 +31,7 @@ class TDTestCase:
def init(self, conn, logSql): def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__) tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor()) tdSql.init(conn.cursor())
self.ts = 1537146000000
def sample_query_form(self, sel="select", func="sample(", col="c1", m_comm =",", k=1,r_comm=")", alias="", fr="from",table_expr="t1", condition=""): def sample_query_form(self, sel="select", func="sample(", col="c1", m_comm =",", k=1,r_comm=")", alias="", fr="from",table_expr="t1", condition=""):
''' '''
...@@ -624,10 +625,11 @@ class TDTestCase: ...@@ -624,10 +625,11 @@ class TDTestCase:
tdLog.info(" sample data is in datas groups ,successed sql is : %s" % sample_query ) tdLog.info(" sample data is in datas groups ,successed sql is : %s" % sample_query )
else: else:
tdLog.exit(" sample data is not in datas groups ,failed sql is : %s" % sample_query ) tdLog.exit(" sample data is not in datas groups ,failed sql is : %s" % sample_query )
def basic_sample_query(self): def basic_sample_query(self):
tdSql.execute(" drop database if exists db ") tdSql.execute(" drop database if exists db ")
tdSql.execute(" create database if not exists db duration 300 ") tdSql.execute(" create database if not exists db duration 300d ")
tdSql.execute(" use db ") tdSql.execute(" use db ")
tdSql.execute( tdSql.execute(
'''create table stb1 '''create table stb1
...@@ -759,14 +761,6 @@ class TDTestCase: ...@@ -759,14 +761,6 @@ class TDTestCase:
self.check_sample("select sample( c1 ,3 ) from t1 where c1 between 1 and 10" ,"select c1 from t1 where c1 between 1 and 10") self.check_sample("select sample( c1 ,3 ) from t1 where c1 between 1 and 10" ,"select c1 from t1 where c1 between 1 and 10")
tdSql.query("select sample(c1,2) ,c2,c3 ,c5 from stb1")
tdSql.checkRows(2)
tdSql.checkCols(4)
self.check_sample("select sample( c1 ,3 ),c2,c3,c4,c5 from t1 where c1 between 1 and 10" ,"select c1,c2,c3,c4,c5 from t1 where c1 between 1 and 10")
self.check_sample("select sample( c1 ,3 ),c2,c3,c4,c5 from stb1 where c1 between 1 and 10" ,"select c1,c2,c3,c4,c5 from stb1 where c1 between 1 and 10")
self.check_sample("select sample( c1 ,3 ),t1 from stb1 where c1 between 1 and 10" ,"select c1,t1 from stb1 where c1 between 1 and 10")
# join # join
tdSql.query("select sample( ct4.c1 , 1 ) from ct1, ct4 where ct4.ts=ct1.ts") tdSql.query("select sample( ct4.c1 , 1 ) from ct1, ct4 where ct4.ts=ct1.ts")
...@@ -779,8 +773,8 @@ class TDTestCase: ...@@ -779,8 +773,8 @@ class TDTestCase:
self.check_sample("select sample(c1,2) from stb1 partition by tbname" , "select c1 from stb1 partition by tbname") self.check_sample("select sample(c1,2) from stb1 partition by tbname" , "select c1 from stb1 partition by tbname")
# nest query # nest query
tdSql.query("select sample(c1,2) from (select c1 from t1); ") # tdSql.query("select sample(c1,2) from (select c1 from t1); ")
tdSql.checkRows(2) # tdSql.checkRows(2)
# union all # union all
tdSql.query("select sample(c1,2) from t1 union all select sample(c1,3) from t1") tdSql.query("select sample(c1,2) from t1 union all select sample(c1,3) from t1")
...@@ -798,36 +792,6 @@ class TDTestCase: ...@@ -798,36 +792,6 @@ class TDTestCase:
tdSql.query("select sample(c1,100)+2 from ct1") tdSql.query("select sample(c1,100)+2 from ct1")
tdSql.query("select abs(sample(c1,100)) from ct1") tdSql.query("select abs(sample(c1,100)) from ct1")
# support stable and tbname
tdSql.query("select tbname ,sample(c1,2) from stb1 partition by tbname order by tbname")
tdSql.checkRows(4)
tdSql.checkData(0,0,'ct1')
tdSql.checkData(3,0,'ct4')
# # bug need fix
# tdSql.query(" select tbname ,c1 ,t1, sample(c1,2) from stb1 partition by tbname order by tbname ")
# tdSql.checkRows(4)
# tdSql.checkData(0,0,'ct1')
# tdSql.checkData(3,0,'ct4')
# tdSql.checkData(0,2,1)
# tdSql.checkData(3,2,4)
tdSql.query(" select tbname ,c1 ,t1, sample(c1,2) from stb1 partition by t1 order by t1 ")
tdSql.checkRows(4)
tdSql.checkData(0,0,'ct1')
tdSql.checkData(3,0,'ct4')
tdSql.checkData(0,2,1)
tdSql.checkData(3,2,4)
# bug need fix
# tdSql.query(" select tbname ,c1 ,t1, sample(c1,2) from stb1 partition by c1 order by c1 ")
# tdSql.checkRows(21)
# bug need fix
# tdSql.query(" select sample(c1,2) from stb1 partition by c1 ")
# tdSql.checkRows(21)
def sample_test_run(self) : def sample_test_run(self) :
tdLog.printNoPrefix("==========support sample function==========") tdLog.printNoPrefix("==========support sample function==========")
tbnum = 10 tbnum = 10
...@@ -891,11 +855,29 @@ class TDTestCase: ...@@ -891,11 +855,29 @@ class TDTestCase:
self.basic_sample_query() self.basic_sample_query()
def sample_big_data(self):
tdSql.execute("create database sample_db")
tdSql.execute("use sample_db")
tdSql.execute("create stable st (ts timestamp ,c1 int ) tags(ind int)" )
tdSql.execute("create table sub_tb using st tags(1)")
for i in range(2000):
ts = self.ts+i*10
tdSql.execute(f"insert into sub_tb values({ts} ,{i})")
tdSql.query("select count(*) from st")
tdSql.checkData(0,0,2000)
tdSql.query("select sample(c1 ,1000) from st")
tdSql.checkRows(1000)
def run(self): def run(self):
import traceback import traceback
try: try:
# run in develop branch # run in develop branch
self.sample_test_run() self.sample_test_run()
self.sample_big_data()
pass pass
except Exception as e: except Exception as e:
traceback.print_exc() traceback.print_exc()
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或