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

Merge branch '3.0' of github.com:taosdata/TDengine into 3.0

This commit is contained in:
plum-lihui 2022-03-28 19:56:34 +08:00
parent baede29bd5 420ccc949c
commit fa9530a126
20 changed files with 1948 additions and 1704 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
include/common/tdataformat.h
View File

@ -63,7 +63,7 @@ extern "C" {
typedef struct {
col_id_t colId; // column ID(start from PRIMARYKEY_TIMESTAMP_COL_ID(1))
int32_t type : 8; // column type
int32_t bytes : 24; // column bytes (restore to int32_t in case of misuse)
int32_t bytes : 24; // column bytes (0~16M)
int32_t sma : 8; // block SMA: 0, no SMA, 1, sum/min/max, 2, ...
int32_t offset : 24; // point offset in STpRow after the header part.
} STColumn;
@ -81,12 +81,12 @@ typedef struct {
// ----------------- TSDB SCHEMA DEFINITION
typedef struct {
int32_t version; // version
int32_t numOfCols; // Number of columns appended
int32_t tlen; // maximum length of a STpRow without the header part (sizeof(VarDataOffsetT) + sizeof(VarDataLenT) +
// (bytes))
uint16_t flen; // First part length in a STpRow after the header part
uint16_t vlen; // pure value part length, excluded the overhead (bytes only)
int32_t numOfCols; // Number of columns appended
schema_ver_t version; // schema version
uint16_t flen; // First part length in a STpRow after the header part
int32_t vlen; // pure value part length, excluded the overhead (bytes only)
int32_t tlen; // maximum length of a STpRow without the header part
// (sizeof(VarDataOffsetT) + sizeof(VarDataLenT) + (bytes))
STColumn columns[];
} STSchema;
@ -120,13 +120,13 @@ static FORCE_INLINE STColumn *tdGetColOfID(STSchema *pSchema, int16_t colId) {
// ----------------- SCHEMA BUILDER DEFINITION
typedef struct {
int32_t tCols;
int32_t nCols;
int32_t tlen;
uint16_t flen;
uint16_t vlen;
int32_t version;
STColumn *columns;
int32_t tCols;
int32_t nCols;
schema_ver_t version;
uint16_t flen;
int32_t vlen;
int32_t tlen;
STColumn *columns;
} STSchemaBuilder;
#define TD_VTYPE_BITS 2 // val type
@ -136,9 +136,9 @@ typedef struct {
#define TD_BITMAP_BYTES(cnt) (ceil((double)cnt / TD_VTYPE_PARTS))
#define TD_BIT_TO_BYTES(cnt) (ceil((double)cnt / 8))
int32_t tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version);
int32_t tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
void tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder);
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version);
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version);
int32_t tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, col_id_t colId, col_bytes_t bytes);
STSchema *tdGetSchemaFromBuilder(STSchemaBuilder *pBuilder);

86
include/common/ttokendef.h
View File

@ -140,45 +140,53 @@
#define TK_AS 122
#define TK_NK_BOOL 123
#define TK_NK_VARIABLE 124
#define TK_BETWEEN 125
#define TK_IS 126
#define TK_NULL 127
#define TK_NK_LT 128
#define TK_NK_GT 129
#define TK_NK_LE 130
#define TK_NK_GE 131
#define TK_NK_NE 132
#define TK_MATCH 133
#define TK_NMATCH 134
#define TK_IN 135
#define TK_JOIN 136
#define TK_INNER 137
#define TK_SELECT 138
#define TK_DISTINCT 139
#define TK_WHERE 140
#define TK_PARTITION 141
#define TK_BY 142
#define TK_SESSION 143
#define TK_STATE_WINDOW 144
#define TK_SLIDING 145
#define TK_FILL 146
#define TK_VALUE 147
#define TK_NONE 148
#define TK_PREV 149
#define TK_LINEAR 150
#define TK_NEXT 151
#define TK_GROUP 152
#define TK_HAVING 153
#define TK_ORDER 154
#define TK_SLIMIT 155
#define TK_SOFFSET 156
#define TK_LIMIT 157
#define TK_OFFSET 158
#define TK_ASC 159
#define TK_DESC 160
#define TK_NULLS 161
#define TK_FIRST 162
#define TK_LAST 163
#define TK_NK_UNDERLINE 125
#define TK_ROWTS 126
#define TK_TBNAME 127
#define TK_QSTARTTS 128
#define TK_QENDTS 129
#define TK_WSTARTTS 130
#define TK_WENDTS 131
#define TK_WDURATION 132
#define TK_BETWEEN 133
#define TK_IS 134
#define TK_NULL 135
#define TK_NK_LT 136
#define TK_NK_GT 137
#define TK_NK_LE 138
#define TK_NK_GE 139
#define TK_NK_NE 140
#define TK_MATCH 141
#define TK_NMATCH 142
#define TK_IN 143
#define TK_JOIN 144
#define TK_INNER 145
#define TK_SELECT 146
#define TK_DISTINCT 147
#define TK_WHERE 148
#define TK_PARTITION 149
#define TK_BY 150
#define TK_SESSION 151
#define TK_STATE_WINDOW 152
#define TK_SLIDING 153
#define TK_FILL 154
#define TK_VALUE 155
#define TK_NONE 156
#define TK_PREV 157
#define TK_LINEAR 158
#define TK_NEXT 159
#define TK_GROUP 160
#define TK_HAVING 161
#define TK_ORDER 162
#define TK_SLIMIT 163
#define TK_SOFFSET 164
#define TK_LIMIT 165
#define TK_OFFSET 166
#define TK_ASC 167
#define TK_DESC 168
#define TK_NULLS 169
#define TK_FIRST 170
#define TK_LAST 171
#define TK_NK_SPACE 300
#define TK_NK_COMMENT 301

1
include/common/ttypes.h
View File

@ -30,6 +30,7 @@ typedef uint8_t TDRowValT;
typedef int16_t col_id_t;
typedef int8_t col_type_t;
typedef int32_t col_bytes_t;
typedef uint16_t schema_ver_t;
#pragma pack(push, 1)
typedef struct {

13
include/libs/function/functionMgt.h
View File

@ -96,7 +96,16 @@ typedef enum EFunctionType {
FUNCTION_TYPE_SERVER_SERSION,
FUNCTION_TYPE_SERVER_STATUS,
FUNCTION_TYPE_CURRENT_USER,
FUNCTION_TYPE_USER
FUNCTION_TYPE_USER,
// pseudo column function
FUNCTION_TYPE_ROWTS = 3500,
FUNCTION_TYPE_TBNAME,
FUNCTION_TYPE_QSTARTTS,
FUNCTION_TYPE_QENDTS,
FUNCTION_TYPE_WSTARTTS,
FUNCTION_TYPE_WENDTS,
FUNCTION_TYPE_WDURATION
} EFunctionType;
struct SqlFunctionCtx;
@ -118,6 +127,8 @@ bool fmIsStringFunc(int32_t funcId);
bool fmIsDatetimeFunc(int32_t funcId);
bool fmIsTimelineFunc(int32_t funcId);
bool fmIsTimeorderFunc(int32_t funcId);
bool fmIsWindowPseudoColumnFunc(int32_t funcId);
bool fmIsWindowClauseFunc(int32_t funcId);
int32_t fmFuncScanType(int32_t funcId);

7
source/common/src/tdataformat.c
View File

@ -123,7 +123,7 @@ void *tdDecodeSchema(void *buf, STSchema **pRSchema) {
return buf;
}
int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version) {
int tdInitTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
if (pBuilder == NULL) return -1;
pBuilder->tCols = 256;
@ -140,7 +140,7 @@ void tdDestroyTSchemaBuilder(STSchemaBuilder *pBuilder) {
}
}
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, int32_t version) {
void tdResetTSchemaBuilder(STSchemaBuilder *pBuilder, schema_ver_t version) {
pBuilder->nCols = 0;
pBuilder->tlen = 0;
pBuilder->flen = 0;
@ -168,6 +168,9 @@ int tdAddColToSchema(STSchemaBuilder *pBuilder, int8_t type, col_id_t colId, col
colSetOffset(pCol, pTCol->offset + TYPE_BYTES[pTCol->type]);
}
// TODO: set sma value by user input
pCol->sma = 1;
if (IS_VAR_DATA_TYPE(type)) {
colSetBytes(pCol, bytes);
pBuilder->tlen += (TYPE_BYTES[type] + bytes);

11
source/dnode/vnode/src/inc/tsdbReadImpl.h
View File

@ -67,12 +67,13 @@ typedef struct {
uint8_t last : 1;
uint8_t blkVer : 7;
uint8_t numOfSubBlocks;
int16_t numOfCols; // not including timestamp column
col_id_t numOfCols; // not including timestamp column
uint32_t len; // data block length
uint32_t keyLen : 24; // key column length, keyOffset = offset+sizeof(SBlockData)+sizeof(SBlockCol)*numOfCols
uint32_t keyLen : 20; // key column length, keyOffset = offset+sizeof(SBlockData)+sizeof(SBlockCol)*numOfCols
uint32_t algorithm : 4;
uint32_t reserve : 8;
int32_t algorithm : 8;
int32_t numOfRows : 24;
col_id_t numOfBSma;
uint16_t numOfRows;
int64_t offset;
uint64_t aggrStat : 1;
uint64_t aggrOffset : 63;
@ -80,7 +81,7 @@ typedef struct {
TSKEY keyLast;
} SBlockV0;
#define SBlock SBlockV0 // latest SBlock definition
#define SBlock SBlockV0 // latest SBlock definition
#endif

50
source/dnode/vnode/src/tsdb/tsdbCommit.c
View File

@ -43,20 +43,20 @@ typedef struct {
#define TSDB_DEFAULT_BLOCK_ROWS(maxRows) ((maxRows)*4 / 5)
#define TSDB_COMMIT_REPO(ch) TSDB_READ_REPO(&(ch->readh))
#define TSDB_COMMIT_REPO_ID(ch) REPO_ID(TSDB_READ_REPO(&(ch->readh)))
#define TSDB_COMMIT_WRITE_FSET(ch) (&((ch)->wSet))
#define TSDB_COMMIT_TABLE(ch) ((ch)->pTable)
#define TSDB_COMMIT_HEAD_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_HEAD)
#define TSDB_COMMIT_DATA_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_DATA)
#define TSDB_COMMIT_LAST_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_LAST)
#define TSDB_COMMIT_SMAD_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAD)
#define TSDB_COMMIT_SMAL_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAL)
#define TSDB_COMMIT_BUF(ch) TSDB_READ_BUF(&((ch)->readh))
#define TSDB_COMMIT_COMP_BUF(ch) TSDB_READ_COMP_BUF(&((ch)->readh))
#define TSDB_COMMIT_EXBUF(ch) TSDB_READ_EXBUF(&((ch)->readh))
#define TSDB_COMMIT_REPO(ch) TSDB_READ_REPO(&(ch->readh))
#define TSDB_COMMIT_REPO_ID(ch) REPO_ID(TSDB_READ_REPO(&(ch->readh)))
#define TSDB_COMMIT_WRITE_FSET(ch) (&((ch)->wSet))
#define TSDB_COMMIT_TABLE(ch) ((ch)->pTable)
#define TSDB_COMMIT_HEAD_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_HEAD)
#define TSDB_COMMIT_DATA_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_DATA)
#define TSDB_COMMIT_LAST_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_LAST)
#define TSDB_COMMIT_SMAD_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAD)
#define TSDB_COMMIT_SMAL_FILE(ch) TSDB_DFILE_IN_SET(TSDB_COMMIT_WRITE_FSET(ch), TSDB_FILE_SMAL)
#define TSDB_COMMIT_BUF(ch) TSDB_READ_BUF(&((ch)->readh))
#define TSDB_COMMIT_COMP_BUF(ch) TSDB_READ_COMP_BUF(&((ch)->readh))
#define TSDB_COMMIT_EXBUF(ch) TSDB_READ_EXBUF(&((ch)->readh))
#define TSDB_COMMIT_DEFAULT_ROWS(ch) TSDB_DEFAULT_BLOCK_ROWS(TSDB_COMMIT_REPO(ch)->config.maxRowsPerFileBlock)
#define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
#define TSDB_COMMIT_TXN_VERSION(ch) FS_TXN_VERSION(REPO_FS(TSDB_COMMIT_REPO(ch)))
static void tsdbStartCommit(STsdb *pRepo);
static void tsdbEndCommit(STsdb *pTsdb, int eno);
@ -1204,9 +1204,10 @@ static int tsdbComparKeyBlock(const void *arg1, const void *arg2) {
int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDFileAggr, SDataCols *pDataCols,
SBlock *pBlock, bool isLast, bool isSuper, void **ppBuf, void **ppCBuf, void **ppExBuf) {
STsdbCfg * pCfg = REPO_CFG(pRepo);
SBlockData * pBlockData = NULL;
STsdbCfg *pCfg = REPO_CFG(pRepo);
SBlockData *pBlockData = NULL;
SAggrBlkData *pAggrBlkData = NULL;
STSchema *pSchema = pTable->pSchema;
int64_t offset = 0, offsetAggr = 0;
int rowsToWrite = pDataCols->numOfRows;
@ -1225,10 +1226,12 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
pAggrBlkData = (SAggrBlkData *)(*ppExBuf);
// Get # of cols not all NULL(not including key column)
int nColsNotAllNull = 0;
col_id_t nColsNotAllNull = 0;
col_id_t nColsOfBlockSma = 0;
for (int ncol = 1; ncol < pDataCols->numOfCols; ++ncol) { // ncol from 1, we skip the timestamp column
SDataCol * pDataCol = pDataCols->cols + ncol;
SBlockCol * pBlockCol = pBlockData->cols + nColsNotAllNull;
STColumn *pColumn = pSchema->columns + ncol;
SDataCol *pDataCol = pDataCols->cols + ncol;
SBlockCol *pBlockCol = pBlockData->cols + nColsNotAllNull;
SAggrBlkCol *pAggrBlkCol = (SAggrBlkCol *)pAggrBlkData + nColsNotAllNull;
if (isAllRowsNull(pDataCol)) { // all data to commit are NULL, just ignore it
@ -1260,7 +1263,12 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
} else {
TD_SET_COL_ROWS_MISC(pBlockCol);
}
nColsNotAllNull++;
++nColsNotAllNull;
if (pColumn->sma) {
++nColsOfBlockSma;
}
}
ASSERT(nColsNotAllNull >= 0 && nColsNotAllNull <= pDataCols->numOfCols);
@ -1357,9 +1365,8 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
return -1;
}
uint32_t aggrStatus = nColsNotAllNull > 0 ? 1 : 0;
uint32_t aggrStatus = nColsOfBlockSma > 0 ? 1 : 0;
if (aggrStatus > 0) {
taosCalcChecksumAppend(0, (uint8_t *)pAggrBlkData, tsizeAggr);
tsdbUpdateDFileMagic(pDFileAggr, POINTER_SHIFT(pAggrBlkData, tsizeAggr - sizeof(TSCKSUM)));
@ -1378,6 +1385,7 @@ int tsdbWriteBlockImpl(STsdb *pRepo, STable *pTable, SDFile *pDFile, SDFile *pDF
pBlock->keyLen = keyLen;
pBlock->numOfSubBlocks = isSuper ? 1 : 0;
pBlock->numOfCols = nColsNotAllNull;
pBlock->numOfBSma = nColsOfBlockSma;
pBlock->keyFirst = dataColsKeyFirst(pDataCols);
pBlock->keyLast = dataColsKeyLast(pDataCols);
pBlock->aggrStat = aggrStatus;

2
source/dnode/vnode/src/tsdb/tsdbReadImpl.c
View File

@ -321,7 +321,7 @@ int tsdbLoadBlockStatis(SReadH *pReadh, SBlock *pBlock) {
return -1;
}
size_t sizeAggr = tsdbBlockAggrSize(pBlock->numOfCols, (uint32_t)pBlock->blkVer);
size_t sizeAggr = tsdbBlockAggrSize(pBlock->numOfBSma, (uint32_t)pBlock->blkVer);
if (tsdbMakeRoom((void **)(&(pReadh->pAggrBlkData)), sizeAggr) < 0) return -1;
int64_t nreadAggr = tsdbReadDFile(pDFileAggr, (void *)(pReadh->pAggrBlkData), sizeAggr);

6
source/dnode/vnode/src/vnd/vnodeWrite.c
View File

@ -189,12 +189,16 @@ int vnodeApplyWMsg(SVnode *pVnode, SRpcMsg *pMsg, SRpcMsg **pRsp) {
}
} break;
case TDMT_VND_CREATE_SMA: { // timeRangeSMA
#if 0
#if 1
SSmaCfg vCreateSmaReq = {0};
if (tDeserializeSVCreateTSmaReq(POINTER_SHIFT(pMsg->pCont, sizeof(SMsgHead)), &vCreateSmaReq) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
vWarn("vgId%d: TDMT_VND_CREATE_SMA received but deserialize failed since %s", pVnode->config.vgId, terrstr(terrno));
return -1;
}
vWarn("vgId%d: TDMT_VND_CREATE_SMA received for %s:%" PRIi64, pVnode->config.vgId, vCreateSmaReq.tSma.indexName,
vCreateSmaReq.tSma.indexUid);
// record current timezone of server side
tstrncpy(vCreateSmaReq.tSma.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);

9
source/dnode/vnode/test/tsdbSmaTest.cpp
View File

@ -330,7 +330,6 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
ASSERT_EQ(metaSaveSmaToDB(pMeta, pSmaCfg), 0);
// step 2: insert data
STSmaDataWrapper *pSmaData = NULL;
STsdb *pTsdb = (STsdb *)taosMemoryCalloc(1, sizeof(STsdb));
STsdbCfg *pCfg = &pTsdb->config;
@ -416,6 +415,8 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
col_id_t numOfCols = 4096;
ASSERT_GT(numOfCols, 0);
#if 0
STSmaDataWrapper *pSmaData = NULL;
pSmaData = (STSmaDataWrapper *)buf;
printf(">> allocate [%d] time to %d and addr is %p\n", ++allocCnt, bufSize, pSmaData);
pSmaData->skey = skey1;
@ -459,9 +460,13 @@ TEST(testCase, tSma_Data_Insert_Query_Test) {
pSmaData->dataLen = (len - sizeof(STSmaDataWrapper));
ASSERT_GE(bufSize, pSmaData->dataLen);
// execute
ASSERT_EQ(tsdbInsertTSmaData(pTsdb, (char *)pSmaData), TSDB_CODE_SUCCESS);
#endif
SSDataBlock *pSmaData = (SSDataBlock *)taosMemoryCalloc(1, sizeof(SSDataBlock));
// step 3: query
uint32_t checkDataCnt = 0;

2
source/libs/function/inc/builtins.h
View File

@ -33,6 +33,8 @@ extern "C" {
#define FUNC_MGT_DATETIME_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(4)
#define FUNC_MGT_TIMELINE_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(5)
#define FUNC_MGT_TIMEORDER_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(6)
#define FUNC_MGT_PSEUDO_COLUMN_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(7)
#define FUNC_MGT_WINDOW_PC_FUNC FUNC_MGT_FUNC_CLASSIFICATION_MASK(8)
#define FUNC_MGT_TEST_MASK(val, mask) (((val) & (mask)) != 0)

77
source/libs/function/src/builtins.c
View File

@ -291,6 +291,76 @@ const SBuiltinFuncDefinition funcMgtBuiltins[] = {
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_rowts",
.type = FUNCTION_TYPE_ROWTS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "tbname",
.type = FUNCTION_TYPE_TBNAME,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_qstartts",
.type = FUNCTION_TYPE_QSTARTTS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_qendts",
.type = FUNCTION_TYPE_QENDTS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_wstartts",
.type = FUNCTION_TYPE_QSTARTTS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_wendts",
.type = FUNCTION_TYPE_QENDTS,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
},
{
.name = "_wduration",
.type = FUNCTION_TYPE_WDURATION,
.classification = FUNC_MGT_PSEUDO_COLUMN_FUNC | FUNC_MGT_WINDOW_PC_FUNC,
.checkFunc = stubCheckAndGetResultType,
.getEnvFunc = NULL,
.initFunc = NULL,
.sprocessFunc = NULL,
.finalizeFunc = NULL
}
};
@ -329,6 +399,13 @@ int32_t stubCheckAndGetResultType(SFunctionNode* pFunc) {
break;
}
case FUNCTION_TYPE_CONCAT:
case FUNCTION_TYPE_ROWTS:
case FUNCTION_TYPE_TBNAME:
case FUNCTION_TYPE_QSTARTTS:
case FUNCTION_TYPE_QENDTS:
case FUNCTION_TYPE_WSTARTTS:
case FUNCTION_TYPE_WENDTS:
case FUNCTION_TYPE_WDURATION:
// todo
break;

24
source/libs/function/src/functionMgt.c
View File

@ -44,6 +44,13 @@ static void doInitFunctionHashTable() {
}
}
static bool isSpecificClassifyFunc(int32_t funcId, uint64_t classification) {
if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
return false;
}
return FUNC_MGT_TEST_MASK(funcMgtBuiltins[funcId].classification, classification);
}
int32_t fmFuncMgtInit() {
taosThreadOnce(&functionHashTableInit, doInitFunctionHashTable);
return initFunctionCode;
@ -89,10 +96,19 @@ int32_t fmGetScalarFuncExecFuncs(int32_t funcId, SScalarFuncExecFuncs* pFpSet) {
}
bool fmIsAggFunc(int32_t funcId) {
if (funcId < 0 || funcId >= funcMgtBuiltinsNum) {
return false;
}
return FUNC_MGT_TEST_MASK(funcMgtBuiltins[funcId].classification, FUNC_MGT_AGG_FUNC);
return isSpecificClassifyFunc(funcId, FUNC_MGT_AGG_FUNC);
}
bool fmIsScalarFunc(int32_t funcId) {
return isSpecificClassifyFunc(funcId, FUNC_MGT_SCALAR_FUNC);
}
bool fmIsWindowPseudoColumnFunc(int32_t funcId) {
return isSpecificClassifyFunc(funcId, FUNC_MGT_WINDOW_PC_FUNC);
}
bool fmIsWindowClauseFunc(int32_t funcId) {
return fmIsAggFunc(funcId) || fmIsWindowPseudoColumnFunc(funcId);
}
void fmFuncMgtDestroy() {

36
source/libs/parser/inc/sql.y
View File

@ -15,6 +15,7 @@
#include <assert.h>
#include <stdbool.h>
#include "functionMgt.h"
#include "nodes.h"
#include "parToken.h"
#include "ttokendef.h"
@ -417,7 +418,7 @@ topic_name(A) ::= NK_ID(B).
/************************************************ expression **********************************************************/
expression(A) ::= literal(B). { A = B; }
//expression(A) ::= NK_QUESTION(B). { A = B; }
//expression(A) ::= pseudo_column(B). { A = B; }
expression(A) ::= pseudo_column(B). { A = B; }
expression(A) ::= column_reference(B). { A = B; }
expression(A) ::= function_name(B) NK_LP expression_list(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, C)); }
expression(A) ::= function_name(B) NK_LP NK_STAR(C) NK_RP(D). { A = createRawExprNodeExt(pCxt, &B, &D, createFunctionNode(pCxt, &B, createNodeList(pCxt, createColumnNode(pCxt, NULL, &C)))); }
@ -467,7 +468,38 @@ expression_list(A) ::= expression_list(B) NK_COMMA expression(C).
column_reference(A) ::= column_name(B). { A = createRawExprNode(pCxt, &B, createColumnNode(pCxt, NULL, &B)); }
column_reference(A) ::= table_name(B) NK_DOT column_name(C). { A = createRawExprNodeExt(pCxt, &B, &C, createColumnNode(pCxt, &B, &C)); }
//pseudo_column(A) ::= NK_NOW. { A = createFunctionNode(pCxt, NULL, NULL); }
//pseudo_column(A) ::= NK_NOW. { A = createFunctionNode(pCxt, NULL, NULL); }
pseudo_column(A) ::= NK_UNDERLINE(B) ROWTS(C). {
SToken t = B;
t.n = (C.z + C.n) - B.z;
A = createRawExprNode(pCxt, &t, createFunctionNode(pCxt, &t, NULL));
}
pseudo_column(A) ::= TBNAME(B). { A = createRawExprNode(pCxt, &B, createFunctionNode(pCxt, &B, NULL)); }
pseudo_column(A) ::= NK_UNDERLINE(B) QSTARTTS(C). {
SToken t = B;
t.n = (C.z + C.n) - B.z;
A = createRawExprNode(pCxt, &t, createFunctionNode(pCxt, &t, NULL));
}
pseudo_column(A) ::= NK_UNDERLINE(B) QENDTS(C). {
SToken t = B;
t.n = (C.z + C.n) - B.z;
A = createRawExprNode(pCxt, &t, createFunctionNode(pCxt, &t, NULL));
}
pseudo_column(A) ::= NK_UNDERLINE(B) WSTARTTS(C). {
SToken t = B;
t.n = (C.z + C.n) - B.z;
A = createRawExprNode(pCxt, &t, createFunctionNode(pCxt, &t, NULL));
}
pseudo_column(A) ::= NK_UNDERLINE(B) WENDTS(C). {
SToken t = B;
t.n = (C.z + C.n) - B.z;
A = createRawExprNode(pCxt, &t, createFunctionNode(pCxt, &t, NULL));
}
pseudo_column(A) ::= NK_UNDERLINE(B) WDURATION(C). {
SToken t = B;
t.n = (C.z + C.n) - B.z;
A = createRawExprNode(pCxt, &t, createFunctionNode(pCxt, &t, NULL));
}
/************************************************ predicate ***********************************************************/
predicate(A) ::= expression(B) compare_op(C) expression(D). {

12
source/libs/parser/src/parTokenizer.c
View File

@ -104,12 +104,15 @@ static SKeyword keywordTable[] = {
{"PRECISION", TK_PRECISION},
{"PRIVILEGE", TK_PRIVILEGE},
{"PREV", TK_PREV},
{"QENDTS", TK_QENDTS},
{"QNODE", TK_QNODE},
{"QNODES", TK_QNODES},
{"QSTARTTS", TK_QSTARTTS},
{"QUORUM", TK_QUORUM},
{"REPLICA", TK_REPLICA},
{"RETENTIONS", TK_RETENTIONS},
{"ROLLUP", TK_ROLLUP},
{"ROWTS", TK_ROWTS},
{"SELECT", TK_SELECT},
{"SESSION", TK_SESSION},
{"SHOW", TK_SHOW},
@ -127,6 +130,7 @@ static SKeyword keywordTable[] = {
{"TABLE", TK_TABLE},
{"TABLES", TK_TABLES},
{"TAGS", TK_TAGS},
{"TBNAME", TK_TBNAME},
{"TIMESTAMP", TK_TIMESTAMP},
{"TINYINT", TK_TINYINT},
{"TOPIC", TK_TOPIC},
@ -141,7 +145,10 @@ static SKeyword keywordTable[] = {
{"VARCHAR", TK_VARCHAR},
{"VGROUPS", TK_VGROUPS},
{"WAL", TK_WAL},
{"WDURATION", TK_WDURATION},
{"WENDTS", TK_WENDTS},
{"WHERE", TK_WHERE},
{"WSTARTTS", TK_WSTARTTS},
// {"ID", TK_ID},
// {"STRING", TK_STRING},
// {"EQ", TK_EQ},
@ -233,7 +240,6 @@ static SKeyword keywordTable[] = {
// {"TRIGGER", TK_TRIGGER},
// {"VIEW", TK_VIEW},
// {"SEMI", TK_SEMI},
// {"TBNAME", TK_TBNAME},
// {"VNODES", TK_VNODES},
// {"PARTITIONS", TK_PARTITIONS},
// {"TOPICS", TK_TOPICS},
@ -427,6 +433,10 @@ uint32_t tGetToken(const char* z, uint32_t* tokenId) {
*tokenId = TK_NK_QUESTION;
return 1;
}
case '_': {
*tokenId = TK_NK_UNDERLINE;
return 1;
}
case '`':
case '\'':
case '"': {

3259
source/libs/parser/src/sql.c
View File

File diff suppressed because it is too large Load Diff

7
source/libs/parser/test/parserAstTest.cpp
View File

@ -206,6 +206,13 @@ TEST_F(ParserTest, selectExpression) {
ASSERT_TRUE(run());
}
TEST_F(ParserTest, selectPseudoColumn) {
setDatabase("root", "test");
bind("SELECT _wstartts, _wendts, count(*) FROM t1 interval(10s)");
ASSERT_TRUE(run());
}
TEST_F(ParserTest, selectClause) {
setDatabase("root", "test");

2
source/libs/planner/src/planLogicCreater.c
View File

@ -411,7 +411,7 @@ static int32_t createAggLogicNode(SLogicPlanContext* pCxt, SSelectStmt* pSelect,
}
static int32_t createWindowLogicNodeFinalize(SLogicPlanContext* pCxt, SSelectStmt* pSelect, SWindowLogicNode* pWindow, SLogicNode** pLogicNode) {
int32_t code = nodesCollectFuncs(pSelect, fmIsAggFunc, &pWindow->pFuncs);
int32_t code = nodesCollectFuncs(pSelect, fmIsWindowClauseFunc, &pWindow->pFuncs);
if (TSDB_CODE_SUCCESS == code) {
code = rewriteExpr(pWindow->pFuncs, pSelect, SQL_CLAUSE_WINDOW);

13
source/libs/planner/src/planPhysiCreater.c
View File

@ -36,10 +36,6 @@ typedef struct SPhysiPlanContext {
} SPhysiPlanContext;
static int32_t getSlotKey(SNode* pNode, const char* pStmtName, char* pKey) {
if (QUERY_NODE_ORDER_BY_EXPR == nodeType(pNode)) {
return getSlotKey(((SOrderByExprNode*)pNode)->pExpr, pStmtName, pKey);
}
if (QUERY_NODE_COLUMN == nodeType(pNode)) {
SColumnNode* pCol = (SColumnNode*)pNode;
if (NULL != pStmtName) {
@ -184,15 +180,16 @@ static int32_t addDataBlockSlotsImpl(SPhysiPlanContext* pCxt, SNodeList* pList,
int16_t nextSlotId = taosHashGetSize(pHash), slotId = 0;
SNode* pNode = NULL;
FOREACH(pNode, pList) {
SNode* pExpr = QUERY_NODE_ORDER_BY_EXPR == nodeType(pNode) ? ((SOrderByExprNode*)pNode)->pExpr : pNode;
char name[TSDB_TABLE_NAME_LEN + TSDB_COL_NAME_LEN] = {0};
int32_t len = getSlotKey(pNode, pStmtName, name);
int32_t len = getSlotKey(pExpr, pStmtName, name);
SSlotIndex* pIndex = taosHashGet(pHash, name, len);
if (NULL == pIndex) {
code = nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, pNode, nextSlotId, output));
code = nodesListStrictAppend(pDataBlockDesc->pSlots, createSlotDesc(pCxt, pExpr, nextSlotId, output));
if (TSDB_CODE_SUCCESS == code) {
code = putSlotToHashImpl(pDataBlockDesc->dataBlockId, nextSlotId, name, len, pHash);
}
pDataBlockDesc->resultRowSize += ((SExprNode*)pNode)->resType.bytes;
pDataBlockDesc->resultRowSize += ((SExprNode*)pExpr)->resType.bytes;
slotId = nextSlotId;
++nextSlotId;
} else {
@ -600,7 +597,7 @@ static EDealRes doRewritePrecalcExprs(SNode** pNode, void* pContext) {
return collectAndRewrite(pContext, pNode);
}
case QUERY_NODE_FUNCTION: {
if (!fmIsAggFunc(((SFunctionNode*)(*pNode))->funcId)) {
if (fmIsScalarFunc(((SFunctionNode*)(*pNode))->funcId)) {
return collectAndRewrite(pContext, pNode);
}
}

3
source/libs/planner/test/plannerTest.cpp
View File

@ -192,6 +192,9 @@ TEST_F(PlannerTest, interval) {
bind("SELECT count(*) FROM t1 interval(10s)");
ASSERT_TRUE(run());
bind("SELECT _wstartts, _wduration, _wendts, count(*) FROM t1 interval(10s)");
ASSERT_TRUE(run());
}
TEST_F(PlannerTest, sessionWindow) {