Skip to content

Commit bbeffc5

Browse files
hjxilinxhjxilinx
authored
fix(gpt): dynamic allocate the option buffer and remove the limitation on the option serialization. (#31482)
1 parent 4376643 commit bbeffc5

File tree

3 files changed

+139
-24
lines changed

3 files changed

+139
-24
lines changed

‎source/common/src/tanalytics.c

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -441,12 +441,25 @@ static int32_t taosAnalyJsonBufWriteOptInt(SAnalyticBuf *pBuf, const char *optNa
441441
}
442442

443443
static int32_t taosAnalyJsonBufWriteOptStr(SAnalyticBuf *pBuf, const char *optName, const char *optVal) {
444-
char buf[128] = {0};
445-
int32_t bufLen = tsnprintf(buf, sizeof(buf), "\"%s\": \"%s\",\n", optName, optVal);
446-
if (taosWriteFile(pBuf->filePtr, buf, bufLen) != bufLen) {
444+
int32_t code = 0;
445+
int32_t keyLen = strlen(optName);
446+
int32_t valLen = strlen(optVal);
447+
448+
int32_t totalLen = keyLen + valLen + 20;
449+
char * buf = taosMemoryMalloc(totalLen);
450+
if (buf == NULL) {
451+
uError("failed to prepare the buffer for serializing the key/value info for analysis, len:%d, code:%s", totalLen,
452+
tstrerror(terrno));
447453
return terrno;
448454
}
449-
return 0;
455+
456+
int32_t bufLen = tsnprintf(buf, totalLen, "\"%s\": \"%s\",\n", optName, optVal);
457+
if (taosWriteFile(pBuf->filePtr, buf, bufLen) != bufLen) {
458+
code = terrno;
459+
}
460+
461+
taosMemoryFree(buf);
462+
return code;
450463
}
451464

452465
static int32_t taosAnalyJsonBufWriteOptFloat(SAnalyticBuf *pBuf, const char *optName, float optVal) {

‎source/libs/executor/src/forecastoperator.c

Lines changed: 74 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,8 @@
2121
#include "tanalytics.h"
2222
#include "taoserror.h"
2323
#include "tcommon.h"
24-
// #include "tcompare.h"
2524
#include "tdatablock.h"
26-
// #include "tfill.h"
2725
#include "tmsg.h"
28-
// #include "ttime.h"
2926

3027
#ifdef USE_ANALYTICS
3128

@@ -45,7 +42,7 @@ typedef struct {
4542
typedef struct {
4643
char algoName[TSDB_ANALYTIC_ALGO_NAME_LEN];
4744
char algoUrl[TSDB_ANALYTIC_ALGO_URL_LEN];
48-
char algoOpt[TSDB_ANALYTIC_ALGO_OPTION_LEN];
45+
char* pOptions;
4946
int64_t maxTs;
5047
int64_t minTs;
5148
int64_t numOfRows;
@@ -195,7 +192,7 @@ static int32_t forecastCloseBuf(SForecastSupp* pSupp, const char* id) {
195192
code = taosAnalyBufWriteDataEnd(pBuf);
196193
if (code != 0) return code;
197194

198-
code = taosAnalyBufWriteOptStr(pBuf, "option", pSupp->algoOpt);
195+
code = taosAnalyBufWriteOptStr(pBuf, "option", pSupp->pOptions);
199196
if (code != 0) return code;
200197

201198
code = taosAnalyBufWriteOptStr(pBuf, "algo", pSupp->algoName);
@@ -226,7 +223,7 @@ static int32_t forecastCloseBuf(SForecastSupp* pSupp, const char* id) {
226223
code = taosAnalyBufWriteOptFloat(pBuf, "conf", pSupp->conf);
227224
if (code != 0) return code;
228225

229-
int32_t len = strlen(pSupp->algoOpt);
226+
int32_t len = strlen(pSupp->pOptions);
230227
int64_t every = (pSupp->setEvery != 0) ? pSupp->every : ((pSupp->maxTs - pSupp->minTs) / (pSupp->numOfRows - 1));
231228
code = taosAnalyBufWriteOptInt(pBuf, "every", every);
232229
if (code != 0) return code;
@@ -538,6 +535,18 @@ static int32_t validInputParams(SFunctionNode* pFunc, const char* id) {
538535
return code;
539536
}
540537

538+
static bool existInList(SForecastSupp* pSupp, int32_t slotId) {
539+
for (int32_t j = 0; j < taosArrayGetSize(pSupp->pCovariateSlotList); ++j) {
540+
SColumn* pCol = taosArrayGet(pSupp->pCovariateSlotList, j);
541+
542+
if (pCol->slotId == slotId) {
543+
return true;
544+
}
545+
}
546+
547+
return false;
548+
}
549+
541550
static int32_t forecastParseInput(SForecastSupp* pSupp, SNodeList* pFuncs, const char* id) {
542551
int32_t code = 0;
543552
SNode* pNode = NULL;
@@ -571,7 +580,11 @@ static int32_t forecastParseInput(SForecastSupp* pSupp, SNodeList* pFuncs, const
571580
pSupp->targetValType = pTarget->node.resType.type;
572581

573582
// let's add the holtwinters as the default forecast algorithm
574-
tstrncpy(pSupp->algoOpt, "algo=holtwinters", TSDB_ANALYTIC_ALGO_OPTION_LEN);
583+
pSupp->pOptions = taosStrdup("algo=holtwinters");
584+
if (pSupp->pOptions == NULL) {
585+
qError("%s failed to dup forecast option, code:%s", id, tstrerror(terrno));
586+
return terrno;
587+
}
575588
} else {
576589
SColumnNode* pTarget = (SColumnNode*)nodesListGetNode(pFunc->pParameterList, 0);
577590
bool assignTs = false;
@@ -594,7 +607,7 @@ static int32_t forecastParseInput(SForecastSupp* pSupp, SNodeList* pFuncs, const
594607
} else if (nodeType(pNode) == QUERY_NODE_VALUE) {
595608
if (!assignOpt) {
596609
SValueNode* pOptNode = (SValueNode*)pNode;
597-
tstrncpy(pSupp->algoOpt, pOptNode->literal, sizeof(pSupp->algoOpt));
610+
pSupp->pOptions = taosStrdup(pOptNode->literal);
598611
assignOpt = true;
599612
continue;
600613
}
@@ -603,7 +616,11 @@ static int32_t forecastParseInput(SForecastSupp* pSupp, SNodeList* pFuncs, const
603616

604617
if (!assignOpt) {
605618
// set the default forecast option
606-
tstrncpy(pSupp->algoOpt, "algo=holtwinters", TSDB_ANALYTIC_ALGO_OPTION_LEN);
619+
pSupp->pOptions = taosStrdup("algo=holtwinters");
620+
if (pSupp->pOptions == NULL) {
621+
qError("%s failed to dup forecast option, code:%s", id, tstrerror(terrno));
622+
return terrno;
623+
}
607624
}
608625

609626
pSupp->pCovariateSlotList = taosArrayInit(4, sizeof(SColumn));
@@ -616,6 +633,15 @@ static int32_t forecastParseInput(SForecastSupp* pSupp, SNodeList* pFuncs, const
616633
break;
617634
}
618635

636+
if (p->slotId == pSupp->targetValSlot) {
637+
continue; // duplicate the target column, ignore it
638+
}
639+
640+
bool exist = existInList(pSupp, p->slotId);
641+
if (exist) {
642+
continue; // duplicate column, ignore it
643+
}
644+
619645
SColumn col = {.slotId = p->slotId,
620646
.colType = p->colType,
621647
.type = p->node.resType.type,
@@ -671,7 +697,7 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
671697

672698
initForecastOpt(pSupp);
673699

674-
code = taosAnalyGetOpts(pSupp->algoOpt, &pHashMap);
700+
code = taosAnalyGetOpts(pSupp->pOptions, &pHashMap);
675701
if (code != TSDB_CODE_SUCCESS) {
676702
return code;
677703
}
@@ -682,7 +708,13 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
682708
return code;
683709
}
684710

685-
code = taosAnalysisParseAlgo(pSupp->algoOpt, pSupp->algoName, pSupp->algoUrl, ANALY_ALGO_TYPE_FORECAST,
711+
if (taosHashGetSize(pHashMap) == 0) {
712+
code = TSDB_CODE_INVALID_PARA;
713+
qError("%s no valid options for forecast, failed to exec", id);
714+
return code;
715+
}
716+
717+
code = taosAnalysisParseAlgo(pSupp->pOptions, pSupp->algoName, pSupp->algoUrl, ANALY_ALGO_TYPE_FORECAST,
686718
tListLen(pSupp->algoUrl), pHashMap, id);
687719
TSDB_CHECK_CODE(code, lino, _end);
688720

@@ -705,7 +737,7 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
705737
pSupp->forecastRows = v;
706738
qDebug("%s forecast rows:%"PRId64, id, pSupp->forecastRows);
707739
} else {
708-
qDebug("%s forecast rows not found:%s, use default:%" PRId64, id, pSupp->algoOpt, pSupp->forecastRows);
740+
qDebug("%s forecast rows not found:%s, use default:%" PRId64, id, pSupp->pOptions, pSupp->forecastRows);
709741
}
710742

711743
if (pSupp->forecastRows > ANALY_FORECAST_RES_MAX_ROWS) {
@@ -730,7 +762,7 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
730762
qDebug("%s forecast conf:%.2f", id, pSupp->conf);
731763
}
732764
} else {
733-
qDebug("%s forecast conf not found:%s, use default:%.2f", id, pSupp->algoOpt, pSupp->conf);
765+
qDebug("%s forecast conf not found:%s, use default:%.2f", id, pSupp->pOptions, pSupp->conf);
734766
}
735767

736768
// extract the start timestamp
@@ -780,8 +812,11 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
780812

781813
void* pCol = taosHashGet(pHashMap, nameBuf, strlen(nameBuf));
782814
if (pCol == NULL) {
783-
qError("%s dynamic real column related:%s column name:%s not specified", id, pKey, nameBuf);
784-
code = TSDB_CODE_ANA_INTERNAL_ERROR;
815+
char* pTmp = taosStrndupi(pKey, keyLen);
816+
qError("%s dynamic real column related:%s column name:%s not specified", id, pTmp, nameBuf);
817+
818+
taosMemoryFree(pTmp);
819+
code = TSDB_CODE_INVALID_PARA;
785820
goto _end;
786821
} else {
787822
// build dynamic_real_feature
@@ -803,7 +838,8 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
803838

804839
if (index == -1) {
805840
qError("%s not found the required future dynamic real column:%s", id, d.pName);
806-
code = TSDB_CODE_ANA_INTERNAL_ERROR;
841+
code = TSDB_CODE_INVALID_PARA;
842+
taosMemoryFree(d.pName);
807843
goto _end;
808844
}
809845

@@ -812,15 +848,20 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
812848
d.data.info.type = pColx->type;
813849
d.data.info.bytes = pColx->bytes;
814850

815-
char* buf = taosStrndup(pVal, taosHashGetValueSize((void*)pVal));
851+
int32_t len = taosHashGetValueSize((void*)pVal);
852+
char* buf = taosStrndupi(pVal, len);
816853
int32_t unused = strdequote((char*)buf);
817854

818855
int32_t num = 0;
819856
char** pList = strsplit(buf, " ", &num);
820857
if (num != pSupp->forecastRows) {
821858
qError("%s the rows:%d of future dynamic real column data is not equalled to the forecasting rows:%" PRId64,
822859
id, num, pSupp->forecastRows);
823-
code = TSDB_CODE_ANA_INTERNAL_ERROR;
860+
code = TSDB_CODE_INVALID_PARA;
861+
862+
taosMemoryFree(d.pName);
863+
taosMemoryFree(pList);
864+
taosMemoryFree(buf);
824865
goto _end;
825866
}
826867

@@ -890,6 +931,9 @@ static int32_t forecastParseOpt(SForecastSupp* pSupp, const char* id) {
890931

891932
}
892933

934+
taosMemoryFree(pList);
935+
taosMemoryFree(buf);
936+
893937
void* noret = taosArrayPush(pSupp->pDynamicRealList, &d);
894938
if (noret == NULL) {
895939
qError("%s failed to add column info in dynamic real column info", id);
@@ -1047,7 +1091,7 @@ int32_t createForecastOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNo
10471091

10481092
*pOptrInfo = pOperator;
10491093

1050-
qDebug("%s forecast env is initialized, option:%s", pId, pSupp->algoOpt);
1094+
qDebug("%s forecast env is initialized, option:%s", pId, pSupp->pOptions);
10511095
return TSDB_CODE_SUCCESS;
10521096

10531097
_error:
@@ -1060,6 +1104,12 @@ int32_t createForecastOperatorInfo(SOperatorInfo* downstream, SPhysiNode* pPhyNo
10601104
return code;
10611105
}
10621106

1107+
static void destroyColFutureData(void* p) {
1108+
SColFutureData* pData = p;
1109+
taosMemoryFree(pData->pName);
1110+
colDataDestroy(&pData->data);
1111+
}
1112+
10631113
static void destroyForecastInfo(void* param) {
10641114
SForecastOperatorInfo* pInfo = (SForecastOperatorInfo*)param;
10651115

@@ -1069,6 +1119,11 @@ static void destroyForecastInfo(void* param) {
10691119
taosArrayDestroy(pInfo->forecastSupp.pCovariateSlotList);
10701120
pInfo->forecastSupp.pCovariateSlotList = NULL;
10711121

1122+
taosArrayDestroyEx(pInfo->forecastSupp.pDynamicRealList, destroyColFutureData);
1123+
pInfo->forecastSupp.pDynamicRealList = NULL;
1124+
1125+
taosMemoryFree(pInfo->forecastSupp.pOptions);
1126+
10721127
cleanupExprSupp(&pInfo->scalarSup);
10731128
taosAnalyBufDestroy(&pInfo->forecastSupp.analyBuf);
10741129
taosMemoryFreeClear(param);

‎tests/script/tsim/analytics/basic0.sim

Lines changed: 48 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -192,7 +192,54 @@ sql select forecast(c2) from pk_stb_ct1;
192192
sql select forecast(c2, 'algo=arima') from pk_stb_ct1;
193193
sql_error select forecast(c2, c3) from pk_stb_ct1;
194194
sql_error select forecast(c2, c3, 'algo=arima') from pk_stb_ct1;
195-
#sql select forecast(c2, c3, 'algo=moirai') from pk_stb_ct1;
195+
196+
print ==================== co-variate query test and future co-variate query test
197+
sql select forecast(c2, c3, c1, 'algo=moirai') from pk_stb_ct1;
198+
199+
sql select forecast(c2, c2, c3, 'algo=moirai') from pk_stb_ct1;
200+
sql select forecast(c2, c2, c2, c2, 'algo=moirai') from pk_stb_ct1;
201+
sql select forecast(c2, c2, c2, c2, 'algo=holtwinters') from pk_stb_ct1;
202+
203+
# not exist column
204+
sql_error select forecast(c2, c3, c1, c4, 'algo=moirai') from pk_stb_ct1;
205+
206+
# not support algorithm
207+
sql_error select forecast(c2, c3, c1, 'algo=holtwinters') from pk_stb_ct1;
208+
209+
#rows not match
210+
sql_error select forecast(c2, c3, c1, 'algo=moirai,dynamic_real_1=[1 1 1 1], rows=22') from pk_stb_ct1;
211+
212+
# missing columns in future dynamic real parameter
213+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[1 1 1],rows=3,dynamic_real_100_col=c4') from pk_stb_ct1;
214+
215+
# name mismatch
216+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[1 1 1],rows=3,dynamic_real_100_col=c2') from pk_stb_ct1;
217+
218+
# name mismatch
219+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[1 1 1],rows=3,dynamic_real_1_col=c1') from pk_stb_ct1;
220+
221+
# invalid input - 1
222+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=(1 1 1),rows=3,dynamic_real_1_col=c1') from pk_stb_ct1;
223+
224+
# invalid input - 2
225+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[1, 1, 1],rows=3,dynamic_real_1_col=c1') from pk_stb_ct1;
226+
227+
# invalid input - 3
228+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=["abc"],rows=1,dynamic_real_1_col=c1') from pk_stb_ct1;
229+
sql_error select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[1 110 31.92],rows=abc,dynamic_real_x_col=c1') from pk_stb_ct1;
230+
231+
print ============== future dynamic real column test
232+
sql select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[1 1 1],rows=3,dynamic_real_100_col=c3') from pk_stb_ct1;
233+
if $rows != 3 then
234+
return -1
235+
endi
236+
237+
print ============== too long parameter test
238+
sql select forecast(c2, c3, c1, 'algo=moirai, dynamic_real_100=[10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000 10000],rows=100,dynamic_real_100_col=c3') from pk_stb_ct1;
239+
240+
if $rows != 100 then
241+
return -1
242+
endi
196243

197244
sql drop anode 1
198245
sql show anodes

0 commit comments

Comments
 (0)