diff --git a/src/cq/src/cqMain.c b/src/cq/src/cqMain.c index 6e81db7db735b49a4a94b9b5a5af76469dc84fb9..9406a2fdce90713d784758d479e953aa18aacdb9 100644 --- a/src/cq/src/cqMain.c +++ b/src/cq/src/cqMain.c @@ -15,17 +15,19 @@ #define _DEFAULT_SOURCE +#include +#include #include #include -#include -#include + +#include "taos.h" #include "taosdef.h" #include "taosmsg.h" +#include "tcq.h" +#include "tdataformat.h" #include "tglobal.h" #include "tlog.h" #include "twal.h" -#include "tcq.h" -#include "taos.h" #define cError(...) { if (cqDebugFlag & DEBUG_ERROR) { taosPrintLog("ERROR CQ ", cqDebugFlag, __VA_ARGS__); }} #define cWarn(...) { if (cqDebugFlag & DEBUG_WARN) { taosPrintLog("WARN CQ ", cqDebugFlag, __VA_ARGS__); }} @@ -46,15 +48,14 @@ typedef struct { } SCqContext; typedef struct SCqObj { - int tid; // table ID - int rowSize; // bytes of a row - char *sqlStr; // SQL string - int columns; // number of columns - SSchema *pSchema; // pointer to schema array - void *pStream; - struct SCqObj *prev; - struct SCqObj *next; - SCqContext *pContext; + int tid; // table ID + int rowSize; // bytes of a row + char * sqlStr; // SQL string + STSchema * pSchema; // pointer to schema array + void * pStream; + struct SCqObj *prev; + struct SCqObj *next; + SCqContext * pContext; } SCqObj; int cqDebugFlag = 135; @@ -152,7 +153,7 @@ void cqStop(void *handle) { pthread_mutex_unlock(&pContext->mutex); } -void *cqCreate(void *handle, int tid, char *sqlStr, SSchema *pSchema, int columns) { +void *cqCreate(void *handle, int tid, char *sqlStr, STSchema *pSchema) { SCqContext *pContext = handle; SCqObj *pObj = calloc(sizeof(SCqObj), 1); @@ -162,11 +163,7 @@ void *cqCreate(void *handle, int tid, char *sqlStr, SSchema *pSchema, int column pObj->sqlStr = malloc(strlen(sqlStr)+1); strcpy(pObj->sqlStr, sqlStr); - pObj->columns = columns; - - int size = sizeof(SSchema) * columns; - pObj->pSchema = malloc(size); - memcpy(pObj->pSchema, pSchema, size); + pObj->pSchema = tdDupSchema(pSchema); cTrace("vgId:%d, id:%d CQ:%s is created", pContext->vgId, pObj->tid, pObj->sqlStr); diff --git a/src/cq/test/cqtest.c b/src/cq/test/cqtest.c index 7977bd85bc18fcf659fa3608b4ccf99923529d92..3aa649ee34ce84cdc79b9d42dffd8656137a618e 100644 --- a/src/cq/test/cqtest.c +++ b/src/cq/test/cqtest.c @@ -59,21 +59,16 @@ int main(int argc, char *argv[]) { exit(-1); } - SSchema schema[2]; - schema[0].type = TSDB_DATA_TYPE_TIMESTAMP; - strcpy(schema[0].name, "ts"); - schema[0].colId = 0; - schema[0].bytes = 8; - - schema[1].type = TSDB_DATA_TYPE_INT; - strcpy(schema[1].name, "avgspeed"); - schema[1].colId = 1; - schema[1].bytes = 4; + STSchema *pSchema = tdNewSchema(2); + tdSchemaAddCol(pSchema, TSDB_DATA_TYPE_TIMESTAMP, 0, 8); + tdSchemaAddCol(pSchema, TSDB_DATA_TYPE_INT, 1, 4); for (int sid =1; sid<10; ++sid) { - cqCreate(pCq, sid, "select avg(speed) from demo.t1 sliding(1s) interval(5s)", schema, 2); + cqCreate(pCq, sid, "select avg(speed) from demo.t1 sliding(1s) interval(5s)", pSchema); } + tdFreeSchema(pSchema); + while (1) { char c = getchar(); diff --git a/src/inc/tcq.h b/src/inc/tcq.h index d0a2097c05d2501e169e3f4347acc122c4f1430f..e025afaa0ae428b9545b5a23308830f8aa455002 100644 --- a/src/inc/tcq.h +++ b/src/inc/tcq.h @@ -19,6 +19,7 @@ extern "C" { #endif +#include "tdataformat.h" typedef int (*FCqWrite)(void *ahandle, void *pHead, int type); @@ -40,7 +41,7 @@ void cqStart(void *handle); void cqStop(void *handle); // cqCreate is called by TSDB to start an instance of CQ -void *cqCreate(void *handle, int sid, char *sqlStr, SSchema *pSchema, int columns); +void *cqCreate(void *handle, int sid, char *sqlStr, STSchema *pSchema); // cqDrop is called by TSDB to stop an instance of CQ, handle is the return value of cqCreate void cqDrop(void *handle); diff --git a/src/inc/tsdb.h b/src/inc/tsdb.h index 341dee14769271103b8e508dab22f287bb329a99..c758d3aea4340f57140eea0d8f3f5ae95f94a2c6 100644 --- a/src/inc/tsdb.h +++ b/src/inc/tsdb.h @@ -43,6 +43,8 @@ typedef struct { void *cqH; int (*notifyStatus)(void *, int status); int (*eventCallBack)(void *); + void *(*cqCreateFunc)(void *handle, int sid, char *sqlStr, STSchema *pSchema); + void (*cqDropFunc)(void *handle); } STsdbAppH; // --------- TSDB REPOSITORY CONFIGURATION DEFINITION @@ -71,7 +73,7 @@ typedef void TsdbRepoT; // use void to hide implementation details from outside int tsdbCreateRepo(char *rootDir, STsdbCfg *pCfg, void *limiter); int32_t tsdbDropRepo(TsdbRepoT *repo); -TsdbRepoT *tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH); +TsdbRepoT *tsdbOpenRepo(char *rootDir, STsdbAppH *pAppH); int32_t tsdbCloseRepo(TsdbRepoT *repo, int toCommit); int32_t tsdbConfigRepo(TsdbRepoT *repo, STsdbCfg *pCfg); diff --git a/src/tsdb/inc/tsdbMain.h b/src/tsdb/inc/tsdbMain.h index 0839e0f8ff786d6d72cfdf301ec58cd20401b685..9dd5136c959f30bc56804b87f9f63692d1cc0eca 100644 --- a/src/tsdb/inc/tsdbMain.h +++ b/src/tsdb/inc/tsdbMain.h @@ -87,6 +87,7 @@ typedef struct STable { struct STable *prev; tstr * name; // NOTE: there a flexible string here char * sql; + void * cqhandle; } STable; #define TSDB_GET_TABLE_LAST_KEY(tb) ((tb)->lastKey) @@ -110,6 +111,7 @@ typedef struct { SMetaFile *mfh; // meta file handle int maxRowBytes; int maxCols; + void * pRepo; } STsdbMeta; // element put in skiplist for each table @@ -118,7 +120,7 @@ typedef struct STableIndexElem { STable* pTable; } STableIndexElem; -STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables); +STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables, void *pRepo); int32_t tsdbFreeMeta(STsdbMeta *pMeta); STSchema * tsdbGetTableSchema(STsdbMeta *pMeta, STable *pTable); STSchema * tsdbGetTableTagSchema(STsdbMeta *pMeta, STable *pTable); diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c index bddb3fcaff997f8589b28bc200c7a33ecf2e8a16..9c8e57d18aa6bcc98d9acb562bd83f070ff97281 100644 --- a/src/tsdb/src/tsdbMain.c +++ b/src/tsdb/src/tsdbMain.c @@ -189,9 +189,9 @@ _err: * * @return a TSDB repository handle on success, NULL for failure and the error number is set */ -TsdbRepoT *tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH) { +TsdbRepoT *tsdbOpenRepo(char *rootDir, STsdbAppH *pAppH) { char dataDir[128] = "\0"; - if (access(tsdbDir, F_OK | W_OK | R_OK) < 0) { + if (access(rootDir, F_OK | W_OK | R_OK) < 0) { return NULL; } @@ -200,12 +200,12 @@ TsdbRepoT *tsdbOpenRepo(char *tsdbDir, STsdbAppH *pAppH) { return NULL; } - pRepo->rootDir = strdup(tsdbDir); + pRepo->rootDir = strdup(rootDir); tsdbRestoreCfg(pRepo, &(pRepo->config)); if (pAppH) pRepo->appH = *pAppH; - pRepo->tsdbMeta = tsdbInitMeta(tsdbDir, pRepo->config.maxTables); + pRepo->tsdbMeta = tsdbInitMeta(rootDir, pRepo->config.maxTables, pRepo); if (pRepo->tsdbMeta == NULL) { free(pRepo->rootDir); free(pRepo); diff --git a/src/tsdb/src/tsdbMeta.c b/src/tsdb/src/tsdbMeta.c index e320de9827c26d02bb2ff6277a90f591e7d06ed9..0d9e6a9cf8c6c5438be023def71657eaaa2e9da6 100644 --- a/src/tsdb/src/tsdbMeta.c +++ b/src/tsdb/src/tsdbMeta.c @@ -142,6 +142,7 @@ int tsdbRestoreTable(void *pHandle, void *cont, int contLen) { void tsdbOrgMeta(void *pHandle) { STsdbMeta *pMeta = (STsdbMeta *)pHandle; + STsdbRepo *pRepo = (STsdbRepo *)pMeta->pRepo; for (int i = 1; i < pMeta->maxTables; i++) { STable *pTable = pMeta->tables[i]; @@ -149,13 +150,20 @@ void tsdbOrgMeta(void *pHandle) { tsdbAddTableIntoIndex(pMeta, pTable); } } + + for (int i = 0; i < pMeta->maxTables; i++) { + STable *pTable = pMeta->tables[i]; + if (pTable && pTable->type == TSDB_STREAM_TABLE) { + pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, i, pTable->sql, tsdbGetTableSchema(pMeta, pTable)); + } + } } /** * Initialize the meta handle * ASSUMPTIONS: VALID PARAMETER */ -STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables) { +STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables, void *pRepo) { STsdbMeta *pMeta = (STsdbMeta *)malloc(sizeof(STsdbMeta)); if (pMeta == NULL) return NULL; @@ -165,6 +173,7 @@ STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables) { pMeta->tables = (STable **)calloc(maxTables, sizeof(STable *)); pMeta->maxRowBytes = 0; pMeta->maxCols = 0; + pMeta->pRepo = pRepo; if (pMeta->tables == NULL) { free(pMeta); return NULL; @@ -189,13 +198,16 @@ STsdbMeta *tsdbInitMeta(char *rootDir, int32_t maxTables) { } int32_t tsdbFreeMeta(STsdbMeta *pMeta) { + STsdbRepo *pRepo = (STsdbRepo *)pMeta->pRepo; if (pMeta == NULL) return 0; tsdbCloseMetaFile(pMeta->mfh); for (int i = 1; i < pMeta->maxTables; i++) { if (pMeta->tables[i] != NULL) { - tsdbFreeTable(pMeta->tables[i]); + STable *pTable = pMeta->tables[i]; + if (pTable->type == TSDB_STREAM_TABLE) (*pRepo->appH.cqDropFunc)(pTable->cqhandle); + tsdbFreeTable(pTable); } } @@ -512,6 +524,7 @@ STable *tsdbGetTableByUid(STsdbMeta *pMeta, uint64_t uid) { } static int tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable, bool addIdx) { + STsdbRepo *pRepo = (STsdbRepo *)pMeta->pRepo; if (pTable->type == TSDB_SUPER_TABLE) { // add super table to the linked list if (pMeta->superList == NULL) { @@ -531,7 +544,7 @@ static int tsdbAddTableToMeta(STsdbMeta *pMeta, STable *pTable, bool addIdx) { tsdbAddTableIntoIndex(pMeta, pTable); } if (pTable->type == TSDB_STREAM_TABLE && addIdx) { - // TODO + pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, pTable->tableId.tid, pTable->sql, tsdbGetTableSchema(pMeta, pTable)); } pMeta->nTables++; diff --git a/src/vnode/src/vnodeMain.c b/src/vnode/src/vnodeMain.c index b25180f0f0e1470405b8e9605f0869640b35d8a7..cc92c0338989718f9ac134eda80b5c96c8255ba1 100644 --- a/src/vnode/src/vnodeMain.c +++ b/src/vnode/src/vnodeMain.c @@ -220,6 +220,8 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) { appH.appH = (void *)pVnode; appH.notifyStatus = vnodeProcessTsdbStatus; appH.cqH = pVnode->cq; + appH.cqCreateFunc = cqCreate; + appH.cqDropFunc = cqDrop; sprintf(temp, "%s/tsdb", rootDir); pVnode->tsdb = tsdbOpenRepo(temp, &appH); if (pVnode->tsdb == NULL) { @@ -391,14 +393,14 @@ static void vnodeCleanUp(SVnodeObj *pVnode) { pVnode->sync = NULL; } - if (pVnode->wal) - walClose(pVnode->wal); - pVnode->wal = NULL; - if (pVnode->tsdb) tsdbCloseRepo(pVnode->tsdb, 1); pVnode->tsdb = NULL; + if (pVnode->wal) + walClose(pVnode->wal); + pVnode->wal = NULL; + if (pVnode->cq) cqClose(pVnode->cq); pVnode->cq = NULL; @@ -467,6 +469,8 @@ static void vnodeNotifyFileSynced(void *ahandle, uint64_t fversion) { appH.appH = (void *)pVnode; appH.notifyStatus = vnodeProcessTsdbStatus; appH.cqH = pVnode->cq; + appH.cqCreateFunc = cqCreate; + appH.cqDropFunc = cqDrop; pVnode->tsdb = tsdbOpenRepo(rootDir, &appH); }