提交 097a047b 编写于 作者: S Shengliang Guan

Merge remote-tracking branch 'origin/3.0' into FIX/TD-19593-3.0

...@@ -337,8 +337,6 @@ typedef struct tDataTypeDescriptor { ...@@ -337,8 +337,6 @@ typedef struct tDataTypeDescriptor {
int32_t nBuf); int32_t nBuf);
int32_t (*decompFunc)(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf, int32_t (*decompFunc)(void *pIn, int32_t nIn, int32_t nEle, void *pOut, int32_t nOut, uint8_t cmprAlg, void *pBuf,
int32_t nBuf); int32_t nBuf);
void (*statisFunc)(int8_t bitmapMode, const void *pBitmap, const void *pData, int32_t numofrow, int64_t *min,
int64_t *max, int64_t *sum, int16_t *minindex, int16_t *maxindex, int16_t *numofnull);
} tDataTypeDescriptor; } tDataTypeDescriptor;
extern tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX]; extern tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX];
...@@ -346,12 +344,9 @@ extern tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX]; ...@@ -346,12 +344,9 @@ extern tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX];
bool isValidDataType(int32_t type); bool isValidDataType(int32_t type);
void setVardataNull(void *val, int32_t type); void setVardataNull(void *val, int32_t type);
void setNull(void *val, int32_t type, int32_t bytes); //void setNull(void *val, int32_t type, int32_t bytes);
void setNullN(void *val, int32_t type, int32_t bytes, int32_t numOfElems); //void setNullN(void *val, int32_t type, int32_t bytes, int32_t numOfElems);
const void *getNullValue(int32_t type);
void assignVal(char *val, const char *src, int32_t len, int32_t type); void assignVal(char *val, const char *src, int32_t len, int32_t type);
void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size, void *buf);
void operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type); void operateVal(void *dst, void *s1, void *s2, int32_t optr, int32_t type);
void *getDataMin(int32_t type); void *getDataMin(int32_t type);
void *getDataMax(int32_t type); void *getDataMax(int32_t type);
......
...@@ -39,8 +39,6 @@ typedef struct SVariant { ...@@ -39,8 +39,6 @@ typedef struct SVariant {
int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value); int32_t toInteger(const char *z, int32_t n, int32_t base, int64_t *value);
int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value); int32_t toUInteger(const char *z, int32_t n, int32_t base, uint64_t *value);
bool taosVariantIsValid(SVariant *pVar);
void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type); void taosVariantCreateFromBinary(SVariant *pVar, const char *pz, size_t len, uint32_t type);
void taosVariantDestroy(SVariant *pV); void taosVariantDestroy(SVariant *pV);
...@@ -49,15 +47,6 @@ void taosVariantAssign(SVariant *pDst, const SVariant *pSrc); ...@@ -49,15 +47,6 @@ void taosVariantAssign(SVariant *pDst, const SVariant *pSrc);
int32_t taosVariantCompare(const SVariant *p1, const SVariant *p2); int32_t taosVariantCompare(const SVariant *p1, const SVariant *p2);
int32_t taosVariantToString(SVariant *pVar, char *dst);
int32_t taosVariantDump(SVariant *pVariant, char *payload, int16_t type, bool includeLengthPrefix);
#if 0
int32_t taosVariantDumpEx(SVariant *pVariant, char *payload, int16_t type, bool includeLengthPrefix, bool *converted, char *extInfo);
#endif
int32_t taosVariantTypeSetType(SVariant *pVariant, char type);
char *taosVariantGet(SVariant *pVar, int32_t type); char *taosVariantGet(SVariant *pVar, int32_t type);
#ifdef __cplusplus #ifdef __cplusplus
......
...@@ -114,7 +114,7 @@ static void setBitMap(uint8_t *pb, uint8_t v, int32_t idx, uint8_t flags) { ...@@ -114,7 +114,7 @@ static void setBitMap(uint8_t *pb, uint8_t v, int32_t idx, uint8_t flags) {
((uint32_t *)(p))[i] = (n); \ ((uint32_t *)(p))[i] = (n); \
} \ } \
} while (0) } while (0)
#if 0
int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) { int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow) {
int32_t code = 0; int32_t code = 0;
#if 0 #if 0
...@@ -432,7 +432,6 @@ void tTSRowFree(STSRow2 *pRow) { ...@@ -432,7 +432,6 @@ void tTSRowFree(STSRow2 *pRow) {
} }
void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) { void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal) {
#if 0
uint8_t isTuple = ((pRow->flags & 0xf0) == 0) ? 1 : 0; uint8_t isTuple = ((pRow->flags & 0xf0) == 0) ? 1 : 0;
STColumn *pTColumn = &pTSchema->columns[iCol]; STColumn *pTColumn = &pTSchema->columns[iCol];
uint8_t flags = pRow->flags & (uint8_t)0xf; uint8_t flags = pRow->flags & (uint8_t)0xf;
...@@ -577,12 +576,10 @@ _return_null: ...@@ -577,12 +576,10 @@ _return_null:
_return_value: _return_value:
*pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, value); *pColVal = COL_VAL_VALUE(pTColumn->colId, pTColumn->type, value);
return; return;
#endif
} }
int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) { int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) {
int32_t code = 0; int32_t code = 0;
#if 0
SColVal cv; SColVal cv;
(*ppArray) = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal)); (*ppArray) = taosArrayInit(pTSchema->numOfCols, sizeof(SColVal));
...@@ -596,11 +593,11 @@ int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) { ...@@ -596,11 +593,11 @@ int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray) {
taosArrayPush(*ppArray, &cv); taosArrayPush(*ppArray, &cv);
} }
#endif
_exit: _exit:
return code; return code;
} }
#endif
int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) { int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow) {
int32_t n = 0; int32_t n = 0;
......
...@@ -41,286 +41,6 @@ static bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCe ...@@ -41,286 +41,6 @@ static bool tdSKvRowGetVal(STSRow *pRow, col_id_t colId, col_id_t colIdx, SCe
static void tdSCellValPrint(SCellVal *pVal, int8_t colType); static void tdSCellValPrint(SCellVal *pVal, int8_t colType);
// implementation // implementation
/**
* @brief Compress bitmap bytes comprised of 2-bits to counterpart of 1-bit.
* e.g.
* TD_VTYPE_NORM 0x00U(00000000) to 00000000 Normal
* TD_VTYPE_NULL 0x01U(00000001) to 00000001 Null
* TD_VTYPE_NONE 0x02U(00000010) to 00000001 Null
*
* 00000000 0x00 0x00
* 01000000 0x40 0x08
* 10000000 0x80 0x08
* ...
* @param byte
* @return uint8_t
*/
static uint8_t tdGetMergedBitmapByte(uint8_t byte) {
switch (byte) {
case 0x00:
return 0x00;
case 0x40:
return 0x08;
case 0x80:
return 0x08;
case 0x10:
return 0x04;
case 0x50:
return 0x0c;
case 0x90:
return 0x0c;
case 0x20:
return 0x04;
case 0x60:
return 0x0c;
case 0xa0:
return 0x0c;
case 0x04:
return 0x02;
case 0x44:
return 0x0a;
case 0x84:
return 0x0a;
case 0x14:
return 0x06;
case 0x54:
return 0x0e;
case 0x94:
return 0x0e;
case 0x24:
return 0x06;
case 0x64:
return 0x0e;
case 0xa4:
return 0x0e;
case 0x08:
return 0x02;
case 0x48:
return 0x0a;
case 0x88:
return 0x0a;
case 0x18:
return 0x06;
case 0x58:
return 0x0e;
case 0x98:
return 0x0e;
case 0x28:
return 0x06;
case 0x68:
return 0x0e;
case 0xa8:
return 0x0e;
case 0x01:
return 0x01;
case 0x41:
return 0x09;
case 0x81:
return 0x09;
case 0x11:
return 0x05;
case 0x51:
return 0x0d;
case 0x91:
return 0x0d;
case 0x21:
return 0x05;
case 0x61:
return 0x0d;
case 0xa1:
return 0x0d;
case 0x05:
return 0x03;
case 0x45:
return 0x0b;
case 0x85:
return 0x0b;
case 0x15:
return 0x07;
case 0x55:
return 0x0f;
case 0x95:
return 0x0f;
case 0x25:
return 0x07;
case 0x65:
return 0x0f;
case 0xa5:
return 0x0f;
case 0x09:
return 0x03;
case 0x49:
return 0x0b;
case 0x89:
return 0x0b;
case 0x19:
return 0x07;
case 0x59:
return 0x0f;
case 0x99:
return 0x0f;
case 0x29:
return 0x07;
case 0x69:
return 0x0f;
case 0xa9:
return 0x0f;
case 0x02:
return 0x01;
case 0x42:
return 0x09;
case 0x82:
return 0x09;
case 0x12:
return 0x05;
case 0x52:
return 0x0d;
case 0x92:
return 0x0d;
case 0x22:
return 0x05;
case 0x62:
return 0x0d;
case 0xa2:
return 0x0d;
case 0x06:
return 0x03;
case 0x46:
return 0x0b;
case 0x86:
return 0x0b;
case 0x16:
return 0x07;
case 0x56:
return 0x0f;
case 0x96:
return 0x0f;
case 0x26:
return 0x07;
case 0x66:
return 0x0f;
case 0xa6:
return 0x0f;
case 0x0a:
return 0x03;
case 0x4a:
return 0x0b;
case 0x8a:
return 0x0b;
case 0x1a:
return 0x07;
case 0x5a:
return 0x0f;
case 0x9a:
return 0x0f;
case 0x2a:
return 0x07;
case 0x6a:
return 0x0f;
case 0xaa:
return 0x0f;
default:
// make sure the bitmap area is set to 0 firstly
ASSERT(0);
return 0x0f; // return NULL bitmap for exception
}
}
/**
* @brief Merge bitmap from 2 bits to 1 bit, and the memory buffer should be guaranteed by the invoker.
*
* @param srcBitmap
* @param nBits
* @param dstBitmap
*/
void tdMergeBitmap(uint8_t *srcBitmap, int32_t nBits, uint8_t *dstBitmap) {
int32_t i = 0, j = 0;
int32_t nBytes = TD_BITMAP_BYTES(nBits);
int32_t nRoundBytes = nBits / 4;
int32_t nRemainderBits = nBits - nRoundBytes * 4;
switch (nRemainderBits) {
case 0:
// NOTHING TODO
break;
case 1: {
void *lastByte = POINTER_SHIFT(srcBitmap, nRoundBytes);
*(uint8_t *)lastByte &= 0xC0;
} break;
case 2: {
void *lastByte = POINTER_SHIFT(srcBitmap, nRoundBytes);
*(uint8_t *)lastByte &= 0xF0;
} break;
case 3: {
void *lastByte = POINTER_SHIFT(srcBitmap, nRoundBytes);
*(uint8_t *)lastByte &= 0xFC;
} break;
default:
ASSERT(0);
}
if (nBytes > 0) {
dstBitmap[j] = (tdGetMergedBitmapByte(srcBitmap[i]) << 4);
}
while ((++i) < nBytes) {
if ((i & 1) == 0) {
dstBitmap[j] = (tdGetMergedBitmapByte(srcBitmap[i]) << 4);
} else {
dstBitmap[j] |= tdGetMergedBitmapByte(srcBitmap[i]);
++j;
}
}
}
/**
* @brief Set bitmap area by byte preferentially and then by bit.
*
* @param pBitmap
* @param nEle
* @param valType
* @param bitmapMode 0 for 2 bits, 1 for 1 bit
* @return int32_t
*/
int32_t tdSetBitmapValTypeN(void *pBitmap, int16_t nEle, TDRowValT valType, int8_t bitmapMode) {
TASSERT(valType < TD_VTYPE_MAX);
int32_t nBytes = (bitmapMode == 0 ? nEle / TD_VTYPE_PARTS : nEle / TD_VTYPE_PARTS_I);
uint8_t vTypeByte = tdVTypeByte[bitmapMode][valType];
for (int i = 0; i < nBytes; ++i) {
*(uint8_t *)pBitmap = vTypeByte;
pBitmap = POINTER_SHIFT(pBitmap, 1);
}
int32_t nLeft = nEle - nBytes * (bitmapMode == 0 ? TD_VTYPE_BITS : TD_VTYPE_BITS_I);
for (int j = 0; j < nLeft; ++j) {
tdSetBitmapValType(pBitmap, j, valType, bitmapMode);
}
return TSDB_CODE_SUCCESS;
}
bool tdIsBitmapBlkNorm(const void *pBitmap, int32_t numOfBits, int8_t bitmapMode) {
int32_t nBytes = (bitmapMode == 0 ? numOfBits / TD_VTYPE_PARTS : numOfBits / TD_VTYPE_PARTS_I);
uint8_t vTypeByte = tdVTypeByte[bitmapMode][TD_VTYPE_NORM];
uint8_t *qBitmap = (uint8_t *)pBitmap;
for (int i = 0; i < nBytes; ++i) {
if (*qBitmap != vTypeByte) {
return false;
}
qBitmap = (uint8_t *)POINTER_SHIFT(pBitmap, i);
}
int32_t nLeft = numOfBits - nBytes * (bitmapMode == 0 ? TD_VTYPE_BITS : TD_VTYPE_BITS_I);
for (int j = 0; j < nLeft; ++j) {
uint8_t vType;
tdGetBitmapValType(qBitmap, j, &vType, bitmapMode);
if (vType != TD_VTYPE_NORM) {
return false;
}
}
return true;
}
STSRow *tdRowDup(STSRow *row) { STSRow *tdRowDup(STSRow *row) {
STSRow *trow = taosMemoryMalloc(TD_ROW_LEN(row)); STSRow *trow = taosMemoryMalloc(TD_ROW_LEN(row));
if (trow == NULL) return NULL; if (trow == NULL) return NULL;
......
此差异已折叠。
此差异已折叠。
...@@ -687,7 +687,12 @@ int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sv ...@@ -687,7 +687,12 @@ int32_t metaGetTbTSchemaEx(SMeta *pMeta, tb_uid_t suid, tb_uid_t uid, int32_t sv
SSchema *pSchema = pSchemaWrapper->pSchema + i; SSchema *pSchema = pSchemaWrapper->pSchema + i;
tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes); tdAddColToSchema(&sb, pSchema->type, pSchema->flags, pSchema->colId, pSchema->bytes);
} }
STSchema *pTSchema = tdGetSchemaFromBuilder(&sb); STSchema *pTSchema = tdGetSchemaFromBuilder(&sb);
if (pTSchema == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
tdDestroyTSchemaBuilder(&sb); tdDestroyTSchemaBuilder(&sb);
*ppTSchema = pTSchema; *ppTSchema = pTSchema;
......
...@@ -84,6 +84,7 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, SArray* pTableIdList ...@@ -84,6 +84,7 @@ int32_t tsdbCacherowsReaderOpen(void* pVnode, int32_t type, SArray* pTableIdList
p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES); p->transferBuf = taosMemoryCalloc(p->pSchema->numOfCols, POINTER_BYTES);
if (p->transferBuf == NULL) { if (p->transferBuf == NULL) {
tsdbCacherowsReaderClose(p);
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
......
...@@ -2554,6 +2554,12 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) { ...@@ -2554,6 +2554,12 @@ static int32_t doBuildDataBlock(STsdbReader* pReader) {
pScanInfo = pReader->status.pTableIter; pScanInfo = pReader->status.pTableIter;
} }
if (pScanInfo == NULL) {
tsdbError("failed to get table, uid:"PRIu64 ", %s", pBlockInfo->uid, pReader->idStr);
code = TSDB_CODE_INVALID_PARA;
return code;
}
if (pBlockInfo != NULL) { if (pBlockInfo != NULL) {
pBlock = getCurrentBlock(pBlockIter); pBlock = getCurrentBlock(pBlockIter);
} }
......
...@@ -250,6 +250,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo ...@@ -250,6 +250,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
qDebug(" %d qualified child tables added into stream scanner", (int32_t)taosArrayGetSize(qa)); qDebug(" %d qualified child tables added into stream scanner", (int32_t)taosArrayGetSize(qa));
code = tqReaderAddTbUidList(pScanInfo->tqReader, qa); code = tqReaderAddTbUidList(pScanInfo->tqReader, qa);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosArrayDestroy(qa);
return code; return code;
} }
...@@ -261,6 +262,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo ...@@ -261,6 +262,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
assignUid = groupbyTbname(pScanInfo->pGroupTags); assignUid = groupbyTbname(pScanInfo->pGroupTags);
keyBuf = taosMemoryMalloc(bufLen); keyBuf = taosMemoryMalloc(bufLen);
if (keyBuf == NULL) { if (keyBuf == NULL) {
taosArrayDestroy(qa);
return TSDB_CODE_OUT_OF_MEMORY; return TSDB_CODE_OUT_OF_MEMORY;
} }
} }
...@@ -277,6 +279,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo ...@@ -277,6 +279,7 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
&keyInfo.groupId); &keyInfo.groupId);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
taosMemoryFree(keyBuf); taosMemoryFree(keyBuf);
taosArrayDestroy(qa);
return code; return code;
} }
} }
...@@ -298,8 +301,6 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo ...@@ -298,8 +301,6 @@ int32_t qUpdateQualifiedTableId(qTaskInfo_t tinfo, const SArray* tableIdList, bo
taosHashPut(pTaskInfo->tableqinfoList.map, uid, sizeof(*uid), &keyInfo.groupId, sizeof(keyInfo.groupId)); taosHashPut(pTaskInfo->tableqinfoList.map, uid, sizeof(*uid), &keyInfo.groupId, sizeof(keyInfo.groupId));
} }
/*}*/
if (keyBuf != NULL) { if (keyBuf != NULL) {
taosMemoryFree(keyBuf); taosMemoryFree(keyBuf);
} }
......
...@@ -462,6 +462,7 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc ...@@ -462,6 +462,7 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
for (int32_t i = 0; i < numOfRows; ++i) { for (int32_t i = 0; i < numOfRows; ++i) {
colDataAppend(pColInfo, i, tmp, false); colDataAppend(pColInfo, i, tmp, false);
} }
taosMemoryFree(tmp);
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
......
...@@ -757,7 +757,6 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition ...@@ -757,7 +757,6 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
goto _error; goto _error;
} }
SSDataBlock* pResBlock = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
int32_t numOfCols = 0; int32_t numOfCols = 0;
SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols); SExprInfo* pExprInfo = createExprInfo(pPartNode->pTargets, NULL, &numOfCols);
...@@ -781,14 +780,18 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition ...@@ -781,14 +780,18 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
uint32_t defaultPgsz = 0; uint32_t defaultPgsz = 0;
uint32_t defaultBufsz = 0; uint32_t defaultBufsz = 0;
SSDataBlock* pResBlock = createResDataBlock(pPartNode->node.pOutputDataBlockDesc);
getBufferPgSize(pResBlock->info.rowSize, &defaultPgsz, &defaultBufsz); getBufferPgSize(pResBlock->info.rowSize, &defaultPgsz, &defaultBufsz);
if (!osTempSpaceAvailable()) { if (!osTempSpaceAvailable()) {
terrno = TSDB_CODE_NO_AVAIL_DISK; terrno = TSDB_CODE_NO_AVAIL_DISK;
pTaskInfo->code = terrno; pTaskInfo->code = terrno;
qError("Create partition operator info failed since %s", terrstr(terrno)); qError("Create partition operator info failed since %s", terrstr(terrno));
blockDataDestroy(pResBlock);
goto _error; goto _error;
} }
int32_t code = createDiskbasedBuf(&pInfo->pBuf, defaultPgsz, defaultBufsz, pTaskInfo->id.str, tsTempDir); int32_t code = createDiskbasedBuf(&pInfo->pBuf, defaultPgsz, defaultBufsz, pTaskInfo->id.str, tsTempDir);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
...@@ -819,7 +822,9 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition ...@@ -819,7 +822,9 @@ SOperatorInfo* createPartitionOperatorInfo(SOperatorInfo* downstream, SPartition
_error: _error:
pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY; pTaskInfo->code = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFreeClear(pInfo); if (pInfo != NULL) {
destroyPartitionOperatorInfo(pInfo);
}
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
return NULL; return NULL;
} }
......
...@@ -737,7 +737,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, ...@@ -737,7 +737,7 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc; SDataBlockDescNode* pDescNode = pTableScanNode->scan.node.pOutputDataBlockDesc;
int32_t numOfCols = 0; int32_t numOfCols = 0;
SArray* pColList = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID); pInfo->pColMatchInfo = extractColMatchInfo(pTableScanNode->scan.pScanCols, pDescNode, &numOfCols, COL_MATCH_FROM_COL_ID);
int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode); int32_t code = initQueryTableDataCond(&pInfo->cond, pTableScanNode);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
...@@ -765,7 +765,6 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode, ...@@ -765,7 +765,6 @@ SOperatorInfo* createTableScanOperatorInfo(STableScanPhysiNode* pTableScanNode,
} }
pInfo->scanFlag = MAIN_SCAN; pInfo->scanFlag = MAIN_SCAN;
pInfo->pColMatchInfo = pColList;
pInfo->currentGroupId = -1; pInfo->currentGroupId = -1;
pInfo->assignBlockUid = pTableScanNode->assignBlockUid; pInfo->assignBlockUid = pTableScanNode->assignBlockUid;
...@@ -1768,8 +1767,9 @@ FETCH_NEXT_BLOCK: ...@@ -1768,8 +1767,9 @@ FETCH_NEXT_BLOCK:
generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes); generateDeleteResultBlock(pInfo, pDelBlock, pInfo->pDeleteDataRes);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT; pInfo->pDeleteDataRes->info.type = STREAM_DELETE_RESULT;
printDataBlock(pDelBlock, "stream scan delete result"); printDataBlock(pDelBlock, "stream scan delete result");
if (pInfo->pDeleteDataRes->info.rows > 0) {
blockDataDestroy(pDelBlock); blockDataDestroy(pDelBlock);
if (pInfo->pDeleteDataRes->info.rows > 0) {
return pInfo->pDeleteDataRes; return pInfo->pDeleteDataRes;
} else { } else {
goto FETCH_NEXT_BLOCK; goto FETCH_NEXT_BLOCK;
...@@ -2090,24 +2090,30 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT ...@@ -2090,24 +2090,30 @@ SOperatorInfo* createRawScanOperatorInfo(SReadHandle* pHandle, SExecTaskInfo* pT
// create meta reader // create meta reader
// create tq reader // create tq reader
int32_t code = TSDB_CODE_SUCCESS;
SStreamRawScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamRawScanInfo)); SStreamRawScanInfo* pInfo = taosMemoryCalloc(1, sizeof(SStreamRawScanInfo));
SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo)); SOperatorInfo* pOperator = taosMemoryCalloc(1, sizeof(SOperatorInfo));
if (pInfo == NULL || pOperator == NULL) { if (pInfo == NULL || pOperator == NULL) {
terrno = TSDB_CODE_QRY_OUT_OF_MEMORY; code = TSDB_CODE_OUT_OF_MEMORY;
return NULL; goto _end;
} }
pInfo->vnode = pHandle->vnode; pInfo->vnode = pHandle->vnode;
pInfo->sContext = pHandle->sContext; pInfo->sContext = pHandle->sContext;
pOperator->name = "RawStreamScanOperator"; pOperator->name = "RawStreamScanOperator";
// pOperator->blocking = false;
// pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo; pOperator->info = pInfo;
pOperator->pTaskInfo = pTaskInfo; pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, NULL, destroyRawScanOperatorInfo, NULL, NULL, NULL); pOperator->fpSet = createOperatorFpSet(NULL, doRawScan, NULL, NULL, destroyRawScanOperatorInfo, NULL, NULL, NULL);
return pOperator; return pOperator;
_end:
taosMemoryFree(pInfo);
taosMemoryFree(pOperator);
pTaskInfo->code = code;
return NULL;
} }
static void destroyStreamScanOperatorInfo(void* param) { static void destroyStreamScanOperatorInfo(void* param) {
...@@ -2285,7 +2291,14 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys ...@@ -2285,7 +2291,14 @@ SOperatorInfo* createStreamScanOperatorInfo(SReadHandle* pHandle, STableScanPhys
return pOperator; return pOperator;
_error: _error:
taosMemoryFreeClear(pInfo); if (pColIds != NULL) {
taosArrayDestroy(pColIds);
}
if (pInfo != NULL) {
destroyStreamScanOperatorInfo(pInfo);
}
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
return NULL; return NULL;
} }
...@@ -3333,7 +3346,7 @@ int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags ...@@ -3333,7 +3346,7 @@ int32_t createScanTableListInfo(SScanPhysiNode* pScanNode, SNodeList* pGroupTags
int64_t st = taosGetTimestampUs(); int64_t st = taosGetTimestampUs();
if (pHandle == NULL) { if (pHandle == NULL) {
qError("invalid handle, in creating operator tree: %s", idStr); qError("invalid handle, in creating operator tree, %s", idStr);
return TSDB_CODE_INVALID_PARA; return TSDB_CODE_INVALID_PARA;
} }
......
...@@ -467,10 +467,12 @@ struct SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t ...@@ -467,10 +467,12 @@ struct SFillInfo* taosCreateFillInfo(TSKEY skey, int32_t numOfFillCols, int32_t
case FILL_MODE_VALUE: case FILL_MODE_VALUE:
pFillInfo->type = TSDB_FILL_SET_VALUE; pFillInfo->type = TSDB_FILL_SET_VALUE;
break; break;
default: default: {
taosMemoryFree(pFillInfo);
terrno = TSDB_CODE_INVALID_PARA; terrno = TSDB_CODE_INVALID_PARA;
return NULL; return NULL;
} }
}
pFillInfo->type = fillType; pFillInfo->type = fillType;
pFillInfo->pFillCol = pCol; pFillInfo->pFillCol = pCol;
......
...@@ -2669,14 +2669,13 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi ...@@ -2669,14 +2669,13 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi
int32_t num = 0; int32_t num = 0;
SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &num); SExprInfo* pExprInfo = createExprInfo(pStateNode->window.pFuncs, NULL, &num);
SSDataBlock* pResBlock = createResDataBlock(pStateNode->window.node.pOutputDataBlockDesc);
initResultSizeInfo(&pOperator->resultInfo, 4096); initResultSizeInfo(&pOperator->resultInfo, 4096);
int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str); int32_t code = initAggInfo(&pOperator->exprSupp, &pInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
SSDataBlock* pResBlock = createResDataBlock(pStateNode->window.node.pOutputDataBlockDesc);
initBasicInfo(&pInfo->binfo, pResBlock); initBasicInfo(&pInfo->binfo, pResBlock);
initResultRowInfo(&pInfo->binfo.resultRowInfo); initResultRowInfo(&pInfo->binfo.resultRowInfo);
...@@ -2704,7 +2703,10 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi ...@@ -2704,7 +2703,10 @@ SOperatorInfo* createStatewindowOperatorInfo(SOperatorInfo* downstream, SStateWi
return pOperator; return pOperator;
_error: _error:
if (pInfo != NULL) {
destroyStateWindowOperatorInfo(pInfo); destroyStateWindowOperatorInfo(pInfo);
}
taosMemoryFreeClear(pOperator); taosMemoryFreeClear(pOperator);
pTaskInfo->code = code; pTaskInfo->code = code;
return NULL; return NULL;
...@@ -4973,9 +4975,7 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, ...@@ -4973,9 +4975,7 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream,
goto _error; goto _error;
} }
int32_t num = 0;
SExprInfo* pExprInfo = createExprInfo(pNode->window.pFuncs, NULL, &num);
SSDataBlock* pResBlock = createResDataBlock(pNode->window.node.pOutputDataBlockDesc);
SInterval interval = {.interval = pNode->interval, SInterval interval = {.interval = pNode->interval,
.sliding = pNode->sliding, .sliding = pNode->sliding,
...@@ -4999,11 +4999,14 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream, ...@@ -4999,11 +4999,14 @@ SOperatorInfo* createMergeAlignedIntervalOperatorInfo(SOperatorInfo* downstream,
size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES; size_t keyBufSize = sizeof(int64_t) + sizeof(int64_t) + POINTER_BYTES;
initResultSizeInfo(&pOperator->resultInfo, 4096); initResultSizeInfo(&pOperator->resultInfo, 4096);
int32_t num = 0;
SExprInfo* pExprInfo = createExprInfo(pNode->window.pFuncs, NULL, &num);
int32_t code = initAggInfo(&pOperator->exprSupp, &iaInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str); int32_t code = initAggInfo(&pOperator->exprSupp, &iaInfo->aggSup, pExprInfo, num, keyBufSize, pTaskInfo->id.str);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
goto _error; goto _error;
} }
SSDataBlock* pResBlock = createResDataBlock(pNode->window.node.pOutputDataBlockDesc);
initBasicInfo(&iaInfo->binfo, pResBlock); initBasicInfo(&iaInfo->binfo, pResBlock);
initExecTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &iaInfo->win); initExecTimeWindowInfo(&iaInfo->twAggSup.timeWindowData, &iaInfo->win);
......
...@@ -181,6 +181,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) { ...@@ -181,6 +181,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
blockDataSplitRows(pDataBlock, pDataBlock->info.hasVarCol, start, &stop, pHandle->pageSize); blockDataSplitRows(pDataBlock, pDataBlock->info.hasVarCol, start, &stop, pHandle->pageSize);
SSDataBlock* p = blockDataExtractBlock(pDataBlock, start, stop - start + 1); SSDataBlock* p = blockDataExtractBlock(pDataBlock, start, stop - start + 1);
if (p == NULL) { if (p == NULL) {
taosArrayDestroy(pPageIdList);
return terrno; return terrno;
} }
...@@ -422,7 +423,7 @@ int32_t msortComparFn(const void* pLeft, const void* pRight, void* param) { ...@@ -422,7 +423,7 @@ int32_t msortComparFn(const void* pLeft, const void* pRight, void* param) {
SColumnInfoData* pRightColInfoData = TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->slotId); SColumnInfoData* pRightColInfoData = TARRAY_GET_ELEM(pRightBlock->pDataBlock, pOrder->slotId);
bool rightNull = false; bool rightNull = false;
if (pRightColInfoData->hasNull) { if (pRightColInfoData->hasNull) {
if (pLeftBlock->pBlockAgg == NULL) { if (pRightBlock->pBlockAgg == NULL) {
rightNull = colDataIsNull_s(pRightColInfoData, pRightSource->src.rowIndex); rightNull = colDataIsNull_s(pRightColInfoData, pRightSource->src.rowIndex);
} else { } else {
rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex, rightNull = colDataIsNull(pRightColInfoData, pRightBlock->info.rows, pRightSource->src.rowIndex,
......
...@@ -6187,99 +6187,6 @@ int32_t groupKeyFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) { ...@@ -6187,99 +6187,6 @@ int32_t groupKeyFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock) {
return pResInfo->numOfRes; return pResInfo->numOfRes;
} }
int32_t interpFunction(SqlFunctionCtx* pCtx) {
#if 0
int32_t fillType = (int32_t) pCtx->param[2].i64;
//bool ascQuery = (pCtx->order == TSDB_ORDER_ASC);
if (pCtx->start.key == pCtx->startTs) {
assert(pCtx->start.key != INT64_MIN);
COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, &pCtx->start.val);
goto interp_success_exit;
} else if (pCtx->end.key == pCtx->startTs && pCtx->end.key != INT64_MIN && fillType == TSDB_FILL_NEXT) {
COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, &pCtx->end.val);
goto interp_success_exit;
}
switch (fillType) {
case TSDB_FILL_NULL:
setNull(pCtx->pOutput, pCtx->outputType, pCtx->outputBytes);
break;
case TSDB_FILL_SET_VALUE:
tVariantDump(&pCtx->param[1], pCtx->pOutput, pCtx->inputType, true);
break;
case TSDB_FILL_LINEAR:
if (pCtx->start.key == INT64_MIN || pCtx->start.key > pCtx->startTs
|| pCtx->end.key == INT64_MIN || pCtx->end.key < pCtx->startTs) {
goto interp_exit;
}
double v1 = -1, v2 = -1;
GET_TYPED_DATA(v1, double, pCtx->inputType, &pCtx->start.val);
GET_TYPED_DATA(v2, double, pCtx->inputType, &pCtx->end.val);
SPoint point1 = {.key = pCtx->start.key, .val = &v1};
SPoint point2 = {.key = pCtx->end.key, .val = &v2};
SPoint point = {.key = pCtx->startTs, .val = pCtx->pOutput};
int32_t srcType = pCtx->inputType;
if (isNull((char *)&pCtx->start.val, srcType) || isNull((char *)&pCtx->end.val, srcType)) {
setNull(pCtx->pOutput, srcType, pCtx->inputBytes);
} else {
bool exceedMax = false, exceedMin = false;
taosGetLinearInterpolationVal(&point, pCtx->outputType, &point1, &point2, TSDB_DATA_TYPE_DOUBLE, &exceedMax, &exceedMin);
if (exceedMax || exceedMin) {
__compar_fn_t func = getComparFunc((int32_t)pCtx->inputType, 0);
if (func(&pCtx->start.val, &pCtx->end.val) <= 0) {
COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, exceedMax ? &pCtx->start.val : &pCtx->end.val);
} else {
COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, exceedMax ? &pCtx->end.val : &pCtx->start.val);
}
}
}
break;
case TSDB_FILL_PREV:
if (pCtx->start.key == INT64_MIN || pCtx->start.key > pCtx->startTs) {
goto interp_exit;
}
COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, &pCtx->start.val);
break;
case TSDB_FILL_NEXT:
if (pCtx->end.key == INT64_MIN || pCtx->end.key < pCtx->startTs) {
goto interp_exit;
}
COPY_TYPED_DATA(pCtx->pOutput, pCtx->inputType, &pCtx->end.val);
break;
case TSDB_FILL_NONE:
// do nothing
default:
goto interp_exit;
}
interp_success_exit:
*(TSKEY*)pCtx->ptsOutputBuf = pCtx->startTs;
INC_INIT_VAL(pCtx, 1);
interp_exit:
pCtx->start.key = INT64_MIN;
pCtx->end.key = INT64_MIN;
pCtx->endTs = pCtx->startTs;
#endif
return TSDB_CODE_SUCCESS;
}
int32_t cachedLastRowFunction(SqlFunctionCtx* pCtx) { int32_t cachedLastRowFunction(SqlFunctionCtx* pCtx) {
int32_t numOfElems = 0; int32_t numOfElems = 0;
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册