xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #16663 from taosdata/3.0 

release: build 3.0.1.0
This commit is contained in:
Shengliang Guan 2022-09-05 17:04:41 +08:00 committed by GitHub
parent 8ec843626c 07c19b736c
commit 3028934f2b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 1518 additions and 1482 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
source/dnode/vnode/src/inc/sma.h
View File

@ -148,6 +148,7 @@ struct SRSmaInfoItem {
}; };
struct SRSmaInfo { struct SRSmaInfo {
SSma *pSma;
STSchema *pTSchema; STSchema *pTSchema;
int64_t suid; int64_t suid;
int64_t lastRecv; // ms int64_t lastRecv; // ms

94
source/dnode/vnode/src/inc/tsdb.h
View File

@ -65,12 +65,14 @@ typedef struct SBlockInfo SBlockInfo;
typedef struct SSmaInfo SSmaInfo; typedef struct SSmaInfo SSmaInfo;
typedef struct SBlockCol SBlockCol; typedef struct SBlockCol SBlockCol;
typedef struct SVersionRange SVersionRange; typedef struct SVersionRange SVersionRange;
typedef struct SLDataIter SLDataIter;
#define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F) #define TSDB_FILE_DLMT ((uint32_t)0xF00AFA0F)
#define TSDB_MAX_SUBBLOCKS 8 #define TSDB_MAX_SUBBLOCKS 8
#define TSDB_MAX_LAST_FILE 16 #define TSDB_MAX_SST_FILE 16
#define TSDB_DEFAULT_LAST_FILE 8 #define TSDB_DEFAULT_SST_FILE 8
#define TSDB_FHDR_SIZE 512 #define TSDB_FHDR_SIZE 512
#define TSDB_DEFAULT_PAGE_SIZE 4096
#define HAS_NONE ((int8_t)0x1) #define HAS_NONE ((int8_t)0x1)
#define HAS_NULL ((int8_t)0x2) #define HAS_NULL ((int8_t)0x2)
@ -82,6 +84,14 @@ typedef struct SVersionRange SVersionRange;
#define TSDBKEY_MIN ((TSDBKEY){.ts = TSKEY_MIN, .version = VERSION_MIN}) #define TSDBKEY_MIN ((TSDBKEY){.ts = TSKEY_MIN, .version = VERSION_MIN})
#define TSDBKEY_MAX ((TSDBKEY){.ts = TSKEY_MAX, .version = VERSION_MAX}) #define TSDBKEY_MAX ((TSDBKEY){.ts = TSKEY_MAX, .version = VERSION_MAX})
#define PAGE_CONTENT_SIZE(PAGE) ((PAGE) - sizeof(TSCKSUM))
#define LOGIC_TO_FILE_OFFSET(LOFFSET, PAGE) \
((LOFFSET) / PAGE_CONTENT_SIZE(PAGE) * (PAGE) + (LOFFSET) % PAGE_CONTENT_SIZE(PAGE))
#define FILE_TO_LOGIC_OFFSET(OFFSET, PAGE) ((OFFSET) / (PAGE)*PAGE_CONTENT_SIZE(PAGE) + (OFFSET) % (PAGE))
#define PAGE_OFFSET(PGNO, PAGE) (((PGNO)-1) * (PAGE))
#define OFFSET_PGNO(OFFSET, PAGE) ((OFFSET) / (PAGE) + 1)
#define LOGIC_TO_FILE_SIZE(LSIZE, PAGE) OFFSET_PGNO(LOGIC_TO_FILE_OFFSET(LSIZE, PAGE), PAGE) * (PAGE)
// tsdbUtil.c ============================================================================================== // tsdbUtil.c ==============================================================================================
// TSDBROW // TSDBROW
#define TSDBROW_TS(ROW) (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow]) #define TSDBROW_TS(ROW) (((ROW)->type == 0) ? (ROW)->pTSRow->ts : (ROW)->pBlockData->aTSKEY[(ROW)->iRow])
@ -195,7 +205,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol
uint8_t **ppBuf); uint8_t **ppBuf);
int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, int32_t nVal, SColData *pColData, int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, int32_t nVal, SColData *pColData,
uint8_t **ppBuf); uint8_t **ppBuf);
int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t size, int8_t toCheck);
// tsdbMemTable ============================================================================================== // tsdbMemTable ==============================================================================================
// SMemTable // SMemTable
int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable); int32_t tsdbMemTableCreate(STsdb *pTsdb, SMemTable **ppMemTable);
@ -563,7 +572,7 @@ struct SDFileSet {
SDataFile *pDataF; SDataFile *pDataF;
SSmaFile *pSmaF; SSmaFile *pSmaF;
uint8_t nSstF; uint8_t nSstF;
SSstFile *aSstF[TSDB_MAX_LAST_FILE]; SSstFile *aSstF[TSDB_MAX_SST_FILE];
}; };
struct SRowIter { struct SRowIter {
@ -578,46 +587,54 @@ struct SRowMerger {
SArray *pArray; // SArray<SColVal> SArray *pArray; // SArray<SColVal>
}; };
typedef struct {
char *path;
int32_t szPage;
int32_t flag;
TdFilePtr pFD;
int64_t pgno;
uint8_t *pBuf;
int64_t szFile;
} STsdbFD;
struct SDelFWriter { struct SDelFWriter {
STsdb *pTsdb; STsdb *pTsdb;
SDelFile fDel; SDelFile fDel;
TdFilePtr pWriteH; STsdbFD *pWriteH;
uint8_t *aBuf[1]; uint8_t *aBuf[1];
}; };
struct SDataFWriter {
STsdb *pTsdb;
SDFileSet wSet;
TdFilePtr pHeadFD;
TdFilePtr pDataFD;
TdFilePtr pSmaFD;
TdFilePtr pLastFD;
SHeadFile fHead;
SDataFile fData;
SSmaFile fSma;
SSstFile fSst[TSDB_MAX_LAST_FILE];
uint8_t *aBuf[4];
};
struct STsdbReadSnap { struct STsdbReadSnap {
SMemTable *pMem; SMemTable *pMem;
SMemTable *pIMem; SMemTable *pIMem;
STsdbFS fs; STsdbFS fs;
}; };
struct SDataFWriter {
STsdb *pTsdb;
SDFileSet wSet;
STsdbFD *pHeadFD;
STsdbFD *pDataFD;
STsdbFD *pSmaFD;
STsdbFD *pSstFD;
SHeadFile fHead;
SDataFile fData;
SSmaFile fSma;
SSstFile fSst[TSDB_MAX_SST_FILE];
uint8_t *aBuf[4];
};
struct SDataFReader { struct SDataFReader {
STsdb *pTsdb; STsdb *pTsdb;
SDFileSet *pSet; SDFileSet *pSet;
TdFilePtr pHeadFD; STsdbFD *pHeadFD;
TdFilePtr pDataFD; STsdbFD *pDataFD;
TdFilePtr pSmaFD; STsdbFD *pSmaFD;
TdFilePtr aLastFD[TSDB_MAX_LAST_FILE]; STsdbFD *aSstFD[TSDB_MAX_SST_FILE];
uint8_t *aBuf[3];
uint8_t *aBuf[3];
}; };
typedef struct { typedef struct {
@ -627,15 +644,16 @@ typedef struct {
} SRowInfo; } SRowInfo;
typedef struct SMergeTree { typedef struct SMergeTree {
int8_t backward; int8_t backward;
SRBTree rbt; SRBTree rbt;
SArray *pIterList; SArray *pIterList;
struct SLDataIter *pIter; SLDataIter *pIter;
} SMergeTree; } SMergeTree;
int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader* pFReader, uint64_t uid, STimeWindow* pTimeWindow, SVersionRange* pVerRange); int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFReader, uint64_t uid,
void tMergeTreeAddIter(SMergeTree *pMTree, struct SLDataIter *pIter); STimeWindow *pTimeWindow, SVersionRange *pVerRange);
bool tMergeTreeNext(SMergeTree *pMTree); void tMergeTreeAddIter(SMergeTree *pMTree, SLDataIter *pIter);
bool tMergeTreeNext(SMergeTree *pMTree);
TSDBROW tMergeTreeGetRow(SMergeTree *pMTree); TSDBROW tMergeTreeGetRow(SMergeTree *pMTree);
void tMergeTreeClose(SMergeTree *pMTree); void tMergeTreeClose(SMergeTree *pMTree);

11
source/dnode/vnode/src/sma/smaCommit.c
View File

@ -188,13 +188,15 @@ static int32_t tdUpdateQTaskInfoFiles(SSma *pSma, SRSmaStat *pStat) {
for (int32_t i = 0; i < taosArrayGetSize(pFS->aQTaskInf);) { for (int32_t i = 0; i < taosArrayGetSize(pFS->aQTaskInf);) {
SQTaskFile *pTaskF = taosArrayGet(pFS->aQTaskInf, i); SQTaskFile *pTaskF = taosArrayGet(pFS->aQTaskInf, i);
if (atomic_sub_fetch_32(&pTaskF->nRef, 1) <= 0) { int32_t oldVal = atomic_fetch_sub_32(&pTaskF->nRef, 1);
if ((oldVal <= 1) && (pTaskF->version < committed)) {
tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->version, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName); tdRSmaQTaskInfoGetFullName(TD_VID(pVnode), pTaskF->version, tfsGetPrimaryPath(pVnode->pTfs), qTaskInfoFullName);
if (taosRemoveFile(qTaskInfoFullName) < 0) { if (taosRemoveFile(qTaskInfoFullName) < 0) {
smaWarn("vgId:%d, cleanup qinf, failed to remove %s since %s", TD_VID(pVnode), qTaskInfoFullName, smaWarn("vgId:%d, cleanup qinf, committed %" PRIi64 ", failed to remove %s since %s", TD_VID(pVnode), committed,
tstrerror(TAOS_SYSTEM_ERROR(errno))); qTaskInfoFullName, tstrerror(TAOS_SYSTEM_ERROR(errno)));
} else { } else {
smaDebug("vgId:%d, cleanup qinf, success to remove %s", TD_VID(pVnode), qTaskInfoFullName); smaDebug("vgId:%d, cleanup qinf, committed %" PRIi64 ", success to remove %s", TD_VID(pVnode), committed,
qTaskInfoFullName);
} }
taosArrayRemove(pFS->aQTaskInf, i); taosArrayRemove(pFS->aQTaskInf, i);
continue; continue;
@ -364,7 +366,6 @@ static int32_t tdProcessRSmaAsyncPostCommitImpl(SSma *pSma) {
} }
SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv); SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pEnv);
SRSmaInfoItem *pItem = NULL;
// step 1: merge qTaskInfo and iQTaskInfo // step 1: merge qTaskInfo and iQTaskInfo
// lock // lock

5
source/dnode/vnode/src/sma/smaEnv.c
View File

@ -62,7 +62,7 @@ int32_t smaInit() {
} }
int32_t type = (8 == POINTER_BYTES) ? TSDB_DATA_TYPE_UBIGINT : TSDB_DATA_TYPE_UINT; int32_t type = (8 == POINTER_BYTES) ? TSDB_DATA_TYPE_UBIGINT : TSDB_DATA_TYPE_UINT;
smaMgmt.refHash = taosHashInit(1, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK); smaMgmt.refHash = taosHashInit(64, taosGetDefaultHashFunction(type), true, HASH_ENTRY_LOCK);
if (!smaMgmt.refHash) { if (!smaMgmt.refHash) {
taosCloseRef(smaMgmt.rsetId); taosCloseRef(smaMgmt.rsetId);
atomic_store_8(&smaMgmt.inited, 0); atomic_store_8(&smaMgmt.inited, 0);
@ -107,6 +107,7 @@ void smaCleanUp() {
if (old == 1) { if (old == 1) {
taosCloseRef(smaMgmt.rsetId); taosCloseRef(smaMgmt.rsetId);
taosHashCleanup(smaMgmt.refHash); taosHashCleanup(smaMgmt.refHash);
smaMgmt.refHash = NULL;
taosTmrCleanUp(smaMgmt.tmrHandle); taosTmrCleanUp(smaMgmt.tmrHandle);
smaInfo("sma mgmt env is cleaned up, rsetId:%d, tmrHandle:%p", smaMgmt.rsetId, smaMgmt.tmrHandle); smaInfo("sma mgmt env is cleaned up, rsetId:%d, tmrHandle:%p", smaMgmt.rsetId, smaMgmt.tmrHandle);
atomic_store_8(&smaMgmt.inited, 0); atomic_store_8(&smaMgmt.inited, 0);
@ -220,7 +221,7 @@ static void tRSmaInfoHashFreeNode(void *data) {
if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 1)) && pItem->level) { if ((pItem = RSMA_INFO_ITEM((SRSmaInfo *)pRSmaInfo, 1)) && pItem->level) {
taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES); taosHashRemove(smaMgmt.refHash, &pItem, POINTER_BYTES);
} }
tdFreeRSmaInfo(NULL, pRSmaInfo, true); tdFreeRSmaInfo(pRSmaInfo->pSma, pRSmaInfo, true);
} }
} }

25
source/dnode/vnode/src/sma/smaRollup.c
View File

@ -121,27 +121,27 @@ static void tdRSmaQTaskInfoFree(qTaskInfo_t *taskHandle, int32_t vgId, int32_t l
*/ */
void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) { void *tdFreeRSmaInfo(SSma *pSma, SRSmaInfo *pInfo, bool isDeepFree) {
if (pInfo) { if (pInfo) {
int32_t vid = pSma ? SMA_VID(pSma) : -1;
for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) { for (int32_t i = 0; i < TSDB_RETENTION_L2; ++i) {
SRSmaInfoItem *pItem = &pInfo->items[i]; SRSmaInfoItem *pItem = &pInfo->items[i];
if (isDeepFree && pItem->tmrId) { if (isDeepFree && pItem->tmrId) {
smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", vid, pItem->tmrId, pInfo->suid, i + 1); smaDebug("vgId:%d, stop fetch timer %p for table %" PRIi64 " level %d", SMA_VID(pSma), pItem->tmrId,
pInfo->suid, i + 1);
taosTmrStopA(&pItem->tmrId); taosTmrStopA(&pItem->tmrId);
} }
if (isDeepFree && pInfo->taskInfo[i]) { if (isDeepFree && pInfo->taskInfo[i]) {
tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], vid, i + 1); tdRSmaQTaskInfoFree(&pInfo->taskInfo[i], SMA_VID(pSma), i + 1);
} else { } else {
smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo", vid, smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty taskInfo", SMA_VID(pSma),
pInfo->suid, i + 1); pInfo->suid, i + 1);
} }
if (pInfo->iTaskInfo[i]) { if (pInfo->iTaskInfo[i]) {
tdRSmaQTaskInfoFree(&pInfo->iTaskInfo[i], vid, i + 1); tdRSmaQTaskInfoFree(&pInfo->iTaskInfo[i], SMA_VID(pSma), i + 1);
} else { } else {
smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty iTaskInfo", vid, smaDebug("vgId:%d, table %" PRIi64 " no need to destroy rsma info level %d since empty iTaskInfo",
pInfo->suid, i + 1); SMA_VID(pSma), pInfo->suid, i + 1);
} }
} }
if (isDeepFree) { if (isDeepFree) {
@ -377,7 +377,10 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION; terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION;
goto _err; goto _err;
} }
pRSmaInfo->pSma = pSma;
pRSmaInfo->pTSchema = pTSchema; pRSmaInfo->pTSchema = pTSchema;
pRSmaInfo->suid = suid;
T_REF_INIT_VAL(pRSmaInfo, 1);
if (!(pRSmaInfo->queue = taosOpenQueue())) { if (!(pRSmaInfo->queue = taosOpenQueue())) {
goto _err; goto _err;
} }
@ -391,8 +394,6 @@ int32_t tdRSmaProcessCreateImpl(SSma *pSma, SRSmaParam *param, int64_t suid, con
if (!(pRSmaInfo->iQall = taosAllocateQall())) { if (!(pRSmaInfo->iQall = taosAllocateQall())) {
goto _err; goto _err;
} }
pRSmaInfo->suid = suid;
T_REF_INIT_VAL(pRSmaInfo, 1);
if (tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0) { if (tdSetRSmaInfoItemParams(pSma, param, pStat, pRSmaInfo, 0) < 0) {
goto _err; goto _err;
@ -1586,8 +1587,8 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) { if (!(pStat = (SRSmaStat *)tdAcquireSmaRef(smaMgmt.rsetId, pRSmaRef->refId))) {
smaDebug("rsma fetch task not start since rsma stat already destroyed, rsetId:%" PRIi64 " refId:%d)", smaDebug("rsma fetch task not start since rsma stat already destroyed, rsetId:%" PRIi64 " refId:%d)",
smaMgmt.rsetId, pRSmaRef->refId); // pRSmaRef freed in taosHashRemove smaMgmt.rsetId, pRSmaRef->refId); // pRSmaRef freed in taosHashRemove
taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES); taosHashRemove(smaMgmt.refHash, &param, POINTER_BYTES);
return; return;
} }
@ -1636,7 +1637,7 @@ static void tdRSmaFetchTrigger(void *param, void *tmrId) {
switch (fetchTriggerStat) { switch (fetchTriggerStat) {
case TASK_TRIGGER_STAT_ACTIVE: { case TASK_TRIGGER_STAT_ACTIVE: {
smaDebug("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active", smaDebug("vgId:%d, rsma fetch task planned for level:%" PRIi8 " suid:%" PRIi64 " since stat is active",
SMA_VID(pSma), pItem->level, pRSmaInfo->suid); SMA_VID(pSma), pItem->level, pRSmaInfo->suid);
// async process // async process
pItem->fetchLevel = pItem->level; pItem->fetchLevel = pItem->level;
#if 0 #if 0

15
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -71,7 +71,7 @@ typedef struct {
SDataIter *pIter; SDataIter *pIter;
SRBTree rbt; SRBTree rbt;
SDataIter dataIter; SDataIter dataIter;
SDataIter aDataIter[TSDB_MAX_LAST_FILE]; SDataIter aDataIter[TSDB_MAX_SST_FILE];
int8_t toLastOnly; int8_t toLastOnly;
}; };
struct { struct {
@ -92,9 +92,6 @@ typedef struct {
SArray *aDelData; // SArray<SDelData> SArray *aDelData; // SArray<SDelData>
} SCommitter; } SCommitter;
extern int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *aSstBlk,
SBlockData *pBlockData); // todo
static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter); static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter);
static int32_t tsdbCommitData(SCommitter *pCommitter); static int32_t tsdbCommitData(SCommitter *pCommitter);
static int32_t tsdbCommitDel(SCommitter *pCommitter); static int32_t tsdbCommitDel(SCommitter *pCommitter);
@ -445,7 +442,7 @@ static int32_t tsdbOpenCommitIter(SCommitter *pCommitter) {
pIter->iSstBlk = 0; pIter->iSstBlk = 0;
SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, 0); SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, 0);
code = tsdbReadSstBlockEx(pCommitter->dReader.pReader, iSst, pSstBlk, &pIter->bData); code = tsdbReadSstBlock(pCommitter->dReader.pReader, iSst, pSstBlk, &pIter->bData);
if (code) goto _err; if (code) goto _err;
pIter->iRow = 0; pIter->iRow = 0;
@ -760,7 +757,7 @@ static int32_t tsdbStartCommit(STsdb *pTsdb, SCommitter *pCommitter) {
pCommitter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows; pCommitter->minRow = pTsdb->pVnode->config.tsdbCfg.minRows;
pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows; pCommitter->maxRow = pTsdb->pVnode->config.tsdbCfg.maxRows;
pCommitter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression; pCommitter->cmprAlg = pTsdb->pVnode->config.tsdbCfg.compression;
pCommitter->maxLast = TSDB_DEFAULT_LAST_FILE; // TODO: make it as a config pCommitter->maxLast = TSDB_DEFAULT_SST_FILE; // TODO: make it as a config
pCommitter->aTbDataP = tsdbMemTableGetTbDataArray(pTsdb->imem); pCommitter->aTbDataP = tsdbMemTableGetTbDataArray(pTsdb->imem);
if (pCommitter->aTbDataP == NULL) { if (pCommitter->aTbDataP == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
@ -790,7 +787,7 @@ static int32_t tsdbCommitDataStart(SCommitter *pCommitter) {
if (code) goto _exit; if (code) goto _exit;
// merger // merger
for (int32_t iSst = 0; iSst < TSDB_MAX_LAST_FILE; iSst++) { for (int32_t iSst = 0; iSst < TSDB_MAX_SST_FILE; iSst++) {
SDataIter *pIter = &pCommitter->aDataIter[iSst]; SDataIter *pIter = &pCommitter->aDataIter[iSst];
pIter->aSstBlk = taosArrayInit(0, sizeof(SSstBlk)); pIter->aSstBlk = taosArrayInit(0, sizeof(SSstBlk));
if (pIter->aSstBlk == NULL) { if (pIter->aSstBlk == NULL) {
@ -832,7 +829,7 @@ static void tsdbCommitDataEnd(SCommitter *pCommitter) {
tBlockDataDestroy(&pCommitter->dReader.bData, 1); tBlockDataDestroy(&pCommitter->dReader.bData, 1);
// merger // merger
for (int32_t iSst = 0; iSst < TSDB_MAX_LAST_FILE; iSst++) { for (int32_t iSst = 0; iSst < TSDB_MAX_SST_FILE; iSst++) {
SDataIter *pIter = &pCommitter->aDataIter[iSst]; SDataIter *pIter = &pCommitter->aDataIter[iSst];
taosArrayDestroy(pIter->aSstBlk); taosArrayDestroy(pIter->aSstBlk);
tBlockDataDestroy(&pIter->bData, 1); tBlockDataDestroy(&pIter->bData, 1);
@ -1059,7 +1056,7 @@ static int32_t tsdbNextCommitRow(SCommitter *pCommitter) {
if (pIter->iSstBlk < taosArrayGetSize(pIter->aSstBlk)) { if (pIter->iSstBlk < taosArrayGetSize(pIter->aSstBlk)) {
SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, pIter->iSstBlk); SSstBlk *pSstBlk = (SSstBlk *)taosArrayGet(pIter->aSstBlk, pIter->iSstBlk);
code = tsdbReadSstBlockEx(pCommitter->dReader.pReader, pIter->iSst, pSstBlk, &pIter->bData); code = tsdbReadSstBlock(pCommitter->dReader.pReader, pIter->iSst, pSstBlk, &pIter->bData);
if (code) goto _exit; if (code) goto _exit;
pIter->iRow = 0; pIter->iRow = 0;

21
source/dnode/vnode/src/tsdb/tsdbFS.c
View File

@ -21,6 +21,9 @@ static int32_t tsdbEncodeFS(uint8_t *p, STsdbFS *pFS) {
int8_t hasDel = pFS->pDelFile ? 1 : 0; int8_t hasDel = pFS->pDelFile ? 1 : 0;
uint32_t nSet = taosArrayGetSize(pFS->aDFileSet); uint32_t nSet = taosArrayGetSize(pFS->aDFileSet);
// version
n += tPutI8(p ? p + n : p, 0);
// SDelFile // SDelFile
n += tPutI8(p ? p + n : p, hasDel); n += tPutI8(p ? p + n : p, hasDel);
if (hasDel) { if (hasDel) {
@ -292,7 +295,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size != pSet->pHeadF->size) { if (size != LOGIC_TO_FILE_SIZE(pSet->pHeadF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} }
@ -303,10 +306,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size < pSet->pDataF->size) { if (size < LOGIC_TO_FILE_SIZE(pSet->pDataF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} else if (size > pSet->pDataF->size) { } else if (size > LOGIC_TO_FILE_SIZE(pSet->pDataF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = tsdbDFileRollback(pTsdb, pSet, TSDB_DATA_FILE); code = tsdbDFileRollback(pTsdb, pSet, TSDB_DATA_FILE);
if (code) goto _err; if (code) goto _err;
} }
@ -317,10 +320,10 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size < pSet->pSmaF->size) { if (size < LOGIC_TO_FILE_SIZE(pSet->pSmaF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} else if (size > pSet->pSmaF->size) { } else if (size > LOGIC_TO_FILE_SIZE(pSet->pSmaF->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE); code = tsdbDFileRollback(pTsdb, pSet, TSDB_SMA_FILE);
if (code) goto _err; if (code) goto _err;
} }
@ -332,7 +335,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }
if (size != pSet->aSstF[iSst]->size) { if (size != LOGIC_TO_FILE_SIZE(pSet->aSstF[iSst]->size, TSDB_DEFAULT_PAGE_SIZE)) {
code = TSDB_CODE_FILE_CORRUPTED; code = TSDB_CODE_FILE_CORRUPTED;
goto _err; goto _err;
} }
@ -364,10 +367,12 @@ static int32_t tsdbRecoverFS(STsdb *pTsdb, uint8_t *pData, int64_t nData) {
int32_t code = 0; int32_t code = 0;
int8_t hasDel; int8_t hasDel;
uint32_t nSet; uint32_t nSet;
int32_t n; int32_t n = 0;
// version
n += tGetI8(pData + n, NULL);
// SDelFile // SDelFile
n = 0;
n += tGetI8(pData + n, &hasDel); n += tGetI8(pData + n, &hasDel);
if (hasDel) { if (hasDel) {
pTsdb->fs.pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile)); pTsdb->fs.pDelFile = (SDelFile *)taosMemoryMalloc(sizeof(SDelFile));

2
source/dnode/vnode/src/tsdb/tsdbFile.c
View File

@ -148,7 +148,7 @@ int32_t tsdbDFileRollback(STsdb *pTsdb, SDFileSet *pSet, EDataFileT ftype) {
} }
// ftruncate // ftruncate
if (taosFtruncateFile(pFD, size) < 0) { if (taosFtruncateFile(pFD, LOGIC_TO_FILE_SIZE(size, TSDB_DEFAULT_PAGE_SIZE)) < 0) {
code = TAOS_SYSTEM_ERROR(errno); code = TAOS_SYSTEM_ERROR(errno);
goto _err; goto _err;
} }

28
source/dnode/vnode/src/tsdb/tsdbMergeTree.c
View File

@ -16,7 +16,7 @@
#include "tsdb.h" #include "tsdb.h"
// SLDataIter ================================================= // SLDataIter =================================================
typedef struct SLDataIter { struct SLDataIter {
SRBTreeNode node; SRBTreeNode node;
SSstBlk *pSstBlk; SSstBlk *pSstBlk;
SDataFReader *pReader; SDataFReader *pReader;
@ -31,13 +31,11 @@ typedef struct SLDataIter {
uint64_t uid; uint64_t uid;
STimeWindow timeWindow; STimeWindow timeWindow;
SVersionRange verRange; SVersionRange verRange;
} SLDataIter; };
static SBlockData* getCurrentBlock(SLDataIter* pIter) { static SBlockData *getCurrentBlock(SLDataIter *pIter) { return &pIter->bData[pIter->loadIndex]; }
return &pIter->bData[pIter->loadIndex];
}
static SBlockData* getNextBlock(SLDataIter* pIter) { static SBlockData *getNextBlock(SLDataIter *pIter) {
pIter->loadIndex ^= 1; pIter->loadIndex ^= 1;
return getCurrentBlock(pIter); return getCurrentBlock(pIter);
} }
@ -116,8 +114,6 @@ void tLDataIterClose(SLDataIter *pIter) {
taosMemoryFree(pIter); taosMemoryFree(pIter);
} }
extern int32_t tsdbReadSstBlockEx(SDataFReader *pReader, int32_t iSst, SSstBlk *pSstBlk, SBlockData *pBlockData);
void tLDataIterNextBlock(SLDataIter *pIter) { void tLDataIterNextBlock(SLDataIter *pIter) {
int32_t step = pIter->backward ? -1 : 1; int32_t step = pIter->backward ? -1 : 1;
pIter->iSstBlk += step; pIter->iSstBlk += step;
@ -150,9 +146,9 @@ void tLDataIterNextBlock(SLDataIter *pIter) {
static void findNextValidRow(SLDataIter *pIter) { static void findNextValidRow(SLDataIter *pIter) {
int32_t step = pIter->backward ? -1 : 1; int32_t step = pIter->backward ? -1 : 1;
bool hasVal = false; bool hasVal = false;
int32_t i = pIter->iRow; int32_t i = pIter->iRow;
SBlockData* pBlockData = getCurrentBlock(pIter); SBlockData *pBlockData = getCurrentBlock(pIter);
for (; i < pBlockData->nRow && i >= 0; i += step) { for (; i < pBlockData->nRow && i >= 0; i += step) {
if (pBlockData->aUid != NULL) { if (pBlockData->aUid != NULL) {
@ -220,12 +216,12 @@ bool tLDataIterNextRow(SLDataIter *pIter) {
return false; return false;
} }
int32_t iBlockL = pIter->iSstBlk; int32_t iBlockL = pIter->iSstBlk;
SBlockData* pBlockData = getCurrentBlock(pIter); SBlockData *pBlockData = getCurrentBlock(pIter);
if (pBlockData->nRow == 0 && pIter->pSstBlk != NULL) { // current block not loaded yet if (pBlockData->nRow == 0 && pIter->pSstBlk != NULL) { // current block not loaded yet
pBlockData = getNextBlock(pIter); pBlockData = getNextBlock(pIter);
code = tsdbReadSstBlockEx(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData); code = tsdbReadSstBlock(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _exit; goto _exit;
} }
@ -249,7 +245,7 @@ bool tLDataIterNextRow(SLDataIter *pIter) {
if (iBlockL != pIter->iSstBlk) { if (iBlockL != pIter->iSstBlk) {
pBlockData = getNextBlock(pIter); pBlockData = getNextBlock(pIter);
code = tsdbReadSstBlockEx(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData); code = tsdbReadSstBlock(pIter->pReader, pIter->iSst, pIter->pSstBlk, pBlockData);
if (code) { if (code) {
goto _exit; goto _exit;
} }
@ -306,7 +302,7 @@ int32_t tMergeTreeOpen(SMergeTree *pMTree, int8_t backward, SDataFReader *pFRead
tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn); tRBTreeCreate(&pMTree->rbt, tLDataIterCmprFn);
int32_t code = TSDB_CODE_OUT_OF_MEMORY; int32_t code = TSDB_CODE_OUT_OF_MEMORY;
struct SLDataIter *pIterList[TSDB_DEFAULT_LAST_FILE] = {0}; struct SLDataIter *pIterList[TSDB_DEFAULT_SST_FILE] = {0};
for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) { // open all last file for (int32_t i = 0; i < pFReader->pSet->nSstF; ++i) { // open all last file
code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange); code = tLDataIterOpen(&pIterList[i], pFReader, i, pMTree->backward, uid, pTimeWindow, pVerRange);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {

1983
source/dnode/vnode/src/tsdb/tsdbReaderWriter.c
View File

File diff suppressed because it is too large Load Diff

74
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -1548,7 +1548,7 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut,
if (code) goto _exit; if (code) goto _exit;
blockCol.offset = aBufN[0]; blockCol.offset = aBufN[0];
aBufN[0] = aBufN[0] + blockCol.szBitmap + blockCol.szOffset + blockCol.szValue + sizeof(TSCKSUM); aBufN[0] = aBufN[0] + blockCol.szBitmap + blockCol.szOffset + blockCol.szValue;
} }
code = tRealloc(&aBuf[1], hdr.szBlkCol + tPutBlockCol(NULL, &blockCol)); code = tRealloc(&aBuf[1], hdr.szBlkCol + tPutBlockCol(NULL, &blockCol));
@ -1556,15 +1556,8 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut,
hdr.szBlkCol += tPutBlockCol(aBuf[1] + hdr.szBlkCol, &blockCol); hdr.szBlkCol += tPutBlockCol(aBuf[1] + hdr.szBlkCol, &blockCol);
} }
aBufN[1] = 0; // SBlockCol
if (hdr.szBlkCol > 0) { aBufN[1] = hdr.szBlkCol;
aBufN[1] = hdr.szBlkCol + sizeof(TSCKSUM);
code = tRealloc(&aBuf[1], aBufN[1]);
if (code) goto _exit;
taosCalcChecksumAppend(0, aBuf[1], aBufN[1]);
}
// uid + version + tskey // uid + version + tskey
aBufN[2] = 0; aBufN[2] = 0;
@ -1585,16 +1578,11 @@ int32_t tCmprBlockData(SBlockData *pBlockData, int8_t cmprAlg, uint8_t **ppOut,
if (code) goto _exit; if (code) goto _exit;
aBufN[2] += hdr.szKey; aBufN[2] += hdr.szKey;
aBufN[2] += sizeof(TSCKSUM);
code = tRealloc(&aBuf[2], aBufN[2]);
if (code) goto _exit;
// hdr // hdr
aBufN[3] = tPutDiskDataHdr(NULL, &hdr); aBufN[3] = tPutDiskDataHdr(NULL, &hdr);
code = tRealloc(&aBuf[3], aBufN[3]); code = tRealloc(&aBuf[3], aBufN[3]);
if (code) goto _exit; if (code) goto _exit;
tPutDiskDataHdr(aBuf[3], &hdr); tPutDiskDataHdr(aBuf[3], &hdr);
taosCalcChecksumAppend(taosCalcChecksum(0, aBuf[3], aBufN[3]), aBuf[2], aBufN[2]);
// aggragate // aggragate
if (ppOut) { if (ppOut) {
@ -1626,10 +1614,6 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin
// SDiskDataHdr // SDiskDataHdr
n += tGetDiskDataHdr(pIn + n, &hdr); n += tGetDiskDataHdr(pIn + n, &hdr);
if (!taosCheckChecksumWhole(pIn, n + hdr.szUid + hdr.szVer + hdr.szKey + sizeof(TSCKSUM))) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
ASSERT(hdr.delimiter == TSDB_FILE_DLMT); ASSERT(hdr.delimiter == TSDB_FILE_DLMT);
pBlockData->suid = hdr.suid; pBlockData->suid = hdr.suid;
@ -1657,7 +1641,7 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin
code = tsdbDecmprData(pIn + n, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY, code = tsdbDecmprData(pIn + n, hdr.szKey, TSDB_DATA_TYPE_TIMESTAMP, hdr.cmprAlg, (uint8_t **)&pBlockData->aTSKEY,
sizeof(TSKEY) * hdr.nRow, &aBuf[0]); sizeof(TSKEY) * hdr.nRow, &aBuf[0]);
if (code) goto _exit; if (code) goto _exit;
n = n + hdr.szKey + sizeof(TSCKSUM); n += hdr.szKey;
// loop to decode each column data // loop to decode each column data
if (hdr.szBlkCol == 0) goto _exit; if (hdr.szBlkCol == 0) goto _exit;
@ -1679,8 +1663,8 @@ int32_t tDecmprBlockData(uint8_t *pIn, int32_t szIn, SBlockData *pBlockData, uin
if (code) goto _exit; if (code) goto _exit;
} }
} else { } else {
code = tsdbDecmprColData(pIn + n + hdr.szBlkCol + sizeof(TSCKSUM) + blockCol.offset, &blockCol, hdr.cmprAlg, code = tsdbDecmprColData(pIn + n + hdr.szBlkCol + blockCol.offset, &blockCol, hdr.cmprAlg, hdr.nRow, pColData,
hdr.nRow, pColData, &aBuf[0]); &aBuf[0]);
if (code) goto _exit; if (code) goto _exit;
} }
} }
@ -2062,12 +2046,6 @@ int32_t tsdbCmprColData(SColData *pColData, int8_t cmprAlg, SBlockCol *pBlockCol
} }
size += pBlockCol->szValue; size += pBlockCol->szValue;
// checksum
size += sizeof(TSCKSUM);
code = tRealloc(ppOut, nOut + size);
if (code) goto _exit;
taosCalcChecksumAppend(0, *ppOut + nOut, size);
_exit: _exit:
return code; return code;
} }
@ -2076,12 +2054,6 @@ int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, in
uint8_t **ppBuf) { uint8_t **ppBuf) {
int32_t code = 0; int32_t code = 0;
int32_t size = pBlockCol->szBitmap + pBlockCol->szOffset + pBlockCol->szValue + sizeof(TSCKSUM);
if (!taosCheckChecksumWhole(pIn, size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
ASSERT(pColData->cid == pBlockCol->cid); ASSERT(pColData->cid == pBlockCol->cid);
ASSERT(pColData->type == pBlockCol->type); ASSERT(pColData->type == pBlockCol->type);
pColData->smaOn = pBlockCol->smaOn; pColData->smaOn = pBlockCol->smaOn;
@ -2153,37 +2125,3 @@ int32_t tsdbDecmprColData(uint8_t *pIn, SBlockCol *pBlockCol, int8_t cmprAlg, in
_exit: _exit:
return code; return code;
} }
int32_t tsdbReadAndCheck(TdFilePtr pFD, int64_t offset, uint8_t **ppOut, int32_t size, int8_t toCheck) {
int32_t code = 0;
// alloc
code = tRealloc(ppOut, size);
if (code) goto _exit;
// seek
int64_t n = taosLSeekFile(pFD, offset, SEEK_SET);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
}
// read
n = taosReadFile(pFD, *ppOut, size);
if (n < 0) {
code = TAOS_SYSTEM_ERROR(errno);
goto _exit;
} else if (n < size) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
// check
if (toCheck && !taosCheckChecksumWhole(*ppOut, size)) {
code = TSDB_CODE_FILE_CORRUPTED;
goto _exit;
}
_exit:
return code;
}

23
source/libs/executor/inc/executorimpl.h
View File

@ -591,6 +591,24 @@ typedef struct SMergeAlignedIntervalAggOperatorInfo {
SNode* pCondition; SNode* pCondition;
} SMergeAlignedIntervalAggOperatorInfo; } SMergeAlignedIntervalAggOperatorInfo;
typedef struct SStreamIntervalOperatorInfo {
// SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
SOptrBasicInfo binfo; // basic info
SAggSupporter aggSup; // aggregate supporter
SExprSupp scalarSupp; // supporter for perform scalar function
SGroupResInfo groupResInfo; // multiple results build supporter
SInterval interval; // interval info
int32_t primaryTsIndex; // primary time stamp slot id from result of downstream operator.
STimeWindowAggSupp twAggSup;
bool invertible;
bool ignoreExpiredData;
SArray* pRecycledPages;
SArray* pDelWins; // SWinRes
int32_t delIndex;
SSDataBlock* pDelRes;
bool isFinal;
} SStreamIntervalOperatorInfo;
typedef struct SStreamFinalIntervalOperatorInfo { typedef struct SStreamFinalIntervalOperatorInfo {
// SOptrBasicInfo should be first, SAggSupporter should be second for stream encode // SOptrBasicInfo should be first, SAggSupporter should be second for stream encode
SOptrBasicInfo binfo; // basic info SOptrBasicInfo binfo; // basic info
@ -909,8 +927,7 @@ void initLimitInfo(const SNode* pLimit, const SNode* pSLimit, SLimitInfo* pLi
void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData, int32_t offset, void doApplyFunctions(SExecTaskInfo* taskInfo, SqlFunctionCtx* pCtx, SColumnInfoData* pTimeWindowData, int32_t offset,
int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput); int32_t forwardStep, int32_t numOfTotal, int32_t numOfOutput);
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, int32_t numOfOutput, SArray* pColList, int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart);
char** pNextStart);
void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs, void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t dataLen, int64_t startTs,
SOperatorInfo* pOperator); SOperatorInfo* pOperator);
@ -1004,6 +1021,8 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
SExecTaskInfo* pTaskInfo); SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SOperatorInfo* createStreamFinalSessionAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo, int32_t numOfChild); SExecTaskInfo* pTaskInfo, int32_t numOfChild);
SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream,
SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo);
SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode, SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo); SExecTaskInfo* pTaskInfo);

8
source/libs/executor/src/executor.c
View File

@ -97,6 +97,8 @@ static int32_t doSetStreamBlock(SOperatorInfo* pOperator, void* input, size_t nu
} }
} }
static FORCE_INLINE void streamInputBlockDataDestory(void* pBlock) { blockDataDestroy((SSDataBlock*)pBlock); }
void tdCleanupStreamInputDataBlock(qTaskInfo_t tinfo) { void tdCleanupStreamInputDataBlock(qTaskInfo_t tinfo) {
SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo; SExecTaskInfo* pTaskInfo = (SExecTaskInfo*)tinfo;
if (!pTaskInfo || !pTaskInfo->pRoot || pTaskInfo->pRoot->numOfDownstream <= 0) { if (!pTaskInfo || !pTaskInfo->pRoot || pTaskInfo->pRoot->numOfDownstream <= 0) {
@ -107,11 +109,7 @@ void tdCleanupStreamInputDataBlock(qTaskInfo_t tinfo) {
if (pOptrInfo->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) { if (pOptrInfo->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
SStreamScanInfo* pInfo = pOptrInfo->info; SStreamScanInfo* pInfo = pOptrInfo->info;
if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) { if (pInfo->blockType == STREAM_INPUT__DATA_BLOCK) {
for (int32_t i = 0; i < taosArrayGetSize(pInfo->pBlockLists); ++i) { taosArrayClearP(pInfo->pBlockLists, streamInputBlockDataDestory);
SSDataBlock* p = *(SSDataBlock**)taosArrayGet(pInfo->pBlockLists, i);
taosArrayDestroy(p->pDataBlock);
taosMemoryFreeClear(p);
}
} else { } else {
ASSERT(0); ASSERT(0);
} }

12
source/libs/executor/src/executorimpl.c
View File

@ -1866,13 +1866,11 @@ void updateLoadRemoteInfo(SLoadRemoteDataInfo* pInfo, int32_t numOfRows, int32_t
pOperator->resultInfo.totalRows += numOfRows; pOperator->resultInfo.totalRows += numOfRows;
} }
int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, int32_t numOfOutput, SArray* pColList, int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, char* pData, SArray* pColList, char** pNextStart) {
char** pNextStart) {
if (pColList == NULL) { // data from other sources if (pColList == NULL) { // data from other sources
blockDataCleanup(pRes); blockDataCleanup(pRes);
*pNextStart = (char*)blockDecode(pRes, pData); *pNextStart = (char*)blockDecode(pRes, pData);
} else { // extract data according to pColList } else { // extract data according to pColList
ASSERT(numOfOutput == taosArrayGetSize(pColList));
char* pStart = pData; char* pStart = pData;
int32_t numOfCols = htonl(*(int32_t*)pStart); int32_t numOfCols = htonl(*(int32_t*)pStart);
@ -1970,7 +1968,7 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
char* pStart = pRetrieveRsp->data; char* pStart = pRetrieveRsp->data;
while (index++ < pRetrieveRsp->numOfBlocks) { while (index++ < pRetrieveRsp->numOfBlocks) {
SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false); SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
code = extractDataBlockFromFetchRsp(pb, pStart, pRetrieveRsp->numOfCols, NULL, &pStart); code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
if (code != 0) { if (code != 0) {
taosMemoryFreeClear(pDataInfo->pRsp); taosMemoryFreeClear(pDataInfo->pRsp);
goto _error; goto _error;
@ -2095,7 +2093,7 @@ static int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp; SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
char* pStart = pRetrieveRsp->data; char* pStart = pRetrieveRsp->data;
int32_t code = extractDataBlockFromFetchRsp(NULL, pStart, pRetrieveRsp->numOfCols, NULL, &pStart); int32_t code = extractDataBlockFromFetchRsp(NULL, pStart, NULL, &pStart);
if (pRsp->completed == 1) { if (pRsp->completed == 1) {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, rowsOfSource:%" PRIu64 qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, rowsOfSource:%" PRIu64
@ -3915,7 +3913,7 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
pOptr = createAggregateOperatorInfo(ops[0], pExprInfo, num, pResBlock, pAggNode->node.pConditions, pOptr = createAggregateOperatorInfo(ops[0], pExprInfo, num, pResBlock, pAggNode->node.pConditions,
pScalarExprInfo, numOfScalarExpr, pAggNode->mergeDataBlock, pTaskInfo); pScalarExprInfo, numOfScalarExpr, pAggNode->mergeDataBlock, pTaskInfo);
} }
} else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type || QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_HASH_INTERVAL == type) {
SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode; SIntervalPhysiNode* pIntervalPhyNode = (SIntervalPhysiNode*)pPhyNode;
SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &num); SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &num);
@ -3940,6 +3938,8 @@ SOperatorInfo* createOperatorTree(SPhysiNode* pPhyNode, SExecTaskInfo* pTaskInfo
pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, &as, pIntervalPhyNode, pOptr = createIntervalOperatorInfo(ops[0], pExprInfo, num, pResBlock, &interval, tsSlotId, &as, pIntervalPhyNode,
pTaskInfo, isStream); pTaskInfo, isStream);
} else if (QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL == type) {
pOptr = createStreamIntervalOperatorInfo(ops[0], pPhyNode, pTaskInfo);
} else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) { } else if (QUERY_NODE_PHYSICAL_PLAN_MERGE_ALIGNED_INTERVAL == type) {
SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode; SMergeAlignedIntervalPhysiNode* pIntervalPhyNode = (SMergeAlignedIntervalPhysiNode*)pPhyNode;
pOptr = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo); pOptr = createMergeAlignedIntervalOperatorInfo(ops[0], pIntervalPhyNode, pTaskInfo);

2
source/libs/executor/src/scanoperator.c
View File

@ -2649,7 +2649,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
} }
char* pStart = pRsp->data; char* pStart = pRsp->data;
extractDataBlockFromFetchRsp(pInfo->pRes, pRsp->data, pOperator->exprSupp.numOfExprs, pInfo->scanCols, &pStart); extractDataBlockFromFetchRsp(pInfo->pRes, pRsp->data, pInfo->scanCols, &pStart);
updateLoadRemoteInfo(&pInfo->loadInfo, pRsp->numOfRows, pRsp->compLen, startTs, pOperator); updateLoadRemoteInfo(&pInfo->loadInfo, pRsp->numOfRows, pRsp->compLen, startTs, pOperator);
// todo log the filter info // todo log the filter info

488
source/libs/executor/src/timewindowoperator.c
View File

@ -939,7 +939,7 @@ bool isOverdue(TSKEY ts, STimeWindowAggSupp* pSup) {
bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) { return isOverdue(pWin->ekey, pSup); } bool isCloseWindow(STimeWindow* pWin, STimeWindowAggSupp* pSup) { return isOverdue(pWin->ekey, pSup); }
static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pBlock, static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo, SSDataBlock* pBlock,
int32_t scanFlag, SHashObj* pUpdatedMap) { int32_t scanFlag) {
SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info; SIntervalAggOperatorInfo* pInfo = (SIntervalAggOperatorInfo*)pOperatorInfo->info;
SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo; SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
@ -955,21 +955,11 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
STimeWindow win = STimeWindow win =
getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->inputOrder); getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, pInfo->inputOrder);
int32_t ret = TSDB_CODE_SUCCESS; int32_t ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) && numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) { if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx, T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM && pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
saveWinResultRow(pResult, tableGroupId, pUpdatedMap);
setResultBufPageDirty(pInfo->aggSup.pResultBuf, &pResultRowInfo->cur);
}
} }
TSKEY ekey = ascScan ? win.ekey : win.skey; TSKEY ekey = ascScan ? win.ekey : win.skey;
int32_t forwardRows = int32_t forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder); getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
@ -991,12 +981,9 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
doWindowBorderInterpolation(pInfo, pBlock, pResult, &win, startPos, forwardRows, pSup); doWindowBorderInterpolation(pInfo, pBlock, pResult, &win, startPos, forwardRows, pSup);
} }
if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) && updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true);
inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) { doApplyFunctions(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, pBlock->info.rows,
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true); numOfOutput);
doApplyFunctions(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, pBlock->info.rows,
numOfOutput);
}
doCloseWindow(pResultRowInfo, pInfo, pResult); doCloseWindow(pResultRowInfo, pInfo, pResult);
@ -1007,13 +994,6 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
if (startPos < 0) { if (startPos < 0) {
break; break;
} }
if (pInfo->ignoreExpiredData && isCloseWindow(&nextWin, &pInfo->twAggSup)) {
ekey = ascScan ? nextWin.ekey : nextWin.skey;
forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
continue;
}
// null data, failed to allocate more memory buffer // null data, failed to allocate more memory buffer
int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo); pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
@ -1021,11 +1001,6 @@ static void hashIntervalAgg(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResul
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY); T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
} }
if (pInfo->execModel == OPTR_EXEC_MODEL_STREAM && pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
saveWinResultRow(pResult, tableGroupId, pUpdatedMap);
setResultBufPageDirty(pInfo->aggSup.pResultBuf, &pResultRowInfo->cur);
}
ekey = ascScan ? nextWin.ekey : nextWin.skey; ekey = ascScan ? nextWin.ekey : nextWin.skey;
forwardRows = forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder); getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, pInfo->inputOrder);
@ -1130,7 +1105,7 @@ static int32_t doOpenIntervalAgg(SOperatorInfo* pOperator) {
setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->inputOrder, scanFlag, true); setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->inputOrder, scanFlag, true);
blockDataUpdateTsWindow(pBlock, pInfo->primaryTsIndex); blockDataUpdateTsWindow(pBlock, pInfo->primaryTsIndex);
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag, NULL); hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, scanFlag);
} }
initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->resultTsOrder); initGroupedResultInfo(&pInfo->groupResInfo, pInfo->aggSup.pResultRowHashTable, pInfo->resultTsOrder);
@ -1581,141 +1556,6 @@ static void doBuildDeleteResult(SArray* pWins, int32_t* index, SSDataBlock* pBlo
} }
} }
static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
SIntervalAggOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
pInfo->inputOrder = TSDB_ORDER_ASC;
SExprSupp* pSup = &pOperator->exprSupp;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
if (pOperator->status == OP_RES_TO_RETURN) {
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single interval");
return pInfo->pDelRes;
}
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pInfo->binfo.pRes->info.rows == 0 || !hasRemainResults(&pInfo->groupResInfo)) {
pOperator->status = OP_EXEC_DONE;
qDebug("===stream===single interval is done");
freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
}
printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
}
SOperatorInfo* downstream = pOperator->pDownstream[0];
SArray* pUpdated = taosArrayInit(4, POINTER_BYTES); // SResKeyPos
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
SHashObj* pUpdatedMap = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);
SStreamState* pState = pTaskInfo->streamInfo.pState;
while (1) {
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) {
break;
}
// qInfo("===stream===%ld", pBlock->info.version);
printDataBlock(pBlock, "single interval recv");
if (pBlock->info.type == STREAM_CLEAR) {
doClearWindows(&pInfo->aggSup, &pOperator->exprSupp, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock,
NULL);
qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
continue;
}
if (pBlock->info.type == STREAM_DELETE_DATA) {
doDeleteSpecifyIntervalWindow(&pInfo->aggSup, pBlock, pInfo->pDelWins, &pInfo->interval, pUpdatedMap);
continue;
} else if (pBlock->info.type == STREAM_GET_ALL) {
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdatedMap);
continue;
}
if (pBlock->info.type == STREAM_NORMAL && pBlock->info.version != 0) {
// set input version
pTaskInfo->version = pBlock->info.version;
}
if (pInfo->scalarSupp.pExprInfo != NULL) {
SExprSupp* pExprSup = &pInfo->scalarSupp;
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
// The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
// caller. Note that all the time window are not close till now.
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, pInfo->inputOrder, MAIN_SCAN, true);
if (pInfo->invertible) {
setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type);
}
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, pBlock->info.window.ekey);
hashIntervalAgg(pOperator, &pInfo->binfo.resultRowInfo, pBlock, MAIN_SCAN, pUpdatedMap);
}
#if 0
if (pState) {
printf(">>>>>>>> stream read backend\n");
SWinKey key = {
.ts = 1,
.groupId = 2,
};
char* val = NULL;
int32_t sz;
if (streamStateGet(pState, &key, (void**)&val, &sz) < 0) {
ASSERT(0);
}
printf("stream read %s %d\n", val, sz);
streamFreeVal(val);
SStreamStateCur* pCur = streamStateGetCur(pState, &key);
ASSERT(pCur);
while (streamStateCurNext(pState, pCur) == 0) {
SWinKey key1;
const void* val1;
if (streamStateGetKVByCur(pCur, &key1, &val1, &sz) < 0) {
break;
}
printf("stream iter key groupId:%d ts:%d, value %s %d\n", key1.groupId, key1.ts, val1, sz);
}
streamStateFreeCur(pCur);
}
#endif
pOperator->status = OP_RES_TO_RETURN;
closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, NULL, pUpdatedMap,
pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
void* pIte = NULL;
while ((pIte = taosHashIterate(pUpdatedMap, pIte)) != NULL) {
taosArrayPush(pUpdated, pIte);
}
taosArraySort(pUpdated, resultrowComparAsc);
finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
removeDeleteResults(pUpdatedMap, pInfo->pDelWins);
taosHashCleanup(pUpdatedMap);
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single interval");
return pInfo->pDelRes;
}
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
}
static void destroyStateWindowOperatorInfo(void* param) { static void destroyStateWindowOperatorInfo(void* param) {
SStateWindowOperatorInfo* pInfo = (SStateWindowOperatorInfo*)param; SStateWindowOperatorInfo* pInfo = (SStateWindowOperatorInfo*)param;
cleanupBasicInfo(&pInfo->binfo); cleanupBasicInfo(&pInfo->binfo);
@ -1925,7 +1765,7 @@ SOperatorInfo* createIntervalOperatorInfo(SOperatorInfo* downstream, SExprInfo*
pOperator->status = OP_NOT_OPENED; pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, doStreamIntervalAgg, NULL, pOperator->fpSet = createOperatorFpSet(doOpenIntervalAgg, doBuildIntervalResult, NULL, NULL,
destroyIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL); destroyIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
if (nodeType(pPhyNode) == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL) { if (nodeType(pPhyNode) == QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL) {
@ -5627,3 +5467,311 @@ _error:
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
} }
static void doStreamIntervalAggImpl(SOperatorInfo* pOperatorInfo, SResultRowInfo* pResultRowInfo,
SSDataBlock* pBlock, int32_t scanFlag, SHashObj* pUpdatedMap) {
SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)pOperatorInfo->info;
SExecTaskInfo* pTaskInfo = pOperatorInfo->pTaskInfo;
SExprSupp* pSup = &pOperatorInfo->exprSupp;
int32_t startPos = 0;
int32_t numOfOutput = pSup->numOfExprs;
SColumnInfoData* pColDataInfo = taosArrayGet(pBlock->pDataBlock, pInfo->primaryTsIndex);
TSKEY* tsCols = (TSKEY*)pColDataInfo->pData;
uint64_t tableGroupId = pBlock->info.groupId;
bool ascScan = true;
TSKEY ts = getStartTsKey(&pBlock->info.window, tsCols);
SResultRow* pResult = NULL;
STimeWindow win =
getActiveTimeWindow(pInfo->aggSup.pResultBuf, pResultRowInfo, ts, &pInfo->interval, TSDB_ORDER_ASC);
int32_t ret = TSDB_CODE_SUCCESS;
if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) &&
inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) {
ret = setTimeWindowOutputBuf(pResultRowInfo, &win, (scanFlag == MAIN_SCAN), &pResult, tableGroupId, pSup->pCtx,
numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
if (ret != TSDB_CODE_SUCCESS || pResult == NULL) {
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
saveWinResultRow(pResult, tableGroupId, pUpdatedMap);
setResultBufPageDirty(pInfo->aggSup.pResultBuf, &pResultRowInfo->cur);
}
}
TSKEY ekey = ascScan ? win.ekey : win.skey;
int32_t forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
ASSERT(forwardRows > 0);
if ((!pInfo->ignoreExpiredData || !isCloseWindow(&win, &pInfo->twAggSup)) &&
inSlidingWindow(&pInfo->interval, &win, &pBlock->info)) {
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &win, true);
doApplyFunctions(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, pBlock->info.rows,
numOfOutput);
}
STimeWindow nextWin = win;
while (1) {
int32_t prevEndPos = forwardRows - 1 + startPos;
startPos = getNextQualifiedWindow(&pInfo->interval, &nextWin, &pBlock->info, tsCols, prevEndPos, TSDB_ORDER_ASC);
if (startPos < 0) {
break;
}
if (pInfo->ignoreExpiredData && isCloseWindow(&nextWin, &pInfo->twAggSup)) {
ekey = ascScan ? nextWin.ekey : nextWin.skey;
forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
continue;
}
// null data, failed to allocate more memory buffer
int32_t code = setTimeWindowOutputBuf(pResultRowInfo, &nextWin, (scanFlag == MAIN_SCAN), &pResult, tableGroupId,
pSup->pCtx, numOfOutput, pSup->rowEntryInfoOffset, &pInfo->aggSup, pTaskInfo);
if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
if (pInfo->twAggSup.calTrigger == STREAM_TRIGGER_AT_ONCE) {
saveWinResultRow(pResult, tableGroupId, pUpdatedMap);
setResultBufPageDirty(pInfo->aggSup.pResultBuf, &pResultRowInfo->cur);
}
ekey = ascScan ? nextWin.ekey : nextWin.skey;
forwardRows =
getNumOfRowsInTimeWindow(&pBlock->info, tsCols, startPos, ekey, binarySearchForKey, NULL, TSDB_ORDER_ASC);
updateTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &nextWin, true);
doApplyFunctions(pTaskInfo, pSup->pCtx, &pInfo->twAggSup.timeWindowData, startPos, forwardRows, pBlock->info.rows,
numOfOutput);
}
}
static SSDataBlock* doStreamIntervalAgg(SOperatorInfo* pOperator) {
SStreamIntervalOperatorInfo* pInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
int64_t maxTs = INT64_MIN;
SExprSupp* pSup = &pOperator->exprSupp;
if (pOperator->status == OP_EXEC_DONE) {
return NULL;
}
if (pOperator->status == OP_RES_TO_RETURN) {
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single interval");
return pInfo->pDelRes;
}
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
if (pInfo->binfo.pRes->info.rows == 0 || !hasRemainResults(&pInfo->groupResInfo)) {
pOperator->status = OP_EXEC_DONE;
qDebug("===stream===single interval is done");
freeAllPages(pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
}
printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
}
SOperatorInfo* downstream = pOperator->pDownstream[0];
SArray* pUpdated = taosArrayInit(4, POINTER_BYTES); // SResKeyPos
_hash_fn_t hashFn = taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY);
SHashObj* pUpdatedMap = taosHashInit(1024, hashFn, false, HASH_NO_LOCK);
SStreamState* pState = pTaskInfo->streamInfo.pState;
while (1) {
SSDataBlock* pBlock = downstream->fpSet.getNextFn(downstream);
if (pBlock == NULL) {
break;
}
printDataBlock(pBlock, "single interval recv");
if (pBlock->info.type == STREAM_CLEAR) {
doClearWindows(&pInfo->aggSup, &pOperator->exprSupp, &pInfo->interval, pOperator->exprSupp.numOfExprs, pBlock,
NULL);
qDebug("%s clear existed time window results for updates checked", GET_TASKID(pTaskInfo));
continue;
} else if (pBlock->info.type == STREAM_DELETE_DATA) {
doDeleteSpecifyIntervalWindow(&pInfo->aggSup, pBlock, pInfo->pDelWins, &pInfo->interval, pUpdatedMap);
continue;
} else if (pBlock->info.type == STREAM_GET_ALL) {
getAllIntervalWindow(pInfo->aggSup.pResultRowHashTable, pUpdatedMap);
continue;
}
if (pBlock->info.type == STREAM_NORMAL && pBlock->info.version != 0) {
// set input version
pTaskInfo->version = pBlock->info.version;
}
if (pInfo->scalarSupp.pExprInfo != NULL) {
SExprSupp* pExprSup = &pInfo->scalarSupp;
projectApplyFunctions(pExprSup->pExprInfo, pBlock, pBlock, pExprSup->pCtx, pExprSup->numOfExprs, NULL);
}
// The timewindow that overlaps the timestamps of the input pBlock need to be recalculated and return to the
// caller. Note that all the time window are not close till now.
// the pDataBlock are always the same one, no need to call this again
setInputDataBlock(pOperator, pSup->pCtx, pBlock, TSDB_ORDER_ASC, MAIN_SCAN, true);
if (pInfo->invertible) {
setInverFunction(pSup->pCtx, pOperator->exprSupp.numOfExprs, pBlock->info.type);
}
maxTs = TMAX(maxTs, pBlock->info.window.ekey);
doStreamIntervalAggImpl(pOperator, &pInfo->binfo.resultRowInfo, pBlock, MAIN_SCAN, pUpdatedMap);
}
pInfo->twAggSup.maxTs = TMAX(pInfo->twAggSup.maxTs, maxTs);
#if 0
if (pState) {
printf(">>>>>>>> stream read backend\n");
SWinKey key = {
.ts = 1,
.groupId = 2,
};
char* val = NULL;
int32_t sz;
if (streamStateGet(pState, &key, (void**)&val, &sz) < 0) {
ASSERT(0);
}
printf("stream read %s %d\n", val, sz);
streamFreeVal(val);
SStreamStateCur* pCur = streamStateGetCur(pState, &key);
ASSERT(pCur);
while (streamStateCurNext(pState, pCur) == 0) {
SWinKey key1;
const void* val1;
if (streamStateGetKVByCur(pCur, &key1, &val1, &sz) < 0) {
break;
}
printf("stream iter key groupId:%d ts:%d, value %s %d\n", key1.groupId, key1.ts, val1, sz);
}
streamStateFreeCur(pCur);
}
#endif
pOperator->status = OP_RES_TO_RETURN;
closeIntervalWindow(pInfo->aggSup.pResultRowHashTable, &pInfo->twAggSup, &pInfo->interval, NULL, pUpdatedMap,
pInfo->pRecycledPages, pInfo->aggSup.pResultBuf);
void* pIte = NULL;
while ((pIte = taosHashIterate(pUpdatedMap, pIte)) != NULL) {
taosArrayPush(pUpdated, pIte);
}
taosArraySort(pUpdated, resultrowComparAsc);
finalizeUpdatedResult(pOperator->exprSupp.numOfExprs, pInfo->aggSup.pResultBuf, pUpdated, pSup->rowEntryInfoOffset);
initMultiResInfoFromArrayList(&pInfo->groupResInfo, pUpdated);
blockDataEnsureCapacity(pInfo->binfo.pRes, pOperator->resultInfo.capacity);
removeDeleteResults(pUpdatedMap, pInfo->pDelWins);
taosHashCleanup(pUpdatedMap);
doBuildDeleteResult(pInfo->pDelWins, &pInfo->delIndex, pInfo->pDelRes);
if (pInfo->pDelRes->info.rows > 0) {
printDataBlock(pInfo->pDelRes, "single interval");
return pInfo->pDelRes;
}
doBuildResultDatablock(pOperator, &pInfo->binfo, &pInfo->groupResInfo, pInfo->aggSup.pResultBuf);
printDataBlock(pInfo->binfo.pRes, "single interval");
return pInfo->binfo.pRes->info.rows == 0 ? NULL : pInfo->binfo.pRes;
}
void destroyStreamIntervalOperatorInfo(void* param) {
SStreamIntervalOperatorInfo* pInfo = (SStreamIntervalOperatorInfo*)param;
cleanupBasicInfo(&pInfo->binfo);
cleanupAggSup(&pInfo->aggSup);
pInfo->pRecycledPages = taosArrayDestroy(pInfo->pRecycledPages);
pInfo->pDelWins = taosArrayDestroy(pInfo->pDelWins);
pInfo->pDelRes = blockDataDestroy(pInfo->pDelRes);
cleanupGroupResInfo(&pInfo->groupResInfo);
colDataDestroy(&pInfo->twAggSup.timeWindowData);
taosMemoryFreeClear(param);
}
SOperatorInfo* createStreamIntervalOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNode,
SExecTaskInfo* pTaskInfo) {
SStreamIntervalOperatorInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamIntervalOperatorInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) {
goto _error;
}
SStreamIntervalPhysiNode* pIntervalPhyNode = (SStreamIntervalPhysiNode*)pPhyNode;
int32_t numOfCols = 0;
SExprInfo* pExprInfo = createExprInfo(pIntervalPhyNode->window.pFuncs, NULL, &numOfCols);
ASSERT(numOfCols > 0);
SSDataBlock* pResBlock = createResDataBlock(pPhyNode->pOutputDataBlockDesc);
SInterval interval = {.interval = pIntervalPhyNode->interval,
.sliding = pIntervalPhyNode->sliding,
.intervalUnit = pIntervalPhyNode->intervalUnit,
.slidingUnit = pIntervalPhyNode->slidingUnit,
.offset = pIntervalPhyNode->offset,
.precision = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->node.resType.precision, };
STimeWindowAggSupp twAggSupp = {.waterMark = pIntervalPhyNode->window.watermark,
.calTrigger = pIntervalPhyNode->window.triggerType,
.maxTs = INT64_MIN, };
ASSERT(twAggSupp.calTrigger != STREAM_TRIGGER_MAX_DELAY);
pOperator->pTaskInfo = pTaskInfo;
pInfo->interval = interval;
pInfo->twAggSup = twAggSupp;
pInfo->ignoreExpiredData = pIntervalPhyNode->window.igExpired;
pInfo->isFinal = false;
if (pIntervalPhyNode->window.pExprs != NULL) {
int32_t numOfScalar = 0;
SExprInfo* pScalarExprInfo = createExprInfo(pIntervalPhyNode->window.pExprs, NULL, &numOfScalar);
int32_t code = initExprSupp(&pInfo->scalarSupp, pScalarExprInfo, numOfScalar);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
}
pInfo->primaryTsIndex = ((SColumnNode*)pIntervalPhyNode->window.pTspk)->slotId;;
initResultSizeInfo(&pOperator->resultInfo, 4096);
SExprSupp* pSup = &pOperator->exprSupp;
size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
int32_t code = initAggInfo(pSup, &pInfo->aggSup, pExprInfo, numOfCols, keyBufSize, pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
initBasicInfo(&pInfo->binfo, pResBlock);
initStreamFunciton(pSup->pCtx, pSup->numOfExprs);
initExecTimeWindowInfo(&pInfo->twAggSup.timeWindowData, &pTaskInfo->window);
pInfo->invertible = allInvertible(pSup->pCtx, numOfCols);
pInfo->invertible = false; // Todo(liuyao): Dependent TSDB API
pInfo->pRecycledPages = taosArrayInit(4, sizeof(int32_t));
pInfo->pDelWins = taosArrayInit(4, sizeof(SWinKey));
pInfo->delIndex = 0;
pInfo->pDelRes = createSpecialDataBlock(STREAM_DELETE_RESULT);
initResultRowInfo(&pInfo->binfo.resultRowInfo);
pOperator->name = "StreamIntervalOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_STREAM_INTERVAL;
pOperator->blocking = true;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->fpSet = createOperatorFpSet(operatorDummyOpenFn, doStreamIntervalAgg, NULL, NULL,
destroyStreamIntervalOperatorInfo, aggEncodeResultRow, aggDecodeResultRow, NULL);
initIntervalDownStream(downstream, pPhyNode->type, &pInfo->aggSup, &pInfo->interval, pInfo->twAggSup.waterMark);
code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
return pOperator;
_error:
destroyStreamIntervalOperatorInfo(pInfo);
taosMemoryFreeClear(pOperator);
pTaskInfo->code = code;
return NULL;
}

34
tests/script/tsim/stream/basic1.sim
View File

@ -588,4 +588,38 @@ if $data00 != 5 then
goto loop3 goto loop3
endi endi
#max,min selectivity
sql create database test3 vgroups 1;
sql use test3;
sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int);
sql create table ts1 using st tags(1,1,1);
sql create stream stream_t3 trigger at_once into streamtST3 as select ts, min(a) c6, a, b, c, ta, tb, tc from ts1 interval(10s) ;
sql insert into ts1 values(1648791211000,1,2,3);
sleep 50
sql insert into ts1 values(1648791222001,2,2,3);
sleep 50
$loop_count = 0
loop3:
sql select * from streamtST3;
sleep 300
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
# row 0
if $data02 != 1 then
print =====data02=$data02
goto loop3
endi
# row 1
if $data12 != 2 then
print =====data12=$data12
goto loop3
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT system sh/exec.sh -n dnode1 -s stop -x SIGINT

41
tests/script/tsim/stream/distributeInterval0.sim
View File

@ -198,7 +198,7 @@ endi
sql select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5, avg(d) from st interval(10s); sql select _wstart, count(*) c1, count(d) c2 , sum(a) c3 , max(b) c4, min(c) c5, avg(d) from st interval(10s);
sql create database test1 vgroups 1; sql create database test1 vgroups 4;
sql use test1; sql use test1;
sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int); sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int);
sql create table ts1 using st tags(1,1,1); sql create table ts1 using st tags(1,1,1);
@ -232,4 +232,43 @@ if $data11 != 2 then
goto loop2 goto loop2
endi endi
#max,min selectivity
sql create database test3 vgroups 4;
sql use test3;
sql create stable st(ts timestamp, a int, b int , c int) tags(ta int,tb int,tc int);
sql create table ts1 using st tags(1,1,1);
sql create table ts2 using st tags(2,2,2);
sql create stream stream_t3 trigger at_once into streamtST3 as select ts, min(a) c6, a, b, c, ta, tb, tc from st interval(10s) ;
sql insert into ts1 values(1648791211000,1,2,3);
sleep 50
sql insert into ts1 values(1648791222001,2,2,3);
sleep 50
sql insert into ts2 values(1648791211000,1,2,3);
sleep 50
sql insert into ts2 values(1648791222001,2,2,3);
sleep 50
$loop_count = 0
loop3:
sql select * from streamtST3;
sleep 300
$loop_count = $loop_count + 1
if $loop_count == 10 then
return -1
endi
# row 0
if $data02 != 1 then
print =====data02=$data02
goto loop3
endi
# row 1
if $data12 != 2 then
print =====data12=$data12
goto loop3
endi
system sh/stop_dnodes.sh system sh/stop_dnodes.sh

28
tests/script/tsim/stream/partitionbyColumn0.sim
View File

@ -24,7 +24,7 @@ sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
$loop_count = 0 $loop_count = 0
loop0: loop0:
sleep 100 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -48,7 +48,7 @@ sql insert into t1 values(1648791213000,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop1: loop1:
sleep 100 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -71,7 +71,7 @@ sql insert into t1 values(1648791213000,2,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop2: loop2:
sleep 100 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -97,7 +97,7 @@ sql insert into t1 values(1648791213002,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop3: loop3:
sleep 100 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -134,7 +134,7 @@ sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (16
$loop_count = 0 $loop_count = 0
loop4: loop4:
sleep 100 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -208,7 +208,7 @@ sql insert into t1 values(1648791213001,1,2,3,2.0);
$loop_count = 0 $loop_count = 0
loop5: loop5:
sleep 100 sleep 50
sql select * from streamt1 order by c1, c4, c2, c3; sql select * from streamt1 order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -229,7 +229,7 @@ sql insert into t1 values(1648791213001,1,1,6,2.0) (1648791223002,1,1,7,2.0);
$loop_count = 0 $loop_count = 0
loop6: loop6:
sleep 100 sleep 50
sql select * from streamt1 order by c1, c4, c2, c3; sql select * from streamt1 order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -294,7 +294,7 @@ sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL);
$loop_count = 0 $loop_count = 0
loop7: loop7:
sleep 100 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -318,7 +318,7 @@ sql insert into t2 values(1648791213000,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop8: loop8:
sleep 100 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -342,7 +342,7 @@ sql insert into t2 values(1648791213000,2,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop9: loop9:
sleep 100 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -372,7 +372,7 @@ sql insert into t2 values(1648791213002,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop10: loop10:
sleep 100 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -414,7 +414,7 @@ sql insert into t2 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (16
$loop_count = 0 $loop_count = 0
loop11: loop11:
sleep 100 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -492,7 +492,7 @@ sql insert into t4 values(1648791213000,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop13: loop13:
sleep 100 sleep 50
sql select * from test.streamt4 order by c1, c2, c3; sql select * from test.streamt4 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
@ -534,7 +534,7 @@ sql insert into t1 values(1648791213000,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop14: loop14:
sleep 100 sleep 50
sql select * from test.streamt4 order by c1, c2, c3; sql select * from test.streamt4 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1

85
tests/script/tsim/stream/partitionbyColumn1.sim
View File

@ -24,11 +24,11 @@ sql insert into t1 values(1648791213000,NULL,NULL,NULL,NULL);
$loop_count = 0 $loop_count = 0
loop0: loop0:
sleep 300 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -45,12 +45,14 @@ endi
sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t1 values(1648791213000,1,2,3,1.0);
$loop_count = 0
loop1: loop1:
sleep 300 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -66,12 +68,14 @@ endi
sql insert into t1 values(1648791213000,2,2,3,1.0); sql insert into t1 values(1648791213000,2,2,3,1.0);
$loop_count = 0
loop2: loop2:
sleep 300 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -90,12 +94,14 @@ sql insert into t1 values(1648791213001,2,2,3,1.0);
sql insert into t1 values(1648791213002,2,2,3,1.0); sql insert into t1 values(1648791213002,2,2,3,1.0);
sql insert into t1 values(1648791213002,1,2,3,1.0); sql insert into t1 values(1648791213002,1,2,3,1.0);
$loop_count = 0
loop3: loop3:
sleep 300 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -125,12 +131,14 @@ sql insert into t1 values(1648791223002,3,2,3,1.0);
sql insert into t1 values(1648791223003,3,2,3,1.0); sql insert into t1 values(1648791223003,3,2,3,1.0);
sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0); sql insert into t1 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
$loop_count = 0
loop4: loop4:
sleep 300 sleep 50
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -199,11 +207,11 @@ sql insert into t1 values(1648791213001,1,2,3,2.0);
$loop_count = 0 $loop_count = 0
loop5: loop5:
sleep 300 sleep 50
sql select * from streamt1 order by c1, c4, c2, c3; sql select * from streamt1 order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -217,12 +225,14 @@ sql insert into t1 values(1648791223001,1,2,5,2.0);
sql insert into t1 values(1648791223002,1,2,5,2.0); sql insert into t1 values(1648791223002,1,2,5,2.0);
sql insert into t1 values(1648791213001,1,1,6,2.0) (1648791223002,1,1,7,2.0); sql insert into t1 values(1648791213001,1,1,6,2.0) (1648791223002,1,1,7,2.0);
$loop_count = 0
loop6: loop6:
sleep 300 sleep 50
sql select * from streamt1 order by c1, c4, c2, c3; sql select * from streamt1 order by c1, c4, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -282,11 +292,11 @@ sql insert into t2 values(1648791213000,NULL,NULL,NULL,NULL);
$loop_count = 0 $loop_count = 0
loop7: loop7:
sleep 300 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -303,12 +313,14 @@ endi
sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t1 values(1648791213000,1,2,3,1.0);
sql insert into t2 values(1648791213000,1,2,3,1.0); sql insert into t2 values(1648791213000,1,2,3,1.0);
$loop_count = 0
loop8: loop8:
sleep 300 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -324,12 +336,15 @@ endi
sql insert into t1 values(1648791213000,2,2,3,1.0); sql insert into t1 values(1648791213000,2,2,3,1.0);
sql insert into t2 values(1648791213000,2,2,3,1.0); sql insert into t2 values(1648791213000,2,2,3,1.0);
$loop_count = 0
loop9: loop9:
sleep 300 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -352,12 +367,14 @@ sql insert into t2 values(1648791213001,2,2,3,1.0);
sql insert into t2 values(1648791213002,2,2,3,1.0); sql insert into t2 values(1648791213002,2,2,3,1.0);
sql insert into t2 values(1648791213002,1,2,3,1.0); sql insert into t2 values(1648791213002,1,2,3,1.0);
$loop_count = 0
loop10: loop10:
sleep 300 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -373,7 +390,7 @@ endi
if $data11 != 2 thenloop4 if $data11 != 2 thenloop4
print =====data11=$data11 print =====data11=$data11
goto loop3 goto loop10
endi endi
if $data12 != 1 then if $data12 != 1 then
@ -392,12 +409,14 @@ sql insert into t2 values(1648791223002,3,2,3,1.0);
sql insert into t2 values(1648791223003,3,2,3,1.0); sql insert into t2 values(1648791223003,3,2,3,1.0);
sql insert into t2 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0); sql insert into t2 values(1648791213001,1,2,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
$loop_count = 0
loop11: loop11:
sleep 300 sleep 50
sql select * from test.streamt2 order by c1, c2, c3; sql select * from test.streamt2 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
@ -470,17 +489,17 @@ sql insert into t4 values(1648791213000,1,2,3,1.0);
$loop_count = 0 $loop_count = 0
loop13: loop13:
sleep 300 sleep 50
sql select * from test.streamt4 order by c1, c2, c3; sql select * from test.streamt4 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi
if $rows != 2 then if $rows != 2 then
print =====rows=$rows print =====rows=$rows
goto loop14 goto loop13
endi endi
if $data01 != 1 then if $data01 != 1 then
@ -495,12 +514,12 @@ endi
if $data11 != 3 then if $data11 != 3 then
print =====data11=$data11 print =====data11=$data11
goto loop11 goto loop13
endi endi
if $data12 != 2 then if $data12 != 2 then
print =====data12=$data12 print =====data12=$data12
goto loop11 goto loop13
endi endi
sql insert into t4 values(1648791213000,2,2,3,1.0); sql insert into t4 values(1648791213000,2,2,3,1.0);
@ -509,12 +528,14 @@ sql insert into t1 values(1648791233000,2,2,3,1.0);
sql insert into t1 values(1648791213000,1,2,3,1.0); sql insert into t1 values(1648791213000,1,2,3,1.0);
$loop_count = 0
loop14: loop14:
sleep 300 sleep 50
sql select * from test.streamt4 order by c1, c2, c3; sql select * from test.streamt4 order by c1, c2, c3;
$loop_count = $loop_count + 1 $loop_count = $loop_count + 1
if $loop_count == 10 then if $loop_count == 20 then
return -1 return -1
endi endi

10
tests/script/tsim/stream/partitionbyColumn2.sim
View File

@ -40,6 +40,8 @@ endi
sql insert into t1 values(1648791213000,1,1,3,1.0); sql insert into t1 values(1648791213000,1,1,3,1.0);
$loop_count = 0
loop1: loop1:
sleep 300 sleep 300
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
@ -61,6 +63,8 @@ endi
sql insert into t1 values(1648791213000,2,1,3,1.0); sql insert into t1 values(1648791213000,2,1,3,1.0);
$loop_count = 0
loop2: loop2:
sleep 300 sleep 300
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
@ -85,6 +89,8 @@ sql insert into t1 values(1648791213001,2,1,3,1.0);
sql insert into t1 values(1648791213002,2,1,3,1.0); sql insert into t1 values(1648791213002,2,1,3,1.0);
sql insert into t1 values(1648791213002,1,1,3,1.0); sql insert into t1 values(1648791213002,1,1,3,1.0);
$loop_count = 0
loop3: loop3:
sleep 300 sleep 300
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
@ -120,6 +126,8 @@ sql insert into t1 values(1648791223002,3,2,3,1.0);
sql insert into t1 values(1648791223003,3,2,3,1.0); sql insert into t1 values(1648791223003,3,2,3,1.0);
sql insert into t1 values(1648791213001,1,1,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0); sql insert into t1 values(1648791213001,1,1,3,1.0) (1648791223001,2,2,3,1.0) (1648791223003,1,2,3,1.0);
$loop_count = 0
loop4: loop4:
sleep 300 sleep 300
sql select * from streamt order by c1, c4, c2, c3; sql select * from streamt order by c1, c4, c2, c3;
@ -212,6 +220,8 @@ sql insert into t1 values(1648791223001,1,2,2,5);
sql insert into t1 values(1648791223002,1,2,2,6); sql insert into t1 values(1648791223002,1,2,2,6);
sql insert into t1 values(1648791213001,1,1,1,7) (1648791223002,1,1,2,8); sql insert into t1 values(1648791213001,1,1,1,7) (1648791223002,1,1,2,8);
$loop_count = 0
loop6: loop6:
sleep 300 sleep 300
sql select * from streamt1 order by c1, c4, c2, c3; sql select * from streamt1 order by c1, c4, c2, c3;

10
utils/test/c/sml_test.c
View File

@ -92,7 +92,7 @@ int smlProcess_telnet_Test() {
int smlProcess_json1_Test() { int smlProcess_json1_Test() {
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0); TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
TAOS_RES *pRes = taos_query(taos, "create database if not exists sml_db schemaless 1"); TAOS_RES *pRes = taos_query(taos, "create database if not exists sml_db");
taos_free_result(pRes); taos_free_result(pRes);
pRes = taos_query(taos, "use sml_db"); pRes = taos_query(taos, "use sml_db");
@ -112,7 +112,7 @@ int smlProcess_json1_Test() {
" }," " },"
" {" " {"
" \"metric\": \"sys.cpu.nice\"," " \"metric\": \"sys.cpu.nice\","
" \"timestamp\": 1346846400," " \"timestamp\": 1662344042,"
" \"value\": 9," " \"value\": 9,"
" \"tags\": {" " \"tags\": {"
" \"host\": \"web02\"," " \"host\": \"web02\","
@ -141,7 +141,7 @@ int smlProcess_json2_Test() {
"{" "{"
" \"metric\": \"meter_current0\"," " \"metric\": \"meter_current0\","
" \"timestamp\": {" " \"timestamp\": {"
" \"value\" : 1346846400," " \"value\" : 1662344042,"
" \"type\" : \"s\"" " \"type\" : \"s\""
" }," " },"
" \"value\": {" " \"value\": {"
@ -181,7 +181,7 @@ int smlProcess_json3_Test() {
"{" "{"
" \"metric\": \"meter_current1\"," " \"metric\": \"meter_current1\","
" \"timestamp\": {" " \"timestamp\": {"
" \"value\" : 1346846400," " \"value\" : 1662344042,"
" \"type\" : \"s\"" " \"type\" : \"s\""
" }," " },"
" \"value\": {" " \"value\": {"
@ -249,7 +249,7 @@ int smlProcess_json4_Test() {
"{" "{"
" \"metric\": \"meter_current2\"," " \"metric\": \"meter_current2\","
" \"timestamp\": {" " \"timestamp\": {"
" \"value\" : 1346846500000," " \"value\" : 1662344042000,"
" \"type\" : \"ms\"" " \"type\" : \"ms\""
" }," " },"
" \"value\": \"ni\"," " \"value\": \"ni\","