Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
taosdata
TDengine
提交
7d8a2822
T
TDengine
项目概览
taosdata
/
TDengine
大约 1 年 前同步成功
通知
1181
Star
22014
Fork
4786
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
1
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
T
TDengine
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
1
Issue
1
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
7d8a2822
编写于
10月 27, 2022
作者:
5
54liuyao
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
fix(stream):tqBuildDeleteReq error
上级
f2538701
变更
6
显示空白变更内容
内联
并排
Showing
6 changed file
with
105 addition
and
9 deletion
+105
-9
include/libs/stream/streamState.h
include/libs/stream/streamState.h
+2
-0
source/dnode/vnode/src/tq/tqSink.c
source/dnode/vnode/src/tq/tqSink.c
+1
-1
source/libs/executor/src/scanoperator.c
source/libs/executor/src/scanoperator.c
+3
-0
source/libs/function/src/builtinsimpl.c
source/libs/function/src/builtinsimpl.c
+4
-3
source/libs/stream/src/streamState.c
source/libs/stream/src/streamState.c
+41
-1
tests/script/tsim/stream/partitionbyColumnInterval.sim
tests/script/tsim/stream/partitionbyColumnInterval.sim
+54
-4
未找到文件。
include/libs/stream/streamState.h
浏览文件 @
7d8a2822
...
...
@@ -99,6 +99,8 @@ int32_t streamStateSeekLast(SStreamState* pState, SStreamStateCur* pCur);
int32_t
streamStateCurNext
(
SStreamState
*
pState
,
SStreamStateCur
*
pCur
);
int32_t
streamStateCurPrev
(
SStreamState
*
pState
,
SStreamStateCur
*
pCur
);
// char* streamStateSessionDump(SStreamState* pState);
#ifdef __cplusplus
}
#endif
...
...
source/dnode/vnode/src/tq/tqSink.c
浏览文件 @
7d8a2822
...
...
@@ -42,7 +42,7 @@ int32_t tqBuildDeleteReq(SVnode* pVnode, const char* stbFullName, const SSDataBl
if
(
metaGetTableEntryByName
(
&
mr
,
name
)
<
0
)
{
metaReaderClear
(
&
mr
);
taosMemoryFree
(
name
);
return
-
1
;
continue
;
}
int64_t
uid
=
mr
.
me
.
uid
;
...
...
source/libs/executor/src/scanoperator.c
浏览文件 @
7d8a2822
...
...
@@ -1299,6 +1299,9 @@ static int32_t generateSessionScanRange(SStreamScanInfo* pInfo, SSDataBlock* pSr
SSessionKey
startWin
=
{
0
};
getCurSessionWindow
(
pInfo
->
windowSup
.
pStreamAggSup
,
startData
[
i
],
endData
[
i
],
groupId
,
&
startWin
);
if
(
IS_INVALID_SESSION_WIN_KEY
(
startWin
))
{
// char* tmp = streamStateSessionDump(pInfo->windowSup.pStreamAggSup->pState);
// qInfo("%s", tmp);
// taosMemoryFree(tmp);
// window has been closed.
continue
;
}
...
...
source/libs/function/src/builtinsimpl.c
浏览文件 @
7d8a2822
...
...
@@ -380,7 +380,7 @@ typedef struct SGroupKeyInfo {
numOfElem += 1; \
pStddevRes->count -= 1; \
sumT -= plist[i]; \
pStddevRes->quadraticISum -=
plist[i] * plist[i];
\
pStddevRes->quadraticISum -=
(int64_t)(plist[i] * plist[i]);
\
} \
} while (0)
...
...
@@ -2526,8 +2526,9 @@ int32_t apercentileFunction(SqlFunctionCtx* pCtx) {
// might be a race condition here that pHisto can be overwritten or setup function
// has not been called, need to relink the buffer pHisto points to.
buildHistogramInfo
(
pInfo
);
qDebug
(
"%s before add %d elements into histogram, total:%"
PRId64
", numOfEntry:%d, pHisto:%p, elems: %p"
,
__FUNCTION__
,
numOfElems
,
pInfo
->
pHisto
->
numOfElems
,
pInfo
->
pHisto
->
numOfEntries
,
pInfo
->
pHisto
,
pInfo
->
pHisto
->
elems
);
qDebug
(
"%s before add %d elements into histogram, total:%"
PRId64
", numOfEntry:%d, pHisto:%p, elems: %p"
,
__FUNCTION__
,
numOfElems
,
pInfo
->
pHisto
->
numOfElems
,
pInfo
->
pHisto
->
numOfEntries
,
pInfo
->
pHisto
,
pInfo
->
pHisto
->
elems
);
for
(
int32_t
i
=
start
;
i
<
pInput
->
numOfRows
+
start
;
++
i
)
{
if
(
colDataIsNull_f
(
pCol
->
nullbitmap
,
i
))
{
continue
;
...
...
source/libs/stream/src/streamState.c
浏览文件 @
7d8a2822
...
...
@@ -497,7 +497,7 @@ SStreamStateCur* streamStateSessionGetRanomCur(SStreamState* pState, const SSess
if
(
pCur
==
NULL
)
return
NULL
;
tdbTbcOpen
(
pState
->
pSessionStateDb
,
&
pCur
->
pCur
,
NULL
);
int32_t
c
=
0
;
int32_t
c
=
-
2
;
SStateSessionKey
sKey
=
{.
key
=
*
key
,
.
opNum
=
pState
->
number
};
tdbTbcMoveTo
(
pCur
->
pCur
,
&
sKey
,
sizeof
(
SStateSessionKey
),
&
c
);
if
(
c
!=
0
)
{
...
...
@@ -727,3 +727,43 @@ _end:
streamStateFreeCur
(
pCur
);
return
res
;
}
#if 0
char* streamStateSessionDump(SStreamState* pState) {
SStreamStateCur* pCur = taosMemoryCalloc(1, sizeof(SStreamStateCur));
if (pCur == NULL) {
return NULL;
}
pCur->number = pState->number;
if (tdbTbcOpen(pState->pSessionStateDb, &pCur->pCur, NULL) < 0) {
streamStateFreeCur(pCur);
return NULL;
}
tdbTbcMoveToFirst(pCur->pCur);
SSessionKey key = {0};
int32_t code = streamStateSessionGetKVByCur(pCur, &key, NULL, 0);
if (code != 0) {
return NULL;
}
int32_t size = 2048;
char* dumpBuf = taosMemoryCalloc(size, 1);
int64_t len = 0;
len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.win.skey);
len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
while (1) {
tdbTbcMoveToNext(pCur->pCur);
key = (SSessionKey){0};
code = streamStateSessionGetKVByCur(pCur, &key, NULL, 0);
if (code != 0) {
return dumpBuf;
}
len += snprintf(dumpBuf + len, size - len, "||s:%15" PRId64 ",", key.win.skey);
len += snprintf(dumpBuf + len, size - len, "e:%15" PRId64 ",", key.win.ekey);
len += snprintf(dumpBuf + len, size - len, "g:%15" PRId64 "||", key.groupId);
}
return dumpBuf;
}
#endif
tests/script/tsim/stream/partitionbyColumnInterval.sim
浏览文件 @
7d8a2822
...
...
@@ -599,12 +599,62 @@ if $loop_count == 20 then
endi
if $rows != 4 then
print =====rows=$rows
print =====rows=$rows
print =====rows=$rows
#goto loop15
print ======rows=$rows
goto loop15
endi
sql insert into t1 values(1648791223001,11,2,3,1.0);
sql insert into t2 values(1648791223001,21,2,3,1.0);
sql insert into t3 values(1648791223001,31,2,3,1.0);
sql insert into t4 values(1648791223001,41,2,3,1.0);
sleep 300
sql delete from st where ts = 1648791223001;
$loop_count = 0
loop16:
sleep 50
sql select * from test.streamt5 order by c1, c2, c3;
$loop_count = $loop_count + 1
if $loop_count == 20 then
print ======rows=$rows
return -1
endi
if $rows != 4 then
goto loop16
endi
sql insert into t1 values(1648791223001,12,2,3,1.0);
sql insert into t2 values(1648791223001,22,2,3,1.0);
sql insert into t3 values(1648791223001,32,2,3,1.0);
sql insert into t4 values(1648791223001,42,2,3,1.0);
sleep 300
sql delete from st where ts = 1648791223001;
$loop_count = 0
loop17:
sleep 50
sql select * from test.streamt5 order by c1, c2, c3;
$loop_count = $loop_count + 1
if $loop_count == 20 then
return -1
endi
if $rows != 4 then
print ======rows=$rows
print ======rows=$rows
print ======rows=$rows
return 1
#goto loop17
endi
$loop_all = $loop_all + 1
print ============loop_all=$loop_all
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录