Skip to content

T
TDengine

taosdata / TDengine
大约 1 年 前同步成功

通知 1184
Star 22015
Fork 4786
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 155b865a 编写于 作者: H Hongze Cheng
浏览文件
操作

Merge branch '3.0' into feature/vnode

上级 a1893dd8 5a4f137b
展开全部 隐藏空白更改
内联 并排
Showing with 16975 addition and 2702 deletion
Jenkinsfile
浏览文件 @ 155b865a
......@@ -142,7 +142,7 @@ pipeline {
}
}
// stage('Parallel test stage') {
// //only build pr
// skip defaultCheckout
// options { skipDefaultCheckout() }
// when {
// allOf{
......
include/common/common.h
浏览文件 @ 155b865a
......@@ -19,7 +19,7 @@
#include "taosdef.h"
#include "taosmsg.h"
#include "tarray.h"
#include "tvariant.h"
//typedef struct STimeWindow {
// TSKEY skey;
// TSKEY ekey;
......@@ -66,4 +66,59 @@ typedef struct SColumnInfoData {
char *pData; // the corresponding block data in memory
} SColumnInfoData;
//======================================================================================================================
// the following structure shared by parser and executor
typedef struct SColumn {
uint64_t uid;
char name[TSDB_COL_NAME_LEN];
int8_t flag; // column type: normal column, tag, or user-input column (integer/float/string)
SColumnInfo info;
} SColumn;
typedef struct SLimit {
int64_t limit;
int64_t offset;
} SLimit;
typedef struct SOrder {
uint32_t order;
int32_t orderColId;
} SOrder;
typedef struct SGroupbyExpr {
SArray* columnInfo; // SArray<SColIndex>, group by columns information
bool groupbyTag; // group by tag or column
} SGroupbyExpr;
// the structure for sql function in select clause
typedef struct SSqlExpr {
char token[TSDB_COL_NAME_LEN]; // original token
SSchema resSchema;
int32_t numOfCols;
SColumn* pColumns; // data columns that are required by query
int32_t interBytes; // inter result buffer size
int16_t numOfParams; // argument value of each function
SVariant param[3]; // parameters are not more than 3
} SSqlExpr;
typedef struct SExprInfo {
struct SSqlExpr base;
struct tExprNode *pExpr;
} SExprInfo;
typedef struct SStateWindow {
SColumn col;
} SStateWindow;
typedef struct SSessionWindow {
int64_t gap; // gap between two session window(in microseconds)
SColumn col;
} SSessionWindow;
#define QUERY_ASC_FORWARD_STEP 1
#define QUERY_DESC_FORWARD_STEP -1
#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
#endif // TDENGINE_COMMON_H
include/common/taosmsg.h
浏览文件 @ 155b865a
......@@ -204,14 +204,14 @@ enum _mgmt_table {
#define TSDB_COL_NORMAL 0x0u // the normal column of the table
#define TSDB_COL_TAG 0x1u // the tag column type
#define TSDB_COL_UDC 0x2u // the user specified normal string column, it is a dummy column
#define TSDB_COL_NULL 0x4u // the column filter NULL or not
#define TSDB_COL_TMP 0x4u // internal column generated by the previous operators
#define TSDB_COL_NULL 0x8u // the column filter NULL or not
#define TSDB_COL_IS_TAG(f) (((f&(~(TSDB_COL_NULL)))&TSDB_COL_TAG) != 0)
#define TSDB_COL_IS_NORMAL_COL(f) ((f&(~(TSDB_COL_NULL))) == TSDB_COL_NORMAL)
#define TSDB_COL_IS_UD_COL(f) ((f&(~(TSDB_COL_NULL))) == TSDB_COL_UDC)
#define TSDB_COL_REQ_NULL(f) (((f)&TSDB_COL_NULL) != 0)
extern char *taosMsg[];
#pragma pack(push, 1)
......@@ -491,11 +491,6 @@ typedef struct SInterval {
int64_t offset;
} SInterval;
typedef struct SSessionWindow {
int64_t gap; // gap between two session window(in microseconds)
int32_t primaryColId; // primary timestamp column
} SSessionWindow;
typedef struct {
SMsgHead head;
char version[TSDB_VERSION_LEN];
......@@ -520,7 +515,7 @@ typedef struct {
int16_t orderColId;
int16_t numOfCols; // the number of columns will be load from vnode
SInterval interval;
SSessionWindow sw; // session window
// SSessionWindow sw; // session window
uint16_t tagCondLen; // tag length in current query
uint16_t colCondLen; // column length in current query
int16_t numOfGroupCols; // num of group by columns
......
include/common/tname.h
浏览文件 @ 155b865a
......@@ -16,6 +16,8 @@
#ifndef TDENGINE_TNAME_H
#define TDENGINE_TNAME_H
#include "taosmsg.h"
#define TSDB_DB_NAME_T 1
#define TSDB_TABLE_NAME_T 2
......@@ -52,6 +54,8 @@ int32_t tNameFromString(SName* dst, const char* str, uint32_t type);
int32_t tNameSetAcctId(SName* dst, const char* acct);
SSchema* tGetTbnameColumnSchema();
#if 0
int32_t tNameSetDbName(SName* dst, const char* acct, SToken* dbToken);
#endif
......
source/libs/function/inc/ttszip.h → include/common/ttszip.h
浏览文件 @ 155b865a
......@@ -21,7 +21,7 @@ extern "C" {
#endif
#include "os.h"
#include "taosdef.h"
#include "tdef.h"
#include "tvariant.h"
#define MEM_BUF_SIZE (1 << 20)
......
include/libs/function/function.h
浏览文件 @ 155b865a
......@@ -26,8 +26,8 @@ extern "C" {
#define MAX_INTERVAL_TIME_WINDOW 1000000 // maximum allowed time windows in final results
#define FUNCTION_SCALAR 1
#define FUNCTION_AGG 2
#define FUNCTION_TYPE_SCALAR 1
#define FUNCTION_TYPE_AGG 2
#define TOP_BOTTOM_QUERY_LIMIT 100
#define FUNCTIONS_NAME_MAX_LENGTH 16
......@@ -78,13 +78,30 @@ extern "C" {
#define FUNCTION_MODE 36
#define FUNCTION_SAMPLE 37
#define FUNCTION_COV 38
// determine the real data need to calculated the result
enum {
BLK_DATA_NO_NEEDED = 0x0,
BLK_DATA_STATIS_NEEDED = 0x1,
BLK_DATA_ALL_NEEDED = 0x3,
BLK_DATA_DISCARD = 0x4, // discard current data block since it is not qualified for filter
};
enum {
MASTER_SCAN = 0x0u,
REVERSE_SCAN = 0x1u,
REPEAT_SCAN = 0x2u, //repeat scan belongs to the master scan
MERGE_STAGE = 0x20u,
};
typedef struct SPoint1 {
int64_t key;
union{double val; char* ptr;};
} SPoint1;
struct SQLFunctionCtx;
struct SResultRowCellInfo;
struct SResultRowEntryInfo;
//for selectivity query, the corresponding tag value is assigned if the data is qualified
typedef struct SExtTagsInfo {
......@@ -93,6 +110,23 @@ typedef struct SExtTagsInfo {
struct SQLFunctionCtx **pTagCtxList;
} SExtTagsInfo;
typedef struct SResultDataInfo {
int16_t type;
int16_t bytes;
int32_t intermediateBytes;
} SResultDataInfo;
#define GET_RES_INFO(ctx) ((ctx)->resultInfo)
typedef struct SFunctionFpSet {
bool (*init)(struct SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo); // setup the execute environment
void (*addInput)(struct SQLFunctionCtx *pCtx);
// finalizer must be called after all exec has been executed to generated final result.
void (*finalize)(struct SQLFunctionCtx *pCtx);
void (*combine)(struct SQLFunctionCtx *pCtx);
} SFunctionFpSet;
// sql function runtime context
typedef struct SQLFunctionCtx {
int32_t size; // number of rows
......@@ -101,9 +135,7 @@ typedef struct SQLFunctionCtx {
int16_t inputType;
int16_t inputBytes;
int16_t outputType;
int16_t outputBytes; // size of results, determined by function and input column data type
int32_t interBufBytes; // internal buffer size
SResultDataInfo resDataInfo;
bool hasNull; // null value exist in current block
bool requireNull; // require null in some function
bool stableQuery;
......@@ -117,18 +149,21 @@ typedef struct SQLFunctionCtx {
void *ptsOutputBuf; // corresponding output buffer for timestamp of each result, e.g., top/bottom*/
SVariant tag;
bool isSmaSet;
SColumnDataAgg sma;
struct SResultRowCellInfo *resultInfo;
bool isAggSet;
SColumnDataAgg agg;
struct SResultRowEntryInfo *resultInfo;
SExtTagsInfo tagInfo;
SPoint1 start;
SPoint1 end;
SFunctionFpSet* fpSet;
} SQLFunctionCtx;
enum {
TEXPR_NODE_DUMMY = 0x0,
TEXPR_BINARYEXPR_NODE= 0x1,
TEXPR_UNARYEXPR_NODE = 0x2,
TEXPR_FUNCTION_NODE = 0x3,
TEXPR_COL_NODE = 0x4,
TEXPR_VALUE_NODE = 0x8,
};
......@@ -137,10 +172,7 @@ typedef struct tExprNode {
uint8_t nodeType;
union {
struct {
union {
int32_t optr; // binary operator
int32_t functionId;// unary operator
};
int32_t optr; // binary operator
void *info; // support filter operation on this expression only available for leaf node
struct tExprNode *pLeft; // left child pointer
struct tExprNode *pRight; // right child pointer
......@@ -148,44 +180,52 @@ typedef struct tExprNode {
SSchema *pSchema;// column node
struct SVariant *pVal; // value node
struct {// function node
char functionName[FUNCTIONS_NAME_MAX_LENGTH];
// int32_t functionId;
int32_t num;
// Note that the attribute of pChild is not the parameter of function, it is the columns that involved in the
// calculation instead.
// E.g., Cov(col1, col2), the column information, w.r.t. the col1 and col2, is kept in pChild nodes.
// The concat function, concat(col1, col2), is a binary scalar
// operator and is kept in the attribute of _node.
struct tExprNode **pChild;
} _function;
};
} tExprNode;
//TODO create?
void exprTreeToBinary(SBufferWriter* bw, tExprNode* pExprTree);
void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *));
typedef struct SAggFunctionInfo {
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // Scalar function or aggregation function
uint8_t functionId; // Function Id
int8_t sFunctionId; // Transfer function for super table query
uint16_t status;
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // Scalar function or aggregation function
uint32_t functionId; // Function Id
int8_t sFunctionId; // Transfer function for super table query
uint16_t status;
bool (*init)(SQLFunctionCtx *pCtx, struct SResultRowCellInfo* pResultCellInfo); // setup the execute environment
void (*exec)(SQLFunctionCtx *pCtx);
bool (*init)(SQLFunctionCtx *pCtx, struct SResultRowEntryInfo* pResultCellInfo); // setup the execute environment
void (*addInput)(SQLFunctionCtx *pCtx);
// finalizer must be called after all exec has been executed to generated final result.
void (*xFinalize)(SQLFunctionCtx *pCtx);
void (*mergeFunc)(SQLFunctionCtx *pCtx);
void (*finalize)(SQLFunctionCtx *pCtx);
void (*combine)(SQLFunctionCtx *pCtx);
int32_t (*dataReqFunc)(SQLFunctionCtx *pCtx, STimeWindow* w, int32_t colId);
} SAggFunctionInfo;
typedef struct SScalarFunctionInfo {
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // scalar function or aggregation function
uint8_t functionId; // index of scalar function
struct SScalarFuncParam;
bool (*init)(SQLFunctionCtx *pCtx, struct SResultRowCellInfo* pResultCellInfo); // setup the execute environment
void (*exec)(SQLFunctionCtx *pCtx);
typedef struct SScalarFunctionInfo {
char name[FUNCTIONS_NAME_MAX_LENGTH];
int8_t type; // scalar function or aggregation function
uint32_t functionId; // index of scalar function
void (*process)(struct SScalarFuncParam* pOutput, size_t numOfInput, const struct SScalarFuncParam *pInput);
} SScalarFunctionInfo;
typedef struct SResultDataInfo {
int16_t type;
int16_t bytes;
int32_t intermediateBytes;
} SResultDataInfo;
typedef struct SMultiFunctionsDesc {
bool stableQuery;
bool groupbyColumn;
......@@ -195,11 +235,12 @@ typedef struct SMultiFunctionsDesc {
bool hasFilter;
bool onlyTagQuery;
bool orderProjectQuery;
bool stateWindow;
bool globalMerge;
bool multigroupResult;
bool blockDistribution;
bool stateWindow;
bool timewindow;
bool sessionWindow;
bool topbotQuery;
bool interpQuery;
bool distinct;
......@@ -215,16 +256,61 @@ int32_t getResultDataInfo(int32_t dataType, int32_t dataBytes, int32_t functionI
* @param len
* @return
*/
int32_t qIsBuiltinFunction(const char* name, int32_t len);
int32_t qIsBuiltinFunction(const char* name, int32_t len, bool* scalarFunction);
bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* functionId);
const char* qGetFunctionName(int32_t functionId);
bool qIsAggregateFunction(const char* functionName);
tExprNode* exprTreeFromBinary(const void* data, size_t size);
void extractFunctionDesc(SArray* pFunctionIdList, SMultiFunctionsDesc* pDesc);
tExprNode* exprdup(tExprNode* pTree);
void resetResultRowEntryResult(SQLFunctionCtx* pCtx, int32_t num);
void cleanupResultRowEntry(struct SResultRowEntryInfo* pCell);
int32_t getNumOfResult(SQLFunctionCtx* pCtx, int32_t num);
bool isRowEntryCompleted(struct SResultRowEntryInfo* pEntry);
bool isRowEntryInitialized(struct SResultRowEntryInfo* pEntry);
struct SScalarFunctionSupport* createScalarFuncSupport(int32_t num);
void destroyScalarFuncSupport(struct SScalarFunctionSupport* pSupport, int32_t num);
struct SScalarFunctionSupport* getScalarFuncSupport(struct SScalarFunctionSupport* pSupport, int32_t index);
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// fill api
struct SFillInfo;
struct SFillColInfo;
typedef struct SPoint {
int64_t key;
void * val;
} SPoint;
void taosFillSetStartInfo(struct SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey);
void taosResetFillInfo(struct SFillInfo* pFillInfo, TSKEY startTimestamp);
void taosFillSetInputDataBlock(struct SFillInfo* pFillInfo, const struct SSDataBlock* pInput);
struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, const int64_t* fillVal);
bool taosFillHasMoreResults(struct SFillInfo* pFillInfo);
struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
struct SFillColInfo* pFillCol, void* handle);
void* taosDestroyFillInfo(struct SFillInfo *pFillInfo);
int64_t taosFillResultDataBlock(struct SFillInfo* pFillInfo, void** output, int32_t capacity);
int64_t getFillInfoStart(struct SFillInfo *pFillInfo);
int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2, int32_t inputType);
///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
// udf api
struct SUdfInfo;
void qAddUdfInfo(uint64_t id, struct SUdfInfo* pUdfInfo);
void qRemoveUdfInfo(uint64_t id, struct SUdfInfo* pUdfInfo);
#ifdef __cplusplus
}
#endif
......
include/libs/parser/parser.h
浏览文件 @ 155b865a
......@@ -26,50 +26,6 @@ extern "C" {
#include "tvariant.h"
#include "function.h"
typedef struct SColumn {
uint64_t tableUid;
int32_t columnIndex;
SColumnInfo info;
} SColumn;
// the structure for sql function in select clause
typedef struct SSqlExpr {
char token[TSDB_COL_NAME_LEN]; // original token
SSchema resSchema;
SColIndex colInfo;
uint64_t uid; // table uid, todo refactor use the pointer
int32_t interBytes; // inter result buffer size
int16_t numOfParams; // argument value of each function
SVariant param[3]; // parameters are not more than 3
} SSqlExpr;
typedef struct SExprInfo {
SSqlExpr base;
struct tExprNode *pExpr;
} SExprInfo;
//typedef struct SInterval {
// int32_t tz; // query client timezone
// char intervalUnit;
// char slidingUnit;
// char offsetUnit;
// int64_t interval;
// int64_t sliding;
// int64_t offset;
//} SInterval;
//
//typedef struct SSessionWindow {
// int64_t gap; // gap between two session window(in microseconds)
// int32_t primaryColId; // primary timestamp column
//} SSessionWindow;
typedef struct SGroupbyExpr {
int16_t tableIndex;
SArray* columnInfo; // SArray<SColIndex>, group by columns information
int16_t orderIndex; // order by column index
int16_t orderType; // order by type: asc/desc
} SGroupbyExpr;
typedef struct SField {
char name[TSDB_COL_NAME_LEN];
uint8_t type;
......@@ -82,16 +38,6 @@ typedef struct SFieldInfo {
SArray *internalField; // SArray<SInternalField>
} SFieldInfo;
typedef struct SLimit {
int64_t limit;
int64_t offset;
} SLimit;
typedef struct SOrder {
uint32_t order;
int32_t orderColId;
} SOrder;
typedef struct SCond {
uint64_t uid;
int32_t len; // length of tag query condition data
......@@ -120,12 +66,6 @@ typedef struct STagCond {
typedef struct STableMetaInfo {
STableMeta *pTableMeta; // table meta, cached in client side and acquired by name
SVgroupsInfo *vgroupList;
/*
* 1. keep the vgroup index during the multi-vnode super table projection query
* 2. keep the vgroup index for multi-vnode insertion
*/
int32_t vgroupIndex;
SName name;
char aliasName[TSDB_TABLE_NAME_LEN]; // alias name of table specified in query sql
SArray *tagColList; // SArray<SColumn*>, involved tag columns
......@@ -137,11 +77,11 @@ typedef struct SQueryStmtInfo {
STimeWindow window; // the whole query time window
SInterval interval; // tumble time window
SSessionWindow sessionWindow; // session time window
SStateWindow stateWindow; // state window query
SGroupbyExpr groupbyExpr; // groupby tags info
SArray * colList; // SArray<SColumn*>
SFieldInfo fieldsInfo;
SArray * exprList; // SArray<SExprInfo*>
SArray * exprList1; // final exprlist in case of arithmetic expression exists
SArray** exprList; // SArray<SExprInfo*>
SLimit limit;
SLimit slimit;
STagCond tagCond;
......@@ -172,6 +112,7 @@ typedef struct SQueryStmtInfo {
struct SQueryStmtInfo *pDownstream;
int32_t havingFieldNum;
SMultiFunctionsDesc info;
int32_t exprListLevelIndex;
} SQueryStmtInfo;
typedef struct SColumnIndex {
......@@ -225,14 +166,23 @@ void assignExprInfo(SExprInfo* dst, const SExprInfo* src);
void columnListCopy(SArray* dst, const SArray* src, uint64_t uid);
void columnListDestroy(SArray* pColumnList);
void dropAllExprInfo(SArray* pExprInfo);
SExprInfo* createExprInfo(STableMetaInfo* pTableMetaInfo, int16_t functionId, SColumnIndex* pColIndex, struct tExprNode* pParamExpr, SSchema* pResSchema, int16_t interSize);
void dropAllExprInfo(SArray** pExprInfo, int32_t numOfLevel);
typedef struct SSourceParam {
SArray *pExprNodeList; //Array<struct tExprNode*>
SArray *pColumnList; //Array<struct SColumn>
int32_t num;
} SSourceParam;
SExprInfo* createExprInfo(STableMetaInfo* pTableMetaInfo, const char* funcName, SSourceParam* pSource, SSchema* pResSchema, int16_t interSize);
int32_t copyExprInfoList(SArray* dst, const SArray* src, uint64_t uid, bool deepcopy);
STableMetaInfo* getMetaInfo(SQueryStmtInfo* pQueryInfo, int32_t tableIndex);
SSchema *getOneColumnSchema(const STableMeta* pTableMeta, int32_t colIndex);
SSchema createSchema(uint8_t type, int16_t bytes, int16_t colId, const char* name);
int32_t getNewResColId();
void addIntoSourceParam(SSourceParam* pSourceParam, tExprNode* pNode, SColumn* pColumn);
#ifdef __cplusplus
}
......
include/os/os.h
浏览文件 @ 155b865a
......@@ -45,6 +45,8 @@ extern "C" {
#include <float.h>
#include <math.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include "osAtomic.h"
#include "osDef.h"
......
include/util/tcompression.h
浏览文件 @ 155b865a
......@@ -20,7 +20,7 @@
extern "C" {
#endif
#include "tdef.h"
#include "taos.h"
#include "tutil.h"
#define COMP_OVERFLOW_BYTES 2
......
include/util/tdef.h
浏览文件 @ 155b865a
......@@ -134,6 +134,15 @@ do { \
#define TSDB_BINARY_OP_REMAINDER 4004
#define TSDB_BINARY_OP_CONCAT 4005
#define FUNCTION_CEIL 4500
#define FUNCTION_FLOOR 4501
#define FUNCTION_ABS 4502
#define FUNCTION_ROUND 4503
#define FUNCTION_LENGTH 4800
#define FUNCTION_CONCAT 4801
#define FUNCTION_LTRIM 4802
#define FUNCTION_RTRIM 4803
#define IS_RELATION_OPTR(op) (((op) >= TSDB_RELATION_LESS) && ((op) < TSDB_RELATION_IN))
#define IS_ARITHMETIC_OPTR(op) (((op) >= TSDB_BINARY_OP_ADD) && ((op) <= TSDB_BINARY_OP_REMAINDER))
......@@ -311,10 +320,6 @@ do { \
#define TSDB_QUERY_TYPE_NON_TYPE 0x00u // none type
#define TSDB_QUERY_TYPE_FREE_RESOURCE 0x01u // free qhandle at vnode
#define TSDB_UDF_TYPE_SCALAR 1
#define TSDB_UDF_TYPE_AGGREGATE 2
/*
* 1. ordinary sub query for select * from super_table
* 2. all sqlobj generated by createSubqueryObj with this flag
......@@ -367,21 +372,6 @@ do { \
#define TSDB_MAX_DISKS_PER_TIER 16
#define TSDB_MAX_DISKS (TSDB_MAX_TIERS * TSDB_MAX_DISKS_PER_TIER)
#define TSDB_DATA_TYPE_NULL 0 // 1 bytes
#define TSDB_DATA_TYPE_BOOL 1 // 1 bytes
#define TSDB_DATA_TYPE_TINYINT 2 // 1 byte
#define TSDB_DATA_TYPE_SMALLINT 3 // 2 bytes
#define TSDB_DATA_TYPE_INT 4 // 4 bytes
#define TSDB_DATA_TYPE_BIGINT 5 // 8 bytes
#define TSDB_DATA_TYPE_FLOAT 6 // 4 bytes
#define TSDB_DATA_TYPE_DOUBLE 7 // 8 bytes
#define TSDB_DATA_TYPE_BINARY 8 // string
#define TSDB_DATA_TYPE_TIMESTAMP 9 // 8 bytes
#define TSDB_DATA_TYPE_NCHAR 10 // unicode string
#define TSDB_DATA_TYPE_UTINYINT 11 // 1 byte
#define TSDB_DATA_TYPE_USMALLINT 12 // 2 bytes
#define TSDB_DATA_TYPE_UINT 13 // 4 bytes
#define TSDB_DATA_TYPE_UBIGINT 14 // 8 bytes
enum { TRANS_STAT_INIT = 0, TRANS_STAT_EXECUTING, TRANS_STAT_EXECUTED, TRANS_STAT_ROLLBACKING, TRANS_STAT_ROLLBACKED };
enum { TRANS_OPER_INIT = 0, TRANS_OPER_EXECUTE, TRANS_OPER_ROLLBACK };
......
include/util/tskiplist.h
浏览文件 @ 155b865a
......@@ -21,7 +21,7 @@ extern "C" {
#endif
#include "os.h"
//#include "tdef.h"
#include "taos.h"
#include "tarray.h"
#include "tfunctional.h"
......
source/libs/executor/CMakeLists.txt
浏览文件 @ 155b865a
......@@ -8,5 +8,5 @@ target_include_directories(
target_link_libraries(
executor
PRIVATE os util common
PRIVATE os util common function parser
)
\ No newline at end of file
src/query/inc/qUtil.h → source/libs/executor/inc/executil.h
浏览文件 @ 155b865a
......@@ -15,6 +15,8 @@
#ifndef TDENGINE_QUERYUTIL_H
#define TDENGINE_QUERYUTIL_H
#include "common.h"
#include "tpagedfile.h"
#include "tbuffer.h"
#define SET_RES_WINDOW_KEY(_k, _ori, _len, _uid) \
......@@ -40,42 +42,92 @@
#define curTimeWindowIndex(_winres) ((_winres)->curIndex)
int32_t getOutputInterResultBufSize(SQueryAttr* pQueryAttr);
size_t getResultRowSize(SQueryRuntimeEnv* pRuntimeEnv);
struct SColumnFilterElem;
typedef bool (*__filter_func_t)(struct SColumnFilterElem* pFilter, const char* val1, const char* val2, int16_t type);
typedef struct SGroupResInfo {
int32_t totalGroup;
int32_t currentGroup;
int32_t index;
SArray* pRows; // SArray<SResultRow*>
bool ordered;
int32_t position;
} SGroupResInfo;
typedef struct SResultRow {
int32_t pageId; // pageId & rowId is the position of current result in disk-based output buffer
int32_t offset:29; // row index in buffer page
bool startInterp; // the time window start timestamp has done the interpolation already.
bool endInterp; // the time window end timestamp has done the interpolation already.
bool closed; // this result status: closed or opened
uint32_t numOfRows; // number of rows of current time window
struct SResultRowEntryInfo* pEntryInfo; // For each result column, there is a resultInfo
STimeWindow win;
char *key; // start key of current result row
} SResultRow;
typedef struct SResultRowInfo {
SResultRow** pResult; // result list
int16_t type:8; // data type for hash key
int32_t size:24; // number of result set
int32_t capacity; // max capacity
int32_t curPos; // current active result row index of pResult list
} SResultRowInfo;
typedef struct SResultRowPool {
int32_t elemSize;
int32_t blockSize;
int32_t numOfElemPerBlock;
struct {
int32_t blockIndex;
int32_t pos;
} position;
SArray* pData; // SArray<void*>
} SResultRowPool;
struct SQueryAttr;
struct SQueryRuntimeEnv;
struct SUdfInfo;
int32_t getOutputInterResultBufSize(struct SQueryAttr* pQueryAttr);
size_t getResultRowSize(struct SQueryRuntimeEnv* pRuntimeEnv);
int32_t initResultRowInfo(SResultRowInfo* pResultRowInfo, int32_t size, int16_t type);
void cleanupResultRowInfo(SResultRowInfo* pResultRowInfo);
void resetResultRowInfo(SQueryRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo);
void resetResultRowInfo(struct SQueryRuntimeEnv* pRuntimeEnv, SResultRowInfo* pResultRowInfo);
int32_t numOfClosedResultRows(SResultRowInfo* pResultRowInfo);
void closeAllResultRows(SResultRowInfo* pResultRowInfo);
int32_t initResultRow(SResultRow *pResultRow);
void closeResultRow(SResultRowInfo* pResultRowInfo, int32_t slot);
bool isResultRowClosed(SResultRowInfo *pResultRowInfo, int32_t slot);
void clearResultRow(SQueryRuntimeEnv* pRuntimeEnv, SResultRow* pResultRow, int16_t type);
void clearResultRow(struct SQueryRuntimeEnv* pRuntimeEnv, SResultRow* pResultRow, int16_t type);
SResultRowCellInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset);
struct SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset);
void* destroyQueryFuncExpr(SExprInfo* pExprInfo, int32_t numOfExpr);
void* freeColumnInfo(SColumnInfo* pColumnInfo, int32_t numOfCols);
int32_t getRowNumForMultioutput(SQueryAttr* pQueryAttr, bool topBottomQuery, bool stable);
int32_t getRowNumForMultioutput(struct SQueryAttr* pQueryAttr, bool topBottomQuery, bool stable);
static FORCE_INLINE SResultRow *getResultRow(SResultRowInfo *pResultRowInfo, int32_t slot) {
assert(pResultRowInfo != NULL && slot >= 0 && slot < pResultRowInfo->size);
return pResultRowInfo->pResult[slot];
}
static FORCE_INLINE char* getPosInResultPage(SQueryAttr* pQueryAttr, tFilePage* page, int32_t rowOffset,
static FORCE_INLINE char* getPosInResultPage(struct SQueryAttr* pQueryAttr, SFilePage* page, int32_t rowOffset,
int32_t offset) {
assert(rowOffset >= 0 && pQueryAttr != NULL);
int32_t numOfRows = (int32_t)getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery);
return ((char *)page->data) + rowOffset + offset * numOfRows;
// int32_t numOfRows = (int32_t)getRowNumForMultioutput(pQueryAttr, pQueryAttr->topBotQuery, pQueryAttr->stableQuery);
// return ((char *)page->data) + rowOffset + offset * numOfRows;
}
bool isNullOperator(SColumnFilterElem *pFilter, const char* minval, const char* maxval, int16_t type);
bool notNullOperator(SColumnFilterElem *pFilter, const char* minval, const char* maxval, int16_t type);
//bool isNullOperator(SColumnFilterElem *pFilter, const char* minval, const char* maxval, int16_t type);
//bool notNullOperator(SColumnFilterElem *pFilter, const char* minval, const char* maxval, int16_t type);
__filter_func_t getFilterOperator(int32_t lowerOptr, int32_t upperOptr);
......@@ -103,8 +155,8 @@ bool hasRemainData(SGroupResInfo* pGroupResInfo);
bool incNextGroup(SGroupResInfo* pGroupResInfo);
int32_t getNumOfTotalRes(SGroupResInfo* pGroupResInfo);
int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, SQueryRuntimeEnv *pRuntimeEnv, int32_t* offset);
int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, struct SQueryRuntimeEnv *pRuntimeEnv, int32_t* offset);
int32_t initUdfInfo(SUdfInfo* pUdfInfo);
int32_t initUdfInfo(struct SUdfInfo* pUdfInfo);
#endif // TDENGINE_QUERYUTIL_H
source/libs/executor/inc/executorimpl.h 0 → 100644
浏览文件 @ 155b865a
此差异已折叠。
src/query/inc/qFilter.h → source/libs/executor/inc/tfilter.h
浏览文件 @ 155b865a
......@@ -20,9 +20,9 @@
extern "C" {
#endif
#include "texpr.h"
#include "hash.h"
#include "thash.h"
#include "tname.h"
#include "function.h"
#define FILTER_DEFAULT_GROUP_SIZE 4
#define FILTER_DEFAULT_UNIT_SIZE 4
......@@ -105,7 +105,7 @@ typedef struct SFilterColRange {
typedef bool (*rangeCompFunc) (const void *, const void *, const void *, const void *, __compar_fn_t);
typedef int32_t(*filter_desc_compare_func)(const void *, const void *);
typedef bool(*filter_exec_func)(void *, int32_t, int8_t**, SDataStatis *, int16_t);
typedef bool(*filter_exec_func)(void *, int32_t, int8_t**, SColumnDataAgg *, int16_t);
typedef struct SFilterRangeCompare {
int64_t s;
......@@ -324,13 +324,13 @@ typedef struct SFilterInfo {
extern int32_t filterInitFromTree(tExprNode* tree, SFilterInfo **pinfo, uint32_t options);
extern bool filterExecute(SFilterInfo *info, int32_t numOfRows, int8_t** p, SDataStatis *statis, int16_t numOfCols);
extern bool filterExecute(SFilterInfo *info, int32_t numOfRows, int8_t** p, SColumnDataAgg *statis, int16_t numOfCols);
extern int32_t filterSetColFieldData(SFilterInfo *info, int32_t numOfCols, SArray* pDataBlock);
extern int32_t filterGetTimeRange(SFilterInfo *info, STimeWindow *win);
extern int32_t filterConverNcharColumns(SFilterInfo* pFilterInfo, int32_t rows, bool *gotNchar);
extern int32_t filterFreeNcharColumns(SFilterInfo* pFilterInfo);
extern void filterFreeInfo(SFilterInfo *info);
extern bool filterRangeExecute(SFilterInfo *info, SDataStatis *pDataStatis, int32_t numOfCols, int32_t numOfRows);
extern bool filterRangeExecute(SFilterInfo *info, SColumnDataAgg *pDataStatis, int32_t numOfCols, int32_t numOfRows);
#ifdef __cplusplus
}
......
src/query/src/qUtil.c → source/libs/executor/src/executil.c
浏览文件 @ 155b865a
......@@ -15,11 +15,11 @@
#include "os.h"
#include "taosmsg.h"
#include "hash.h"
#include "thash.h"
#include "qExecutor.h"
#include "qUtil.h"
#include "queryLog.h"
#include "executil.h"
#include "executorimpl.h"
//#include "queryLog.h"
#include "tbuffer.h"
#include "tcompression.h"
#include "tlosertree.h"
......@@ -33,9 +33,9 @@ typedef struct SCompSupporter {
int32_t getRowNumForMultioutput(SQueryAttr* pQueryAttr, bool topBottomQuery, bool stable) {
if (pQueryAttr && (!stable)) {
for (int16_t i = 0; i < pQueryAttr->numOfOutput; ++i) {
if (pQueryAttr->pExpr1[i].base.functionId == TSDB_FUNC_TOP || pQueryAttr->pExpr1[i].base.functionId == TSDB_FUNC_BOTTOM) {
return (int32_t)pQueryAttr->pExpr1[i].base.param[0].i64;
}
// if (pQueryAttr->pExpr1[i].base. == FUNCTION_TOP || pQueryAttr->pExpr1[i].base.functionId == FUNCTION_BOTTOM) {
// return (int32_t)pQueryAttr->pExpr1[i].base.param[0].i;
// }
}
}
......@@ -143,18 +143,18 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResultRow, int16
// the result does not put into the SDiskbasedResultBuf, ignore it.
if (pResultRow->pageId >= 0) {
tFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pResultRow->pageId);
SFilePage *page = getResBufPage(pRuntimeEnv->pResultBuf, pResultRow->pageId);
int16_t offset = 0;
for (int32_t i = 0; i < pRuntimeEnv->pQueryAttr->numOfOutput; ++i) {
SResultRowCellInfo *pResultInfo = &pResultRow->pCellInfo[i];
struct SResultRowEntryInfo *pEntryInfo = NULL;//pResultRow->pEntryInfo[i];
int16_t size = pRuntimeEnv->pQueryAttr->pExpr1[i].base.resType;
int16_t size = pRuntimeEnv->pQueryAttr->pExpr1[i].base.resSchema.bytes;
char * s = getPosInResultPage(pRuntimeEnv->pQueryAttr, page, pResultRow->offset, offset);
memset(s, 0, size);
offset += size;
RESET_RESULT_INFO(pResultInfo);
cleanupResultRowEntry(pEntryInfo);
}
}
......@@ -168,14 +168,16 @@ void clearResultRow(SQueryRuntimeEnv *pRuntimeEnv, SResultRow *pResultRow, int16
}
// TODO refactor: use macro
SResultRowCellInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset) {
struct SResultRowEntryInfo* getResultCell(const SResultRow* pRow, int32_t index, int32_t* offset) {
assert(index >= 0 && offset != NULL);
return (SResultRowCellInfo*)((char*) pRow->pCellInfo + offset[index]);
// return (SResultRowEntryInfo*)((char*) pRow->pCellInfo + offset[index]);
return NULL;
}
size_t getResultRowSize(SQueryRuntimeEnv* pRuntimeEnv) {
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
return (pQueryAttr->numOfOutput * sizeof(SResultRowCellInfo)) + pQueryAttr->interBufSize + sizeof(SResultRow);
return 0;
// return (pQueryAttr->numOfOutput * sizeof(SResultRowEntryInfo)) + pQueryAttr->interBufSize + sizeof(SResultRow);
}
SResultRowPool* initResultRowPool(size_t size) {
......@@ -271,9 +273,9 @@ void interResToBinary(SBufferWriter* bw, SArray* pRes, int32_t tagLen) {
tbufWriteUint32(bw, numOfRows);
for(int32_t k = 0; k < numOfRows; ++k) {
SResPair v = *(SResPair*) taosArrayGet(p->pResult, k);
tbufWriteDouble(bw, v.avg);
tbufWriteInt64(bw, v.key);
// SResPair v = *(SResPair*) taosArrayGet(p->pResult, k);
// tbufWriteDouble(bw, v.avg);
// tbufWriteInt64(bw, v.key);
}
}
}
......@@ -301,19 +303,19 @@ SArray* interResFromBinary(const char* data, int32_t len) {
SArray* p = taosArrayInit(numOfCols, sizeof(SStddevInterResult));
for(int32_t j = 0; j < numOfCols; ++j) {
int16_t colId = tbufReadUint16(&br);
// int16_t colId = tbufReadUint16(&br);
int32_t numOfRows = tbufReadUint32(&br);
SStddevInterResult interRes = {.colId = colId, .pResult = taosArrayInit(4, sizeof(struct SResPair)),};
// SStddevInterResult interRes = {.colId = colId, .pResult = taosArrayInit(4, sizeof(struct SResPair)),};
for(int32_t k = 0; k < numOfRows; ++k) {
SResPair px = {0};
px.avg = tbufReadDouble(&br);
px.key = tbufReadInt64(&br);
taosArrayPush(interRes.pResult, &px);
// SResPair px = {0};
// px.avg = tbufReadDouble(&br);
// px.key = tbufReadInt64(&br);
//
// taosArrayPush(interRes.pResult, &px);
}
taosArrayPush(p, &interRes);
// taosArrayPush(p, &interRes);
}
char* p1 = NULL;
......@@ -395,22 +397,22 @@ static int64_t getNumOfResultWindowRes(SQueryRuntimeEnv* pRuntimeEnv, SResultRow
SQueryAttr* pQueryAttr = pRuntimeEnv->pQueryAttr;
for (int32_t j = 0; j < pQueryAttr->numOfOutput; ++j) {
int32_t functionId = pQueryAttr->pExpr1[j].base.functionId;
int32_t functionId = 0;//pQueryAttr->pExpr1[j].base.functionId;
/*
* ts, tag, tagprj function can not decide the output number of current query
* the number of output result is decided by main output
*/
if (functionId == TSDB_FUNC_TS || functionId == TSDB_FUNC_TAG || functionId == TSDB_FUNC_TAGPRJ) {
if (functionId == FUNCTION_TS || functionId == FUNCTION_TAG || functionId == FUNCTION_TAGPRJ) {
continue;
}
SResultRowCellInfo *pResultInfo = getResultCell(pResultRow, j, rowCellInfoOffset);
assert(pResultInfo != NULL);
if (pResultInfo->numOfRes > 0) {
return pResultInfo->numOfRes;
}
// SResultRowEntryInfo *pResultInfo = getResultCell(pResultRow, j, rowCellInfoOffset);
// assert(pResultInfo != NULL);
//
// if (pResultInfo->numOfRes > 0) {
// return pResultInfo->numOfRes;
// }
}
return 0;
......@@ -545,7 +547,7 @@ static UNUSED_FUNC int32_t mergeIntoGroupResultImpl(SQueryRuntimeEnv *pRuntimeEn
pTableQueryInfoList = malloc(POINTER_BYTES * size);
if (pTableQueryInfoList == NULL || posList == NULL || pGroupResInfo->pRows == NULL || pGroupResInfo->pRows == NULL) {
qError("QInfo:%"PRIu64" failed alloc memory", GET_QID(pRuntimeEnv));
// qError("QInfo:%"PRIu64" failed alloc memory", GET_QID(pRuntimeEnv));
code = TSDB_CODE_QRY_OUT_OF_MEMORY;
goto _end;
}
......@@ -617,8 +619,8 @@ static UNUSED_FUNC int32_t mergeIntoGroupResultImpl(SQueryRuntimeEnv *pRuntimeEn
int64_t endt = taosGetTimestampMs();
qDebug("QInfo:%"PRIx64" result merge completed for group:%d, elapsed time:%" PRId64 " ms", GET_QID(pRuntimeEnv),
pGroupResInfo->currentGroup, endt - startt);
// qDebug("QInfo:%"PRIx64" result merge completed for group:%d, elapsed time:%" PRId64 " ms", GET_QID(pRuntimeEnv),
// pGroupResInfo->currentGroup, endt - startt);
_end:
tfree(pTableQueryInfoList);
......@@ -639,90 +641,90 @@ int32_t mergeIntoGroupResult(SGroupResInfo* pGroupResInfo, SQueryRuntimeEnv* pRu
break;
}
qDebug("QInfo:%"PRIu64" no result in group %d, continue", GET_QID(pRuntimeEnv), pGroupResInfo->currentGroup);
// qDebug("QInfo:%"PRIu64" no result in group %d, continue", GET_QID(pRuntimeEnv), pGroupResInfo->currentGroup);
cleanupGroupResInfo(pGroupResInfo);
incNextGroup(pGroupResInfo);
}
int64_t elapsedTime = taosGetTimestampUs() - st;
qDebug("QInfo:%"PRIu64" merge res data into group, index:%d, total group:%d, elapsed time:%" PRId64 "us", GET_QID(pRuntimeEnv),
pGroupResInfo->currentGroup, pGroupResInfo->totalGroup, elapsedTime);
// int64_t elapsedTime = taosGetTimestampUs() - st;
// qDebug("QInfo:%"PRIu64" merge res data into group, index:%d, total group:%d, elapsed time:%" PRId64 "us", GET_QID(pRuntimeEnv),
// pGroupResInfo->currentGroup, pGroupResInfo->totalGroup, elapsedTime);
return TSDB_CODE_SUCCESS;
}
void blockDistInfoToBinary(STableBlockDist* pDist, struct SBufferWriter* bw) {
tbufWriteUint32(bw, pDist->numOfTables);
tbufWriteUint16(bw, pDist->numOfFiles);
tbufWriteUint64(bw, pDist->totalSize);
tbufWriteUint64(bw, pDist->totalRows);
tbufWriteInt32(bw, pDist->maxRows);
tbufWriteInt32(bw, pDist->minRows);
tbufWriteUint32(bw, pDist->numOfRowsInMemTable);
tbufWriteUint32(bw, pDist->numOfSmallBlocks);
tbufWriteUint64(bw, taosArrayGetSize(pDist->dataBlockInfos));
// compress the binary string
char* p = TARRAY_GET_START(pDist->dataBlockInfos);
// compress extra bytes
size_t x = taosArrayGetSize(pDist->dataBlockInfos) * pDist->dataBlockInfos->elemSize;
char* tmp = malloc(x + 2);
bool comp = false;
int32_t len = tsCompressString(p, (int32_t)x, 1, tmp, (int32_t)x, ONE_STAGE_COMP, NULL, 0);
if (len == -1 || len >= x) { // compress failed, do not compress this binary data
comp = false;
len = (int32_t)x;
} else {
comp = true;
}
tbufWriteUint8(bw, comp);
tbufWriteUint32(bw, len);
if (comp) {
tbufWriteBinary(bw, tmp, len);
} else {
tbufWriteBinary(bw, p, len);
}
tfree(tmp);
}
void blockDistInfoFromBinary(const char* data, int32_t len, STableBlockDist* pDist) {
SBufferReader br = tbufInitReader(data, len, false);
pDist->numOfTables = tbufReadUint32(&br);
pDist->numOfFiles = tbufReadUint16(&br);
pDist->totalSize = tbufReadUint64(&br);
pDist->totalRows = tbufReadUint64(&br);
pDist->maxRows = tbufReadInt32(&br);
pDist->minRows = tbufReadInt32(&br);
pDist->numOfRowsInMemTable = tbufReadUint32(&br);
pDist->numOfSmallBlocks = tbufReadUint32(&br);
int64_t numSteps = tbufReadUint64(&br);
bool comp = tbufReadUint8(&br);
uint32_t compLen = tbufReadUint32(&br);
size_t originalLen = (size_t) (numSteps *sizeof(SFileBlockInfo));
char* outputBuf = NULL;
if (comp) {
outputBuf = malloc(originalLen);
size_t actualLen = compLen;
const char* compStr = tbufReadBinary(&br, &actualLen);
int32_t orignalLen = tsDecompressString(compStr, compLen, 1, outputBuf,
(int32_t)originalLen , ONE_STAGE_COMP, NULL, 0);
assert(orignalLen == numSteps *sizeof(SFileBlockInfo));
} else {
outputBuf = (char*) tbufReadBinary(&br, &originalLen);
}
pDist->dataBlockInfos = taosArrayFromList(outputBuf, (uint32_t)numSteps, sizeof(SFileBlockInfo));
if (comp) {
tfree(outputBuf);
}
}
//void blockDistInfoToBinary(STableBlockDist* pDist, struct SBufferWriter* bw) {
// tbufWriteUint32(bw, pDist->numOfTables);
// tbufWriteUint16(bw, pDist->numOfFiles);
// tbufWriteUint64(bw, pDist->totalSize);
// tbufWriteUint64(bw, pDist->totalRows);
// tbufWriteInt32(bw, pDist->maxRows);
// tbufWriteInt32(bw, pDist->minRows);
// tbufWriteUint32(bw, pDist->numOfRowsInMemTable);
// tbufWriteUint32(bw, pDist->numOfSmallBlocks);
// tbufWriteUint64(bw, taosArrayGetSize(pDist->dataBlockInfos));
//
// // compress the binary string
// char* p = TARRAY_GET_START(pDist->dataBlockInfos);
//
// // compress extra bytes
// size_t x = taosArrayGetSize(pDist->dataBlockInfos) * pDist->dataBlockInfos->elemSize;
// char* tmp = malloc(x + 2);
//
// bool comp = false;
// int32_t len = tsCompressString(p, (int32_t)x, 1, tmp, (int32_t)x, ONE_STAGE_COMP, NULL, 0);
// if (len == -1 || len >= x) { // compress failed, do not compress this binary data
// comp = false;
// len = (int32_t)x;
// } else {
// comp = true;
// }
//
// tbufWriteUint8(bw, comp);
// tbufWriteUint32(bw, len);
// if (comp) {
// tbufWriteBinary(bw, tmp, len);
// } else {
// tbufWriteBinary(bw, p, len);
// }
// tfree(tmp);
//}
//void blockDistInfoFromBinary(const char* data, int32_t len, STableBlockDist* pDist) {
// SBufferReader br = tbufInitReader(data, len, false);
//
// pDist->numOfTables = tbufReadUint32(&br);
// pDist->numOfFiles = tbufReadUint16(&br);
// pDist->totalSize = tbufReadUint64(&br);
// pDist->totalRows = tbufReadUint64(&br);
// pDist->maxRows = tbufReadInt32(&br);
// pDist->minRows = tbufReadInt32(&br);
// pDist->numOfRowsInMemTable = tbufReadUint32(&br);
// pDist->numOfSmallBlocks = tbufReadUint32(&br);
// int64_t numSteps = tbufReadUint64(&br);
//
// bool comp = tbufReadUint8(&br);
// uint32_t compLen = tbufReadUint32(&br);
//
// size_t originalLen = (size_t) (numSteps *sizeof(SFileBlockInfo));
//
// char* outputBuf = NULL;
// if (comp) {
// outputBuf = malloc(originalLen);
//
// size_t actualLen = compLen;
// const char* compStr = tbufReadBinary(&br, &actualLen);
//
// int32_t orignalLen = tsDecompressString(compStr, compLen, 1, outputBuf,
// (int32_t)originalLen , ONE_STAGE_COMP, NULL, 0);
// assert(orignalLen == numSteps *sizeof(SFileBlockInfo));
// } else {
// outputBuf = (char*) tbufReadBinary(&br, &originalLen);
// }
//
// pDist->dataBlockInfos = taosArrayFromList(outputBuf, (uint32_t)numSteps, sizeof(SFileBlockInfo));
// if (comp) {
// tfree(outputBuf);
// }
//}
source/libs/executor/src/executorimpl.c 0 → 100644
浏览文件 @ 155b865a
此差异已折叠。
source/libs/executor/src/tarithoperator.c 已删除 100644 → 0
浏览文件 @ a1893dd8
此差异已折叠。
source/libs/executor/src/tfilter.c 0 → 100644
浏览文件 @ 155b865a
此差异已折叠。
source/libs/function/inc/taggfunction.h
浏览文件 @ 155b865a
......@@ -28,14 +28,14 @@ extern "C" {
#include "function.h"
#include "tudf.h"
extern SAggFunctionInfo aggFunc[34];
extern SAggFunctionInfo aggFunc[35];
typedef struct SResultRowCellInfo {
typedef struct SResultRowEntryInfo {
int8_t hasResult; // result generated, not NULL value
bool initialized; // output buffer has been initialized
bool complete; // query has completed
uint32_t numOfRes; // num of output result in current buffer
} SResultRowCellInfo;
} SResultRowEntryInfo;
#define FUNCSTATE_SO 0x0u
#define FUNCSTATE_MO 0x1u // dynamic number of output, not multinumber of output e.g., TOP/BOTTOM
......@@ -52,54 +52,24 @@ typedef struct SResultRowCellInfo {
#define DATA_SET_FLAG ',' // to denote the output area has data, not null value
#define DATA_SET_FLAG_SIZE sizeof(DATA_SET_FLAG)
#define QUERY_ASC_FORWARD_STEP 1
#define QUERY_DESC_FORWARD_STEP -1
#define GET_FORWARD_DIRECTION_FACTOR(ord) (((ord) == TSDB_ORDER_ASC) ? QUERY_ASC_FORWARD_STEP : QUERY_DESC_FORWARD_STEP)
#define TOP_BOTTOM_QUERY_LIMIT 100
enum {
MASTER_SCAN = 0x0u,
REVERSE_SCAN = 0x1u,
REPEAT_SCAN = 0x2u, //repeat scan belongs to the master scan
MERGE_STAGE = 0x20u,
};
#define QUERY_IS_STABLE_QUERY(type) (((type)&TSDB_QUERY_TYPE_STABLE_QUERY) != 0)
#define QUERY_IS_JOIN_QUERY(type) (TSDB_QUERY_HAS_TYPE(type, TSDB_QUERY_TYPE_JOIN_QUERY))
#define QUERY_IS_PROJECTION_QUERY(type) (((type)&TSDB_QUERY_TYPE_PROJECTION_QUERY) != 0)
#define QUERY_IS_FREE_RESOURCE(type) (((type)&TSDB_QUERY_TYPE_FREE_RESOURCE) != 0)
typedef struct SArithmeticSupport {
struct SExprInfo *pExprInfo;
int32_t numOfCols;
SColumnInfo *colList;
void *exprList; // client side used
int32_t offset;
char** data;
} SArithmeticSupport;
typedef struct SInterpInfoDetail {
TSKEY ts; // interp specified timestamp
int8_t type;
int8_t primaryCol;
} SInterpInfoDetail;
#define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowCellInfo)))
#define GET_RES_INFO(ctx) ((ctx)->resultInfo)
#define GET_ROWCELL_INTERBUF(_c) ((void*) ((char*)(_c) + sizeof(SResultRowEntryInfo)))
#define IS_STREAM_QUERY_VALID(x) (((x)&TSDB_FUNCSTATE_STREAM) != 0)
#define IS_MULTIOUTPUT(x) (((x)&TSDB_FUNCSTATE_MO) != 0)
// determine the real data need to calculated the result
enum {
BLK_DATA_NO_NEEDED = 0x0,
BLK_DATA_STATIS_NEEDED = 0x1,
BLK_DATA_ALL_NEEDED = 0x3,
BLK_DATA_DISCARD = 0x4, // discard current data block since it is not qualified for filter
};
typedef struct STwaInfo {
int8_t hasResult; // flag to denote has value
double dOutput;
......@@ -115,12 +85,7 @@ bool topbot_datablock_filter(SQLFunctionCtx *pCtx, const char *minval, const cha
* the numOfRes should be kept, since it may be used later
* and allow the ResultInfo to be re initialized
*/
#define RESET_RESULT_INFO(_r) \
do { \
(_r)->initialized = false; \
} while (0)
static FORCE_INLINE void initResultInfo(SResultRowCellInfo *pResInfo, int32_t bufLen) {
static FORCE_INLINE void initResultRowEntry(SResultRowEntryInfo *pResInfo, int32_t bufLen) {
pResInfo->initialized = true; // the this struct has been initialized flag
pResInfo->complete = false;
......
source/libs/executor/inc/tarithoperator.h → source/libs/function/inc/tbinoperator.h
浏览文件 @ 155b865a
......@@ -13,20 +13,21 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_COMMON_QARITHMETICOPERATOR_H_
#define _TD_COMMON_QARITHMETICOPERATOR_H_
#ifndef _TD_COMMON_BIN_SCALAR_OPERATOR_H_
#define _TD_COMMON_BIN_SCALAR_OPERATOR_H_
#ifdef __cplusplus
extern "C" {
#endif
typedef void (*_arithmetic_operator_fn_t)(void *left, int32_t numLeft, int32_t leftType, void *right, int32_t numRight,
int32_t rightType, void *output, int32_t order);
#include "tscalarfunction.h"
_arithmetic_operator_fn_t getArithmeticOperatorFn(int32_t arithmeticOptr);
typedef void (*_bin_scalar_fn_t)(SScalarFuncParam* pLeft, SScalarFuncParam* pRight, void *output, int32_t order);
_bin_scalar_fn_t getBinScalarOperatorFn(int32_t binOperator);
bool isBinaryStringOp(int32_t op);
#ifdef __cplusplus
}
#endif
#endif /*_TD_COMMON_QARITHMETICOPERATOR_H_*/
#endif /*_TD_COMMON_BIN_SCALAR_OPERATOR_H_*/
source/libs/function/inc/texpr.h
浏览文件 @ 155b865a
......@@ -60,14 +60,10 @@ typedef struct SExprTraverseSupp {
void *pExtInfo;
} SExprTraverseSupp;
tExprNode* exprTreeFromBinary(const void* data, size_t size);
tExprNode* exprTreeFromTableName(const char* tbnameCond);
bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp *param);
void arithmeticTreeTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
char *(*cb)(void *, const char*, int32_t));
void buildFilterSetFromBinary(void **q, const char *buf, int32_t len);
#ifdef __cplusplus
......
source/libs/function/inc/tfill.h
浏览文件 @ 155b865a
......@@ -25,7 +25,7 @@ extern "C" {
struct SSDataBlock;
typedef struct {
typedef struct SFillColInfo {
STColumn col; // column info
int16_t functionId; // sql function id
int16_t flag; // column flag: TAG COLUMN|NORMAL COLUMN
......@@ -64,30 +64,11 @@ typedef struct SFillInfo {
void* handle; // for debug purpose
} SFillInfo;
typedef struct SPoint {
int64_t key;
void * val;
} SPoint;
SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
SFillColInfo* pFillCol, void* handle);
void taosResetFillInfo(SFillInfo* pFillInfo, TSKEY startTimestamp);
void* taosDestroyFillInfo(SFillInfo *pFillInfo);
void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey);
void taosFillSetInputDataBlock(SFillInfo* pFillInfo, const struct SSDataBlock* pInput);
int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t maxNumOfRows);
bool taosFillHasMoreResults(SFillInfo* pFillInfo);
int64_t getNumOfResultsAfterFillGap(SFillInfo* pFillInfo, int64_t ekey, int32_t maxNumOfRows);
int32_t taosGetLinearInterpolationVal(SPoint* point, int32_t outputType, SPoint* point1, SPoint* point2, int32_t inputType);
int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, void** output, int32_t capacity);
#ifdef __cplusplus
}
......
source/libs/function/inc/tscalarfunction.h
浏览文件 @ 155b865a
......@@ -21,14 +21,28 @@ extern "C" {
#include "function.h"
extern struct SScalarFunctionInfo scalarFunc[1];
#define FUNCTION_CEIL 38
#define FUNCTION_FLOOR 39
#define FUNCTION_ROUND 40
#define FUNCTION_MAVG 41
#define FUNCTION_CSUM 42
#define FUNCCTION_CONCAT 43
typedef struct SScalarFuncParam {
void* data;
int32_t num;
int32_t type;
int32_t bytes;
} SScalarFuncParam;
typedef struct SScalarFunctionSupport {
struct SExprInfo *pExprInfo;
int32_t numOfCols;
SColumnInfo *colList;
void *exprList; // client side used
int32_t offset;
char** data;
} SScalarFunctionSupport;
extern struct SScalarFunctionInfo scalarFunc[8];
int32_t evaluateExprNodeTree(tExprNode* pExprs, int32_t numOfRows, SScalarFuncParam* pOutput,
void* param, char* (*getSourceDataBlock)(void*, const char*, int32_t));
#ifdef __cplusplus
}
......
src/query/inc/qScript.h → source/libs/function/inc/tscript.h
浏览文件 @ 155b865a
......@@ -16,6 +16,7 @@
#ifndef TDENGINE_QSCRIPT_H
#define TDENGINE_QSCRIPT_H
#if 0
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>
......@@ -23,7 +24,7 @@
#include "tutil.h"
#include "hash.h"
#include "tlist.h"
#include "qUdf.h"
#include "tudf.h"
#define MAX_FUNC_NAME 64
......@@ -78,5 +79,6 @@ void destroyScriptCtx(void *pScriptCtx);
int32_t scriptEnvPoolInit();
void scriptEnvPoolCleanup();
bool isValidScript(char *script, int32_t len);
#endif
#endif //TDENGINE_QSCRIPT_H
source/libs/function/inc/tudf.h
浏览文件 @ 155b865a
......@@ -16,6 +16,13 @@
#ifndef TDENGINE_TUDF_H
#define TDENGINE_TUDF_H
#ifdef __cplusplus
extern "C" {
#endif
#include "os.h"
#include "taoserror.h"
enum {
TSDB_UDF_FUNC_NORMAL = 0,
TSDB_UDF_FUNC_INIT,
......@@ -76,4 +83,8 @@ typedef void (*udfFinalizeFunc)(char* dataOutput, char* interBuf, int32_t* numOf
typedef void (*udfMergeFunc)(char* data, int32_t numOfRows, char* dataOutput, int32_t* numOfOutput, SUdfInit* buf);
typedef void (*udfDestroyFunc)(SUdfInit* buf);
#ifdef __cplusplus
}
#endif
#endif // TDENGINE_TUDF_H
source/libs/function/inc/tunaryoperator.h 0 → 100644
浏览文件 @ 155b865a
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef _TD_COMMON_UNARY_SCALAR_OPERATOR_H_
#define _TD_COMMON_UNARY_SCALAR_OPERATOR_H_
#ifdef __cplusplus
extern "C" {
#endif
#include "tscalarfunction.h"
typedef void (*_unary_scalar_fn_t)(SScalarFuncParam *pLeft, SScalarFuncParam* pOutput);
_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t binOperator);
#ifdef __cplusplus
}
#endif
#endif /*_TD_COMMON_BIN_SCALAR_OPERATOR_H_*/
source/libs/function/src/tbinoperator.c 0 → 100644
浏览文件 @ 155b865a
此差异已折叠。
source/libs/function/src/texpr.c
浏览文件 @ 155b865a
......@@ -21,11 +21,8 @@
#include "tarray.h"
#include "tbuffer.h"
#include "tcompare.h"
#include "tname.h"
#include "thash.h"
#include "tskiplist.h"
#include "texpr.h"
//#include "tarithoperator.h"
#include "tvariant.h"
//static uint8_t UNUSED_FUNC isQueryOnPrimaryKey(const char *primaryColumnName, const tExprNode *pLeft, const tExprNode *pRight) {
......@@ -41,71 +38,6 @@
// }
//}
static void reverseCopy(char* dest, const char* src, int16_t type, int32_t numOfRows) {
switch(type) {
case TSDB_DATA_TYPE_TINYINT:
case TSDB_DATA_TYPE_UTINYINT:{
int8_t* p = (int8_t*) dest;
int8_t* pSrc = (int8_t*) src;
for(int32_t i = 0; i < numOfRows; ++i) {
p[i] = pSrc[numOfRows - i - 1];
}
return;
}
case TSDB_DATA_TYPE_SMALLINT:
case TSDB_DATA_TYPE_USMALLINT:{
int16_t* p = (int16_t*) dest;
int16_t* pSrc = (int16_t*) src;
for(int32_t i = 0; i < numOfRows; ++i) {
p[i] = pSrc[numOfRows - i - 1];
}
return;
}
case TSDB_DATA_TYPE_INT:
case TSDB_DATA_TYPE_UINT: {
int32_t* p = (int32_t*) dest;
int32_t* pSrc = (int32_t*) src;
for(int32_t i = 0; i < numOfRows; ++i) {
p[i] = pSrc[numOfRows - i - 1];
}
return;
}
case TSDB_DATA_TYPE_BIGINT:
case TSDB_DATA_TYPE_UBIGINT: {
int64_t* p = (int64_t*) dest;
int64_t* pSrc = (int64_t*) src;
for(int32_t i = 0; i < numOfRows; ++i) {
p[i] = pSrc[numOfRows - i - 1];
}
return;
}
case TSDB_DATA_TYPE_FLOAT: {
float* p = (float*) dest;
float* pSrc = (float*) src;
for(int32_t i = 0; i < numOfRows; ++i) {
p[i] = pSrc[numOfRows - i - 1];
}
return;
}
case TSDB_DATA_TYPE_DOUBLE: {
double* p = (double*) dest;
double* pSrc = (double*) src;
for(int32_t i = 0; i < numOfRows; ++i) {
p[i] = pSrc[numOfRows - i - 1];
}
return;
}
default: assert(0);
}
}
static void doExprTreeDestroy(tExprNode **pExpr, void (*fp)(void *));
void tExprTreeDestroy(tExprNode *pNode, void (*fp)(void *)) {
......@@ -182,114 +114,7 @@ bool exprTreeApplyFilter(tExprNode *pExpr, const void *pItem, SExprTraverseSupp
return param->nodeFilterFn(pItem, pExpr->_node.info);
}
void arithmeticTreeTraverse(tExprNode *pExprs, int32_t numOfRows, char *pOutput, void *param, int32_t order,
char *(*getSourceDataBlock)(void *, const char*, int32_t)) {
if (pExprs == NULL) {
return;
}
#if 0
tExprNode *pLeft = pExprs->_node.pLeft;
tExprNode *pRight = pExprs->_node.pRight;
/* the left output has result from the left child syntax tree */
char *pLeftOutput = (char*)malloc(sizeof(int64_t) * numOfRows);
if (pLeft->nodeType == TEXPR_BINARYEXPR_NODE) {
arithmeticTreeTraverse(pLeft, numOfRows, pLeftOutput, param, order, getSourceDataBlock);
}
// the right output has result from the right child syntax tree
char *pRightOutput = malloc(sizeof(int64_t) * numOfRows);
char *pdata = malloc(sizeof(int64_t) * numOfRows);
if (pRight->nodeType == TEXPR_BINARYEXPR_NODE) {
arithmeticTreeTraverse(pRight, numOfRows, pRightOutput, param, order, getSourceDataBlock);
}
if (pLeft->nodeType == TEXPR_BINARYEXPR_NODE) {
if (pRight->nodeType == TEXPR_BINARYEXPR_NODE) {
/*
* exprLeft + exprRight
* the type of returned value of one expression is always double float precious
*/
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
OperatorFn(pLeftOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pRightOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pOutput, TSDB_ORDER_ASC);
} else if (pRight->nodeType == TEXPR_COL_NODE) { // exprLeft + columnRight
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
// set input buffer
char *pInputData = getSourceDataBlock(param, pRight->pSchema->name, pRight->pSchema->colId);
if (order == TSDB_ORDER_DESC) {
reverseCopy(pdata, pInputData, pRight->pSchema->type, numOfRows);
OperatorFn(pLeftOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pdata, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
} else {
OperatorFn(pLeftOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pInputData, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
}
} else if (pRight->nodeType == TEXPR_VALUE_NODE) { // exprLeft + 12
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
OperatorFn(pLeftOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
}
} else if (pLeft->nodeType == TEXPR_COL_NODE) {
// column data specified on left-hand-side
char *pLeftInputData = getSourceDataBlock(param, pLeft->pSchema->name, pLeft->pSchema->colId);
if (pRight->nodeType == TEXPR_BINARYEXPR_NODE) { // columnLeft + expr2
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
if (order == TSDB_ORDER_DESC) {
reverseCopy(pdata, pLeftInputData, pLeft->pSchema->type, numOfRows);
OperatorFn(pdata, numOfRows, pLeft->pSchema->type, pRightOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pOutput, TSDB_ORDER_ASC);
} else {
OperatorFn(pLeftInputData, numOfRows, pLeft->pSchema->type, pRightOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pOutput, TSDB_ORDER_ASC);
}
} else if (pRight->nodeType == TEXPR_COL_NODE) { // columnLeft + columnRight
// column data specified on right-hand-side
char *pRightInputData = getSourceDataBlock(param, pRight->pSchema->name, pRight->pSchema->colId);
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
// both columns are descending order, do not reverse the source data
OperatorFn(pLeftInputData, numOfRows, pLeft->pSchema->type, pRightInputData, numOfRows, pRight->pSchema->type, pOutput, order);
} else if (pRight->nodeType == TEXPR_VALUE_NODE) { // columnLeft + 12
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
if (order == TSDB_ORDER_DESC) {
reverseCopy(pdata, pLeftInputData, pLeft->pSchema->type, numOfRows);
OperatorFn(pdata, numOfRows, pLeft->pSchema->type, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
} else {
OperatorFn(pLeftInputData, numOfRows, pLeft->pSchema->type, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
}
}
} else {
// column data specified on left-hand-side
if (pRight->nodeType == TEXPR_BINARYEXPR_NODE) { // 12 + expr2
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, pRightOutput, numOfRows, TSDB_DATA_TYPE_DOUBLE, pOutput, TSDB_ORDER_ASC);
} else if (pRight->nodeType == TEXPR_COL_NODE) { // 12 + columnRight
// column data specified on right-hand-side
char *pRightInputData = getSourceDataBlock(param, pRight->pSchema->name, pRight->pSchema->colId);
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
if (order == TSDB_ORDER_DESC) {
reverseCopy(pdata, pRightInputData, pRight->pSchema->type, numOfRows);
OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, pdata, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
} else {
OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, pRightInputData, numOfRows, pRight->pSchema->type, pOutput, TSDB_ORDER_ASC);
}
} else if (pRight->nodeType == TEXPR_VALUE_NODE) { // 12 + 12
_arithmetic_operator_fn_t OperatorFn = getArithmeticOperatorFn(pExprs->_node.optr);
OperatorFn(&pLeft->pVal->i, 1, pLeft->pVal->nType, &pRight->pVal->i, 1, pRight->pVal->nType, pOutput, TSDB_ORDER_ASC);
}
}
tfree(pdata);
tfree(pLeftOutput);
tfree(pRightOutput);
#endif
}
static void exprTreeToBinaryImpl(SBufferWriter* bw, tExprNode* expr) {
tbufWriteUint8(bw, expr->nodeType);
......
source/libs/function/src/tfill.c
浏览文件 @ 155b865a
......@@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <function.h>
#include "os.h"
#include "taosdef.h"
......@@ -27,7 +28,6 @@
#define FILL_IS_ASC_FILL(_f) ((_f)->order == TSDB_ORDER_ASC)
#define DO_INTERPOLATION(_v1, _v2, _k1, _k2, _k) ((_v1) + ((_v2) - (_v1)) * (((double)(_k)) - ((double)(_k1))) / (((double)(_k2)) - ((double)(_k1))))
#define GET_FORWARD_DIRECTION_FACTOR(_ord) (((_ord) == TSDB_ORDER_ASC)? 1:-1)
static void setTagsValue(SFillInfo* pFillInfo, void** data, int32_t genRows) {
for(int32_t j = 0; j < pFillInfo->numOfCols; ++j) {
......@@ -340,9 +340,9 @@ static int32_t taosNumOfRemainRows(SFillInfo* pFillInfo) {
return pFillInfo->numOfRows - pFillInfo->index;
}
SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
struct SFillInfo* taosCreateFillInfo(int32_t order, TSKEY skey, int32_t numOfTags, int32_t capacity, int32_t numOfCols,
int64_t slidingTime, int8_t slidingUnit, int8_t precision, int32_t fillType,
SFillColInfo* pCol, void* handle) {
struct SFillColInfo* pCol, void* handle) {
if (fillType == TSDB_FILL_NONE) {
return NULL;
}
......@@ -522,3 +522,33 @@ int64_t taosFillResultDataBlock(SFillInfo* pFillInfo, void** output, int32_t cap
return numOfRes;
}
int64_t getFillInfoStart(struct SFillInfo *pFillInfo) {
return pFillInfo->start;
}
struct SFillColInfo* createFillColInfo(SExprInfo* pExpr, int32_t numOfOutput, const int64_t* fillVal) {
int32_t offset = 0;
struct SFillColInfo* pFillCol = calloc(numOfOutput, sizeof(SFillColInfo));
if (pFillCol == NULL) {
return NULL;
}
for(int32_t i = 0; i < numOfOutput; ++i) {
SExprInfo* pExprInfo = &pExpr[i];
pFillCol[i].col.bytes = pExprInfo->base.resSchema.bytes;
pFillCol[i].col.type = (int8_t)pExprInfo->base.resSchema.type;
pFillCol[i].col.offset = offset;
pFillCol[i].col.colId = pExprInfo->base.resSchema.colId;
pFillCol[i].tagIndex = -2;
pFillCol[i].flag = pExprInfo->base.pColumns->flag; // always be the normal column for table query
// pFillCol[i].functionId = pExprInfo->pExpr->_function.functionId;
pFillCol[i].fillVal.i = fillVal[i];
offset += pExprInfo->base.resSchema.bytes;
}
return pFillCol;
}
\ No newline at end of file
source/libs/function/src/tfunction.c
浏览文件 @ 155b865a
......@@ -6,6 +6,7 @@
#include "tscalarfunction.h"
static SHashObj* functionHashTable = NULL;
static SHashObj* udfHashTable = NULL;
static void doInitFunctionHashTable() {
int numOfEntries = tListLen(aggFunc);
......@@ -23,15 +24,18 @@ static void doInitFunctionHashTable() {
SScalarFunctionInfo* ptr = &scalarFunc[i];
taosHashPut(functionHashTable, scalarFunc[i].name, len, (void*)&ptr, POINTER_BYTES);
}
udfHashTable = taosHashInit(numOfEntries, MurmurHash3_32, true, true);
}
static pthread_once_t functionHashTableInit = PTHREAD_ONCE_INIT;
int32_t qIsBuiltinFunction(const char* name, int32_t len) {
int32_t qIsBuiltinFunction(const char* name, int32_t len, bool* scalarFunction) {
pthread_once(&functionHashTableInit, doInitFunctionHashTable);
SAggFunctionInfo** pInfo = taosHashGet(functionHashTable, name, len);
if (pInfo != NULL) {
*scalarFunction = ((*pInfo)->type == FUNCTION_TYPE_SCALAR);
return (*pInfo)->functionId;
} else {
return -1;
......@@ -42,8 +46,34 @@ bool qIsValidUdf(SArray* pUdfInfo, const char* name, int32_t len, int32_t* funct
return true;
}
const char* qGetFunctionName(int32_t functionId) {
bool qIsAggregateFunction(const char* functionName) {
assert(functionName != NULL);
bool scalarfunc = false;
qIsBuiltinFunction(functionName, strlen(functionName), &scalarfunc);
return !scalarfunc;
}
SAggFunctionInfo* qGetFunctionInfo(const char* name, int32_t len) {
pthread_once(&functionHashTableInit, doInitFunctionHashTable);
SAggFunctionInfo** pInfo = taosHashGet(functionHashTable, name, len);
if (pInfo != NULL) {
return (*pInfo);
} else {
return NULL;
}
}
void qAddUdfInfo(uint64_t id, SUdfInfo* pUdfInfo) {
int32_t len = (uint32_t)strlen(pUdfInfo->name);
taosHashPut(udfHashTable, pUdfInfo->name, len, (void*)&pUdfInfo, POINTER_BYTES);
}
void qRemoveUdfInfo(uint64_t id, SUdfInfo* pUdfInfo) {
int32_t len = (uint32_t)strlen(pUdfInfo->name);
taosHashRemove(udfHashTable, pUdfInfo->name, len);
}
bool isTagsQuery(SArray* pFunctionIdList) {
......
src/query/src/qScript.c → source/libs/function/src/tscript.c
浏览文件 @ 155b865a
......@@ -14,12 +14,12 @@
*/
#include "os.h"
#include "qScript.h"
#include "ttype.h"
#include "tscript.h"
#include "ttypes.h"
#include "tstrbuild.h"
#include "queryLog.h"
//#include "queryLog.h"
#include "ttokendef.h"
#if 0
static ScriptEnvPool *pool = NULL;
static ScriptEnv* getScriptEnvFromPool();
......@@ -444,3 +444,4 @@ bool isValidScript(char *script, int32_t len) {
return ret;
}
#endif
source/libs/function/src/tudf.c 0 → 100644
浏览文件 @ 155b865a
#include "tudf.h"
static char* getUdfFuncName(char* funcname, char* name, int type) {
switch (type) {
case TSDB_UDF_FUNC_NORMAL:
strcpy(funcname, name);
break;
case TSDB_UDF_FUNC_INIT:
sprintf(funcname, "%s_init", name);
break;
case TSDB_UDF_FUNC_FINALIZE:
sprintf(funcname, "%s_finalize", name);
break;
case TSDB_UDF_FUNC_MERGE:
sprintf(funcname, "%s_merge", name);
break;
case TSDB_UDF_FUNC_DESTROY:
sprintf(funcname, "%s_destroy", name);
break;
default:
assert(0);
break;
}
return funcname;
}
#if 0
int32_t initUdfInfo(SUdfInfo* pUdfInfo) {
if (pUdfInfo == NULL) {
return TSDB_CODE_SUCCESS;
}
////qError("script len: %d", pUdfInfo->contLen);
if (isValidScript(pUdfInfo->content, pUdfInfo->contLen)) {
pUdfInfo->isScript = 1;
pUdfInfo->pScriptCtx = createScriptCtx(pUdfInfo->content, pUdfInfo->resType, pUdfInfo->resBytes);
if (pUdfInfo->pScriptCtx == NULL) {
return TSDB_CODE_QRY_SYS_ERROR;
}
tfree(pUdfInfo->content);
pUdfInfo->funcs[TSDB_UDF_FUNC_INIT] = taosLoadScriptInit;
if (pUdfInfo->funcs[TSDB_UDF_FUNC_INIT] == NULL
|| (*(scriptInitFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_INIT])(pUdfInfo->pScriptCtx) != TSDB_CODE_SUCCESS) {
return TSDB_CODE_QRY_SYS_ERROR;
}
pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL] = taosLoadScriptNormal;
if (pUdfInfo->funcType == FUNCTION_TYPE_AGG) {
pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE] = taosLoadScriptFinalize;
pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE] = taosLoadScriptMerge;
}
pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY] = taosLoadScriptDestroy;
} else {
char path[PATH_MAX] = {0};
taosGetTmpfilePath("script", path, tsTempDir);
FILE* file = fopen(path, "w+");
// TODO check for failure of flush to disk
/*size_t t = */ fwrite(pUdfInfo->content, pUdfInfo->contLen, 1, file);
fclose(file);
tfree(pUdfInfo->content);
pUdfInfo->path = strdup(path);
pUdfInfo->handle = taosLoadDll(path);
if (NULL == pUdfInfo->handle) {
return TSDB_CODE_QRY_SYS_ERROR;
}
char funcname[FUNCTIONS_NAME_MAX_LENGTH + 10] = {0};
pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(funcname, pUdfInfo->name, TSDB_UDF_FUNC_NORMAL));
if (NULL == pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL]) {
return TSDB_CODE_QRY_SYS_ERROR;
}
pUdfInfo->funcs[TSDB_UDF_FUNC_INIT] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(funcname, pUdfInfo->name, TSDB_UDF_FUNC_INIT));
if (pUdfInfo->funcType == FUNCTION_TYPE_AGG) {
pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(funcname, pUdfInfo->name, TSDB_UDF_FUNC_FINALIZE));
pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(funcname, pUdfInfo->name, TSDB_UDF_FUNC_MERGE));
}
pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY] = taosLoadSym(pUdfInfo->handle, getUdfFuncName(funcname, pUdfInfo->name, TSDB_UDF_FUNC_DESTROY));
if (pUdfInfo->funcs[TSDB_UDF_FUNC_INIT]) {
return (*(udfInitFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_INIT])(&pUdfInfo->init);
}
}
return TSDB_CODE_SUCCESS;
}
void destroyUdfInfo(SUdfInfo* pUdfInfo) {
if (pUdfInfo == NULL) {
return;
}
if (pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY]) {
if (pUdfInfo->isScript) {
(*(scriptDestroyFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY])(pUdfInfo->pScriptCtx);
tfree(pUdfInfo->content);
}else{
(*(udfDestroyFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_DESTROY])(&pUdfInfo->init);
}
}
tfree(pUdfInfo->name);
if (pUdfInfo->path) {
unlink(pUdfInfo->path);
}
tfree(pUdfInfo->path);
tfree(pUdfInfo->content);
taosCloseDll(pUdfInfo->handle);
tfree(pUdfInfo);
}
void doInvokeUdf(struct SUdfInfo* pUdfInfo, SQLFunctionCtx *pCtx, int32_t idx, int32_t type) {
int32_t output = 0;
if (pUdfInfo == NULL || pUdfInfo->funcs[type] == NULL) {
//qError("empty udf function, type:%d", type);
return;
}
// //qDebug("invoke udf function:%s,%p", pUdfInfo->name, pUdfInfo->funcs[type]);
switch (type) {
case TSDB_UDF_FUNC_NORMAL:
if (pUdfInfo->isScript) {
(*(scriptNormalFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL])(pUdfInfo->pScriptCtx,
(char *)pCtx->pInput + idx * pCtx->inputType, pCtx->inputType, pCtx->inputBytes, pCtx->size, pCtx->ptsList, pCtx->startTs, pCtx->pOutput,
(char *)pCtx->ptsOutputBuf, &output, pCtx->resDataInfo.type, pCtx->resDataInfo.bytes);
} else {
SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
void *interBuf = (void *)GET_ROWCELL_INTERBUF(pResInfo);
(*(udfNormalFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_NORMAL])((char *)pCtx->pInput + idx * pCtx->inputType, pCtx->inputType, pCtx->inputBytes, pCtx->size, pCtx->ptsList,
pCtx->pOutput, interBuf, (char *)pCtx->ptsOutputBuf, &output, pCtx->resDataInfo.type, pCtx->resDataInfo.bytes, &pUdfInfo->init);
}
if (pUdfInfo->funcType == TSDB_UDF_TYPE_AGGREGATE) {
pCtx->resultInfo->numOfRes = output;
} else {
pCtx->resultInfo->numOfRes += output;
}
if (pCtx->resultInfo->numOfRes > 0) {
pCtx->resultInfo->hasResult = DATA_SET_FLAG;
}
break;
case TSDB_UDF_FUNC_MERGE:
if (pUdfInfo->isScript) {
(*(scriptMergeFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE])(pUdfInfo->pScriptCtx, pCtx->pInput, pCtx->size, pCtx->pOutput, &output);
} else {
(*(udfMergeFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_MERGE])(pCtx->pInput, pCtx->size, pCtx->pOutput, &output, &pUdfInfo->init);
}
// set the output value exist
pCtx->resultInfo->numOfRes = output;
if (output > 0) {
pCtx->resultInfo->hasResult = DATA_SET_FLAG;
}
break;
case TSDB_UDF_FUNC_FINALIZE: {
SResultRowEntryInfo *pResInfo = GET_RES_INFO(pCtx);
void *interBuf = (void *)GET_ROWCELL_INTERBUF(pResInfo);
if (pUdfInfo->isScript) {
(*(scriptFinalizeFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pUdfInfo->pScriptCtx, pCtx->startTs, pCtx->pOutput, &output);
} else {
(*(udfFinalizeFunc)pUdfInfo->funcs[TSDB_UDF_FUNC_FINALIZE])(pCtx->pOutput, interBuf, &output, &pUdfInfo->init);
}
// set the output value exist
pCtx->resultInfo->numOfRes = output;
if (output > 0) {
pCtx->resultInfo->hasResult = DATA_SET_FLAG;
}
break;
}
}
}
#endif
\ No newline at end of file
source/libs/function/src/tunaryoperator.c 0 → 100644
浏览文件 @ 155b865a
#include "tunaryoperator.h"
// TODO dynamic define these functions
_unary_scalar_fn_t getUnaryScalarOperatorFn(int32_t operator) {
assert(0);
}
bool isStringOperatorFn(int32_t op) {
return op == FUNCTION_LENGTH;
}
source/libs/parser/inc/astGenerator.h
浏览文件 @ 155b865a
......@@ -37,9 +37,14 @@ enum SQL_NODE_TYPE {
SQL_NODE_EXPR = 4,
};
enum SQL_NODE_FROM_TYPE {
SQL_NODE_FROM_SUBQUERY = 1,
SQL_NODE_FROM_TABLELIST = 2,
enum SQL_FROM_NODE_TYPE {
SQL_FROM_NODE_SUBQUERY = 1,
SQL_FROM_NODE_TABLES = 2,
};
enum SQL_UNION_TYPE {
SQL_TYPE_UNIONALL = 1,
SQL_TYPE_UNION = 2,
};
extern char tTokenTypeSwitcher[13];
......@@ -79,8 +84,8 @@ typedef struct SWindowStateVal {
struct SRelationInfo;
typedef struct SSqlNode {
struct SArray *pSelNodeList; // select clause
struct SRelationInfo *from; // from clause SArray<SSqlNode>
struct SArray *pSelNodeList; // select clause
struct tSqlExpr *pWhere; // where clause [optional]
SArray *pGroupby; // groupby clause, only for tags[optional], SArray<SListItem>
SArray *pSortOrder; // orderby [optional], SArray<SListItem>
......@@ -95,18 +100,23 @@ typedef struct SSqlNode {
struct tSqlExpr *pHaving; // having clause [optional]
} SSqlNode;
typedef struct SRelElementPair {
typedef struct SSubclause {
int32_t unionType;
SArray *node;
} SSubclause;
typedef struct SRelElement {
union {
SToken tableName;
SArray *pSubquery;
SToken tableName;
SSubclause *pSubquery;
};
SToken aliasName;
} SRelElementPair;
} SRelElement;
typedef struct SRelationInfo {
int32_t type; // nested query|table name list
SArray *list; // SArray<SRelElementPair>
SArray *list; // SArray<SRelElement>
} SRelationInfo;
typedef struct SCreatedTableInfo {
......@@ -216,7 +226,7 @@ typedef struct SMiscInfo {
typedef struct SSqlInfo {
int32_t type;
bool valid;
SArray *list; // todo refactor
SSubclause sub;
char msg[256];
SArray *funcs;
union {
......@@ -257,7 +267,7 @@ SArray *tListItemAppendToken(SArray *pList, SToken *pAliasToken, uint8_t sortOrd
SRelationInfo *setTableNameList(SRelationInfo *pRelationInfo, SToken *pName, SToken *pAlias);
void * destroyRelationInfo(SRelationInfo *pFromInfo);
SRelationInfo *addSubquery(SRelationInfo *pRelationInfo, SArray *pSub, SToken *pAlias);
SRelationInfo *addSubquery(SRelationInfo *pRelationInfo, SSubclause *pSub, SToken *pAlias);
// sql expr leaf node
tSqlExpr *tSqlExprCreateIdValue(SToken *pToken, int32_t optrType);
......@@ -285,13 +295,13 @@ SAlterTableInfo * tSetAlterTableInfo(SToken *pTableName, SArray *pCols, SArray *
SCreatedTableInfo createNewChildTableInfo(SToken *pTableName, SArray *pTagNames, SArray *pTagVals, SToken *pToken,
SToken *igExists);
void destroyAllSqlNode(SArray *pSqlNode);
void destroyAllSqlNode(struct SSubclause *pSqlNode);
void destroySqlNode(SSqlNode *pSql);
void freeCreateTableInfo(void* p);
SSqlInfo *setSqlInfo(SSqlInfo *pInfo, void *pSqlExprInfo, SToken *pTableName, int32_t type);
SArray *setSubclause(SArray *pList, void *pSqlNode);
SArray *appendSelectClause(SArray *pList, void *pSubclause);
SSubclause* setSubclause(SSubclause* sub, void *pSqlNode);
SSubclause* appendSelectClause(SSubclause *sub, int32_t unionType, void *pSubclause);
void setCreatedTableName(SSqlInfo *pInfo, SToken *pTableNameToken, SToken *pIfNotExists);
void* destroyCreateTableSql(SCreateTableSql* pCreate);
......
source/libs/parser/inc/parserUtil.h
浏览文件 @ 155b865a
......@@ -37,7 +37,8 @@ extern "C" {
(((metaInfo)->pTableMeta != NULL) && ((metaInfo)->pTableMeta->tableType == TSDB_TEMP_TABLE))
TAOS_FIELD createField(const SSchema* pSchema);
SSchema createSchema(uint8_t type, int16_t bytes, int16_t colId, const char* name);
void setColumn(SColumn* pColumn, uint64_t uid, const char* tableName, int8_t flag, const SSchema* pSchema);
SColumn createColumn(uint64_t uid, const char* tableName, int8_t flag, const SSchema* pSchema);
SInternalField* insertFieldInfo(SFieldInfo* pFieldInfo, int32_t index, SSchema* field);
int32_t getNumOfFields(SFieldInfo* pFieldInfo);
......@@ -47,20 +48,18 @@ int32_t parserValidateIdToken(SToken* pToken);
int32_t buildInvalidOperationMsg(SMsgBuf* pMsgBuf, const char* msg);
int32_t buildSyntaxErrMsg(char* dst, int32_t dstBufLen, const char* additionalInfo, const char* sourceStr);
int32_t createProjectionExpr(SQueryStmtInfo* pQueryInfo, STableMetaInfo* pTableMetaInfo, SExprInfo*** pExpr, int32_t* num);
STableMetaInfo* addEmptyMetaInfo(SQueryStmtInfo* pQueryInfo);
void columnListCopyAll(SArray* dst, const SArray* src);
void columnListDestroy(SArray* pColumnList);
SColumn* columnListInsert(SArray* pColumnList, int32_t columnIndex, uint64_t uid, SSchema* pSchema);
SColumn* insertPrimaryTsColumn(SArray* pColumnList, uint64_t tableUid);
SColumn* columnListInsert(SArray* pColumnList, uint64_t uid, SSchema* pSchema, int32_t flag);
SColumn* insertPrimaryTsColumn(SArray* pColumnList, const char* colName, uint64_t tableUid);
void cleanupTagCond(STagCond* pTagCond);
void cleanupColumnCond(SArray** pCond);
uint32_t convertRelationalOperator(SToken *pToken);
int32_t getExprFunctionId(SExprInfo *pExprInfo);
#ifdef __cplusplus
}
......
source/libs/parser/inc/queryInfoUtil.h
浏览文件 @ 155b865a
......@@ -28,10 +28,11 @@ int32_t getNumOfTags(const STableMeta* pTableMeta);
SSchema *getTableColumnSchema(const STableMeta *pTableMeta);
SSchema *getTableTagSchema(const STableMeta* pTableMeta);
size_t getNumOfExprs(SQueryStmtInfo* pQueryInfo);
SArray *getCurrentExprList(SQueryStmtInfo* pQueryInfo);
size_t getNumOfExprs(SQueryStmtInfo* pQueryInfo);
SExprInfo* createBinaryExprInfo(struct tExprNode* pNode, SSchema* pResSchema);
void addExprInfo(SQueryStmtInfo* pQueryInfo, int32_t index, SExprInfo* pExprInfo);
void addExprInfo(SArray* pExprList, int32_t index, SExprInfo* pExprInfo, int32_t level);
void updateExprInfo(SExprInfo* pExprInfo, int16_t functionId, int32_t colId, int16_t srcColumnIndex, int16_t resType, int16_t resSize);
SExprInfo* getExprInfo(SQueryStmtInfo* pQueryInfo, int32_t index);
......@@ -39,11 +40,10 @@ int32_t copyAllExprInfo(SArray* dst, const SArray* src, bool deepcopy);
void addExprInfoParam(SSqlExpr* pExpr, char* argument, int32_t type, int32_t bytes);
int32_t getExprFunctionId(SExprInfo *pExprInfo);
void cleanupFieldInfo(SFieldInfo* pFieldInfo);
STableComInfo getTableInfo(const STableMeta* pTableMeta);
SArray* extractFunctionIdList(SArray* pExprInfoList);
SArray *extractFunctionList(SArray* pExprInfoList);
#ifdef __cplusplus
}
......
source/libs/parser/inc/sql.y
浏览文件 @ 155b865a
此差异已折叠。
source/libs/parser/src/parser.c
浏览文件 @ 155b865a
此差异已折叠。
source/libs/parser/src/parserUtil.c
浏览文件 @ 155b865a
此差异已折叠。
source/libs/parser/src/sql.c
浏览文件 @ 155b865a
此差异已折叠。
source/libs/parser/src/ttokenizer.c
浏览文件 @ 155b865a
......@@ -70,7 +70,7 @@ static SKeyword keywordTable[] = {
{"STAR", TK_STAR},
{"SLASH", TK_SLASH},
{"REM ", TK_REM},
{"CONCAT", TK_CONCAT},
{"||", TK_CONCAT},
{"UMINUS", TK_UMINUS},
{"UPLUS", TK_UPLUS},
{"BITNOT", TK_BITNOT},
......
source/libs/planner/src/planner.c
浏览文件 @ 155b865a
此差异已折叠。
source/util/src/thash.c
浏览文件 @ 155b865a
......@@ -16,6 +16,7 @@
#include "os.h"
#include "thash.h"
#include "ulog.h"
#include "taos.h"
#include "tdef.h"
#define EXT_SIZE 1024
......
source/util/src/thashutil.c
浏览文件 @ 155b865a
此差异已折叠。
src/client/src/tscSQLParser.c
浏览文件 @ 155b865a
此差异已折叠。
src/client/src/tscSystem.c
浏览文件 @ 155b865a
此差异已折叠。
src/query/inc/qAggMain.h
浏览文件 @ 155b865a
此差异已折叠。
src/query/inc/qSqlparser.h
浏览文件 @ 155b865a
此差异已折叠。
src/query/inc/qTableMeta.h
浏览文件 @ 155b865a
此差异已折叠。
src/query/src/qAggMain.c
浏览文件 @ 155b865a
此差异已折叠。
src/query/src/qExecutor.c
浏览文件 @ 155b865a
此差异已折叠。
src/query/src/qFilter.c
浏览文件 @ 155b865a
此差异已折叠。
src/query/src/qPlan.c
浏览文件 @ 155b865a
此差异已折叠。
src/query/src/qSqlParser.c
浏览文件 @ 155b865a
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或