提交 a9b18d20 编写于 作者: H Haojun Liao

fix:fix error.

上级 01f6f74f
...@@ -59,7 +59,7 @@ typedef struct SDataSinkMgtCfg { ...@@ -59,7 +59,7 @@ typedef struct SDataSinkMgtCfg {
uint32_t maxDataBlockNumPerQuery; uint32_t maxDataBlockNumPerQuery;
} SDataSinkMgtCfg; } SDataSinkMgtCfg;
int32_t dsDataSinkMgtInit(SDataSinkMgtCfg* cfg); int32_t dsDataSinkMgtInit(SDataSinkMgtCfg* cfg, SStorageAPI* pAPI);
typedef struct SInputData { typedef struct SInputData {
const struct SSDataBlock* pData; const struct SSDataBlock* pData;
......
...@@ -118,7 +118,7 @@ const void *metaGetTableTagVal(const void *tag, int16_t type, STagVal *tagVal); ...@@ -118,7 +118,7 @@ const void *metaGetTableTagVal(const void *tag, int16_t type, STagVal *tagVal);
int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName); int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName);
int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName); int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName);
int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid); int metaGetTableUidByName(void *pVnode, char *tbName, uint64_t *uid);
int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType); int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType);
bool metaIsTableExist(void* pVnode, tb_uid_t uid); bool metaIsTableExist(void* pVnode, tb_uid_t uid);
int32_t metaGetCachedTableUidList(SMeta *pMeta, tb_uid_t suid, const uint8_t *key, int32_t keyLen, SArray *pList, int32_t metaGetCachedTableUidList(SMeta *pMeta, tb_uid_t suid, const uint8_t *key, int32_t keyLen, SArray *pList,
...@@ -192,7 +192,7 @@ int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList, ...@@ -192,7 +192,7 @@ int32_t tsdbCacherowsReaderOpen(void *pVnode, int32_t type, void *pTableIdList,
int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, const int32_t *dstSlotIds, int32_t tsdbRetrieveCacheRows(void *pReader, SSDataBlock *pResBlock, const int32_t *slotIds, const int32_t *dstSlotIds,
SArray *pTableUids); SArray *pTableUids);
void *tsdbCacherowsReaderClose(void *pReader); void *tsdbCacherowsReaderClose(void *pReader);
int32_t tsdbGetTableSchema(SVnode *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid); int32_t tsdbGetTableSchema(void *pVnode, int64_t uid, STSchema **pSchema, int64_t *suid);
void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity); void tsdbCacheSetCapacity(SVnode *pVnode, size_t capacity);
size_t tsdbCacheGetCapacity(SVnode *pVnode); size_t tsdbCacheGetCapacity(SVnode *pVnode);
......
...@@ -139,10 +139,10 @@ tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) { ...@@ -139,10 +139,10 @@ tb_uid_t metaGetTableEntryUidByName(SMeta *pMeta, const char *name) {
return uid; return uid;
} }
int metaGetTableNameByUid(void *meta, uint64_t uid, char *tbName) { int metaGetTableNameByUid(void *pVnode, uint64_t uid, char *tbName) {
int code = 0; int code = 0;
SMetaReader mr = {0}; SMetaReader mr = {0};
metaReaderInit(&mr, (SMeta *)meta, 0); metaReaderInit(&mr, ((SVnode*)pVnode)->pMeta, 0);
code = metaReaderGetTableEntryByUid(&mr, uid); code = metaReaderGetTableEntryByUid(&mr, uid);
if (code < 0) { if (code < 0) {
metaReaderClear(&mr); metaReaderClear(&mr);
...@@ -170,10 +170,10 @@ int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName) { ...@@ -170,10 +170,10 @@ int metaGetTableSzNameByUid(void *meta, uint64_t uid, char *tbName) {
return 0; return 0;
} }
int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) { int metaGetTableUidByName(void *pVnode, char *tbName, uint64_t *uid) {
int code = 0; int code = 0;
SMetaReader mr = {0}; SMetaReader mr = {0};
metaReaderInit(&mr, (SMeta *)meta, 0); metaReaderInit(&mr, ((SVnode *)pVnode)->pMeta, 0);
SMetaReader *pReader = &mr; SMetaReader *pReader = &mr;
...@@ -191,10 +191,10 @@ int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) { ...@@ -191,10 +191,10 @@ int metaGetTableUidByName(void *meta, char *tbName, uint64_t *uid) {
return 0; return 0;
} }
int metaGetTableTypeByName(void *meta, char *tbName, ETableType *tbType) { int metaGetTableTypeByName(void *pVnode, char *tbName, ETableType *tbType) {
int code = 0; int code = 0;
SMetaReader mr = {0}; SMetaReader mr = {0};
metaReaderInit(&mr, (SMeta *)meta, 0); metaReaderInit(&mr, ((SVnode*)pVnode)->pMeta, 0);
code = metaGetTableEntryByName(&mr, tbName); code = metaGetTableEntryByName(&mr, tbName);
if (code == 0) *tbType = mr.me.type; if (code == 0) *tbType = mr.me.type;
......
...@@ -5373,9 +5373,9 @@ int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) { ...@@ -5373,9 +5373,9 @@ int64_t tsdbGetNumOfRowsInMemTable(STsdbReader* pReader) {
return rows; return rows;
} }
int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid) { int32_t tsdbGetTableSchema(void* pVnode, int64_t uid, STSchema** pSchema, int64_t* suid) {
SMetaReader mr = {0}; SMetaReader mr = {0};
metaReaderInit(&mr, pVnode->pMeta, 0); metaReaderInit(&mr, ((SVnode*)pVnode)->pMeta, 0);
int32_t code = metaReaderGetTableEntryByUidCache(&mr, uid); int32_t code = metaReaderGetTableEntryByUidCache(&mr, uid);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
terrno = TSDB_CODE_TDB_INVALID_TABLE_ID; terrno = TSDB_CODE_TDB_INVALID_TABLE_ID;
...@@ -5405,7 +5405,7 @@ int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int6 ...@@ -5405,7 +5405,7 @@ int32_t tsdbGetTableSchema(SVnode* pVnode, int64_t uid, STSchema** pSchema, int6
metaReaderClear(&mr); metaReaderClear(&mr);
// get the newest table schema version // get the newest table schema version
code = metaGetTbTSchemaEx(pVnode->pMeta, *suid, uid, -1, pSchema); code = metaGetTbTSchemaEx(((SVnode*)pVnode)->pMeta, *suid, uid, -1, pSchema);
return code; return code;
} }
......
...@@ -78,6 +78,8 @@ void initMetadataAPI(SStoreMeta* pMeta) { ...@@ -78,6 +78,8 @@ void initMetadataAPI(SStoreMeta* pMeta) {
pMeta->getTableUidByName = metaGetTableUidByName; pMeta->getTableUidByName = metaGetTableUidByName;
pMeta->getTableTypeByName = metaGetTableTypeByName; pMeta->getTableTypeByName = metaGetTableTypeByName;
pMeta->getTableNameByUid = metaGetTableNameByUid; pMeta->getTableNameByUid = metaGetTableNameByUid;
pMeta->getTableSchema = tsdbGetTableSchema; // todo refactor
} }
void initTqAPI(SStoreTqReader* pTq) { void initTqAPI(SStoreTqReader* pTq) {
......
...@@ -30,7 +30,7 @@ struct SDataSinkHandle; ...@@ -30,7 +30,7 @@ struct SDataSinkHandle;
typedef struct SDataSinkManager { typedef struct SDataSinkManager {
SDataSinkMgtCfg cfg; SDataSinkMgtCfg cfg;
SStorageAPI storeFn; SStorageAPI* pAPI;
} SDataSinkManager; } SDataSinkManager;
typedef int32_t (*FPutDataBlock)(struct SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue); typedef int32_t (*FPutDataBlock)(struct SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue);
......
...@@ -224,7 +224,7 @@ typedef struct STableScanInfo { ...@@ -224,7 +224,7 @@ typedef struct STableScanInfo {
int8_t assignBlockUid; int8_t assignBlockUid;
bool hasGroupByTag; bool hasGroupByTag;
bool countOnly; bool countOnly;
TsdReader readerAPI; // TsdReader readerAPI;
} STableScanInfo; } STableScanInfo;
typedef struct STableMergeScanInfo { typedef struct STableMergeScanInfo {
......
...@@ -429,7 +429,7 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat ...@@ -429,7 +429,7 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
inserter->explain = pInserterNode->explain; inserter->explain = pInserterNode->explain;
int64_t suid = 0; int64_t suid = 0;
int32_t code = pManager->storeFn.metaFn.getTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId, &inserter->pSchema, &suid); int32_t code = pManager->pAPI->metaFn.getTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId, &inserter->pSchema, &suid);
if (code) { if (code) {
destroyDataSinker((SDataSinkHandle*)inserter); destroyDataSinker((SDataSinkHandle*)inserter);
taosMemoryFree(inserter); taosMemoryFree(inserter);
......
...@@ -21,8 +21,9 @@ ...@@ -21,8 +21,9 @@
static SDataSinkManager gDataSinkManager = {0}; static SDataSinkManager gDataSinkManager = {0};
SDataSinkStat gDataSinkStat = {0}; SDataSinkStat gDataSinkStat = {0};
int32_t dsDataSinkMgtInit(SDataSinkMgtCfg* cfg) { int32_t dsDataSinkMgtInit(SDataSinkMgtCfg* cfg, SStorageAPI* pAPI) {
gDataSinkManager.cfg = *cfg; gDataSinkManager.cfg = *cfg;
gDataSinkManager.pAPI = pAPI;
return 0; // to avoid compiler eror return 0; // to avoid compiler eror
} }
......
...@@ -507,7 +507,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId, ...@@ -507,7 +507,7 @@ int32_t qCreateExecTask(SReadHandle* readHandle, int32_t vgId, uint64_t taskId,
} }
SDataSinkMgtCfg cfg = {.maxDataBlockNum = 500, .maxDataBlockNumPerQuery = 50}; SDataSinkMgtCfg cfg = {.maxDataBlockNum = 500, .maxDataBlockNumPerQuery = 50};
code = dsDataSinkMgtInit(&cfg); code = dsDataSinkMgtInit(&cfg, &(*pTask)->storageAPI);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str); qError("failed to dsDataSinkMgtInit, code:%s, %s", tstrerror(code), (*pTask)->id.str);
goto _error; goto _error;
......
...@@ -929,7 +929,7 @@ static void destroyTableScanOperatorInfo(void* param) { ...@@ -929,7 +929,7 @@ static void destroyTableScanOperatorInfo(void* param) {
STableScanInfo* pTableScanInfo = (STableScanInfo*)param; STableScanInfo* pTableScanInfo = (STableScanInfo*)param;
blockDataDestroy(pTableScanInfo->pResBlock); blockDataDestroy(pTableScanInfo->pResBlock);
taosHashCleanup(pTableScanInfo->pIgnoreTables); taosHashCleanup(pTableScanInfo->pIgnoreTables);
destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->readerAPI); destroyTableScanBase(&pTableScanInfo->base, &pTableScanInfo->base.readerAPI);
taosMemoryFreeClear(param); taosMemoryFreeClear(param);
} }
...@@ -975,7 +975,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, ...@@ -975,7 +975,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
pInfo->sample.sampleRatio = pTableScanNode->ratio; pInfo->sample.sampleRatio = pTableScanNode->ratio;
pInfo->sample.seed = taosGetTimestampSec(); pInfo->sample.seed = taosGetTimestampSec();
pInfo->readerAPI = pTaskInfo->storageAPI.tsdReader; pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;
initResultSizeInfo(&pOperator->resultInfo, 4096); initResultSizeInfo(&pOperator->resultInfo, 4096);
pInfo->pResBlock = createDataBlockFromDescNode(pDescNode); pInfo->pResBlock = createDataBlockFromDescNode(pDescNode);
// blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity); // blockDataEnsureCapacity(pInfo->pResBlock, pOperator->resultInfo.capacity);
...@@ -1075,7 +1075,7 @@ void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin, uint6 ...@@ -1075,7 +1075,7 @@ void resetTableScanInfo(STableScanInfo* pTableScanInfo, STimeWindow* pWin, uint6
pTableScanInfo->base.cond.endVersion = version; pTableScanInfo->base.cond.endVersion = version;
pTableScanInfo->scanTimes = 0; pTableScanInfo->scanTimes = 0;
pTableScanInfo->currentGroupId = -1; pTableScanInfo->currentGroupId = -1;
pTableScanInfo->readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader); pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);
pTableScanInfo->base.dataReader = NULL; pTableScanInfo->base.dataReader = NULL;
} }
...@@ -1250,7 +1250,7 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32 ...@@ -1250,7 +1250,7 @@ static SSDataBlock* doRangeScan(SStreamScanInfo* pInfo, SSDataBlock* pSDB, int32
*pRowIndex = 0; *pRowIndex = 0;
pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX}; pInfo->updateWin = (STimeWindow){.skey = INT64_MIN, .ekey = INT64_MAX};
STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info; STableScanInfo* pTableScanInfo = pInfo->pTableScanOp->info;
pTableScanInfo->readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader); pTableScanInfo->base.readerAPI.tsdReaderClose(pTableScanInfo->base.dataReader);
pTableScanInfo->base.dataReader = NULL; pTableScanInfo->base.dataReader = NULL;
return NULL; return NULL;
} }
...@@ -3045,6 +3045,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN ...@@ -3045,6 +3045,7 @@ SOperatorInfo* createTableMergeScanOperatorInfo(STableScanPhysiNode* pTableScanN
goto _error; goto _error;
} }
pInfo->base.readerAPI = pTaskInfo->storageAPI.tsdReader;
pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD; pInfo->base.dataBlockLoadFlag = FUNC_DATA_REQUIRED_DATA_LOAD;
pInfo->base.scanFlag = MAIN_SCAN; pInfo->base.scanFlag = MAIN_SCAN;
pInfo->base.readHandle = *readHandle; pInfo->base.readHandle = *readHandle;
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册