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

support timestamp before 1970

This commit is contained in:
dapan1121 2021-02-04 09:23:35 +08:00
parent bd57fd013a
commit 38e1312b1c
2 changed files with 34 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/common/inc/tdataformat.h
View File

@ -134,6 +134,22 @@ typedef uint64_t TKEY;
#define tdGetTKEY(key) (((TKEY)ABS(key)) | (TKEY_NEGATIVE_FLAG & (TKEY)(key))) #define tdGetTKEY(key) (((TKEY)ABS(key)) | (TKEY_NEGATIVE_FLAG & (TKEY)(key)))
#define tdGetKey(tkey) (((TSKEY)((tkey)&TKEY_VALUE_FILTER)) * (TKEY_IS_NEGATIVE(tkey) ? -1 : 1)) #define tdGetKey(tkey) (((TSKEY)((tkey)&TKEY_VALUE_FILTER)) * (TKEY_IS_NEGATIVE(tkey) ? -1 : 1))
#define MIN_TS_KEY ((TSKEY)0x8000000000000001)
#define MAX_TS_KEY ((TSKEY)0x3fffffffffffffff)
#define TD_TO_TKEY(key) tdGetTKEY(((key) < MIN_TS_KEY) ? MIN_TS_KEY : (((key) > MAX_TS_KEY) ? MAX_TS_KEY : key))
static FORCE_INLINE TKEY keyToTkey(TSKEY key) {
TSKEY lkey = key;
if (key > MAX_TS_KEY) {
lkey = MAX_TS_KEY;
} else if (key < MIN_TS_KEY) {
lkey = MIN_TS_KEY;
}
return tdGetTKEY(lkey);
}
static FORCE_INLINE int tkeyComparFn(const void *tkey1, const void *tkey2) { static FORCE_INLINE int tkeyComparFn(const void *tkey1, const void *tkey2) {
TSKEY key1 = tdGetKey(*(TKEY *)tkey1); TSKEY key1 = tdGetKey(*(TKEY *)tkey1);
TSKEY key2 = tdGetKey(*(TKEY *)tkey2); TSKEY key2 = tdGetKey(*(TKEY *)tkey2);

27
src/tsdb/src/tsdbRead.c
View File

@ -317,7 +317,7 @@ static STsdbQueryHandle* tsdbQueryTablesImpl(STsdbRepo* tsdb, STsdbQueryCond* pC
pQueryHandle->window = pCond->twindow; pQueryHandle->window = pCond->twindow;
pQueryHandle->pTsdb = tsdb; pQueryHandle->pTsdb = tsdb;
pQueryHandle->type = TSDB_QUERY_TYPE_ALL; pQueryHandle->type = TSDB_QUERY_TYPE_ALL;
pQueryHandle->cur.fid = -1; pQueryHandle->cur.fid = INT32_MIN;
pQueryHandle->cur.win = TSWINDOW_INITIALIZER; pQueryHandle->cur.win = TSWINDOW_INITIALIZER;
pQueryHandle->checkFiles = true; pQueryHandle->checkFiles = true;
pQueryHandle->activeIndex = 0; // current active table index pQueryHandle->activeIndex = 0; // current active table index
@ -478,16 +478,18 @@ static bool initTableMemIterator(STsdbQueryHandle* pHandle, STableCheckInfo* pCh
if (pMemT && pCheckInfo->tableId.tid < pMemT->maxTables) { if (pMemT && pCheckInfo->tableId.tid < pMemT->maxTables) {
pMem = pMemT->tData[pCheckInfo->tableId.tid]; pMem = pMemT->tData[pCheckInfo->tableId.tid];
if (pMem != NULL && pMem->uid == pCheckInfo->tableId.uid) { // check uid if (pMem != NULL && pMem->uid == pCheckInfo->tableId.uid) { // check uid
TKEY tLastKey = keyToTkey(pCheckInfo->lastKey);
pCheckInfo->iter = pCheckInfo->iter =
tSkipListCreateIterFromVal(pMem->pData, (const char*)&pCheckInfo->lastKey, TSDB_DATA_TYPE_TIMESTAMP, order); tSkipListCreateIterFromVal(pMem->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
} }
} }
if (pIMemT && pCheckInfo->tableId.tid < pIMemT->maxTables) { if (pIMemT && pCheckInfo->tableId.tid < pIMemT->maxTables) {
pIMem = pIMemT->tData[pCheckInfo->tableId.tid]; pIMem = pIMemT->tData[pCheckInfo->tableId.tid];
if (pIMem != NULL && pIMem->uid == pCheckInfo->tableId.uid) { // check uid if (pIMem != NULL && pIMem->uid == pCheckInfo->tableId.uid) { // check uid
TKEY tLastKey = keyToTkey(pCheckInfo->lastKey);
pCheckInfo->iiter = pCheckInfo->iiter =
tSkipListCreateIterFromVal(pIMem->pData, (const char*)&pCheckInfo->lastKey, TSDB_DATA_TYPE_TIMESTAMP, order); tSkipListCreateIterFromVal(pIMem->pData, (const char*)&tLastKey, TSDB_DATA_TYPE_TIMESTAMP, order);
} }
} }
@ -652,7 +654,7 @@ static bool hasMoreDataInCache(STsdbQueryHandle* pHandle) {
STsdbCfg *pCfg = &pHandle->pTsdb->config; STsdbCfg *pCfg = &pHandle->pTsdb->config;
size_t size = taosArrayGetSize(pHandle->pTableCheckInfo); size_t size = taosArrayGetSize(pHandle->pTableCheckInfo);
assert(pHandle->activeIndex < size && pHandle->activeIndex >= 0 && size >= 1); assert(pHandle->activeIndex < size && pHandle->activeIndex >= 0 && size >= 1);
pHandle->cur.fid = -1; pHandle->cur.fid = INT32_MIN;
STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex); STableCheckInfo* pCheckInfo = taosArrayGet(pHandle->pTableCheckInfo, pHandle->activeIndex);
@ -1117,7 +1119,12 @@ int32_t doCopyRowsFromFileBlock(STsdbQueryHandle* pQueryHandle, int32_t capacity
if (pColInfo->info.colId == src->colId) { if (pColInfo->info.colId == src->colId) {
if (pColInfo->info.type != TSDB_DATA_TYPE_BINARY && pColInfo->info.type != TSDB_DATA_TYPE_NCHAR) { if (pColInfo->info.type == TSDB_DATA_TYPE_TIMESTAMP) {
for (int32_t n = 0; n < num; n++) {
TKEY tkey = *(TKEY *)((char*)src->pData + bytes * start + n * sizeof(TKEY));
*(TSKEY *)(pData + n * sizeof(TSKEY)) = tdGetKey(tkey);
}
} else if (pColInfo->info.type != TSDB_DATA_TYPE_BINARY && pColInfo->info.type != TSDB_DATA_TYPE_NCHAR) {
memmove(pData, (char*)src->pData + bytes * start, bytes * num); memmove(pData, (char*)src->pData + bytes * start, bytes * num);
} else { // handle the var-string } else { // handle the var-string
char* dst = pData; char* dst = pData;
@ -1222,7 +1229,6 @@ static void copyOneRowFromMem(STsdbQueryHandle* pQueryHandle, int32_t capacity,
break; break;
case TSDB_DATA_TYPE_BIGINT: case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT: case TSDB_DATA_TYPE_UBIGINT:
case TSDB_DATA_TYPE_TIMESTAMP:
*(uint64_t *)pData = *(uint64_t *)value; *(uint64_t *)pData = *(uint64_t *)value;
break; break;
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
@ -1231,6 +1237,9 @@ static void copyOneRowFromMem(STsdbQueryHandle* pQueryHandle, int32_t capacity,
case TSDB_DATA_TYPE_DOUBLE: case TSDB_DATA_TYPE_DOUBLE:
SET_DOUBLE_PTR(pData, value); SET_DOUBLE_PTR(pData, value);
break; break;
case TSDB_DATA_TYPE_TIMESTAMP:
*(TSKEY *)pData = tdGetKey(*(TKEY *)value);
break;
default: default:
memcpy(pData, value, pColInfo->info.bytes); memcpy(pData, value, pColInfo->info.bytes);
} }
@ -1890,7 +1899,7 @@ static int32_t getFirstFileDataBlock(STsdbQueryHandle* pQueryHandle, bool* exist
assert(pQueryHandle->pFileGroup == NULL); assert(pQueryHandle->pFileGroup == NULL);
} }
cur->fid = -1; // denote that there are no data in file anymore cur->fid = INT32_MIN; // denote that there are no data in file anymore
*exists = false; *exists = false;
return code; return code;
} }
@ -2493,7 +2502,7 @@ void tsdbRetrieveDataBlockInfo(TsdbQueryHandleT* pQueryHandle, SDataBlockInfo* p
STable* pTable = NULL; STable* pTable = NULL;
// there are data in file // there are data in file
if (pHandle->cur.fid >= 0) { if (pHandle->cur.fid != INT32_MIN) {
STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[cur->slot]; STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[cur->slot];
pTable = pBlockInfo->pTableCheckInfo->pTableObj; pTable = pBlockInfo->pTableCheckInfo->pTableObj;
} else { } else {
@ -2577,7 +2586,7 @@ SArray* tsdbRetrieveDataBlock(TsdbQueryHandleT* pQueryHandle, SArray* pIdList) {
*/ */
STsdbQueryHandle* pHandle = (STsdbQueryHandle*)pQueryHandle; STsdbQueryHandle* pHandle = (STsdbQueryHandle*)pQueryHandle;
if (pHandle->cur.fid < 0) { if (pHandle->cur.fid == INT32_MIN) {
return pHandle->pColumns; return pHandle->pColumns;
} else { } else {
STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[pHandle->cur.slot]; STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[pHandle->cur.slot];