Merge pull request #11082 from taosdata/feature/3.0_liaohj
Feature/3.0 liaohj
This commit is contained in:
commit
9b6f008cf8
|
@ -574,11 +574,11 @@ typedef struct SMultiwayMergeInfo {
|
||||||
} SMultiwayMergeInfo;
|
} SMultiwayMergeInfo;
|
||||||
|
|
||||||
// todo support the disk-based sort
|
// todo support the disk-based sort
|
||||||
typedef struct SOrderOperatorInfo {
|
typedef struct SSortOperatorInfo {
|
||||||
int32_t colIndex;
|
int32_t colIndex;
|
||||||
int32_t order;
|
int32_t order;
|
||||||
SSDataBlock *pDataBlock;
|
SSDataBlock *pDataBlock;
|
||||||
} SOrderOperatorInfo;
|
} SSortOperatorInfo;
|
||||||
|
|
||||||
void appendUpstream(SOperatorInfo* p, SOperatorInfo* pUpstream);
|
void appendUpstream(SOperatorInfo* p, SOperatorInfo* pUpstream);
|
||||||
|
|
||||||
|
@ -609,7 +609,7 @@ SOperatorInfo* createFilterOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperator
|
||||||
int32_t numOfOutput, SColumnInfo* pCols, int32_t numOfFilter);
|
int32_t numOfOutput, SColumnInfo* pCols, int32_t numOfFilter);
|
||||||
|
|
||||||
SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pUpstream, int32_t numOfUpstream, SSchema* pSchema, int32_t numOfOutput);
|
SOperatorInfo* createJoinOperatorInfo(SOperatorInfo** pUpstream, int32_t numOfUpstream, SSchema* pSchema, int32_t numOfOutput);
|
||||||
SOperatorInfo* createOrderOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SOrderVal* pOrderVal);
|
SOperatorInfo* createSortOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SOrderVal* pOrderVal);
|
||||||
|
|
||||||
SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
|
SSDataBlock* doGlobalAggregate(void* param, bool* newgroup);
|
||||||
SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
|
SSDataBlock* doMultiwayMergeSort(void* param, bool* newgroup);
|
||||||
|
|
|
@ -2301,7 +2301,7 @@ static int32_t setupQueryRuntimeEnv(SQueryRuntimeEnv *pRuntimeEnv, int32_t numOf
|
||||||
}
|
}
|
||||||
|
|
||||||
case OP_Order: {
|
case OP_Order: {
|
||||||
pRuntimeEnv->proot = createOrderOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput, &pQueryAttr->order);
|
pRuntimeEnv->proot = createSortOperatorInfo(pRuntimeEnv, pRuntimeEnv->proot, pQueryAttr->pExpr1, pQueryAttr->numOfOutput, &pQueryAttr->order);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -5516,7 +5516,7 @@ static SSDataBlock* doSort(void* param, bool* newgroup) {
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
SOrderOperatorInfo* pInfo = pOperator->info;
|
SSortOperatorInfo* pInfo = pOperator->info;
|
||||||
|
|
||||||
SSDataBlock* pBlock = NULL;
|
SSDataBlock* pBlock = NULL;
|
||||||
while(1) {
|
while(1) {
|
||||||
|
@ -5556,8 +5556,8 @@ static SSDataBlock* doSort(void* param, bool* newgroup) {
|
||||||
return (pInfo->pDataBlock->info.rows > 0)? pInfo->pDataBlock:NULL;
|
return (pInfo->pDataBlock->info.rows > 0)? pInfo->pDataBlock:NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
SOperatorInfo *createOrderOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SOrderVal* pOrderVal) {
|
SOperatorInfo *createSortOperatorInfo(SQueryRuntimeEnv* pRuntimeEnv, SOperatorInfo* upstream, SExprInfo* pExpr, int32_t numOfOutput, SOrderVal* pOrderVal) {
|
||||||
SOrderOperatorInfo* pInfo = calloc(1, sizeof(SOrderOperatorInfo));
|
SSortOperatorInfo* pInfo = calloc(1, sizeof(SSortOperatorInfo));
|
||||||
|
|
||||||
{
|
{
|
||||||
SSDataBlock* pDataBlock = calloc(1, sizeof(SSDataBlock));
|
SSDataBlock* pDataBlock = calloc(1, sizeof(SSDataBlock));
|
||||||
|
@ -6611,7 +6611,7 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
|
||||||
}
|
}
|
||||||
|
|
||||||
static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) {
|
static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) {
|
||||||
SOrderOperatorInfo* pInfo = (SOrderOperatorInfo*) param;
|
SSortOperatorInfo* pInfo = (SSortOperatorInfo*) param;
|
||||||
pInfo->pDataBlock = blockDataDestroy(pInfo->pDataBlock);
|
pInfo->pDataBlock = blockDataDestroy(pInfo->pDataBlock);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -29,8 +29,9 @@ typedef struct SCorEpSet {
|
||||||
} SCorEpSet;
|
} SCorEpSet;
|
||||||
|
|
||||||
typedef struct SBlockOrderInfo {
|
typedef struct SBlockOrderInfo {
|
||||||
|
bool nullFirst;
|
||||||
int32_t order;
|
int32_t order;
|
||||||
int32_t colIndex;
|
int32_t slotId;
|
||||||
SColumnInfoData* pColData;
|
SColumnInfoData* pColData;
|
||||||
} SBlockOrderInfo;
|
} SBlockOrderInfo;
|
||||||
|
|
||||||
|
@ -176,7 +177,7 @@ size_t blockDataGetSerialMetaSize(const SSDataBlock* pBlock);
|
||||||
|
|
||||||
SSchema* blockDataExtractSchema(const SSDataBlock* pBlock, int32_t* numOfCols);
|
SSchema* blockDataExtractSchema(const SSDataBlock* pBlock, int32_t* numOfCols);
|
||||||
|
|
||||||
int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
|
int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo);
|
||||||
int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
|
int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
|
||||||
|
|
||||||
int32_t blockDataEnsureColumnCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
|
int32_t blockDataEnsureColumnCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
|
||||||
|
|
|
@ -54,6 +54,16 @@ qTaskInfo_t qCreateStreamExecTaskInfo(void* msg, void* streamReadHandle);
|
||||||
*/
|
*/
|
||||||
int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type);
|
int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Set multiple input data blocks for the stream scan.
|
||||||
|
* @param tinfo
|
||||||
|
* @param pBlocks
|
||||||
|
* @param numOfInputBlock
|
||||||
|
* @param type
|
||||||
|
* @return
|
||||||
|
*/
|
||||||
|
int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, void** pBlocks, size_t numOfBlocks, int32_t type);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Update the table id list, add or remove.
|
* Update the table id list, add or remove.
|
||||||
*
|
*
|
||||||
|
@ -86,16 +96,6 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
|
||||||
*/
|
*/
|
||||||
int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds);
|
int32_t qExecTask(qTaskInfo_t tinfo, SSDataBlock** pRes, uint64_t* useconds);
|
||||||
|
|
||||||
/**
|
|
||||||
* Retrieve the produced results information, if current query is not paused or completed,
|
|
||||||
* this function will be blocked to wait for the query execution completed or paused,
|
|
||||||
* in which case enough results have been produced already.
|
|
||||||
*
|
|
||||||
* @param tinfo
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
int32_t qRetrieveQueryResultInfo(qTaskInfo_t tinfo, bool* buildRes, void* pRspContext);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* kill the ongoing query and free the query handle and corresponding resources automatically
|
* kill the ongoing query and free the query handle and corresponding resources automatically
|
||||||
* @param tinfo qhandle
|
* @param tinfo qhandle
|
||||||
|
@ -158,50 +158,6 @@ int32_t qGetQualifiedTableIdList(void* pTableList, const char* tagCond, int32_t
|
||||||
*/
|
*/
|
||||||
int32_t qUpdateQueriedTableIdList(qTaskInfo_t tinfo, int64_t uid, int32_t type);
|
int32_t qUpdateQueriedTableIdList(qTaskInfo_t tinfo, int64_t uid, int32_t type);
|
||||||
|
|
||||||
//================================================================================================
|
|
||||||
// query handle management
|
|
||||||
/**
|
|
||||||
* Query handle mgmt object
|
|
||||||
* @param vgId
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
void* qOpenTaskMgmt(int32_t vgId);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* broadcast the close information and wait for all query stop.
|
|
||||||
* @param pExecutor
|
|
||||||
*/
|
|
||||||
void qTaskMgmtNotifyClosing(void* pExecutor);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Re-open the query handle management module when opening the vnode again.
|
|
||||||
* @param pExecutor
|
|
||||||
*/
|
|
||||||
void qQueryMgmtReOpen(void* pExecutor);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Close query mgmt and clean up resources.
|
|
||||||
* @param pExecutor
|
|
||||||
*/
|
|
||||||
void qCleanupTaskMgmt(void* pExecutor);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Add the query into the query mgmt object
|
|
||||||
* @param pMgmt
|
|
||||||
* @param qId
|
|
||||||
* @param qInfo
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
void** qRegisterTask(void* pMgmt, uint64_t qId, void* qInfo);
|
|
||||||
|
|
||||||
/**
|
|
||||||
* acquire the query handle according to the key from query mgmt object.
|
|
||||||
* @param pMgmt
|
|
||||||
* @param key
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
void** qAcquireTask(void* pMgmt, uint64_t key);
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* release the query handle and decrease the reference count in cache
|
* release the query handle and decrease the reference count in cache
|
||||||
* @param pMgmt
|
* @param pMgmt
|
||||||
|
@ -211,13 +167,6 @@ void** qAcquireTask(void* pMgmt, uint64_t key);
|
||||||
*/
|
*/
|
||||||
void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle);
|
void** qReleaseTask(void* pMgmt, void* pQInfo, bool freeHandle);
|
||||||
|
|
||||||
/**
|
|
||||||
* De-register the query handle from the management module and free it immediately.
|
|
||||||
* @param pMgmt
|
|
||||||
* @param pQInfo
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
void** qDeregisterQInfo(void* pMgmt, void* pQInfo);
|
|
||||||
|
|
||||||
void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
|
void qProcessFetchRsp(void* parent, struct SRpcMsg* pMsg, struct SEpSet* pEpSet);
|
||||||
|
|
||||||
|
|
|
@ -647,7 +647,6 @@ double blockDataGetSerialRowSize(const SSDataBlock* pBlock) {
|
||||||
typedef struct SSDataBlockSortHelper {
|
typedef struct SSDataBlockSortHelper {
|
||||||
SArray* orderInfo; // SArray<SBlockOrderInfo>
|
SArray* orderInfo; // SArray<SBlockOrderInfo>
|
||||||
SSDataBlock* pDataBlock;
|
SSDataBlock* pDataBlock;
|
||||||
bool nullFirst;
|
|
||||||
} SSDataBlockSortHelper;
|
} SSDataBlockSortHelper;
|
||||||
|
|
||||||
int32_t dataBlockCompar(const void* p1, const void* p2, const void* param) {
|
int32_t dataBlockCompar(const void* p1, const void* p2, const void* param) {
|
||||||
|
@ -672,11 +671,11 @@ int32_t dataBlockCompar(const void* p1, const void* p2, const void* param) {
|
||||||
}
|
}
|
||||||
|
|
||||||
if (rightNull) {
|
if (rightNull) {
|
||||||
return pHelper->nullFirst ? 1 : -1;
|
return pOrder->nullFirst ? 1 : -1;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (leftNull) {
|
if (leftNull) {
|
||||||
return pHelper->nullFirst ? -1 : 1;
|
return pOrder->nullFirst ? -1 : 1;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -907,7 +906,7 @@ static __compar_fn_t getComparFn(int32_t type, int32_t order) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst) {
|
int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo) {
|
||||||
ASSERT(pDataBlock != NULL && pOrderInfo != NULL);
|
ASSERT(pDataBlock != NULL && pOrderInfo != NULL);
|
||||||
if (pDataBlock->info.rows <= 1) {
|
if (pDataBlock->info.rows <= 1) {
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
|
@ -922,7 +921,7 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirs
|
||||||
for (int32_t i = 0; i < taosArrayGetSize(pOrderInfo); ++i) {
|
for (int32_t i = 0; i < taosArrayGetSize(pOrderInfo); ++i) {
|
||||||
SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);
|
SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);
|
||||||
|
|
||||||
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, pInfo->colIndex);
|
SColumnInfoData* pColInfoData = taosArrayGet(pDataBlock->pDataBlock, pInfo->slotId);
|
||||||
if (pColInfoData->hasNull) {
|
if (pColInfoData->hasNull) {
|
||||||
sortColumnHasNull = true;
|
sortColumnHasNull = true;
|
||||||
}
|
}
|
||||||
|
@ -961,10 +960,10 @@ int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirs
|
||||||
|
|
||||||
int64_t p0 = taosGetTimestampUs();
|
int64_t p0 = taosGetTimestampUs();
|
||||||
|
|
||||||
SSDataBlockSortHelper helper = {.nullFirst = nullFirst, .pDataBlock = pDataBlock, .orderInfo = pOrderInfo};
|
SSDataBlockSortHelper helper = {.pDataBlock = pDataBlock, .orderInfo = pOrderInfo};
|
||||||
for (int32_t i = 0; i < taosArrayGetSize(helper.orderInfo); ++i) {
|
for (int32_t i = 0; i < taosArrayGetSize(helper.orderInfo); ++i) {
|
||||||
struct SBlockOrderInfo* pInfo = taosArrayGet(helper.orderInfo, i);
|
struct SBlockOrderInfo* pInfo = taosArrayGet(helper.orderInfo, i);
|
||||||
pInfo->pColData = taosArrayGet(pDataBlock->pDataBlock, pInfo->colIndex);
|
pInfo->pColData = taosArrayGet(pDataBlock->pDataBlock, pInfo->slotId);
|
||||||
}
|
}
|
||||||
|
|
||||||
taosqsort(index, rows, sizeof(int32_t), &helper, dataBlockCompar);
|
taosqsort(index, rows, sizeof(int32_t), &helper, dataBlockCompar);
|
||||||
|
@ -1012,7 +1011,7 @@ SHelper* createTupleIndex_rv(int32_t numOfRows, SArray* pOrderInfo, SSDataBlock*
|
||||||
|
|
||||||
for (int32_t i = 0; i < numOfCols; ++i) {
|
for (int32_t i = 0; i < numOfCols; ++i) {
|
||||||
SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);
|
SBlockOrderInfo* pInfo = taosArrayGet(pOrderInfo, i);
|
||||||
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, pInfo->colIndex);
|
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, pInfo->slotId);
|
||||||
pInfo->pColData = pColInfo;
|
pInfo->pColData = pColInfo;
|
||||||
sortValLengthPerRow += pColInfo->info.bytes;
|
sortValLengthPerRow += pColInfo->info.bytes;
|
||||||
}
|
}
|
||||||
|
@ -1106,7 +1105,7 @@ int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullF
|
||||||
// Allocate the additional buffer.
|
// Allocate the additional buffer.
|
||||||
int64_t p0 = taosGetTimestampUs();
|
int64_t p0 = taosGetTimestampUs();
|
||||||
|
|
||||||
SSDataBlockSortHelper helper = {.nullFirst = nullFirst, .pDataBlock = pDataBlock, .orderInfo = pOrderInfo};
|
SSDataBlockSortHelper helper = {.pDataBlock = pDataBlock, .orderInfo = pOrderInfo};
|
||||||
|
|
||||||
uint32_t rows = pDataBlock->info.rows;
|
uint32_t rows = pDataBlock->info.rows;
|
||||||
SHelper* index = createTupleIndex_rv(rows, helper.orderInfo, pDataBlock);
|
SHelper* index = createTupleIndex_rv(rows, helper.orderInfo, pDataBlock);
|
||||||
|
|
|
@ -167,10 +167,10 @@ TEST(testCase, Datablock_test) {
|
||||||
printf("the second row of binary:%s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
|
printf("the second row of binary:%s, length:%d\n", (char*)varDataVal(pData), varDataLen(pData));
|
||||||
|
|
||||||
SArray* pOrderInfo = taosArrayInit(3, sizeof(SBlockOrderInfo));
|
SArray* pOrderInfo = taosArrayInit(3, sizeof(SBlockOrderInfo));
|
||||||
SBlockOrderInfo order = {.order = TSDB_ORDER_ASC, .colIndex = 0};
|
SBlockOrderInfo order = {.nullFirst = true, .order = TSDB_ORDER_ASC, .slotId = 0};
|
||||||
taosArrayPush(pOrderInfo, &order);
|
taosArrayPush(pOrderInfo, &order);
|
||||||
|
|
||||||
blockDataSort(b, pOrderInfo, true);
|
blockDataSort(b, pOrderInfo);
|
||||||
blockDataDestroy(b);
|
blockDataDestroy(b);
|
||||||
|
|
||||||
taosArrayDestroy(pOrderInfo);
|
taosArrayDestroy(pOrderInfo);
|
||||||
|
|
|
@ -430,9 +430,10 @@ typedef struct STagScanInfo {
|
||||||
} STagScanInfo;
|
} STagScanInfo;
|
||||||
|
|
||||||
typedef struct SStreamBlockScanInfo {
|
typedef struct SStreamBlockScanInfo {
|
||||||
|
SArray* pBlockLists; // multiple SSDatablock.
|
||||||
SSDataBlock* pRes; // result SSDataBlock
|
SSDataBlock* pRes; // result SSDataBlock
|
||||||
int32_t blockType; // current block type
|
int32_t blockType; // current block type
|
||||||
bool blockValid; // Is current data has returned?
|
int32_t validBlockIndex; // Is current data has returned?
|
||||||
SColumnInfo* pCols; // the output column info
|
SColumnInfo* pCols; // the output column info
|
||||||
uint64_t numOfRows; // total scanned rows
|
uint64_t numOfRows; // total scanned rows
|
||||||
uint64_t numOfExec; // execution times
|
uint64_t numOfExec; // execution times
|
||||||
|
@ -572,11 +573,13 @@ typedef struct SGroupbyOperatorInfo {
|
||||||
typedef struct SSessionAggOperatorInfo {
|
typedef struct SSessionAggOperatorInfo {
|
||||||
SOptrBasicInfo binfo;
|
SOptrBasicInfo binfo;
|
||||||
SAggSupporter aggSup;
|
SAggSupporter aggSup;
|
||||||
|
SGroupResInfo groupResInfo;
|
||||||
STimeWindow curWindow; // current time window
|
STimeWindow curWindow; // current time window
|
||||||
TSKEY prevTs; // previous timestamp
|
TSKEY prevTs; // previous timestamp
|
||||||
int32_t numOfRows; // number of rows
|
int32_t numOfRows; // number of rows
|
||||||
int32_t start; // start row index
|
int32_t start; // start row index
|
||||||
bool reptScan; // next round scan
|
bool reptScan; // next round scan
|
||||||
|
int64_t gap; // session window gap
|
||||||
} SSessionAggOperatorInfo;
|
} SSessionAggOperatorInfo;
|
||||||
|
|
||||||
typedef struct SStateWindowOperatorInfo {
|
typedef struct SStateWindowOperatorInfo {
|
||||||
|
@ -593,8 +596,7 @@ typedef struct SSortedMergeOperatorInfo {
|
||||||
SOptrBasicInfo binfo;
|
SOptrBasicInfo binfo;
|
||||||
bool hasVarCol;
|
bool hasVarCol;
|
||||||
|
|
||||||
SArray *orderInfo; // SArray<SBlockOrderInfo>
|
SArray* pSortInfo;
|
||||||
bool nullFirst;
|
|
||||||
int32_t numOfSources;
|
int32_t numOfSources;
|
||||||
|
|
||||||
SSortHandle *pSortHandle;
|
SSortHandle *pSortHandle;
|
||||||
|
@ -613,12 +615,10 @@ typedef struct SSortedMergeOperatorInfo {
|
||||||
SAggSupporter aggSup;
|
SAggSupporter aggSup;
|
||||||
} SSortedMergeOperatorInfo;
|
} SSortedMergeOperatorInfo;
|
||||||
|
|
||||||
typedef struct SOrderOperatorInfo {
|
typedef struct SSortOperatorInfo {
|
||||||
uint32_t sortBufSize; // max buffer size for in-memory sort
|
uint32_t sortBufSize; // max buffer size for in-memory sort
|
||||||
SSDataBlock *pDataBlock;
|
SSDataBlock *pDataBlock;
|
||||||
bool hasVarCol; // has variable length column, such as binary/varchar/nchar
|
SArray* pSortInfo;
|
||||||
SArray *orderInfo;
|
|
||||||
bool nullFirst;
|
|
||||||
SSortHandle *pSortHandle;
|
SSortHandle *pSortHandle;
|
||||||
int32_t bufPageSize;
|
int32_t bufPageSize;
|
||||||
int32_t numOfRowsInRes;
|
int32_t numOfRowsInRes;
|
||||||
|
@ -629,7 +629,7 @@ typedef struct SOrderOperatorInfo {
|
||||||
uint64_t totalSize; // total load bytes from remote
|
uint64_t totalSize; // total load bytes from remote
|
||||||
uint64_t totalRows; // total number of rows
|
uint64_t totalRows; // total number of rows
|
||||||
uint64_t totalElapsed; // total elapsed time
|
uint64_t totalElapsed; // total elapsed time
|
||||||
} SOrderOperatorInfo;
|
} SSortOperatorInfo;
|
||||||
|
|
||||||
typedef struct SDistinctDataInfo {
|
typedef struct SDistinctDataInfo {
|
||||||
int32_t index;
|
int32_t index;
|
||||||
|
@ -655,15 +655,15 @@ SOperatorInfo* createAggregateOperatorInfo(SOperatorInfo* downstream, SExprInfo*
|
||||||
SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
|
SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
|
||||||
SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
|
SOperatorInfo* createMultiTableAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
|
||||||
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
|
SOperatorInfo* createProjectOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t num, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
|
||||||
SOperatorInfo* createOrderOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SArray* pOrderVal, SExecTaskInfo* pTaskInfo);
|
SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExecTaskInfo* pTaskInfo);
|
||||||
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pOrderVal, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
|
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo);
|
||||||
SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataBlock* pResBlock, const SName* pName,
|
SOperatorInfo* createSysTableScanOperatorInfo(void* pSysTableReadHandle, SSDataBlock* pResBlock, const SName* pName,
|
||||||
SNode* pCondition, SEpSet epset, SArray* colList, SExecTaskInfo* pTaskInfo, bool showRewrite, int32_t accountId);
|
SNode* pCondition, SEpSet epset, SArray* colList, SExecTaskInfo* pTaskInfo, bool showRewrite, int32_t accountId);
|
||||||
SOperatorInfo* createLimitOperatorInfo(SOperatorInfo* downstream, SLimit* pLimit, SExecTaskInfo* pTaskInfo);
|
SOperatorInfo* createLimitOperatorInfo(SOperatorInfo* downstream, SLimit* pLimit, SExecTaskInfo* pTaskInfo);
|
||||||
|
|
||||||
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval,
|
SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SInterval* pInterval,
|
||||||
const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
|
const STableGroupInfo* pTableGroupInfo, SExecTaskInfo* pTaskInfo);
|
||||||
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo);
|
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap, SExecTaskInfo* pTaskInfo);
|
||||||
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
|
SOperatorInfo* createGroupOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResultBlock,
|
||||||
SArray* pGroupColList, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
|
SArray* pGroupColList, SExecTaskInfo* pTaskInfo, const STableGroupInfo* pTableGroupInfo);
|
||||||
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
|
SOperatorInfo* createFillOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExpr, int32_t numOfCols, SInterval* pInterval, SSDataBlock* pResBlock,
|
||||||
|
|
|
@ -63,7 +63,7 @@ typedef int32_t (*_sort_merge_compar_fn_t)(const void* p1, const void* p2, void*
|
||||||
* @param type
|
* @param type
|
||||||
* @return
|
* @return
|
||||||
*/
|
*/
|
||||||
SSortHandle* tsortCreateSortHandle(SArray* pOrderInfo, bool nullFirst, int32_t type, int32_t pageSize, int32_t numOfPages, SSchema* pSchema, int32_t numOfCols, const char* idstr);
|
SSortHandle* tsortCreateSortHandle(SArray* pOrderInfo, int32_t type, int32_t pageSize, int32_t numOfPages, SSDataBlock* pBlock, const char* idstr);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
*
|
*
|
||||||
|
|
|
@ -19,8 +19,6 @@
|
||||||
|
|
||||||
#include "executil.h"
|
#include "executil.h"
|
||||||
#include "executorimpl.h"
|
#include "executorimpl.h"
|
||||||
//#include "queryLog.h"
|
|
||||||
#include "tbuffer.h"
|
|
||||||
#include "tcompression.h"
|
#include "tcompression.h"
|
||||||
#include "tlosertree.h"
|
#include "tlosertree.h"
|
||||||
|
|
||||||
|
|
|
@ -14,11 +14,12 @@
|
||||||
*/
|
*/
|
||||||
|
|
||||||
#include "executor.h"
|
#include "executor.h"
|
||||||
|
#include "tdatablock.h"
|
||||||
#include "executorimpl.h"
|
#include "executorimpl.h"
|
||||||
#include "planner.h"
|
#include "planner.h"
|
||||||
#include "vnode.h"
|
#include "vnode.h"
|
||||||
|
|
||||||
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, int32_t type, char* id) {
|
static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void** input, size_t numOfBlocks, int32_t type, char* id) {
|
||||||
ASSERT(pOperator != NULL);
|
ASSERT(pOperator != NULL);
|
||||||
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
|
if (pOperator->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
|
||||||
if (pOperator->numOfDownstream == 0) {
|
if (pOperator->numOfDownstream == 0) {
|
||||||
|
@ -31,7 +32,7 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, int32_t t
|
||||||
return TSDB_CODE_QRY_APP_ERROR;
|
return TSDB_CODE_QRY_APP_ERROR;
|
||||||
}
|
}
|
||||||
pOperator->status = OP_NOT_OPENED;
|
pOperator->status = OP_NOT_OPENED;
|
||||||
return doSetStreamBlock(pOperator->pDownstream[0], input, type, id);
|
return doSetStreamBlock(pOperator->pDownstream[0], input, numOfBlocks, type, id);
|
||||||
} else {
|
} else {
|
||||||
SStreamBlockScanInfo* pInfo = pOperator->info;
|
SStreamBlockScanInfo* pInfo = pOperator->info;
|
||||||
|
|
||||||
|
@ -43,20 +44,20 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, int32_t t
|
||||||
}
|
}
|
||||||
|
|
||||||
if (type == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
|
if (type == STREAM_DATA_TYPE_SUBMIT_BLOCK) {
|
||||||
if (tqReadHandleSetMsg(pInfo->readerHandle, input, 0) < 0) {
|
if (tqReadHandleSetMsg(pInfo->readerHandle, input[0], 0) < 0) {
|
||||||
qError("submit msg messed up when initing stream block, %s" PRIx64, id);
|
qError("submit msg messed up when initing stream block, %s" PRIx64, id);
|
||||||
return TSDB_CODE_QRY_APP_ERROR;
|
return TSDB_CODE_QRY_APP_ERROR;
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
ASSERT(!pInfo->blockValid);
|
for (int32_t i = 0; i < numOfBlocks; ++i) {
|
||||||
|
SSDataBlock* pDataBlock = input[i];
|
||||||
|
|
||||||
SSDataBlock* pDataBlock = input;
|
SSDataBlock* p = createOneDataBlock(pDataBlock);
|
||||||
pInfo->pRes->info = pDataBlock->info;
|
p->info = pDataBlock->info;
|
||||||
taosArrayClear(pInfo->pRes->pDataBlock);
|
|
||||||
taosArrayAddAll(pInfo->pRes->pDataBlock, pDataBlock->pDataBlock);
|
|
||||||
|
|
||||||
// set current block valid.
|
taosArrayAddAll(p->pDataBlock, pDataBlock->pDataBlock);
|
||||||
pInfo->blockValid = true;
|
taosArrayPush(pInfo->pBlockLists, &p);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
|
@ -64,17 +65,21 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, int32_t t
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type) {
|
int32_t qSetStreamInput(qTaskInfo_t tinfo, const void* input, int32_t type) {
|
||||||
|
qSetMultiStreamInput(tinfo, (void**) &input, 1, type);
|
||||||
|
}
|
||||||
|
|
||||||
|
int32_t qSetMultiStreamInput(qTaskInfo_t tinfo, void** pBlocks, size_t numOfBlocks, int32_t type) {
|
||||||
if (tinfo == NULL) {
|
if (tinfo == NULL) {
|
||||||
return TSDB_CODE_QRY_APP_ERROR;
|
return TSDB_CODE_QRY_APP_ERROR;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (input == NULL) {
|
if (pBlocks == NULL || numOfBlocks == 0) {
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
||||||
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
|
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
|
||||||
|
|
||||||
int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void*)input, type, GET_TASKID(pTaskInfo));
|
int32_t code = doSetStreamBlock(pTaskInfo->pRoot, (void**)pBlocks, numOfBlocks, type, GET_TASKID(pTaskInfo));
|
||||||
if (code != TSDB_CODE_SUCCESS) {
|
if (code != TSDB_CODE_SUCCESS) {
|
||||||
qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
|
qError("%s failed to set the stream block data", GET_TASKID(pTaskInfo));
|
||||||
} else {
|
} else {
|
||||||
|
|
|
@ -69,6 +69,7 @@ typedef enum SResultTsInterpType {
|
||||||
typedef struct SColMatchInfo {
|
typedef struct SColMatchInfo {
|
||||||
int32_t colId;
|
int32_t colId;
|
||||||
int32_t targetSlotId;
|
int32_t targetSlotId;
|
||||||
|
bool output;
|
||||||
} SColMatchInfo;
|
} SColMatchInfo;
|
||||||
|
|
||||||
#if 0
|
#if 0
|
||||||
|
@ -322,26 +323,6 @@ static void sortGroupResByOrderList(SGroupResInfo *pGroupResInfo, STaskRuntimeEn
|
||||||
}
|
}
|
||||||
|
|
||||||
//setup the output buffer for each operator
|
//setup the output buffer for each operator
|
||||||
SSDataBlock* createOutputBuf(SExprInfo* pExpr, int32_t numOfOutput, int32_t numOfRows) {
|
|
||||||
const static int32_t minSize = 8;
|
|
||||||
|
|
||||||
SSDataBlock *res = taosMemoryCalloc(1, sizeof(SSDataBlock));
|
|
||||||
res->info.numOfCols = numOfOutput;
|
|
||||||
res->pDataBlock = taosArrayInit(numOfOutput, sizeof(SColumnInfoData));
|
|
||||||
for (int32_t i = 0; i < numOfOutput; ++i) {
|
|
||||||
SColumnInfoData idata = {{0}};
|
|
||||||
idata.info.type = pExpr[i].base.resSchema.type;
|
|
||||||
idata.info.bytes = pExpr[i].base.resSchema.bytes;
|
|
||||||
idata.info.colId = pExpr[i].base.resSchema.colId;
|
|
||||||
|
|
||||||
int32_t size = TMAX(idata.info.bytes * numOfRows, minSize);
|
|
||||||
idata.pData = taosMemoryCalloc(1, size); // at least to hold a pointer on x64 platform
|
|
||||||
taosArrayPush(res->pDataBlock, &idata);
|
|
||||||
}
|
|
||||||
|
|
||||||
return res;
|
|
||||||
}
|
|
||||||
|
|
||||||
SSDataBlock* createOutputBuf_rv1(SDataBlockDescNode* pNode) {
|
SSDataBlock* createOutputBuf_rv1(SDataBlockDescNode* pNode) {
|
||||||
int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
|
int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
|
||||||
|
|
||||||
|
@ -350,11 +331,15 @@ SSDataBlock* createOutputBuf_rv1(SDataBlockDescNode* pNode) {
|
||||||
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
|
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
|
||||||
|
|
||||||
pBlock->info.blockId = pNode->dataBlockId;
|
pBlock->info.blockId = pNode->dataBlockId;
|
||||||
pBlock->info.rowSize = pNode->resultRowSize;
|
pBlock->info.rowSize = pNode->resultRowSize; // todo ??
|
||||||
|
|
||||||
for(int32_t i = 0; i < numOfCols; ++i) {
|
for(int32_t i = 0; i < numOfCols; ++i) {
|
||||||
SColumnInfoData idata = {{0}};
|
SColumnInfoData idata = {{0}};
|
||||||
SSlotDescNode* pDescNode = nodesListGetNode(pNode->pSlots, i);
|
SSlotDescNode* pDescNode = nodesListGetNode(pNode->pSlots, i);
|
||||||
|
if (!pDescNode->output) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
idata.info.type = pDescNode->dataType.type;
|
idata.info.type = pDescNode->dataType.type;
|
||||||
idata.info.bytes = pDescNode->dataType.bytes;
|
idata.info.bytes = pDescNode->dataType.bytes;
|
||||||
idata.info.scale = pDescNode->dataType.scale;
|
idata.info.scale = pDescNode->dataType.scale;
|
||||||
|
@ -1813,25 +1798,27 @@ static void doHashGroupbyAgg(SOperatorInfo* pOperator, SSDataBlock *pBlock) {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperatorInfo *pInfo, SSDataBlock *pSDataBlock) {
|
// todo handle multiple tables cases.
|
||||||
STaskRuntimeEnv* pRuntimeEnv = pOperator->pRuntimeEnv;
|
static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperatorInfo *pInfo, SSDataBlock *pBlock) {
|
||||||
STableQueryInfo* item = pRuntimeEnv->current;
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
||||||
|
|
||||||
// primary timestamp column
|
// primary timestamp column
|
||||||
SColumnInfoData* pColInfoData = taosArrayGet(pSDataBlock->pDataBlock, 0);
|
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, 0);
|
||||||
|
|
||||||
bool masterScan = IS_MAIN_SCAN(pRuntimeEnv);
|
bool masterScan = true;
|
||||||
SOptrBasicInfo* pBInfo = &pInfo->binfo;
|
STimeWindow window = {0};
|
||||||
|
int32_t numOfOutput = pOperator->numOfOutput;
|
||||||
|
int64_t gid = pBlock->info.groupId;
|
||||||
|
|
||||||
int64_t gap = pOperator->pRuntimeEnv->pQueryAttr->sw.gap;
|
int64_t gap = pInfo->gap;
|
||||||
pInfo->numOfRows = 0;
|
pInfo->numOfRows = 0;
|
||||||
if (IS_REPEAT_SCAN(pRuntimeEnv) && !pInfo->reptScan) {
|
if (/*IS_REPEAT_SCAN(pRuntimeEnv) && */!pInfo->reptScan) {
|
||||||
pInfo->reptScan = true;
|
pInfo->reptScan = true;
|
||||||
pInfo->prevTs = INT64_MIN;
|
pInfo->prevTs = INT64_MIN;
|
||||||
}
|
}
|
||||||
|
|
||||||
TSKEY* tsList = (TSKEY*)pColInfoData->pData;
|
TSKEY* tsList = (TSKEY*)pColInfoData->pData;
|
||||||
for (int32_t j = 0; j < pSDataBlock->info.rows; ++j) {
|
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
|
||||||
if (pInfo->prevTs == INT64_MIN) {
|
if (pInfo->prevTs == INT64_MIN) {
|
||||||
pInfo->curWindow.skey = tsList[j];
|
pInfo->curWindow.skey = tsList[j];
|
||||||
pInfo->curWindow.ekey = tsList[j];
|
pInfo->curWindow.ekey = tsList[j];
|
||||||
|
@ -1848,17 +1835,15 @@ static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperator
|
||||||
}
|
}
|
||||||
} else { // start a new session window
|
} else { // start a new session window
|
||||||
SResultRow* pResult = NULL;
|
SResultRow* pResult = NULL;
|
||||||
|
|
||||||
pInfo->curWindow.ekey = pInfo->curWindow.skey;
|
pInfo->curWindow.ekey = pInfo->curWindow.skey;
|
||||||
int32_t ret = setResultOutputBufByKey(pRuntimeEnv, &pBInfo->resultRowInfo, pSDataBlock->info.uid, &pInfo->curWindow, masterScan,
|
int32_t ret = setResultOutputBufByKey_rv(&pInfo->binfo.resultRowInfo, pBlock->info.uid, &window, masterScan, &pResult, gid, pInfo->binfo.pCtx,
|
||||||
&pResult, item->groupIndex, pBInfo->pCtx, pOperator->numOfOutput,
|
numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup, pTaskInfo);
|
||||||
pBInfo->rowCellInfoOffset);
|
|
||||||
if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
|
if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
|
||||||
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_APP_ERROR);
|
longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
|
||||||
}
|
}
|
||||||
|
|
||||||
// doApplyFunctions(pRuntimeEnv, pBInfo->pCtx, &pInfo->curWindow, pInfo->start, pInfo->numOfRows, tsList,
|
// pInfo->numOfRows data belong to the current session window
|
||||||
// pSDataBlock->info.rows, pOperator->numOfOutput);
|
doApplyFunctions(pInfo->binfo.pCtx, &window, pInfo->start, pInfo->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
|
||||||
|
|
||||||
pInfo->curWindow.skey = tsList[j];
|
pInfo->curWindow.skey = tsList[j];
|
||||||
pInfo->curWindow.ekey = tsList[j];
|
pInfo->curWindow.ekey = tsList[j];
|
||||||
|
@ -1871,15 +1856,13 @@ static void doSessionWindowAggImpl(SOperatorInfo* pOperator, SSessionAggOperator
|
||||||
SResultRow* pResult = NULL;
|
SResultRow* pResult = NULL;
|
||||||
|
|
||||||
pInfo->curWindow.ekey = pInfo->curWindow.skey;
|
pInfo->curWindow.ekey = pInfo->curWindow.skey;
|
||||||
int32_t ret = setResultOutputBufByKey(pRuntimeEnv, &pBInfo->resultRowInfo, pSDataBlock->info.uid, &pInfo->curWindow, masterScan,
|
int32_t ret = setResultOutputBufByKey_rv(&pInfo->binfo.resultRowInfo, pBlock->info.uid, &window, masterScan, &pResult, gid, pInfo->binfo.pCtx,
|
||||||
&pResult, item->groupIndex, pBInfo->pCtx, pOperator->numOfOutput,
|
numOfOutput, pInfo->binfo.rowCellInfoOffset, &pInfo->aggSup, pTaskInfo);
|
||||||
pBInfo->rowCellInfoOffset);
|
|
||||||
if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
|
if (ret != TSDB_CODE_SUCCESS) { // null data, too many state code
|
||||||
longjmp(pRuntimeEnv->env, TSDB_CODE_QRY_APP_ERROR);
|
longjmp(pTaskInfo->env, TSDB_CODE_QRY_APP_ERROR);
|
||||||
}
|
}
|
||||||
|
|
||||||
// doApplyFunctions(pRuntimeEnv, pBInfo->pCtx, &pInfo->curWindow, pInfo->start, pInfo->numOfRows, tsList,
|
doApplyFunctions(pInfo->binfo.pCtx, &window, pInfo->start, pInfo->numOfRows, NULL, pBlock->info.rows, numOfOutput, TSDB_ORDER_ASC);
|
||||||
// pSDataBlock->info.rows, pOperator->numOfOutput);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
static void setResultRowKey(SResultRow* pResultRow, char* pData, int16_t type) {
|
static void setResultRowKey(SResultRow* pResultRow, char* pData, int16_t type) {
|
||||||
|
@ -2883,16 +2866,19 @@ int32_t loadDataBlock(SExecTaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo,
|
||||||
return terrno;
|
return terrno;
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t numOfCols = pBlock->info.numOfCols;
|
int32_t numOfCols = pBlock->info.numOfCols;
|
||||||
for(int32_t i = 0; i < numOfCols; ++i) {
|
for (int32_t i = 0; i < numOfCols; ++i) {
|
||||||
SColumnInfoData* p = taosArrayGet(pCols, i);
|
SColumnInfoData* p = taosArrayGet(pCols, i);
|
||||||
SColMatchInfo* pColMatchInfo = taosArrayGet(pTableScanInfo->pColMatchInfo, i);
|
SColMatchInfo* pColMatchInfo = taosArrayGet(pTableScanInfo->pColMatchInfo, i);
|
||||||
ASSERT(pColMatchInfo->colId == p->info.colId);
|
if (!pColMatchInfo->output) {
|
||||||
|
continue;
|
||||||
taosArraySet(pBlock->pDataBlock, pColMatchInfo->targetSlotId, p);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return TSDB_CODE_SUCCESS;
|
ASSERT(pColMatchInfo->colId == p->info.colId);
|
||||||
|
taosArraySet(pBlock->pDataBlock, pColMatchInfo->targetSlotId, p);
|
||||||
|
}
|
||||||
|
|
||||||
|
return TSDB_CODE_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t loadDataBlockOnDemand(SExecTaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
|
int32_t loadDataBlockOnDemand(SExecTaskInfo *pTaskInfo, STableScanInfo* pTableScanInfo, SSDataBlock* pBlock, uint32_t* status) {
|
||||||
|
@ -4738,6 +4724,17 @@ static SSDataBlock* doBlockInfoScan(SOperatorInfo *pOperator, bool* newgroup) {
|
||||||
#endif
|
#endif
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static void doClearBufferedBlocks(SStreamBlockScanInfo* pInfo) {
|
||||||
|
size_t total = taosArrayGetSize(pInfo->pBlockLists);
|
||||||
|
|
||||||
|
pInfo->validBlockIndex = 0;
|
||||||
|
for(int32_t i = 0; i < total; ++i) {
|
||||||
|
SSDataBlock* p = taosArrayGet(pInfo->pBlockLists, i);
|
||||||
|
blockDataDestroy(p);
|
||||||
|
}
|
||||||
|
taosArrayClear(pInfo->pBlockLists);
|
||||||
|
}
|
||||||
|
|
||||||
static SSDataBlock* doStreamBlockScan(SOperatorInfo *pOperator, bool* newgroup) {
|
static SSDataBlock* doStreamBlockScan(SOperatorInfo *pOperator, bool* newgroup) {
|
||||||
// NOTE: this operator does never check if current status is done or not
|
// NOTE: this operator does never check if current status is done or not
|
||||||
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
||||||
|
@ -4750,43 +4747,45 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo *pOperator, bool* newgroup)
|
||||||
}
|
}
|
||||||
|
|
||||||
if (pInfo->blockType == STREAM_DATA_TYPE_SSDATA_BLOCK) {
|
if (pInfo->blockType == STREAM_DATA_TYPE_SSDATA_BLOCK) {
|
||||||
if (pInfo->blockValid) {
|
size_t total = taosArrayGetSize(pInfo->pBlockLists);
|
||||||
pInfo->blockValid = false; // this block can only be used once.
|
if (pInfo->validBlockIndex >= total) {
|
||||||
return pInfo->pRes;
|
doClearBufferedBlocks(pInfo);
|
||||||
} else {
|
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
int32_t current = pInfo->validBlockIndex++;
|
||||||
|
return taosArrayGet(pInfo->pBlockLists, current);
|
||||||
|
} else {
|
||||||
|
SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
|
||||||
|
blockDataCleanup(pInfo->pRes);
|
||||||
|
|
||||||
|
while (tqNextDataBlock(pInfo->readerHandle)) {
|
||||||
|
pTaskInfo->code = tqRetrieveDataBlockInfo(pInfo->readerHandle, pBlockInfo);
|
||||||
|
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
|
||||||
|
terrno = pTaskInfo->code;
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (pBlockInfo->rows == 0) {
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
pInfo->pRes->pDataBlock = tqRetrieveDataBlock(pInfo->readerHandle);
|
||||||
|
if (pInfo->pRes->pDataBlock == NULL) {
|
||||||
|
// TODO add log
|
||||||
|
pTaskInfo->code = terrno;
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
// record the scan action.
|
||||||
|
pInfo->numOfExec++;
|
||||||
|
pInfo->numOfRows += pBlockInfo->rows;
|
||||||
|
|
||||||
|
return (pBlockInfo->rows == 0) ? NULL : pInfo->pRes;
|
||||||
}
|
}
|
||||||
|
|
||||||
SDataBlockInfo* pBlockInfo = &pInfo->pRes->info;
|
|
||||||
blockDataCleanup(pInfo->pRes);
|
|
||||||
|
|
||||||
while (tqNextDataBlock(pInfo->readerHandle)) {
|
|
||||||
pTaskInfo->code = tqRetrieveDataBlockInfo(pInfo->readerHandle, pBlockInfo);
|
|
||||||
if (pTaskInfo->code != TSDB_CODE_SUCCESS) {
|
|
||||||
terrno = pTaskInfo->code;
|
|
||||||
return NULL;
|
|
||||||
}
|
|
||||||
|
|
||||||
if (pBlockInfo->rows == 0) {
|
|
||||||
return NULL;
|
|
||||||
}
|
|
||||||
|
|
||||||
pInfo->pRes->pDataBlock = tqRetrieveDataBlock(pInfo->readerHandle);
|
|
||||||
if (pInfo->pRes->pDataBlock == NULL) {
|
|
||||||
// TODO add log
|
|
||||||
pTaskInfo->code = terrno;
|
|
||||||
return NULL;
|
|
||||||
}
|
|
||||||
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
|
|
||||||
// record the scan action.
|
|
||||||
pInfo->numOfExec++;
|
|
||||||
pInfo->numOfRows += pBlockInfo->rows;
|
|
||||||
|
|
||||||
return (pBlockInfo->rows == 0)? NULL:pInfo->pRes;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t loadRemoteDataCallback(void* param, const SDataBuf* pMsg, int32_t code) {
|
int32_t loadRemoteDataCallback(void* param, const SDataBuf* pMsg, int32_t code) {
|
||||||
|
@ -5423,6 +5422,13 @@ SOperatorInfo* createStreamScanOperatorInfo(void *streamReadHandle, SSDataBlock*
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pInfo->pBlockLists = taosArrayInit(4, POINTER_BYTES);
|
||||||
|
if (pInfo->pBlockLists == NULL) {
|
||||||
|
taosMemoryFreeClear(pInfo);
|
||||||
|
taosMemoryFreeClear(pOperator);
|
||||||
|
return NULL;
|
||||||
|
}
|
||||||
|
|
||||||
pInfo->readerHandle = streamReadHandle;
|
pInfo->readerHandle = streamReadHandle;
|
||||||
pInfo->pRes = pResBlock;
|
pInfo->pRes = pResBlock;
|
||||||
|
|
||||||
|
@ -5840,14 +5846,14 @@ static void cleanupAggSup(SAggSupporter* pAggSup);
|
||||||
|
|
||||||
static void destroySortedMergeOperatorInfo(void* param, int32_t numOfOutput) {
|
static void destroySortedMergeOperatorInfo(void* param, int32_t numOfOutput) {
|
||||||
SSortedMergeOperatorInfo* pInfo = (SSortedMergeOperatorInfo*) param;
|
SSortedMergeOperatorInfo* pInfo = (SSortedMergeOperatorInfo*) param;
|
||||||
taosArrayDestroy(pInfo->orderInfo);
|
taosArrayDestroy(pInfo->pSortInfo);
|
||||||
taosArrayDestroy(pInfo->groupInfo);
|
taosArrayDestroy(pInfo->groupInfo);
|
||||||
|
|
||||||
if (pInfo->pSortHandle != NULL) {
|
if (pInfo->pSortHandle != NULL) {
|
||||||
tsortDestroySortHandle(pInfo->pSortHandle);
|
tsortDestroySortHandle(pInfo->pSortHandle);
|
||||||
}
|
}
|
||||||
blockDataDestroy(pInfo->binfo.pRes);
|
|
||||||
|
|
||||||
|
blockDataDestroy(pInfo->binfo.pRes);
|
||||||
cleanupAggSup(&pInfo->aggSup);
|
cleanupAggSup(&pInfo->aggSup);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -6105,12 +6111,10 @@ static SSDataBlock* doSortedMerge(SOperatorInfo *pOperator, bool* newgroup) {
|
||||||
return getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pInfo->hasVarCol, pInfo->binfo.capacity);
|
return getSortedBlockData(pInfo->pSortHandle, pInfo->binfo.pRes, pInfo->hasVarCol, pInfo->binfo.capacity);
|
||||||
}
|
}
|
||||||
|
|
||||||
SSchema* p = blockDataExtractSchema(pInfo->binfo.pRes, NULL);
|
|
||||||
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
|
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
|
||||||
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->orderInfo, pInfo->nullFirst, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize,
|
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_MULTISOURCE_MERGE, pInfo->bufPageSize,
|
||||||
numOfBufPage, p, pInfo->binfo.pRes->info.numOfCols, "GET_TASKID(pTaskInfo)");
|
numOfBufPage, pInfo->binfo.pRes, "GET_TASKID(pTaskInfo)");
|
||||||
|
|
||||||
taosMemoryFreeClear(p);
|
|
||||||
tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock);
|
tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock);
|
||||||
|
|
||||||
for(int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
|
for(int32_t i = 0; i < pOperator->numOfDownstream; ++i) {
|
||||||
|
@ -6128,29 +6132,6 @@ static SSDataBlock* doSortedMerge(SOperatorInfo *pOperator, bool* newgroup) {
|
||||||
return doMerge(pOperator);
|
return doMerge(pOperator);
|
||||||
}
|
}
|
||||||
|
|
||||||
static SArray* createBlockOrder(SExprInfo* pExprInfo, int32_t numOfCols, SArray* pOrderVal) {
|
|
||||||
SArray* pOrderInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));
|
|
||||||
|
|
||||||
size_t numOfOrder = taosArrayGetSize(pOrderVal);
|
|
||||||
for (int32_t j = 0; j < numOfOrder; ++j) {
|
|
||||||
SBlockOrderInfo orderInfo = {0};
|
|
||||||
SOrder* pOrder = taosArrayGet(pOrderVal, j);
|
|
||||||
orderInfo.order = pOrder->order;
|
|
||||||
|
|
||||||
for (int32_t i = 0; i < numOfCols; ++i) {
|
|
||||||
SExprInfo* pExpr = &pExprInfo[i];
|
|
||||||
if (pExpr->base.resSchema.colId == pOrder->col.colId) {
|
|
||||||
orderInfo.colIndex = i;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
taosArrayPush(pOrderInfo, &orderInfo);
|
|
||||||
}
|
|
||||||
|
|
||||||
return pOrderInfo;
|
|
||||||
}
|
|
||||||
|
|
||||||
static int32_t initGroupCol(SExprInfo* pExprInfo, int32_t numOfCols, SArray* pGroupInfo, SSortedMergeOperatorInfo* pInfo) {
|
static int32_t initGroupCol(SExprInfo* pExprInfo, int32_t numOfCols, SArray* pGroupInfo, SSortedMergeOperatorInfo* pInfo) {
|
||||||
if (pGroupInfo == NULL || taosArrayGetSize(pGroupInfo) == 0) {
|
if (pGroupInfo == NULL || taosArrayGetSize(pGroupInfo) == 0) {
|
||||||
return 0;
|
return 0;
|
||||||
|
@ -6199,7 +6180,7 @@ static int32_t initGroupCol(SExprInfo* pExprInfo, int32_t numOfCols, SArray* pGr
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
||||||
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pOrderVal, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo) {
|
SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t numOfDownstream, SExprInfo* pExprInfo, int32_t num, SArray* pSortInfo, SArray* pGroupInfo, SExecTaskInfo* pTaskInfo) {
|
||||||
SSortedMergeOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortedMergeOperatorInfo));
|
SSortedMergeOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortedMergeOperatorInfo));
|
||||||
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
||||||
if (pInfo == NULL || pOperator == NULL) {
|
if (pInfo == NULL || pOperator == NULL) {
|
||||||
|
@ -6228,7 +6209,7 @@ SOperatorInfo* createSortedMergeOperatorInfo(SOperatorInfo** downstream, int32_t
|
||||||
// pRuntimeEnv->pQueryAttr->topBotQuery, false));
|
// pRuntimeEnv->pQueryAttr->topBotQuery, false));
|
||||||
pInfo->sortBufSize = 1024 * 16; // 1MB
|
pInfo->sortBufSize = 1024 * 16; // 1MB
|
||||||
pInfo->bufPageSize = 1024;
|
pInfo->bufPageSize = 1024;
|
||||||
pInfo->orderInfo = createBlockOrder(pExprInfo, num, pOrderVal);
|
pInfo->pSortInfo = pSortInfo;
|
||||||
|
|
||||||
pInfo->binfo.capacity = blockDataGetCapacityInRow(pInfo->binfo.pRes, pInfo->bufPageSize);
|
pInfo->binfo.capacity = blockDataGetCapacityInRow(pInfo->binfo.pRes, pInfo->bufPageSize);
|
||||||
|
|
||||||
|
@ -6268,35 +6249,34 @@ static SSDataBlock* doSort(SOperatorInfo *pOperator, bool* newgroup) {
|
||||||
}
|
}
|
||||||
|
|
||||||
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
||||||
SOrderOperatorInfo* pInfo = pOperator->info;
|
SSortOperatorInfo* pInfo = pOperator->info;
|
||||||
|
bool hasVarCol = pInfo->pDataBlock->info.hasVarCol;
|
||||||
|
|
||||||
if (pOperator->status == OP_RES_TO_RETURN) {
|
if (pOperator->status == OP_RES_TO_RETURN) {
|
||||||
return getSortedBlockData(pInfo->pSortHandle, pInfo->pDataBlock, pInfo->hasVarCol, pInfo->numOfRowsInRes);
|
return getSortedBlockData(pInfo->pSortHandle, pInfo->pDataBlock, hasVarCol, pInfo->numOfRowsInRes);
|
||||||
}
|
}
|
||||||
|
|
||||||
SSchema* p = blockDataExtractSchema(pInfo->pDataBlock, NULL);
|
|
||||||
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
|
int32_t numOfBufPage = pInfo->sortBufSize / pInfo->bufPageSize;
|
||||||
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->orderInfo, pInfo->nullFirst, SORT_SINGLESOURCE_SORT, pInfo->bufPageSize,
|
pInfo->pSortHandle = tsortCreateSortHandle(pInfo->pSortInfo, SORT_SINGLESOURCE_SORT, pInfo->bufPageSize,
|
||||||
numOfBufPage, p, pInfo->pDataBlock->info.numOfCols, "GET_TASKID(pTaskInfo)");
|
numOfBufPage, pInfo->pDataBlock, "GET_TASKID(pTaskInfo)");
|
||||||
|
|
||||||
taosMemoryFreeClear(p);
|
|
||||||
tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock);
|
tsortSetFetchRawDataFp(pInfo->pSortHandle, loadNextDataBlock);
|
||||||
|
|
||||||
SGenericSource* ps = taosMemoryCalloc(1, sizeof(SGenericSource));
|
SGenericSource* ps = taosMemoryCalloc(1, sizeof(SGenericSource));
|
||||||
ps->param = pOperator;
|
ps->param = pOperator->pDownstream[0];
|
||||||
tsortAddSource(pInfo->pSortHandle, ps);
|
tsortAddSource(pInfo->pSortHandle, ps);
|
||||||
|
|
||||||
// TODO set error code;
|
|
||||||
int32_t code = tsortOpen(pInfo->pSortHandle);
|
int32_t code = tsortOpen(pInfo->pSortHandle);
|
||||||
if (code != TSDB_CODE_SUCCESS) {
|
if (code != TSDB_CODE_SUCCESS) {
|
||||||
longjmp(pTaskInfo->env, terrno);
|
longjmp(pTaskInfo->env, terrno);
|
||||||
}
|
}
|
||||||
|
|
||||||
pOperator->status = OP_RES_TO_RETURN;
|
pOperator->status = OP_RES_TO_RETURN;
|
||||||
return getSortedBlockData(pInfo->pSortHandle, pInfo->pDataBlock, pInfo->hasVarCol, pInfo->numOfRowsInRes);
|
return getSortedBlockData(pInfo->pSortHandle, pInfo->pDataBlock, hasVarCol, pInfo->numOfRowsInRes);
|
||||||
}
|
}
|
||||||
|
|
||||||
SOperatorInfo *createOrderOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SArray* pOrderVal, SExecTaskInfo* pTaskInfo) {
|
SOperatorInfo *createSortOperatorInfo(SOperatorInfo* downstream, SSDataBlock* pResBlock, SArray* pSortInfo, SExecTaskInfo* pTaskInfo) {
|
||||||
SOrderOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SOrderOperatorInfo));
|
SSortOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSortOperatorInfo));
|
||||||
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
||||||
if (pInfo == NULL || pOperator == NULL) {
|
if (pInfo == NULL || pOperator == NULL) {
|
||||||
taosMemoryFreeClear(pInfo);
|
taosMemoryFreeClear(pInfo);
|
||||||
|
@ -6305,37 +6285,21 @@ SOperatorInfo *createOrderOperatorInfo(SOperatorInfo* downstream, SExprInfo* pEx
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
pInfo->sortBufSize = 1024 * 16; // 1MB
|
pInfo->sortBufSize = 1024 * 16; // 1MB, TODO dynamic set the available sort buffer
|
||||||
pInfo->bufPageSize = 1024;
|
pInfo->bufPageSize = 1024;
|
||||||
pInfo->numOfRowsInRes = 1024;
|
pInfo->numOfRowsInRes = 1024;
|
||||||
|
pInfo->pDataBlock = pResBlock;
|
||||||
|
pInfo->pSortInfo = pSortInfo;
|
||||||
|
|
||||||
pInfo->orderInfo = createBlockOrder(pExprInfo, numOfCols, pOrderVal);
|
pOperator->name = "Sort";
|
||||||
|
|
||||||
for(int32_t i = 0; i < numOfCols; ++i) {
|
|
||||||
if (IS_VAR_DATA_TYPE(pExprInfo[i].base.resSchema.type)) {
|
|
||||||
pInfo->hasVarCol = true;
|
|
||||||
break;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (pInfo->orderInfo == NULL || pInfo->pDataBlock == NULL) {
|
|
||||||
taosMemoryFreeClear(pOperator);
|
|
||||||
destroyOrderOperatorInfo(pInfo, numOfCols);
|
|
||||||
taosMemoryFreeClear(pInfo);
|
|
||||||
|
|
||||||
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY;
|
|
||||||
return NULL;
|
|
||||||
}
|
|
||||||
|
|
||||||
pOperator->name = "Order";
|
|
||||||
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT;
|
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SORT;
|
||||||
pOperator->blockingOptr = true;
|
pOperator->blockingOptr = true;
|
||||||
pOperator->status = OP_NOT_OPENED;
|
pOperator->status = OP_NOT_OPENED;
|
||||||
pOperator->info = pInfo;
|
pOperator->info = pInfo;
|
||||||
|
|
||||||
pOperator->pTaskInfo = pTaskInfo;
|
pOperator->pTaskInfo = pTaskInfo;
|
||||||
pOperator->getNextFn = doSort;
|
pOperator->getNextFn = doSort;
|
||||||
pOperator->closeFn = destroyOrderOperatorInfo;
|
pOperator->closeFn = destroyOrderOperatorInfo;
|
||||||
|
|
||||||
int32_t code = appendDownstream(pOperator, &downstream, 1);
|
int32_t code = appendDownstream(pOperator, &downstream, 1);
|
||||||
return pOperator;
|
return pOperator;
|
||||||
|
@ -6496,7 +6460,6 @@ static bool aggDecodeResultRow(SOperatorInfo* pOperator, char *result, int32_t l
|
||||||
offset += valueLen;
|
offset += valueLen;
|
||||||
|
|
||||||
initResultRow(resultRow);
|
initResultRow(resultRow);
|
||||||
|
|
||||||
pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] = (SResultRowPosition) {.pageId = resultRow->pageId, .offset = resultRow->offset};
|
pInfo->resultRowInfo.pPosition[pInfo->resultRowInfo.size++] = (SResultRowPosition) {.pageId = resultRow->pageId, .offset = resultRow->offset};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -6741,7 +6704,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo *pOperator) {
|
||||||
|
|
||||||
STableIntervalOperatorInfo* pInfo = pOperator->info;
|
STableIntervalOperatorInfo* pInfo = pOperator->info;
|
||||||
|
|
||||||
// int32_t order = pQueryAttr->order.order;
|
int32_t order = TSDB_ORDER_ASC;
|
||||||
// STimeWindow win = pQueryAttr->window;
|
// STimeWindow win = pQueryAttr->window;
|
||||||
bool newgroup = false;
|
bool newgroup = false;
|
||||||
SOperatorInfo* downstream = pOperator->pDownstream[0];
|
SOperatorInfo* downstream = pOperator->pDownstream[0];
|
||||||
|
@ -6758,7 +6721,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo *pOperator) {
|
||||||
// setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->pCtx, pOperator->numOfOutput);
|
// setTagValue(pOperator, pRuntimeEnv->current->pTable, pInfo->pCtx, pOperator->numOfOutput);
|
||||||
|
|
||||||
// the pDataBlock are always the same one, no need to call this again
|
// the pDataBlock are always the same one, no need to call this again
|
||||||
setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, TSDB_ORDER_ASC);
|
setInputDataBlock(pOperator, pInfo->binfo.pCtx, pBlock, order);
|
||||||
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
|
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, 0);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -6770,7 +6733,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo *pOperator) {
|
||||||
return TSDB_CODE_SUCCESS;
|
return TSDB_CODE_SUCCESS;
|
||||||
}
|
}
|
||||||
|
|
||||||
static SSDataBlock* doIntervalAgg(SOperatorInfo *pOperator, bool* newgroup) {
|
static SSDataBlock* doBuildIntervalResult(SOperatorInfo *pOperator, bool* newgroup) {
|
||||||
STableIntervalOperatorInfo* pInfo = pOperator->info;
|
STableIntervalOperatorInfo* pInfo = pOperator->info;
|
||||||
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
|
||||||
|
|
||||||
|
@ -7102,13 +7065,14 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo *pOperator, bool* newgroup)
|
||||||
return NULL;
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
SSessionAggOperatorInfo* pWindowInfo = pOperator->info;
|
SSessionAggOperatorInfo* pInfo = pOperator->info;
|
||||||
SOptrBasicInfo* pBInfo = &pWindowInfo->binfo;
|
SOptrBasicInfo* pBInfo = &pInfo->binfo;
|
||||||
|
|
||||||
if (pOperator->status == OP_RES_TO_RETURN) {
|
if (pOperator->status == OP_RES_TO_RETURN) {
|
||||||
// toSDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pBInfo->pRes);
|
toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pBInfo->pRes, pBInfo->capacity, pBInfo->rowCellInfoOffset);
|
||||||
if (pBInfo->pRes->info.rows == 0/* || !hasRemainDataInCurrentGroup(&pRuntimeEnv->groupResInfo)*/) {
|
if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
|
||||||
pOperator->status = OP_EXEC_DONE;
|
doSetOperatorCompleted(pOperator);
|
||||||
|
return NULL;
|
||||||
}
|
}
|
||||||
|
|
||||||
return pBInfo->pRes;
|
return pBInfo->pRes;
|
||||||
|
@ -7127,19 +7091,20 @@ static SSDataBlock* doSessionWindowAgg(SOperatorInfo *pOperator, bool* newgroup)
|
||||||
|
|
||||||
// the pDataBlock are always the same one, no need to call this again
|
// the pDataBlock are always the same one, no need to call this again
|
||||||
setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order);
|
setInputDataBlock(pOperator, pBInfo->pCtx, pBlock, order);
|
||||||
doSessionWindowAggImpl(pOperator, pWindowInfo, pBlock);
|
doSessionWindowAggImpl(pOperator, pInfo, pBlock);
|
||||||
}
|
}
|
||||||
|
|
||||||
// restore the value
|
// restore the value
|
||||||
pOperator->status = OP_RES_TO_RETURN;
|
pOperator->status = OP_RES_TO_RETURN;
|
||||||
closeAllResultRows(&pBInfo->resultRowInfo);
|
closeAllResultRows(&pBInfo->resultRowInfo);
|
||||||
// setTaskStatus(pOperator->pTaskInfo, QUERY_COMPLETED);
|
finalizeMultiTupleQueryResult(pBInfo->pCtx, pOperator->numOfOutput, pInfo->aggSup.pResultBuf, &pBInfo->resultRowInfo, pBInfo->rowCellInfoOffset);
|
||||||
finalizeQueryResult(pBInfo->pCtx, pOperator->numOfOutput);
|
|
||||||
|
|
||||||
// initGroupResInfo(&pBInfo->groupResInfo, &pBInfo->resultRowInfo);
|
initGroupResInfo(&pInfo->groupResInfo, &pBInfo->resultRowInfo);
|
||||||
// toSDatablock(&pRuntimeEnv->groupResInfo, pRuntimeEnv, pBInfo->pRes);
|
|
||||||
if (pBInfo->pRes->info.rows == 0/* || !hasRemainDataInCurrentGroup(&pRuntimeEnv->groupResInfo)*/) {
|
blockDataEnsureCapacity(pBInfo->pRes, pBInfo->capacity);
|
||||||
pOperator->status = OP_EXEC_DONE;
|
toSDatablock(&pInfo->groupResInfo, pInfo->aggSup.pResultBuf, pBInfo->pRes, pBInfo->capacity, pBInfo->rowCellInfoOffset);
|
||||||
|
if (pBInfo->pRes->info.rows == 0 || !hasRemainDataInCurrentGroup(&pInfo->groupResInfo)) {
|
||||||
|
doSetOperatorCompleted(pOperator);
|
||||||
}
|
}
|
||||||
|
|
||||||
return pBInfo->pRes->info.rows == 0? NULL:pBInfo->pRes;
|
return pBInfo->pRes->info.rows == 0? NULL:pBInfo->pRes;
|
||||||
|
@ -7505,10 +7470,10 @@ static void destroyTagScanOperatorInfo(void* param, int32_t numOfOutput) {
|
||||||
}
|
}
|
||||||
|
|
||||||
static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) {
|
static void destroyOrderOperatorInfo(void* param, int32_t numOfOutput) {
|
||||||
SOrderOperatorInfo* pInfo = (SOrderOperatorInfo*) param;
|
SSortOperatorInfo* pInfo = (SSortOperatorInfo*) param;
|
||||||
pInfo->pDataBlock = blockDataDestroy(pInfo->pDataBlock);
|
pInfo->pDataBlock = blockDataDestroy(pInfo->pDataBlock);
|
||||||
|
|
||||||
taosArrayDestroy(pInfo->orderInfo);
|
taosArrayDestroy(pInfo->pSortInfo);
|
||||||
}
|
}
|
||||||
|
|
||||||
static void destroyConditionOperatorInfo(void* param, int32_t numOfOutput) {
|
static void destroyConditionOperatorInfo(void* param, int32_t numOfOutput) {
|
||||||
|
@ -7716,7 +7681,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
|
||||||
pOperator->numOfOutput = numOfCols;
|
pOperator->numOfOutput = numOfCols;
|
||||||
pOperator->info = pInfo;
|
pOperator->info = pInfo;
|
||||||
pOperator->_openFn = doOpenIntervalAgg;
|
pOperator->_openFn = doOpenIntervalAgg;
|
||||||
pOperator->getNextFn = doIntervalAgg;
|
pOperator->getNextFn = doBuildIntervalResult;
|
||||||
pOperator->closeFn = destroyIntervalOperatorInfo;
|
pOperator->closeFn = destroyIntervalOperatorInfo;
|
||||||
|
|
||||||
code = appendDownstream(pOperator, &downstream, 1);
|
code = appendDownstream(pOperator, &downstream, 1);
|
||||||
|
@ -7782,25 +7747,27 @@ SOperatorInfo* createStatewindowOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOper
|
||||||
return pOperator;
|
return pOperator;
|
||||||
}
|
}
|
||||||
|
|
||||||
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, SExecTaskInfo* pTaskInfo) {
|
SOperatorInfo* createSessionAggOperatorInfo(SOperatorInfo* downstream, SExprInfo* pExprInfo, int32_t numOfCols, SSDataBlock* pResBlock, int64_t gap, SExecTaskInfo* pTaskInfo) {
|
||||||
SSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSessionAggOperatorInfo));
|
SSessionAggOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SSessionAggOperatorInfo));
|
||||||
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;
|
||||||
}
|
}
|
||||||
|
|
||||||
int32_t code = doInitAggInfoSup(&pInfo->aggSup, pInfo->binfo.pCtx, numOfCols, pTaskInfo->id.str);
|
int32_t numOfRows = 4096;
|
||||||
|
int32_t code = initAggInfo(&pInfo->binfo, &pInfo->aggSup, pExprInfo, numOfCols, numOfRows, pResBlock, pTaskInfo->id.str);
|
||||||
if (code != TSDB_CODE_SUCCESS) {
|
if (code != TSDB_CODE_SUCCESS) {
|
||||||
goto _error;
|
goto _error;
|
||||||
}
|
}
|
||||||
|
|
||||||
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
|
initResultRowInfo(&pInfo->binfo.resultRowInfo, 8);
|
||||||
|
|
||||||
|
pInfo->gap = gap;
|
||||||
pInfo->binfo.pRes = pResBlock;
|
pInfo->binfo.pRes = pResBlock;
|
||||||
pInfo->prevTs = INT64_MIN;
|
pInfo->prevTs = INT64_MIN;
|
||||||
pInfo->reptScan = false;
|
pInfo->reptScan = false;
|
||||||
pOperator->name = "SessionWindowAggOperator";
|
pOperator->name = "SessionWindowAggOperator";
|
||||||
// pOperator->operatorType = OP_SessionWindow;
|
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW;
|
||||||
pOperator->blockingOptr = true;
|
pOperator->blockingOptr = true;
|
||||||
pOperator->status = OP_NOT_OPENED;
|
pOperator->status = OP_NOT_OPENED;
|
||||||
pOperator->pExpr = pExprInfo;
|
pOperator->pExpr = pExprInfo;
|
||||||
|
@ -8035,7 +8002,7 @@ SOperatorInfo* createSLimitOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperatorI
|
||||||
offset += pExpr[index->colIndex].base.resSchema.bytes;
|
offset += pExpr[index->colIndex].base.resSchema.bytes;
|
||||||
}
|
}
|
||||||
|
|
||||||
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pOperator->resultInfo.capacity);
|
// pInfo->pRes = createOutputBuf(pExpr, numOfOutput, pOperator->resultInfo.capacity);
|
||||||
|
|
||||||
|
|
||||||
pOperator->name = "SLimitOperator";
|
pOperator->name = "SLimitOperator";
|
||||||
|
@ -8328,7 +8295,7 @@ SOperatorInfo* createDistinctOperatorInfo(STaskRuntimeEnv* pRuntimeEnv, SOperato
|
||||||
pInfo->outputCapacity = 4096;
|
pInfo->outputCapacity = 4096;
|
||||||
pInfo->pDistinctDataInfo = taosArrayInit(numOfOutput, sizeof(SDistinctDataInfo));
|
pInfo->pDistinctDataInfo = taosArrayInit(numOfOutput, sizeof(SDistinctDataInfo));
|
||||||
pInfo->pSet = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
|
pInfo->pSet = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), false, HASH_NO_LOCK);
|
||||||
pInfo->pRes = createOutputBuf(pExpr, numOfOutput, (int32_t) pInfo->outputCapacity);
|
// pInfo->pRes = createOutputBuf(pExpr, numOfOutput, (int32_t) pInfo->outputCapacity);
|
||||||
|
|
||||||
|
|
||||||
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
|
||||||
|
@ -8535,16 +8502,17 @@ static int32_t doCreateTableGroup(void* metaHandle, int32_t tableType, uint64_t
|
||||||
static SArray* extractTableIdList(const STableGroupInfo* pTableGroupInfo);
|
static SArray* extractTableIdList(const STableGroupInfo* pTableGroupInfo);
|
||||||
static SArray* extractScanColumnId(SNodeList* pNodeList);
|
static SArray* extractScanColumnId(SNodeList* pNodeList);
|
||||||
static SArray* extractColumnInfo(SNodeList* pNodeList);
|
static SArray* extractColumnInfo(SNodeList* pNodeList);
|
||||||
static SArray* extractColMatchInfo(SNodeList* pNodeList);
|
static SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols);
|
||||||
|
static SArray* createSortInfo(SNodeList* pNodeList);
|
||||||
|
|
||||||
SOperatorInfo* doCreateOperatorTreeNode(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, uint64_t queryId, uint64_t taskId, STableGroupInfo* pTableGroupInfo) {
|
SOperatorInfo* doCreateOperatorTreeNode(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo, SReadHandle* pHandle, uint64_t queryId, uint64_t taskId, STableGroupInfo* pTableGroupInfo) {
|
||||||
if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
|
if (pPhyNode->pChildren == NULL || LIST_LENGTH(pPhyNode->pChildren) == 0) {
|
||||||
if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pPhyNode)) {
|
if (QUERY_NODE_PHYSICAL_PLAN_TABLE_SCAN == nodeType(pPhyNode)) {
|
||||||
SScanPhysiNode* pScanPhyNode = (SScanPhysiNode*)pPhyNode;
|
SScanPhysiNode* pScanPhyNode = (SScanPhysiNode*)pPhyNode;
|
||||||
|
|
||||||
size_t numOfCols = LIST_LENGTH(pScanPhyNode->pScanCols);
|
int32_t numOfCols = 0;
|
||||||
tsdbReaderT pDataReader = doCreateDataReader((STableScanPhysiNode*)pPhyNode, pHandle, pTableGroupInfo, (uint64_t)queryId, taskId);
|
tsdbReaderT pDataReader = doCreateDataReader((STableScanPhysiNode*)pPhyNode, pHandle, pTableGroupInfo, (uint64_t)queryId, taskId);
|
||||||
SArray* pColList = extractColMatchInfo(pScanPhyNode->pScanCols);
|
SArray* pColList = extractColMatchInfo(pScanPhyNode->pScanCols, pScanPhyNode->node.pOutputDataBlockDesc, &numOfCols);
|
||||||
|
|
||||||
return createTableScanOperatorInfo(pDataReader, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pScanPhyNode->reverse, pColList, pTaskInfo);
|
return createTableScanOperatorInfo(pDataReader, pScanPhyNode->order, numOfCols, pScanPhyNode->count, pScanPhyNode->reverse, pColList, pTaskInfo);
|
||||||
} else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pPhyNode)) {
|
} else if (QUERY_NODE_PHYSICAL_PLAN_EXCHANGE == nodeType(pPhyNode)) {
|
||||||
|
@ -8583,15 +8551,13 @@ SOperatorInfo* doCreateOperatorTreeNode(SPhysiNode* pPhyNode, SExecTaskInfo* pTa
|
||||||
size_t size = LIST_LENGTH(pPhyNode->pChildren);
|
size_t size = LIST_LENGTH(pPhyNode->pChildren);
|
||||||
assert(size == 1);
|
assert(size == 1);
|
||||||
|
|
||||||
for (int32_t i = 0; i < size; ++i) {
|
SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, 0);
|
||||||
SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, i);
|
SOperatorInfo* op = doCreateOperatorTreeNode(pChildNode, pTaskInfo, pHandle, queryId, taskId, pTableGroupInfo);
|
||||||
SOperatorInfo* op = doCreateOperatorTreeNode(pChildNode, pTaskInfo, pHandle, queryId, taskId, pTableGroupInfo);
|
|
||||||
|
|
||||||
int32_t num = 0;
|
int32_t num = 0;
|
||||||
SExprInfo* pExprInfo = createExprInfo(((SProjectPhysiNode*)pPhyNode)->pProjections, NULL, &num);
|
SExprInfo* pExprInfo = createExprInfo(((SProjectPhysiNode*)pPhyNode)->pProjections, NULL, &num);
|
||||||
SSDataBlock* pResBlock = createOutputBuf_rv1(pPhyNode->pOutputDataBlockDesc);
|
SSDataBlock* pResBlock = createOutputBuf_rv1(pPhyNode->pOutputDataBlockDesc);
|
||||||
return createProjectOperatorInfo(op, pExprInfo, num, pResBlock, pTaskInfo);
|
return createProjectOperatorInfo(op, pExprInfo, num, pResBlock, pTaskInfo);
|
||||||
}
|
|
||||||
} else if (QUERY_NODE_PHYSICAL_PLAN_AGG == nodeType(pPhyNode)) {
|
} else if (QUERY_NODE_PHYSICAL_PLAN_AGG == nodeType(pPhyNode)) {
|
||||||
size_t size = LIST_LENGTH(pPhyNode->pChildren);
|
size_t size = LIST_LENGTH(pPhyNode->pChildren);
|
||||||
assert(size == 1);
|
assert(size == 1);
|
||||||
|
@ -8623,16 +8589,46 @@ SOperatorInfo* doCreateOperatorTreeNode(SPhysiNode* pPhyNode, SExecTaskInfo* pTa
|
||||||
|
|
||||||
SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
|
SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
|
||||||
|
|
||||||
|
//todo: set the correct primary timestamp key column
|
||||||
int32_t num = 0;
|
int32_t num = 0;
|
||||||
SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &num);
|
SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &num);
|
||||||
SSDataBlock* pResBlock = createOutputBuf_rv1(pPhyNode->pOutputDataBlockDesc);
|
SSDataBlock* pResBlock = createOutputBuf_rv1(pPhyNode->pOutputDataBlockDesc);
|
||||||
|
|
||||||
SInterval interval = {.interval = pIntervalPhyNode->interval, .sliding = pIntervalPhyNode->sliding,
|
SInterval interval = {.interval = pIntervalPhyNode->interval,
|
||||||
|
.sliding = pIntervalPhyNode->sliding,
|
||||||
.intervalUnit = pIntervalPhyNode->intervalUnit,
|
.intervalUnit = pIntervalPhyNode->intervalUnit,
|
||||||
.slidingUnit = pIntervalPhyNode->slidingUnit, .offset = pIntervalPhyNode->offset};
|
.slidingUnit = pIntervalPhyNode->slidingUnit,
|
||||||
|
.offset = pIntervalPhyNode->offset};
|
||||||
return createIntervalOperatorInfo(op, pExprInfo, num, pResBlock, &interval, pTableGroupInfo, pTaskInfo);
|
return createIntervalOperatorInfo(op, pExprInfo, num, pResBlock, &interval, pTableGroupInfo, pTaskInfo);
|
||||||
}
|
}
|
||||||
} /*else if (pPhyNode->info.type == OP_MultiTableAggregate) {
|
} else if (QUERY_NODE_PHYSICAL_PLAN_SORT == nodeType(pPhyNode)) {
|
||||||
|
size_t size = LIST_LENGTH(pPhyNode->pChildren);
|
||||||
|
assert(size == 1);
|
||||||
|
|
||||||
|
SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, 0);
|
||||||
|
SOperatorInfo* op = doCreateOperatorTreeNode(pChildNode, pTaskInfo, pHandle, queryId, taskId, pTableGroupInfo);
|
||||||
|
|
||||||
|
SSortPhysiNode* pSortPhyNode = (SSortPhysiNode*)pPhyNode;
|
||||||
|
|
||||||
|
SSDataBlock* pResBlock = createOutputBuf_rv1(pPhyNode->pOutputDataBlockDesc);
|
||||||
|
SArray* info = createSortInfo(pSortPhyNode->pSortKeys);
|
||||||
|
return createSortOperatorInfo(op, pResBlock, info, pTaskInfo);
|
||||||
|
} else if (QUERY_NODE_PHYSICAL_PLAN_SESSION_WINDOW == nodeType(pPhyNode)) {
|
||||||
|
size_t size = LIST_LENGTH(pPhyNode->pChildren);
|
||||||
|
assert(size == 1);
|
||||||
|
|
||||||
|
SPhysiNode* pChildNode = (SPhysiNode*)nodesListGetNode(pPhyNode->pChildren, 0);
|
||||||
|
SOperatorInfo* op = doCreateOperatorTreeNode(pChildNode, pTaskInfo, pHandle, queryId, taskId, pTableGroupInfo);
|
||||||
|
|
||||||
|
SSessionWinodwPhysiNode* pSessionNode = (SSessionWinodwPhysiNode*)pPhyNode;
|
||||||
|
|
||||||
|
int32_t num = 0;
|
||||||
|
SExprInfo* pExprInfo = createExprInfo(pSessionNode->window.pFuncs, NULL, &num);
|
||||||
|
SSDataBlock* pResBlock = createOutputBuf_rv1(pPhyNode->pOutputDataBlockDesc);
|
||||||
|
return createSessionAggOperatorInfo(op, pExprInfo, num, pResBlock, pSessionNode->gap, pTaskInfo);
|
||||||
|
} else {
|
||||||
|
ASSERT(0);
|
||||||
|
}/*else if (pPhyNode->info.type == OP_MultiTableAggregate) {
|
||||||
size_t size = taosArrayGetSize(pPhyNode->pChildren);
|
size_t size = taosArrayGetSize(pPhyNode->pChildren);
|
||||||
assert(size == 1);
|
assert(size == 1);
|
||||||
|
|
||||||
|
@ -8725,7 +8721,39 @@ SArray* extractColumnInfo(SNodeList* pNodeList) {
|
||||||
return pList;
|
return pList;
|
||||||
}
|
}
|
||||||
|
|
||||||
SArray* extractColMatchInfo(SNodeList* pNodeList) {
|
SArray* createSortInfo(SNodeList* pNodeList) {
|
||||||
|
size_t numOfCols = LIST_LENGTH(pNodeList);
|
||||||
|
SArray* pList = taosArrayInit(numOfCols, sizeof(SBlockOrderInfo));
|
||||||
|
if (pList == NULL) {
|
||||||
|
terrno = TSDB_CODE_OUT_OF_MEMORY;
|
||||||
|
return pList;
|
||||||
|
}
|
||||||
|
|
||||||
|
for(int32_t i = 0; i < numOfCols; ++i) {
|
||||||
|
STargetNode* pNode = (STargetNode*) nodesListGetNode(pNodeList, i);
|
||||||
|
SOrderByExprNode* pSortKey = (SOrderByExprNode*) pNode->pExpr;
|
||||||
|
SBlockOrderInfo bi = {0};
|
||||||
|
bi.order = (pSortKey->order == ORDER_ASC)? TSDB_ORDER_ASC:TSDB_ORDER_DESC;
|
||||||
|
bi.nullFirst = (pSortKey->nullOrder == NULL_ORDER_FIRST);
|
||||||
|
|
||||||
|
SColumnNode* pColNode = (SColumnNode*)pSortKey->pExpr;
|
||||||
|
bi.slotId = pColNode->slotId;
|
||||||
|
// pColNode->order;
|
||||||
|
// SColumn c = {0};
|
||||||
|
// c.slotId = pColNode->slotId;
|
||||||
|
// c.colId = pColNode->colId;
|
||||||
|
// c.type = pColNode->node.resType.type;
|
||||||
|
// c.bytes = pColNode->node.resType.bytes;
|
||||||
|
// c.precision = pColNode->node.resType.precision;
|
||||||
|
// c.scale = pColNode->node.resType.scale;
|
||||||
|
|
||||||
|
taosArrayPush(pList, &bi);
|
||||||
|
}
|
||||||
|
|
||||||
|
return pList;
|
||||||
|
}
|
||||||
|
|
||||||
|
SArray* extractColMatchInfo(SNodeList* pNodeList, SDataBlockDescNode* pOutputNodeList, int32_t* numOfOutputCols) {
|
||||||
size_t numOfCols = LIST_LENGTH(pNodeList);
|
size_t numOfCols = LIST_LENGTH(pNodeList);
|
||||||
SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchInfo));
|
SArray* pList = taosArrayInit(numOfCols, sizeof(SColMatchInfo));
|
||||||
if (pList == NULL) {
|
if (pList == NULL) {
|
||||||
|
@ -8738,12 +8766,25 @@ SArray* extractColMatchInfo(SNodeList* pNodeList) {
|
||||||
SColumnNode* pColNode = (SColumnNode*) pNode->pExpr;
|
SColumnNode* pColNode = (SColumnNode*) pNode->pExpr;
|
||||||
|
|
||||||
SColMatchInfo c = {0};
|
SColMatchInfo c = {0};
|
||||||
c.colId = pColNode->colId;
|
c.colId = pColNode->colId;
|
||||||
c.targetSlotId = pNode->slotId;
|
c.targetSlotId = pNode->slotId;
|
||||||
|
c.output = true;
|
||||||
taosArrayPush(pList, &c);
|
taosArrayPush(pList, &c);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
*numOfOutputCols = 0;
|
||||||
|
|
||||||
|
int32_t num = LIST_LENGTH(pOutputNodeList->pSlots);
|
||||||
|
for(int32_t i = 0; i < num; ++i) {
|
||||||
|
SSlotDescNode* pNode = (SSlotDescNode*) nodesListGetNode(pOutputNodeList->pSlots, i);
|
||||||
|
SColMatchInfo* info = taosArrayGet(pList, pNode->slotId);
|
||||||
|
// if (pNode->output) {
|
||||||
|
(*numOfOutputCols) += 1;
|
||||||
|
// } else {
|
||||||
|
// info->output = false;
|
||||||
|
// }
|
||||||
|
}
|
||||||
|
|
||||||
return pList;
|
return pList;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -23,20 +23,19 @@
|
||||||
#include "tsort.h"
|
#include "tsort.h"
|
||||||
#include "tutil.h"
|
#include "tutil.h"
|
||||||
|
|
||||||
typedef struct STupleHandle {
|
struct STupleHandle {
|
||||||
SSDataBlock* pBlock;
|
SSDataBlock* pBlock;
|
||||||
int32_t rowIndex;
|
int32_t rowIndex;
|
||||||
} STupleHandle;
|
};
|
||||||
|
|
||||||
typedef struct SSortHandle {
|
struct SSortHandle {
|
||||||
int32_t type;
|
int32_t type;
|
||||||
|
|
||||||
int32_t pageSize;
|
int32_t pageSize;
|
||||||
int32_t numOfPages;
|
int32_t numOfPages;
|
||||||
SDiskbasedBuf *pBuf;
|
SDiskbasedBuf *pBuf;
|
||||||
|
|
||||||
SArray *pOrderInfo;
|
SArray *pSortInfo;
|
||||||
bool nullFirst;
|
|
||||||
SArray *pOrderedSource;
|
SArray *pOrderedSource;
|
||||||
|
|
||||||
_sort_fetch_block_fn_t fetchfp;
|
_sort_fetch_block_fn_t fetchfp;
|
||||||
|
@ -60,7 +59,7 @@ typedef struct SSortHandle {
|
||||||
bool inMemSort;
|
bool inMemSort;
|
||||||
bool needAdjust;
|
bool needAdjust;
|
||||||
STupleHandle tupleHandle;
|
STupleHandle tupleHandle;
|
||||||
} SSortHandle;
|
};
|
||||||
|
|
||||||
static int32_t msortComparFn(const void *pLeft, const void *pRight, void *param);
|
static int32_t msortComparFn(const void *pLeft, const void *pRight, void *param);
|
||||||
|
|
||||||
|
@ -90,18 +89,18 @@ static SSDataBlock* createDataBlock_rv(SSchema* pSchema, int32_t numOfCols) {
|
||||||
* @param type
|
* @param type
|
||||||
* @return
|
* @return
|
||||||
*/
|
*/
|
||||||
SSortHandle* tsortCreateSortHandle(SArray* pOrderInfo, bool nullFirst, int32_t type, int32_t pageSize, int32_t numOfPages, SSchema* pSchema, int32_t numOfCols, const char* idstr) {
|
SSortHandle* tsortCreateSortHandle(SArray* pSortInfo, int32_t type, int32_t pageSize, int32_t numOfPages, SSDataBlock* pBlock, const char* idstr) {
|
||||||
SSortHandle* pSortHandle = taosMemoryCalloc(1, sizeof(SSortHandle));
|
SSortHandle* pSortHandle = taosMemoryCalloc(1, sizeof(SSortHandle));
|
||||||
|
|
||||||
pSortHandle->type = type;
|
pSortHandle->type = type;
|
||||||
pSortHandle->pageSize = pageSize;
|
pSortHandle->pageSize = pageSize;
|
||||||
pSortHandle->numOfPages = numOfPages;
|
pSortHandle->numOfPages = numOfPages;
|
||||||
pSortHandle->pOrderedSource = taosArrayInit(4, POINTER_BYTES);
|
pSortHandle->pSortInfo = pSortInfo;
|
||||||
pSortHandle->pOrderInfo = pOrderInfo;
|
pSortHandle->pDataBlock = createOneDataBlock(pBlock);
|
||||||
pSortHandle->nullFirst = nullFirst;
|
|
||||||
pSortHandle->cmpParam.orderInfo = pOrderInfo;
|
pSortHandle->pOrderedSource = taosArrayInit(4, POINTER_BYTES);
|
||||||
|
pSortHandle->cmpParam.orderInfo = pSortInfo;
|
||||||
|
|
||||||
pSortHandle->pDataBlock = createDataBlock_rv(pSchema, numOfCols);
|
|
||||||
tsortSetComparFp(pSortHandle, msortComparFn);
|
tsortSetComparFp(pSortHandle, msortComparFn);
|
||||||
|
|
||||||
if (idstr != NULL) {
|
if (idstr != NULL) {
|
||||||
|
@ -364,14 +363,14 @@ int32_t msortComparFn(const void *pLeft, const void *pRight, void *param) {
|
||||||
for(int32_t i = 0; i < pInfo->size; ++i) {
|
for(int32_t i = 0; i < pInfo->size; ++i) {
|
||||||
SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, i);
|
SBlockOrderInfo* pOrder = TARRAY_GET_ELEM(pInfo, i);
|
||||||
|
|
||||||
SColumnInfoData* pLeftColInfoData = TARRAY_GET_ELEM(pLeftBlock->pDataBlock, pOrder->colIndex);
|
SColumnInfoData* pLeftColInfoData = TARRAY_GET_ELEM(pLeftBlock->pDataBlock, pOrder->slotId);
|
||||||
|
|
||||||
bool leftNull = false;
|
bool leftNull = false;
|
||||||
if (pLeftColInfoData->hasNull) {
|
if (pLeftColInfoData->hasNull) {
|
||||||
leftNull = colDataIsNull(pLeftColInfoData, pLeftBlock->info.rows, pLeftSource->src.rowIndex, pLeftBlock->pBlockAgg);
|
leftNull = colDataIsNull(pLeftColInfoData, pLeftBlock->info.rows, pLeftSource->src.rowIndex, pLeftBlock->pBlockAgg);
|
||||||
}
|
}
|
||||||
|
|
||||||
SColumnInfoData* pRightColInfoData = TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->colIndex);
|
SColumnInfoData* pRightColInfoData = TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->slotId);
|
||||||
bool rightNull = false;
|
bool rightNull = false;
|
||||||
if (pRightColInfoData->hasNull) {
|
if (pRightColInfoData->hasNull) {
|
||||||
rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex, pRightBlock->pBlockAgg);
|
rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex, pRightBlock->pBlockAgg);
|
||||||
|
@ -415,6 +414,9 @@ int32_t msortComparFn(const void *pLeft, const void *pRight, void *param) {
|
||||||
|
|
||||||
static int32_t doInternalMergeSort(SSortHandle* pHandle) {
|
static int32_t doInternalMergeSort(SSortHandle* pHandle) {
|
||||||
size_t numOfSources = taosArrayGetSize(pHandle->pOrderedSource);
|
size_t numOfSources = taosArrayGetSize(pHandle->pOrderedSource);
|
||||||
|
if (numOfSources == 0) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
// Calculate the I/O counts to complete the data sort.
|
// Calculate the I/O counts to complete the data sort.
|
||||||
double sortPass = floorl(log2(numOfSources) / log2(pHandle->numOfPages));
|
double sortPass = floorl(log2(numOfSources) / log2(pHandle->numOfPages));
|
||||||
|
@ -542,7 +544,7 @@ static int32_t createInitialSortedMultiSources(SSortHandle* pHandle) {
|
||||||
if (size > sortBufSize) {
|
if (size > sortBufSize) {
|
||||||
// Perform the in-memory sort and then flush data in the buffer into disk.
|
// Perform the in-memory sort and then flush data in the buffer into disk.
|
||||||
int64_t p = taosGetTimestampUs();
|
int64_t p = taosGetTimestampUs();
|
||||||
blockDataSort(pHandle->pDataBlock, pHandle->pOrderInfo, pHandle->nullFirst);
|
blockDataSort(pHandle->pDataBlock, pHandle->pSortInfo);
|
||||||
|
|
||||||
int64_t el = taosGetTimestampUs() - p;
|
int64_t el = taosGetTimestampUs() - p;
|
||||||
pHandle->sortElapsed += el;
|
pHandle->sortElapsed += el;
|
||||||
|
@ -555,7 +557,7 @@ static int32_t createInitialSortedMultiSources(SSortHandle* pHandle) {
|
||||||
size_t size = blockDataGetSize(pHandle->pDataBlock);
|
size_t size = blockDataGetSize(pHandle->pDataBlock);
|
||||||
|
|
||||||
// Perform the in-memory sort and then flush data in the buffer into disk.
|
// Perform the in-memory sort and then flush data in the buffer into disk.
|
||||||
blockDataSort(pHandle->pDataBlock, pHandle->pOrderInfo, pHandle->nullFirst);
|
blockDataSort(pHandle->pDataBlock, pHandle->pSortInfo);
|
||||||
|
|
||||||
// All sorted data can fit in memory, external memory sort is not needed. Return to directly
|
// All sorted data can fit in memory, external memory sort is not needed. Return to directly
|
||||||
if (size <= sortBufSize) {
|
if (size <= sortBufSize) {
|
||||||
|
@ -603,6 +605,10 @@ int32_t tsortOpen(SSortHandle* pHandle) {
|
||||||
ASSERT(numOfSources <= getNumOfInMemBufPages(pHandle->pBuf));
|
ASSERT(numOfSources <= getNumOfInMemBufPages(pHandle->pBuf));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (numOfSources == 0) {
|
||||||
|
return 0;
|
||||||
|
}
|
||||||
|
|
||||||
code = sortComparInit(&pHandle->cmpParam, pHandle->pOrderedSource, 0, numOfSources - 1, pHandle);
|
code = sortComparInit(&pHandle->cmpParam, pHandle->pOrderedSource, 0, numOfSources - 1, pHandle);
|
||||||
if (code != TSDB_CODE_SUCCESS) {
|
if (code != TSDB_CODE_SUCCESS) {
|
||||||
return code;
|
return code;
|
||||||
|
|
|
@ -966,7 +966,7 @@ TEST(testCase, inMem_sort_Test) {
|
||||||
exp1->base.resSchema = createSchema(TSDB_DATA_TYPE_BINARY, 40, 2, "res1");
|
exp1->base.resSchema = createSchema(TSDB_DATA_TYPE_BINARY, 40, 2, "res1");
|
||||||
taosArrayPush(pExprInfo, &exp1);
|
taosArrayPush(pExprInfo, &exp1);
|
||||||
|
|
||||||
SOperatorInfo* pOperator = createOrderOperatorInfo(createDummyOperator(10000, 5, 1000, data_asc, 1), pExprInfo, pOrderVal, NULL);
|
SOperatorInfo* pOperator = createSortOperatorInfo(createDummyOperator(10000, 5, 1000, data_asc, 1), pExprInfo, pOrderVal, NULL);
|
||||||
|
|
||||||
bool newgroup = false;
|
bool newgroup = false;
|
||||||
SSDataBlock* pRes = pOperator->getNextFn(pOperator, &newgroup);
|
SSDataBlock* pRes = pOperator->getNextFn(pOperator, &newgroup);
|
||||||
|
@ -1035,7 +1035,7 @@ TEST(testCase, external_sort_Test) {
|
||||||
exp1->base.resSchema = createSchema(TSDB_DATA_TYPE_BINARY, 40, 2, "res1");
|
exp1->base.resSchema = createSchema(TSDB_DATA_TYPE_BINARY, 40, 2, "res1");
|
||||||
// taosArrayPush(pExprInfo, &exp1);
|
// taosArrayPush(pExprInfo, &exp1);
|
||||||
|
|
||||||
SOperatorInfo* pOperator = createOrderOperatorInfo(createDummyOperator(10000, 1500, 1000, data_desc, 1), pExprInfo, pOrderVal, NULL);
|
SOperatorInfo* pOperator = createSortOperatorInfo(createDummyOperator(10000, 1500, 1000, data_desc, 1), pExprInfo, pOrderVal, NULL);
|
||||||
|
|
||||||
bool newgroup = false;
|
bool newgroup = false;
|
||||||
SSDataBlock* pRes = NULL;
|
SSDataBlock* pRes = NULL;
|
||||||
|
|
|
@ -98,14 +98,14 @@ int32_t docomp(const void* p1, const void* p2, void* param) {
|
||||||
for(int32_t i = 0; i < pInfo->size; ++i) {
|
for(int32_t i = 0; i < pInfo->size; ++i) {
|
||||||
SBlockOrderInfo* pOrder = (SBlockOrderInfo*)TARRAY_GET_ELEM(pInfo, i);
|
SBlockOrderInfo* pOrder = (SBlockOrderInfo*)TARRAY_GET_ELEM(pInfo, i);
|
||||||
|
|
||||||
SColumnInfoData* pLeftColInfoData = (SColumnInfoData*)TARRAY_GET_ELEM(pLeftBlock->pDataBlock, pOrder->colIndex);
|
SColumnInfoData* pLeftColInfoData = (SColumnInfoData*)TARRAY_GET_ELEM(pLeftBlock->pDataBlock, pOrder->slotId);
|
||||||
|
|
||||||
bool leftNull = false;
|
bool leftNull = false;
|
||||||
if (pLeftColInfoData->hasNull) {
|
if (pLeftColInfoData->hasNull) {
|
||||||
leftNull = colDataIsNull(pLeftColInfoData, pLeftBlock->info.rows, pLeftSource->src.rowIndex, pLeftBlock->pBlockAgg);
|
leftNull = colDataIsNull(pLeftColInfoData, pLeftBlock->info.rows, pLeftSource->src.rowIndex, pLeftBlock->pBlockAgg);
|
||||||
}
|
}
|
||||||
|
|
||||||
SColumnInfoData* pRightColInfoData = (SColumnInfoData*) TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->colIndex);
|
SColumnInfoData* pRightColInfoData = (SColumnInfoData*) TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->slotId);
|
||||||
bool rightNull = false;
|
bool rightNull = false;
|
||||||
if (pRightColInfoData->hasNull) {
|
if (pRightColInfoData->hasNull) {
|
||||||
rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex, pRightBlock->pBlockAgg);
|
rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex, pRightBlock->pBlockAgg);
|
||||||
|
|
Loading…
Reference in New Issue