diff --git a/docs/en/12-taos-sql/12-distinguished.md b/docs/en/12-taos-sql/12-distinguished.md index bfc9ca32c0..8eecb706c0 100644 --- a/docs/en/12-taos-sql/12-distinguished.md +++ b/docs/en/12-taos-sql/12-distinguished.md @@ -102,6 +102,7 @@ The detailed beaviors of `NULL`, `NULL_F`, `VALUE`, and VALUE_F are described be 1. A huge volume of interpolation output may be returned using `FILL`, so it's recommended to specify the time range when using `FILL`. The maximum number of interpolation values that can be returned in a single query is 10,000,000. 2. The result set is in ascending order of timestamp when you aggregate by time window. 3. If aggregate by window is used on STable, the aggregate function is performed on all the rows matching the filter conditions. If `PARTITION BY` is not used in the query, the result set will be returned in strict ascending order of timestamp; otherwise the result set will be returned in the order of ascending timestamp in each group. +4. The output windows of Fill are related with time range of WHERE Clause. For asc fill, the first output window is the first window that conains the start time of WHERE clause. The last output window is the last window that contains the end time of WHERE clause. ::: diff --git a/docs/zh/12-taos-sql/12-distinguished.md b/docs/zh/12-taos-sql/12-distinguished.md index 0eaeb0dfa7..50bf36d2e1 100755 --- a/docs/zh/12-taos-sql/12-distinguished.md +++ b/docs/zh/12-taos-sql/12-distinguished.md @@ -97,6 +97,7 @@ NULL, NULL_F, VALUE, VALUE_F 这几种填充模式针对不同场景区别如下 1. 使用 FILL 语句的时候可能生成大量的填充输出，务必指定查询的时间区间。针对每次查询，系统可返回不超过 1 千万条具有插值的结果。 2. 在时间维度聚合中，返回的结果中时间序列严格单调递增。 3. 如果查询对象是超级表，则聚合函数会作用于该超级表下满足值过滤条件的所有表的数据。如果查询中没有使用 PARTITION BY 语句，则返回的结果按照时间序列严格单调递增；如果查询中使用了 PARTITION BY 语句分组，则返回结果中每个 PARTITION 内按照时间序列严格单调递增。 +4. Fill输出的起始和结束窗口与WHERE条件的时间范围有关, 如增序Fill时, 第一个窗口是包含WHERE条件开始时间的第一个窗口, 最后一个窗口是包含WHERE条件结束时间的最后一个窗口。 ::: diff --git a/source/libs/executor/src/filloperator.c b/source/libs/executor/src/filloperator.c index 4b71c5ee3f..c4ef74608a 100644 --- a/source/libs/executor/src/filloperator.c +++ b/source/libs/executor/src/filloperator.c @@ -59,6 +59,7 @@ static void revisedFillStartKey(SFillOperatorInfo* pInfo, SSDataBlock* pBlock, i static void destroyFillOperatorInfo(void* param); static void doApplyScalarCalculation(SOperatorInfo* pOperator, SSDataBlock* pBlock, int32_t order, int32_t scanFlag); static void fillResetPrevForNewGroup(SFillInfo* pFillInfo); +static void reviseFillStartAndEndKey(SFillOperatorInfo* pInfo, int32_t order); static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo* pOperator, SFillOperatorInfo* pInfo, SResultInfo* pResultInfo, int32_t order) { @@ -74,7 +75,7 @@ static void doHandleRemainBlockForNewGroupImpl(SOperatorInfo* pOperator, SFillOp blockDataCleanup(pInfo->pRes); doApplyScalarCalculation(pOperator, pInfo->existNewGroupBlock, order, scanFlag); - revisedFillStartKey(pInfo, pInfo->existNewGroupBlock, order); + reviseFillStartAndEndKey(pOperator->info, order); int64_t ts = (order == TSDB_ORDER_ASC) ? pInfo->existNewGroupBlock->info.window.ekey : pInfo->existNewGroupBlock->info.window.skey; @@ -258,7 +259,7 @@ static SSDataBlock* doFillImpl(SOperatorInfo* pOperator) { if (pInfo->curGroupId == 0 || (pInfo->curGroupId == pInfo->pRes->info.id.groupId)) { if (pInfo->curGroupId == 0 && taosFillNotStarted(pInfo->pFillInfo)) { - revisedFillStartKey(pInfo, pBlock, order); + reviseFillStartAndEndKey(pInfo, order); } pInfo->curGroupId = pInfo->pRes->info.id.groupId; // the first data block @@ -549,3 +550,31 @@ _error: taosMemoryFreeClear(pOperator); return code; } + +static void reviseFillStartAndEndKey(SFillOperatorInfo* pInfo, int32_t order) { + int64_t skey, ekey, next; + if (order == TSDB_ORDER_ASC) { + skey = taosTimeTruncate(pInfo->win.skey, &pInfo->pFillInfo->interval); + taosFillUpdateStartTimestampInfo(pInfo->pFillInfo, skey); + + ekey = taosTimeTruncate(pInfo->win.ekey, &pInfo->pFillInfo->interval); + next = ekey; + while (next < pInfo->win.ekey) { + next = taosTimeAdd(ekey, pInfo->pFillInfo->interval.sliding, pInfo->pFillInfo->interval.slidingUnit, + pInfo->pFillInfo->interval.precision); + ekey = next > pInfo->win.ekey ? ekey : next; + } + pInfo->win.ekey = ekey; + } else { + assert(order == TSDB_ORDER_DESC); + skey = taosTimeTruncate(pInfo->win.skey, &pInfo->pFillInfo->interval); + next = skey; + while (next < pInfo->win.skey) { + next = taosTimeAdd(skey, pInfo->pFillInfo->interval.sliding, pInfo->pFillInfo->interval.slidingUnit, + pInfo->pFillInfo->interval.precision); + skey = next > pInfo->win.skey ? skey : next; + } + taosFillUpdateStartTimestampInfo(pInfo->pFillInfo, skey); + pInfo->win.ekey = taosTimeTruncate(pInfo->win.ekey, &pInfo->pFillInfo->interval); + } +} diff --git a/tests/system-test/2-query/fill.py b/tests/system-test/2-query/fill.py index f5cd2d5855..64a43bd80a 100644 --- a/tests/system-test/2-query/fill.py +++ b/tests/system-test/2-query/fill.py @@ -1,6 +1,11 @@ +import queue +import random +from fabric2.runners import threading +from pandas._libs import interval import taos import sys +from util.common import TDCom from util.log import * from util.sql import * from util.cases import * @@ -8,6 +13,7 @@ from util.cases import * class TDTestCase: + updatecfgDict = {'asynclog': 0, 'ttlUnit': 1, 'ttlPushInterval': 5, 'ratioOfVnodeStreamThrea': 4, 'numOfVnodeQueryThreads': 80} def init(self, conn, logSql, replicaVar=1): self.replicaVar = int(replicaVar) @@ -15,7 +21,115 @@ class TDTestCase: #tdSql.init(conn.cursor()) tdSql.init(conn.cursor(), logSql) # output sql.txt file + def generate_fill_range(self, data_start: int, data_end: int, interval: int, step: int): + ret = [] + begin = data_start - 10 * interval + end = data_end + 10 * interval + for i in range(begin, end, step): + for j in range(begin, end, step): + ret.append((i,j)) + return ret + + def check_fill_range(self, where_start, where_end, res_asc, res_desc, sql: str, interval): + if len(res_asc) != len(res_desc): + tdLog.exit(f"err, asc desc with different rows, asc: {len(res_asc)}, desc: {len(res_desc)} sql: {sql}") + if len(res_asc) == 0: + tdLog.info(f'from {where_start} to {where_end} no rows returned') + return + asc_first = res_asc[0] + asc_last = res_asc[-1] + desc_first = res_desc[0] + desc_last = res_desc[-1] + if asc_first[0] != desc_last[0] or asc_last[0] != desc_first[0]: + tdLog.exit(f'fill sql different row data {sql}: asc<{asc_first[0].timestamp()}, {asc_last[0].timestamp()}>, desc<{desc_last[0].timestamp()}, {desc_first[0].timestamp()}>') + else: + tdLog.info(f'from {where_start} to {where_end} same time returned asc<{asc_first[0].timestamp()}, {asc_last[0].timestamp()}>, desc<{desc_last[0].timestamp()}, {desc_first[0].timestamp()}> interval: {interval}') + + def generate_partition_by(self): + val = random.random() + if val < 0.6: + return "" + elif val < 0.8: + return "partition by location" + else: + return "partition by tbname" + + def generate_fill_interval(self): + ret = [] + #intervals = [60, 90, 120, 300, 3600] + intervals = [120, 300, 3600] + for i in range(0, len(intervals)): + for j in range(0, i+1): + ret.append((intervals[i], intervals[j])) + return ret + + def generate_fill_sql(self, where_start, where_end, fill_interval: tuple): + partition_by = self.generate_partition_by() + where = f'where ts >= {where_start} and ts < {where_end}' + sql = f'select first(_wstart), last(_wstart) from (select _wstart, _wend, count(*) from test.meters {where} {partition_by} interval({fill_interval[0]}s) sliding({fill_interval[1]}s) fill(NULL)' + sql_asc = sql + " order by _wstart asc) t" + sql_desc = sql + " order by _wstart desc) t" + return sql_asc, sql_desc + + def fill_test_thread_routine(self, cli: TDSql, interval, data_start, data_end, step): + ranges = self.generate_fill_range(data_start, data_end, interval[0], step) + for range in ranges: + sql_asc, sql_desc = self.generate_fill_sql(range[0], range[1], interval) + cli.query(sql_asc, queryTimes=1) + asc_res = cli.queryResult + cli.query(sql_desc, queryTimes=1) + desc_res = cli.queryResult + self.check_fill_range(range[0], range[1], asc_res,desc_res , sql_asc, interval) + + def fill_test_task_routine(self, tdCom: TDCom, queue: queue.Queue): + cli = tdCom.newTdSql() + while True: + m: list = queue.get() + if len(m) == 0: + break + interval = m[0] + range = m[1] + sql_asc, sql_desc = self.generate_fill_sql(range[0], range[1], interval) + cli.query(sql_asc, queryTimes=1) + asc_res = cli.queryResult + cli.query(sql_desc, queryTimes=1) + desc_res = cli.queryResult + self.check_fill_range(range[0], range[1], asc_res,desc_res , sql_asc, interval) + cli.close() + + def schedule_fill_test_tasks(self): + num: int = 20 + threads = [] + tdCom = TDCom() + q: queue.Queue = queue.Queue() + for _ in range(num): + t = threading.Thread(target=self.fill_test_task_routine, args=(tdCom, q)) + t.start() + threads.append(t) + + data_start = 1500000000000 + data_end = 1500319968000 + step = 30000000 + + fill_intervals: list[tuple] = self.generate_fill_interval() + for interval in fill_intervals: + ranges = self.generate_fill_range(data_start, data_end, interval[0], step) + for r in ranges: + q.put([interval, r]) + + for _ in range(num): + q.put([]) + + for t in threads: + t.join() + + def test_fill_range(self): + os.system('taosBenchmark -t 10 -n 10000 -v 8 -S 32000 -y') + self.schedule_fill_test_tasks() + tdSql.execute('drop database test') + def run(self): + self.test_fill_range() dbname = "db" tbname = "tb" diff --git a/tests/system-test/2-query/test_td28163.py b/tests/system-test/2-query/test_td28163.py index a101549b66..005d78d075 100644 --- a/tests/system-test/2-query/test_td28163.py +++ b/tests/system-test/2-query/test_td28163.py @@ -176,7 +176,7 @@ class TDTestCase: def test_query_with_window(self): # time window tdSql.query("select sum(c_int_empty) from st where ts > '2024-01-01 00:00:00.000' and ts <= '2024-01-01 14:00:00.000' interval(5m) sliding(1m) fill(value, 10);") - tdSql.checkRows(841) + tdSql.checkRows(845) tdSql.checkData(0, 0, 10) tdSql.query("select _wstart, _wend, sum(c_int) from st where ts > '2024-01-01 00:00:00.000' and ts <= '2024-01-01 14:00:00.000' interval(5m) sliding(1m);")