diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index b9aaea8469795771854919a2584d28d5c3f4e9e3..3ab9e2f66faafc8d4bda0b6993c0070e0ec14ea6 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -580,7 +580,7 @@ static FORCE_INLINE void convertToSKVRow(SMemRow dest, SMemRow src, SSchema *pSc SKVRow kvRow = memRowKvBody(dest); memRowSetType(dest, SMEM_ROW_KV); - memRowSetKvVersion(kvRow, dataRowVersion(dataRow)); + memRowSetKvVersion(dest, dataRowVersion(dataRow)); kvRowSetNCols(kvRow, nBoundCols); kvRowSetLen(kvRow, (TDRowLenT)(TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * nBoundCols)); diff --git a/src/tsdb/inc/tsdbMeta.h b/src/tsdb/inc/tsdbMeta.h index 8ce5e7ade80b2006ac8c39fec178994073c5a26d..a13b522eece382fc9c8f0d1de471e2f1c9421840 100644 --- a/src/tsdb/inc/tsdbMeta.h +++ b/src/tsdb/inc/tsdbMeta.h @@ -74,7 +74,7 @@ void tsdbFreeMeta(STsdbMeta* pMeta); int tsdbOpenMeta(STsdbRepo* pRepo); int tsdbCloseMeta(STsdbRepo* pRepo); STable* tsdbGetTableByUid(STsdbMeta* pMeta, uint64_t uid); -STSchema* tsdbGetTableSchemaByVersion(STable* pTable, int16_t _version); +STSchema* tsdbGetTableSchemaByVersion(STable* pTable, int16_t _version, int8_t rowType); int tsdbWLockRepoMeta(STsdbRepo* pRepo); int tsdbRLockRepoMeta(STsdbRepo* pRepo); int tsdbUnlockRepoMeta(STsdbRepo* pRepo); @@ -99,7 +99,9 @@ static FORCE_INLINE int tsdbCompareSchemaVersion(const void *key1, const void *k } } -static FORCE_INLINE STSchema* tsdbGetTableSchemaImpl(STable* pTable, bool lock, bool copy, int16_t _version) { +// set rowType to -1 at default if have no relationship with row +static FORCE_INLINE STSchema* tsdbGetTableSchemaImpl(STable* pTable, bool lock, bool copy, int16_t _version, + int8_t rowType) { STable* pDTable = (pTable->pSuper != NULL) ? pTable->pSuper : pTable; // for performance purpose STSchema* pSchema = NULL; STSchema* pTSchema = NULL; @@ -110,8 +112,12 @@ static FORCE_INLINE STSchema* tsdbGetTableSchemaImpl(STable* pTable, bool lock, } else { // get the schema with version void* ptr = taosArraySearch(pDTable->schema, &_version, tsdbCompareSchemaVersion, TD_EQ); if (ptr == NULL) { - terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION; - goto _exit; + if (rowType == SMEM_ROW_KV) { + ptr = taosArrayGetLast(pDTable->schema); + } else { + terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION; + goto _exit; + } } pTSchema = *(STSchema**)ptr; } @@ -130,7 +136,7 @@ _exit: } static FORCE_INLINE STSchema* tsdbGetTableSchema(STable* pTable) { - return tsdbGetTableSchemaImpl(pTable, false, false, -1); + return tsdbGetTableSchemaImpl(pTable, false, false, -1, -1); } static FORCE_INLINE STSchema *tsdbGetTableTagSchema(STable *pTable) { diff --git a/src/tsdb/src/tsdbCommit.c b/src/tsdb/src/tsdbCommit.c index 6c41571c2d2274b0bbda57460e1fe93505038c97..2c36e1b67ac7efd845b0e818d9cfa45b214795b3 100644 --- a/src/tsdb/src/tsdbCommit.c +++ b/src/tsdb/src/tsdbCommit.c @@ -866,7 +866,7 @@ static int tsdbCommitToTable(SCommitH *pCommith, int tid) { } static int tsdbSetCommitTable(SCommitH *pCommith, STable *pTable) { - STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1); + STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1, -1); pCommith->pTable = pTable; @@ -1283,7 +1283,8 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt (*iter)++; } else if (key1 > key2) { if (pSchema == NULL || schemaVersion(pSchema) != memRowVersion(row)) { - pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, memRowVersion(row)); + pSchema = + tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, memRowVersion(row), (int8_t)memRowType(row)); ASSERT(pSchema != NULL); } @@ -1304,7 +1305,8 @@ static void tsdbLoadAndMergeFromCache(SDataCols *pDataCols, int *iter, SCommitIt if (update != TD_ROW_DISCARD_UPDATE) { //copy mem data if (pSchema == NULL || schemaVersion(pSchema) != memRowVersion(row)) { - pSchema = tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, memRowVersion(row)); + pSchema = + tsdbGetTableSchemaImpl(pCommitIter->pTable, false, false, memRowVersion(row), (int8_t)memRowType(row)); ASSERT(pSchema != NULL); } diff --git a/src/tsdb/src/tsdbCompact.c b/src/tsdb/src/tsdbCompact.c index 5ccb9e90f2407561709d36a85ac3e992e5d5a8ba..9b890ec113600420c179f4d13159706d79dd322a 100644 --- a/src/tsdb/src/tsdbCompact.c +++ b/src/tsdb/src/tsdbCompact.c @@ -431,7 +431,7 @@ static int tsdbCompactMeta(STsdbRepo *pRepo) { if (pTh->pTable == NULL || pTh->pBlkIdx == NULL) continue; - pSchema = tsdbGetTableSchemaImpl(pTh->pTable, true, true, -1); + pSchema = tsdbGetTableSchemaImpl(pTh->pTable, true, true, -1, -1); taosArrayClear(pComph->aSupBlk); if ((tdInitDataCols(pComph->pDataCols, pSchema) < 0) || (tdInitDataCols(pReadh->pDCols[0], pSchema) < 0) || (tdInitDataCols(pReadh->pDCols[1], pSchema) < 0)) { diff --git a/src/tsdb/src/tsdbMain.c b/src/tsdb/src/tsdbMain.c index b2e6fe89161d0e9bceaf74a46807f51ec402fb2a..a22f9265e0a829f91292f427cdb81b20c8172f98 100644 --- a/src/tsdb/src/tsdbMain.c +++ b/src/tsdb/src/tsdbMain.c @@ -617,7 +617,7 @@ static void tsdbStartStream(STsdbRepo *pRepo) { STable *pTable = pMeta->tables[i]; if (pTable && pTable->type == TSDB_STREAM_TABLE) { pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, TABLE_UID(pTable), TABLE_TID(pTable), TABLE_NAME(pTable)->data, pTable->sql, - tsdbGetTableSchemaImpl(pTable, false, false, -1), 0); + tsdbGetTableSchemaImpl(pTable, false, false, -1, -1), 0); } } } diff --git a/src/tsdb/src/tsdbMemTable.c b/src/tsdb/src/tsdbMemTable.c index e766d97a97a5905db87691426d282a219eef9d68..c82490c3e8aa0c5ffabda78a9fc3222c958fd79e 100644 --- a/src/tsdb/src/tsdbMemTable.c +++ b/src/tsdb/src/tsdbMemTable.c @@ -582,7 +582,7 @@ static int tsdbAdjustMemMaxTables(SMemTable *pMemTable, int maxTables) { static int tsdbAppendTableRowToCols(STable *pTable, SDataCols *pCols, STSchema **ppSchema, SMemRow row) { if (pCols) { if (*ppSchema == NULL || schemaVersion(*ppSchema) != memRowVersion(row)) { - *ppSchema = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row)); + *ppSchema = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row), (int8_t)memRowType(row)); if (*ppSchema == NULL) { ASSERT(false); return -1; @@ -730,7 +730,7 @@ static SMemRow tsdbInsertDupKeyMerge(SMemRow row1, SMemRow row2, STsdbRepo* pRep if(pSchema2 != NULL && schemaVersion(pSchema2) == dv1) { *ppSchema1 = pSchema2; } else { - *ppSchema1 = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row1)); + *ppSchema1 = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row1), (int8_t)memRowType(row1)); } pSchema1 = *ppSchema1; } @@ -739,7 +739,7 @@ static SMemRow tsdbInsertDupKeyMerge(SMemRow row1, SMemRow row2, STsdbRepo* pRep if(schemaVersion(pSchema1) == dv2) { pSchema2 = pSchema1; } else { - *ppSchema2 = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row2)); + *ppSchema2 = tsdbGetTableSchemaImpl(pTable, false, false, memRowVersion(row2), (int8_t)memRowType(row2)); pSchema2 = *ppSchema2; } } @@ -847,7 +847,7 @@ static int tsdbInsertDataToTable(STsdbRepo* pRepo, SSubmitBlk* pBlock, int32_t * } } - STSchema *pSchema = tsdbGetTableSchemaByVersion(pTable, pBlock->sversion); + STSchema *pSchema = tsdbGetTableSchemaByVersion(pTable, pBlock->sversion, -1); pRepo->stat.pointsWritten += points * schemaNCols(pSchema); pRepo->stat.totalStorage += points * schemaVLen(pSchema); @@ -894,7 +894,7 @@ static int tsdbGetSubmitMsgNext(SSubmitMsgIter *pIter, SSubmitBlk **pPBlock) { static int tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pTable) { ASSERT(pTable != NULL); - STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1); + STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1, -1); int sversion = schemaVersion(pSchema); if (pBlock->sversion == sversion) { @@ -951,7 +951,7 @@ static int tsdbCheckTableSchema(STsdbRepo *pRepo, SSubmitBlk *pBlock, STable *pT } } else { ASSERT(pBlock->sversion >= 0); - if (tsdbGetTableSchemaImpl(pTable, false, false, pBlock->sversion) == NULL) { + if (tsdbGetTableSchemaImpl(pTable, false, false, pBlock->sversion, -1) == NULL) { tsdbError("vgId:%d invalid submit schema version %d to table %s tid %d from client", REPO_ID(pRepo), pBlock->sversion, TABLE_CHAR_NAME(pTable), TABLE_TID(pTable)); terrno = TSDB_CODE_TDB_IVD_TB_SCHEMA_VERSION; @@ -972,7 +972,7 @@ static void updateTableLatestColumn(STsdbRepo *pRepo, STable *pTable, SMemRow ro return; } - pSchema = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row)); + pSchema = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row), (int8_t)memRowType(row)); if (pSchema == NULL) { return; } diff --git a/src/tsdb/src/tsdbMeta.c b/src/tsdb/src/tsdbMeta.c index 8407a0519a5595b750e0fd45cc82dcd9d1f6b5b1..e604b07cc1dd0d37574235263a018baecbe670c1 100644 --- a/src/tsdb/src/tsdbMeta.c +++ b/src/tsdb/src/tsdbMeta.c @@ -534,8 +534,8 @@ STable *tsdbGetTableByUid(STsdbMeta *pMeta, uint64_t uid) { return *(STable **)ptr; } -STSchema *tsdbGetTableSchemaByVersion(STable *pTable, int16_t _version) { - return tsdbGetTableSchemaImpl(pTable, true, false, _version); +STSchema *tsdbGetTableSchemaByVersion(STable *pTable, int16_t _version, int8_t rowType) { + return tsdbGetTableSchemaImpl(pTable, true, false, _version, rowType); } int tsdbWLockRepoMeta(STsdbRepo *pRepo) { @@ -652,7 +652,7 @@ int tsdbInitColIdCacheWithSchema(STable* pTable, STSchema* pSchema) { } STSchema* tsdbGetTableLatestSchema(STable *pTable) { - return tsdbGetTableSchemaByVersion(pTable, -1); + return tsdbGetTableSchemaByVersion(pTable, -1, -1); } int tsdbUpdateLastColSchema(STable *pTable, STSchema *pNewSchema) { @@ -957,7 +957,7 @@ static int tsdbAddTableToMeta(STsdbRepo *pRepo, STable *pTable, bool addIdx, boo } if (TABLE_TYPE(pTable) != TSDB_CHILD_TABLE) { - STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1); + STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1, -1); if (schemaNCols(pSchema) > pMeta->maxCols) pMeta->maxCols = schemaNCols(pSchema); if (schemaTLen(pSchema) > pMeta->maxRowBytes) pMeta->maxRowBytes = schemaTLen(pSchema); } @@ -965,7 +965,7 @@ static int tsdbAddTableToMeta(STsdbRepo *pRepo, STable *pTable, bool addIdx, boo if (lock && tsdbUnlockRepoMeta(pRepo) < 0) return -1; if (TABLE_TYPE(pTable) == TSDB_STREAM_TABLE && addIdx) { pTable->cqhandle = (*pRepo->appH.cqCreateFunc)(pRepo->appH.cqH, TABLE_UID(pTable), TABLE_TID(pTable), TABLE_NAME(pTable)->data, pTable->sql, - tsdbGetTableSchemaImpl(pTable, false, false, -1), 1); + tsdbGetTableSchemaImpl(pTable, false, false, -1, -1), 1); } tsdbDebug("vgId:%d table %s tid %d uid %" PRIu64 " is added to meta", REPO_ID(pRepo), TABLE_CHAR_NAME(pTable), @@ -984,7 +984,7 @@ static void tsdbRemoveTableFromMeta(STsdbRepo *pRepo, STable *pTable, bool rmFro SListNode *pNode = NULL; STable * tTable = NULL; - STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1); + STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1, -1); int maxCols = schemaNCols(pSchema); int maxRowBytes = schemaTLen(pSchema); @@ -1018,7 +1018,7 @@ static void tsdbRemoveTableFromMeta(STsdbRepo *pRepo, STable *pTable, bool rmFro for (int i = 0; i < pMeta->maxTables; i++) { STable *_pTable = pMeta->tables[i]; if (_pTable != NULL) { - pSchema = tsdbGetTableSchemaImpl(_pTable, false, false, -1); + pSchema = tsdbGetTableSchemaImpl(_pTable, false, false, -1, -1); maxCols = MAX(maxCols, schemaNCols(pSchema)); maxRowBytes = MAX(maxRowBytes, schemaTLen(pSchema)); } diff --git a/src/tsdb/src/tsdbRead.c b/src/tsdb/src/tsdbRead.c index 0f1554a3f35007caad7a3047cc7631836bdb7365..02dbe00bf7d5647ca410683647c69e62b248b758 100644 --- a/src/tsdb/src/tsdbRead.c +++ b/src/tsdb/src/tsdbRead.c @@ -1582,7 +1582,7 @@ static void mergeTwoRowFromMem(STsdbQueryHandle* pQueryHandle, int32_t capacity, int32_t numOfColsOfRow1 = 0; if (pSchema1 == NULL) { - pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1)); + pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1), (int8_t)memRowType(row1)); } if(isRow1DataRow) { numOfColsOfRow1 = schemaNCols(pSchema1); @@ -1594,7 +1594,7 @@ static void mergeTwoRowFromMem(STsdbQueryHandle* pQueryHandle, int32_t capacity, if(row2) { isRow2DataRow = isDataRow(row2); if (pSchema2 == NULL) { - pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2)); + pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2), (int8_t)memRowType(row2)); } if(isRow2DataRow) { numOfColsOfRow2 = schemaNCols(pSchema2); @@ -1961,11 +1961,11 @@ static void doMergeTwoLevelData(STsdbQueryHandle* pQueryHandle, STableCheckInfo* if ((key < tsArray[pos] && ASCENDING_TRAVERSE(pQueryHandle->order)) || (key > tsArray[pos] && !ASCENDING_TRAVERSE(pQueryHandle->order))) { if (rv1 != memRowVersion(row1)) { - pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1)); + pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1), (int8_t)memRowType(row1)); rv1 = memRowVersion(row1); } if(row2 && rv2 != memRowVersion(row2)) { - pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2)); + pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2), (int8_t)memRowType(row2)); rv2 = memRowVersion(row2); } @@ -1986,11 +1986,11 @@ static void doMergeTwoLevelData(STsdbQueryHandle* pQueryHandle, STableCheckInfo* doCopyRowsFromFileBlock(pQueryHandle, pQueryHandle->outputCapacity, numOfRows, pos, pos); } if (rv1 != memRowVersion(row1)) { - pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1)); + pSchema1 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row1), (int8_t)memRowType(row1)); rv1 = memRowVersion(row1); } if(row2 && rv2 != memRowVersion(row2)) { - pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2)); + pSchema2 = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row2), (int8_t)memRowType(row2)); rv2 = memRowVersion(row2); } @@ -2654,7 +2654,7 @@ static int tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int win->ekey = key; if (rv != memRowVersion(row)) { - pSchema = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row)); + pSchema = tsdbGetTableSchemaByVersion(pTable, memRowVersion(row), (int8_t)memRowType(row)); rv = memRowVersion(row); } mergeTwoRowFromMem(pQueryHandle, maxRowsToRead, numOfRows, row, NULL, numOfCols, pTable, pSchema, NULL, true); diff --git a/src/tsdb/src/tsdbReadImpl.c b/src/tsdb/src/tsdbReadImpl.c index 74d41cce194f9921ee0c521de9e329bad5eeb3f9..8ab0130218b8e1c52877a3072d7a5ae1f773dd17 100644 --- a/src/tsdb/src/tsdbReadImpl.c +++ b/src/tsdb/src/tsdbReadImpl.c @@ -153,7 +153,7 @@ int tsdbLoadBlockIdx(SReadH *pReadh) { } int tsdbSetReadTable(SReadH *pReadh, STable *pTable) { - STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1); + STSchema *pSchema = tsdbGetTableSchemaImpl(pTable, false, false, -1, -1); pReadh->pTable = pTable;