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

more code format

This commit is contained in:
Hongze Cheng 2022-10-13 14:15:42 +08:00
parent c79cc7e454
commit c4b3da5032
17 changed files with 564 additions and 613 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

162
source/libs/function/inc/builtinsimpl.h
View File

@ -23,37 +23,37 @@ extern "C" {
#include "function.h"
#include "functionMgt.h"
bool dummyGetEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* UNUSED_PARAM(pEnv));
bool dummyInit(SqlFunctionCtx* UNUSED_PARAM(pCtx), SResultRowEntryInfo* UNUSED_PARAM(pResultInfo));
bool dummyGetEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* UNUSED_PARAM(pEnv));
bool dummyInit(SqlFunctionCtx* UNUSED_PARAM(pCtx), SResultRowEntryInfo* UNUSED_PARAM(pResultInfo));
int32_t dummyProcess(SqlFunctionCtx* UNUSED_PARAM(pCtx));
int32_t dummyFinalize(SqlFunctionCtx* UNUSED_PARAM(pCtx), SSDataBlock* UNUSED_PARAM(pBlock));
bool functionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool functionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t functionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t functionFinalizeWithResultBuf(SqlFunctionCtx* pCtx, SSDataBlock* pBlock, char* finalResult);
int32_t combineFunction(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
EFuncDataRequired countDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t countFunction(SqlFunctionCtx *pCtx);
int32_t countInvertFunction(SqlFunctionCtx *pCtx);
bool getCountFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t countFunction(SqlFunctionCtx* pCtx);
int32_t countInvertFunction(SqlFunctionCtx* pCtx);
EFuncDataRequired statisDataRequired(SFunctionNode* pFunc, STimeWindow* pTimeWindow);
bool getSumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t sumFunction(SqlFunctionCtx *pCtx);
int32_t sumInvertFunction(SqlFunctionCtx *pCtx);
int32_t sumCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getSumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t sumFunction(SqlFunctionCtx* pCtx);
int32_t sumInvertFunction(SqlFunctionCtx* pCtx);
int32_t sumCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool minmaxFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getMinmaxFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool minmaxFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
bool getMinmaxFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t minFunction(SqlFunctionCtx* pCtx);
int32_t maxFunction(SqlFunctionCtx *pCtx);
int32_t maxFunction(SqlFunctionCtx* pCtx);
int32_t minmaxFunctionFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t minCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t maxCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getAvgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool avgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getAvgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool avgFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t avgFunction(SqlFunctionCtx* pCtx);
int32_t avgFunctionMerge(SqlFunctionCtx* pCtx);
int32_t avgFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
@ -62,8 +62,8 @@ int32_t avgInvertFunction(SqlFunctionCtx* pCtx);
int32_t avgCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getAvgInfoSize();
bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stddevFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getStddevFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stddevFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t stddevFunction(SqlFunctionCtx* pCtx);
int32_t stddevFunctionMerge(SqlFunctionCtx* pCtx);
int32_t stddevFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
@ -72,63 +72,63 @@ int32_t stddevInvertFunction(SqlFunctionCtx* pCtx);
int32_t stddevCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getStddevInfoSize();
bool getLeastSQRFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool leastSQRFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getLeastSQRFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool leastSQRFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t leastSQRFunction(SqlFunctionCtx* pCtx);
int32_t leastSQRFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t leastSQRInvertFunction(SqlFunctionCtx* pCtx);
int32_t leastSQRCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getPercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool percentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t percentileFunction(SqlFunctionCtx *pCtx);
bool getPercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool percentileFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t percentileFunction(SqlFunctionCtx* pCtx);
int32_t percentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getApercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool apercentileFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t apercentileFunction(SqlFunctionCtx *pCtx);
bool getApercentileFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool apercentileFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t apercentileFunction(SqlFunctionCtx* pCtx);
int32_t apercentileFunctionMerge(SqlFunctionCtx* pCtx);
int32_t apercentileFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t apercentilePartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t apercentileCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getApercentileMaxSize();
bool getDiffFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool diffFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
int32_t diffFunction(SqlFunctionCtx *pCtx);
bool getDiffFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool diffFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo);
int32_t diffFunction(SqlFunctionCtx* pCtx);
bool getDerivativeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool derivativeFuncSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
int32_t derivativeFunction(SqlFunctionCtx *pCtx);
bool getDerivativeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool derivativeFuncSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo);
int32_t derivativeFunction(SqlFunctionCtx* pCtx);
bool getIrateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool irateFuncSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResInfo);
int32_t irateFunction(SqlFunctionCtx *pCtx);
bool getIrateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool irateFuncSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResInfo);
int32_t irateFunction(SqlFunctionCtx* pCtx);
int32_t irateFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t cachedLastRowFunction(SqlFunctionCtx* pCtx);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t firstFunction(SqlFunctionCtx *pCtx);
int32_t firstFunctionMerge(SqlFunctionCtx *pCtx);
int32_t lastFunction(SqlFunctionCtx *pCtx);
int32_t lastFunctionMerge(SqlFunctionCtx *pCtx);
int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getFirstLastInfoSize(int32_t resBytes);
bool getFirstLastFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t firstFunction(SqlFunctionCtx* pCtx);
int32_t firstFunctionMerge(SqlFunctionCtx* pCtx);
int32_t lastFunction(SqlFunctionCtx* pCtx);
int32_t lastFunctionMerge(SqlFunctionCtx* pCtx);
int32_t firstLastFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t firstLastPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t firstCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t lastCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getFirstLastInfoSize(int32_t resBytes);
EFuncDataRequired lastDynDataReq(void* pRes, STimeWindow* pTimeWindow);
int32_t lastRowFunction(SqlFunctionCtx *pCtx);
int32_t lastRowFunction(SqlFunctionCtx* pCtx);
bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
bool getTopBotMergeFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
bool topBotFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t topFunction(SqlFunctionCtx *pCtx);
int32_t topFunctionMerge(SqlFunctionCtx *pCtx);
int32_t bottomFunction(SqlFunctionCtx *pCtx);
int32_t bottomFunctionMerge(SqlFunctionCtx *pCtx);
bool getTopBotFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
bool getTopBotMergeFuncEnv(SFunctionNode* UNUSED_PARAM(pFunc), SFuncExecEnv* pEnv);
bool topBotFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t topFunction(SqlFunctionCtx* pCtx);
int32_t topFunctionMerge(SqlFunctionCtx* pCtx);
int32_t bottomFunction(SqlFunctionCtx* pCtx);
int32_t bottomFunctionMerge(SqlFunctionCtx* pCtx);
int32_t topBotFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t topBotPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t topBotMergeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
@ -136,8 +136,8 @@ int32_t topCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t bottomCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
int32_t getTopBotInfoSize(int64_t numOfItems);
bool getSpreadFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool spreadFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getSpreadFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool spreadFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t spreadFunction(SqlFunctionCtx* pCtx);
int32_t spreadFunctionMerge(SqlFunctionCtx* pCtx);
int32_t spreadFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
@ -145,8 +145,8 @@ int32_t spreadPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t getSpreadInfoSize();
int32_t spreadCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getElapsedFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool elapsedFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getElapsedFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool elapsedFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t elapsedFunction(SqlFunctionCtx* pCtx);
int32_t elapsedFunctionMerge(SqlFunctionCtx* pCtx);
int32_t elapsedFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
@ -154,8 +154,8 @@ int32_t elapsedPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t getElapsedInfoSize();
int32_t elapsedCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getHistogramFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool histogramFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getHistogramFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool histogramFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t histogramFunction(SqlFunctionCtx* pCtx);
int32_t histogramFunctionPartial(SqlFunctionCtx* pCtx);
int32_t histogramFunctionMerge(SqlFunctionCtx* pCtx);
@ -164,7 +164,7 @@ int32_t histogramPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t getHistogramInfoSize();
int32_t histogramCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getHLLFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getHLLFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t hllFunction(SqlFunctionCtx* pCtx);
int32_t hllFunctionMerge(SqlFunctionCtx* pCtx);
int32_t hllFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
@ -172,48 +172,48 @@ int32_t hllPartialFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
int32_t getHLLInfoSize();
int32_t hllCombine(SqlFunctionCtx* pDestCtx, SqlFunctionCtx* pSourceCtx);
bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stateFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getStateFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool stateFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t stateCountFunction(SqlFunctionCtx* pCtx);
int32_t stateDurationFunction(SqlFunctionCtx* pCtx);
bool getCsumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getCsumFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t csumFunction(SqlFunctionCtx* pCtx);
bool getMavgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool mavgFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getMavgFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool mavgFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t mavgFunction(SqlFunctionCtx* pCtx);
bool getSampleFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool sampleFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getSampleFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool sampleFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t sampleFunction(SqlFunctionCtx* pCtx);
int32_t sampleFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getTailFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool tailFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
bool getTailFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool tailFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t tailFunction(SqlFunctionCtx* pCtx);
bool getUniqueFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool uniqueFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t uniqueFunction(SqlFunctionCtx *pCtx);
bool getUniqueFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool uniqueFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t uniqueFunction(SqlFunctionCtx* pCtx);
bool getModeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool modeFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t modeFunction(SqlFunctionCtx *pCtx);
bool getModeFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool modeFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t modeFunction(SqlFunctionCtx* pCtx);
int32_t modeFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool twaFunctionSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t twaFunction(SqlFunctionCtx *pCtx);
int32_t twaFinalize(struct SqlFunctionCtx *pCtx, SSDataBlock* pBlock);
bool getTwaFuncEnv(struct SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool twaFunctionSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t twaFunction(SqlFunctionCtx* pCtx);
int32_t twaFinalize(struct SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getSelectivityFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool blockDistSetup(SqlFunctionCtx *pCtx, SResultRowEntryInfo* pResultInfo);
int32_t blockDistFunction(SqlFunctionCtx *pCtx);
bool blockDistSetup(SqlFunctionCtx* pCtx, SResultRowEntryInfo* pResultInfo);
int32_t blockDistFunction(SqlFunctionCtx* pCtx);
int32_t blockDistFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);
bool getGroupKeyFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
bool getGroupKeyFuncEnv(SFunctionNode* pFunc, SFuncExecEnv* pEnv);
int32_t groupKeyFunction(SqlFunctionCtx* pCtx);
int32_t groupKeyFinalize(SqlFunctionCtx* pCtx, SSDataBlock* pBlock);

10
source/libs/function/inc/tfunctionInt.h
View File

@ -22,11 +22,11 @@ extern "C" {
#include "os.h"
#include "tname.h"
#include "taosdef.h"
#include "tvariant.h"
#include "function.h"
#include "taosdef.h"
#include "tname.h"
#include "tudf.h"
#include "tvariant.h"
bool topbot_datablock_filter(SqlFunctionCtx *pCtx, const char *minval, const char *maxval);
@ -37,8 +37,8 @@ bool topbot_datablock_filter(SqlFunctionCtx *pCtx, const char *minval, const cha
static FORCE_INLINE void initResultRowEntry(SResultRowEntryInfo *pResInfo, int32_t bufLen) {
pResInfo->initialized = true; // the this struct has been initialized flag
pResInfo->complete = false;
pResInfo->numOfRes = 0;
pResInfo->complete = false;
pResInfo->numOfRes = 0;
memset(GET_ROWCELL_INTERBUF(pResInfo), 0, bufLen);
}

16
source/libs/function/inc/thistogram.h
View File

@ -43,15 +43,15 @@ typedef struct SHistogramInfo {
int64_t numOfElems;
int32_t numOfEntries;
int32_t maxEntries;
double min;
double max;
double min;
double max;
#if defined(USE_ARRAYLIST)
SHistBin* elems;
#else
tSkipList* pList;
tSkipList* pList;
SMultiwayMergeTreeInfo* pLoserTree;
int32_t maxIndex;
bool ordered;
int32_t maxIndex;
bool ordered;
#endif
} SHistogramInfo;
@ -61,16 +61,16 @@ SHistogramInfo* tHistogramCreateFrom(void* pBuf, int32_t numOfBins);
int32_t tHistogramAdd(SHistogramInfo** pHisto, double val);
int64_t tHistogramSum(SHistogramInfo* pHisto, double v);
double* tHistogramUniform(SHistogramInfo* pHisto, double* ratio, int32_t num);
double* tHistogramUniform(SHistogramInfo* pHisto, double* ratio, int32_t num);
SHistogramInfo* tHistogramMerge(SHistogramInfo* pHisto1, SHistogramInfo* pHisto2, int32_t numOfEntries);
void tHistogramDestroy(SHistogramInfo** pHisto);
void tHistogramDestroy(SHistogramInfo** pHisto);
void tHistogramPrint(SHistogramInfo* pHisto);
int32_t histoBinarySearch(SHistBin* pEntry, int32_t len, double val);
SHeapEntry* tHeapCreate(int32_t numOfEntries);
void tHeapSort(SHeapEntry* pEntry, int32_t len);
void tHeapSort(SHeapEntry* pEntry, int32_t len);
#ifdef __cplusplus
}

6
source/libs/function/inc/tpercentile.h
View File

@ -61,10 +61,10 @@ typedef struct tMemBucket {
MinMaxEntry range; // value range
int32_t times; // count that has been checked for deciding the correct data value buckets.
__compar_fn_t comparFn;
tMemBucketSlot* pSlots;
SDiskbasedBuf* pBuffer;
tMemBucketSlot *pSlots;
SDiskbasedBuf *pBuffer;
__perc_hash_func_t hashFunc;
SHashObj* groupPagesMap; // disk page map for different groups;
SHashObj *groupPagesMap; // disk page map for different groups;
} tMemBucket;
tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval, double maxval);

8
source/libs/function/inc/tscript.h
View File

@ -17,18 +17,18 @@
#define TDENGINE_QSCRIPT_H
#if 0
#include <lua.h>
#include <lauxlib.h>
#include <lua.h>
#include <lualib.h>
#include "tutil.h"
#include "hash.h"
#include "tlist.h"
#include "tudf.h"
#include "tutil.h"
#define MAX_FUNC_NAME 64
#define USER_FUNC_NAME "funcName"
#define USER_FUNC_NAME "funcName"
#define USER_FUNC_NAME_LIMIT 48
enum ScriptState {
@ -81,4 +81,4 @@ void scriptEnvPoolCleanup();
bool isValidScript(char *script, int32_t len);
#endif
#endif //TDENGINE_QSCRIPT_H
#endif // TDENGINE_QSCRIPT_H

34
source/libs/function/inc/tudfInt.h
View File

@ -40,33 +40,31 @@ typedef struct SUdfSetupRequest {
typedef struct SUdfSetupResponse {
int64_t udfHandle;
int8_t outputType;
int8_t outputType;
int32_t outputLen;
int32_t bufSize;
} SUdfSetupResponse;
typedef struct SUdfCallRequest {
int64_t udfHandle;
int8_t callType;
int8_t callType;
SSDataBlock block;
SSDataBlock block;
SUdfInterBuf interBuf;
SUdfInterBuf interBuf2;
int8_t initFirst;
int8_t initFirst;
} SUdfCallRequest;
typedef struct SUdfCallResponse {
int8_t callType;
SSDataBlock resultData;
int8_t callType;
SSDataBlock resultData;
SUdfInterBuf resultBuf;
} SUdfCallResponse;
typedef struct SUdfTeardownRequest {
int64_t udfHandle;
} SUdfTeardownRequest;
typedef struct SUdfTeardownResponse {
#ifdef WINDOWS
size_t avoidCompilationErrors;
@ -79,8 +77,8 @@ typedef struct SUdfRequest {
int8_t type;
union {
SUdfSetupRequest setup;
SUdfCallRequest call;
SUdfSetupRequest setup;
SUdfCallRequest call;
SUdfTeardownRequest teardown;
};
} SUdfRequest;
@ -89,29 +87,29 @@ typedef struct SUdfResponse {
int32_t msgLen;
int64_t seqNum;
int8_t type;
int8_t type;
int32_t code;
union {
SUdfSetupResponse setupRsp;
SUdfCallResponse callRsp;
SUdfSetupResponse setupRsp;
SUdfCallResponse callRsp;
SUdfTeardownResponse teardownRsp;
};
} SUdfResponse;
int32_t encodeUdfRequest(void **buf, const SUdfRequest* request);
void* decodeUdfRequest(const void *buf, SUdfRequest* request);
int32_t encodeUdfRequest(void **buf, const SUdfRequest *request);
void *decodeUdfRequest(const void *buf, SUdfRequest *request);
int32_t encodeUdfResponse(void **buf, const SUdfResponse *response);
void* decodeUdfResponse(const void* buf, SUdfResponse *response);
void *decodeUdfResponse(const void *buf, SUdfResponse *response);
void freeUdfColumnData(SUdfColumnData *data, SUdfColumnMeta *meta);
void freeUdfColumn(SUdfColumn* col);
void freeUdfColumn(SUdfColumn *col);
void freeUdfDataDataBlock(SUdfDataBlock *block);
int32_t convertDataBlockToUdfDataBlock(SSDataBlock *block, SUdfDataBlock *udfBlock);
int32_t convertUdfColumnToDataBlock(SUdfColumn *udfCol, SSDataBlock *block);
int32_t getUdfdPipeName(char* pipeName, int32_t size);
int32_t getUdfdPipeName(char *pipeName, int32_t size);
#ifdef __cplusplus
}
#endif

78
source/libs/function/inc/udfc.h
View File

@ -14,20 +14,20 @@
#endif
enum {
UDF_TASK_SETUP = 0,
UDF_TASK_CALL = 1,
UDF_TASK_TEARDOWN = 2
UDF_TASK_SETUP = 0,
UDF_TASK_CALL = 1,
UDF_TASK_TEARDOWN = 2
};
typedef struct SSDataBlock{
char *data;
int32_t size;
typedef struct SSDataBlock {
char *data;
int32_t size;
} SSDataBlock;
typedef struct SUdfInfo {
char *udfName;
char *path;
char *udfName;
char *path;
} SUdfInfo;
typedef void *UdfcFuncHandle;
@ -36,9 +36,9 @@ int32_t createUdfdProxy();
int32_t destroyUdfdProxy();
//int32_t setupUdf(SUdfInfo *udf, int32_t numOfUdfs, UdfcFuncHandle *handles);
// int32_t setupUdf(SUdfInfo *udf, int32_t numOfUdfs, UdfcFuncHandle *handles);
int32_t setupUdf(SUdfInfo* udf, UdfcFuncHandle* handle);
int32_t setupUdf(SUdfInfo *udf, UdfcFuncHandle *handle);
int32_t callUdf(UdfcFuncHandle handle, int8_t step, char *state, int32_t stateSize, SSDataBlock input, char **newstate,
int32_t *newStateSize, SSDataBlock *output);
@ -46,43 +46,39 @@ int32_t callUdf(UdfcFuncHandle handle, int8_t step, char *state, int32_t stateSi
int32_t doTeardownUdf(UdfcFuncHandle handle);
typedef struct SUdfSetupRequest {
char udfName[16]; //
int8_t scriptType; // 0:c, 1: lua, 2:js
int8_t udfType; //udaf, udf, udtf
int16_t pathSize;
char *path;
char udfName[16]; //
int8_t scriptType; // 0:c, 1: lua, 2:js
int8_t udfType; // udaf, udf, udtf
int16_t pathSize;
char *path;
} SUdfSetupRequest;
typedef struct SUdfSetupResponse {
int64_t udfHandle;
int64_t udfHandle;
} SUdfSetupResponse;
typedef struct SUdfCallRequest {
int64_t udfHandle;
int8_t step;
int64_t udfHandle;
int8_t step;
int32_t inputBytes;
char *input;
int32_t inputBytes;
char *input;
int32_t stateBytes;
char *state;
int32_t stateBytes;
char *state;
} SUdfCallRequest;
typedef struct SUdfCallResponse {
int32_t outputBytes;
char *output;
int32_t newStateBytes;
char *newState;
int32_t outputBytes;
char *output;
int32_t newStateBytes;
char *newState;
} SUdfCallResponse;
typedef struct SUdfTeardownRequest {
int64_t udfHandle;
int64_t udfHandle;
} SUdfTeardownRequest;
typedef struct SUdfTeardownResponse {
#ifdef WINDOWS
size_t avoidCompilationErrors;
@ -90,24 +86,24 @@ typedef struct SUdfTeardownResponse {
} SUdfTeardownResponse;
typedef struct SUdfRequest {
int32_t msgLen;
int64_t seqNum;
int32_t msgLen;
int64_t seqNum;
int8_t type;
void *subReq;
int8_t type;
void *subReq;
} SUdfRequest;
typedef struct SUdfResponse {
int32_t msgLen;
int64_t seqNum;
int32_t msgLen;
int64_t seqNum;
int8_t type;
int32_t code;
void *subRsp;
int8_t type;
int32_t code;
void *subRsp;
} SUdfResponse;
int32_t decodeRequest(char *buf, int32_t bufLen, SUdfRequest **pRequest);
int32_t encodeResponse(char **buf, int32_t *bufLen, SUdfResponse *response);
int32_t encodeRequest(char **buf, int32_t *bufLen, SUdfRequest *request);
int32_t decodeResponse(char *buf, int32_t bufLen, SUdfResponse **pResponse);
#endif //UDF_UDF_H
#endif // UDF_UDF_H

7
source/libs/function/src/builtinsimpl.c
View File

@ -1557,8 +1557,8 @@ int32_t doMinMaxHelper(SqlFunctionCtx* pCtx, int32_t isMinFunc) {
numOfElems += 1;
}
} else if (type == TSDB_DATA_TYPE_FLOAT) {
float* pData = (float*)pCol->pData;
float* val = (float*)&pBuf->v;
float* pData = (float*)pCol->pData;
float* val = (float*)&pBuf->v;
for (int32_t i = start; i < start + numOfRows; ++i) {
if ((pCol->hasNull) && colDataIsNull_f(pCol->nullbitmap, i)) {
@ -2977,7 +2977,8 @@ int32_t lastFunction(SqlFunctionCtx* pCtx) {
return TSDB_CODE_SUCCESS;
}
static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst, int32_t rowIndex) {
static void firstLastTransferInfo(SqlFunctionCtx* pCtx, SFirstLastRes* pInput, SFirstLastRes* pOutput, bool isFirst,
int32_t rowIndex) {
SInputColumnInfoData* pColInfo = &pCtx->input;
if (pOutput->hasResult) {

18
source/libs/function/src/tfunctionInt.c
View File

@ -15,8 +15,8 @@
#include "os.h"
#include "taosdef.h"
#include "tmsg.h"
#include "thash.h"
#include "tmsg.h"
#include "ttypes.h"
#include "function.h"
@ -29,15 +29,13 @@
#include "ttszip.h"
#include "tudf.h"
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell) {
pCell->initialized = false;
}
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell) { pCell->initialized = false; }
int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock) {
int32_t maxRows = 0;
for (int32_t j = 0; j < num; ++j) {
SResultRowEntryInfo *pResInfo = GET_RES_INFO(&pCtx[j]);
SResultRowEntryInfo* pResInfo = GET_RES_INFO(&pCtx[j]);
if (pResInfo != NULL && maxRows < pResInfo->numOfRes) {
maxRows = pResInfo->numOfRes;
}
@ -46,12 +44,12 @@ int32_t getNumOfResult(SqlFunctionCtx* pCtx, int32_t num, SSDataBlock* pResBlock
assert(maxRows >= 0);
blockDataEnsureCapacity(pResBlock, maxRows);
for(int32_t i = 0; i < num; ++i) {
for (int32_t i = 0; i < num; ++i) {
SColumnInfoData* pCol = taosArrayGet(pResBlock->pDataBlock, i);
SResultRowEntryInfo *pResInfo = GET_RES_INFO(&pCtx[i]);
SResultRowEntryInfo* pResInfo = GET_RES_INFO(&pCtx[i]);
if (pResInfo->numOfRes == 0) {
for(int32_t j = 0; j < pResInfo->numOfRes; ++j) {
for (int32_t j = 0; j < pResInfo->numOfRes; ++j) {
colDataAppend(pCol, j, NULL, true); // TODO add set null data api
}
} else {
@ -70,6 +68,4 @@ bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry) {
return pEntry->complete;
}
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry) {
return pEntry->initialized;
}
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry) { return pEntry->initialized; }

14
source/libs/function/src/thistogram.c
View File

@ -14,10 +14,10 @@
*/
#include "os.h"
#include "thistogram.h"
#include "taosdef.h"
#include "tmsg.h"
#include "thistogram.h"
#include "tlosertree.h"
#include "tmsg.h"
/**
*
@ -54,7 +54,7 @@ SHistogramInfo* tHistogramCreateFrom(void* pBuf, int32_t numOfBins) {
SHistogramInfo* pHisto = (SHistogramInfo*)pBuf;
pHisto->elems = (SHistBin*)((char*)pBuf + sizeof(SHistogramInfo));
for(int32_t i = 0; i < numOfBins; ++i) {
for (int32_t i = 0; i < numOfBins; ++i) {
pHisto->elems[i].val = -DBL_MAX;
}
@ -116,7 +116,7 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
pEntry1->delta = ((SHistBin*)pResNode->pForward[0]->pData)->val - val;
if ((*pHisto)->ordered) {
int32_t lastIndex = (*pHisto)->maxIndex;
int32_t lastIndex = (*pHisto)->maxIndex;
SMultiwayMergeTreeInfo* pTree = (*pHisto)->pLoserTree;
(*pHisto)->pLoserTree->pNode[lastIndex + pTree->numOfEntries].pData = pResNode;
@ -156,7 +156,7 @@ int32_t tHistogramAdd(SHistogramInfo** pHisto, double val) {
SSkipListPrint((*pHisto)->pList, 1);
SMultiwayMergeTreeInfo* pTree = (*pHisto)->pLoserTree;
tSkipListNode* pHead = (*pHisto)->pList->pHead.pForward[0];
tSkipListNode* pHead = (*pHisto)->pList->pHead.pForward[0];
tSkipListNode* p1 = pHead;
@ -357,7 +357,7 @@ void tHistogramDestroy(SHistogramInfo** pHisto) {
}
void tHistogramPrint(SHistogramInfo* pHisto) {
printf("total entries: %d, elements: %"PRId64 "\n", pHisto->numOfEntries, pHisto->numOfElems);
printf("total entries: %d, elements: %" PRId64 "\n", pHisto->numOfEntries, pHisto->numOfElems);
#if defined(USE_ARRAYLIST)
for (int32_t i = 0; i < pHisto->numOfEntries; ++i) {
printf("%d: (%f, %" PRId64 ")\n", i + 1, pHisto->elems[i].val, pHisto->elems[i].num);
@ -536,7 +536,7 @@ SHistogramInfo* tHistogramMerge(SHistogramInfo* pHisto1, SHistogramInfo* pHisto2
}
SHistBin* pHistoBins = taosMemoryCalloc(1, sizeof(SHistBin) * (pHisto1->numOfEntries + pHisto2->numOfEntries));
int32_t i = 0, j = 0, k = 0;
int32_t i = 0, j = 0, k = 0;
while (i < pHisto1->numOfEntries && j < pHisto2->numOfEntries) {
if (pHisto1->elems[i].val < pHisto2->elems[j].val) {

159
source/libs/function/src/tpercentile.c
View File

@ -14,8 +14,8 @@
*/
#include "taoserror.h"
#include "tglobal.h"
#include "tcompare.h"
#include "tglobal.h"
#include "taosdef.h"
#include "tcompare.h"
@ -25,21 +25,20 @@
#define DEFAULT_NUM_OF_SLOT 1024
int32_t getGroupId(int32_t numOfSlots, int32_t slotIndex, int32_t times) {
return (times * numOfSlots) + slotIndex;
}
int32_t getGroupId(int32_t numOfSlots, int32_t slotIndex, int32_t times) { return (times * numOfSlots) + slotIndex; }
static SFilePage *loadDataFromFilePage(tMemBucket *pMemBucket, int32_t slotIdx) {
SFilePage *buffer = (SFilePage *)taosMemoryCalloc(1, pMemBucket->bytes * pMemBucket->pSlots[slotIdx].info.size + sizeof(SFilePage));
SFilePage *buffer =
(SFilePage *)taosMemoryCalloc(1, pMemBucket->bytes * pMemBucket->pSlots[slotIdx].info.size + sizeof(SFilePage));
int32_t groupId = getGroupId(pMemBucket->numOfSlots, slotIdx, pMemBucket->times);
SArray* pIdList = *(SArray**)taosHashGet(pMemBucket->groupPagesMap, &groupId, sizeof(groupId));
SArray *pIdList = *(SArray **)taosHashGet(pMemBucket->groupPagesMap, &groupId, sizeof(groupId));
int32_t offset = 0;
for(int32_t i = 0; i < taosArrayGetSize(pIdList); ++i) {
int32_t* pageId = taosArrayGet(pIdList, i);
for (int32_t i = 0; i < taosArrayGetSize(pIdList); ++i) {
int32_t *pageId = taosArrayGet(pIdList, i);
SFilePage* pg = getBufPage(pMemBucket->pBuffer, *pageId);
SFilePage *pg = getBufPage(pMemBucket->pBuffer, *pageId);
memcpy(buffer->data + offset, pg->data, (size_t)(pg->num * pMemBucket->bytes));
offset += (int32_t)(pg->num * pMemBucket->bytes);
@ -49,7 +48,7 @@ static SFilePage *loadDataFromFilePage(tMemBucket *pMemBucket, int32_t slotIdx)
return buffer;
}
static void resetBoundingBox(MinMaxEntry* range, int32_t type) {
static void resetBoundingBox(MinMaxEntry *range, int32_t type) {
if (IS_SIGNED_NUMERIC_TYPE(type)) {
range->i64MaxVal = INT64_MIN;
range->i64MinVal = INT64_MAX;
@ -62,17 +61,17 @@ static void resetBoundingBox(MinMaxEntry* range, int32_t type) {
}
}
static int32_t setBoundingBox(MinMaxEntry* range, int16_t type, double minval, double maxval) {
static int32_t setBoundingBox(MinMaxEntry *range, int16_t type, double minval, double maxval) {
if (minval > maxval) {
return -1;
}
if (IS_SIGNED_NUMERIC_TYPE(type)) {
range->i64MinVal = (int64_t) minval;
range->i64MaxVal = (int64_t) maxval;
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)){
range->u64MinVal = (uint64_t) minval;
range->u64MaxVal = (uint64_t) maxval;
range->i64MinVal = (int64_t)minval;
range->i64MaxVal = (int64_t)maxval;
} else if (IS_UNSIGNED_NUMERIC_TYPE(type)) {
range->u64MinVal = (uint64_t)minval;
range->u64MaxVal = (uint64_t)maxval;
} else {
range->dMinVal = minval;
range->dMaxVal = maxval;
@ -81,10 +80,10 @@ static int32_t setBoundingBox(MinMaxEntry* range, int16_t type, double minval, d
return 0;
}
static void resetPosInfo(SSlotInfo* pInfo) {
pInfo->size = 0;
static void resetPosInfo(SSlotInfo *pInfo) {
pInfo->size = 0;
pInfo->pageId = -1;
pInfo->data = NULL;
pInfo->data = NULL;
}
double findOnlyResult(tMemBucket *pMemBucket) {
@ -92,16 +91,16 @@ double findOnlyResult(tMemBucket *pMemBucket) {
for (int32_t i = 0; i < pMemBucket->numOfSlots; ++i) {
tMemBucketSlot *pSlot = &pMemBucket->pSlots[i];
if (pSlot->info.size == 0) {
if (pSlot->info.size == 0) {
continue;
}
int32_t groupId = getGroupId(pMemBucket->numOfSlots, i, pMemBucket->times);
SArray* list = *(SArray**)taosHashGet(pMemBucket->groupPagesMap, &groupId, sizeof(groupId));
SArray *list = *(SArray **)taosHashGet(pMemBucket->groupPagesMap, &groupId, sizeof(groupId));
assert(list->size == 1);
int32_t* pageId = taosArrayGet(list, 0);
SFilePage* pPage = getBufPage(pMemBucket->pBuffer, *pageId);
int32_t *pageId = taosArrayGet(list, 0);
SFilePage *pPage = getBufPage(pMemBucket->pBuffer, *pageId);
assert(pPage->num == 1);
double v = 0;
@ -121,14 +120,14 @@ int32_t tBucketIntHash(tMemBucket *pBucket, const void *value) {
if (v > pBucket->range.i64MaxVal || v < pBucket->range.i64MinVal) {
return index;
}
// divide the value range into 1024 buckets
uint64_t span = pBucket->range.i64MaxVal - pBucket->range.i64MinVal;
if (span < pBucket->numOfSlots) {
int64_t delta = v - pBucket->range.i64MinVal;
index = (delta % pBucket->numOfSlots);
} else {
double slotSpan = ((double)span) / pBucket->numOfSlots;
double slotSpan = ((double)span) / pBucket->numOfSlots;
uint64_t delta = v - pBucket->range.i64MinVal;
index = (int32_t)(delta / slotSpan);
@ -150,12 +149,12 @@ int32_t tBucketUintHash(tMemBucket *pBucket, const void *value) {
if (v > pBucket->range.u64MaxVal || v < pBucket->range.u64MinVal) {
return index;
}
// divide the value range into 1024 buckets
uint64_t span = pBucket->range.u64MaxVal - pBucket->range.u64MinVal;
if (span < pBucket->numOfSlots) {
int64_t delta = v - pBucket->range.u64MinVal;
index = (int32_t) (delta % pBucket->numOfSlots);
index = (int32_t)(delta % pBucket->numOfSlots);
} else {
double slotSpan = (double)span / pBucket->numOfSlots;
index = (int32_t)((v - pBucket->range.u64MinVal) / slotSpan);
@ -209,9 +208,9 @@ static __perc_hash_func_t getHashFunc(int32_t type) {
}
}
static void resetSlotInfo(tMemBucket* pBucket) {
static void resetSlotInfo(tMemBucket *pBucket) {
for (int32_t i = 0; i < pBucket->numOfSlots; ++i) {
tMemBucketSlot* pSlot = &pBucket->pSlots[i];
tMemBucketSlot *pSlot = &pBucket->pSlots[i];
resetBoundingBox(&pSlot->range, pBucket->type);
resetPosInfo(&pSlot->info);
@ -225,9 +224,9 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
}
pBucket->numOfSlots = DEFAULT_NUM_OF_SLOT;
pBucket->bufPageSize = 16384 * 4; // 16k per page
pBucket->bufPageSize = 16384 * 4; // 16k per page
pBucket->type = dataType;
pBucket->type = dataType;
pBucket->bytes = nElemSize;
pBucket->total = 0;
pBucket->times = 1;
@ -235,17 +234,17 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
pBucket->maxCapacity = 200000;
pBucket->groupPagesMap = taosHashInit(128, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
if (setBoundingBox(&pBucket->range, pBucket->type, minval, maxval) != 0) {
// qError("MemBucket:%p, invalid value range: %f-%f", pBucket, minval, maxval);
// qError("MemBucket:%p, invalid value range: %f-%f", pBucket, minval, maxval);
taosMemoryFree(pBucket);
return NULL;
}
pBucket->elemPerPage = (pBucket->bufPageSize - sizeof(SFilePage))/pBucket->bytes;
pBucket->elemPerPage = (pBucket->bufPageSize - sizeof(SFilePage)) / pBucket->bytes;
pBucket->comparFn = getKeyComparFunc(pBucket->type, TSDB_ORDER_ASC);
pBucket->hashFunc = getHashFunc(pBucket->type);
if (pBucket->hashFunc == NULL) {
// qError("MemBucket:%p, not support data type %d, failed", pBucket, pBucket->type);
// qError("MemBucket:%p, not support data type %d, failed", pBucket, pBucket->type);
taosMemoryFree(pBucket);
return NULL;
}
@ -270,8 +269,8 @@ tMemBucket *tMemBucketCreate(int16_t nElemSize, int16_t dataType, double minval,
tMemBucketDestroy(pBucket);
return NULL;
}
// qDebug("MemBucket:%p, elem size:%d", pBucket, pBucket->bytes);
// qDebug("MemBucket:%p, elem size:%d", pBucket, pBucket->bytes);
return pBucket;
}
@ -280,9 +279,9 @@ void tMemBucketDestroy(tMemBucket *pBucket) {
return;
}
void* p = taosHashIterate(pBucket->groupPagesMap, NULL);
while(p) {
SArray** p1 = p;
void *p = taosHashIterate(pBucket->groupPagesMap, NULL);
while (p) {
SArray **p1 = p;
p = taosHashIterate(pBucket->groupPagesMap, p);
taosArrayDestroy(*p1);
}
@ -341,7 +340,7 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
int32_t count = 0;
int32_t bytes = pBucket->bytes;
for (int32_t i = 0; i < size; ++i) {
char *d = (char *) data + i * bytes;
char *d = (char *)data + i * bytes;
int32_t index = (pBucket->hashFunc)(pBucket, d);
if (index < 0) {
continue;
@ -365,11 +364,11 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
pSlot->info.data = NULL;
}
SArray* pPageIdList = (SArray*)taosHashGet(pBucket->groupPagesMap, &groupId, sizeof(groupId));
SArray *pPageIdList = (SArray *)taosHashGet(pBucket->groupPagesMap, &groupId, sizeof(groupId));
if (pPageIdList == NULL) {
SArray* pList = taosArrayInit(4, sizeof(int32_t));
SArray *pList = taosArrayInit(4, sizeof(int32_t));
taosHashPut(pBucket->groupPagesMap, &groupId, sizeof(groupId), &pList, POINTER_BYTES);
pPageIdList = pList;
pPageIdList = pList;
}
pSlot->info.data = getNewBufPage(pBucket->pBuffer, &pageId);
@ -396,29 +395,29 @@ int32_t tMemBucketPut(tMemBucket *pBucket, const void *data, size_t size) {
* slot of the next segment.
*/
static MinMaxEntry getMinMaxEntryOfNextSlotWithData(tMemBucket *pMemBucket, int32_t slotIdx) {
int32_t j = slotIdx + 1;
while (j < pMemBucket->numOfSlots && (pMemBucket->pSlots[j].info.size == 0)) {
++j;
}
int32_t j = slotIdx + 1;
while (j < pMemBucket->numOfSlots && (pMemBucket->pSlots[j].info.size == 0)) {
++j;
}
assert(j < pMemBucket->numOfSlots);
return pMemBucket->pSlots[j].range;
assert(j < pMemBucket->numOfSlots);
return pMemBucket->pSlots[j].range;
}
static bool isIdenticalData(tMemBucket *pMemBucket, int32_t index);
static double getIdenticalDataVal(tMemBucket* pMemBucket, int32_t slotIndex) {
static double getIdenticalDataVal(tMemBucket *pMemBucket, int32_t slotIndex) {
assert(isIdenticalData(pMemBucket, slotIndex));
tMemBucketSlot *pSlot = &pMemBucket->pSlots[slotIndex];
double finalResult = 0.0;
if (IS_SIGNED_NUMERIC_TYPE(pMemBucket->type)) {
finalResult = (double) pSlot->range.i64MinVal;
finalResult = (double)pSlot->range.i64MinVal;
} else if (IS_UNSIGNED_NUMERIC_TYPE(pMemBucket->type)) {
finalResult = (double) pSlot->range.u64MinVal;
finalResult = (double)pSlot->range.u64MinVal;
} else {
finalResult = (double) pSlot->range.dMinVal;
finalResult = (double)pSlot->range.dMinVal;
}
return finalResult;
@ -445,14 +444,14 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
double maxOfThisSlot = 0;
double minOfNextSlot = 0;
if (IS_SIGNED_NUMERIC_TYPE(pMemBucket->type)) {
maxOfThisSlot = (double) pSlot->range.i64MaxVal;
minOfNextSlot = (double) next.i64MinVal;
maxOfThisSlot = (double)pSlot->range.i64MaxVal;
minOfNextSlot = (double)next.i64MinVal;
} else if (IS_UNSIGNED_NUMERIC_TYPE(pMemBucket->type)) {
maxOfThisSlot = (double) pSlot->range.u64MaxVal;
minOfNextSlot = (double) next.u64MinVal;
maxOfThisSlot = (double)pSlot->range.u64MaxVal;
minOfNextSlot = (double)next.u64MinVal;
} else {
maxOfThisSlot = (double) pSlot->range.dMaxVal;
minOfNextSlot = (double) next.dMinVal;
maxOfThisSlot = (double)pSlot->range.dMaxVal;
minOfNextSlot = (double)next.dMinVal;
}
assert(minOfNextSlot > maxOfThisSlot);
@ -478,32 +477,32 @@ double getPercentileImpl(tMemBucket *pMemBucket, int32_t count, double fraction)
return val;
} else { // incur a second round bucket split
if (isIdenticalData(pMemBucket, i)) {
return getIdenticalDataVal(pMemBucket, i);
}
if (isIdenticalData(pMemBucket, i)) {
return getIdenticalDataVal(pMemBucket, i);
}
// try next round
pMemBucket->times += 1;
// qDebug("MemBucket:%p, start next round data bucketing, time:%d", pMemBucket, pMemBucket->times);
// try next round
pMemBucket->times += 1;
// qDebug("MemBucket:%p, start next round data bucketing, time:%d", pMemBucket, pMemBucket->times);
pMemBucket->range = pSlot->range;
pMemBucket->total = 0;
pMemBucket->range = pSlot->range;
pMemBucket->total = 0;
resetSlotInfo(pMemBucket);
resetSlotInfo(pMemBucket);
int32_t groupId = getGroupId(pMemBucket->numOfSlots, i, pMemBucket->times - 1);
int32_t groupId = getGroupId(pMemBucket->numOfSlots, i, pMemBucket->times - 1);
SIDList list = taosHashGet(pMemBucket->groupPagesMap, &groupId, sizeof(groupId));
assert(list->size > 0);
assert(list->size > 0);
for (int32_t f = 0; f < list->size; ++f) {
SPageInfo *pgInfo = *(SPageInfo **)taosArrayGet(list, f);
SFilePage *pg = getBufPage(pMemBucket->pBuffer, getPageId(pgInfo));
for (int32_t f = 0; f < list->size; ++f) {
SPageInfo *pgInfo = *(SPageInfo **)taosArrayGet(list, f);
SFilePage *pg = getBufPage(pMemBucket->pBuffer, getPageId(pgInfo));
tMemBucketPut(pMemBucket, pg->data, (int32_t)pg->num);
releaseBufPageInfo(pMemBucket->pBuffer, pgInfo);
}
tMemBucketPut(pMemBucket, pg->data, (int32_t)pg->num);
releaseBufPageInfo(pMemBucket->pBuffer, pgInfo);
}
return getPercentileImpl(pMemBucket, count - num, fraction);
return getPercentileImpl(pMemBucket, count - num, fraction);
}
} else {
num += pSlot->info.size;
@ -527,7 +526,7 @@ double getPercentile(tMemBucket *pMemBucket, double percent) {
// find the min/max value, no need to scan all data in bucket
if (fabs(percent - 100.0) < DBL_EPSILON || (percent < DBL_EPSILON)) {
MinMaxEntry* pRange = &pMemBucket->range;
MinMaxEntry *pRange = &pMemBucket->range;
if (IS_SIGNED_NUMERIC_TYPE(pMemBucket->type)) {
double v = (double)(fabs(percent - 100) < DBL_EPSILON ? pRange->i64MaxVal : pRange->i64MinVal);
@ -536,11 +535,11 @@ double getPercentile(tMemBucket *pMemBucket, double percent) {
double v = (double)(fabs(percent - 100) < DBL_EPSILON ? pRange->u64MaxVal : pRange->u64MinVal);
return v;
} else {
return fabs(percent - 100) < DBL_EPSILON? pRange->dMaxVal:pRange->dMinVal;
return fabs(percent - 100) < DBL_EPSILON ? pRange->dMaxVal : pRange->dMinVal;
}
}
double percentVal = (percent * (pMemBucket->total - 1)) / ((double)100.0);
double percentVal = (percent * (pMemBucket->total - 1)) / ((double)100.0);
// do put data by using buckets
int32_t orderIdx = (int32_t)percentVal;

4
source/libs/function/src/tscript.c
View File

@ -13,10 +13,10 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "os.h"
#include "tscript.h"
#include "ttypes.h"
#include "os.h"
#include "tstrbuild.h"
#include "ttypes.h"
//#include "queryLog.h"
#include "ttokendef.h"
#if 0

594
source/libs/function/src/tudf.c
View File

File diff suppressed because it is too large Load Diff

13
source/libs/function/test/runUdf.c
View File

@ -53,7 +53,7 @@ int scalarFuncTest() {
blockDataEnsureCapacity(pBlock, 1024);
pBlock->info.rows = 1024;
SColumnInfoData* pCol = taosArrayGet(pBlock->pDataBlock, 0);
SColumnInfoData *pCol = taosArrayGet(pBlock->pDataBlock, 0);
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(pCol, j, &j);
}
@ -68,14 +68,13 @@ int scalarFuncTest() {
SColumnInfoData *col = output.columnData;
for (int32_t i = 0; i < output.numOfRows; ++i) {
if (i % 100 == 0)
fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t)));
if (i % 100 == 0) fprintf(stderr, "%d\t%d\n", i, *(int32_t *)(col->pData + i * sizeof(int32_t)));
}
colDataDestroy(output.columnData);
taosMemoryFree(output.columnData);
}
int64_t end = taosGetTimestampUs();
fprintf(stderr, "time: %f\n", (end-beg)/1000.0);
fprintf(stderr, "time: %f\n", (end - beg) / 1000.0);
doTeardownUdf(handle);
return 0;
@ -92,13 +91,13 @@ int aggregateFuncTest() {
SSDataBlock *pBlock = createDataBlock();
for (int32_t i = 0; i < taosArrayGetSize(pBlock->pDataBlock); ++i) {
SColumnInfoData colInfo = createColumnInfoData(TSDB_DATA_TYPE_INT, sizeof(int32_t), 1);
blockDataAppendColInfo(pBlock, &colInfo);
blockDataAppendColInfo(pBlock, &colInfo);
}
blockDataEnsureCapacity(pBlock, 1024);
pBlock->info.rows = 1024;
SColumnInfoData* pColInfo = bdGetColumnInfoData(pBlock, 0);
SColumnInfoData *pColInfo = bdGetColumnInfoData(pBlock, 0);
for (int32_t j = 0; j < pBlock->info.rows; ++j) {
colDataAppendInt32(pColInfo, j, &j);
}
@ -111,7 +110,7 @@ int aggregateFuncTest() {
taosArrayDestroy(pBlock->pDataBlock);
doCallUdfAggFinalize(handle, &newBuf, &resultBuf);
fprintf(stderr, "agg result: %f\n", *(double*)resultBuf.buf);
fprintf(stderr, "agg result: %f\n", *(double *)resultBuf.buf);
freeUdfInterBuf(&buf);
freeUdfInterBuf(&newBuf);

21
source/libs/function/test/udf1.c
View File

@ -1,6 +1,6 @@
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#ifdef LINUX
#include <unistd.h>
#endif
@ -9,16 +9,11 @@
#endif
#include "taosudf.h"
DLL_EXPORT int32_t udf1_init() { return 0; }
DLL_EXPORT int32_t udf1_init() {
return 0;
}
DLL_EXPORT int32_t udf1_destroy() { return 0; }
DLL_EXPORT int32_t udf1_destroy() {
return 0;
}
DLL_EXPORT int32_t udf1(SUdfDataBlock* block, SUdfColumn *resultCol) {
DLL_EXPORT int32_t udf1(SUdfDataBlock *block, SUdfColumn *resultCol) {
SUdfColumnMeta *meta = &resultCol->colMeta;
meta->bytes = 4;
meta->type = TSDB_DATA_TYPE_INT;
@ -35,14 +30,14 @@ DLL_EXPORT int32_t udf1(SUdfDataBlock* block, SUdfColumn *resultCol) {
break;
}
}
if ( j == block->numOfCols) {
if (j == block->numOfCols) {
int32_t luckyNum = 88;
udfColDataSet(resultCol, i, (char *)&luckyNum, false);
}
}
//to simulate actual processing delay by udf
// to simulate actual processing delay by udf
#ifdef LINUX
usleep(1 * 1000); // usleep takes sleep time in us (1 millionth of a second)
usleep(1 * 1000); // usleep takes sleep time in us (1 millionth of a second)
#endif
#ifdef WINDOWS
Sleep(1);

29
source/libs/function/test/udf2.c
View File

@ -1,32 +1,27 @@
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "taosudf.h"
DLL_EXPORT int32_t udf2_init() {
return 0;
}
DLL_EXPORT int32_t udf2_init() { return 0; }
DLL_EXPORT int32_t udf2_destroy() {
return 0;
}
DLL_EXPORT int32_t udf2_destroy() { return 0; }
DLL_EXPORT int32_t udf2_start(SUdfInterBuf *buf) {
DLL_EXPORT int32_t udf2_start(SUdfInterBuf* buf) {
*(int64_t*)(buf->buf) = 0;
buf->bufLen = sizeof(double);
buf->numOfResult = 0;
return 0;
}
DLL_EXPORT int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
DLL_EXPORT int32_t udf2(SUdfDataBlock* block, SUdfInterBuf* interBuf, SUdfInterBuf* newInterBuf) {
double sumSquares = *(double*)interBuf->buf;
int8_t numNotNull = 0;
for (int32_t i = 0; i < block->numOfCols; ++i) {
SUdfColumn* col = block->udfCols[i];
if (!(col->colMeta.type == TSDB_DATA_TYPE_INT ||
col->colMeta.type == TSDB_DATA_TYPE_DOUBLE)) {
if (!(col->colMeta.type == TSDB_DATA_TYPE_INT || col->colMeta.type == TSDB_DATA_TYPE_DOUBLE)) {
return TSDB_CODE_UDF_INVALID_INPUT;
}
}
@ -38,18 +33,18 @@ DLL_EXPORT int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterB
}
switch (col->colMeta.type) {
case TSDB_DATA_TYPE_INT: {
char* cell = udfColDataGetData(col, j);
char* cell = udfColDataGetData(col, j);
int32_t num = *(int32_t*)cell;
sumSquares += (double)num * num;
break;
}
case TSDB_DATA_TYPE_DOUBLE: {
char* cell = udfColDataGetData(col, j);
char* cell = udfColDataGetData(col, j);
double num = *(double*)cell;
sumSquares += num * num;
break;
}
default:
default:
break;
}
++numNotNull;
@ -67,7 +62,7 @@ DLL_EXPORT int32_t udf2(SUdfDataBlock* block, SUdfInterBuf *interBuf, SUdfInterB
return 0;
}
DLL_EXPORT int32_t udf2_finish(SUdfInterBuf* buf, SUdfInterBuf *resultData) {
DLL_EXPORT int32_t udf2_finish(SUdfInterBuf* buf, SUdfInterBuf* resultData) {
if (buf->numOfResult == 0) {
resultData->numOfResult = 0;
return 0;

4
tools/scripts/codeFormat.sh
View File

@ -16,7 +16,7 @@ FORMAT_DIR_LIST=(
"${PRJ_ROOT_DIR}/source/libs/catalog"
"${PRJ_ROOT_DIR}/source/libs/command"
"${PRJ_ROOT_DIR}/source/libs/executor"
# "${PRJ_ROOT_DIR}/source/libs/function"
"${PRJ_ROOT_DIR}/source/libs/function"
"${PRJ_ROOT_DIR}/source/libs/index"
"${PRJ_ROOT_DIR}/source/libs/monitor"
"${PRJ_ROOT_DIR}/source/libs/nodes"
@ -49,3 +49,5 @@ for d in ${FORMAT_DIR_LIST[@]}; do
done
cd ${ORIGIN_DIR}
# find source/libs/ -path ./source/libs/qworker -prune -o -regex '.*\.\(cpp\|hpp\|c\|h\)' -print