diff --git a/docs/en/07-develop/03-insert-data/01-sql-writing.mdx b/docs/en/07-develop/03-insert-data/01-sql-writing.mdx index d8c4453f409dfaf1db1ec154e9ba35f8db74862e..da0e87abbc2c83ca940dd596ffbf5746a6b65823 100644 --- a/docs/en/07-develop/03-insert-data/01-sql-writing.mdx +++ b/docs/en/07-develop/03-insert-data/01-sql-writing.mdx @@ -42,7 +42,7 @@ INSERT INTO d1001 VALUES (1538548684000, 10.2, 220, 0.23) (1538548696650, 10.3, ### Insert into Multiple Tables -Data can be inserted into multiple tables in the same SQL statement. The example below inserts 2 rows into table "d1001" and 1 row into table "d1002". +Data can be inserted into multiple tables in single SQL statement. The example below inserts 2 rows into table "d1001" and 1 row into table "d1002". ```sql INSERT INTO d1001 VALUES (1538548685000, 10.3, 219, 0.31) (1538548695000, 12.6, 218, 0.33) d1002 VALUES (1538548696800, 12.3, 221, 0.31); @@ -52,15 +52,15 @@ For more details about `INSERT` please refer to [INSERT](/taos-sql/insert). :::info -- Inserting in batches can improve performance. Normally, the higher the batch size, the better the performance. Please note that a single row can't exceed 48K bytes and each SQL statement can't exceed 1MB. -- Inserting with multiple threads can also improve performance. However, depending on the system resources on the application side and the server side, when the number of inserting threads grows beyond a specific point the performance may drop instead of improving. The proper number of threads needs to be tested in a specific environment to find the best number. +- Inserting in batches can improve performance. Normally, the higher the batch size, the better the performance. Please note that a single row can't exceed 48 KB bytes and each SQL statement can't exceed 1 MB. +- Inserting with multiple threads can also improve performance. However, depending on the system resources on the application side and the server side, when the number of inserting threads grows beyond a specific point the performance may drop instead of improving. The proper number of threads needs to be tested in a specific environment to find the best number. The proper number of threads may be impacted by the system resources on the server side, the system resources on the client side, the table schemas, etc. ::: :::warning -- If the timestamp for the row to be inserted already exists in the table, the behavior depends on the value of parameter `UPDATE`. If it's set to 0 (the default value), the row will be discarded. If it's set to 1, the new values will override the old values for the same row. -- The timestamp to be inserted must be newer than the timestamp of subtracting current time by the parameter `KEEP`. If `KEEP` is set to 3650 days, then the data older than 3650 days ago can't be inserted. The timestamp to be inserted can't be newer than the timestamp of current time plus parameter `DAYS`. If `DAYS` is set to 2, the data newer than 2 days later can't be inserted. +- If the timestamp for the row to be inserted already exists in the table, the old data will be overritten by the new values for the columns for which new values are provided, columns for which no new values are provided are not impacted. +- The timestamp to be inserted must be newer than the timestamp of subtracting current time by the parameter `KEEP`. If `KEEP` is set to 3650 days, then the data older than 3650 days ago can't be inserted. The timestamp to be inserted can't be newer than the timestamp of current time plus parameter `DURATION`. If `DAYS` is set to 2, the data newer than 2 days later can't be inserted. ::: @@ -101,7 +101,7 @@ For more details about `INSERT` please refer to [INSERT](/taos-sql/insert). ### Insert with Parameter Binding -TDengine also provides API support for parameter binding. Similar to MySQL, only `?` can be used in these APIs to represent the parameters to bind. From version 2.1.1.0 and 2.1.2.0, parameter binding support for inserting data has improved significantly to improve the insert performance by avoiding the cost of parsing SQL statements. +TDengine also provides API support for parameter binding. Similar to MySQL, only `?` can be used in these APIs to represent the parameters to bind. Parameter binding support for inserting data has improved significantly to improve the insert performance by avoiding the cost of parsing SQL statements. Parameter binding is available only with native connection. diff --git a/docs/zh/07-develop/03-insert-data/01-sql-writing.mdx b/docs/zh/07-develop/03-insert-data/01-sql-writing.mdx index 99a92573c87d0f90f699a8d1352619f4df4aef39..214cbdaa96d02e0cd1251eeda97c6a897887cc7e 100644 --- a/docs/zh/07-develop/03-insert-data/01-sql-writing.mdx +++ b/docs/zh/07-develop/03-insert-data/01-sql-writing.mdx @@ -52,15 +52,15 @@ INSERT INTO d1001 VALUES (1538548685000, 10.3, 219, 0.31) (1538548695000, 12.6, :::info -- 要提高写入效率，需要批量写入。一批写入的记录条数越多，插入效率就越高。但一条记录不能超过 48K，一条 SQL 语句总长度不能超过 1M 。 -- TDengine 支持多线程同时写入，要进一步提高写入速度，一个客户端需要打开 20 个以上的线程同时写。但线程数达到一定数量后，无法再提高，甚至还会下降，因为线程频繁切换，带来额外开销。 +- 要提高写入效率，需要批量写入。一般来说一批写入的记录条数越多，插入效率就越高。但一条记录不能超过 48K，一条 SQL 语句总长度不能超过 1M 。 +- TDengine 支持多线程同时写入，要进一步提高写入速度，一个客户端需要打开多个同时写。但线程数达到一定数量后，无法再提高，甚至还会下降，因为线程频繁切换，会带来额外开销，合适的线程数量与服务端的处理能力，服务端的具体配置，数据库的参数，数据定义的 Schema，写入数据的 Batch Size 等很多因素相关。一般来说，服务端和客户端处理能力越强，所能支持的并发写入的线程可以越多；数据库配置时的 vgroups 越多（但仍然要在服务端的处理能力以内）则所能支持的并发写入越多；数据定义的 Schema 越简单，所能支持的并发写入越多。 ::: :::warning -- 对同一张表，如果新插入记录的时间戳已经存在，默认情形下（UPDATE=0）新记录将被直接抛弃，也就是说，在一张表里，时间戳必须是唯一的。如果应用自动生成记录，很有可能生成的时间戳是一样的，这样，成功插入的记录条数会小于应用插入的记录条数。如果在创建数据库时使用了 UPDATE 1 选项，插入相同时间戳的新记录将覆盖原有记录。 -- 写入的数据的时间戳必须大于当前时间减去配置参数 keep 的时间。如果 keep 配置为 3650 天，那么无法写入比 3650 天还早的数据。写入数据的时间戳也不能大于当前时间加配置参数 days。如果 days 为 2，那么无法写入比当前时间还晚 2 天的数据。 +- 对同一张表，如果新插入记录的时间戳已经存在，则指定了新值的列会用新值覆盖旧值，而没有指定新值的列则不受影响。 +- 写入的数据的时间戳必须大于当前时间减去配置参数 keep 的时间。如果 keep 配置为 3650 天，那么无法写入比 3650 天还早的数据。写入数据的时间戳也不能大于当前时间加配置参数 duration。如果 duration 为 2，那么无法写入比当前时间还晚 2 天的数据。 ::: @@ -104,7 +104,7 @@ INSERT INTO d1001 VALUES (1538548685000, 10.3, 219, 0.31) (1538548695000, 12.6, ### 参数绑定写入 -TDengine 也提供了支持参数绑定的 Prepare API，与 MySQL 类似，这些 API 目前也仅支持用问号 `?` 来代表待绑定的参数。从 2.1.1.0 和 2.1.2.0 版本开始，TDengine 大幅改进了参数绑定接口对数据写入（INSERT）场景的支持。这样在通过参数绑定接口写入数据时，就避免了 SQL 语法解析的资源消耗，从而在绝大多数情况下显著提升写入性能。 +TDengine 也提供了支持参数绑定的 Prepare API，与 MySQL 类似，这些 API 目前也仅支持用问号 `?` 来代表待绑定的参数。在通过参数绑定接口写入数据时，就避免了 SQL 语法解析的资源消耗，从而在绝大多数情况下显著提升写入性能。 需要注意的是，只有使用原生连接的连接器，才能使用参数绑定功能。 diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index ba92ed238b02489bb09d45fb65a0b0071f6fac9a..6805e4d501e5128fb967bfd2e868a9751d17b10b 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -189,12 +189,15 @@ void destroyTscObj(void *pObj) { SClientHbKey connKey = {.tscRid = pTscObj->id, .connType = pTscObj->connType}; hbDeregisterConn(pTscObj->pAppInfo->pAppHbMgr, connKey); - int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1); + destroyAllRequests(pTscObj->pRequests); + taosHashCleanup(pTscObj->pRequests); + schedulerStopQueryHb(pTscObj->pAppInfo->pTransporter); tscDebug("connObj 0x%" PRIx64 " p:%p destroyed, remain inst totalConn:%" PRId64, pTscObj->id, pTscObj, pTscObj->pAppInfo->numOfConns); + int64_t connNum = atomic_sub_fetch_64(&pTscObj->pAppInfo->numOfConns, 1); if (0 == connNum) { destroyAppInst(pTscObj->pAppInfo); } diff --git a/source/client/src/clientHb.c b/source/client/src/clientHb.c index e2d75d39e34c9077591101e636255027e4b672d7..2a9d113108a565694091de4c3ea0c78431c50b57 100644 --- a/source/client/src/clientHb.c +++ b/source/client/src/clientHb.c @@ -671,8 +671,7 @@ static void *hbThreadFunc(void *param) { } #endif while (1) { - int8_t threadStop = atomic_val_compare_exchange_8(&clientHbMgr.threadStop, 1, 2); - if (1 == threadStop) { + if (1 == clientHbMgr.threadStop) { break; } @@ -760,9 +759,7 @@ static void hbStopThread() { return; } - while (2 != atomic_load_8(&clientHbMgr.threadStop)) { - taosUsleep(10); - } + taosThreadJoin(clientHbMgr.thread, NULL); tscDebug("hb thread stopped"); } diff --git a/source/common/src/systable.c b/source/common/src/systable.c index ba8a8e122049ea567e4d8f04396094b68326f3b0..08997bcaf8ba4f745db9f7d13b131cad47db901b 100644 --- a/source/common/src/systable.c +++ b/source/common/src/systable.c @@ -123,6 +123,9 @@ static const SSysDbTableSchema userStbsSchema[] = { {.name = "tags", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, {.name = "last_update", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP}, {.name = "table_comment", .bytes = TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + {.name = "watermark", .bytes = 64 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + {.name = "max_delay", .bytes = 64 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + {.name = "rollup", .bytes = 128 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, }; static const SSysDbTableSchema streamSchema[] = { @@ -146,8 +149,8 @@ static const SSysDbTableSchema userTblsSchema[] = { {.name = "uid", .bytes = 8, .type = TSDB_DATA_TYPE_BIGINT}, {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, {.name = "ttl", .bytes = 4, .type = TSDB_DATA_TYPE_INT}, - {.name = "table_comment", .bytes = TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, - {.name = "type", .bytes = 20 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + {.name = "table_comment", .bytes = TSDB_TB_COMMENT_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + {.name = "type", .bytes = 21 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, }; static const SSysDbTableSchema userTblDistSchema[] = { diff --git a/source/dnode/mnode/impl/src/mndStb.c b/source/dnode/mnode/impl/src/mndStb.c index 1c234cf2802cb51e1480d7241b3ac7602c2ef565..1e576250285b88aaab8f4aad134bbb240e2646f3 100644 --- a/source/dnode/mnode/impl/src/mndStb.c +++ b/source/dnode/mnode/impl/src/mndStb.c @@ -2109,7 +2109,7 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, numOfRows, (const char *)&pStb->updateTime, false); // number of tables - pColInfo = taosArrayGet(pBlock->pDataBlock, cols); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); if (pStb->commentLen > 0) { char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0}; STR_TO_VARSTR(comment, pStb->comment); @@ -2122,6 +2122,34 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc colDataAppendNULL(pColInfo, numOfRows); } + char watermark[64 + VARSTR_HEADER_SIZE] = {0}; + sprintf(varDataVal(watermark), "%" PRId64 "a,%" PRId64 "a", pStb->watermark[0], pStb->watermark[1]); + varDataSetLen(watermark, strlen(varDataVal(watermark))); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataAppend(pColInfo, numOfRows, (const char *)watermark, false); + + char maxDelay[64 + VARSTR_HEADER_SIZE] = {0}; + sprintf(varDataVal(maxDelay), "%" PRId64 "a,%" PRId64 "a", pStb->maxdelay[0], pStb->maxdelay[1]); + varDataSetLen(maxDelay, strlen(varDataVal(maxDelay))); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataAppend(pColInfo, numOfRows, (const char *)maxDelay, false); + + char rollup[128 + VARSTR_HEADER_SIZE] = {0}; + int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs); + for (int32_t i = 0; i < rollupNum; ++i) { + char *funcName = taosArrayGet(pStb->pFuncs, i); + if (i) { + strcat(varDataVal(rollup), ", "); + } + strcat(varDataVal(rollup), funcName); + } + varDataSetLen(rollup, strlen(varDataVal(rollup))); + + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataAppend(pColInfo, numOfRows, (const char *)rollup, false); + numOfRows++; sdbRelease(pSdb, pStb); } diff --git a/source/libs/catalog/src/catalog.c b/source/libs/catalog/src/catalog.c index 3a7ad4a2d613c22eabbc134acaae1b59175004a1..ca50702894a2a473099fc336c385d01588354173 100644 --- a/source/libs/catalog/src/catalog.c +++ b/source/libs/catalog/src/catalog.c @@ -1293,6 +1293,7 @@ void catalogDestroy(void) { if (!taosCheckCurrentInDll()) { ctgClearCacheEnqueue(NULL, true, true, true); + taosThreadJoin(gCtgMgmt.updateThread, NULL); } taosHashCleanup(gCtgMgmt.pCluster); diff --git a/source/libs/executor/inc/tfill.h b/source/libs/executor/inc/tfill.h index 3b80b262ca41830c569fefb4f4230fa3362b9911..0349632b9aa177a140c21b39e96232bfed8ce8b1 100644 --- a/source/libs/executor/inc/tfill.h +++ b/source/libs/executor/inc/tfill.h @@ -42,6 +42,7 @@ typedef struct SFillInfo { TSKEY start; // start timestamp TSKEY end; // endKey for fill TSKEY currentKey; // current active timestamp, the value may be changed during the fill procedure. + int32_t tsSlotId; // primary time stamp slot id int32_t order; // order [TSDB_ORDER_ASC|TSDB_ORDER_DESC] int32_t type; // fill type int32_t numOfRows; // number of rows in the input data block @@ -74,8 +75,8 @@ struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, co bool taosFillHasMoreResults(struct SFillInfo* pFillInfo); SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols, - SInterval* pInterval, int32_t fillType, - struct SFillColInfo* pCol, const char* id); + SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t slotId, + const char* id); void* taosDestroyFillInfo(struct SFillInfo *pFillInfo); int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, SSDataBlock* p, int32_t capacity); diff --git a/source/libs/executor/src/executorimpl.c b/source/libs/executor/src/executorimpl.c index ac8043205279339a22a957d15d7a516db89f97f5..e9a244b573626e67700d1b3e1d347f1104754277 100644 --- a/source/libs/executor/src/executorimpl.c +++ b/source/libs/executor/src/executorimpl.c @@ -4013,10 +4013,12 @@ static int32_t initFillInfo(SFillOperatorInfo* pInfo, SExprInfo* pExpr, int32_t w = getFirstQualifiedTimeWindow(win.skey, &w, pInterval, TSDB_ORDER_ASC); int32_t order = TSDB_ORDER_ASC; - pInfo->pFillInfo = taosCreateFillInfo(order, w.skey, 0, capacity, numOfCols, pInterval, fillType, pColInfo, id); + pInfo->pFillInfo = taosCreateFillInfo(order, w.skey, 0, capacity, numOfCols, pInterval, + fillType, pColInfo, pInfo->primaryTsCol, id); pInfo->win = win; pInfo->p = taosMemoryCalloc(numOfCols, POINTER_BYTES); + if (pInfo->pFillInfo == NULL || pInfo->p == NULL) { taosMemoryFree(pInfo->pFillInfo); taosMemoryFree(pInfo->p); diff --git a/source/libs/executor/src/tfill.c b/source/libs/executor/src/tfill.c index b0e2166baff8d22d6e7f00421f46704f69415025..c008c7c4a9d0fec83188ae3eee55144242125db3 100644 --- a/source/libs/executor/src/tfill.c +++ b/source/libs/executor/src/tfill.c @@ -14,6 +14,7 @@ */ #include "os.h" +#include "query.h" #include "taosdef.h" #include "tmsg.h" #include "ttypes.h" @@ -48,14 +49,15 @@ static void setTagsValue(SFillInfo* pFillInfo, void** data, int32_t genRows) { } } -static void setNullRow(SSDataBlock* pBlock, int32_t numOfCol, int32_t rowIndex) { +static void setNullRow(SSDataBlock* pBlock, int64_t ts, int32_t rowIndex) { // the first are always the timestamp column, so start from the second column. for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) { SColumnInfoData* p = taosArrayGet(pBlock->pDataBlock, i); - if (p->info.type == TSDB_DATA_TYPE_TIMESTAMP && i == 0) { - continue; + if (p->info.type == TSDB_DATA_TYPE_TIMESTAMP) { + colDataAppend(p, rowIndex, (const char*)&ts, false); + } else { + colDataAppendNULL(p, rowIndex); } - colDataAppendNULL(p, rowIndex); } } @@ -64,16 +66,17 @@ static void setNullRow(SSDataBlock* pBlock, int32_t numOfCol, int32_t rowIndex) static void doSetVal(SColumnInfoData* pDstColInfoData, int32_t rowIndex, const SGroupKeys* pKey); -static void doFillOneRowResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSDataBlock* pSrcBlock, int64_t ts, - bool outOfBound) { +static void doFillOneRow(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSDataBlock* pSrcBlock, int64_t ts, + bool outOfBound) { SPoint point1, point2, point; int32_t step = GET_FORWARD_DIRECTION_FACTOR(pFillInfo->order); // set the primary timestamp column value int32_t index = pFillInfo->numOfCurrent; - SColumnInfoData* pCol0 = taosArrayGet(pBlock->pDataBlock, 0); + SColumnInfoData* pCol0 = taosArrayGet(pBlock->pDataBlock, pFillInfo->tsSlotId); char* val = colDataGetData(pCol0, index); + // set the primary timestamp value *(TSKEY*)val = pFillInfo->currentKey; // set the other values @@ -92,7 +95,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSData } } else if (pFillInfo->type == TSDB_FILL_NEXT) { SArray* p = FILL_IS_ASC_FILL(pFillInfo) ? pFillInfo->next : pFillInfo->prev; - + // todo refactor: start from 0 not 1 for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) { SFillColInfo* pCol = &pFillInfo->pFillCol[i]; if (TSDB_COL_IS_TAG(pCol->flag)) { @@ -106,7 +109,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSData } else if (pFillInfo->type == TSDB_FILL_LINEAR) { // TODO : linear interpolation supports NULL value if (outOfBound) { - setNullRow(pBlock, pFillInfo->numOfCols, index); + setNullRow(pBlock, pFillInfo->currentKey, index); } else { for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) { SFillColInfo* pCol = &pFillInfo->pFillCol[i]; @@ -143,7 +146,7 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSData } } } else if (pFillInfo->type == TSDB_FILL_NULL) { // fill with NULL - setNullRow(pBlock, pFillInfo->numOfCols, index); + setNullRow(pBlock, pFillInfo->currentKey, index); } else { // fill with user specified value for each column for (int32_t i = 1; i < pFillInfo->numOfCols; ++i) { SFillColInfo* pCol = &pFillInfo->pFillCol[i]; @@ -166,6 +169,8 @@ static void doFillOneRowResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, SSData int64_t v = 0; GET_TYPED_DATA(v, int64_t, pVar->nType, &pVar->i); colDataAppend(pDst, index, (char*)&v, false); + } else if (pDst->info.type == TSDB_DATA_TYPE_TIMESTAMP) { + colDataAppend(pDst, index, (const char*)&pFillInfo->currentKey, false); } } } @@ -247,7 +252,7 @@ static int32_t fillResultImpl(SFillInfo* pFillInfo, SSDataBlock* pBlock, int32_t // fill the gap between two input rows while (((pFillInfo->currentKey < ts && ascFill) || (pFillInfo->currentKey > ts && !ascFill)) && pFillInfo->numOfCurrent < outputRows) { - doFillOneRowResult(pFillInfo, pBlock, pFillInfo->pSrcBlock, ts, false); + doFillOneRow(pFillInfo, pBlock, pFillInfo->pSrcBlock, ts, false); } // output buffer is full, abort @@ -343,7 +348,7 @@ static int64_t appendFilledResult(SFillInfo* pFillInfo, SSDataBlock* pBlock, int */ pFillInfo->numOfCurrent = 0; while (pFillInfo->numOfCurrent < resultCapacity) { - doFillOneRowResult(pFillInfo, pBlock, pFillInfo->pSrcBlock, pFillInfo->start, true); + doFillOneRow(pFillInfo, pBlock, pFillInfo->pSrcBlock, pFillInfo->start, true); } pFillInfo->numOfTotal += pFillInfo->numOfCurrent; @@ -408,7 +413,7 @@ static int32_t taosNumOfRemainRows(SFillInfo* pFillInfo) { } struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols, - SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, + SInterval* pInterval, int32_t fillType, struct SFillColInfo* pCol, int32_t primaryTsSlotId, const char* id) { if (fillType == TSDB_FILL_NONE) { return NULL; @@ -420,6 +425,8 @@ struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTag return NULL; } + pFillInfo->tsSlotId = primaryTsSlotId; + taosResetFillInfo(pFillInfo, skey); pFillInfo->order = order; @@ -589,11 +596,10 @@ int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, SSDataBlock* p, int32_t ca assert(numOfRes == pFillInfo->numOfCurrent); } - // qDebug("fill:%p, generated fill result, src block:%d, index:%d, brange:%"PRId64"-%"PRId64", currentKey:%"PRId64", - // current:%d, total:%d, %p", - // pFillInfo, pFillInfo->numOfRows, pFillInfo->index, pFillInfo->start, pFillInfo->end, pFillInfo->currentKey, - // pFillInfo->numOfCurrent, - // pFillInfo->numOfTotal, pFillInfo->handle); + qDebug("fill:%p, generated fill result, src block:%d, index:%d, brange:%" PRId64 "-%" PRId64 ", currentKey:%" PRId64 + ", current : % d, total : % d, %s", pFillInfo, + pFillInfo->numOfRows, pFillInfo->index, pFillInfo->start, pFillInfo->end, pFillInfo->currentKey, + pFillInfo->numOfCurrent, pFillInfo->numOfTotal, pFillInfo->id); return numOfRes; } diff --git a/source/libs/function/src/builtinsimpl.c b/source/libs/function/src/builtinsimpl.c index 359352951f00d2df930f7d10c19b9d5a2f14504e..9aad34d609f0df420ee93b66b893ed37dfaa89be 100644 --- a/source/libs/function/src/builtinsimpl.c +++ b/source/libs/function/src/builtinsimpl.c @@ -907,11 +907,13 @@ int32_t avgFunctionMerge(SqlFunctionCtx* pCtx) { SAvgRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); - int32_t start = pInput->startRowIndex; - char* data = colDataGetData(pCol, start); - SAvgRes* pInputInfo = (SAvgRes*)varDataVal(data); + int32_t start = pInput->startRowIndex; - avgTransferInfo(pInputInfo, pInfo); + for(int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SAvgRes* pInputInfo = (SAvgRes*)varDataVal(data); + avgTransferInfo(pInputInfo, pInfo); + } SET_VAL(GET_RES_INFO(pCtx), 1, 1); @@ -2512,11 +2514,13 @@ int32_t apercentileFunctionMerge(SqlFunctionCtx* pCtx) { SAPercentileInfo* pInfo = GET_ROWCELL_INTERBUF(pResInfo); - int32_t start = pInput->startRowIndex; - char* data = colDataGetData(pCol, start); - SAPercentileInfo* pInputInfo = (SAPercentileInfo*)varDataVal(data); + int32_t start = pInput->startRowIndex; - apercentileTransferInfo(pInputInfo, pInfo); + for(int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SAPercentileInfo* pInputInfo = (SAPercentileInfo*)varDataVal(data); + apercentileTransferInfo(pInputInfo, pInfo); + } SET_VAL(pResInfo, 1, 1); return TSDB_CODE_SUCCESS; @@ -2877,13 +2881,17 @@ static int32_t firstLastFunctionMergeImpl(SqlFunctionCtx* pCtx, bool isFirstQuer SFirstLastRes* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); - int32_t start = pInput->startRowIndex; - char* data = colDataGetData(pCol, start); - SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data); - - firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery); + int32_t start = pInput->startRowIndex; + int32_t numOfElems = 0; - int32_t numOfElems = pInputInfo->hasResult ? 1 : 0; + for(int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SFirstLastRes* pInputInfo = (SFirstLastRes*)varDataVal(data); + firstLastTransferInfo(pCtx, pInputInfo, pInfo, isFirstQuery); + if (!numOfElems) { + numOfElems = pInputInfo->hasResult ? 1 : 0; + } + } SET_VAL(GET_RES_INFO(pCtx), numOfElems, 1); @@ -3703,11 +3711,13 @@ int32_t spreadFunctionMerge(SqlFunctionCtx* pCtx) { SSpreadInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); - int32_t start = pInput->startRowIndex; - char* data = colDataGetData(pCol, start); - SSpreadInfo* pInputInfo = (SSpreadInfo*)varDataVal(data); + int32_t start = pInput->startRowIndex; - spreadTransferInfo(pInputInfo, pInfo); + for(int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SSpreadInfo* pInputInfo = (SSpreadInfo*)varDataVal(data); + spreadTransferInfo(pInputInfo, pInfo); + } SET_VAL(GET_RES_INFO(pCtx), 1, 1); @@ -3873,11 +3883,13 @@ int32_t elapsedFunctionMerge(SqlFunctionCtx* pCtx) { SElapsedInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); - int32_t start = pInput->startRowIndex; - char* data = colDataGetData(pCol, start); - SElapsedInfo* pInputInfo = (SElapsedInfo*)varDataVal(data); + int32_t start = pInput->startRowIndex; - elapsedTransferInfo(pInputInfo, pInfo); + for(int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SElapsedInfo* pInputInfo = (SElapsedInfo*)varDataVal(data); + elapsedTransferInfo(pInputInfo, pInfo); + } SET_VAL(GET_RES_INFO(pCtx), 1, 1); return TSDB_CODE_SUCCESS; @@ -4164,10 +4176,10 @@ int32_t histogramFunctionMerge(SqlFunctionCtx* pCtx) { SHistoFuncInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); - int32_t start = pInput->startRowIndex; + int32_t start = pInput->startRowIndex; for(int32_t i = start; i < start + pInput->numOfRows; ++i) { - char* data = colDataGetData(pCol, i); + char* data = colDataGetData(pCol, i); SHistoFuncInfo* pInputInfo = (SHistoFuncInfo*)varDataVal(data); histogramTransferInfo(pInputInfo, pInfo); } @@ -4385,11 +4397,13 @@ int32_t hllFunctionMerge(SqlFunctionCtx* pCtx) { SHLLInfo* pInfo = GET_ROWCELL_INTERBUF(GET_RES_INFO(pCtx)); - int32_t start = pInput->startRowIndex; - char* data = colDataGetData(pCol, start); - SHLLInfo* pInputInfo = (SHLLInfo*)varDataVal(data); + int32_t start = pInput->startRowIndex; - hllTransferInfo(pInputInfo, pInfo); + for(int32_t i = start; i < start + pInput->numOfRows; ++i) { + char* data = colDataGetData(pCol, i); + SHLLInfo* pInputInfo = (SHLLInfo*)varDataVal(data); + hllTransferInfo(pInputInfo, pInfo); + } SET_VAL(GET_RES_INFO(pCtx), 1, 1); return TSDB_CODE_SUCCESS;