Merge branch 'feat/TS-4243-3.0' of https://github.com/taosdata/TDengine into feat/TS-4243-3.0
This commit is contained in:
commit
dcf3cf6e71
|
@ -126,7 +126,7 @@ int32_t tRowMerge(SArray *aRowP, STSchema *pTSchema, int8_t flag);
|
|||
int32_t tRowUpsertColData(SRow *pRow, STSchema *pTSchema, SColData *aColData, int32_t nColData, int32_t flag);
|
||||
void tRowGetKey(SRow *pRow, SRowKey *key);
|
||||
int32_t tRowKeyCompare(const void *p1, const void *p2);
|
||||
int32_t tRowKeyAssign(SRowKey* pDst, SRowKey* pSrc);
|
||||
int32_t tRowKeyAssign(SRowKey *pDst, SRowKey *pSrc);
|
||||
|
||||
// SRowIter ================================
|
||||
int32_t tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter);
|
||||
|
@ -174,6 +174,7 @@ int32_t tColDataUpdateValue(SColData *pColData, SColVal *pColVal, bool forward);
|
|||
void tColDataGetValue(SColData *pColData, int32_t iVal, SColVal *pColVal);
|
||||
uint8_t tColDataGetBitValue(const SColData *pColData, int32_t iVal);
|
||||
int32_t tColDataCopy(SColData *pColDataFrom, SColData *pColData, xMallocFn xMalloc, void *arg);
|
||||
void tColDataArrGetRowKey(SColData *aColData, int32_t nColData, int32_t iRow, SRowKey *key);
|
||||
|
||||
extern void (*tColDataCalcSMA[])(SColData *pColData, int64_t *sum, int64_t *max, int64_t *min, int16_t *numOfNull);
|
||||
|
||||
|
@ -188,8 +189,8 @@ void tColDataSortMerge(SArray *colDataArr);
|
|||
int32_t tColDataAddValueByDataBlock(SColData *pColData, int8_t type, int32_t bytes, int32_t nRows, char *lengthOrbitmap,
|
||||
char *data);
|
||||
// for encode/decode
|
||||
int32_t tPutColData(uint8_t *pBuf, SColData *pColData);
|
||||
int32_t tGetColData(uint8_t *pBuf, SColData *pColData);
|
||||
int32_t tPutColData(uint8_t version, uint8_t *pBuf, SColData *pColData);
|
||||
int32_t tGetColData(uint8_t version, uint8_t *pBuf, SColData *pColData);
|
||||
|
||||
// STRUCT ================================
|
||||
struct STColumn {
|
||||
|
|
|
@ -4037,6 +4037,7 @@ int32_t tDeserializeSMqSeekReq(void* buf, int32_t bufLen, SMqSeekReq* pReq);
|
|||
#define SUBMIT_REQ_COLUMN_DATA_FORMAT 0x2
|
||||
#define SUBMIT_REQ_FROM_FILE 0x4
|
||||
#define TD_REQ_FROM_TAOX 0x8
|
||||
#define SUBMIT_REQUEST_VERSION (1)
|
||||
|
||||
#define TD_REQ_FROM_TAOX_OLD 0x1 // for compatibility
|
||||
|
||||
|
|
|
@ -16,7 +16,6 @@
|
|||
#ifndef _TD_COMMON_TOKEN_H_
|
||||
#define _TD_COMMON_TOKEN_H_
|
||||
|
||||
|
||||
#define TK_OR 1
|
||||
#define TK_AND 2
|
||||
#define TK_UNION 3
|
||||
|
|
|
@ -3111,7 +3111,7 @@ static int32_t tColDataCopyRowAppend(SColData *aFromColData, int32_t iFromRow, S
|
|||
return code;
|
||||
}
|
||||
|
||||
static FORCE_INLINE void tColDataArrGetRowKey(SColData *aColData, int32_t nColData, int32_t iRow, SRowKey *key) {
|
||||
void tColDataArrGetRowKey(SColData *aColData, int32_t nColData, int32_t iRow, SRowKey *key) {
|
||||
SColVal cv;
|
||||
|
||||
key->ts = ((TSKEY *)aColData[0].pData)[iRow];
|
||||
|
@ -3490,7 +3490,7 @@ _exit:
|
|||
return;
|
||||
}
|
||||
|
||||
int32_t tPutColData(uint8_t *pBuf, SColData *pColData) {
|
||||
static int32_t tPutColDataVersion0(uint8_t *pBuf, SColData *pColData) {
|
||||
int32_t n = 0;
|
||||
|
||||
n += tPutI16v(pBuf ? pBuf + n : NULL, pColData->cid);
|
||||
|
@ -3532,7 +3532,7 @@ int32_t tPutColData(uint8_t *pBuf, SColData *pColData) {
|
|||
return n;
|
||||
}
|
||||
|
||||
int32_t tGetColData(uint8_t *pBuf, SColData *pColData) {
|
||||
static int32_t tGetColDataVersion0(uint8_t *pBuf, SColData *pColData) {
|
||||
int32_t n = 0;
|
||||
|
||||
n += tGetI16v(pBuf + n, &pColData->cid);
|
||||
|
@ -3571,10 +3571,45 @@ int32_t tGetColData(uint8_t *pBuf, SColData *pColData) {
|
|||
n += pColData->nData;
|
||||
}
|
||||
}
|
||||
pColData->cflag = 0;
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
static int32_t tPutColDataVersion1(uint8_t *pBuf, SColData *pColData) {
|
||||
int32_t n = tPutColDataVersion0(pBuf, pColData);
|
||||
n += tPutI8(pBuf ? pBuf + n : NULL, pColData->cflag);
|
||||
return n;
|
||||
}
|
||||
|
||||
static int32_t tGetColDataVersion1(uint8_t *pBuf, SColData *pColData) {
|
||||
int32_t n = tGetColDataVersion0(pBuf, pColData);
|
||||
n += tGetI8(pBuf ? pBuf + n : NULL, &pColData->cflag);
|
||||
return n;
|
||||
}
|
||||
|
||||
int32_t tPutColData(uint8_t version, uint8_t *pBuf, SColData *pColData) {
|
||||
if (version == 0) {
|
||||
return tPutColDataVersion0(pBuf, pColData);
|
||||
} else if (version == 1) {
|
||||
return tPutColDataVersion1(pBuf, pColData);
|
||||
} else {
|
||||
ASSERT(0);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
int32_t tGetColData(uint8_t version, uint8_t *pBuf, SColData *pColData) {
|
||||
if (version == 0) {
|
||||
return tGetColDataVersion0(pBuf, pColData);
|
||||
} else if (version == 1) {
|
||||
return tGetColDataVersion1(pBuf, pColData);
|
||||
} else {
|
||||
ASSERT(0);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
#define CALC_SUM_MAX_MIN(SUM, MAX, MIN, VAL) \
|
||||
do { \
|
||||
(SUM) += (VAL); \
|
||||
|
|
|
@ -9074,7 +9074,8 @@ int32_t tDecodeSBatchDeleteReqSetCtime(SDecoder *pDecoder, SBatchDeleteReq *pReq
|
|||
static int32_t tEncodeSSubmitTbData(SEncoder *pCoder, const SSubmitTbData *pSubmitTbData) {
|
||||
if (tStartEncode(pCoder) < 0) return -1;
|
||||
|
||||
if (tEncodeI32v(pCoder, pSubmitTbData->flags) < 0) return -1;
|
||||
int32_t flags = pSubmitTbData->flags | ((SUBMIT_REQUEST_VERSION) << 8);
|
||||
if (tEncodeI32v(pCoder, flags) < 0) return -1;
|
||||
|
||||
// auto create table
|
||||
if (pSubmitTbData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {
|
||||
|
@ -9094,7 +9095,8 @@ static int32_t tEncodeSSubmitTbData(SEncoder *pCoder, const SSubmitTbData *pSubm
|
|||
if (tEncodeU64v(pCoder, nColData) < 0) return -1;
|
||||
|
||||
for (uint64_t i = 0; i < nColData; i++) {
|
||||
pCoder->pos += tPutColData(pCoder->data ? pCoder->data + pCoder->pos : NULL, &aColData[i]);
|
||||
pCoder->pos +=
|
||||
tPutColData(SUBMIT_REQUEST_VERSION, pCoder->data ? pCoder->data + pCoder->pos : NULL, &aColData[i]);
|
||||
}
|
||||
} else {
|
||||
if (tEncodeU64v(pCoder, TARRAY_SIZE(pSubmitTbData->aRowP)) < 0) return -1;
|
||||
|
@ -9113,13 +9115,18 @@ static int32_t tEncodeSSubmitTbData(SEncoder *pCoder, const SSubmitTbData *pSubm
|
|||
|
||||
static int32_t tDecodeSSubmitTbData(SDecoder *pCoder, SSubmitTbData *pSubmitTbData) {
|
||||
int32_t code = 0;
|
||||
int32_t flags;
|
||||
uint8_t version;
|
||||
|
||||
if (tStartDecode(pCoder) < 0) {
|
||||
code = TSDB_CODE_INVALID_MSG;
|
||||
goto _exit;
|
||||
}
|
||||
|
||||
if (tDecodeI32v(pCoder, &pSubmitTbData->flags) < 0) return -1;
|
||||
if (tDecodeI32v(pCoder, &flags) < 0) return -1;
|
||||
|
||||
pSubmitTbData->flags = flags & 0xff;
|
||||
version = (flags >> 8) & 0xff;
|
||||
|
||||
if (pSubmitTbData->flags & SUBMIT_REQ_AUTO_CREATE_TABLE) {
|
||||
pSubmitTbData->pCreateTbReq = taosMemoryCalloc(1, sizeof(SVCreateTbReq));
|
||||
|
@ -9163,7 +9170,7 @@ static int32_t tDecodeSSubmitTbData(SDecoder *pCoder, SSubmitTbData *pSubmitTbDa
|
|||
}
|
||||
|
||||
for (int32_t i = 0; i < nColData; ++i) {
|
||||
pCoder->pos += tGetColData(pCoder->data + pCoder->pos, taosArrayReserve(pSubmitTbData->aCol, 1));
|
||||
pCoder->pos += tGetColData(version, pCoder->data + pCoder->pos, taosArrayReserve(pSubmitTbData->aCol, 1));
|
||||
}
|
||||
} else {
|
||||
uint64_t nRow;
|
||||
|
|
|
@ -57,7 +57,6 @@ static void saveOneRowForLastRaw(SLastCol* pColVal, SCacheRowsReader* pReader, c
|
|||
static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* pReader, const int32_t* slotIds,
|
||||
const int32_t* dstSlotIds, void** pRes, const char* idStr) {
|
||||
int32_t numOfRows = pBlock->info.rows;
|
||||
// bool allNullRow = true;
|
||||
|
||||
if (HASTYPE(pReader->type, CACHESCAN_RETRIEVE_LAST)) {
|
||||
uint64_t ts = TSKEY_MIN;
|
||||
|
@ -108,11 +107,12 @@ static int32_t saveOneRow(SArray* pRow, SSDataBlock* pBlock, SCacheRowsReader* p
|
|||
}
|
||||
}
|
||||
|
||||
// pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to substruct it
|
||||
// pColInfoData->info.bytes includes the VARSTR_HEADER_SIZE, need to subtract it
|
||||
p->hasResult = true;
|
||||
varDataSetLen(pRes[i], pColInfoData->info.bytes - VARSTR_HEADER_SIZE);
|
||||
colDataSetVal(pColInfoData, numOfRows, (const char*)pRes[i], false);
|
||||
}
|
||||
|
||||
for (int32_t idx = 0; idx < taosArrayGetSize(pBlock->pDataBlock); ++idx) {
|
||||
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, idx);
|
||||
if (idx < funcTypeBlockArray->size) {
|
||||
|
@ -233,6 +233,8 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, void* pTableIdList,
|
|||
if (IS_VAR_DATA_TYPE(pPkCol->type)) {
|
||||
p->rowKey.pks[0].pData = taosMemoryCalloc(1, pPkCol->bytes);
|
||||
}
|
||||
|
||||
p->pkColumn = *pPkCol;
|
||||
}
|
||||
|
||||
if (numOfTables == 0) {
|
||||
|
@ -366,15 +368,15 @@ int32_t tsdbRetrieveCacheRows(void* pReader, SSDataBlock* pResBlock, const int32
|
|||
goto _end;
|
||||
}
|
||||
|
||||
for (int32_t j = 0; j < pr->numOfCols; ++j) {
|
||||
int32_t bytes;
|
||||
if (slotIds[j] == -1) {
|
||||
bytes = 1;
|
||||
} else {
|
||||
bytes = pr->pSchema->columns[slotIds[j]].bytes;
|
||||
}
|
||||
int32_t pkBufLen = 0;
|
||||
if (pr->rowKey.numOfPKs > 0) {
|
||||
pkBufLen = pr->pkColumn.bytes;
|
||||
}
|
||||
|
||||
pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + bytes + VARSTR_HEADER_SIZE);
|
||||
for (int32_t j = 0; j < pr->numOfCols; ++j) {
|
||||
int32_t bytes = (slotIds[j] == -1) ? 1 : pr->pSchema->columns[slotIds[j]].bytes;
|
||||
|
||||
pRes[j] = taosMemoryCalloc(1, sizeof(SFirstLastRes) + bytes + pkBufLen + VARSTR_HEADER_SIZE);
|
||||
SFirstLastRes* p = (SFirstLastRes*)varDataVal(pRes[j]);
|
||||
p->ts = INT64_MIN;
|
||||
}
|
||||
|
|
|
@ -53,6 +53,13 @@ SSttBlockLoadInfo *tCreateSttBlockLoadInfo(STSchema *pSchema, int16_t *colList,
|
|||
return pLoadInfo;
|
||||
}
|
||||
|
||||
static void freeItem(void* pValue) {
|
||||
SValue* p = (SValue*) pValue;
|
||||
if (IS_VAR_DATA_TYPE(p->type)) {
|
||||
taosMemoryFree(p->pData);
|
||||
}
|
||||
}
|
||||
|
||||
void *destroySttBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo) {
|
||||
if (pLoadInfo == NULL) {
|
||||
return NULL;
|
||||
|
@ -72,8 +79,8 @@ void *destroySttBlockLoadInfo(SSttBlockLoadInfo *pLoadInfo) {
|
|||
|
||||
if (pLoadInfo->info.pCount != NULL) {
|
||||
taosArrayDestroy(pLoadInfo->info.pUid);
|
||||
taosArrayDestroy(pLoadInfo->info.pFirstKey);
|
||||
taosArrayDestroy(pLoadInfo->info.pLastKey);
|
||||
taosArrayDestroyEx(pLoadInfo->info.pFirstKey, freeItem);
|
||||
taosArrayDestroyEx(pLoadInfo->info.pLastKey, freeItem);
|
||||
taosArrayDestroy(pLoadInfo->info.pCount);
|
||||
taosArrayDestroy(pLoadInfo->info.pFirstTs);
|
||||
taosArrayDestroy(pLoadInfo->info.pLastTs);
|
||||
|
@ -319,6 +326,21 @@ static int32_t extractSttBlockInfo(SLDataIter *pIter, const TSttBlkArray *pArray
|
|||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t tValueDupPayload(SValue *pVal) {
|
||||
if (IS_VAR_DATA_TYPE(pVal->type)) {
|
||||
char *p = (char *)pVal->pData;
|
||||
char *pBuf = taosMemoryMalloc(pVal->nData);
|
||||
if (pBuf == NULL) {
|
||||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
memcpy(pBuf, p, pVal->nData);
|
||||
pVal->pData = (uint8_t *)pBuf;
|
||||
}
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBlockLoadInfo *pBlockLoadInfo,
|
||||
TStatisBlkArray *pStatisBlkArray, uint64_t suid, const char *id) {
|
||||
int32_t numOfBlocks = TARRAY2_SIZE(pStatisBlkArray);
|
||||
|
@ -384,25 +406,16 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
|
|||
break;
|
||||
}
|
||||
|
||||
if (IS_VAR_DATA_TYPE(vFirst.type)) {
|
||||
char *p = (char *)vFirst.pData;
|
||||
char *pBuf = taosMemoryMalloc(vFirst.nData);
|
||||
memcpy(pBuf, p, vFirst.nData);
|
||||
vFirst.pData = (uint8_t *)pBuf;
|
||||
}
|
||||
tValueDupPayload(&vFirst);
|
||||
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &vFirst);
|
||||
|
||||
// todo add api to clone the original data
|
||||
code = tValueColumnGet(&block.lastKeyPKs[0], f, &vLast);
|
||||
if (code) {
|
||||
break;
|
||||
}
|
||||
|
||||
if (IS_VAR_DATA_TYPE(vLast.type)) {
|
||||
char *p = (char *)vLast.pData;
|
||||
char *pBuf = taosMemoryMalloc(vLast.nData);
|
||||
memcpy(pBuf, p, vLast.nData);
|
||||
vLast.pData = (uint8_t *)pBuf;
|
||||
}
|
||||
tValueDupPayload(&vLast);
|
||||
taosArrayPush(pBlockLoadInfo->info.pLastKey, &vLast);
|
||||
}
|
||||
|
||||
|
@ -420,8 +433,15 @@ static int32_t loadSttStatisticsBlockData(SSttFileReader *pSttFileReader, SSttBl
|
|||
taosArrayPush(pBlockLoadInfo->info.pFirstTs, &record.firstKey.ts);
|
||||
taosArrayPush(pBlockLoadInfo->info.pLastTs, &record.lastKey.ts);
|
||||
|
||||
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &record.firstKey.pks[0]);
|
||||
taosArrayPush(pBlockLoadInfo->info.pLastKey, &record.lastKey.pks[0]);
|
||||
SValue s = record.firstKey.pks[0];
|
||||
tValueDupPayload(&s);
|
||||
|
||||
taosArrayPush(pBlockLoadInfo->info.pFirstKey, &s);
|
||||
|
||||
s = record.lastKey.pks[0];
|
||||
tValueDupPayload(&s);
|
||||
|
||||
taosArrayPush(pBlockLoadInfo->info.pLastKey, &s);
|
||||
i += 1;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -382,6 +382,7 @@ typedef struct SCacheRowsReader {
|
|||
SArray* pFuncTypeList;
|
||||
__compar_fn_t pkComparFn;
|
||||
SRowKey rowKey;
|
||||
SColumnInfo pkColumn;
|
||||
} SCacheRowsReader;
|
||||
|
||||
int32_t tsdbCacheGetBatch(STsdb* pTsdb, tb_uid_t uid, SArray* pLastArray, SCacheRowsReader* pr, int8_t ltype);
|
||||
|
|
|
@ -240,10 +240,13 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
|
|||
}
|
||||
|
||||
SSubmitTbData submitTbData;
|
||||
uint8_t version;
|
||||
if (tDecodeI32v(pCoder, &submitTbData.flags) < 0) {
|
||||
code = TSDB_CODE_INVALID_MSG;
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
}
|
||||
version = (submitTbData.flags >> 8) & 0xff;
|
||||
submitTbData.flags = submitTbData.flags & 0xff;
|
||||
|
||||
if (submitTbData.flags & SUBMIT_REQ_FROM_FILE) {
|
||||
code = grantCheck(TSDB_GRANT_CSV);
|
||||
|
@ -307,7 +310,7 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
|
|||
}
|
||||
|
||||
SColData colData = {0};
|
||||
pCoder->pos += tGetColData(pCoder->data + pCoder->pos, &colData);
|
||||
pCoder->pos += tGetColData(version, pCoder->data + pCoder->pos, &colData);
|
||||
if (colData.flag != HAS_VALUE) {
|
||||
code = TSDB_CODE_INVALID_MSG;
|
||||
goto _exit;
|
||||
|
@ -321,7 +324,7 @@ static int32_t vnodePreProcessSubmitTbData(SVnode *pVnode, SDecoder *pCoder, int
|
|||
}
|
||||
|
||||
for (uint64_t i = 1; i < nColData; i++) {
|
||||
pCoder->pos += tGetColData(pCoder->data + pCoder->pos, &colData);
|
||||
pCoder->pos += tGetColData(version, pCoder->data + pCoder->pos, &colData);
|
||||
}
|
||||
} else {
|
||||
uint64_t nRow;
|
||||
|
@ -1572,17 +1575,18 @@ static int32_t vnodeProcessSubmitReq(SVnode *pVnode, int64_t ver, void *pReq, in
|
|||
goto _exit;
|
||||
}
|
||||
|
||||
SColData *pColData = (SColData *)taosArrayGet(pSubmitTbData->aCol, 0);
|
||||
TSKEY *aKey = (TSKEY *)(pColData->pData);
|
||||
|
||||
for (int32_t iRow = 0; iRow < pColData->nVal; iRow++) {
|
||||
if (aKey[iRow] < minKey || aKey[iRow] > maxKey || (iRow > 0 && aKey[iRow] <= aKey[iRow - 1])) {
|
||||
SColData *colDataArr = TARRAY_DATA(pSubmitTbData->aCol);
|
||||
SRowKey lastKey;
|
||||
tColDataArrGetRowKey(colDataArr, TARRAY_SIZE(pSubmitTbData->aCol), 0, &lastKey);
|
||||
for (int32_t iRow = 1; iRow < colDataArr[0].nVal; iRow++) {
|
||||
SRowKey key;
|
||||
tColDataArrGetRowKey(TARRAY_DATA(pSubmitTbData->aCol), TARRAY_SIZE(pSubmitTbData->aCol), iRow, &key);
|
||||
if (tRowKeyCompare(&lastKey, &key) >= 0) {
|
||||
code = TSDB_CODE_INVALID_MSG;
|
||||
vError("vgId:%d %s failed since %s, version:%" PRId64, TD_VID(pVnode), __func__, tstrerror(terrno), ver);
|
||||
goto _exit;
|
||||
}
|
||||
}
|
||||
|
||||
} else {
|
||||
int32_t nRow = TARRAY_SIZE(pSubmitTbData->aRowP);
|
||||
SRow **aRow = (SRow **)TARRAY_DATA(pSubmitTbData->aRowP);
|
||||
|
|
|
@ -40,7 +40,7 @@ typedef struct SCacheRowsScanInfo {
|
|||
SExprSupp pseudoExprSup;
|
||||
int32_t retrieveType;
|
||||
int32_t currentGroupIndex;
|
||||
SSDataBlock* pBufferredRes;
|
||||
SSDataBlock* pBufferedRes;
|
||||
SArray* pUidList;
|
||||
SArray* pCidList;
|
||||
int32_t indexOfBufferedRes;
|
||||
|
@ -160,9 +160,9 @@ SOperatorInfo* createCacherowsScanOperator(SLastRowScanPhysiNode* pScanNode, SRe
|
|||
|
||||
capacity = TMIN(totalTables, 4096);
|
||||
|
||||
pInfo->pBufferredRes = createOneDataBlock(pInfo->pRes, false);
|
||||
setColIdForCacheReadBlock(pInfo->pBufferredRes, pScanNode);
|
||||
blockDataEnsureCapacity(pInfo->pBufferredRes, capacity);
|
||||
pInfo->pBufferedRes = createOneDataBlock(pInfo->pRes, false);
|
||||
setColIdForCacheReadBlock(pInfo->pBufferedRes, pScanNode);
|
||||
blockDataEnsureCapacity(pInfo->pBufferedRes, capacity);
|
||||
} else { // by tags
|
||||
pInfo->retrieveType = CACHESCAN_RETRIEVE_TYPE_SINGLE | SCAN_ROW_TYPE(pScanNode->ignoreNull);
|
||||
capacity = 1; // only one row output
|
||||
|
@ -219,18 +219,18 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
|
|||
T_LONG_JMP(pTaskInfo->env, pTaskInfo->code);
|
||||
}
|
||||
|
||||
if (pInfo->indexOfBufferedRes >= pInfo->pBufferredRes->info.rows) {
|
||||
blockDataCleanup(pInfo->pBufferredRes);
|
||||
if (pInfo->indexOfBufferedRes >= pInfo->pBufferedRes->info.rows) {
|
||||
blockDataCleanup(pInfo->pBufferedRes);
|
||||
taosArrayClear(pInfo->pUidList);
|
||||
|
||||
int32_t code = pInfo->readHandle.api.cacheFn.retrieveRows(pInfo->pLastrowReader, pInfo->pBufferredRes, pInfo->pSlotIds,
|
||||
int32_t code = pInfo->readHandle.api.cacheFn.retrieveRows(pInfo->pLastrowReader, pInfo->pBufferedRes, pInfo->pSlotIds,
|
||||
pInfo->pDstSlotIds, pInfo->pUidList);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
T_LONG_JMP(pTaskInfo->env, code);
|
||||
}
|
||||
|
||||
// check for tag values
|
||||
int32_t resultRows = pInfo->pBufferredRes->info.rows;
|
||||
int32_t resultRows = pInfo->pBufferedRes->info.rows;
|
||||
|
||||
// the results may be null, if last values are all null
|
||||
ASSERT(resultRows == 0 || resultRows == taosArrayGetSize(pInfo->pUidList));
|
||||
|
@ -239,12 +239,12 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
|
|||
|
||||
SSDataBlock* pRes = pInfo->pRes;
|
||||
|
||||
if (pInfo->indexOfBufferedRes < pInfo->pBufferredRes->info.rows) {
|
||||
for (int32_t i = 0; i < taosArrayGetSize(pInfo->pBufferredRes->pDataBlock); ++i) {
|
||||
if (pInfo->indexOfBufferedRes < pInfo->pBufferedRes->info.rows) {
|
||||
for (int32_t i = 0; i < taosArrayGetSize(pInfo->pBufferedRes->pDataBlock); ++i) {
|
||||
SColumnInfoData* pCol = taosArrayGet(pRes->pDataBlock, i);
|
||||
int32_t slotId = pCol->info.slotId;
|
||||
|
||||
SColumnInfoData* pSrc = taosArrayGet(pInfo->pBufferredRes->pDataBlock, slotId);
|
||||
SColumnInfoData* pSrc = taosArrayGet(pInfo->pBufferedRes->pDataBlock, slotId);
|
||||
SColumnInfoData* pDst = taosArrayGet(pRes->pDataBlock, slotId);
|
||||
|
||||
if (colDataIsNull_s(pSrc, pInfo->indexOfBufferedRes)) {
|
||||
|
@ -350,7 +350,7 @@ SSDataBlock* doScanCache(SOperatorInfo* pOperator) {
|
|||
void destroyCacheScanOperator(void* param) {
|
||||
SCacheRowsScanInfo* pInfo = (SCacheRowsScanInfo*)param;
|
||||
blockDataDestroy(pInfo->pRes);
|
||||
blockDataDestroy(pInfo->pBufferredRes);
|
||||
blockDataDestroy(pInfo->pBufferedRes);
|
||||
taosMemoryFree(pInfo->pSlotIds);
|
||||
taosMemoryFree(pInfo->pDstSlotIds);
|
||||
taosArrayDestroy(pInfo->pCidList);
|
||||
|
|
|
@ -2018,9 +2018,17 @@ static int32_t translateCast(SFunctionNode* pFunc, char* pErrBuf, int32_t len) {
|
|||
if (IS_STR_DATA_TYPE(para2Type)) {
|
||||
para2Bytes -= VARSTR_HEADER_SIZE;
|
||||
}
|
||||
if (para2Bytes <= 0 || para2Bytes > 4096) { // cast dst var type length limits to 4096 bytes
|
||||
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
|
||||
"CAST function converted length should be in range (0, 4096] bytes");
|
||||
if (para2Bytes <= 0 ||
|
||||
para2Bytes > TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE) { // cast dst var type length limits to 4096 bytes
|
||||
if (TSDB_DATA_TYPE_NCHAR == para2Type) {
|
||||
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
|
||||
"CAST function converted length should be in range (0, %d] NCHARS",
|
||||
(TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE)/TSDB_NCHAR_SIZE);
|
||||
} else {
|
||||
return buildFuncErrMsg(pErrBuf, len, TSDB_CODE_FUNC_FUNTION_ERROR,
|
||||
"CAST function converted length should be in range (0, %d] bytes",
|
||||
TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE);
|
||||
}
|
||||
}
|
||||
|
||||
// add database precision as param
|
||||
|
|
|
@ -416,6 +416,13 @@ type_name(A) ::= DECIMAL.
|
|||
type_name(A) ::= DECIMAL NK_LP NK_INTEGER NK_RP. { A = createDataType(TSDB_DATA_TYPE_DECIMAL); }
|
||||
type_name(A) ::= DECIMAL NK_LP NK_INTEGER NK_COMMA NK_INTEGER NK_RP. { A = createDataType(TSDB_DATA_TYPE_DECIMAL); }
|
||||
|
||||
%type type_name_default_len { SDataType }
|
||||
%destructor type_name_default_len { }
|
||||
type_name_default_len(A) ::= BINARY. { A = createVarLenDataType(TSDB_DATA_TYPE_BINARY, NULL); }
|
||||
type_name_default_len(A) ::= NCHAR. { A = createVarLenDataType(TSDB_DATA_TYPE_NCHAR, NULL); }
|
||||
type_name_default_len(A) ::= VARCHAR. { A = createVarLenDataType(TSDB_DATA_TYPE_VARCHAR, NULL); }
|
||||
type_name_default_len(A) ::= VARBINARY. { A = createVarLenDataType(TSDB_DATA_TYPE_VARBINARY, NULL); }
|
||||
|
||||
%type tags_def_opt { SNodeList* }
|
||||
%destructor tags_def_opt { nodesDestroyList($$); }
|
||||
tags_def_opt(A) ::= . { A = NULL; }
|
||||
|
@ -1122,6 +1129,9 @@ function_expression(A) ::= function_name(B) NK_LP expression_list(C) NK_RP(D).
|
|||
function_expression(A) ::= star_func(B) NK_LP star_func_para_list(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, C)); }
|
||||
function_expression(A) ::=
|
||||
CAST(B) NK_LP expr_or_subquery(C) AS type_name(D) NK_RP(E). { A = createRawExprNodeExt(pCxt, &B, &E, createCastFunctionNode(pCxt, releaseRawExprNode(pCxt, C), D)); }
|
||||
function_expression(A) ::=
|
||||
CAST(B) NK_LP expr_or_subquery(C) AS type_name_default_len(D) NK_RP(E). { A = createRawExprNodeExt(pCxt, &B, &E, createCastFunctionNode(pCxt, releaseRawExprNode(pCxt, C), D)); }
|
||||
|
||||
function_expression(A) ::= literal_func(B). { A = B; }
|
||||
|
||||
literal_func(A) ::= noarg_func(B) NK_LP NK_RP(C). { A = createRawExprNodeExt(pCxt, &B, &C, createFunctionNode(pCxt, &B, NULL)); }
|
||||
|
|
|
@ -1602,7 +1602,10 @@ SDataType createDataType(uint8_t type) {
|
|||
}
|
||||
|
||||
SDataType createVarLenDataType(uint8_t type, const SToken* pLen) {
|
||||
SDataType dt = {.type = type, .precision = 0, .scale = 0, .bytes = taosStr2Int32(pLen->z, NULL, 10)};
|
||||
int32_t len = TSDB_MAX_BINARY_LEN - VARSTR_HEADER_SIZE;
|
||||
if (type == TSDB_DATA_TYPE_NCHAR) len /= TSDB_NCHAR_SIZE;
|
||||
if(pLen) len = taosStr2Int32(pLen->z, NULL, 10);
|
||||
SDataType dt = {.type = type, .precision = 0, .scale = 0, .bytes = len};
|
||||
return dt;
|
||||
}
|
||||
|
||||
|
|
File diff suppressed because it is too large
Load Diff
|
@ -70,7 +70,7 @@ int32_t transDecompressMsg(char** msg, int32_t len) {
|
|||
char* buf = taosMemoryCalloc(1, oriLen + sizeof(STransMsgHead));
|
||||
STransMsgHead* pNewHead = (STransMsgHead*)buf;
|
||||
int32_t decompLen = LZ4_decompress_safe(pCont + sizeof(STransCompMsg), (char*)pNewHead->content,
|
||||
len - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen);
|
||||
len - sizeof(STransMsgHead) - sizeof(STransCompMsg), oriLen);
|
||||
memcpy((char*)pNewHead, (char*)pHead, sizeof(STransMsgHead));
|
||||
|
||||
pNewHead->msgLen = htonl(oriLen + sizeof(STransMsgHead));
|
||||
|
@ -158,6 +158,10 @@ int transResetBuffer(SConnBuffer* connBuf) {
|
|||
p->left = -1;
|
||||
p->total = 0;
|
||||
p->len = 0;
|
||||
if (p->cap > BUFFER_CAP) {
|
||||
p->cap = BUFFER_CAP;
|
||||
p->buf = taosMemoryRealloc(p->buf, p->cap);
|
||||
}
|
||||
} else {
|
||||
ASSERTS(0, "invalid read from sock buf");
|
||||
return -1;
|
||||
|
|
|
@ -242,6 +242,7 @@ void taosCloseLog() {
|
|||
taosMemoryFreeClear(tsLogObj.logHandle->buffer);
|
||||
taosThreadMutexDestroy(&tsLogObj.logMutex);
|
||||
taosMemoryFreeClear(tsLogObj.logHandle);
|
||||
tsLogObj.logHandle = NULL;
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -285,8 +286,11 @@ static void taosKeepOldLog(char *oldName) {
|
|||
taosRemoveOldFiles(tsLogDir, tsLogKeepDays);
|
||||
}
|
||||
}
|
||||
|
||||
static void *taosThreadToOpenNewFile(void *param) {
|
||||
typedef struct {
|
||||
TdFilePtr pOldFile;
|
||||
char keepName[LOG_FILE_NAME_LEN + 20];
|
||||
} OldFileKeeper;
|
||||
static OldFileKeeper *taosOpenNewFile() {
|
||||
char keepName[LOG_FILE_NAME_LEN + 20];
|
||||
sprintf(keepName, "%s.%d", tsLogObj.logName, tsLogObj.flag);
|
||||
|
||||
|
@ -312,13 +316,26 @@ static void *taosThreadToOpenNewFile(void *param) {
|
|||
tsLogObj.logHandle->pFile = pFile;
|
||||
tsLogObj.lines = 0;
|
||||
tsLogObj.openInProgress = 0;
|
||||
taosSsleep(20);
|
||||
taosCloseLogByFd(pOldFile);
|
||||
OldFileKeeper* oldFileKeeper = taosMemoryMalloc(sizeof(OldFileKeeper));
|
||||
if (oldFileKeeper == NULL) {
|
||||
uError("create old log keep info faild! mem is not enough.");
|
||||
return NULL;
|
||||
}
|
||||
oldFileKeeper->pOldFile = pOldFile;
|
||||
memcpy(oldFileKeeper->keepName, keepName, LOG_FILE_NAME_LEN + 20);
|
||||
|
||||
uInfo(" new log file:%d is opened", tsLogObj.flag);
|
||||
uInfo("==================================");
|
||||
taosKeepOldLog(keepName);
|
||||
return oldFileKeeper;
|
||||
}
|
||||
|
||||
static void *taosThreadToCloseOldFile(void* param) {
|
||||
if(!param) return NULL;
|
||||
OldFileKeeper* oldFileKeeper = (OldFileKeeper*)param;
|
||||
taosSsleep(20);
|
||||
taosCloseLogByFd(oldFileKeeper->pOldFile);
|
||||
taosKeepOldLog(oldFileKeeper->keepName);
|
||||
taosMemoryFree(oldFileKeeper);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
|
@ -334,7 +351,8 @@ static int32_t taosOpenNewLogFile() {
|
|||
taosThreadAttrInit(&attr);
|
||||
taosThreadAttrSetDetachState(&attr, PTHREAD_CREATE_DETACHED);
|
||||
|
||||
taosThreadCreate(&thread, &attr, taosThreadToOpenNewFile, NULL);
|
||||
OldFileKeeper* oldFileKeeper = taosOpenNewFile();
|
||||
taosThreadCreate(&thread, &attr, taosThreadToCloseOldFile, oldFileKeeper);
|
||||
taosThreadAttrDestroy(&attr);
|
||||
}
|
||||
|
||||
|
@ -347,10 +365,11 @@ void taosResetLog() {
|
|||
// force create a new log file
|
||||
tsLogObj.lines = tsNumOfLogLines + 10;
|
||||
|
||||
taosOpenNewLogFile();
|
||||
|
||||
uInfo("==================================");
|
||||
uInfo(" reset log file ");
|
||||
if (tsLogObj.logHandle) {
|
||||
taosOpenNewLogFile();
|
||||
uInfo("==================================");
|
||||
uInfo(" reset log file ");
|
||||
}
|
||||
}
|
||||
|
||||
static bool taosCheckFileIsOpen(char *logFileName) {
|
||||
|
|
|
@ -79,6 +79,10 @@ class TDTestCase:
|
|||
tdSql.query(f"select cast(c1 as binary(32)) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c1)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c1[i]) )
|
||||
|
||||
tdSql.query(f"select cast(c1 as binary) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c1)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c1[i]) )
|
||||
|
||||
tdLog.printNoPrefix("==========step6: cast int to nchar, expect changes to str(int) ")
|
||||
|
||||
|
@ -130,6 +134,13 @@ class TDTestCase:
|
|||
tdSql.query(f"select cast(c2 as binary(32)) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c2)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c2[i]) )
|
||||
|
||||
tdSql.query(f"select cast(c2 as binary) as b from {self.dbname}.ct4")
|
||||
for i in range(len(data_ct4_c2)):
|
||||
tdSql.checkData( i, 0, str(data_ct4_c2[i]) )
|
||||
tdSql.query(f"select cast(c2 as binary) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c2)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c2[i]) )
|
||||
|
||||
tdLog.printNoPrefix("==========step10: cast bigint to nchar, expect changes to str(int) ")
|
||||
|
||||
|
@ -184,6 +195,13 @@ class TDTestCase:
|
|||
tdSql.query(f"select cast(c3 as binary(32)) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c3)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c3[i]) )
|
||||
|
||||
tdSql.query(f"select cast(c3 as binary) as b from {self.dbname}.ct4")
|
||||
for i in range(len(data_ct4_c3)):
|
||||
tdSql.checkData( i, 0, str(data_ct4_c3[i]) )
|
||||
tdSql.query(f"select cast(c3 as binary) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c3)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c3[i]) )
|
||||
|
||||
tdLog.printNoPrefix("==========step14: cast smallint to nchar, expect changes to str(int) ")
|
||||
|
||||
|
@ -235,6 +253,13 @@ class TDTestCase:
|
|||
tdSql.query(f"select cast(c4 as binary(32)) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c4)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c4[i]) )
|
||||
|
||||
tdSql.query(f"select cast(c4 as binary) as b from {self.dbname}.ct4")
|
||||
for i in range(len(data_ct4_c4)):
|
||||
tdSql.checkData( i, 0, str(data_ct4_c4[i]) )
|
||||
tdSql.query(f"select cast(c4 as binary) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c4)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c4[i]) )
|
||||
|
||||
tdLog.printNoPrefix("==========step18: cast tinyint to nchar, expect changes to str(int) ")
|
||||
|
||||
|
@ -282,6 +307,12 @@ class TDTestCase:
|
|||
for i in range(len(data_ct4_c5)):
|
||||
tdSql.checkData( i, 0, str(data_ct4_c5[i]) ) if data_ct4_c5[i] is None else tdSql.checkData( i, 0, f'{data_ct4_c5[i]:.6f}' )
|
||||
tdSql.query(f"select cast(c5 as binary(32)) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c5)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c5[i]) ) if data_t1_c5[i] is None else tdSql.checkData( i, 0, f'{data_t1_c5[i]:.6f}' )
|
||||
tdSql.query(f"select cast(c5 as binary) as b from {self.dbname}.ct4")
|
||||
for i in range(len(data_ct4_c5)):
|
||||
tdSql.checkData( i, 0, str(data_ct4_c5[i]) ) if data_ct4_c5[i] is None else tdSql.checkData( i, 0, f'{data_ct4_c5[i]:.6f}' )
|
||||
tdSql.query(f"select cast(c5 as binary) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c5)):
|
||||
tdSql.checkData( i, 0, str(data_t1_c5[i]) ) if data_t1_c5[i] is None else tdSql.checkData( i, 0, f'{data_t1_c5[i]:.6f}' )
|
||||
|
||||
|
@ -290,6 +321,12 @@ class TDTestCase:
|
|||
for i in range(len(data_ct4_c5)):
|
||||
tdSql.checkData( i, 0, None ) if data_ct4_c5[i] is None else tdSql.checkData( i, 0, f'{data_ct4_c5[i]:.6f}' )
|
||||
tdSql.query(f"select cast(c5 as nchar(32)) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c5)):
|
||||
tdSql.checkData( i, 0, None ) if data_t1_c5[i] is None else tdSql.checkData( i, 0, f'{data_t1_c5[i]:.6f}' )
|
||||
tdSql.query(f"select cast(c5 as nchar) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c5)):
|
||||
tdSql.checkData( i, 0, None ) if data_t1_c5[i] is None else tdSql.checkData( i, 0, f'{data_t1_c5[i]:.6f}' )
|
||||
tdSql.query(f"select cast(c5 as varchar) as b from {self.dbname}.t1")
|
||||
for i in range(len(data_t1_c5)):
|
||||
tdSql.checkData( i, 0, None ) if data_t1_c5[i] is None else tdSql.checkData( i, 0, f'{data_t1_c5[i]:.6f}' )
|
||||
|
||||
|
@ -580,6 +617,10 @@ class TDTestCase:
|
|||
( tdSql.checkData(i, 0, '12121.233231') for i in range(tdSql.queryRows) )
|
||||
tdSql.query(f"select cast(12121.23323131 + 'test~!@`#$%^&*(){'}'}{'{'}][;><.,' as binary(2)) as b from {self.dbname}.ct4")
|
||||
( tdSql.checkData(i, 0, '12') for i in range(tdSql.queryRows) )
|
||||
tdSql.query(f"select cast(12121.23323131 + 'test~!@`#$%^&*(){'}'}{'{'}][;><.,' as binary) as b from {self.dbname}.ct4")
|
||||
( tdSql.checkData(i, 0, '12121.233231') for i in range(tdSql.queryRows) )
|
||||
tdSql.query(f"select cast(12121.23323131 + 'test~!@`#$%^&*(){'}'}{'{'}][;><.,' as binary) as b from {self.dbname}.ct4")
|
||||
( tdSql.checkData(i, 0, '12') for i in range(tdSql.queryRows) )
|
||||
tdSql.query(f"select cast(12121.23323131 + 'test~!@`#$%^&*(){'}'}{'{'}][;><.,' as nchar(16)) as b from {self.dbname}.ct4")
|
||||
( tdSql.checkData(i, 0, '12121.233231') for i in range(tdSql.queryRows) )
|
||||
tdSql.query(f"select cast(12121.23323131 + 'test~!@`#$%^&*(){'}'}{'{'}][;><.,' as nchar(2)) as b from {self.dbname}.ct4")
|
||||
|
|
Loading…
Reference in New Issue