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

Merge branch '3.0' of https://github.com/taosdata/TDengine into main

This commit is contained in:
jiajingbin 2022-11-24 17:08:31 +08:00
parent f1a73c47aa 130d6e3b6c
commit fbcccf840a
47 changed files with 1440 additions and 865 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
docs/en/05-get-started/index.md
View File

@ -8,6 +8,7 @@ import DiscordSVG from './discord.svg'
import TwitterSVG from './twitter.svg'
import YouTubeSVG from './youtube.svg'
import LinkedInSVG from './linkedin.svg'
import StackOverflowSVG from './stackoverflow.svg'
You can install and run TDengine on Linux/Windows/macOS machines as well as Docker containers. You can also deploy TDengine as a managed service with TDengine Cloud.
@ -35,10 +36,19 @@ The TDengine Knowledge Map covers the various knowledge points of TDengine, reve
<table width="100%">
<tr align="center" style={{border:0}}>
<td width="20%" style={{border:0}}><a href="https://github.com/taosdata/TDengine" target="_blank"><GitHubSVG /><p>Star GitHub</p></a></td>
<td width="20%" style={{border:0}}><a href="https://discord.com/invite/VZdSuUg4pS" target="_blank"><DiscordSVG /><p>Join Discord</p></a></td>
<td width="20%" style={{border:0}}><a href="https://twitter.com/TDengineDB" target="_blank"><TwitterSVG /><p>Follow Twitter</p></a></td>
<td width="20%" style={{border:0}}><a href="https://www.youtube.com/@tdengine" target="_blank"><YouTubeSVG /><p>Subscribe YouTube</p></a></td>
<td width="20%" style={{border:0}}><a href="https://www.linkedin.com/company/tdengine" target="_blank"><LinkedInSVG /><p>Follow LinkedIn</p></a></td>
<td width="16%" style={{border:0}}><a href="https://github.com/taosdata/TDengine" target="_blank"><GitHubSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://discord.com/invite/VZdSuUg4pS" target="_blank"><DiscordSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://twitter.com/TDengineDB" target="_blank"><TwitterSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://www.youtube.com/@tdengine" target="_blank"><YouTubeSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://www.linkedin.com/company/tdengine" target="_blank"><LinkedInSVG /></a></td>
<td width="16%" style={{border:0}}><a href="https://stackoverflow.com/questions/tagged/tdengine" target="_blank"><StackOverflowSVG /></a></td>
</tr>
<tr align="center" style={{border:0,backgroundColor:'transparent'}}>
<td width="16%" style={{border:0,padding:0}}><a href="https://github.com/taosdata/TDengine" target="_blank">Star GitHub</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://discord.com/invite/VZdSuUg4pS" target="_blank">Join Discord</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://twitter.com/TDengineDB" target="_blank">Follow Twitter</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://www.youtube.com/@tdengine" target="_blank">Subscribe YouTube</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://www.linkedin.com/company/tdengine" target="_blank">Follow LinkedIn</a></td>
<td width="16%" style={{border:0,padding:0}}><a href="https://stackoverflow.com/questions/tagged/tdengine" target="_blank">Ask StackOverflow</a></td>
</tr>
</table>

7
docs/en/05-get-started/stackoverflow.svg Normal file
View File

@ -0,0 +1,7 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="-8 0 48 48" width="50" height="50">
<path d="M26 41v-9h4v13H0V32h4v9h22z" fill="#BCBBBB" />
<path
d="M23 34l.8-3-16.1-3.3L7 31l16 3zM9.2 23.2l15 7 1.4-3-15-7-1.4 3zm4.2-7.4L26 26.4l2.1-2.5-12.7-10.6-2.1 2.5zM21.5 8l-2.7 2 9.9 13.3 2.7-2L21.5 8zM7 38h16v-3H7v3z"
fill="#F48024"
/>
</svg>
Side by Side

After

Width:  |  Height:  |  Size: 350 B

103
include/common/tdataformat.h
View File

@ -27,17 +27,17 @@
extern "C" {
#endif
typedef struct SBuffer SBuffer;
typedef struct SSchema SSchema;
typedef struct STColumn STColumn;
typedef struct STSchema STSchema;
typedef struct SValue SValue;
typedef struct SColVal SColVal;
typedef struct STSRow2 STSRow2;
typedef struct STSRowBuilder STSRowBuilder;
typedef struct STagVal STagVal;
typedef struct STag STag;
typedef struct SColData SColData;
typedef struct SBuffer SBuffer;
typedef struct SSchema SSchema;
typedef struct STColumn STColumn;
typedef struct STSchema STSchema;
typedef struct SValue SValue;
typedef struct SColVal SColVal;
typedef struct SRow SRow;
typedef struct SRowIter SRowIter;
typedef struct STagVal STagVal;
typedef struct STag STag;
typedef struct SColData SColData;
#define HAS_NONE ((uint8_t)0x1)
#define HAS_NULL ((uint8_t)0x2)
@ -68,13 +68,10 @@ struct SBuffer {
void tBufferDestroy(SBuffer *pBuffer);
int32_t tBufferInit(SBuffer *pBuffer, int64_t size);
int32_t tBufferPut(SBuffer *pBuffer, const void *pData, int64_t nData);
int32_t tBufferReserve(SBuffer *pBuffer, int64_t nData, void **ppData);
// STSchema ================================
int32_t tTSchemaCreate(int32_t sver, SSchema *pSchema, int32_t nCols, STSchema **ppTSchema);
void tTSchemaDestroy(STSchema *pTSchema);
// SValue ================================
static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type);
void tDestroyTSchema(STSchema *pTSchema);
// SColVal ================================
#define CV_FLAG_VALUE ((int8_t)0x0)
@ -89,26 +86,14 @@ static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type);
#define COL_VAL_IS_NULL(CV) ((CV)->flag == CV_FLAG_NULL)
#define COL_VAL_IS_VALUE(CV) ((CV)->flag == CV_FLAG_VALUE)
// STSRow2 ================================
#define TSROW_LEN(PROW, V) tGetI32v((uint8_t *)(PROW)->data, (V) ? &(V) : NULL)
#define TSROW_SVER(PROW, V) tGetI32v((PROW)->data + TSROW_LEN(PROW, NULL), (V) ? &(V) : NULL)
// SRow ================================
int32_t tRowBuild(SArray *aColVal, STSchema *pTSchema, SBuffer *pBuffer);
void tRowGet(SRow *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
int32_t tTSRowNew(STSRowBuilder *pBuilder, SArray *pArray, STSchema *pTSchema, STSRow2 **ppRow);
int32_t tTSRowClone(const STSRow2 *pRow, STSRow2 **ppRow);
void tTSRowFree(STSRow2 *pRow);
void tTSRowGet(STSRow2 *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
int32_t tTSRowToArray(STSRow2 *pRow, STSchema *pTSchema, SArray **ppArray);
int32_t tPutTSRow(uint8_t *p, STSRow2 *pRow);
int32_t tGetTSRow(uint8_t *p, STSRow2 **ppRow);
// STSRowBuilder ================================
#define tsRowBuilderInit() ((STSRowBuilder){0})
#define tsRowBuilderClear(B) \
do { \
if ((B)->pBuf) { \
taosMemoryFree((B)->pBuf); \
} \
} while (0)
// SRowIter ================================
int32_t tRowIterOpen(SRow *pRow, STSchema *pTSchema, SRowIter **ppIter);
void tRowIterClose(SRowIter **ppIter);
SColVal *tRowIterNext(SRowIter *pIter);
// STag ================================
int32_t tTagNew(SArray *pArray, int32_t version, int8_t isJson, STag **ppTag);
@ -147,29 +132,17 @@ struct STSchema {
int32_t numOfCols;
int32_t version;
int32_t flen;
int32_t vlen;
int32_t tlen;
STColumn columns[];
};
#define TSROW_HAS_NONE ((uint8_t)0x1)
#define TSROW_HAS_NULL ((uint8_t)0x2U)
#define TSROW_HAS_VAL ((uint8_t)0x4U)
#define TSROW_KV_SMALL ((uint8_t)0x10U)
#define TSROW_KV_MID ((uint8_t)0x20U)
#define TSROW_KV_BIG ((uint8_t)0x40U)
#pragma pack(push, 1)
struct STSRow2 {
TSKEY ts;
uint8_t flags;
uint8_t data[];
};
#pragma pack(pop)
struct STSRowBuilder {
// STSRow2 tsRow;
int32_t szBuf;
uint8_t *pBuf;
struct SRow {
uint8_t flag;
uint8_t rsv;
uint16_t sver;
uint32_t len;
TSKEY ts;
uint8_t data[];
};
struct SValue {
@ -258,37 +231,17 @@ typedef struct {
int32_t nCols;
schema_ver_t version;
uint16_t flen;
int32_t vlen;
int32_t tlen;
STColumn *columns;
} STSchemaBuilder;
// use 2 bits for bitmap(default: STSRow/sub block)
#define TD_VTYPE_BITS 2
#define TD_VTYPE_PARTS 4 // PARTITIONS: 1 byte / 2 bits
#define TD_VTYPE_OPTR 3 // OPERATOR: 4 - 1, utilize to get remainder
#define TD_BITMAP_BYTES(cnt) (((cnt) + TD_VTYPE_OPTR) >> 2)
// use 1 bit for bitmap(super block)
#define TD_VTYPE_BITS_I 1
#define TD_VTYPE_PARTS_I 8 // PARTITIONS: 1 byte / 1 bit
#define TD_VTYPE_OPTR_I 7 // OPERATOR: 8 - 1, utilize to get remainder
#define TD_BITMAP_BYTES_I(cnt) (((cnt) + TD_VTYPE_OPTR_I) >> 3)
int32_t tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
void tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder);
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, int8_t flags, col_id_t colId, col_bytes_t bytes);
STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);
static FORCE_INLINE int32_t tGetValue(uint8_t *p, SValue *pValue, int8_t type) {
if (IS_VAR_DATA_TYPE(type)) {
return tGetBinary(p, &pValue->pData, pValue ? &pValue->nData : NULL);
} else {
memcpy(&pValue->val, p, tDataTypes[type].bytes);
return tDataTypes[type].bytes;
}
}
STSchema *tBuildTSchema(SSchema *aSchema, int32_t numOfCols, int32_t version);
#endif

10
include/common/trow.h
View File

@ -55,6 +55,14 @@ typedef struct STSRow {
#define TD_ROW_TP 0x0U // default
#define TD_ROW_KV 0x01U
#define TD_VTYPE_PARTS 4 // PARTITIONS: 1 byte / 2 bits
#define TD_VTYPE_OPTR 3 // OPERATOR: 4 - 1, utilize to get remainder
#define TD_BITMAP_BYTES(cnt) (((cnt) + TD_VTYPE_OPTR) >> 2)
#define TD_VTYPE_PARTS_I 8 // PARTITIONS: 1 byte / 1 bit
#define TD_VTYPE_OPTR_I 7 // OPERATOR: 8 - 1, utilize to get remainder
#define TD_BITMAP_BYTES_I(cnt) (((cnt) + TD_VTYPE_OPTR_I) >> 3)
/**
* @brief value type
* - for data from client input and STSRow in memory, 3 types of value none/null/norm available
@ -244,7 +252,7 @@ int32_t tdGetBitmapValTypeI(const void *pBitmap, int16_t colIdx, TDRowValT *pVal
*/
static FORCE_INLINE void *tdGetBitmapAddrTp(STSRow *pRow, uint32_t flen) {
// The primary TS key is stored separatedly.
return POINTER_SHIFT(TD_ROW_DATA(pRow), flen - sizeof(TSKEY));
return POINTER_SHIFT(TD_ROW_DATA(pRow), flen);
// return POINTER_SHIFT(pRow->ts, flen);
}

2
include/libs/nodes/plannodes.h
View File

@ -171,6 +171,7 @@ typedef struct SExchangeLogicNode {
SLogicNode node;
int32_t srcStartGroupId;
int32_t srcEndGroupId;
bool seqRecvData;
} SExchangeLogicNode;
typedef struct SMergeLogicNode {
@ -416,6 +417,7 @@ typedef struct SExchangePhysiNode {
int32_t srcEndGroupId;
bool singleChannel;
SNodeList* pSrcEndPoints; // element is SDownstreamSource, scheduler fill by calling qSetSuplanExecutionNode
bool seqRecvData;
} SExchangePhysiNode;
typedef struct SMergePhysiNode {

17
source/client/src/clientRawBlockWrite.c
View File

@ -1275,6 +1275,7 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname)
nVar++;
}
}
fLen -= sizeof(TSKEY);
int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
(int32_t)TD_BITMAP_BYTES(numOfCols - 1);
@ -1333,7 +1334,9 @@ int taos_write_raw_block(TAOS* taos, int rows, char* pData, const char* tbname)
}
}
offset += TYPE_BYTES[pColumn->type];
if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
offset += TYPE_BYTES[pColumn->type];
}
}
tdSRowEnd(&rb);
int32_t rowLen = TD_ROW_LEN(rowData);
@ -1503,6 +1506,7 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
nVar++;
}
}
fLen -= sizeof(TSKEY);
int32_t rows = rspObj.resInfo.numOfRows;
int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
@ -1585,8 +1589,9 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, offset, k);
}
}
offset += TYPE_BYTES[pColumn->type];
if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
offset += TYPE_BYTES[pColumn->type];
}
}
tdSRowEnd(&rb);
int32_t rowLen = TD_ROW_LEN(rowData);
@ -1803,6 +1808,7 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
nVar++;
}
}
fLen -= sizeof(TSKEY);
int32_t rows = rspObj.resInfo.numOfRows;
int32_t extendedRowSize = rowSize + TD_ROW_HEAD_LEN - sizeof(TSKEY) + nVar * sizeof(VarDataOffsetT) +
@ -1888,8 +1894,9 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
tdAppendColValToRow(&rb, pColumn->colId, pColumn->type, TD_VTYPE_NORM, colData, true, offset, k);
}
}
offset += TYPE_BYTES[pColumn->type];
if (pColumn->colId != PRIMARYKEY_TIMESTAMP_COL_ID) {
offset += TYPE_BYTES[pColumn->type];
}
}
tdSRowEnd(&rb);
int32_t rowLen = TD_ROW_LEN(rowData);

14
source/client/test/clientTests.cpp
View File

@ -692,7 +692,6 @@ TEST(testCase, insert_test) {
taos_free_result(pRes);
taos_close(pConn);
}
#endif
TEST(testCase, projection_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
@ -752,9 +751,6 @@ TEST(testCase, projection_query_tables) {
taos_close(pConn);
}
#if 0
TEST(testCase, tsbs_perf_test) {
TdThread qid[20] = {0};
@ -764,15 +760,16 @@ TEST(testCase, tsbs_perf_test) {
getchar();
}
#endif
TEST(testCase, projection_query_stables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);
TAOS_RES* pRes = taos_query(pConn, "use abc1");
TAOS_RES* pRes = taos_query(pConn, "use test");
taos_free_result(pRes);
pRes = taos_query(pConn, "select ts from st1");
pRes = taos_query(pConn, "select * from meters limit 50000000");
if (taos_errno(pRes) != 0) {
printf("failed to select from table, reason:%s\n", taos_errstr(pRes));
taos_free_result(pRes);
@ -785,14 +782,15 @@ TEST(testCase, projection_query_stables) {
char str[512] = {0};
while ((pRow = taos_fetch_row(pRes)) != NULL) {
int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
printf("%s\n", str);
// int32_t code = taos_print_row(str, pRow, pFields, numOfFields);
// printf("%s\n", str);
}
taos_free_result(pRes);
taos_close(pConn);
}
#if 0
TEST(testCase, agg_query_tables) {
TAOS* pConn = taos_connect("localhost", "root", "taosdata", NULL, 0);
ASSERT_NE(pConn, nullptr);

1
source/common/src/tdatablock.c
View File

@ -2055,6 +2055,7 @@ int32_t buildSubmitReqFromDataBlock(SSubmitReq** pReq, const SSDataBlock* pDataB
isStartKey = true;
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NORM, var, true,
offset, k);
continue; // offset should keep 0 for next column
} else if (colDataIsNull_s(pColInfoData, j)) {
tdAppendColValToRow(&rb, PRIMARYKEY_TIMESTAMP_COL_ID + k, TSDB_DATA_TYPE_TIMESTAMP, TD_VTYPE_NULL, NULL,

1200
source/common/src/tdataformat.c
View File

File diff suppressed because it is too large Load Diff

13
source/common/src/trow.c
View File

@ -192,7 +192,7 @@ bool tdSTpRowGetVal(STSRow *pRow, col_id_t colId, col_type_t colType, int32_t fl
return true;
}
void *pBitmap = tdGetBitmapAddrTp(pRow, flen);
tdGetTpRowValOfCol(pVal, pRow, pBitmap, colType, offset - sizeof(TSKEY), colIdx);
tdGetTpRowValOfCol(pVal, pRow, pBitmap, colType, offset, colIdx);
return true;
}
@ -217,7 +217,7 @@ bool tdSTSRowIterFetch(STSRowIter *pIter, col_id_t colId, col_type_t colType, SC
return false;
}
}
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset, pVal);
++pIter->colIdx;
} else if (TD_IS_KV_ROW(pIter->pRow)) {
return tdSTSRowIterGetKvVal(pIter, colId, &pIter->kvIdx, pVal);
@ -244,7 +244,7 @@ bool tdSTSRowIterNext(STSRowIter *pIter, SCellVal *pVal) {
}
if (TD_IS_TP_ROW(pIter->pRow)) {
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset - sizeof(TSKEY), pVal);
tdSTSRowIterGetTpVal(pIter, pCol->type, pCol->offset, pVal);
} else if (TD_IS_KV_ROW(pIter->pRow)) {
tdSTSRowIterGetKvVal(pIter, pCol->colId, &pIter->kvIdx, pVal);
} else {
@ -469,7 +469,7 @@ bool tdSTSRowGetVal(STSRowIter *pIter, col_id_t colId, col_type_t colType, SCell
#ifdef TD_SUPPORT_BITMAP
colIdx = POINTER_DISTANCE(pCol, pSchema->columns) / sizeof(STColumn);
#endif
tdGetTpRowValOfCol(pVal, pRow, pIter->pBitmap, pCol->type, pCol->offset - sizeof(TSKEY), colIdx - 1);
tdGetTpRowValOfCol(pVal, pRow, pIter->pBitmap, pCol->type, pCol->offset, colIdx - 1);
} else if (TD_IS_KV_ROW(pRow)) {
SKvRowIdx *pIdx = (SKvRowIdx *)taosbsearch(&colId, TD_ROW_COL_IDX(pRow), tdRowGetNCols(pRow), sizeof(SKvRowIdx),
compareKvRowColId, TD_EQ);
@ -757,11 +757,10 @@ int32_t tdAppendColValToKvRow(SRowBuilder *pBuilder, TDRowValT valType, const vo
int32_t tdAppendColValToTpRow(SRowBuilder *pBuilder, TDRowValT valType, const void *val, bool isCopyVarData,
int8_t colType, int16_t colIdx, int32_t offset) {
if ((offset < (int32_t)sizeof(TSKEY)) || (colIdx < 1)) {
if (colIdx < 1) {
terrno = TSDB_CODE_INVALID_PARA;
return terrno;
}
offset -= sizeof(TSKEY);
--colIdx;
#ifdef TD_SUPPORT_BITMAP
@ -853,7 +852,7 @@ int32_t tdSRowResetBuf(SRowBuilder *pBuilder, void *pBuf) {
memset(pBuilder->pBitmap, TD_VTYPE_NONE_BYTE_II, pBuilder->nBitmaps);
#endif
// the primary TS key is stored separatedly
len = TD_ROW_HEAD_LEN + pBuilder->flen - sizeof(TSKEY) + pBuilder->nBitmaps;
len = TD_ROW_HEAD_LEN + pBuilder->flen + pBuilder->nBitmaps;
TD_ROW_SET_LEN(pBuilder->pBuf, len);
TD_ROW_SET_SVER(pBuilder->pBuf, pBuilder->sver);
break;

4
source/common/src/ttypes.c
View File

@ -61,7 +61,7 @@ tDataTypeDescriptor tDataTypes[TSDB_DATA_TYPE_MAX] = {
static float floatMin = -FLT_MAX, floatMax = FLT_MAX;
static double doubleMin = -DBL_MAX, doubleMax = DBL_MAX;
FORCE_INLINE void *getDataMin(int32_t type, void* value) {
FORCE_INLINE void *getDataMin(int32_t type, void *value) {
switch (type) {
case TSDB_DATA_TYPE_FLOAT:
*(float *)value = floatMin;
@ -77,7 +77,7 @@ FORCE_INLINE void *getDataMin(int32_t type, void* value) {
return value;
}
FORCE_INLINE void *getDataMax(int32_t type, void* value) {
FORCE_INLINE void *getDataMax(int32_t type, void *value) {
switch (type) {
case TSDB_DATA_TYPE_FLOAT:
*(float *)value = floatMax;

6
source/dnode/mnode/impl/src/mndScheduler.c
View File

@ -224,7 +224,7 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
ASSERT(taosArrayGetSize(pStream->tasks) == 1);
while (1) {
SVgObj* pVgroup;
SVgObj* pVgroup = NULL;
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
if (pIter == NULL) break;
if (!mndVgroupInDb(pVgroup, pStream->targetDbUid)) {
@ -258,6 +258,7 @@ int32_t mndAddShuffleSinkTasksToStream(SMnode* pMnode, SStreamObj* pStream) {
pTask->tbSink.pSchemaWrapper = tCloneSSchemaWrapper(&pStream->outputSchema);
ASSERT(pTask->tbSink.pSchemaWrapper);
}
sdbRelease(pSdb, pVgroup);
}
return 0;
}
@ -382,6 +383,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
qDestroyQueryPlan(pPlan);
return -1;
}
sdbRelease(pSdb, pVgroup);
} else {
if (mndAssignTaskToSnode(pMnode, pInnerTask, plan, pSnode) < 0) {
sdbRelease(pSdb, pSnode);
@ -396,6 +398,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
qDestroyQueryPlan(pPlan);
return -1;
}
sdbRelease(pSdb, pVgroup);
}
}
@ -459,6 +462,7 @@ int32_t mndScheduleStream(SMnode* pMnode, SStreamObj* pStream) {
pEpInfo->nodeId = pTask->nodeId;
pEpInfo->taskId = pTask->taskId;
taosArrayPush(pInnerTask->childEpInfo, &pEpInfo);
sdbRelease(pSdb, pVgroup);
}
}

18
source/dnode/mnode/impl/src/mndStb.c
View File

@ -1187,11 +1187,11 @@ static int32_t mndCheckAlterColForTopic(SMnode *pMnode, const char *stbFullName,
goto NEXT;
}
if (pCol->colId > 0 && pCol->colId == colId) {
sdbRelease(pSdb, pTopic);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TOPIC;
mError("topic:%s, check colId:%d conflicted", pTopic->name, pCol->colId);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pTopic);
return -1;
}
mInfo("topic:%s, check colId:%d passed", pTopic->name, pCol->colId);
@ -1230,11 +1230,11 @@ static int32_t mndCheckAlterColForStream(SMnode *pMnode, const char *stbFullName
goto NEXT;
}
if (pCol->colId > 0 && pCol->colId == colId) {
sdbRelease(pSdb, pStream);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
terrno = TSDB_CODE_MND_STREAM_MUST_BE_DELETED;
mError("stream:%s, check colId:%d conflicted", pStream->name, pCol->colId);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pStream);
return -1;
}
mInfo("stream:%s, check colId:%d passed", pStream->name, pCol->colId);
@ -1279,11 +1279,11 @@ static int32_t mndCheckAlterColForTSma(SMnode *pMnode, const char *stbFullName,
goto NEXT;
}
if ((pCol->colId) > 0 && (pCol->colId == colId)) {
sdbRelease(pSdb, pSma);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
terrno = TSDB_CODE_MND_FIELD_CONFLICT_WITH_TSMA;
mError("tsma:%s, check colId:%d conflicted", pSma->name, pCol->colId);
nodesDestroyNode(pAst);
nodesDestroyList(pNodeList);
sdbRelease(pSdb, pSma);
return -1;
}
mInfo("tsma:%s, check colId:%d passed", pSma->name, pCol->colId);

10
source/dnode/vnode/src/inc/tsdb.h
View File

@ -56,7 +56,7 @@ typedef struct SDataFWriter SDataFWriter;
typedef struct SDataFReader SDataFReader;
typedef struct SDelFWriter SDelFWriter;
typedef struct SDelFReader SDelFReader;
typedef struct SRowIter SRowIter;
typedef struct STSDBRowIter STSDBRowIter;
typedef struct STsdbFS STsdbFS;
typedef struct SRowMerger SRowMerger;
typedef struct STsdbReadSnap STsdbReadSnap;
@ -111,9 +111,9 @@ static FORCE_INLINE int64_t tsdbLogicToFileSize(int64_t lSize, int32_t szPage) {
void tsdbRowGetColVal(TSDBROW *pRow, STSchema *pTSchema, int32_t iCol, SColVal *pColVal);
// int32_t tPutTSDBRow(uint8_t *p, TSDBROW *pRow);
int32_t tsdbRowCmprFn(const void *p1, const void *p2);
// SRowIter
void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema);
SColVal *tRowIterNext(SRowIter *pIter);
// STSDBRowIter
void tsdbRowIterInit(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema);
SColVal *tsdbRowIterNext(STSDBRowIter *pIter);
// SRowMerger
int32_t tRowMergerInit2(SRowMerger *pMerger, STSchema *pResTSchema, TSDBROW *pRow, STSchema *pTSchema);
int32_t tRowMergerAdd(SRowMerger *pMerger, TSDBROW *pRow, STSchema *pTSchema);
@ -562,7 +562,7 @@ struct SDFileSet {
SSttFile *aSttF[TSDB_MAX_STT_TRIGGER];
};
struct SRowIter {
struct STSDBRowIter {
TSDBROW *pRow;
STSchema *pTSchema;
SColVal colVal;

19
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -341,7 +341,7 @@ int32_t tsdbUpdateTableSchema(SMeta *pMeta, int64_t suid, int64_t uid, SSkmInfo
pSkmInfo->suid = suid;
pSkmInfo->uid = uid;
tTSchemaDestroy(pSkmInfo->pTSchema);
tDestroyTSchema(pSkmInfo->pTSchema);
code = metaGetTbTSchemaEx(pMeta, suid, uid, -1, &pSkmInfo->pTSchema);
TSDB_CHECK_CODE(code, lino, _exit);
@ -365,7 +365,7 @@ static int32_t tsdbCommitterUpdateRowSchema(SCommitter *pCommitter, int64_t suid
pCommitter->skmRow.suid = suid;
pCommitter->skmRow.uid = uid;
tTSchemaDestroy(pCommitter->skmRow.pTSchema);
tDestroyTSchema(pCommitter->skmRow.pTSchema);
code = metaGetTbTSchemaEx(pCommitter->pTsdb->pVnode->pMeta, suid, uid, sver, &pCommitter->skmRow.pTSchema);
TSDB_CHECK_CODE(code, lino, _exit);
@ -498,7 +498,7 @@ static int32_t tsdbCommitFileDataStart(SCommitter *pCommitter) {
#if 0
ASSERT(pCommitter->minKey <= pCommitter->nextKey && pCommitter->maxKey >= pCommitter->nextKey);
#endif
pCommitter->nextKey = TSKEY_MAX;
// Reader
@ -623,7 +623,8 @@ int32_t tsdbWriteDataBlock(SDataFWriter *pWriter, SBlockData *pBlockData, SMapDa
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino,
tstrerror(code));
}
return code;
}
@ -666,7 +667,8 @@ int32_t tsdbWriteSttBlock(SDataFWriter *pWriter, SBlockData *pBlockData, SArray
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino,
tstrerror(code));
}
return code;
}
@ -706,7 +708,8 @@ static int32_t tsdbCommitSttBlk(SDataFWriter *pWriter, SDiskDataBuilder *pBuilde
_exit:
if (code) {
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino, tstrerror(code));
tsdbError("vgId:%d, %s failed at line %d since %s", TD_VID(pWriter->pTsdb->pVnode), __func__, lino,
tstrerror(code));
}
return code;
}
@ -919,8 +922,8 @@ static void tsdbCommitDataEnd(SCommitter *pCommitter) {
#else
tBlockDataDestroy(&pCommitter->dWriter.bDatal, 1);
#endif
tTSchemaDestroy(pCommitter->skmTable.pTSchema);
tTSchemaDestroy(pCommitter->skmRow.pTSchema);
tDestroyTSchema(pCommitter->skmTable.pTSchema);
tDestroyTSchema(pCommitter->skmRow.pTSchema);
}
static int32_t tsdbCommitData(SCommitter *pCommitter) {

10
source/dnode/vnode/src/tsdb/tsdbDiskData.c
View File

@ -595,21 +595,21 @@ int32_t tDiskDataAddRow(SDiskDataBuilder *pBuilder, TSDBROW *pRow, STSchema *pTS
if (pBuilder->bi.minKey > kRow.ts) pBuilder->bi.minKey = kRow.ts;
if (pBuilder->bi.maxKey < kRow.ts) pBuilder->bi.maxKey = kRow.ts;
SRowIter iter = {0};
tRowIterInit(&iter, pRow, pTSchema);
STSDBRowIter iter = {0};
tsdbRowIterInit(&iter, pRow, pTSchema);
SColVal *pColVal = tRowIterNext(&iter);
SColVal *pColVal = tsdbRowIterNext(&iter);
for (int32_t iBuilder = 0; iBuilder < pBuilder->nBuilder; iBuilder++) {
SDiskColBuilder *pDCBuilder = (SDiskColBuilder *)taosArrayGet(pBuilder->aBuilder, iBuilder);
while (pColVal && pColVal->cid < pDCBuilder->cid) {
pColVal = tRowIterNext(&iter);
pColVal = tsdbRowIterNext(&iter);
}
if (pColVal && pColVal->cid == pDCBuilder->cid) {
code = tDiskColAddVal(pDCBuilder, pColVal);
if (code) return code;
pColVal = tRowIterNext(&iter);
pColVal = tsdbRowIterNext(&iter);
} else {
code = tDiskColAddVal(pDCBuilder, &COL_VAL_NONE(pDCBuilder->cid, pDCBuilder->type));
if (code) return code;

4
source/dnode/vnode/src/tsdb/tsdbSnapshot.c
View File

@ -555,7 +555,7 @@ int32_t tsdbSnapReaderClose(STsdbSnapReader** ppReader) {
}
tBlockDataDestroy(&pReader->bData, 1);
tTSchemaDestroy(pReader->skmTable.pTSchema);
tDestroyTSchema(pReader->skmTable.pTSchema);
// del
if (pReader->pDelFReader) tsdbDelFReaderClose(&pReader->pDelFReader);
@ -1416,7 +1416,7 @@ int32_t tsdbSnapWriterClose(STsdbSnapWriter** ppWriter, int8_t rollback) {
taosArrayDestroy(pWriter->dReader.aBlockIdx);
tBlockDataDestroy(&pWriter->bData, 1);
tTSchemaDestroy(pWriter->skmTable.pTSchema);
tDestroyTSchema(pWriter->skmTable.pTSchema);
for (int32_t iBuf = 0; iBuf < sizeof(pWriter->aBuf) / sizeof(uint8_t*); iBuf++) {
tFree(pWriter->aBuf[iBuf]);

14
source/dnode/vnode/src/tsdb/tsdbUtil.c
View File

@ -579,8 +579,8 @@ int32_t tsdbRowCmprFn(const void *p1, const void *p2) {
return tsdbKeyCmprFn(&TSDBROW_KEY((TSDBROW *)p1), &TSDBROW_KEY((TSDBROW *)p2));
}
// SRowIter ======================================================
void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
// STSDBRowIter ======================================================
void tsdbRowIterInit(STSDBRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
pIter->pRow = pRow;
if (pRow->type == 0) {
ASSERT(pTSchema);
@ -594,7 +594,7 @@ void tRowIterInit(SRowIter *pIter, TSDBROW *pRow, STSchema *pTSchema) {
}
}
SColVal *tRowIterNext(SRowIter *pIter) {
SColVal *tsdbRowIterNext(STSDBRowIter *pIter) {
if (pIter->pRow->type == 0) {
if (pIter->i < pIter->pTSchema->numOfCols) {
tTSRowGetVal(pIter->pRow->pTSRow, pIter->pTSchema, pIter->i, &pIter->colVal);
@ -1084,11 +1084,11 @@ static int32_t tBlockDataAppendTPRow(SBlockData *pBlockData, STSRow *pRow, STSch
cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset - sizeof(TSKEY));
void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset);
cv.value.nData = varDataLen(pData);
cv.value.pData = varDataVal(pData);
} else {
memcpy(&cv.value.val, pRow->data + pTColumn->offset - sizeof(TSKEY), pTColumn->bytes);
memcpy(&cv.value.val, pRow->data + pTColumn->offset, pTColumn->bytes);
}
code = tColDataAppendValue(pColData, &cv);
@ -1106,11 +1106,11 @@ static int32_t tBlockDataAppendTPRow(SBlockData *pBlockData, STSRow *pRow, STSch
cv.flag = CV_FLAG_VALUE;
if (IS_VAR_DATA_TYPE(pTColumn->type)) {
void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset - sizeof(TSKEY));
void *pData = (char *)pRow + *(int32_t *)(pRow->data + pTColumn->offset);
cv.value.nData = varDataLen(pData);
cv.value.pData = varDataVal(pData);
} else {
memcpy(&cv.value.val, pRow->data + pTColumn->offset - sizeof(TSKEY), pTColumn->bytes);
memcpy(&cv.value.val, pRow->data + pTColumn->offset, pTColumn->bytes);
}
code = tColDataAppendValue(pColData, &cv);

4
source/libs/executor/src/dataDispatcher.c
View File

@ -138,7 +138,9 @@ static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput,
toDataCacheEntry(pDispatcher, pInput, pBuf);
taosWriteQitem(pDispatcher->pDataBlocks, pBuf);
*pContinue = (DS_BUF_LOW == updateStatus(pDispatcher) ? true : false);
int32_t status = updateStatus(pDispatcher);
*pContinue = (status == DS_BUF_LOW || status == DS_BUF_EMPTY);
return TSDB_CODE_SUCCESS;
}

129
source/libs/executor/src/exchangeoperator.c
View File

@ -51,9 +51,9 @@ typedef struct SSourceDataInfo {
const char* taskId;
} SSourceDataInfo;
static void destroyExchangeOperatorInfo(void* param);
static void freeBlock(void* pParam);
static void freeSourceDataInfo(void* param);
static void destroyExchangeOperatorInfo(void* param);
static void freeBlock(void* pParam);
static void freeSourceDataInfo(void* param);
static void* setAllSourcesCompleted(SOperatorInfo* pOperator);
static int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code);
@ -62,7 +62,9 @@ static int32_t getCompletedSources(const SArray* pArray);
static int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator);
static int32_t seqLoadRemoteData(SOperatorInfo* pOperator);
static int32_t prepareLoadRemoteData(SOperatorInfo* pOperator);
static int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock, bool holdDataInBuf);
static int32_t handleLimitOffset(SOperatorInfo* pOperator, SLimitInfo* pLimitInfo, SSDataBlock* pBlock,
bool holdDataInBuf);
static int32_t doExtractResultBlocks(SExchangeInfo* pExchangeInfo, SSourceDataInfo* pDataInfo);
static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeInfo* pExchangeInfo,
SExecTaskInfo* pTaskInfo) {
@ -75,7 +77,9 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
}
while (1) {
qDebug("prepare wait for ready, %p, %s", pExchangeInfo, GET_TASKID(pTaskInfo));
tsem_wait(&pExchangeInfo->ready);
if (isTaskKilled(pTaskInfo)) {
longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
@ -103,41 +107,33 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
if (pRsp->numOfRows == 0) {
pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
qDebug("%s vgId:%d, taskId:0x%" PRIx64 " execId:%d index:%d completed, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 ", try next %d/%" PRIzu,
", totalRows:%" PRIu64 ", try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pDataInfo->totalRows,
pExchangeInfo->loadInfo.totalRows, i + 1, totalSources);
taosMemoryFreeClear(pDataInfo->pRsp);
break;
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
int32_t index = 0;
char* pStart = pRetrieveRsp->data;
while (index++ < pRetrieveRsp->numOfBlocks) {
SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
if (code != 0) {
taosMemoryFreeClear(pDataInfo->pRsp);
goto _error;
}
taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
code = doExtractResultBlocks(pExchangeInfo, pDataInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
updateLoadRemoteInfo(pLoadInfo, pRetrieveRsp->numOfRows, pRetrieveRsp->compLen, pDataInfo->startTime, pOperator);
pDataInfo->totalRows += pRetrieveRsp->numOfRows;
if (pRsp->completed == 1) {
pDataInfo->status = EX_SOURCE_DATA_EXHAUSTED;
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
" execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
", total:%.2f Kb, try next %d/%" PRIzu,
" execId:%d index:%d completed, blocks:%d, numOfRows:%d, rowsOfSource:%" PRIu64 ", totalRows:%" PRIu64
", total:%.2f Kb, try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, i, pRsp->numOfBlocks,
pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
i + 1, totalSources);
pRsp->numOfRows, pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0, i + 1,
totalSources);
} else {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64
" execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb",
" execId:%d blocks:%d, numOfRows:%d, totalRows:%" PRIu64 ", total:%.2f Kb",
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRsp->numOfBlocks,
pRsp->numOfRows, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0);
}
@ -162,7 +158,7 @@ static void concurrentlyLoadRemoteDataImpl(SOperatorInfo* pOperator, SExchangeIn
}
}
_error:
_error:
pTaskInfo->code = code;
}
@ -303,17 +299,19 @@ SOperatorInfo* createExchangeOperatorInfo(void* pTransporter, SExchangePhysiNode
SExchangeOpStopInfo stopInfo = {QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, pInfo->self};
qAppendTaskStopInfo(pTaskInfo, &stopInfo);
pInfo->seqLoadData = false;
pInfo->seqLoadData = pExNode->seqRecvData;
pInfo->pTransporter = pTransporter;
setOperatorInfo(pOperator, "ExchangeOperator", QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, false, OP_NOT_OPENED, pInfo, pTaskInfo);
setOperatorInfo(pOperator, "ExchangeOperator", QUERY_NODE_PHYSICAL_PLAN_EXCHANGE, false, OP_NOT_OPENED, pInfo,
pTaskInfo);
pOperator->exprSupp.numOfExprs = taosArrayGetSize(pInfo->pDummyBlock->pDataBlock);
pOperator->fpSet = createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, destroyExchangeOperatorInfo, NULL);
pOperator->fpSet =
createOperatorFpSet(prepareLoadRemoteData, doLoadRemoteData, NULL, destroyExchangeOperatorInfo, NULL);
return pOperator;
_error:
_error:
if (pInfo != NULL) {
doDestroyExchangeOperatorInfo(pInfo);
}
@ -360,7 +358,7 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
SExchangeInfo* pExchangeInfo = taosAcquireRef(exchangeObjRefPool, pWrapper->exchangeId);
if (pExchangeInfo == NULL) {
qWarn("failed to acquire exchange operator, since it may have been released");
qWarn("failed to acquire exchange operator, since it may have been released, %p", pExchangeInfo);
taosMemoryFree(pMsg->pData);
return TSDB_CODE_SUCCESS;
}
@ -379,20 +377,24 @@ int32_t loadRemoteDataCallback(void* param, SDataBuf* pMsg, int32_t code) {
pRsp->numOfBlocks = htonl(pRsp->numOfBlocks);
ASSERT(pRsp != NULL);
qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d", pSourceDataInfo->taskId, index, pRsp->numOfBlocks,
pRsp->numOfRows);
qDebug("%s fetch rsp received, index:%d, blocks:%d, rows:%d, %p", pSourceDataInfo->taskId, index, pRsp->numOfBlocks,
pRsp->numOfRows, pExchangeInfo);
} else {
taosMemoryFree(pMsg->pData);
pSourceDataInfo->code = code;
qDebug("%s fetch rsp received, index:%d, error:%s", pSourceDataInfo->taskId, index, tstrerror(code));
qDebug("%s fetch rsp received, index:%d, error:%s, %p", pSourceDataInfo->taskId, index, tstrerror(code),
pExchangeInfo);
}
pSourceDataInfo->status = EX_SOURCE_DATA_READY;
code = tsem_post(&pExchangeInfo->ready);
if (code != TSDB_CODE_SUCCESS) {
code = TAOS_SYSTEM_ERROR(code);
qError("failed to invoke post when fetch rsp is ready, code:%s, %p", tstrerror(code), pExchangeInfo);
}
tsem_post(&pExchangeInfo->ready);
taosReleaseRef(exchangeObjRefPool, pWrapper->exchangeId);
return TSDB_CODE_SUCCESS;
return code;
}
int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTaskInfo, int32_t sourceIndex) {
@ -429,14 +431,14 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
taosMemoryFree(pWrapper);
return pTaskInfo->code;
}
void* msg = taosMemoryCalloc(1, msgSize);
if (NULL == msg) {
pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
taosMemoryFree(pWrapper);
return pTaskInfo->code;
}
if (tSerializeSResFetchReq(msg, msgSize, &req) < 0) {
pTaskInfo->code = TSDB_CODE_QRY_OUT_OF_MEMORY;
taosMemoryFree(pWrapper);
@ -444,9 +446,9 @@ int32_t doSendFetchDataRequest(SExchangeInfo* pExchangeInfo, SExecTaskInfo* pTas
return pTaskInfo->code;
}
qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %d/%" PRIzu,
qDebug("%s build fetch msg and send to vgId:%d, ep:%s, taskId:0x%" PRIx64 ", execId:%d, %p, %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->addr.epSet.eps[0].fqdn, pSource->taskId,
pSource->execId, sourceIndex, totalSources);
pSource->execId, pExchangeInfo, sourceIndex, totalSources);
// send the fetch remote task result reques
SMsgSendInfo* pMsgSendInfo = taosMemoryCalloc(1, sizeof(SMsgSendInfo));
@ -526,7 +528,7 @@ void* setAllSourcesCompleted(SOperatorInfo* pOperator) {
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
qDebug("%s all %" PRIzu " sources are exhausted, total rows: %" PRIu64 ", %.2f Kb, elapsed:%.2f ms",
GET_TASKID(pTaskInfo), totalSources, pLoadInfo->totalRows, pLoadInfo->totalSize / 1024.0,
pLoadInfo->totalElapsed / 1000.0);
@ -576,15 +578,37 @@ int32_t prepareConcurrentlyLoad(SOperatorInfo* pOperator) {
if (isTaskKilled(pTaskInfo)) {
longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
tsem_post(&pExchangeInfo->ready);
return TSDB_CODE_SUCCESS;
}
int32_t doExtractResultBlocks(SExchangeInfo* pExchangeInfo, SSourceDataInfo* pDataInfo) {
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
char* pStart = pRetrieveRsp->data;
int32_t index = 0;
int32_t code = 0;
while (index++ < pRetrieveRsp->numOfBlocks) {
SSDataBlock* pb = createOneDataBlock(pExchangeInfo->pDummyBlock, false);
code = extractDataBlockFromFetchRsp(pb, pStart, NULL, &pStart);
if (code != 0) {
taosMemoryFreeClear(pDataInfo->pRsp);
return code;
}
taosArrayPush(pExchangeInfo->pResultBlockList, &pb);
}
return code;
}
int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
SExchangeInfo* pExchangeInfo = pOperator->info;
SExecTaskInfo* pTaskInfo = pOperator->pTaskInfo;
int32_t code = 0;
size_t totalSources = taosArrayGetSize(pExchangeInfo->pSources);
int64_t startTs = taosGetTimestampUs();
@ -594,13 +618,15 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
return TSDB_CODE_SUCCESS;
}
SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
pDataInfo->status = EX_SOURCE_DATA_NOT_READY;
doSendFetchDataRequest(pExchangeInfo, pTaskInfo, pExchangeInfo->current);
tsem_wait(&pExchangeInfo->ready);
if (isTaskKilled(pTaskInfo)) {
longjmp(pTaskInfo->env, TSDB_CODE_TSC_QUERY_CANCELLED);
}
SSourceDataInfo* pDataInfo = taosArrayGet(pExchangeInfo->pSourceDataInfo, pExchangeInfo->current);
SDownstreamSourceNode* pSource = taosArrayGet(pExchangeInfo->pSources, pExchangeInfo->current);
if (pDataInfo->code != TSDB_CODE_SUCCESS) {
@ -614,7 +640,7 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
SLoadRemoteDataInfo* pLoadInfo = &pExchangeInfo->loadInfo;
if (pRsp->numOfRows == 0) {
qDebug("%s vgId:%d, taskID:0x%" PRIx64 " execId:%d %d of total completed, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 " try next",
", totalRows:%" PRIu64 " try next",
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pExchangeInfo->current + 1,
pDataInfo->totalRows, pLoadInfo->totalRows);
@ -624,14 +650,15 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
continue;
}
code = doExtractResultBlocks(pExchangeInfo, pDataInfo);
if (code != TSDB_CODE_SUCCESS) {
goto _error;
}
SRetrieveTableRsp* pRetrieveRsp = pDataInfo->pRsp;
char* pStart = pRetrieveRsp->data;
int32_t code = extractDataBlockFromFetchRsp(NULL, pStart, NULL, &pStart);
if (pRsp->completed == 1) {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, rowsOfSource:%" PRIu64
", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu,
", totalRows:%" PRIu64 ", totalBytes:%" PRIu64 " try next %d/%" PRIzu,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows,
pDataInfo->totalRows, pLoadInfo->totalRows, pLoadInfo->totalSize, pExchangeInfo->current + 1,
totalSources);
@ -640,7 +667,7 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
pExchangeInfo->current += 1;
} else {
qDebug("%s fetch msg rsp from vgId:%d, taskId:0x%" PRIx64 " execId:%d numOfRows:%d, totalRows:%" PRIu64
", totalBytes:%" PRIu64,
", totalBytes:%" PRIu64,
GET_TASKID(pTaskInfo), pSource->addr.nodeId, pSource->taskId, pSource->execId, pRetrieveRsp->numOfRows,
pLoadInfo->totalRows, pLoadInfo->totalSize);
}
@ -651,6 +678,10 @@ int32_t seqLoadRemoteData(SOperatorInfo* pOperator) {
taosMemoryFreeClear(pDataInfo->pRsp);
return TSDB_CODE_SUCCESS;
}
_error:
pTaskInfo->code = code;
return code;
}
int32_t prepareLoadRemoteData(SOperatorInfo* pOperator) {

2
source/libs/executor/src/scanoperator.c
View File

@ -1849,12 +1849,12 @@ FETCH_NEXT_BLOCK:
prepareRangeScan(pInfo, pInfo->pUpdateRes, &pInfo->updateResIndex);
copyDataBlock(pInfo->pDeleteDataRes, pInfo->pUpdateRes);
pInfo->pDeleteDataRes->info.type = STREAM_DELETE_DATA;
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
printDataBlock(pDelBlock, "stream scan delete data");
if (pInfo->tqReader) {
blockDataDestroy(pDelBlock);
}
if (pInfo->pDeleteDataRes->info.rows > 0) {
pInfo->scanMode = STREAM_SCAN_FROM_DATAREADER_RANGE;
return pInfo->pDeleteDataRes;
} else {
goto FETCH_NEXT_BLOCK;

20
source/libs/executor/src/timewindowoperator.c
View File

@ -3364,22 +3364,23 @@ void initDummyFunction(SqlFunctionCtx* pDummy, SqlFunctionCtx* pCtx, int32_t num
}
}
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, int64_t waterMark, uint16_t type,
int32_t tsColIndex) {
void initDownStream(SOperatorInfo* downstream, SStreamAggSupporter* pAggSup, uint16_t type, int32_t tsColIndex,
STimeWindowAggSupp* pTwSup) {
if (downstream->operatorType == QUERY_NODE_PHYSICAL_PLAN_STREAM_PARTITION) {
SStreamPartitionOperatorInfo* pScanInfo = downstream->info;
pScanInfo->tsColIndex = tsColIndex;
}
if (downstream->operatorType != QUERY_NODE_PHYSICAL_PLAN_STREAM_SCAN) {
initDownStream(downstream->pDownstream[0], pAggSup, waterMark, type, tsColIndex);
initDownStream(downstream->pDownstream[0], pAggSup, type, tsColIndex, pTwSup);
return;
}
SStreamScanInfo* pScanInfo = downstream->info;
pScanInfo->windowSup = (SWindowSupporter){.pStreamAggSup = pAggSup, .gap = pAggSup->gap, .parentType = type};
if (!pScanInfo->pUpdateInfo) {
pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, waterMark);
pScanInfo->pUpdateInfo = updateInfoInit(60000, TSDB_TIME_PRECISION_MILLI, pTwSup->waterMark);
}
pScanInfo->twAggSup = *pTwSup;
}
int32_t initStreamAggSupporter(SStreamAggSupporter* pSup, SqlFunctionCtx* pCtx, int32_t numOfOutput, int64_t gap,
@ -3654,6 +3655,11 @@ static void doStreamSessionAggImpl(SOperatorInfo* pOperator, SSDataBlock* pSData
setSessionWinOutputInfo(pStUpdated, &winInfo);
winRows = updateSessionWindowInfo(&winInfo, startTsCols, endTsCols, groupId, rows, i, pAggSup->gap,
pAggSup->pResultRows, pStUpdated, pStDeleted);
// coverity scan error
if (!winInfo.pOutputBuf) {
T_LONG_JMP(pTaskInfo->env, TSDB_CODE_QRY_OUT_OF_MEMORY);
}
code = doOneWindowAggImpl(&pInfo->twAggSup.timeWindowData, &winInfo, &pResult, i, winRows, rows, numOfOutput,
pOperator);
if (code != TSDB_CODE_SUCCESS || pResult == NULL) {
@ -4102,8 +4108,7 @@ SOperatorInfo* createStreamSessionAggOperatorInfo(SOperatorInfo* downstream, SPh
createOperatorFpSet(operatorDummyOpenFn, doStreamSessionAgg, NULL, destroyStreamSessionAggOperatorInfo, NULL);
if (downstream) {
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType,
pInfo->primaryTsIndex);
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
code = appendDownstream(pOperator, &downstream, 1);
}
return pOperator;
@ -4606,8 +4611,7 @@ SOperatorInfo* createStreamStateAggOperatorInfo(SOperatorInfo* downstream, SPhys
pInfo, pTaskInfo);
pOperator->fpSet =
createOperatorFpSet(operatorDummyOpenFn, doStreamStateAgg, NULL, destroyStreamStateOperatorInfo, NULL);
initDownStream(downstream, &pInfo->streamAggSup, pInfo->twAggSup.waterMark, pOperator->operatorType,
pInfo->primaryTsIndex);
initDownStream(downstream, &pInfo->streamAggSup, pOperator->operatorType, pInfo->primaryTsIndex, &pInfo->twAggSup);
code = appendDownstream(pOperator, &downstream, 1);
if (code != TSDB_CODE_SUCCESS) {
goto _error;

1
source/libs/nodes/src/nodesCloneFuncs.c
View File

@ -434,6 +434,7 @@ static int32_t logicExchangeCopy(const SExchangeLogicNode* pSrc, SExchangeLogicN
COPY_BASE_OBJECT_FIELD(node, logicNodeCopy);
COPY_SCALAR_FIELD(srcStartGroupId);
COPY_SCALAR_FIELD(srcEndGroupId);
COPY_SCALAR_FIELD(seqRecvData);
return TSDB_CODE_SUCCESS;
}

7
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -1864,6 +1864,7 @@ static int32_t jsonToPhysiAggNode(const SJson* pJson, void* pObj) {
static const char* jkExchangePhysiPlanSrcStartGroupId = "SrcStartGroupId";
static const char* jkExchangePhysiPlanSrcEndGroupId = "SrcEndGroupId";
static const char* jkExchangePhysiPlanSrcEndPoints = "SrcEndPoints";
static const char* jkExchangePhysiPlanSeqRecvData = "SeqRecvData";
static int32_t physiExchangeNodeToJson(const void* pObj, SJson* pJson) {
const SExchangePhysiNode* pNode = (const SExchangePhysiNode*)pObj;
@ -1878,6 +1879,9 @@ static int32_t physiExchangeNodeToJson(const void* pObj, SJson* pJson) {
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkExchangePhysiPlanSrcEndPoints, pNode->pSrcEndPoints);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddBoolToObject(pJson, jkExchangePhysiPlanSeqRecvData, pNode->seqRecvData);
}
return code;
}
@ -1895,6 +1899,9 @@ static int32_t jsonToPhysiExchangeNode(const SJson* pJson, void* pObj) {
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkExchangePhysiPlanSrcEndPoints, &pNode->pSrcEndPoints);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetBoolValue(pJson, jkExchangePhysiPlanSeqRecvData, &pNode->seqRecvData);
}
return code;
}

9
source/libs/nodes/src/nodesMsgFuncs.c
View File

@ -2428,7 +2428,8 @@ enum {
PHY_EXCHANGE_CODE_SRC_START_GROUP_ID,
PHY_EXCHANGE_CODE_SRC_END_GROUP_ID,
PHY_EXCHANGE_CODE_SINGLE_CHANNEL,
PHY_EXCHANGE_CODE_SRC_ENDPOINTS
PHY_EXCHANGE_CODE_SRC_ENDPOINTS,
PHY_EXCHANGE_CODE_SEQ_RECV_DATA
};
static int32_t physiExchangeNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
@ -2447,6 +2448,9 @@ static int32_t physiExchangeNodeToMsg(const void* pObj, STlvEncoder* pEncoder) {
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeObj(pEncoder, PHY_EXCHANGE_CODE_SRC_ENDPOINTS, nodeListToMsg, pNode->pSrcEndPoints);
}
if (TSDB_CODE_SUCCESS == code) {
code = tlvEncodeBool(pEncoder, PHY_EXCHANGE_CODE_SEQ_RECV_DATA, pNode->seqRecvData);
}
return code;
}
@ -2473,6 +2477,9 @@ static int32_t msgToPhysiExchangeNode(STlvDecoder* pDecoder, void* pObj) {
case PHY_EXCHANGE_CODE_SRC_ENDPOINTS:
code = msgToNodeListFromTlv(pTlv, (void**)&pNode->pSrcEndPoints);
break;
case PHY_EXCHANGE_CODE_SEQ_RECV_DATA:
code = tlvDecodeBool(pTlv, &pNode->seqRecvData);
break;
default:
break;
}

2
source/libs/parser/src/parInsertUtil.c
View File

@ -139,8 +139,8 @@ void insSetBoundColumnInfo(SParsedDataColInfo* pColList, SSchema* pSchema, col_i
if (i > 0) {
pColList->cols[i].offset = pColList->cols[i - 1].offset + pSchema[i - 1].bytes;
pColList->cols[i].toffset = pColList->flen;
pColList->flen += TYPE_BYTES[type];
}
pColList->flen += TYPE_BYTES[type];
switch (type) {
case TSDB_DATA_TYPE_BINARY:
pColList->allNullLen += (VARSTR_HEADER_SIZE + CHAR_BYTES);

17
source/libs/parser/src/parTranslater.c
View File

@ -3685,9 +3685,19 @@ static int32_t translateInsertProject(STranslateContext* pCxt, SInsertStmt* pIns
return addOrderByPrimaryKeyToQuery(pCxt, pPrimaryKeyExpr, pInsert->pQuery);
}
static int32_t translateInsertTable(STranslateContext* pCxt, SNode* pTable) {
int32_t code = translateFrom(pCxt, pTable);
if (TSDB_CODE_SUCCESS == code && TSDB_CHILD_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType &&
TSDB_NORMAL_TABLE != ((SRealTableNode*)pTable)->pMeta->tableType) {
code = generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_SYNTAX_ERROR,
"insert data into super table is not supported");
}
return code;
}
static int32_t translateInsert(STranslateContext* pCxt, SInsertStmt* pInsert) {
pCxt->pCurrStmt = (SNode*)pInsert;
int32_t code = translateFrom(pCxt, pInsert->pTable);
int32_t code = translateInsertTable(pCxt, pInsert->pTable);
if (TSDB_CODE_SUCCESS == code) {
code = translateInsertCols(pCxt, pInsert);
}
@ -7089,9 +7099,10 @@ static int32_t rewriteDropTable(STranslateContext* pCxt, SQuery* pQuery) {
static int32_t buildUpdateTagValReq(STranslateContext* pCxt, SAlterTableStmt* pStmt, STableMeta* pTableMeta,
SVAlterTbReq* pReq) {
SSchema* pSchema = getColSchema(pTableMeta, pStmt->colName);
SSchema* pSchema = getTagSchema(pTableMeta, pStmt->colName);
if (NULL == pSchema) {
return generateSyntaxErrMsg(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE);
return generateSyntaxErrMsgExt(&pCxt->msgBuf, TSDB_CODE_PAR_INVALID_ALTER_TABLE, "Invalid tag name: %s",
pStmt->colName);
}
pReq->tagName = strdup(pStmt->colName);

98
source/libs/planner/src/planLogicCreater.c
View File

@ -36,6 +36,7 @@ static int32_t createQueryLogicNode(SLogicPlanContext* pCxt, SNode* pStmt, SLogi
typedef struct SRewriteExprCxt {
int32_t errCode;
SNodeList* pExprs;
bool* pOutputs;
} SRewriteExprCxt;
static void setColumnInfo(SFunctionNode* pFunc, SColumnNode* pCol) {
@ -63,14 +64,30 @@ static void setColumnInfo(SFunctionNode* pFunc, SColumnNode* pCol) {
}
static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
SRewriteExprCxt* pCxt = (SRewriteExprCxt*)pContext;
switch (nodeType(*pNode)) {
case QUERY_NODE_COLUMN: {
if (NULL != pCxt->pOutputs) {
SNode* pExpr;
int32_t index = 0;
FOREACH(pExpr, pCxt->pExprs) {
if (QUERY_NODE_GROUPING_SET == nodeType(pExpr)) {
pExpr = nodesListGetNode(((SGroupingSetNode*)pExpr)->pParameterList, 0);
}
if (nodesEqualNode(pExpr, *pNode)) {
pCxt->pOutputs[index] = true;
break;
}
}
}
break;
}
case QUERY_NODE_OPERATOR:
case QUERY_NODE_LOGIC_CONDITION:
case QUERY_NODE_FUNCTION:
case QUERY_NODE_CASE_WHEN: {
SRewriteExprCxt* pCxt = (SRewriteExprCxt*)pContext;
SNode* pExpr;
int32_t index = 0;
SNode* pExpr;
int32_t index = 0;
FOREACH(pExpr, pCxt->pExprs) {
if (QUERY_NODE_GROUPING_SET == nodeType(pExpr)) {
pExpr = nodesListGetNode(((SGroupingSetNode*)pExpr)->pParameterList, 0);
@ -89,6 +106,9 @@ static EDealRes doRewriteExpr(SNode** pNode, void* pContext) {
}
nodesDestroyNode(*pNode);
*pNode = (SNode*)pCol;
if (NULL != pCxt->pOutputs) {
pCxt->pOutputs[index] = true;
}
return DEAL_RES_IGNORE_CHILD;
}
++index;
@ -121,7 +141,7 @@ static EDealRes doNameExpr(SNode* pNode, void* pContext) {
static int32_t rewriteExprForSelect(SNode* pExpr, SSelectStmt* pSelect, ESqlClause clause) {
nodesWalkExpr(pExpr, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = NULL};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = NULL, .pOutputs = NULL};
cxt.errCode = nodesListMakeAppend(&cxt.pExprs, pExpr);
if (TSDB_CODE_SUCCESS == cxt.errCode) {
nodesRewriteSelectStmt(pSelect, clause, doRewriteExpr, &cxt);
@ -130,23 +150,50 @@ static int32_t rewriteExprForSelect(SNode* pExpr, SSelectStmt* pSelect, ESqlClau
return cxt.errCode;
}
static int32_t rewriteExprsForSelect(SNodeList* pExprs, SSelectStmt* pSelect, ESqlClause clause) {
static int32_t cloneRewriteExprs(SNodeList* pExprs, bool* pOutputs, SNodeList** pRewriteExpr) {
int32_t code = TSDB_CODE_SUCCESS;
int32_t index = 0;
SNode* pExpr = NULL;
FOREACH(pExpr, pExprs) {
if (pOutputs[index]) {
code = nodesListMakeStrictAppend(pRewriteExpr, nodesCloneNode(pExpr));
if (TSDB_CODE_SUCCESS != code) {
NODES_DESTORY_LIST(*pRewriteExpr);
break;
}
}
}
return code;
}
static int32_t rewriteExprsForSelect(SNodeList* pExprs, SSelectStmt* pSelect, ESqlClause clause,
SNodeList** pRewriteExprs) {
nodesWalkExprs(pExprs, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs, .pOutputs = NULL};
if (NULL != pRewriteExprs) {
cxt.pOutputs = taosMemoryCalloc(LIST_LENGTH(pExprs), sizeof(bool));
if (NULL == cxt.pOutputs) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
nodesRewriteSelectStmt(pSelect, clause, doRewriteExpr, &cxt);
if (TSDB_CODE_SUCCESS == cxt.errCode && NULL != pRewriteExprs) {
cxt.errCode = cloneRewriteExprs(pExprs, cxt.pOutputs, pRewriteExprs);
}
taosMemoryFree(cxt.pOutputs);
return cxt.errCode;
}
static int32_t rewriteExpr(SNodeList* pExprs, SNode** pTarget) {
nodesWalkExprs(pExprs, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs, .pOutputs = NULL};
nodesRewriteExpr(pTarget, doRewriteExpr, &cxt);
return cxt.errCode;
}
static int32_t rewriteExprs(SNodeList* pExprs, SNodeList* pTarget) {
nodesWalkExprs(pExprs, doNameExpr, NULL);
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs};
SRewriteExprCxt cxt = {.errCode = TSDB_CODE_SUCCESS, .pExprs = pExprs, .pOutputs = NULL};
nodesRewriteExprs(pTarget, doRewriteExpr, &cxt);
return cxt.errCode;
}
@ -311,7 +358,7 @@ static int32_t createScanLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pScan->pScanPseudoCols, pSelect, SQL_CLAUSE_FROM);
code = rewriteExprsForSelect(pScan->pScanPseudoCols, pSelect, SQL_CLAUSE_FROM, NULL);
}
pScan->scanType = getScanType(pCxt, pScan->pScanPseudoCols, pScan->pScanCols, pScan->tableType, pSelect->tagScan);
@ -509,23 +556,20 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pAggFuncs) {
code = rewriteExprsForSelect(pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY);
code = rewriteExprsForSelect(pAgg->pAggFuncs, pSelect, SQL_CLAUSE_GROUP_BY, NULL);
}
if (NULL != pSelect->pGroupByList) {
if (NULL != pAgg->pGroupKeys) {
code = nodesListStrictAppendList(pAgg->pGroupKeys, nodesCloneList(pSelect->pGroupByList));
} else {
pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
if (NULL == pAgg->pGroupKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
pAgg->pGroupKeys = nodesCloneList(pSelect->pGroupByList);
if (NULL == pAgg->pGroupKeys) {
code = TSDB_CODE_OUT_OF_MEMORY;
}
}
// rewrite the expression in subsequent clauses
SNodeList* pOutputGroupKeys = NULL;
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY);
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_GROUP_BY, &pOutputGroupKeys);
}
if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pHaving) {
@ -536,9 +580,11 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
}
// set the output
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pGroupKeys) {
code = createColumnByRewriteExprs(pAgg->pGroupKeys, &pAgg->node.pTargets);
if (TSDB_CODE_SUCCESS == code && NULL != pOutputGroupKeys) {
code = createColumnByRewriteExprs(pOutputGroupKeys, &pAgg->node.pTargets);
}
nodesDestroyList(pOutputGroupKeys);
if (TSDB_CODE_SUCCESS == code && NULL != pAgg->pAggFuncs) {
code = createColumnByRewriteExprs(pAgg->pAggFuncs, &pAgg->node.pTargets);
}
@ -574,7 +620,7 @@ static int32_t createIndefRowsFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt
// indefinite rows functions and _select_values functions
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, fmIsVectorFunc, &pIdfRowsFunc->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pIdfRowsFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT);
code = rewriteExprsForSelect(pIdfRowsFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT, NULL);
}
// set the output
@ -612,7 +658,7 @@ static int32_t createInterpFuncLogicNode(SLogicPlanContext* pCxt, SSelectStmt* p
// interp functions and _group_key functions
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_SELECT, isInterpFunc, &pInterpFunc->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pInterpFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT);
code = rewriteExprsForSelect(pInterpFunc->pFuncs, pSelect, SQL_CLAUSE_SELECT, NULL);
}
if (TSDB_CODE_SUCCESS == code && NULL != pSelect->pFill) {
@ -656,7 +702,7 @@ static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStm
int32_t code = nodesCollectFuncs(pSelect, SQL_CLAUSE_WINDOW, fmIsWindowClauseFunc, &pWindow->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW);
code = rewriteExprsForSelect(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW, NULL);
}
if (TSDB_CODE_SUCCESS == code) {
@ -854,10 +900,10 @@ static int32_t createFillLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect
int32_t code = partFillExprs(pSelect, &pFill->pFillExprs, &pFill->pNotFillExprs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFill->pFillExprs, pSelect, SQL_CLAUSE_FILL);
code = rewriteExprsForSelect(pFill->pFillExprs, pSelect, SQL_CLAUSE_FILL, NULL);
}
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pFill->pNotFillExprs, pSelect, SQL_CLAUSE_FILL);
code = rewriteExprsForSelect(pFill->pNotFillExprs, pSelect, SQL_CLAUSE_FILL, NULL);
}
if (TSDB_CODE_SUCCESS == code) {
code = createColumnByRewriteExprs(pFill->pFillExprs, &pFill->node.pTargets);
@ -1066,7 +1112,7 @@ static int32_t createDistinctLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSe
// rewrite the expression in subsequent clauses
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_DISTINCT);
code = rewriteExprsForSelect(pAgg->pGroupKeys, pSelect, SQL_CLAUSE_DISTINCT, NULL);
}
// set the output

26
source/libs/planner/src/planOptimizer.c
View File

@ -1476,19 +1476,33 @@ static bool partTagsHasIndefRowsSelectFunc(SNodeList* pFuncs) {
return false;
}
static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, int32_t start, SNodeList* pAggFuncs) {
bool hasIndefRowsSelectFunc = partTagsHasIndefRowsSelectFunc(pAggFuncs);
static bool partTagsNeedOutput(SNode* pExpr, SNodeList* pTargets) {
SNode* pOutput = NULL;
FOREACH(pOutput, pTargets) {
if (QUERY_NODE_COLUMN == nodeType(pExpr)) {
if (nodesEqualNode(pExpr, pOutput)) {
return true;
}
} else if (0 == strcmp(((SExprNode*)pExpr)->aliasName, ((SColumnNode*)pOutput)->colName)) {
return true;
}
}
return false;
}
static int32_t partTagsRewriteGroupTagsToFuncs(SNodeList* pGroupTags, int32_t start, SAggLogicNode* pAgg) {
bool hasIndefRowsSelectFunc = partTagsHasIndefRowsSelectFunc(pAgg->pAggFuncs);
int32_t code = TSDB_CODE_SUCCESS;
int32_t index = 0;
SNode* pNode = NULL;
FOREACH(pNode, pGroupTags) {
if (index++ < start) {
if (index++ < start || !partTagsNeedOutput(pNode, pAgg->node.pTargets)) {
continue;
}
if (hasIndefRowsSelectFunc) {
code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_select_value", pNode));
code = nodesListStrictAppend(pAgg->pAggFuncs, partTagsCreateWrapperFunc("_select_value", pNode));
} else {
code = nodesListStrictAppend(pAggFuncs, partTagsCreateWrapperFunc("_group_key", pNode));
code = nodesListStrictAppend(pAgg->pAggFuncs, partTagsCreateWrapperFunc("_group_key", pNode));
}
if (TSDB_CODE_SUCCESS != code) {
break;
@ -1541,7 +1555,7 @@ static int32_t partTagsOptimize(SOptimizeContext* pCxt, SLogicSubplan* pLogicSub
}
NODES_DESTORY_LIST(pAgg->pGroupKeys);
if (TSDB_CODE_SUCCESS == code && start >= 0) {
code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, start, pAgg->pAggFuncs);
code = partTagsRewriteGroupTagsToFuncs(pScan->pGroupTags, start, pAgg);
}
}
if (TSDB_CODE_SUCCESS == code) {

1
source/libs/planner/src/planPhysiCreater.c
View File

@ -1064,6 +1064,7 @@ static int32_t doCreateExchangePhysiNode(SPhysiPlanContext* pCxt, SExchangeLogic
pExchange->srcStartGroupId = pExchangeLogicNode->srcStartGroupId;
pExchange->srcEndGroupId = pExchangeLogicNode->srcEndGroupId;
pExchange->seqRecvData = pExchangeLogicNode->seqRecvData;
*pPhyNode = (SPhysiNode*)pExchange;
return TSDB_CODE_SUCCESS;

93
source/libs/planner/src/planSpliter.c
View File

@ -292,6 +292,43 @@ static bool stbSplNeedSplitJoin(bool streamQuery, SJoinLogicNode* pJoin) {
return true;
}
static SNodeList* stbSplGetPartKeys(SLogicNode* pNode) {
if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) {
return ((SScanLogicNode*)pNode)->pGroupTags;
} else if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
return ((SPartitionLogicNode*)pNode)->pPartitionKeys;
} else {
return NULL;
}
}
static bool stbSplHasPartTbname(SNodeList* pPartKeys) {
if (NULL == pPartKeys) {
return false;
}
SNode* pPartKey = NULL;
FOREACH(pPartKey, pPartKeys) {
if (QUERY_NODE_GROUPING_SET == nodeType(pPartKey)) {
pPartKey = nodesListGetNode(((SGroupingSetNode*)pPartKey)->pParameterList, 0);
}
if ((QUERY_NODE_FUNCTION == nodeType(pPartKey) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pPartKey)->funcType) ||
(QUERY_NODE_COLUMN == nodeType(pPartKey) && COLUMN_TYPE_TBNAME == ((SColumnNode*)pPartKey)->colType)) {
return true;
}
}
return false;
}
static bool stbSplIsPartTableAgg(SAggLogicNode* pAgg) {
if (NULL != pAgg->pGroupKeys) {
return stbSplHasPartTbname(pAgg->pGroupKeys);
}
if (1 != LIST_LENGTH(pAgg->node.pChildren)) {
return false;
}
return stbSplHasPartTbname(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pAgg->node.pChildren, 0)));
}
static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
switch (nodeType(pNode)) {
case QUERY_NODE_LOGIC_PLAN_SCAN:
@ -301,7 +338,9 @@ static bool stbSplNeedSplit(bool streamQuery, SLogicNode* pNode) {
case QUERY_NODE_LOGIC_PLAN_PARTITION:
return streamQuery ? false : stbSplIsMultiTbScanChild(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_AGG:
return !stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) && stbSplHasMultiTbScan(streamQuery, pNode);
return (!stbSplHasGatherExecFunc(((SAggLogicNode*)pNode)->pAggFuncs) ||
stbSplIsPartTableAgg((SAggLogicNode*)pNode)) &&
stbSplHasMultiTbScan(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_WINDOW:
return stbSplNeedSplitWindow(streamQuery, pNode);
case QUERY_NODE_LOGIC_PLAN_SORT:
@ -676,27 +715,8 @@ static int32_t stbSplSplitState(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
}
}
static SNodeList* stbSplGetPartKeys(SLogicNode* pNode) {
if (QUERY_NODE_LOGIC_PLAN_SCAN == nodeType(pNode)) {
return ((SScanLogicNode*)pNode)->pGroupTags;
} else if (QUERY_NODE_LOGIC_PLAN_PARTITION == nodeType(pNode)) {
return ((SPartitionLogicNode*)pNode)->pPartitionKeys;
} else {
return NULL;
}
}
static bool stbSplIsPartTbanme(SNodeList* pPartKeys) {
if (NULL == pPartKeys || 1 != LIST_LENGTH(pPartKeys)) {
return false;
}
SNode* pPartKey = nodesListGetNode(pPartKeys, 0);
return (QUERY_NODE_FUNCTION == nodeType(pPartKey) && FUNCTION_TYPE_TBNAME == ((SFunctionNode*)pPartKey)->funcType) ||
(QUERY_NODE_COLUMN == nodeType(pPartKey) && COLUMN_TYPE_TBNAME == ((SColumnNode*)pPartKey)->colType);
}
static bool stbSplIsPartTableWinodw(SWindowLogicNode* pWindow) {
return stbSplIsPartTbanme(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pWindow->node.pChildren, 0)));
return stbSplHasPartTbname(stbSplGetPartKeys((SLogicNode*)nodesListGetNode(pWindow->node.pChildren, 0)));
}
static int32_t stbSplSplitWindowForCrossTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
@ -713,6 +733,17 @@ static int32_t stbSplSplitWindowForCrossTable(SSplitContext* pCxt, SStableSplitI
return TSDB_CODE_PLAN_INTERNAL_ERROR;
}
static bool stbSplNeedSeqRecvData(SLogicNode* pNode) {
if (NULL == pNode) {
return false;
}
if (NULL != pNode->pLimit || NULL != pNode->pSlimit) {
return true;
}
return stbSplNeedSeqRecvData(pNode->pParent);
}
static int32_t stbSplSplitWindowForPartTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
if (pCxt->pPlanCxt->streamQuery) {
SPLIT_FLAG_SET_MASK(pInfo->pSubplan->splitFlag, SPLIT_FLAG_STABLE_SPLIT);
@ -728,6 +759,7 @@ static int32_t stbSplSplitWindowForPartTable(SSplitContext* pCxt, SStableSplitIn
code = replaceLogicNode(pInfo->pSubplan, pInfo->pSplitNode, (SLogicNode*)pExchange);
}
if (TSDB_CODE_SUCCESS == code) {
pExchange->seqRecvData = stbSplNeedSeqRecvData((SLogicNode*)pExchange);
code = nodesListMakeStrictAppend(&pInfo->pSubplan->pChildren,
(SNode*)splCreateScanSubplan(pCxt, pInfo->pSplitNode, SPLIT_FLAG_STABLE_SPLIT));
}
@ -797,7 +829,17 @@ static int32_t stbSplCreatePartAggNode(SAggLogicNode* pMergeAgg, SLogicNode** pO
return code;
}
static int32_t stbSplSplitAggNode(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
static int32_t stbSplSplitAggNodeForPartTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
int32_t code = splCreateExchangeNodeForSubplan(pCxt, pInfo->pSubplan, pInfo->pSplitNode, SUBPLAN_TYPE_MERGE);
if (TSDB_CODE_SUCCESS == code) {
code = nodesListMakeStrictAppend(&pInfo->pSubplan->pChildren,
(SNode*)splCreateScanSubplan(pCxt, pInfo->pSplitNode, SPLIT_FLAG_STABLE_SPLIT));
}
++(pCxt->groupId);
return code;
}
static int32_t stbSplSplitAggNodeForCrossTable(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
SLogicNode* pPartAgg = NULL;
int32_t code = stbSplCreatePartAggNode((SAggLogicNode*)pInfo->pSplitNode, &pPartAgg);
if (TSDB_CODE_SUCCESS == code) {
@ -812,6 +854,13 @@ static int32_t stbSplSplitAggNode(SSplitContext* pCxt, SStableSplitInfo* pInfo)
return code;
}
static int32_t stbSplSplitAggNode(SSplitContext* pCxt, SStableSplitInfo* pInfo) {
if (stbSplIsPartTableAgg((SAggLogicNode*)pInfo->pSplitNode)) {
return stbSplSplitAggNodeForPartTable(pCxt, pInfo);
}
return stbSplSplitAggNodeForCrossTable(pCxt, pInfo);
}
static SNode* stbSplCreateColumnNode(SExprNode* pExpr) {
SColumnNode* pCol = (SColumnNode*)nodesMakeNode(QUERY_NODE_COLUMN);
if (NULL == pCol) {

2
source/libs/sync/inc/syncInt.h
View File

@ -232,7 +232,7 @@ int32_t syncNodeSendMsgById(const SRaftId* destRaftId, SSyncNode* pSyncNode, S
int32_t syncNodeSendMsgByInfo(const SNodeInfo* nodeInfo, SSyncNode* pSyncNode, SRpcMsg* pMsg);
SyncIndex syncMinMatchIndex(SSyncNode* pSyncNode);
int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHandle** h);
bool syncNodeHeartbeatTimeout(SSyncNode* pSyncNode);
bool syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode);
// raft state change --------------
void syncNodeUpdateTerm(SSyncNode* pSyncNode, SyncTerm term);

1
source/libs/sync/inc/syncMessage.h
View File

@ -124,6 +124,7 @@ typedef struct SyncHeartbeat {
SyncIndex commitIndex;
SyncTerm privateTerm;
SyncTerm minMatchIndex;
int64_t timeStamp;
} SyncHeartbeat;
typedef struct SyncHeartbeatReply {

17
source/libs/sync/src/syncMain.c
View File

@ -640,7 +640,7 @@ int32_t syncNodePropose(SSyncNode* pSyncNode, SRpcMsg* pMsg, bool isWeak) {
}
// heartbeat timeout
if (syncNodeHeartbeatTimeout(pSyncNode)) {
if (syncNodeHeartbeatReplyTimeout(pSyncNode)) {
terrno = TSDB_CODE_SYN_PROPOSE_NOT_READY;
sNError(pSyncNode, "failed to sync propose since hearbeat timeout, type:%s, last:%" PRId64 ", cmt:%" PRId64,
TMSG_INFO(pMsg->msgType), syncNodeGetLastIndex(pSyncNode), pSyncNode->commitIndex);
@ -2039,6 +2039,7 @@ static void syncNodeEqPeerHeartbeatTimer(void* param, void* tmrId) {
pSyncMsg->commitIndex = pSyncNode->commitIndex;
pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
pSyncMsg->privateTerm = 0;
pSyncMsg->timeStamp = taosGetTimestampMs();
// send msg
syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);
@ -2094,7 +2095,7 @@ int32_t syncCacheEntry(SSyncLogStore* pLogStore, SSyncRaftEntry* pEntry, LRUHand
return code;
}
bool syncNodeHeartbeatTimeout(SSyncNode* pSyncNode) {
bool syncNodeHeartbeatReplyTimeout(SSyncNode* pSyncNode) {
if (pSyncNode->replicaNum == 1) {
return false;
}
@ -2148,7 +2149,11 @@ static int32_t syncNodeAppendNoop(SSyncNode* ths) {
int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncHeartbeat* pMsg = pRpcMsg->pCont;
syncLogRecvHeartbeat(ths, pMsg, "");
int64_t tsMs = taosGetTimestampMs();
char buf[128];
snprintf(buf, sizeof(buf), "recv local time:%" PRId64, tsMs);
syncLogRecvHeartbeat(ths, pMsg, buf);
SRpcMsg rpcMsg = {0};
(void)syncBuildHeartbeatReply(&rpcMsg, ths->vgId);
@ -2161,6 +2166,8 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
pMsgReply->timeStamp = taosGetTimestampMs();
if (pMsg->term == ths->pRaftStore->currentTerm && ths->state != TAOS_SYNC_STATE_LEADER) {
syncIndexMgrSetRecvTime(ths->pNextIndex, &(pMsg->srcId), tsMs);
syncNodeResetElectTimer(ths);
ths->minMatchIndex = pMsg->minMatchIndex;
@ -2220,9 +2227,11 @@ int32_t syncNodeOnHeartbeat(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
int32_t syncNodeOnHeartbeatReply(SSyncNode* ths, const SRpcMsg* pRpcMsg) {
SyncHeartbeatReply* pMsg = pRpcMsg->pCont;
syncLogRecvHeartbeatReply(ths, pMsg, "");
int64_t tsMs = taosGetTimestampMs();
char buf[128];
snprintf(buf, sizeof(buf), "recv local time:%" PRId64, tsMs);
syncLogRecvHeartbeatReply(ths, pMsg, buf);
// update last reply time, make decision whether the other node is alive or not
syncIndexMgrSetRecvTime(ths->pMatchIndex, &pMsg->srcId, tsMs);

3
source/libs/sync/src/syncReplication.c
View File

@ -213,9 +213,11 @@ int32_t syncNodeSendHeartbeat(SSyncNode* pSyncNode, const SRaftId* destId, SRpcM
}
int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
int64_t ts = taosGetTimestampMs();
for (int32_t i = 0; i < pSyncNode->peersNum; ++i) {
SRpcMsg rpcMsg = {0};
if (syncBuildHeartbeat(&rpcMsg, pSyncNode->vgId) != 0) {
sError("vgId:%d, build sync-heartbeat error", pSyncNode->vgId);
continue;
}
@ -226,6 +228,7 @@ int32_t syncNodeHeartbeatPeers(SSyncNode* pSyncNode) {
pSyncMsg->commitIndex = pSyncNode->commitIndex;
pSyncMsg->minMatchIndex = syncMinMatchIndex(pSyncNode);
pSyncMsg->privateTerm = 0;
pSyncMsg->timeStamp = ts;
// send msg
syncNodeSendHeartbeat(pSyncNode, &pSyncMsg->destId, &rpcMsg);

55
source/libs/sync/src/syncUtil.c
View File

@ -15,6 +15,7 @@
#define _DEFAULT_SOURCE
#include "syncUtil.h"
#include "syncIndexMgr.h"
#include "syncMessage.h"
#include "syncRaftCfg.h"
#include "syncRaftStore.h"
@ -175,6 +176,36 @@ void syncCfg2SimpleStr(const SSyncCfg* pCfg, char* buf, int32_t bufLen) {
}
}
// for leader
static void syncHearbeatReplyTime2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int32_t len = 5;
for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
int64_t tsMs = syncIndexMgrGetRecvTime(pSyncNode->pMatchIndex, &(pSyncNode->replicasId[i]));
if (i < pSyncNode->replicaNum - 1) {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 ",", i, tsMs);
} else {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 "}", i, tsMs);
}
}
}
// for follower
static void syncHearbeatTime2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int32_t len = 4;
for (int32_t i = 0; i < pSyncNode->replicaNum; ++i) {
int64_t tsMs = syncIndexMgrGetRecvTime(pSyncNode->pNextIndex, &(pSyncNode->replicasId[i]));
if (i < pSyncNode->replicaNum - 1) {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 ",", i, tsMs);
} else {
len += snprintf(buf + len, bufLen - len, "%d:%" PRId64 "}", i, tsMs);
}
}
}
static void syncPeerState2Str(SSyncNode* pSyncNode, char* buf, int32_t bufLen) {
int32_t len = 1;
@ -221,6 +252,12 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
char peerStr[1024] = "{";
syncPeerState2Str(pNode, peerStr, sizeof(peerStr));
char hbrTimeStr[256] = "hbr:{";
syncHearbeatReplyTime2Str(pNode, hbrTimeStr, sizeof(hbrTimeStr));
char hbTimeStr[256] = "hb:{";
syncHearbeatTime2Str(pNode, hbTimeStr, sizeof(hbTimeStr));
int32_t quorum = syncNodeDynamicQuorum(pNode);
char eventLog[512]; // {0};
@ -243,12 +280,13 @@ void syncPrintNodeLog(const char* flags, ELogLevel level, int32_t dflag, SSyncNo
"%s"
", tm:%" PRIu64 ", cmt:%" PRId64 ", fst:%" PRId64 ", lst:%" PRId64 ", min:%" PRId64 ", snap:%" PRId64
", snap-tm:%" PRIu64 ", sby:%d, aq:%d, snaping:%" PRId64 ", r-num:%d, lcfg:%" PRId64
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s",
", chging:%d, rsto:%d, dquorum:%d, elt:%" PRId64 ", hb:%" PRId64 ", %s, %s, %s, %s",
pNode->vgId, syncStr(pNode->state), eventLog, currentTerm, pNode->commitIndex, logBeginIndex,
logLastIndex, pNode->minMatchIndex, snapshot.lastApplyIndex, snapshot.lastApplyTerm,
pNode->pRaftCfg->isStandBy, aqItems, pNode->snapshottingIndex, pNode->replicaNum,
pNode->pRaftCfg->lastConfigIndex, pNode->changing, pNode->restoreFinish, quorum,
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr);
pNode->electTimerLogicClock, pNode->heartbeatTimerLogicClockUser, peerStr, cfgStr, hbTimeStr,
hbrTimeStr);
}
}
@ -395,9 +433,8 @@ void syncLogSendHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const
syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode,
"send sync-heartbeat to %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", pterm:%" PRId64
"}, %s",
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->privateTerm, s);
"send sync-heartbeat to %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64 "}, %s",
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, s);
}
void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const char* s) {
@ -406,9 +443,9 @@ void syncLogRecvHeartbeat(SSyncNode* pSyncNode, const SyncHeartbeat* pMsg, const
syncUtilU642Addr(pMsg->srcId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode,
"recv sync-heartbeat from %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", pterm:%" PRId64
"recv sync-heartbeat from %s:%d {term:%" PRId64 ", cmt:%" PRId64 ", min-match:%" PRId64 ", ts:%" PRId64
"}, %s",
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->privateTerm, s);
host, port, pMsg->term, pMsg->commitIndex, pMsg->minMatchIndex, pMsg->timeStamp, s);
}
void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {
@ -416,8 +453,8 @@ void syncLogSendHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* p
uint16_t port;
syncUtilU642Addr(pMsg->destId.addr, host, sizeof(host), &port);
sNTrace(pSyncNode, "send sync-heartbeat-reply from %s:%d {term:%" PRId64 ", pterm:%" PRId64 "}, %s", host, port,
pMsg->term, pMsg->privateTerm, s);
sNTrace(pSyncNode, "send sync-heartbeat-reply from %s:%d {term:%" PRId64 ", ts:%" PRId64 "}, %s", host, port,
pMsg->term, pMsg->timeStamp, s);
}
void syncLogRecvHeartbeatReply(SSyncNode* pSyncNode, const SyncHeartbeatReply* pMsg, const char* s) {

22
tests/parallel_test/cases.task
View File

@ -417,14 +417,14 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/user_control.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/fsync.py
,,n,system-test,python3 ./test.py -f 0-others/compatibility.py
#,,,system-test,python3 ./test.py -f 1-insert/alter_database.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
,,,system-test,python3 ./test.py -f 1-insert/alter_database.py
,,,system-test,python3 ./test.py -f 1-insert/influxdb_line_taosc_insert.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/opentsdb_telnet_line_taosc_insert.py
,,,system-test,python3 ./test.py -f 1-insert/opentsdb_json_taosc_insert.py
,,,system-test,python3 ./test.py -f 1-insert/test_stmt_muti_insert_query.py
,,,system-test,python3 ./test.py -f 1-insert/test_stmt_set_tbname_tag.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_stable.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_table.py
#,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/alter_table.py
,,n,system-test,python3 ./test.py -f 1-insert/boundary.py
,,n,system-test,python3 ./test.py -f 1-insert/insertWithMoreVgroup.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/table_comment.py
@ -436,7 +436,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/table_param_ttl.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/update_data_muti_rows.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/db_tb_name_check.py
,,,system-test,python3 ./test.py -f 1-insert/database_pre_suf.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/database_pre_suf.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/InsertFuturets.py
,,y,system-test,./pytest.sh python3 ./test.py -f 0-others/show.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/abs.py
@ -561,8 +561,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sample.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sin.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sin.py -R
,,,system-test,python3 ./test.py -f 2-query/smaTest.py
,,,system-test,python3 ./test.py -f 2-query/smaTest.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/smaTest.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/smaTest.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/spread.py
@ -595,8 +595,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/Today.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/top.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/top.py -R
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ttl_comment.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/ttl_comment.py -R
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/twa.py
@ -617,6 +617,8 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/delete_childtable.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/delete_normaltable.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/keep_expired.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/drop.py
,,y,system-test,./pytest.sh python3 ./test.py -f 1-insert/drop.py -N 3 -M 3 -i False -n 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/join2.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/union1.py
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/concat2.py
@ -818,7 +820,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/count_partition.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/max_partition.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/last_row.py -Q 2
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/case_when.py -Q 2
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/between.py -Q 3
@ -911,7 +913,7 @@
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/count_partition.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/max_partition.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/last_row.py -Q 3
,,,system-test,python3 ./test.py -f 2-query/tsbsQuery.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/tsbsQuery.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/sml.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/interp.py -Q 3
,,y,system-test,./pytest.sh python3 ./test.py -f 2-query/case_when.py -Q 3

14
tests/pytest/util/dnodes.py
View File

@ -312,8 +312,14 @@ class TDDnode:
cmd = "mintty -h never %s -c %s" % (
binPath, self.cfgDir)
else:
cmd = "nohup %s -c %s > /dev/null 2>&1 & " % (
binPath, self.cfgDir)
if self.asan:
asanDir = "%s/sim/asan/dnode%d.asan" % (
self.path, self.index)
cmd = "nohup %s -c %s > /dev/null 2> %s & " % (
binPath, self.cfgDir, asanDir)
else:
cmd = "nohup %s -c %s > /dev/null 2>&1 & " % (
binPath, self.cfgDir)
else:
valgrindCmdline = "valgrind --log-file=\"%s/../log/valgrind.log\" --tool=memcheck --leak-check=full --show-reachable=no --track-origins=yes --show-leak-kinds=all -v --workaround-gcc296-bugs=yes"%self.cfgDir
@ -748,7 +754,7 @@ class TDDnodes:
tdLog.exit("index:%d should on a scale of [1, 10]" % (index))
def StopAllSigint(self):
tdLog.info("stop all dnodes sigint")
tdLog.info("stop all dnodes sigint, asan:%d" % self.asan)
if self.asan:
tdLog.info("execute script: %s" % self.stopDnodesSigintPath)
os.system(self.stopDnodesSigintPath)
@ -756,7 +762,7 @@ class TDDnodes:
return
def stopAll(self):
tdLog.info("stop all dnodes")
tdLog.info("stop all dnodes, asan:%d" % self.asan)
if self.asan:
tdLog.info("execute script: %s" % self.stopDnodesPath)
os.system(self.stopDnodesPath)

60
tests/script/tsim/stream/state0.sim
View File

@ -731,5 +731,65 @@ if $data32 != 1 then
goto loop9
endi
sql drop stream if exists streams5;
sql drop database if exists test5;
sql create database test5;
sql use test5;
sql create table tb (ts timestamp, a int);
sql insert into tb values (now + 1m , 1 );
sql create table b (c timestamp, d int, e int , f int, g double);
sql create stream streams0 trigger at_once into streamt as select _wstart c1, count(*) c2, max(a) c3 from tb state_window(a);
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3;
print data00:$data00
print data01:$data01
sql insert into b values(1648791213000,NULL,NULL,NULL,NULL);
sql select * from streamt order by c1, c2, c3;
print data00:$data00
print data01:$data01
sql insert into b values(1648791213001,1,2,2,2.0);
sql insert into b values(1648791213002,1,3,3,3.0);
sql insert into tb values(1648791213003,1);
sql select * from streamt;
print data00:$data00
print data01:$data01
sql delete from b where c >= 1648791213001 and c <= 1648791213002;
sql insert into b values(1648791223003,2,2,3,1.0); insert into b values(1648791223002,2,3,3,3.0);
sql insert into tb values (now + 1m , 1 );
sql select * from streamt;
print data00:$data00
print data01:$data01
sql insert into b(c,d) values (now + 6m , 6 );
sql delete from b where c >= 1648791213001 and c <= 1648791233005;;
$loop_count = 0
loop10:
sleep 200
sql select c2 from streamt;
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
if $rows != 1 then
print =====rows=$rows
goto loop10
endi
if $data00 != 2 then
print =====data00=$data00
goto loop10
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT

4
tests/system-test/1-insert/alter_database.py
View File

@ -18,7 +18,7 @@ class TDTestCase:
tdSql.init(conn.cursor(), logSql)
self.buffer_boundary = [3, 4097, 8193, 12289, 16384]
self.buffer_error = [self.buffer_boundary[0] -
1, self.buffer_boundary[-1]+1, 12289, 256]
1, self.buffer_boundary[-1]+1, 256]
# pages_boundary >= 64
self.pages_boundary = [64, 128, 512]
self.pages_error = [self.pages_boundary[0]-1]
@ -53,7 +53,7 @@ class TDTestCase:
tdSql.execute('drop database db')
def run(self):
tdSql.error('create database db1 vgroups 10 buffer 12289')
self.alter_buffer()
self.alter_pages()

7
tests/system-test/1-insert/alter_table.py
View File

@ -87,7 +87,6 @@ class TDTestCase:
}
def alter_check_ntb(self):
tdSql.prepare()
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
for i in self.values_list:
@ -107,6 +106,7 @@ class TDTestCase:
v = f'binary({self.binary_length+1})'
v_error = f'binary({self.binary_length-1})'
tdSql.error(f'alter table {self.ntbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.ntbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.ntbname} modify column {key} {v}')
tdSql.query(f'describe {self.ntbname}')
result = tdCom.getOneRow(1,'VARCHAR')
@ -115,6 +115,7 @@ class TDTestCase:
v = f'nchar({self.binary_length+1})'
v_error = f'nchar({self.binary_length-1})'
tdSql.error(f'alter table {self.ntbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.ntbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.ntbname} modify column {key} {v}')
tdSql.query(f'describe {self.ntbname}')
result = tdCom.getOneRow(1,'NCHAR')
@ -122,6 +123,7 @@ class TDTestCase:
else:
for v in self.column_dict.values():
tdSql.error(f'alter table {self.ntbname} modify column {key} {v}')
tdSql.error(f'alter table {self.ntbname} set tag {key} = "abcd1"')
for key,values in self.column_dict.items():
rename_str = f'{tdCom.getLongName(constant.COL_NAME_LENGTH_MAX,"letters")}'
tdSql.execute(f'alter table {self.ntbname} rename column {key} {rename_str}')
@ -285,6 +287,7 @@ class TDTestCase:
v = f'binary({self.binary_length+1})'
v_error = f'binary({self.binary_length-1})'
tdSql.error(f'alter table {self.stbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.stbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.stbname} modify column {key} {v}')
tdSql.query(f'describe {self.stbname}')
result = tdCom.getOneRow(1,'VARCHAR')
@ -297,6 +300,7 @@ class TDTestCase:
v = f'nchar({self.binary_length+1})'
v_error = f'nchar({self.binary_length-1})'
tdSql.error(f'alter table {self.stbname} modify column {key} {v_error}')
tdSql.error(f'alter table {self.stbname} set tag {key} = "abcd1"')
tdSql.execute(f'alter table {self.stbname} modify column {key} {v}')
tdSql.query(f'describe {self.stbname}')
result = tdCom.getOneRow(1,'NCHAR')
@ -308,6 +312,7 @@ class TDTestCase:
else:
for v in self.column_dict.values():
tdSql.error(f'alter table {self.stbname} modify column {key} {v}')
tdSql.error(f'alter table {self.stbname} set tag {key} = "abcd1"')
for key,values in self.column_dict.items():
rename_str = f'{tdCom.getLongName(constant.COL_NAME_LENGTH_MAX,"letters")}'
tdSql.error(f'alter table {self.stbname} rename column {key} {rename_str}')

161
tests/system-test/1-insert/drop.py Normal file
View File

@ -0,0 +1,161 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import math
from util.log import *
from util.cases import *
from util.sql import *
from util.common import *
from util.sqlset import *
class TDTestCase:
def init(self, conn, logSql, replicaVar=1):
self.replicaVar = int(replicaVar)
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor())
self.setsql = TDSetSql()
self.dbname = 'db'
self.ntbname = f"{self.dbname}.ntb"
self.rowNum = 10
self.tbnum = 20
self.ts = 1537146000000
self.binary_str = 'taosdata'
self.nchar_str = '涛思数据'
self.column_dict = {
'ts' : 'timestamp',
'col1': 'tinyint',
'col2': 'smallint',
'col3': 'int',
'col4': 'bigint',
'col5': 'tinyint unsigned',
'col6': 'smallint unsigned',
'col7': 'int unsigned',
'col8': 'bigint unsigned',
'col9': 'float',
'col10': 'double',
'col11': 'bool',
'col12': 'binary(20)',
'col13': 'nchar(20)'
}
def insert_data(self,column_dict,tbname,row_num):
insert_sql = self.setsql.set_insertsql(column_dict,tbname,self.binary_str,self.nchar_str)
for i in range(row_num):
insert_list = []
self.setsql.insert_values(column_dict,i,insert_sql,insert_list,self.ts)
def drop_ntb_check(self):
tdSql.execute(f'create database if not exists {self.dbname} replica {self.replicaVar}')
tdSql.execute(f'use {self.dbname}')
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
self.insert_data(self.column_dict,self.ntbname,self.rowNum)
for k,v in self.column_dict.items():
if v.lower() == "timestamp":
tdSql.query(f'select * from {self.ntbname} where {k} = {self.ts}')
tdSql.checkRows(1)
tdSql.execute(f'drop table {self.ntbname}')
tdSql.execute(f'flush database {self.dbname}')
tdSql.execute(self.setsql.set_create_normaltable_sql(self.ntbname,self.column_dict))
self.insert_data(self.column_dict,self.ntbname,self.rowNum)
for k,v in self.column_dict.items():
if v.lower() == "timestamp":
tdSql.query(f'select * from {self.ntbname} where {k} = {self.ts}')
tdSql.checkRows(1)
tdSql.execute(f'drop database {self.dbname}')
def drop_stb_ctb_check(self):
stbname = f'{self.dbname}.{tdCom.getLongName(5,"letters")}'
tag_dict = {
't0':'int'
}
tag_values = [
f'1'
]
tdSql.execute(f"create database if not exists {self.dbname} replica {self.replicaVar}")
tdSql.execute(f'use {self.dbname}')
tdSql.execute(self.setsql.set_create_stable_sql(stbname,self.column_dict,tag_dict))
for i in range(self.tbnum):
tdSql.execute(f"create table {stbname}_{i} using {stbname} tags({tag_values[0]})")
self.insert_data(self.column_dict,f'{stbname}_{i}',self.rowNum)
for k,v in self.column_dict.items():
for i in range(self.tbnum):
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname}_{i} where {k} = {self.ts}')
tdSql.checkRows(1)
tdSql.execute(f'drop table {stbname}_{i}')
tdSql.execute(f'flush database {self.dbname}')
for i in range(self.tbnum):
tdSql.execute(f"create table {stbname}_{i} using {stbname} tags({tag_values[0]})")
self.insert_data(self.column_dict,f'{stbname}_{i}',self.rowNum)
for k,v in self.column_dict.items():
for i in range(self.tbnum):
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname}_{i} where {k} = {self.ts}')
tdSql.checkRows(1)
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname} where {k} = {self.ts}')
tdSql.checkRows(self.tbnum)
tdSql.execute(f'drop table {stbname}')
tdSql.execute(f'flush database {self.dbname}')
tdSql.execute(self.setsql.set_create_stable_sql(stbname,self.column_dict,tag_dict))
for i in range(self.tbnum):
tdSql.execute(f"create table {stbname}_{i} using {stbname} tags({tag_values[0]})")
self.insert_data(self.column_dict,f'{stbname}_{i}',self.rowNum)
for k,v in self.column_dict.items():
if v.lower() == "timestamp":
tdSql.query(f'select * from {stbname} where {k} = {self.ts}')
tdSql.checkRows(self.tbnum)
tdSql.execute(f'drop database {self.dbname}')
def drop_topic_check(self):
tdSql.execute(f'create database {self.dbname} replica {self.replicaVar}')
tdSql.execute(f'use {self.dbname}')
stbname = tdCom.getLongName(5,"letters")
topic_name = tdCom.getLongName(5,"letters")
tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)')
tdSql.execute(f'create topic {topic_name} as select * from {self.dbname}.{stbname}')
tdSql.query(f'select * from information_schema.ins_topics where topic_name = "{topic_name}"')
tdSql.checkEqual(tdSql.queryResult[0][3],f'create topic {topic_name} as select * from {self.dbname}.{stbname}')
tdSql.execute(f'drop topic {topic_name}')
tdSql.execute(f'create topic {topic_name} as select c0 from {self.dbname}.{stbname}')
tdSql.query(f'select * from information_schema.ins_topics where topic_name = "{topic_name}"')
tdSql.checkEqual(tdSql.queryResult[0][3],f'create topic {topic_name} as select c0 from {self.dbname}.{stbname}')
tdSql.execute(f'drop topic {topic_name}')
tdSql.execute(f'drop database {self.dbname}')
def drop_stream_check(self):
tdSql.execute(f'create database {self.dbname} replica {self.replicaVar}')
tdSql.execute(f'use {self.dbname}')
stbname = tdCom.getLongName(5,"letters")
stream_name = tdCom.getLongName(5,"letters")
tdSql.execute(f'create table {stbname} (ts timestamp,c0 int) tags(t0 int)')
tdSql.execute(f'create table tb using {stbname} tags(1)')
tdSql.execute(f'create stream {stream_name} into stb as select * from {self.dbname}.{stbname} partition by tbname')
tdSql.query(f'select * from information_schema.ins_streams where stream_name = "{stream_name}"')
print(tdSql.queryResult)
tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} into stb as select * from {self.dbname}.{stbname} partition by tbname')
tdSql.execute(f'drop stream {stream_name}')
tdSql.execute(f'create stream {stream_name} into stb1 as select * from tb')
tdSql.query(f'select * from information_schema.ins_streams where stream_name = "{stream_name}"')
tdSql.checkEqual(tdSql.queryResult[0][2],f'create stream {stream_name} into stb1 as select * from tb')
tdSql.execute(f'drop database {self.dbname}')
def run(self):
self.drop_ntb_check()
self.drop_stb_ctb_check()
self.drop_topic_check()
self.drop_stream_check()
pass
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

25
tests/system-test/2-query/smaTest.py
View File

@ -44,8 +44,8 @@ class TDTestCase:
def run(self):
# insert data
dbname = "db"
self.insert_data1(f"{dbname}.t1", self.ts, 1000*10000)
self.insert_data1(f"{dbname}.t4", self.ts, 1000*10000)
self.insert_data1(f"{dbname}.t1", self.ts, 10*10000)
self.insert_data1(f"{dbname}.t4", self.ts, 10*10000)
# test base case
# self.test_case1()
tdLog.debug(" LIMIT test_case1 ............ [OK]")
@ -53,7 +53,6 @@ class TDTestCase:
# self.test_case2()
tdLog.debug(" LIMIT test_case2 ............ [OK]")
# stop
def stop(self):
tdSql.close()
@ -77,15 +76,17 @@ class TDTestCase:
# insert data1
def insert_data(self, tbname, ts_start, count):
pre_insert = "insert into %s values"%tbname
pre_insert = "insert into %s values" % tbname
sql = pre_insert
tdLog.debug("insert table %s rows=%d ..."%(tbname, count))
tdLog.debug("insert table %s rows=%d ..." % (tbname, count))
for i in range(count):
sql += " (%d,%d)"%(ts_start + i*1000, i )
if i >0 and i%30000 == 0:
sql += " (%d,%d)" % (ts_start + i*1000, i)
if i > 0 and i % 20000 == 0:
tdLog.info("%d rows inserted" % i)
tdSql.execute(sql)
sql = pre_insert
# end sql
tdLog.info("insert_data end")
if sql != pre_insert:
tdSql.execute(sql)
@ -93,15 +94,17 @@ class TDTestCase:
return
def insert_data1(self, tbname, ts_start, count):
pre_insert = "insert into %s values"%tbname
pre_insert = "insert into %s values" % tbname
sql = pre_insert
tdLog.debug("insert table %s rows=%d ..."%(tbname, count))
tdLog.debug("insert table %s rows=%d ..." % (tbname, count))
for i in range(count):
sql += " (%d,%d,%d)"%(ts_start + i*1000, i , i+1)
if i >0 and i%30000 == 0:
sql += " (%d,%d,%d)" % (ts_start + i*1000, i, i+1)
if i > 0 and i % 20000 == 0:
tdLog.info("%d rows inserted" % i)
tdSql.execute(sql)
sql = pre_insert
# end sql
tdLog.info("insert_data1 end")
if sql != pre_insert:
tdSql.execute(sql)

1
tests/system-test/test.py
View File

@ -464,6 +464,7 @@ if __name__ == "__main__":
tdDnodes.init(deployPath, masterIp)
tdDnodes.setTestCluster(testCluster)
tdDnodes.setValgrind(valgrind)
tdDnodes.setAsan(asan)
tdDnodes.stopAll()
for dnode in tdDnodes.dnodes:
tdDnodes.deploy(dnode.index,{})

3
tools/shell/inc/shellAuto.h
View File

@ -39,4 +39,7 @@ void callbackAutoTab(char* sqlstr, TAOS* pSql, bool usedb);
// introduction
void printfIntroduction();
// show all commands help
void showHelp();
#endif

43
tools/shell/src/shellAuto.c
View File

@ -108,6 +108,7 @@ SWords shellCommands[] = {
{"drop topic <topic_name> ;", 0, 0, NULL},
{"drop stream <stream_name> ;", 0, 0, NULL},
{"explain select", 0, 0, NULL}, // 44 append sub sql
{"help;", 0, 0, NULL},
{"grant all on <anyword> to <user_name> ;", 0, 0, NULL},
{"grant read on <anyword> to <user_name> ;", 0, 0, NULL},
{"grant write on <anyword> to <user_name> ;", 0, 0, NULL},
@ -386,6 +387,8 @@ void showHelp() {
drop stream <stream_name> ;\n\
----- E ----- \n\
explain select clause ...\n\
----- H ----- \n\
help;\n\
----- I ----- \n\
insert into <tb_name> values(...) ;\n\
insert into <tb_name> using <stb_name> tags(...) values(...) ;\n\
@ -1478,24 +1481,36 @@ bool matchSelectQuery(TAOS* con, SShellCmd* cmd) {
// if is input create fields or tags area, return true
bool isCreateFieldsArea(char* p) {
char* left = strrchr(p, '(');
if (left == NULL) {
// like 'create table st'
return false;
}
// put to while, support like create table st(ts timestamp, bin1 binary(16), bin2 + blank + TAB
char* p1 = strdup(p);
bool ret = false;
while (1) {
char* left = strrchr(p1, '(');
if (left == NULL) {
// like 'create table st'
ret = false;
break;
}
char* right = strrchr(p, ')');
if (right == NULL) {
// like 'create table st( '
return true;
}
char* right = strrchr(p1, ')');
if (right == NULL) {
// like 'create table st( '
ret = true;
break;
}
if (left > right) {
// like 'create table st( ts timestamp, age int) tags(area '
return true;
if (left > right) {
// like 'create table st( ts timestamp, age int) tags(area '
ret = true;
break;
}
// set string end by small for next strrchr search
*left = 0;
}
taosMemoryFree(p1);
return false;
return ret;
}
bool matchCreateTable(TAOS* con, SShellCmd* cmd) {

6
tools/shell/src/shellEngine.c
View File

@ -134,6 +134,12 @@ int32_t shellRunCommand(char *command, bool recordHistory) {
return 0;
}
// add help or help;
if(strcmp(command, "help") == 0 || strcmp(command, "help;") == 0) {
showHelp();
return 0;
}
if (recordHistory) shellRecordCommandToHistory(command);
char quote = 0, *cmd = command;