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

fix: fix no data in db issue

This commit is contained in:
dapan1121 2023-06-13 11:44:13 +08:00
parent 7adea35706
commit 48b9cac75c
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6932,20 +6932,26 @@ int32_t translatePostCreateStream(SParseContext* pParseCxt, SQuery* pQuery, void
SCreateStreamStmt* pStmt = (SCreateStreamStmt*)pQuery->pRoot; SCreateStreamStmt* pStmt = (SCreateStreamStmt*)pQuery->pRoot;
STranslateContext cxt = {0}; STranslateContext cxt = {0};
SInterval interval = {0}; SInterval interval = {0};
int64_t lastTs = 0;
if (NULL == pResRow || NULL == pResRow[0]) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
int32_t code = initTranslateContext(pParseCxt, NULL, &cxt); int32_t code = initTranslateContext(pParseCxt, NULL, &cxt);
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
code = buildIntervalForCreateStream(pStmt, &interval); code = buildIntervalForCreateStream(pStmt, &interval);
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
if (interval.interval > 0) { if (pResRow && pResRow[0]) {
pStmt->pReq->lastTs = taosTimeTruncate(*(int64_t*)pResRow[0], &interval); lastTs = *(int64_t*)pResRow[0];
} else if (interval.interval > 0) {
lastTs = convertTimePrecision(taosGetTimestampMs(), TSDB_TIME_PRECISION_MILLI, interval.precision);
} else { } else {
pStmt->pReq->lastTs = *(int64_t*)pResRow[0]; lastTs = taosGetTimestampMs();
}
}
if (TSDB_CODE_SUCCESS == code) {
if (interval.interval > 0) {
pStmt->pReq->lastTs = taosTimeTruncate(lastTs, &interval);
} else {
pStmt->pReq->lastTs = lastTs;
} }
code = buildCmdMsg(&cxt, TDMT_MND_CREATE_STREAM, (FSerializeFunc)tSerializeSCMCreateStreamReq, pStmt->pReq); code = buildCmdMsg(&cxt, TDMT_MND_CREATE_STREAM, (FSerializeFunc)tSerializeSCMCreateStreamReq, pStmt->pReq);
} }

6
source/libs/parser/test/parInitialCTest.cpp
View File

@ -885,12 +885,12 @@ TEST_F(ParserInitialCTest, createStream) {
setCreateStreamReq( setCreateStreamReq(
"s1", "test", "s1", "test",
"create stream if not exists s1 trigger max_delay 20s watermark 10s ignore expired 0 fill_history 1 ignore " "create stream if not exists s1 trigger max_delay 20s watermark 10s ignore expired 0 fill_history 0 ignore "
"update 1 into st3 as select count(*) from t1 interval(10s)", "update 1 into st3 as select count(*) from t1 interval(10s)",
"st3", 1); "st3", 1);
setStreamOptions(STREAM_CREATE_STABLE_TRUE, STREAM_TRIGGER_MAX_DELAY, 20 * MILLISECOND_PER_SECOND, setStreamOptions(STREAM_CREATE_STABLE_TRUE, STREAM_TRIGGER_MAX_DELAY, 20 * MILLISECOND_PER_SECOND,
10 * MILLISECOND_PER_SECOND, 0, 1, 1); 10 * MILLISECOND_PER_SECOND, 0, 0, 1);
run("CREATE STREAM IF NOT EXISTS s1 TRIGGER MAX_DELAY 20s WATERMARK 10s IGNORE EXPIRED 0 FILL_HISTORY 1 IGNORE " run("CREATE STREAM IF NOT EXISTS s1 TRIGGER MAX_DELAY 20s WATERMARK 10s IGNORE EXPIRED 0 FILL_HISTORY 0 IGNORE "
"UPDATE 1 INTO st3 AS SELECT COUNT(*) FROM t1 INTERVAL(10S)"); "UPDATE 1 INTO st3 AS SELECT COUNT(*) FROM t1 INTERVAL(10S)");
clearCreateStreamReq(); clearCreateStreamReq();