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

refactor(stream): opt stream sink perf.

This commit is contained in:
Haojun Liao 2024-04-30 09:37:53 +08:00
parent 9ba8e6bd34
commit 54e9522c62
1 changed files with 6 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
source/dnode/vnode/src/tq/tqSink.c
View File

@ -333,9 +333,10 @@ int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, c
} }
while (j < newLen && k < oldLen) { while (j < newLen && k < oldLen) {
SRow* pNewRow = taosArrayGetP(pNew->aRowP, j); SRow* pNewRow = TARRAY_GET_ELEM(pNew->aRowP, j);
SRow* pOldRow = taosArrayGetP(pExisted->aRowP, k); SRow* pOldRow = TARRAY_GET_ELEM(pExisted->aRowP, k);
if (pNewRow->ts < pOldRow->ts) {
if (pNewRow->ts <= pOldRow->ts) {
taosArrayPush(pFinal, &pNewRow); taosArrayPush(pFinal, &pNewRow);
j += 1; j += 1;
} else if (pNewRow->ts > pOldRow->ts) { } else if (pNewRow->ts > pOldRow->ts) {
@ -373,12 +374,12 @@ int32_t doMergeExistedRows(SSubmitTbData* pExisted, const SSubmitTbData* pNew, c
} }
while (j < newLen) { while (j < newLen) {
SRow* pRow = taosArrayGetP(pNew->aRowP, j++); SRow* pRow = TARRAY_GET_ELEM(pNew->aRowP, j++);
taosArrayPush(pFinal, &pRow); taosArrayPush(pFinal, &pRow);
} }
while (k < oldLen) { while (k < oldLen) {
SRow* pRow = taosArrayGetP(pExisted->aRowP, k++); SRow* pRow = TARRAY_GET_ELEM(pExisted->aRowP, k++);
taosArrayPush(pFinal, &pRow); taosArrayPush(pFinal, &pRow);
} }