Skip to content

T
TDengine

taosdata / TDengine
接近 2 年 前同步成功

通知 1192
Star 22018
Fork 4786
T
TDengine
未验证 提交 5cdef06f 编写于 作者: L Liu Jicong 提交者： GitHub
浏览文件
操作

Merge pull request #14119 from taosdata/feature/3_liaohj 

refactor(query): do some internal refactor.
上级 62653a77 800098a7
展开全部 隐藏空白更改
内联 并排
Showing with 622 addition and 795 deletion
include/common/tcommon.h
浏览文件 @ 5cdef06f
......@@ -69,14 +69,13 @@ typedef struct SColumnDataAgg {
typedef struct SDataBlockInfo {
STimeWindow window;
int32_t rows;
int32_t rows; // todo hide this attribute
int32_t rowSize;
int64_t uid; // the uid of table, from which current data block comes
int64_t blockId; // block id, generated by physical planner
uint64_t uid; // the uid of table, from which current data block comes
uint16_t blockId; // block id, generated by physical planner
uint64_t groupId; // no need to serialize
int16_t numOfCols;
int16_t hasVarCol;
int32_t capacity;
uint32_t capacity;
// TODO: optimize and remove following
int32_t childId; // used for stream, do not serialize
EStreamType type; // used for stream, do not serialize
......@@ -97,7 +96,7 @@ typedef struct SVarColAttr {
// pBlockAgg->numOfNull == info.rows, all data are null
// pBlockAgg->numOfNull == 0, no data are null.
typedef struct SColumnInfoData {
SColumnInfo info; // TODO filter info needs to be removed
SColumnInfo info; // column info
bool hasNull; // if current column data has null value.
char* pData; // the corresponding block data in memory
union {
......
include/common/tdatablock.h
浏览文件 @ 5cdef06f
......@@ -184,9 +184,9 @@ static FORCE_INLINE void colDataAppendDouble(SColumnInfoData* pColumnInfoData, u
int32_t getJsonValueLen(const char* data);
int32_t colDataAppend(SColumnInfoData* pColumnInfoData, uint32_t currentRow, const char* pData, bool isNull);
int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, int32_t* capacity,
int32_t colDataMergeCol(SColumnInfoData* pColumnInfoData, uint32_t numOfRow1, uint32_t* capacity,
const SColumnInfoData* pSource, uint32_t numOfRow2);
int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows);
int32_t colDataAssign(SColumnInfoData* pColumnInfoData, const SColumnInfoData* pSource, int32_t numOfRows, const SDataBlockInfo* pBlockInfo);
int32_t blockDataUpdateTsWindow(SSDataBlock* pDataBlock, int32_t tsColumnIndex);
int32_t colDataGetLength(const SColumnInfoData* pColumnInfoData, int32_t numOfRows);
......@@ -212,7 +212,7 @@ size_t blockDataGetSerialMetaSize(uint32_t numOfCols);
int32_t blockDataSort(SSDataBlock* pDataBlock, SArray* pOrderInfo);
int32_t blockDataSort_rv(SSDataBlock* pDataBlock, SArray* pOrderInfo, bool nullFirst);
int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, size_t existRows, uint32_t numOfRows);
int32_t colInfoDataEnsureCapacity(SColumnInfoData* pColumn, uint32_t numOfRows);
int32_t blockDataEnsureCapacity(SSDataBlock* pDataBlock, uint32_t numOfRows);
void colInfoDataCleanup(SColumnInfoData* pColumn, uint32_t numOfRows);
......@@ -223,6 +223,11 @@ size_t blockDataGetCapacityInRow(const SSDataBlock* pBlock, size_t pageSize);
int32_t blockDataTrimFirstNRows(SSDataBlock* pBlock, size_t n);
SSDataBlock* createOneDataBlock(const SSDataBlock* pDataBlock, bool copyData);
SSDataBlock* createDataBlock();
int32_t blockDataAppendColInfo(SSDataBlock* pBlock, SColumnInfoData* pColInfoData);
SColumnInfoData createColumnInfoData(int16_t type, int32_t bytes, int16_t colId);
SColumnInfoData* bdGetColumnInfoData(SSDataBlock* pBlock, int32_t index);
void blockCompressEncode(const SSDataBlock* pBlock, char* data, int32_t* dataLen, int32_t numOfCols,
int8_t needCompress);
......@@ -236,7 +241,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SArray* pDataBlocks
char* buildCtbNameByGroupId(const char* stbName, uint64_t groupId);
static FORCE_INLINE int32_t blockGetEncodeSize(const SSDataBlock* pBlock) {
return blockDataGetSerialMetaSize(pBlock->info.numOfCols) + blockDataGetSize(pBlock);
return blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock)) + blockDataGetSize(pBlock);
}
static FORCE_INLINE int32_t blockCompressColData(SColumnInfoData* pColRes, int32_t numOfRows, char* data,
......
include/common/tmsg.h
浏览文件 @ 5cdef06f
......@@ -629,11 +629,6 @@ typedef struct {
uint8_t scale;
} SColumnInfo;
typedef struct {
int64_t uid;
TSKEY key; // last accessed ts, for subscription
} STableIdInfo;
typedef struct STimeWindow {
TSKEY skey;
TSKEY ekey;
......
include/libs/function/function.h
浏览文件 @ 5cdef06f
......@@ -172,23 +172,6 @@ typedef struct tExprNode {
void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *));
typedef struct SAggFunctionInfo {
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // Scalar function or aggregation function
uint32_t functionId; // Function Id
int8_t sFunctionId; // Transfer function for super table query
uint16_t status;
bool (*init)(SqlFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo); // setup the execute environment
void (*addInput)(SqlFunctionCtx *pCtx);
// finalizer must be called after all exec has been executed to generated final result.
void (*finalize)(SqlFunctionCtx *pCtx);
void (*combine)(SqlFunctionCtx *pCtx);
int32_t (*dataReqFunc)(SqlFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
} SAggFunctionInfo;
struct SScalarParam {
SColumnInfoData *columnData;
SHashObj *pHashFilter;
......
source/common/src/tdatablock.c
浏览文件 @ 5cdef06f
此差异已折叠。
source/common/test/commonTests.cpp
浏览文件 @ 5cdef06f
......@@ -88,26 +88,15 @@ TEST(testCase, toInteger_test) {
}
TEST(testCase, Datablock_test) {
SSDataBlock* b = static_cast<SSDataBlock*>(taosMemoryCalloc(1, sizeof(SSDataBlock)));
b->info.numOfCols = 2;
b->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0};
infoData.info.bytes = 4;
infoData.info.type = TSDB_DATA_TYPE_INT;
infoData.info.colId = 1;
SSDataBlock* b = createDataBlock();
infoData.pData = (char*)taosMemoryCalloc(40, infoData.info.bytes);
infoData.nullbitmap = (char*)taosMemoryCalloc(1, sizeof(char) * (40 / 8));
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 1);
taosArrayPush(b->pDataBlock, &infoData);
blockDataAppendColInfo(b, &infoData);
SColumnInfoData infoData1 = {0};
infoData1.info.bytes = 40;
infoData1.info.type = TSDB_DATA_TYPE_BINARY;
infoData1.info.colId = 2;
infoData1.varmeta.offset = (int32_t*)taosMemoryCalloc(40, sizeof(uint32_t));
taosArrayPush(b->pDataBlock, &infoData1);
SColumnInfoData infoData1 = createColumnInfoData(TSDB_DATA_TYPE_BINARY, 40, 2);
blockDataAppendColInfo(b, &infoData1);
blockDataEnsureCapacity(b, 40);
char* str = "the value of: %d";
char buf[128] = {0};
......@@ -222,28 +211,17 @@ TEST(testCase, non_var_dataBlock_split_test) {
#endif
TEST(testCase, var_dataBlock_split_test) {
SSDataBlock* b = static_cast<SSDataBlock*>(taosMemoryCalloc(1, sizeof(SSDataBlock)));
b->info.numOfCols = 2;
b->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
int32_t numOfRows = 1000000;
SColumnInfoData infoData = {0};
infoData.info.bytes = 4;
infoData.info.type = TSDB_DATA_TYPE_INT;
infoData.info.colId = 1;
SSDataBlock* b = createDataBlock();
infoData.pData = (char*)taosMemoryCalloc(numOfRows, infoData.info.bytes);
infoData.nullbitmap = (char*)taosMemoryCalloc(1, sizeof(char) * (numOfRows / 8));
taosArrayPush(b->pDataBlock, &infoData);
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 1);
blockDataAppendColInfo(b, &infoData);
SColumnInfoData infoData1 = {0};
infoData1.info.bytes = 40;
infoData1.info.type = TSDB_DATA_TYPE_BINARY;
infoData1.info.colId = 2;
SColumnInfoData infoData1 = createColumnInfoData(TSDB_DATA_TYPE_BINARY, 40, 2);
blockDataAppendColInfo(b, &infoData1);
infoData1.varmeta.offset = (int32_t*)taosMemoryCalloc(numOfRows, sizeof(uint32_t));
taosArrayPush(b->pDataBlock, &infoData1);
blockDataEnsureCapacity(b, numOfRows);
char buf[41] = {0};
char buf1[100] = {0};
......
source/dnode/mgmt/mgmt_dnode/src/dmHandle.c
浏览文件 @ 5cdef06f
......@@ -210,7 +210,6 @@ SSDataBlock* dmBuildVariablesBlock(void) {
taosArrayPush(pBlock->pDataBlock, &colInfoData);
}
pBlock->info.numOfCols = pMeta[index].colNum;
pBlock->info.hasVarCol = true;
return pBlock;
......@@ -269,8 +268,9 @@ int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
dmAppendVariablesToBlock(pBlock, pMgmt->pData->dnodeId);
size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pBlock->info.numOfCols +
blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(pBlock->info.numOfCols);
size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * numOfCols +
blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(numOfCols);
SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
if (pRsp == NULL) {
......@@ -281,10 +281,10 @@ int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
char *pStart = pRsp->data;
*(int32_t *)pStart = htonl(pBlock->info.numOfCols);
*(int32_t *)pStart = htonl(numOfCols);
pStart += sizeof(int32_t); // number of columns
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
for (int32_t i = 0; i < numOfCols; ++i) {
SSysTableSchema *pSchema = (SSysTableSchema *)pStart;
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, i);
......@@ -296,7 +296,7 @@ int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
}
int32_t len = 0;
blockCompressEncode(pBlock, pStart, &len, pBlock->info.numOfCols, false);
blockCompressEncode(pBlock, pStart, &len, numOfCols, false);
pRsp->numOfRows = htonl(pBlock->info.rows);
pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision
......
source/dnode/mnode/impl/src/mndShow.c
浏览文件 @ 5cdef06f
......@@ -236,7 +236,6 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
int32_t numOfCols = pShow->pMeta->numOfColumns;
SSDataBlock *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
pBlock->info.numOfCols = numOfCols;
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {0};
......@@ -271,7 +270,7 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
}
size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pShow->pMeta->numOfColumns +
blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(pBlock->info.numOfCols);
blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock));
SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
if (pRsp == NULL) {
......
source/dnode/vnode/inc/vnode.h
浏览文件 @ 5cdef06f
......@@ -149,8 +149,8 @@ int32_t tqReadHandleRemoveTbUidList(STqReadHandle *pHandle, const SArray *tbUidL
int32_t tqReadHandleSetMsg(STqReadHandle *pHandle, SSubmitReq *pMsg, int64_t ver);
bool tqNextDataBlock(STqReadHandle *pHandle);
bool tqNextDataBlockFilterOut(STqReadHandle *pHandle, SHashObj *filterOutUids);
int32_t tqRetrieveDataBlock(SArray **ppCols, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t *pUid,
int32_t *pNumOfRows, int16_t *pNumOfCols);
int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle *pHandle, uint64_t *pGroupId, uint64_t *pUid,
int32_t *pNumOfRows);
// sma
int32_t smaGetTSmaDays(SVnodeCfg *pCfg, void *pCont, uint32_t contLen, int32_t *days);
......
source/dnode/vnode/src/tq/tqExec.c
浏览文件 @ 5cdef06f
......@@ -29,7 +29,7 @@ static int32_t tqAddBlockDataToRsp(const SSDataBlock* pBlock, SMqDataBlkRsp* pRs
// TODO enable compress
int32_t actualLen = 0;
blockCompressEncode(pBlock, pRetrieve->data, &actualLen, pBlock->info.numOfCols, false);
blockCompressEncode(pBlock, pRetrieve->data, &actualLen, taosArrayGetSize(pBlock->pDataBlock), false);
actualLen += sizeof(SRetrieveTableRsp);
ASSERT(actualLen <= dataStrLen);
taosArrayPush(pRsp->blockDataLen, &actualLen);
......@@ -75,7 +75,7 @@ int32_t tqScanSnapshot(STQ* pTq, const STqExecHandle* pExec, SMqDataBlkRsp* pRsp
if (pDataBlock == NULL) break;
ASSERT(pDataBlock->info.rows != 0);
ASSERT(pDataBlock->info.numOfCols != 0);
ASSERT(taosArrayGetSize(pDataBlock->pDataBlock) != 0);
tqAddBlockDataToRsp(pDataBlock, pRsp);
pRsp->blockNum++;
......@@ -98,7 +98,6 @@ int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkR
if (pDataBlock == NULL) break;
ASSERT(pDataBlock->info.rows != 0);
ASSERT(pDataBlock->info.numOfCols != 0);
tqAddBlockDataToRsp(pDataBlock, pRsp);
if (pRsp->withTbName) {
......@@ -112,8 +111,7 @@ int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkR
tqReadHandleSetMsg(pReader, pReq, 0);
while (tqNextDataBlock(pReader)) {
SSDataBlock block = {0};
if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
&block.info.numOfCols) < 0) {
if (tqRetrieveDataBlock(&block, pReader, &block.info.groupId, &block.info.uid, &block.info.rows) < 0) {
if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;
ASSERT(0);
}
......@@ -130,8 +128,7 @@ int32_t tqDataExec(STQ* pTq, STqExecHandle* pExec, SSubmitReq* pReq, SMqDataBlkR
tqReadHandleSetMsg(pReader, pReq, 0);
while (tqNextDataBlockFilterOut(pReader, pExec->execDb.pFilterOutTbUid)) {
SSDataBlock block = {0};
if (tqRetrieveDataBlock(&block.pDataBlock, pReader, &block.info.groupId, &block.info.uid, &block.info.rows,
&block.info.numOfCols) < 0) {
if (tqRetrieveDataBlock(&block, pReader, &block.info.groupId, &block.info.uid, &block.info.rows) < 0) {
if (terrno == TSDB_CODE_TQ_TABLE_SCHEMA_NOT_FOUND) continue;
ASSERT(0);
}
......
source/dnode/vnode/src/tq/tqRead.c
浏览文件 @ 5cdef06f
......@@ -142,8 +142,8 @@ bool tqNextDataBlockFilterOut(STqReadHandle* pHandle, SHashObj* filterOutUids) {
return false;
}
int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* pGroupId, uint64_t* pUid,
int32_t* pNumOfRows, int16_t* pNumOfCols) {
int32_t tqRetrieveDataBlock(SSDataBlock* pBlock, STqReadHandle* pHandle, uint64_t* pGroupId, uint64_t* pUid,
int32_t* pNumOfRows) {
*pUid = 0;
// TODO set to real sversion
......@@ -180,23 +180,14 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
int32_t colNumNeed = taosArrayGetSize(pHandle->pColIdList);
if (colNumNeed == 0) {
*ppCols = taosArrayInit(pSchemaWrapper->nCols, sizeof(SColumnInfoData));
if (*ppCols == NULL) {
return -1;
}
int32_t colMeta = 0;
while (colMeta < pSchemaWrapper->nCols) {
SSchema* pColSchema = &pSchemaWrapper->pSchema[colMeta];
SColumnInfoData colInfo = {0};
colInfo.info.bytes = pColSchema->bytes;
colInfo.info.colId = pColSchema->colId;
colInfo.info.type = pColSchema->type;
if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
int32_t code = blockDataAppendColInfo(pBlock, &colInfo);
if (code != TSDB_CODE_SUCCESS) {
goto FAIL;
}
taosArrayPush(*ppCols, &colInfo);
colMeta++;
}
} else {
......@@ -204,11 +195,6 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
colNumNeed = pSchemaWrapper->nCols;
}
*ppCols = taosArrayInit(colNumNeed, sizeof(SColumnInfoData));
if (*ppCols == NULL) {
return -1;
}
int32_t colMeta = 0;
int32_t colNeed = 0;
while (colMeta < pSchemaWrapper->nCols && colNeed < colNumNeed) {
......@@ -220,23 +206,22 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
} else if (colIdSchema > colIdNeed) {
colNeed++;
} else {
SColumnInfoData colInfo = {0};
colInfo.info.bytes = pColSchema->bytes;
colInfo.info.colId = pColSchema->colId;
colInfo.info.type = pColSchema->type;
if (colInfoDataEnsureCapacity(&colInfo, 0, *pNumOfRows) < 0) {
SColumnInfoData colInfo = createColumnInfoData(pColSchema->type, pColSchema->bytes, pColSchema->colId);
int32_t code = blockDataAppendColInfo(pBlock, &colInfo);
if (code != TSDB_CODE_SUCCESS) {
goto FAIL;
}
taosArrayPush(*ppCols, &colInfo);
colMeta++;
colNeed++;
}
}
}
int32_t colActual = taosArrayGetSize(*ppCols);
*pNumOfCols = colActual;
if (blockDataEnsureCapacity(pBlock, *pNumOfRows) < 0) {
goto FAIL;
}
int32_t colActual = blockDataGetNumOfCols(pBlock);
// TODO in stream shuffle case, fetch groupId
*pGroupId = 0;
......@@ -253,7 +238,7 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
tdSTSRowIterReset(&iter, row);
// get all wanted col of that block
for (int32_t i = 0; i < colActual; i++) {
SColumnInfoData* pColData = taosArrayGet(*ppCols, i);
SColumnInfoData* pColData = taosArrayGet(pBlock->pDataBlock, i);
SCellVal sVal = {0};
if (!tdSTSRowIterNext(&iter, pColData->info.colId, pColData->info.type, &sVal)) {
break;
......@@ -265,8 +250,9 @@ int32_t tqRetrieveDataBlock(SArray** ppCols, STqReadHandle* pHandle, uint64_t* p
curRow++;
}
return 0;
FAIL:
if (*ppCols) taosArrayDestroy(*ppCols);
FAIL: // todo refactor here
// if (*ppCols) taosArrayDestroy(*ppCols);
return -1;
}
......
source/dnode/vnode/src/tq/tqSink.c
浏览文件 @ 5cdef06f
......@@ -44,7 +44,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
createTbReq.ctb.suid = suid;
STagVal tagVal = {
.cid = pDataBlock->info.numOfCols + 1,
.cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
.type = TSDB_DATA_TYPE_UBIGINT,
.i64 = (int64_t)pDataBlock->info.groupId,
};
......@@ -107,7 +107,7 @@ SSubmitReq* tdBlockToSubmit(const SArray* pBlocks, const STSchema* pTSchema, boo
createTbReq.ctb.suid = suid;
STagVal tagVal = {
.cid = pDataBlock->info.numOfCols + 1,
.cid = taosArrayGetSize(pDataBlock->pDataBlock) + 1,
.type = TSDB_DATA_TYPE_UBIGINT,
.i64 = (int64_t)pDataBlock->info.groupId,
};
......
source/dnode/vnode/src/tsdb/tsdbRead.c
浏览文件 @ 5cdef06f
......@@ -16,13 +16,11 @@
#include "tsdb.h"
#include "vnode.h"
#define EXTRA_BYTES 2
#define ASCENDING_TRAVERSE(o) (o == TSDB_ORDER_ASC)
#define QH_GET_NUM_OF_COLS(handle) ((size_t)(taosArrayGetSize((handle)->pColumns)))
#define QH_GET_NUM_OF_COLS(handle) ((size_t)(taosArrayGetSize((handle)->pResBlock->pDataBlock)))
#define GET_FILE_DATA_BLOCK_INFO(_checkInfo, _block) \
((SDataBlockInfo){.window = {.skey = (_block)->minKey.ts, .ekey = (_block)->maxKey.ts}, \
.numOfCols = (_block)->numOfCols, \
.rows = (_block)->numOfRows, \
.uid = (_checkInfo)->tableId})
......@@ -115,7 +113,8 @@ typedef struct STsdbReadHandle {
// SColumnDataAgg* statis; // query level statistics, only one table block statistics info exists at any time
// SColumnDataAgg** pstatis;// the ptr array list to return to caller
int32_t numOfBlocks;
SArray* pColumns; // column list, SColumnInfoData array list
SSDataBlock* pResBlock;
// SArray* pColumns; // column list, SColumnInfoData array list
bool locateStart;
int32_t outputCapacity;
int32_t realNumOfRows;
......@@ -177,7 +176,7 @@ static SArray* getColumnIdList(STsdbReadHandle* pTsdbReadHandle) {
SArray* pIdList = taosArrayInit(numOfCols, sizeof(int16_t));
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pCol = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pCol = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
taosArrayPush(pIdList, &pCol->info.colId);
}
......@@ -425,23 +424,21 @@ static STsdbReadHandle* tsdbQueryTablesImpl(SVnode* pVnode, SQueryTableDataCond*
}
// todo: use list instead of array?
pReadHandle->pColumns = taosArrayInit(pCond->numOfCols, sizeof(SColumnInfoData));
if (pReadHandle->pColumns == NULL) {
pReadHandle->pResBlock = createDataBlock();
if (pReadHandle->pResBlock == NULL) {
goto _end;
}
for (int32_t i = 0; i < pCond->numOfCols; ++i) {
SColumnInfoData colInfo = {{0}, 0};
colInfo.info = pCond->colList[i];
int32_t code = colInfoDataEnsureCapacity(&colInfo, 0, pReadHandle->outputCapacity);
if (code != TSDB_CODE_SUCCESS) {
SColumnInfoData colInfo = {.info = pCond->colList[i], 0};
int32_t code = blockDataAppendColInfo(pReadHandle->pResBlock, &colInfo);
if (code != TSDB_CODE_SUCCESS){
goto _end;
}
taosArrayPush(pReadHandle->pColumns, &colInfo);
}
blockDataEnsureCapacity(pReadHandle->pResBlock, pReadHandle->outputCapacity);
pReadHandle->suppInfo.defaultLoadColumn = getDefaultLoadColumns(pReadHandle, true);
size_t size = taosArrayGetSize(pReadHandle->suppInfo.defaultLoadColumn);
......@@ -1536,12 +1533,12 @@ static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t
int32_t trueEnd = ascScan ? end : start;
int32_t step = ascScan ? 1 : -1;
int32_t requiredNumOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
int32_t requiredNumOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pResBlock->pDataBlock);
// data in buffer has greater timestamp, copy data in file block
int32_t i = 0, j = 0;
while (i < requiredNumOfCols && j < pCols->numOfCols) {
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
SDataCol* src = &pCols->cols[j];
if (src->colId < pColInfo->info.colId) {
......@@ -1592,7 +1589,7 @@ static int32_t doCopyRowsFromFileBlock(STsdbReadHandle* pTsdbReadHandle, int32_t
}
while (i < requiredNumOfCols) { // the remain columns are all null data
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
colDataAppendNNULL(pColInfo, numOfRows, num);
i++;
}
......@@ -1674,7 +1671,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
int32_t i = 0, j = 0, k = 0;
while (i < numOfCols && (j < numOfColsOfRow1 || k < numOfColsOfRow2)) {
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
int32_t colIdOfRow1;
if (j >= numOfColsOfRow1) {
......@@ -1805,7 +1802,7 @@ static int32_t mergeTwoRowFromMem(STsdbReadHandle* pTsdbReadHandle, int32_t capa
if (mergeOption == 1) {
while (i < numOfCols) { // the remain columns are all null data
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
colDataAppend(pColInfo, *curRow, NULL, true);
++i;
}
......@@ -1861,7 +1858,7 @@ static void doCheckGeneratedBlockRange(STsdbReadHandle* pTsdbReadHandle) {
assert(cur->win.skey >= pTsdbReadHandle->window.ekey && cur->win.ekey <= pTsdbReadHandle->window.skey);
}
SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, 0);
SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, 0);
assert(cur->win.skey == ((TSKEY*)pColInfoData->pData)[0] &&
cur->win.ekey == ((TSKEY*)pColInfoData->pData)[cur->rows - 1]);
} else {
......@@ -2771,7 +2768,7 @@ static int tsdbReadRowsFromCache(STableCheckInfo* pCheckInfo, TSKEY maxKey, int
STsdbReadHandle* pTsdbReadHandle) {
int numOfRows = 0;
int curRows = 0;
int32_t numOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pColumns);
int32_t numOfCols = (int32_t)taosArrayGetSize(pTsdbReadHandle->pResBlock->pDataBlock);
STsdbCfg* pCfg = REPO_CFG(pTsdbReadHandle->pTsdb);
win->skey = TSKEY_INITIAL_VAL;
......@@ -2901,7 +2898,7 @@ static bool loadBlockOfActiveTable(STsdbReadHandle* pTsdbReadHandle) {
if (exists) {
tsdbRetrieveDataBlock((tsdbReaderT*)pTsdbReadHandle, NULL);
if (pTsdbReadHandle->currentLoadExternalRows && pTsdbReadHandle->window.skey == pTsdbReadHandle->window.ekey) {
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, 0);
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, 0);
assert(*(int64_t*)pColInfo->pData == pTsdbReadHandle->window.skey);
}
......@@ -3003,7 +3000,7 @@ static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
//
// int32_t i = 0, j = 0;
// while(i < tgNumOfCols && j < numOfCols) {
// pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
// pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
// if (pTable->lastCols[j].colId < pColInfo->info.colId) {
// j++;
// continue;
......@@ -3066,7 +3063,7 @@ static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
// continue;
// }
//
// pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, n);
// pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, n);
// pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;;
//
// if (pColInfo->info.colId == PRIMARYKEY_TIMESTAMP_COL_ID) {
......@@ -3091,7 +3088,7 @@ static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
//
// // leave the real ts column as the last row, because last function only (not stable) use the last row as res
// if (priKey != TSKEY_INITIAL_VAL) {
// pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, priIdx);
// pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, priIdx);
// pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;
//
// *(TSKEY *)pData = priKey;
......@@ -3101,7 +3098,7 @@ static bool loadCachedLastRow(STsdbReadHandle* pTsdbReadHandle) {
// continue;
// }
//
// pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, n);
// pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, n);
// pData = (char*)pColInfo->pData + numOfRows * pColInfo->info.bytes;;
//
// assert (pColInfo->info.colId != PRIMARYKEY_TIMESTAMP_COL_ID);
......@@ -3161,9 +3158,9 @@ static bool loadDataBlockFromTableSeq(STsdbReadHandle* pTsdbReadHandle) {
bool tsdbNextDataBlock(tsdbReaderT pHandle) {
STsdbReadHandle* pTsdbReadHandle = (STsdbReadHandle*)pHandle;
size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pColumns);
size_t numOfCols = taosArrayGetSize(pTsdbReadHandle->pResBlock->pDataBlock);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pColInfo = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
colInfoDataCleanup(pColInfo, pTsdbReadHandle->outputCapacity);
}
......@@ -3250,7 +3247,7 @@ bool tsdbNextDataBlock(tsdbReaderT pHandle) {
// SArray* row = (type == TSDB_PREV_ROW)? pTsdbReadHandle->prev : pTsdbReadHandle->next;
//
// for (int32_t i = 0; i < numOfCols; ++i) {
// SColumnInfoData* pCol = taosArrayGet(pTsdbReadHandle->pColumns, i);
// SColumnInfoData* pCol = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
//
// SColumnInfoData colInfo = {{0}, 0};
// colInfo.info = pCol->info;
......@@ -3280,7 +3277,7 @@ bool tsdbNextDataBlock(tsdbReaderT pHandle) {
// }
//
// for (int32_t i = 0; i < cond.numOfCols; ++i) {
// SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
// SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
// memcpy(&cond.colList[i], &pColInfoData->info, sizeof(SColumnInfo));
// }
//
......@@ -3342,7 +3339,7 @@ bool tsdbGetExternalRow(tsdbReaderT pHandle) {
int32_t numOfCols = (int32_t)QH_GET_NUM_OF_COLS(pTsdbReadHandle);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pColumns, i);
SColumnInfoData* pColInfoData = taosArrayGet(pTsdbReadHandle->pResBlock->pDataBlock, i);
SColumnInfoData* first = taosArrayGet(pTsdbReadHandle->prev, i);
memcpy(pColInfoData->pData, first->pData, pColInfoData->info.bytes);
......@@ -3617,13 +3614,13 @@ SArray* tsdbRetrieveDataBlock(tsdbReaderT* pTsdbReadHandle, SArray* pIdList) {
*/
STsdbReadHandle* pHandle = (STsdbReadHandle*)pTsdbReadHandle;
if (pHandle->cur.fid == INT32_MIN) {
return pHandle->pColumns;
return pHandle->pResBlock->pDataBlock;
} else {
STableBlockInfo* pBlockInfo = &pHandle->pDataBlockInfo[pHandle->cur.slot];
STableCheckInfo* pCheckInfo = pBlockInfo->pTableCheckInfo;
if (pHandle->cur.mixBlock) {
return pHandle->pColumns;
return pHandle->pResBlock->pDataBlock;
} else {
SDataBlockInfo binfo = GET_FILE_DATA_BLOCK_INFO(pCheckInfo, pBlockInfo->compBlock);
assert(pHandle->realNumOfRows <= binfo.rows);
......@@ -3633,7 +3630,7 @@ SArray* tsdbRetrieveDataBlock(tsdbReaderT* pTsdbReadHandle, SArray* pIdList) {
if (pBlockLoadInfo->slot == pHandle->cur.slot && pBlockLoadInfo->fileGroup->fid == pHandle->cur.fid &&
pBlockLoadInfo->uid == pCheckInfo->tableId) {
return pHandle->pColumns;
return pHandle->pResBlock->pDataBlock;
} else { // only load the file block
SBlock* pBlock = pBlockInfo->compBlock;
if (doLoadFileDataBlock(pHandle, pBlock, pCheckInfo, pHandle->cur.slot) != TSDB_CODE_SUCCESS) {
......@@ -3641,7 +3638,7 @@ SArray* tsdbRetrieveDataBlock(tsdbReaderT* pTsdbReadHandle, SArray* pIdList) {
}
int32_t numOfRows = doCopyRowsFromFileBlock(pHandle, pHandle->outputCapacity, 0, 0, pBlock->numOfRows - 1);
return pHandle->pColumns;
return pHandle->pResBlock->pDataBlock;
}
}
}
......@@ -3692,7 +3689,7 @@ void tsdbCleanupReadHandle(tsdbReaderT queryHandle) {
return;
}
pTsdbReadHandle->pColumns = doFreeColumnInfoData(pTsdbReadHandle->pColumns);
pTsdbReadHandle->pResBlock->pDataBlock = doFreeColumnInfoData(pTsdbReadHandle->pResBlock->pDataBlock);
taosArrayDestroy(pTsdbReadHandle->suppInfo.defaultLoadColumn);
taosMemoryFreeClear(pTsdbReadHandle->pDataBlockInfo);
......
source/dnode/vnode/test/tsdbSmaTest.cpp
浏览文件 @ 5cdef06f
......@@ -435,7 +435,7 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
SSDataBlock *pDataBlock = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
EXPECT_NE(pDataBlock, nullptr);
pDataBlock->pBlockAgg = NULL;
pDataBlock->info.numOfCols = tSmaNumOfCols;
taosArrayGetSize(pDataBlock->pDataBlock) = tSmaNumOfCols;
pDataBlock->info.rows = tSmaNumOfRows;
pDataBlock->info.groupId = tSmaGroupId + g;
......
source/libs/command/src/command.c
浏览文件 @ 5cdef06f
......@@ -32,31 +32,19 @@ static int32_t getSchemaBytes(const SSchema* pSchema) {
}
static SSDataBlock* buildDescResultDataBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = DESCRIBE_RESULT_COLS;
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
SSDataBlock* pBlock = createDataBlock();
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = DESCRIBE_RESULT_FIELD_LEN;
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, DESCRIBE_RESULT_FIELD_LEN, 1);
blockDataAppendColInfo(pBlock, &infoData);
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = DESCRIBE_RESULT_TYPE_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, DESCRIBE_RESULT_TYPE_LEN, 2);
blockDataAppendColInfo(pBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_INT;
infoData.info.bytes = tDataTypes[TSDB_DATA_TYPE_INT].bytes;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = DESCRIBE_RESULT_NOTE_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData = createColumnInfoData(TSDB_DATA_TYPE_INT, tDataTypes[TSDB_DATA_TYPE_INT].bytes, 3);
blockDataAppendColInfo(pBlock, &infoData);
infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, DESCRIBE_RESULT_NOTE_LEN, 4);
blockDataAppendColInfo(pBlock, &infoData);
return pBlock;
}
......@@ -127,22 +115,12 @@ static int32_t execResetQueryCache() { return catalogClearCache(); }
static SSDataBlock* buildCreateDBResultDataBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = SHOW_CREATE_DB_RESULT_COLS;
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_DB_RESULT_FIELD1_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_DB_RESULT_FIELD2_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
SSDataBlock* pBlock = createDataBlock();
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, SHOW_CREATE_DB_RESULT_COLS, 1);
blockDataAppendColInfo(pBlock, &infoData);
infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, SHOW_CREATE_DB_RESULT_FIELD2_LEN, 2);
blockDataAppendColInfo(pBlock, &infoData);
return pBlock;
}
......@@ -284,21 +262,13 @@ static int32_t execShowCreateDatabase(SShowCreateDatabaseStmt* pStmt, SRetrieveT
}
static SSDataBlock* buildCreateTbResultDataBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = SHOW_CREATE_TB_RESULT_COLS;
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_TB_RESULT_FIELD1_LEN;
SSDataBlock* pBlock = createDataBlock();
taosArrayPush(pBlock->pDataBlock, &infoData);
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, SHOW_CREATE_TB_RESULT_FIELD1_LEN, 1);
blockDataAppendColInfo(pBlock, &infoData);
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = SHOW_CREATE_TB_RESULT_FIELD2_LEN;
taosArrayPush(pBlock->pDataBlock, &infoData);
infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, SHOW_CREATE_TB_RESULT_FIELD2_LEN, 2);
blockDataAppendColInfo(pBlock, &infoData);
return pBlock;
}
......@@ -554,10 +524,9 @@ static int32_t execAlterLocal(SAlterLocalStmt* pStmt) { return TSDB_CODE_FAILED;
static SSDataBlock* buildLocalVariablesResultDataBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->info.numOfCols = SHOW_LOCAL_VARIABLES_RESULT_COLS;
pBlock->info.hasVarCol = true;
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
pBlock->pDataBlock = taosArrayInit(SHOW_LOCAL_VARIABLES_RESULT_COLS, sizeof(SColumnInfoData));
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
......
source/libs/command/src/explain.c
浏览文件 @ 5cdef06f
......@@ -1190,17 +1190,12 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
QRY_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
}
SSDataBlock *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_VARCHAR;
infoData.info.bytes = TSDB_EXPLAIN_RESULT_ROW_SIZE;
pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
taosArrayPush(pBlock->pDataBlock, &infoData);
SSDataBlock *pBlock = createDataBlock();
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_VARCHAR, TSDB_EXPLAIN_RESULT_ROW_SIZE, 1);
blockDataAppendColInfo(pBlock, &infoData);
blockDataEnsureCapacity(pBlock, rowNum);
SColumnInfoData* pInfoData = taosArrayGet(pBlock->pDataBlock, 0);
pInfoData->hasNull = false;
colInfoDataEnsureCapacity(pInfoData, 0, rowNum);
char buf[1024] = {0};
for (int32_t i = 0; i < rowNum; ++i) {
......@@ -1210,9 +1205,7 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
colDataAppend(pInfoData, i, buf, false);
}
pBlock->info.numOfCols = 1;
pBlock->info.rows = rowNum;
pBlock->info.hasVarCol = true;
int32_t rspSize = sizeof(SRetrieveTableRsp) + blockGetEncodeSize(pBlock);
......@@ -1226,7 +1219,7 @@ int32_t qExplainGetRspFromCtx(void *ctx, SRetrieveTableRsp **pRsp) {
rsp->numOfRows = htonl(rowNum);
int32_t len = 0;
blockCompressEncode(pBlock, rsp->data, &len, pBlock->info.numOfCols, 0);
blockCompressEncode(pBlock, rsp->data, &len, taosArrayGetSize(pBlock->pDataBlock), 0);
ASSERT(len == rspSize - sizeof(SRetrieveTableRsp));
rsp->compLen = htonl(len);
......
source/libs/executor/src/dataDeleter.c
浏览文件 @ 5cdef06f
......@@ -75,7 +75,7 @@ static void toDataCacheEntry(SDataDeleterHandle* pHandle, const SInputData* pInp
SDataCacheEntry* pEntry = (SDataCacheEntry*)pBuf->pData;
pEntry->compressed = 0;
pEntry->numOfRows = pInput->pData->info.rows;
pEntry->numOfCols = pInput->pData->info.numOfCols;
pEntry->numOfCols = taosArrayGetSize(pInput->pData->pDataBlock);
pEntry->dataLen = sizeof(SDeleterRes);
ASSERT(1 == pEntry->numOfRows);
......
source/libs/executor/src/executil.c
浏览文件 @ 5cdef06f
......@@ -183,34 +183,24 @@ SArray* createSortInfo(SNodeList* pNodeList) {
SSDataBlock* createResDataBlock(SDataBlockDescNode* pNode) {
int32_t numOfCols = LIST_LENGTH(pNode->pSlots);
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
SSDataBlock* pBlock = createDataBlock();
pBlock->info.blockId = pNode->dataBlockId;
pBlock->info.rowSize = pNode->totalRowSize; // todo ??
pBlock->info.type = STREAM_INVALID;
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {{0}};
SSlotDescNode* pDescNode = (SSlotDescNode*)nodesListGetNode(pNode->pSlots, i);
// if (!pDescNode->output) { // todo disable it temporarily
// continue;
// }
idata.info.type = pDescNode->dataType.type;
idata.info.bytes = pDescNode->dataType.bytes;
SColumnInfoData idata = createColumnInfoData(pDescNode->dataType.type, pDescNode->dataType.bytes, pDescNode->slotId);
idata.info.scale = pDescNode->dataType.scale;
idata.info.slotId = pDescNode->slotId;
idata.info.precision = pDescNode->dataType.precision;
if (IS_VAR_DATA_TYPE(idata.info.type)) {
pBlock->info.hasVarCol = true;
}
taosArrayPush(pBlock->pDataBlock, &idata);
blockDataAppendColInfo(pBlock, &idata);
}
pBlock->info.numOfCols = taosArrayGetSize(pBlock->pDataBlock);
return pBlock;
}
......@@ -698,7 +688,7 @@ void relocateColumnData(SSDataBlock* pBlock, const SArray* pColMatchInfo, SArray
if (p->info.colId == pmInfo->colId) {
SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, pmInfo->targetSlotId);
colDataAssign(pDst, p, pBlock->info.rows);
colDataAssign(pDst, p, pBlock->info.rows, &pBlock->info);
i++;
j++;
} else if (p->info.colId < pmInfo->colId) {
......
source/libs/executor/src/executorimpl.c
浏览文件 @ 5cdef06f
......@@ -356,7 +356,7 @@ void initExecTimeWindowInfo(SColumnInfoData* pColData, STimeWindow* pQueryWindow
pColData->info.type = TSDB_DATA_TYPE_TIMESTAMP;
pColData->info.bytes = sizeof(int64_t);
colInfoDataEnsureCapacity(pColData, 0, 5);
colInfoDataEnsureCapacity(pColData, 5);
colDataAppendInt64(pColData, 0, &pQueryWindow->skey);
colDataAppendInt64(pColData, 1, &pQueryWindow->ekey);
......@@ -458,7 +458,7 @@ static int32_t doCreateConstantValColumnInfo(SInputColumnInfoData* pInput, SFunc
pColInfo = pInput->pData[paramIndex];
}
colInfoDataEnsureCapacity(pColInfo, 0, numOfRows);
colInfoDataEnsureCapacity(pColInfo, numOfRows);
int8_t type = pFuncParam->param.nType;
if (type == TSDB_DATA_TYPE_BIGINT || type == TSDB_DATA_TYPE_UBIGINT) {
......@@ -579,7 +579,7 @@ int32_t projectApplyFunctions(SExprInfo* pExpr, SSDataBlock* pResult, SSDataBloc
colDataMergeCol(pColInfoData, pResult->info.rows, &pResult->info.capacity, pfCtx->input.pData[0],
pfCtx->input.numOfRows);
} else {
colDataAssign(pColInfoData, pfCtx->input.pData[0], pfCtx->input.numOfRows);
colDataAssign(pColInfoData, pfCtx->input.pData[0], pfCtx->input.numOfRows, &pResult->info);
}
numOfRows = pfCtx->input.numOfRows;
......@@ -1186,7 +1186,7 @@ int32_t loadDataBlockOnDemand(SExecTaskInfo* pTaskInfo, STableScanInfo* pTableSc
// }
// if (pQueryAttr->pFilters != NULL) {
// filterSetColFieldData(pQueryAttr->pFilters, pBlock->info.numOfCols, pBlock->pDataBlock);
// filterSetColFieldData(pQueryAttr->pFilters, taosArrayGetSize(pBlock->pDataBlock), pBlock->pDataBlock);
// }
// if (pQueryAttr->pFilters != NULL || pRuntimeEnv->pTsBuf != NULL) {
......@@ -1321,7 +1321,8 @@ void doFilter(const SNode* pFilterNode, SSDataBlock* pBlock) {
// todo move to the initialization function
int32_t code = filterInitFromNode((SNode*)pFilterNode, &filter, 0);
SFilterColumnParam param1 = {.numOfCols = pBlock->info.numOfCols, .pDataBlock = pBlock->pDataBlock};
size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
SFilterColumnParam param1 = {.numOfCols = numOfCols, .pDataBlock = pBlock->pDataBlock};
code = filterSetDataFromSlotId(filter, &param1);
int8_t* rowRes = NULL;
......@@ -1343,7 +1344,8 @@ void extractQualifiedTupleByFilterResult(SSDataBlock* pBlock, const int8_t* rowR
int32_t totalRows = pBlock->info.rows;
SSDataBlock* px = createOneDataBlock(pBlock, true);
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pSrc = taosArrayGet(px->pDataBlock, i);
SColumnInfoData* pDst = taosArrayGet(pBlock->pDataBlock, i);
// it is a reserved column for scalar function, and no data in this column yet.
......@@ -2045,8 +2047,9 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLo
int32_t compLen, int32_t numOfOutput, int64_t startTs, uint64_t* total,
SArray* pColList) {
if (pColList == NULL) { // data from other sources
blockDataCleanup(pRes);
// blockDataEnsureCapacity(pRes, numOfRows);
blockCompressDecode(pRes, numOfOutput, numOfRows, pData);
pRes->info.rows = numOfRows;
} else { // extract data according to pColList
ASSERT(numOfOutput == taosArrayGetSize(pColList));
char* pStart = pData;
......@@ -2064,32 +2067,19 @@ int32_t extractDataBlockFromFetchRsp(SSDataBlock* pRes, SLoadRemoteDataInfo* pLo
pStart += sizeof(SSysTableSchema);
}
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
pBlock->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
pBlock->info.numOfCols = numOfCols;
SSDataBlock* pBlock = createDataBlock();
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData idata = {0};
idata.info.type = pSchema[i].type;
idata.info.bytes = pSchema[i].bytes;
idata.info.colId = pSchema[i].colId;
idata.hasNull = true;
taosArrayPush(pBlock->pDataBlock, &idata);
if (IS_VAR_DATA_TYPE(idata.info.type)) {
pBlock->info.hasVarCol = true;
}
SColumnInfoData idata = createColumnInfoData(pSchema[i].type, pSchema[i].bytes, pSchema[i].colId);
blockDataAppendColInfo(pBlock, &idata);
}
blockCompressDecode(pBlock, numOfCols, numOfRows, pStart);
blockDataEnsureCapacity(pRes, numOfRows);
// data from mnode
pRes->info.rows = numOfRows;
relocateColumnData(pRes, pColList, pBlock->pDataBlock, false);
taosArrayDestroy(pBlock->pDataBlock);
taosMemoryFree(pBlock);
// blockDataDestroy(pBlock);
blockDataDestroy(pBlock);
}
// todo move this to time window aggregator, since the primary timestamp may not be known by exchange operator.
......@@ -2412,15 +2402,15 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
tsem_init(&pInfo->ready, 0, 0);
pInfo->seqLoadData = false;
pInfo->seqLoadData = false;
pInfo->pTransporter = pTransporter;
pInfo->pResult = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
pOperator->name = "ExchangeOperator";
pInfo->pResult = createResDataBlock(pExNode->node.pOutputDataBlockDesc);
pOperator->name = "ExchangeOperator";
pOperator->operatorType = QUERY_NODE_PHYSICAL_PLAN_EXCHANGE;
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->exprSupp.numOfExprs = pInfo->pResult->info.numOfCols;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pResult->pDataBlock);
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, NULL,
......@@ -2542,9 +2532,6 @@ static void doMergeImpl(SOperatorInfo* pOperator, int32_t numOfExpr, SSDataBlock
SSortedMergeOperatorInfo* pInfo = pOperator->info;
SqlFunctionCtx* pCtx = pOperator->exprSupp.pCtx;
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
// pCtx[i].size = 1;
}
for (int32_t i = 0; i < pBlock->info.rows; ++i) {
if (!pInfo->hasGroupVal) {
......@@ -2653,7 +2640,7 @@ SSDataBlock* getSortedMergeBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlo
SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId);
SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->targetSlotId);
colDataAssign(pDst, pSrc, p->info.rows);
colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
}
pDataBlock->info.rows = p->info.rows;
......
source/libs/executor/src/groupoperator.c
浏览文件 @ 5cdef06f
......@@ -559,8 +559,8 @@ uint64_t calcGroupId(char* pData, int32_t len) {
}
int32_t* setupColumnOffset(const SSDataBlock* pBlock, int32_t rowCapacity) {
size_t numOfCols = pBlock->info.numOfCols;
int32_t* offset = taosMemoryCalloc(pBlock->info.numOfCols, sizeof(int32_t));
size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
int32_t* offset = taosMemoryCalloc(numOfCols, sizeof(int32_t));
offset[0] = sizeof(int32_t) + sizeof(uint64_t); // the number of rows in current page, ref to SSDataBlock paged serialization format
......
source/libs/executor/src/scanoperator.c
浏览文件 @ 5cdef06f
......@@ -203,7 +203,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
pCost->skipBlocks += 1;
// clear all data in pBlock that are set when handing the previous block
for (int32_t i = 0; i < pBlockInfo->numOfCols; ++i) {
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i);
pcol->pData = NULL;
}
......@@ -217,7 +217,7 @@ static int32_t loadDataBlock(SOperatorInfo* pOperator, STableScanInfo* pTableSca
tsdbRetrieveDataBlockStatisInfo(pTableScanInfo->dataReader, &pColAgg, &allColumnsHaveAgg);
if (allColumnsHaveAgg == true) {
int32_t numOfCols = pBlock->info.numOfCols;
int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
// todo create this buffer during creating operator
if (pBlock->pBlockAgg == NULL) {
......@@ -316,7 +316,7 @@ void addTagPseudoColumnData(SReadHandle* pHandle, SExprInfo* pPseudoExpr, int32_
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, dstSlotId);
colInfoDataEnsureCapacity(pColInfoData, 0, pBlock->info.rows);
colInfoDataEnsureCapacity(pColInfoData, pBlock->info.rows);
colInfoDataCleanup(pColInfoData, pBlock->info.rows);
int32_t functionId = pExpr->pExpr->_function.functionId;
......@@ -355,10 +355,8 @@ void setTbNameColData(void* pMeta, const SSDataBlock* pBlock, SColumnInfoData* p
struct SScalarFuncExecFuncs fpSet = {0};
fmGetScalarFuncExecFuncs(functionId, &fpSet);
SColumnInfoData infoData = {0};
infoData.info.type = TSDB_DATA_TYPE_BIGINT;
infoData.info.bytes = sizeof(uint64_t);
colInfoDataEnsureCapacity(&infoData, 0, 1);
SColumnInfoData infoData = createColumnInfoData(TSDB_DATA_TYPE_BIGINT, sizeof(uint64_t), 1);
colInfoDataEnsureCapacity(&infoData, 1);
colDataAppendInt64(&infoData, 0, (int64_t*)&pBlock->info.uid);
SScalarParam srcParam = {.numOfRows = pBlock->info.rows, .param = pMeta, .columnData = &infoData};
......@@ -786,7 +784,7 @@ static bool prepareDataScan(SStreamBlockScanInfo* pInfo) {
}
static void copyOneRow(SSDataBlock* dest, SSDataBlock* source, int32_t sourceRowId) {
for (int32_t j = 0; j < source->info.numOfCols; j++) {
for (int32_t j = 0; j < taosArrayGetSize(source->pDataBlock); j++) {
SColumnInfoData* pDestCol = (SColumnInfoData*)taosArrayGet(dest->pDataBlock, j);
SColumnInfoData* pSourceCol = (SColumnInfoData*)taosArrayGet(source->pDataBlock, j);
if (colDataIsNull_s(pSourceCol, sourceRowId)) {
......@@ -858,7 +856,6 @@ static void setUpdateData(SStreamBlockScanInfo* pInfo, SSDataBlock* pBlock, SSDa
SColumnInfoData* pCol = (SColumnInfoData*)taosArrayGet(pUpdateBlock->pDataBlock, pInfo->primaryTsIndex);
ASSERT(pCol->info.type == TSDB_DATA_TYPE_TIMESTAMP);
blockDataEnsureCapacity(pUpdateBlock, size);
ASSERT(pBlock->info.numOfCols == pUpdateBlock->info.numOfCols);
int32_t rowId = *(int32_t*)taosArrayGet(pInfo->tsArray, pInfo->tsArrayIndex);
pInfo->groupId = getGroupId(pInfo->pSnapshotReadOp, pBlock, rowId);
......@@ -969,13 +966,13 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
blockDataCleanup(pInfo->pRes);
while (tqNextDataBlock(pInfo->streamBlockReader)) {
SArray* pCols = NULL;
SSDataBlock block = {0};
uint64_t groupId = 0;
uint64_t uid = 0;
int32_t numOfRows = 0;
int16_t outputCol = 0;
int32_t code = tqRetrieveDataBlock(&pCols, pInfo->streamBlockReader, &groupId, &uid, &numOfRows, &outputCol);
// todo refactor
int32_t code = tqRetrieveDataBlock(&block, pInfo->streamBlockReader, &groupId, &uid, &numOfRows);
if (code != TSDB_CODE_SUCCESS || numOfRows == 0) {
pTaskInfo->code = code;
......@@ -986,6 +983,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
pInfo->pRes->info.rows = numOfRows;
pInfo->pRes->info.uid = uid;
pInfo->pRes->info.type = STREAM_NORMAL;
pInfo->pRes->info.capacity = numOfRows;
// for generating rollup SMA result, each time is an independent time serie.
// TODO temporarily used, when the statement of "partition by tbname" is ready, remove this
......@@ -1000,6 +998,7 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
pInfo->pRes->info.groupId = *groupIdPre;
}
// todo extract method
for (int32_t i = 0; i < taosArrayGetSize(pInfo->pColMatchInfo); ++i) {
SColMatchInfo* pColMatchInfo = taosArrayGet(pInfo->pColMatchInfo, i);
if (!pColMatchInfo->output) {
......@@ -1007,8 +1006,8 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
}
bool colExists = false;
for (int32_t j = 0; j < taosArrayGetSize(pCols); ++j) {
SColumnInfoData* pResCol = taosArrayGet(pCols, j);
for (int32_t j = 0; j < blockDataGetNumOfCols(&block); ++j) {
SColumnInfoData* pResCol = bdGetColumnInfoData(&block, j);
if (pResCol->info.colId == pColMatchInfo->colId) {
taosArraySet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId, pResCol);
colExists = true;
......@@ -1019,7 +1018,6 @@ static SSDataBlock* doStreamBlockScan(SOperatorInfo* pOperator) {
// the required column does not exists in submit block, let's set it to be all null value
if (!colExists) {
SColumnInfoData* pDst = taosArrayGet(pInfo->pRes->pDataBlock, pColMatchInfo->targetSlotId);
colInfoDataEnsureCapacity(pDst, 0, pBlockInfo->rows);
colDataAppendNNULL(pDst, 0, pBlockInfo->rows);
}
}
......@@ -1173,7 +1171,7 @@ SOperatorInfo* createStreamScanOperatorInfo(void* pDataReader, SReadHandle* pHan
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->exprSupp.numOfExprs = pInfo->pRes->info.numOfCols;
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pRes->pDataBlock);
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet =
......@@ -1301,7 +1299,7 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) {
SColumnInfoData* pSrc = taosArrayGet(pInfo->pRes->pDataBlock, i);
if (keep) {
colDataAssign(pDest, pSrc, pInfo->pRes->info.rows);
colDataAssign(pDest, pSrc, pInfo->pRes->info.rows, &px->info);
numOfRow = pInfo->pRes->info.rows;
} else if (NULL != rowRes) {
numOfRow = 0;
......@@ -1331,8 +1329,6 @@ static SSDataBlock* doFilterResult(SSysTableScanInfo* pInfo) {
}
static SSDataBlock* buildSysTableMetaBlock() {
SSDataBlock* pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
size_t size = 0;
const SSysTableMeta* pMeta = NULL;
getInfosDbMeta(&pMeta, &size);
......@@ -1345,19 +1341,12 @@ static SSDataBlock* buildSysTableMetaBlock() {
}
}
pBlock->pDataBlock = taosArrayInit(pBlock->info.numOfCols, sizeof(SColumnInfoData));
SSDataBlock* pBlock = createDataBlock();
for (int32_t i = 0; i < pMeta[index].colNum; ++i) {
SColumnInfoData colInfoData = {0};
colInfoData.info.colId = i + 1;
colInfoData.info.type = pMeta[index].schema[i].type;
colInfoData.info.bytes = pMeta[index].schema[i].bytes;
taosArrayPush(pBlock->pDataBlock, &colInfoData);
SColumnInfoData colInfoData = createColumnInfoData(pMeta[index].schema[i].type, pMeta[index].schema[i].bytes, i + 1);
blockDataAppendColInfo(pBlock, &colInfoData);
}
pBlock->info.numOfCols = pMeta[index].colNum;
pBlock->info.hasVarCol = true;
return pBlock;
}
......@@ -1388,7 +1377,6 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
}
blockDataCleanup(pInfo->pRes);
int32_t numOfRows = 0;
const char* db = NULL;
......@@ -1584,7 +1572,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
if (pRsp->numOfRows == 0 || pRsp->completed) {
pOperator->status = OP_EXEC_DONE;
qDebug("%s load meta data from mnode completed, rowsOfSource:%d, totalRows:%" PRIu64 " ", GET_TASKID(pTaskInfo),
qDebug("%s load meta data from mnode completed, rowsOfSource:%d, totalRows:%" PRIu64, GET_TASKID(pTaskInfo),
pRsp->numOfRows, pInfo->loadInfo.totalRows);
if (pRsp->numOfRows == 0) {
......@@ -1711,7 +1699,7 @@ SOperatorInfo* createSysTableScanOperatorInfo(void* readHandle, SSystemTableScan
pOperator->blocking = false;
pOperator->status = OP_NOT_OPENED;
pOperator->info = pInfo;
pOperator->exprSupp.numOfExprs = pResBlock->info.numOfCols;
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pResBlock->pDataBlock);
pOperator->pTaskInfo = pTaskInfo;
pOperator->fpSet =
......@@ -2049,6 +2037,7 @@ int32_t createMultipleDataReaders(SQueryTableDataCond* pQueryCond, SReadHandle*
return TSDB_CODE_SUCCESS;
}
// todo refactor
static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeScanInfo* pTableScanInfo,
int32_t readerIdx, SSDataBlock* pBlock, uint32_t* status) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
......@@ -2079,7 +2068,7 @@ static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeSc
pCost->skipBlocks += 1;
// clear all data in pBlock that are set when handing the previous block
for (int32_t i = 0; i < pBlockInfo->numOfCols; ++i) {
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData* pcol = taosArrayGet(pBlock->pDataBlock, i);
pcol->pData = NULL;
}
......@@ -2094,7 +2083,7 @@ static int32_t loadDataBlockFromOneTable(SOperatorInfo* pOperator, STableMergeSc
tsdbRetrieveDataBlockStatisInfo(reader, &pColAgg, &allColumnsHaveAgg);
if (allColumnsHaveAgg == true) {
int32_t numOfCols = pBlock->info.numOfCols;
int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
// todo create this buffer during creating operator
if (pBlock->pBlockAgg == NULL) {
......@@ -2189,7 +2178,13 @@ static SSDataBlock* getTableDataBlock(void* param) {
continue;
}
tsdbRetrieveDataBlockInfo(reader, &pBlock->info);
blockDataCleanup(pBlock);
SDataBlockInfo binfo = pBlock->info;
tsdbRetrieveDataBlockInfo(reader, &binfo);
binfo.capacity = binfo.rows;
blockDataEnsureCapacity(pBlock, binfo.capacity);
pBlock->info = binfo;
uint32_t status = 0;
int32_t code = loadDataBlockFromOneTable(pOperator, pTableScanInfo, readerIdx, pBlock, &status);
......
source/libs/executor/src/sortoperator.c
浏览文件 @ 5cdef06f
......@@ -80,7 +80,7 @@ _error:
}
void appendOneRowToDataBlock(SSDataBlock* pBlock, STupleHandle* pTupleHandle) {
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
bool isNull = tsortIsNullVal(pTupleHandle, i);
if (isNull) {
......@@ -120,6 +120,9 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
}
if (p->info.rows > 0) {
blockDataEnsureCapacity(pDataBlock, capacity);
// todo extract function to handle this
int32_t numOfCols = taosArrayGetSize(pColMatchInfo);
for (int32_t i = 0; i < numOfCols; ++i) {
SColMatchInfo* pmInfo = taosArrayGet(pColMatchInfo, i);
......@@ -127,11 +130,10 @@ SSDataBlock* getSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, i
SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId);
SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->targetSlotId);
colDataAssign(pDst, pSrc, p->info.rows);
colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
}
pDataBlock->info.rows = p->info.rows;
pDataBlock->info.capacity = p->info.rows;
}
blockDataDestroy(p);
......@@ -257,6 +259,7 @@ typedef struct SGroupSortOperatorInfo {
SSDataBlock* getGroupSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlock, int32_t capacity,
SArray* pColMatchInfo, SGroupSortOperatorInfo* pInfo) {
blockDataCleanup(pDataBlock);
blockDataEnsureCapacity(pDataBlock, capacity);
SSDataBlock* p = tsortGetSortedDataBlock(pHandle);
if (p == NULL) {
......@@ -285,7 +288,7 @@ SSDataBlock* getGroupSortedBlockData(SSortHandle* pHandle, SSDataBlock* pDataBlo
SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId);
SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->targetSlotId);
colDataAssign(pDst, pSrc, p->info.rows);
colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
}
pDataBlock->info.rows = p->info.rows;
......@@ -564,7 +567,8 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
}
}
if (p->info.rows > 0) {
if (p->info.rows > 0) {// todo extract method
blockDataEnsureCapacity(pDataBlock, p->info.rows);
int32_t numOfCols = taosArrayGetSize(pColMatchInfo);
for (int32_t i = 0; i < numOfCols; ++i) {
SColMatchInfo* pmInfo = taosArrayGet(pColMatchInfo, i);
......@@ -572,11 +576,10 @@ SSDataBlock* getMultiwaySortedBlockData(SSortHandle* pHandle, SSDataBlock* pData
SColumnInfoData* pSrc = taosArrayGet(p->pDataBlock, pmInfo->srcSlotId);
SColumnInfoData* pDst = taosArrayGet(pDataBlock->pDataBlock, pmInfo->targetSlotId);
colDataAssign(pDst, pSrc, p->info.rows);
colDataAssign(pDst, pSrc, p->info.rows, &pDataBlock->info);
}
pDataBlock->info.rows = p->info.rows;
pDataBlock->info.capacity = p->info.rows;
pDataBlock->info.groupId = pInfo->groupId;
}
......
source/libs/executor/src/tfill.c
浏览文件 @ 5cdef06f
......@@ -50,7 +50,7 @@ static void setTagsValue(SFillInfo* pFillInfo, void** data, int32_t genRows) {
static void setNullRow(SSDataBlock* pBlock, int32_t numOfCol, int32_t rowIndex) {
// the first are always the timestamp column, so start from the second column.
for (int32_t i = 1; i < pBlock->info.numOfCols; ++i) {
for (int32_t i = 1; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData* p = taosArrayGet(pBlock->pDataBlock, i);
colDataAppendNULL(p, rowIndex);
}
......
source/libs/executor/src/timewindowoperator.c
浏览文件 @ 5cdef06f
......@@ -1814,7 +1814,7 @@ static int32_t initPrevRowsKeeper(STimeSliceOperatorInfo* pInfo, SSDataBlock* pB
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t numOfCols = pBlock->info.numOfCols;
int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
......@@ -2231,9 +2231,6 @@ static void clearStreamIntervalOperator(SStreamFinalIntervalOperatorInfo* pInfo)
}
static void clearUpdateDataBlock(SSDataBlock* pBlock) {
if (pBlock->info.rows <= 0) {
return;
}
blockDataCleanup(pBlock);
}
......@@ -2243,12 +2240,14 @@ void copyUpdateDataBlock(SSDataBlock* pDest, SSDataBlock* pSource, int32_t tsCol
clearUpdateDataBlock(pDest);
SColumnInfoData* pDestCol = taosArrayGet(pDest->pDataBlock, 0);
SColumnInfoData* pSourceCol = taosArrayGet(pSource->pDataBlock, tsColIndex);
// copy timestamp column
colDataAssign(pDestCol, pSourceCol, pSource->info.rows);
for (int32_t i = 1; i < pDest->info.numOfCols; i++) {
colDataAssign(pDestCol, pSourceCol, pSource->info.rows, &pDest->info);
for (int32_t i = 1; i < taosArrayGetSize(pDest->pDataBlock); i++) {
SColumnInfoData* pCol = taosArrayGet(pDest->pDataBlock, i);
colDataAppendNNULL(pCol, 0, pSource->info.rows);
}
pDest->info.rows = pSource->info.rows;
pDest->info.groupId = pSource->info.groupId;
pDest->info.type = pSource->info.type;
......@@ -2918,7 +2917,7 @@ void doBuildDeleteDataBlock(SHashObj* pStDeleted, SSDataBlock* pBlock, void** It
while (((*Ite) = taosHashIterate(pStDeleted, *Ite)) != NULL) {
SColumnInfoData* pColInfoData = taosArrayGet(pBlock->pDataBlock, 0);
colDataAppend(pColInfoData, pBlock->info.rows, *Ite, false);
for (int32_t i = 1; i < pBlock->info.numOfCols; i++) {
for (int32_t i = 1; i < taosArrayGetSize(pBlock->pDataBlock); i++) {
pColInfoData = taosArrayGet(pBlock->pDataBlock, i);
colDataAppendNULL(pColInfoData, pBlock->info.rows);
}
......
source/libs/executor/src/tsort.c
浏览文件 @ 5cdef06f
......@@ -147,7 +147,7 @@ static int32_t doAddNewExternalMemSource(SDiskbasedBuf *pBuf, SArray* pAllSource
int32_t rowSize = blockDataGetSerialRowSize(pSource->src.pBlock);
// The value of numOfRows must be greater than 0, which is guaranteed by the previous memory allocation
int32_t numOfRows = (getBufPageSize(pBuf) - blockDataGetSerialMetaSize(pBlock->info.numOfCols))/rowSize;
int32_t numOfRows = (getBufPageSize(pBuf) - blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock)))/rowSize;
ASSERT(numOfRows > 0);
return blockDataEnsureCapacity(pSource->src.pBlock, numOfRows);
}
......@@ -178,7 +178,7 @@ static int32_t doAddToBuf(SSDataBlock* pDataBlock, SSortHandle* pHandle) {
return terrno;
}
int32_t size = blockDataGetSize(p) + sizeof(int32_t) + p->info.numOfCols * sizeof(int32_t);
int32_t size = blockDataGetSize(p) + sizeof(int32_t) + taosArrayGetSize(p->pDataBlock) * sizeof(int32_t);
assert(size <= getBufPageSize(pHandle->pBuf));
blockDataToBuf(pPage, p);
......@@ -252,7 +252,7 @@ static int32_t sortComparInit(SMsortComparParam* cmpParam, SArray* pSources, int
}
static void appendOneRowToDataBlock(SSDataBlock *pBlock, const SSDataBlock* pSource, int32_t* rowIndex) {
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData* pColInfo = taosArrayGet(pBlock->pDataBlock, i);
SColumnInfoData* pSrcColInfo = taosArrayGet(pSource->pDataBlock, i);
......@@ -500,7 +500,7 @@ static int32_t doInternalMergeSort(SSortHandle* pHandle) {
return terrno;
}
int32_t size = blockDataGetSize(pDataBlock) + sizeof(int32_t) + pDataBlock->info.numOfCols * sizeof(int32_t);
int32_t size = blockDataGetSize(pDataBlock) + sizeof(int32_t) + taosArrayGetSize(pDataBlock->pDataBlock) * sizeof(int32_t);
assert(size <= getBufPageSize(pHandle->pBuf));
blockDataToBuf(pPage, pDataBlock);
......
source/libs/executor/test/executorTests.cpp
浏览文件 @ 5cdef06f
......@@ -62,18 +62,11 @@ SSDataBlock* getDummyBlock(SOperatorInfo* pOperator) {
}
if (pInfo->pBlock == NULL) {
pInfo->pBlock = static_cast<SSDataBlock*>(taosMemoryCalloc(1, sizeof(SSDataBlock)));
pInfo->pBlock = createDataBlock();
pInfo->pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
SColumnInfoData colInfo = {0};
colInfo.info.type = TSDB_DATA_TYPE_INT;
colInfo.info.bytes = sizeof(int32_t);
colInfo.info.colId = 1;
colInfo.pData = static_cast<char*>(taosMemoryCalloc(pInfo->numOfRowsPerPage, sizeof(int32_t)));
colInfo.nullbitmap = static_cast<char*>(taosMemoryCalloc(1, (pInfo->numOfRowsPerPage + 7) / 8));
taosArrayPush(pInfo->pBlock->pDataBlock, &colInfo);
SColumnInfoData colInfo = createColumnInfoData(TSDB_DATA_TYPE_INT, sizeof(int32_t), 1);
blockDataAppendColInfo(pInfo->pBlock, &colInfo);
blockDataEnsureCapacity(pInfo->pBlock, pInfo->numOfRowsPerPage);
// SColumnInfoData colInfo1 = {0};
// colInfo1.info.type = TSDB_DATA_TYPE_BINARY;
......@@ -115,7 +108,6 @@ SSDataBlock* getDummyBlock(SOperatorInfo* pOperator) {
}
pBlock->info.rows = pInfo->numOfRowsPerPage;
pBlock->info.numOfCols = 1;
pInfo->current += 1;
return pBlock;
......@@ -128,28 +120,15 @@ SSDataBlock* get2ColsDummyBlock(SOperatorInfo* pOperator) {
}
if (pInfo->pBlock == NULL) {
pInfo->pBlock = static_cast<SSDataBlock*>(taosMemoryCalloc(1, sizeof(SSDataBlock)));
pInfo->pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
SColumnInfoData colInfo = {0};
colInfo.info.type = TSDB_DATA_TYPE_TIMESTAMP;
colInfo.info.bytes = sizeof(int64_t);
colInfo.info.colId = 1;
colInfo.pData = static_cast<char*>(taosMemoryCalloc(pInfo->numOfRowsPerPage, sizeof(int64_t)));
// colInfo.nullbitmap = static_cast<char*>(taosMemoryCalloc(1, (pInfo->numOfRowsPerPage + 7) / 8));
taosArrayPush(pInfo->pBlock->pDataBlock, &colInfo);
pInfo->pBlock = createDataBlock();
SColumnInfoData colInfo1 = {0};
colInfo1.info.type = TSDB_DATA_TYPE_INT;
colInfo1.info.bytes = 4;
colInfo1.info.colId = 2;
SColumnInfoData colInfo = createColumnInfoData(TSDB_DATA_TYPE_TIMESTAMP, sizeof(int64_t), 1);
blockDataAppendColInfo(pInfo->pBlock, &colInfo);
colInfo1.pData = static_cast<char*>(taosMemoryCalloc(pInfo->numOfRowsPerPage, sizeof(int32_t)));
colInfo1.nullbitmap = static_cast<char*>(taosMemoryCalloc(1, (pInfo->numOfRowsPerPage + 7) / 8));
SColumnInfoData colInfo1 = createColumnInfoData(TSDB_DATA_TYPE_INT, 4, 2);
blockDataAppendColInfo(pInfo->pBlock, &colInfo1);
taosArrayPush(pInfo->pBlock->pDataBlock, &colInfo1);
blockDataEnsureCapacity(pInfo->pBlock, pInfo->numOfRowsPerPage);
} else {
blockDataCleanup(pInfo->pBlock);
}
......@@ -185,7 +164,6 @@ SSDataBlock* get2ColsDummyBlock(SOperatorInfo* pOperator) {
}
pBlock->info.rows = pInfo->numOfRowsPerPage;
pBlock->info.numOfCols = 1;
pInfo->current += 1;
......
source/libs/executor/test/sortTests.cpp
浏览文件 @ 5cdef06f
......@@ -62,25 +62,21 @@ SSDataBlock* getSingleColDummyBlock(void* param) {
return NULL;
}
SSDataBlock* pBlock = static_cast<SSDataBlock*>(taosMemoryCalloc(1, sizeof(SSDataBlock)));
pBlock->pDataBlock = taosArrayInit(4, sizeof(SColumnInfoData));
SSDataBlock* pBlock = createDataBlock();
SColumnInfoData colInfo = {0};
colInfo.info.type = pInfo->type;
if (pInfo->type == TSDB_DATA_TYPE_NCHAR){
colInfo.info.bytes = TSDB_NCHAR_SIZE * VARCOUNT + VARSTR_HEADER_SIZE;
colInfo.varmeta.offset = static_cast<int32_t *>(taosMemoryCalloc(pInfo->pageRows, sizeof(int32_t)));
} else if(pInfo->type == TSDB_DATA_TYPE_BINARY) {
colInfo.info.bytes = VARCOUNT + VARSTR_HEADER_SIZE;
colInfo.varmeta.offset = static_cast<int32_t *>(taosMemoryCalloc(pInfo->pageRows, sizeof(int32_t)));
} else{
colInfo.info.bytes = tDataTypes[pInfo->type].bytes;
colInfo.pData = static_cast<char*>(taosMemoryCalloc(pInfo->pageRows, colInfo.info.bytes));
colInfo.nullbitmap = static_cast<char*>(taosMemoryCalloc(1, (pInfo->pageRows + 7) / 8));
}
colInfo.info.colId = 1;
taosArrayPush(pBlock->pDataBlock, &colInfo);
blockDataAppendColInfo(pBlock, &colInfo);
blockDataEnsureCapacity(pBlock, pInfo->pageRows);
for (int32_t i = 0; i < pInfo->pageRows; ++i) {
SColumnInfoData* pColInfo = static_cast<SColumnInfoData*>(TARRAY_GET_ELEM(pBlock->pDataBlock, 0));
......@@ -128,7 +124,6 @@ SSDataBlock* getSingleColDummyBlock(void* param) {
}
pBlock->info.rows = pInfo->pageRows;
pBlock->info.numOfCols = 1;
return pBlock;
}
......@@ -354,15 +349,10 @@ TEST(testCase, ordered_merge_sort_Test) {
SArray* orderInfo = taosArrayInit(1, sizeof(SBlockOrderInfo));
taosArrayPush(orderInfo, &oi);
SSDataBlock* pBlock = static_cast<SSDataBlock*>(taosMemoryCalloc(1, sizeof(SSDataBlock)));
pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
pBlock->info.numOfCols = 1;
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
SColumnInfoData colInfo = {0};
colInfo.info.type = TSDB_DATA_TYPE_INT;
colInfo.info.bytes = sizeof(int32_t);
colInfo.info.colId = 1;
taosArrayPush(pBlock->pDataBlock, &colInfo);
SSDataBlock* pBlock = createDataBlock();
for (int32_t i = 0; i < 1; ++i) {
SColumnInfoData colInfo = createColumnInfoData(TSDB_DATA_TYPE_INT, sizeof(int32_t), 1);
blockDataAppendColInfo(pBlock, &colInfo);
}
SSortHandle* phandle = tsortCreateSortHandle(orderInfo, SORT_MULTISOURCE_MERGE, 1024, 5, pBlock,"test_abc");
......
source/libs/function/inc/taggfunction.h
浏览文件 @ 5cdef06f
......@@ -28,19 +28,6 @@ extern "C" {
#include "function.h"
#include "tudf.h"
extern SAggFunctionInfo aggFunc[35];
#define FUNCSTATE_SO 0x0u
#define FUNCSTATE_MO 0x1u // dynamic number of output, not multinumber of output e.g., TOP/BOTTOM
#define FUNCSTATE_STREAM 0x2u // function avail for stream
#define FUNCSTATE_STABLE 0x4u // function avail for super table
#define FUNCSTATE_NEED_TS 0x8u // timestamp is required during query processing
#define FUNCSTATE_SELECTIVITY 0x10u // selectivity functions, can exists along with tag columns
#define BASIC_FUNC_SO FUNCSTATE_SO | FUNCSTATE_STREAM | FUNCSTATE_STABLE
#define BASIC_FUNC_MO FUNCSTATE_MO | FUNCSTATE_STREAM | FUNCSTATE_STABLE
#define AVG_FUNCTION_INTER_BUFFER_SIZE 50
#define DATA_SET_FLAG ',' // to denote the output area has data, not null value
......
source/libs/function/src/builtinsimpl.c
浏览文件 @ 5cdef06f
......@@ -1449,11 +1449,13 @@ void setNullSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, int32_t
void setSelectivityValue(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, const STuplePos* pTuplePos, int32_t rowIndex) {
int32_t pageId = pTuplePos->pageId;
int32_t offset = pTuplePos->offset;
if (pTuplePos->pageId != -1) {
int32_t numOfCols = taosArrayGetSize(pCtx->pSrcBlock->pDataBlock);
SFilePage* pPage = getBufPage(pCtx->pBuf, pageId);
bool* nullList = (bool*)((char*)pPage + offset);
char* pStart = (char*)(nullList + pCtx->pSrcBlock->info.numOfCols * sizeof(bool));
char* pStart = (char*)(nullList + numOfCols * sizeof(bool));
// todo set the offset value to optimize the performance.
for (int32_t j = 0; j < pCtx->subsidiaries.num; ++j) {
......@@ -3038,7 +3040,7 @@ void doAddIntoResult(SqlFunctionCtx* pCtx, void* pData, int32_t rowIndex, SSData
void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos) {
SFilePage* pPage = NULL;
int32_t completeRowSize = pSrcBlock->info.rowSize + pSrcBlock->info.numOfCols * sizeof(bool);
int32_t completeRowSize = pSrcBlock->info.rowSize + (int32_t) taosArrayGetSize(pSrcBlock->pDataBlock) * sizeof(bool);
if (pCtx->curBufPage == -1) {
pPage = getNewBufPage(pCtx->pBuf, 0, &pCtx->curBufPage);
......@@ -3056,8 +3058,8 @@ void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
// keep the current row data, extract method
int32_t offset = 0;
bool* nullList = (bool*)((char*)pPage + pPage->num);
char* pStart = (char*)(nullList + sizeof(bool) * pSrcBlock->info.numOfCols);
for (int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
char* pStart = (char*)(nullList + sizeof(bool) * (int32_t) taosArrayGetSize(pSrcBlock->pDataBlock));
for (int32_t i = 0; i < (int32_t) taosArrayGetSize(pSrcBlock->pDataBlock); ++i) {
SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
bool isNull = colDataIsNull_s(pCol, rowIndex);
if (isNull) {
......@@ -3086,11 +3088,13 @@ void saveTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pS
void copyTupleData(SqlFunctionCtx* pCtx, int32_t rowIndex, const SSDataBlock* pSrcBlock, STuplePos* pPos) {
SFilePage* pPage = getBufPage(pCtx->pBuf, pPos->pageId);
int32_t numOfCols = taosArrayGetSize(pSrcBlock->pDataBlock);
bool* nullList = (bool*)((char*)pPage + pPos->offset);
char* pStart = (char*)(nullList + pSrcBlock->info.numOfCols * sizeof(bool));
char* pStart = (char*)(nullList + numOfCols * sizeof(bool));
int32_t offset = 0;
for (int32_t i = 0; i < pSrcBlock->info.numOfCols; ++i) {
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pCol = taosArrayGet(pSrcBlock->pDataBlock, i);
if ((nullList[i] = colDataIsNull_s(pCol, rowIndex)) == true) {
offset += pCol->info.bytes;
......
source/libs/function/src/tudf.c
浏览文件 @ 5cdef06f
......@@ -763,8 +763,8 @@ void freeUdfInterBuf(SUdfInterBuf *buf) {
int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlock) {
udfBlock->numOfRows = block->info.rows;
udfBlock->numOfCols = block->info.numOfCols;
udfBlock->udfCols = taosMemoryCalloc(udfBlock->numOfCols, sizeof(SUdfColumn*));
udfBlock->numOfCols = taosArrayGetSize(block->pDataBlock);
udfBlock->udfCols = taosMemoryCalloc(taosArrayGetSize(block->pDataBlock), sizeof(SUdfColumn*));
for (int32_t i = 0; i < udfBlock->numOfCols; ++i) {
udfBlock->udfCols[i] = taosMemoryCalloc(1, sizeof(SUdfColumn));
SColumnInfoData *col= (SColumnInfoData*)taosArrayGet(block->pDataBlock, i);
......@@ -799,7 +799,6 @@ int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlo
}
int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block) {
block->info.numOfCols = 1;
block->info.rows = udfCol->colData.numOfRows;
block->info.hasVarCol = IS_VAR_DATA_TYPE(udfCol->colMeta.type);
......@@ -830,25 +829,19 @@ int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block) {
int32_t convertScalarParamToDataBlock(SScalarParam *input, int32_t numOfCols, SSDataBlock *output) {
output->info.rows = input->numOfRows;
output->info.numOfCols = numOfCols;
bool hasVarCol = false;
for (int32_t i = 0; i < numOfCols; ++i) {
if (IS_VAR_DATA_TYPE((input+i)->columnData->info.type)) {
hasVarCol = true;
break;
}
}
output->info.hasVarCol = hasVarCol;
output->pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
for (int32_t i = 0; i < numOfCols; ++i) {
taosArrayPush(output->pDataBlock, (input + i)->columnData);
if (IS_VAR_DATA_TYPE((input+i)->columnData->info.type)) {
output->info.hasVarCol = true;
}
}
return 0;
}
int32_t convertDataBlockToScalarParm(SSDataBlock *input, SScalarParam *output) {
if (input->info.numOfCols != 1) {
if (taosArrayGetSize(input->pDataBlock) != 1) {
fnError("scalar function only support one column");
return -1;
}
......@@ -1089,24 +1082,14 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
int32_t start = pInput->startRowIndex;
int32_t numOfRows = pInput->numOfRows;
SSDataBlock tempBlock = {0};
tempBlock.info.numOfCols = numOfCols;
tempBlock.info.rows = pInput->totalRows;
tempBlock.info.uid = pInput->uid;
bool hasVarCol = false;
tempBlock.pDataBlock = taosArrayInit(numOfCols, sizeof(SColumnInfoData));
SSDataBlock* pTempBlock = createDataBlock();
pTempBlock->info.rows = pInput->totalRows;
pTempBlock->info.uid = pInput->uid;
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData *col = pInput->pData[i];
if (IS_VAR_DATA_TYPE(col->info.type)) {
hasVarCol = true;
}
taosArrayPush(tempBlock.pDataBlock, col);
blockDataAppendColInfo(pTempBlock, pInput->pData[i]);
}
tempBlock.info.hasVarCol = hasVarCol;
SSDataBlock *inputBlock = blockDataExtractBlock(&tempBlock, start, numOfRows);
SSDataBlock *inputBlock = blockDataExtractBlock(pTempBlock, start, numOfRows);
SUdfInterBuf state = {.buf = udfRes->interResBuf,
.bufLen = session->bufSize,
......@@ -1131,7 +1114,9 @@ int32_t udfAggProcess(struct SqlFunctionCtx *pCtx) {
}
blockDataDestroy(inputBlock);
taosArrayDestroy(tempBlock.pDataBlock);
taosArrayDestroy(pTempBlock->pDataBlock);
taosMemoryFree(pTempBlock);
releaseUdfFuncHandle(pCtx->udfName);
freeUdfInterBuf(&newState);
......
source/libs/function/test/runUdf.c
浏览文件 @ 5cdef06f
......@@ -45,27 +45,27 @@ int scalarFuncTest() {
for (int k = 0; k < 1; ++k) {
SSDataBlock block = {0};
SSDataBlock *pBlock = &block;
pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
pBlock->info.numOfCols = 1;
for (int32_t i = 0; i < 1; ++i) {
SColumnInfoData colInfo = createColumnInfoData(TSDB_DATA_TYPE_INT, sizeof(int32_t), 1);
blockDataAppendColInfo(pBlock, &colInfo);
}
blockDataEnsureCapacity(pBlock, 1024);
pBlock->info.rows = 1024;
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
SColumnInfoData colInfo = {0};
colInfo.info.type = TSDB_DATA_TYPE_INT;
colInfo.info.bytes = sizeof(int32_t);
colInfo.info.colId = 1;
colInfoDataEnsureCapacity(&colInfo, 0, pBlock->info.rows);
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(&colInfo, j, &j);
}
taosArrayPush(pBlock->pDataBlock, &colInfo);
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, 0);
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(pCol, j, &j);
}
SScalarParam input = {0};
input.numOfRows = pBlock->info.rows;
input.columnData = taosArrayGet(pBlock->pDataBlock, 0);
SScalarParam output = {0};
doCallUdfScalarFunc(handle, &input, 1, &output);
taosArrayDestroy(pBlock->pDataBlock);
SColumnInfoData *col = output.columnData;
for (int32_t i = 0; i < output.numOfRows; ++i) {
if (i % 100 == 0)
......@@ -89,21 +89,18 @@ int aggregateFuncTest() {
return -1;
}
SSDataBlock block = {0};
SSDataBlock *pBlock = &block;
pBlock->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
pBlock->info.numOfCols = 1;
SSDataBlock *pBlock = createDataBlock();
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData colInfo = createColumnInfoData(TSDB_DATA_TYPE_INT, sizeof(int32_t), 1);
blockDataAppendColInfo(pBlock, &colInfo);
}
blockDataEnsureCapacity(pBlock, 1024);
pBlock->info.rows = 1024;
for (int32_t i = 0; i < pBlock->info.numOfCols; ++i) {
SColumnInfoData colInfo = {0};
colInfo.info.type = TSDB_DATA_TYPE_INT;
colInfo.info.bytes = sizeof(int32_t);
colInfo.info.colId = 1;
colInfoDataEnsureCapacity(&colInfo, 0, pBlock->info.rows);
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(&colInfo, j, &j);
}
taosArrayPush(pBlock->pDataBlock, &colInfo);
SColumnInfoData* pColInfo = bdGetColumnInfoData(pBlock, 0);
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(pColInfo, j, &j);
}
SUdfInterBuf buf = {0};
......@@ -121,6 +118,7 @@ int aggregateFuncTest() {
freeUdfInterBuf(&resultBuf);
doTeardownUdf(handle);
blockDataDestroy(pBlock);
return 0;
}
......
source/libs/index/test/index_executor_tests.cpp
浏览文件 @ 5cdef06f
......@@ -86,7 +86,7 @@ void sifAppendReservedSlot(SArray *pBlockList, int16_t *dataBlockId, int16_t *sl
SColumnInfoData idata = {0};
idata.info = *colInfo;
colInfoDataEnsureCapacity(&idata, 0, rows);
colInfoDataEnsureCapacity(&idata, rows);
taosArrayPush(res->pDataBlock, &idata);
......
source/libs/scalar/inc/sclInt.h
浏览文件 @ 5cdef06f
......@@ -57,7 +57,7 @@ typedef struct SScalarCtx {
#define SCL_ERR_JRET(c) do { code = c; if (code != TSDB_CODE_SUCCESS) { terrno = code; goto _return; } } while (0)
int32_t doConvertDataType(SValueNode* pValueNode, SScalarParam* out);
SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows);
SColumnInfoData* sclCreateColumnInfoData(SDataType* pType, int32_t numOfRows);
int32_t sclConvertToTsValueNode(int8_t precision, SValueNode* valueNode);
#define GET_PARAM_TYPE(_c) ((_c)->columnData ? (_c)->columnData->info.type : (_c)->hashValueType)
......
source/libs/scalar/src/filter.c
浏览文件 @ 5cdef06f
......@@ -3821,7 +3821,7 @@ bool filterExecute(SFilterInfo *info, SSDataBlock *pSrc, int8_t** p, SColumnData
SScalarParam output = {0};
SDataType type = {.type = TSDB_DATA_TYPE_BOOL, .bytes = sizeof(bool)};
output.columnData = createColumnInfoData(&type, pSrc->info.rows);
output.columnData = sclCreateColumnInfoData(&type, pSrc->info.rows);
SArray *pList = taosArrayInit(1, POINTER_BYTES);
taosArrayPush(pList, &pSrc);
......
source/libs/scalar/src/scalar.c
浏览文件 @ 5cdef06f
......@@ -36,7 +36,7 @@ int32_t sclConvertToTsValueNode(int8_t precision, SValueNode* valueNode) {
}
SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows) {
SColumnInfoData* sclCreateColumnInfoData(SDataType* pType, int32_t numOfRows) {
SColumnInfoData* pColumnData = taosMemoryCalloc(1, sizeof(SColumnInfoData));
if (pColumnData == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
......@@ -48,7 +48,7 @@ SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows) {
pColumnData->info.scale = pType->scale;
pColumnData->info.precision = pType->precision;
int32_t code = colInfoDataEnsureCapacity(pColumnData, 0, numOfRows);
int32_t code = colInfoDataEnsureCapacity(pColumnData, numOfRows);
if (code != TSDB_CODE_SUCCESS) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
taosMemoryFree(pColumnData);
......@@ -60,10 +60,10 @@ SColumnInfoData* createColumnInfoData(SDataType* pType, int32_t numOfRows) {
int32_t doConvertDataType(SValueNode* pValueNode, SScalarParam* out) {
SScalarParam in = {.numOfRows = 1};
in.columnData = createColumnInfoData(&pValueNode->node.resType, 1);
in.columnData = sclCreateColumnInfoData(&pValueNode->node.resType, 1);
colDataAppend(in.columnData, 0, nodesGetValueFromNode(pValueNode), false);
colInfoDataEnsureCapacity(out->columnData, 0, 1);
colInfoDataEnsureCapacity(out->columnData, 1);
int32_t code = vectorConvertImpl(&in, out);
sclFreeParam(&in);
......@@ -193,7 +193,7 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
SValueNode *valueNode = (SValueNode *)node;
param->numOfRows = 1;
param->columnData = createColumnInfoData(&valueNode->node.resType, 1);
param->columnData = sclCreateColumnInfoData(&valueNode->node.resType, 1);
if (TSDB_DATA_TYPE_NULL == valueNode->node.resType.type) {
colDataAppendNULL(param->columnData, 0);
} else {
......@@ -245,7 +245,7 @@ int32_t sclInitParam(SNode* node, SScalarParam *param, SScalarCtx *ctx, int32_t
}
SSDataBlock *block = *(SSDataBlock **)taosArrayGet(ctx->pBlockList, index);
if (NULL == block || ref->slotId >= block->info.numOfCols) {
if (NULL == block || ref->slotId >= taosArrayGetSize(block->pDataBlock)) {
sclError("column slotId is too big, slodId:%d, dataBlockSize:%d", ref->slotId, (int32_t)taosArrayGetSize(block->pDataBlock));
SCL_ERR_RET(TSDB_CODE_QRY_INVALID_INPUT);
}
......@@ -431,7 +431,7 @@ int32_t sclExecFunction(SFunctionNode *node, SScalarCtx *ctx, SScalarParam *outp
SCL_ERR_JRET(code);
}
output->columnData = createColumnInfoData(&node->node.resType, rowNum);
output->columnData = sclCreateColumnInfoData(&node->node.resType, rowNum);
if (output->columnData == NULL) {
sclError("calloc %d failed", (int32_t)(rowNum * output->columnData->info.bytes));
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
......@@ -484,7 +484,7 @@ int32_t sclExecLogic(SLogicConditionNode *node, SScalarCtx *ctx, SScalarParam *o
output->numOfRows = rowNum;
SDataType t = {.type = type, .bytes = tDataTypes[type].bytes};
output->columnData = createColumnInfoData(&t, rowNum);
output->columnData = sclCreateColumnInfoData(&t, rowNum);
if (output->columnData == NULL) {
sclError("calloc %d failed", (int32_t)(rowNum * sizeof(bool)));
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
......@@ -539,7 +539,7 @@ int32_t sclExecOperator(SOperatorNode *node, SScalarCtx *ctx, SScalarParam *outp
int32_t code = 0;
SCL_ERR_RET(sclInitOperatorParams(&params, node, ctx, &rowNum));
output->columnData = createColumnInfoData(&node->node.resType, rowNum);
output->columnData = sclCreateColumnInfoData(&node->node.resType, rowNum);
if (output->columnData == NULL) {
sclError("calloc failed, size:%d", (int32_t)rowNum * node->node.resType.bytes);
SCL_ERR_JRET(TSDB_CODE_QRY_OUT_OF_MEMORY);
......@@ -901,7 +901,7 @@ EDealRes sclWalkTarget(SNode* pNode, SScalarCtx *ctx) {
return DEAL_RES_ERROR;
}
colDataAssign(col, res->columnData, res->numOfRows);
colDataAssign(col, res->columnData, res->numOfRows, NULL);
block->info.rows = res->numOfRows;
sclFreeParam(res);
......@@ -982,8 +982,9 @@ int32_t scalarCalculate(SNode *pNode, SArray *pBlockList, SScalarParam *pDst) {
sclError("no valid res in hash, node:%p, type:%d", pNode, nodeType(pNode));
SCL_ERR_JRET(TSDB_CODE_QRY_APP_ERROR);
}
colDataAssign(pDst->columnData, res->columnData, res->numOfRows);
colInfoDataEnsureCapacity(pDst->columnData, res->numOfRows);
colDataAssign(pDst->columnData, res->columnData, res->numOfRows, NULL);
pDst->numOfRows = res->numOfRows;
taosHashRemove(ctx.pRes, (void *)&pNode, POINTER_BYTES);
}
......
source/libs/scalar/src/sclfunc.c
浏览文件 @ 5cdef06f
......@@ -120,7 +120,7 @@ int32_t absFunction(SScalarParam *pInput, int32_t inputNum, SScalarParam *pOutpu
}
default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows);
colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL);
}
}
......@@ -273,7 +273,7 @@ static int32_t doScalarFunction(SScalarParam *pInput, int32_t inputNum, SScalarP
}
default: {
colDataAssign(pOutputData, pInputData, pInput->numOfRows);
colDataAssign(pOutputData, pInputData, pInput->numOfRows, NULL);
}
}
......
source/libs/scalar/src/sclvector.c
浏览文件 @ 5cdef06f
......@@ -852,7 +852,7 @@ int32_t vectorConvertScalarParam(SScalarParam *input, SScalarParam *output, int3
SDataType t = {.type = type, .bytes = tDataTypes[type].bytes};
output->numOfRows = input->numOfRows;
output->columnData = createColumnInfoData(&t, input->numOfRows);
output->columnData = sclCreateColumnInfoData(&t, input->numOfRows);
if (output->columnData == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
......@@ -923,7 +923,7 @@ static int32_t doConvertHelper(SScalarParam* pDest, int32_t* convert, const SSca
pDest->numOfRows = pParam->numOfRows;
SDataType t = {.type = type, .bytes = tDataTypes[type].bytes};
pDest->columnData = createColumnInfoData(&t, pParam->numOfRows);
pDest->columnData = sclCreateColumnInfoData(&t, pParam->numOfRows);
if (pDest->columnData == NULL) {
sclError("malloc %d failed", (int32_t)(pParam->numOfRows * sizeof(double)));
return TSDB_CODE_OUT_OF_MEMORY;
......
source/libs/scalar/test/scalar/scalarTests.cpp
浏览文件 @ 5cdef06f
......@@ -84,30 +84,29 @@ void scltInitLogFile() {
void scltAppendReservedSlot(SArray *pBlockList, int16_t *dataBlockId, int16_t *slotId, bool newBlock, int32_t rows, SColumnInfo *colInfo) {
if (newBlock) {
SSDataBlock *res = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
res->info.numOfCols = 1;
res->info.rows = rows;
res->pDataBlock = taosArrayInit(1, sizeof(SColumnInfoData));
SSDataBlock *res = createDataBlock();
SColumnInfoData idata = {0};
idata.info = *colInfo;
idata.info = *colInfo;
colInfoDataEnsureCapacity(&idata, rows);
taosArrayPush(res->pDataBlock, &idata);
taosArrayPush(pBlockList, &res);
blockDataAppendColInfo(res, &idata);
blockDataEnsureCapacity(res, rows);
res->info.capacity = rows;
res->info.rows = rows;
SColumnInfoData* p = static_cast<SColumnInfoData *>(taosArrayGet(res->pDataBlock, 0));
ASSERT(p->pData != NULL && p->nullbitmap != NULL);
taosArrayPush(pBlockList, &res);
*dataBlockId = taosArrayGetSize(pBlockList) - 1;
res->info.blockId = *dataBlockId;
*slotId = 0;
} else {
SSDataBlock *res = *(SSDataBlock **)taosArrayGetLast(pBlockList);
res->info.numOfCols++;
SColumnInfoData idata = {0};
idata.info = *colInfo;
colInfoDataEnsureCapacity(&idata, 0, rows);
taosArrayPush(res->pDataBlock, &idata);
colInfoDataEnsureCapacity(&idata, rows);
blockDataAppendColInfo(res, &idata);
*dataBlockId = taosArrayGetSize(pBlockList) - 1;
*slotId = taosArrayGetSize(res->pDataBlock) - 1;
......@@ -144,29 +143,19 @@ void scltMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
}
if (NULL == *block) {
SSDataBlock *res = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
res->info.numOfCols = 3;
res->info.rows = rowNum;
res->pDataBlock = taosArrayInit(3, sizeof(SColumnInfoData));
SSDataBlock *res = createDataBlock();
for (int32_t i = 0; i < 2; ++i) {
SColumnInfoData idata = {{0}};
idata.info.type = TSDB_DATA_TYPE_NULL;
idata.info.bytes = 10;
idata.info.colId = i + 1;
int32_t size = idata.info.bytes * rowNum;
idata.pData = (char *)taosMemoryCalloc(1, size);
taosArrayPush(res->pDataBlock, &idata);
SColumnInfoData idata = createColumnInfoData(TSDB_DATA_TYPE_INT, 10, i + 1);
colInfoDataEnsureCapacity(&idata, rowNum);
blockDataAppendColInfo(res, &idata);
}
SColumnInfoData idata = {{0}};
idata.info.type = dataType;
idata.info.bytes = dataBytes;
idata.info.colId = 3;
int32_t size = idata.info.bytes * rowNum;
idata.pData = (char *)taosMemoryCalloc(1, size);
colInfoDataEnsureCapacity(&idata, 0, rowNum);
taosArrayPush(res->pDataBlock, &idata);
SColumnInfoData idata = createColumnInfoData(dataType, dataBytes, 3);
colInfoDataEnsureCapacity(&idata, rowNum);
blockDataAppendColInfo(res, &idata);
res->info.capacity = rowNum;
res->info.rows = rowNum;
SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock);
for (int32_t i = 0; i < rowNum; ++i) {
colDataAppend(pColumn, i, (const char *)value, false);
......@@ -185,17 +174,13 @@ void scltMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
SSDataBlock *res = *block;
int32_t idx = taosArrayGetSize(res->pDataBlock);
SColumnInfoData idata = {{0}};
idata.info.type = dataType;
idata.info.bytes = dataBytes;
idata.info.colId = 1 + idx;
int32_t size = idata.info.bytes * rowNum;
idata.pData = (char *)taosMemoryCalloc(1, size);
taosArrayPush(res->pDataBlock, &idata);
res->info.numOfCols++;
SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock);
SColumnInfoData idata = createColumnInfoData(dataType, dataBytes, 1 + idx);
colInfoDataEnsureCapacity(&idata, rowNum);
colInfoDataEnsureCapacity(pColumn, 0, rowNum);
res->info.capacity = rowNum;
blockDataAppendColInfo(res, &idata);
SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock);
for (int32_t i = 0; i < rowNum; ++i) {
colDataAppend(pColumn, i, (const char *)value, false);
......@@ -279,9 +264,6 @@ void scltMakeTargetNode(SNode **pNode, int16_t dataBlockId, int16_t slotId, SNod
*pNode = (SNode *)onode;
}
}
TEST(constantTest, bigint_add_bigint) {
......@@ -1529,6 +1511,7 @@ TEST(columnTest, bigint_column_multi_binary_column) {
SArray *blockList = taosArrayInit(1, POINTER_BYTES);
taosArrayPush(blockList, &src);
SColumnInfo colInfo = createColumnInfo(1, TSDB_DATA_TYPE_DOUBLE, sizeof(double));
int16_t dataBlockId = 0, slotId = 0;
scltAppendReservedSlot(blockList, &dataBlockId, &slotId, false, rowNum, &colInfo);
......@@ -2033,7 +2016,7 @@ void scltMakeDataBlock(SScalarParam **pInput, int32_t type, void *pVal, int32_t
input->numOfRows = num;
input->columnData->info = createColumnInfo(0, type, bytes);
colInfoDataEnsureCapacity(input->columnData, 0, num);
colInfoDataEnsureCapacity(input->columnData, num);
if (setVal) {
for (int32_t i = 0; i < num; ++i) {
......
source/libs/stream/src/streamDispatch.c
浏览文件 @ 5cdef06f
......@@ -96,16 +96,17 @@ int32_t streamBroadcastToChildren(SStreamTask* pTask, const SSDataBlock* pBlock)
pRetrieve = taosMemoryCalloc(1, dataStrLen);
if (pRetrieve == NULL) return -1;
int32_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
pRetrieve->useconds = 0;
pRetrieve->precision = TSDB_DEFAULT_PRECISION;
pRetrieve->compressed = 0;
pRetrieve->completed = 1;
pRetrieve->streamBlockType = pBlock->info.type;
pRetrieve->numOfRows = htonl(pBlock->info.rows);
pRetrieve->numOfCols = htonl(pBlock->info.numOfCols);
pRetrieve->numOfCols = htonl(numOfCols);
int32_t actualLen = 0;
blockCompressEncode(pBlock, pRetrieve->data, &actualLen, pBlock->info.numOfCols, false);
blockCompressEncode(pBlock, pRetrieve->data, &actualLen, numOfCols, false);
SStreamRetrieveReq req = {
.streamId = pTask->streamId,
......@@ -170,10 +171,12 @@ static int32_t streamAddBlockToDispatchMsg(const SSDataBlock* pBlock, SStreamDis
pRetrieve->completed = 1;
pRetrieve->streamBlockType = pBlock->info.type;
pRetrieve->numOfRows = htonl(pBlock->info.rows);
pRetrieve->numOfCols = htonl(pBlock->info.numOfCols);
int32_t numOfCols = (int32_t) taosArrayGetSize(pBlock->pDataBlock);
pRetrieve->numOfCols = htonl(numOfCols);
int32_t actualLen = 0;
blockCompressEncode(pBlock, pRetrieve->data, &actualLen, pBlock->info.numOfCols, false);
blockCompressEncode(pBlock, pRetrieve->data, &actualLen, numOfCols, false);
actualLen += sizeof(SRetrieveTableRsp);
ASSERT(actualLen <= dataStrLen);
taosArrayPush(pReq->dataLen, &actualLen);
......
tests/script/general/parser/udf.sim
浏览文件 @ 5cdef06f
......@@ -44,6 +44,7 @@ endi
sql create function n02 as '/tmp/normal' outputtype bool;
sql show functions;
if $rows != 1 then
print expect 1, actual $rows
return -1
endi
......
tests/script/tsim/query/udf.sim
浏览文件 @ 5cdef06f
......@@ -44,6 +44,7 @@ endi
sql select udf2(f) from t;
if $rows != 1 then
print expect 1, actual $rows
return -1
endi
if $data00 != 2.236067977 then
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或