Merge branch '3.0' of https://github.com/taosdata/TDengine into fix/commit_txn
This commit is contained in:
Hongze Cheng
commit
20b3e2168b
|
|
@ -863,6 +863,7 @@ INTERP(expr)
|
|||
|
||||
- `INTERP` is used to get the value that matches the specified time slice from a column. If no such value exists an interpolation value will be returned based on `FILL` parameter.
|
||||
- The input data of `INTERP` is the value of the specified column and a `where` clause can be used to filter the original data. If no `where` condition is specified then all original data is the input.
|
||||
- `INTERP` must be used along with `RANGE`, `EVERY`, `FILL` keywords.
|
||||
- The output time range of `INTERP` is specified by `RANGE(timestamp1,timestamp2)` parameter, with timestamp1<=timestamp2. timestamp1 is the starting point of the output time range and must be specified. timestamp2 is the ending point of the output time range and must be specified.
|
||||
- The number of rows in the result set of `INTERP` is determined by the parameter `EVERY`. Starting from timestamp1, one interpolation is performed for every time interval specified `EVERY` parameter. The parameter `EVERY` must be an integer, with no quotes, with a time unit of: b(nanosecond), u(microsecond), a(millisecond)), s(second), m(minute), h(hour), d(day), or w(week). For example, `EVERY(500a)` will interpolate every 500 milliseconds.
|
||||
- Interpolation is performed based on `FILL` parameter.
|
||||
|
|
|
|||
|
|
@ -864,8 +864,9 @@ INTERP(expr)
|
|||
|
||||
- INTERP 用于在指定时间断面获取指定列的记录值，如果该时间断面不存在符合条件的行数据，那么会根据 FILL 参数的设定进行插值。
|
||||
- INTERP 的输入数据为指定列的数据，可以通过条件语句（where 子句）来对原始列数据进行过滤，如果没有指定过滤条件则输入为全部数据。
|
||||
- INTERP 需要同时与 RANGE，EVERY 和 FILL 关键字一起使用。
|
||||
- INTERP 的输出时间范围根据 RANGE(timestamp1,timestamp2)字段来指定，需满足 timestamp1<=timestamp2。其中 timestamp1（必选值）为输出时间范围的起始值，即如果 timestamp1 时刻符合插值条件则 timestamp1 为输出的第一条记录，timestamp2（必选值）为输出时间范围的结束值，即输出的最后一条记录的 timestamp 不能大于 timestamp2。
|
||||
- INTERP 根据 EVERY 字段来确定输出时间范围内的结果条数，即从 timestamp1 开始每隔固定长度的时间（EVERY 值）进行插值。如果没有指定 EVERY，则默认窗口大小为无穷大，即从 timestamp1 开始只有一个窗口。
|
||||
- INTERP 根据 EVERY 字段来确定输出时间范围内的结果条数，即从 timestamp1 开始每隔固定长度的时间（EVERY 值）进行插值。
|
||||
- INTERP 根据 FILL 字段来决定在每个符合输出条件的时刻如何进行插值。
|
||||
- INTERP 只能在一个时间序列内进行插值，因此当作用于超级表时必须跟 partition by tbname 一起使用。
|
||||
- INTERP 可以与伪列 _irowts 一起使用，返回插值点所对应的时间戳(3.0.1.4版本以后支持)。
|
||||
|
|
|
|||
|
|
@ -240,7 +240,7 @@ dataDir /mnt/data6 2 0
|
|||
|
||||
## 数据查询
|
||||
|
||||
TDengine 提供了多种多样针对表和超级表的查询处理功能，除了常规的聚合查询之外，还提供针对时序数据的窗口查询、统计聚合等功能。TDengine 的查询处理需要客户端、vnode、qnode、mnode 节点协同完成，一个复杂的超级表聚合查询可能需要多个 vnode 和 qnode 节点公共分担查询和计算任务。
|
||||
TDengine 提供了多种多样针对表和超级表的查询处理功能，除了常规的聚合查询之外，还提供针对时序数据的窗口查询、统计聚合等功能。TDengine 的查询处理需要客户端、vnode、qnode、mnode 节点协同完成，一个复杂的超级表聚合查询可能需要多个 vnode 和 qnode 节点共同分担查询和计算任务。
|
||||
|
||||
### 查询基本流程
|
||||
|
||||
|
|
|
|||
|
|
@ -199,7 +199,7 @@ int32_t buildRequest(uint64_t connId, const char* sql, int sqlLen, void* param,
|
|||
if (tsQueryUseNodeAllocator && !qIsInsertValuesSql((*pRequest)->sqlstr, (*pRequest)->sqlLen)) {
|
||||
if (TSDB_CODE_SUCCESS !=
|
||||
nodesCreateAllocator((*pRequest)->requestId, tsQueryNodeChunkSize, &((*pRequest)->allocatorRefId))) {
|
||||
tscError("%d failed to create node allocator, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", (*pRequest)->self,
|
||||
tscError("%" PRId64 " failed to create node allocator, reqId:0x%" PRIx64 ", conn:%" PRId64 ", %s", (*pRequest)->self,
|
||||
(*pRequest)->requestId, pTscObj->id, sql);
|
||||
|
||||
destroyRequest(*pRequest);
|
||||
|
|
|
|||
|
|
@ -2555,7 +2555,7 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
|
|||
int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
|
||||
char *sep = ", ";
|
||||
int32_t sepLen = strlen(sep);
|
||||
int32_t rollupLen = sizeof(rollup) - 2;
|
||||
int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;
|
||||
for (int32_t i = 0; i < rollupNum; ++i) {
|
||||
char *funcName = taosArrayGet(pStb->pFuncs, i);
|
||||
if (i) {
|
||||
|
|
|
|||
|
|
@ -209,9 +209,9 @@ int32_t smaBegin(SSma* pSma);
|
|||
int32_t smaSyncPreCommit(SSma* pSma);
|
||||
int32_t smaSyncCommit(SSma* pSma);
|
||||
int32_t smaSyncPostCommit(SSma* pSma);
|
||||
int32_t smaAsyncPreCommit(SSma* pSma);
|
||||
int32_t smaAsyncCommit(SSma* pSma);
|
||||
int32_t smaAsyncPostCommit(SSma* pSma);
|
||||
int32_t smaPreCommit(SSma* pSma);
|
||||
int32_t smaCommit(SSma* pSma);
|
||||
int32_t smaPostCommit(SSma* pSma);
|
||||
int32_t smaDoRetention(SSma* pSma, int64_t now);
|
||||
|
||||
int32_t tdProcessTSmaCreate(SSma* pSma, int64_t version, const char* msg);
|
||||
|
|
|
|||
|
|
@ -54,28 +54,28 @@ int32_t smaSyncPostCommit(SSma *pSma) { return tdProcessRSmaSyncPostCommitImpl(p
|
|||
#endif
|
||||
|
||||
/**
|
||||
* @brief Only applicable to Rollup SMA
|
||||
* @brief async commit, only applicable to Rollup SMA
|
||||
*
|
||||
* @param pSma
|
||||
* @return int32_t
|
||||
*/
|
||||
int32_t smaAsyncPreCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma); }
|
||||
int32_t smaPreCommit(SSma *pSma) { return tdProcessRSmaAsyncPreCommitImpl(pSma); }
|
||||
|
||||
/**
|
||||
* @brief Only applicable to Rollup SMA
|
||||
* @brief async commit, only applicable to Rollup SMA
|
||||
*
|
||||
* @param pSma
|
||||
* @return int32_t
|
||||
*/
|
||||
int32_t smaAsyncCommit(SSma *pSma) { return tdProcessRSmaAsyncCommitImpl(pSma); }
|
||||
int32_t smaCommit(SSma *pSma) { return tdProcessRSmaAsyncCommitImpl(pSma); }
|
||||
|
||||
/**
|
||||
* @brief Only applicable to Rollup SMA
|
||||
* @brief async commit, only applicable to Rollup SMA
|
||||
*
|
||||
* @param pSma
|
||||
* @return int32_t
|
||||
*/
|
||||
int32_t smaAsyncPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); }
|
||||
int32_t smaPostCommit(SSma *pSma) { return tdProcessRSmaAsyncPostCommitImpl(pSma); }
|
||||
|
||||
/**
|
||||
* @brief set rsma trigger stat active
|
||||
|
|
@ -366,9 +366,11 @@ static int32_t tdProcessRSmaAsyncPreCommitImpl(SSma *pSma) {
|
|||
* @return int32_t
|
||||
*/
|
||||
static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) {
|
||||
int32_t code = 0;
|
||||
SVnode *pVnode = pSma->pVnode;
|
||||
SSmaEnv *pSmaEnv = SMA_RSMA_ENV(pSma);
|
||||
if (!pSmaEnv) {
|
||||
return TSDB_CODE_SUCCESS;
|
||||
goto _exit;
|
||||
}
|
||||
#if 0
|
||||
SRSmaStat *pRSmaStat = (SRSmaStat *)SMA_ENV_STAT(pSmaEnv);
|
||||
|
|
@ -378,8 +380,21 @@ static int32_t tdProcessRSmaAsyncCommitImpl(SSma *pSma) {
|
|||
return TSDB_CODE_FAILED;
|
||||
}
|
||||
#endif
|
||||
|
||||
return TSDB_CODE_SUCCESS;
|
||||
if ((code = tsdbCommit(VND_RSMA0(pVnode))) < 0) {
|
||||
smaError("vgId:%d, failed to commit tsdb rsma0 since %s", TD_VID(pVnode), tstrerror(code));
|
||||
goto _exit;
|
||||
}
|
||||
if ((code = tsdbCommit(VND_RSMA1(pVnode))) < 0) {
|
||||
smaError("vgId:%d, failed to commit tsdb rsma1 since %s", TD_VID(pVnode), tstrerror(code));
|
||||
goto _exit;
|
||||
}
|
||||
if ((code = tsdbCommit(VND_RSMA2(pVnode))) < 0) {
|
||||
smaError("vgId:%d, failed to commit tsdb rsma2 since %s", TD_VID(pVnode), tstrerror(code));
|
||||
goto _exit;
|
||||
}
|
||||
_exit:
|
||||
terrno = code;
|
||||
return code;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -239,10 +239,10 @@ int vnodeCommit(SVnode *pVnode) {
|
|||
}
|
||||
walBeginSnapshot(pVnode->pWal, pVnode->state.applied);
|
||||
|
||||
// preCommit
|
||||
// smaSyncPreCommit(pVnode->pSma);
|
||||
code = smaAsyncPreCommit(pVnode->pSma);
|
||||
TSDB_CHECK_CODE(code, lino, _exit);
|
||||
if (smaPreCommit(pVnode->pSma) < 0) {
|
||||
vError("vgId:%d, failed to pre-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
vnodeBufPoolUnRef(pVnode->inUse);
|
||||
pVnode->inUse = NULL;
|
||||
|
|
@ -254,21 +254,8 @@ int vnodeCommit(SVnode *pVnode) {
|
|||
}
|
||||
|
||||
if (VND_IS_RSMA(pVnode)) {
|
||||
if (smaAsyncCommit(pVnode->pSma) < 0) {
|
||||
vError("vgId:%d, failed to async commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
if (tsdbCommit(VND_RSMA0(pVnode)) < 0) {
|
||||
vError("vgId:%d, failed to commit tsdb rsma0 since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
if (tsdbCommit(VND_RSMA1(pVnode)) < 0) {
|
||||
vError("vgId:%d, failed to commit tsdb rsma1 since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
if (tsdbCommit(VND_RSMA2(pVnode)) < 0) {
|
||||
vError("vgId:%d, failed to commit tsdb rsma2 since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
if (smaCommit(pVnode->pSma) < 0) {
|
||||
vError("vgId:%d, failed to commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
} else {
|
||||
|
|
@ -296,10 +283,8 @@ int vnodeCommit(SVnode *pVnode) {
|
|||
|
||||
pVnode->state.committed = info.state.committed;
|
||||
|
||||
// postCommit
|
||||
// smaSyncPostCommit(pVnode->pSma);
|
||||
if (smaAsyncPostCommit(pVnode->pSma) < 0) {
|
||||
vError("vgId:%d, failed to async post-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
if (smaPostCommit(pVnode->pSma) < 0) {
|
||||
vError("vgId:%d, failed to post-commit sma since %s", TD_VID(pVnode), tstrerror(terrno));
|
||||
return -1;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -150,9 +150,15 @@ static int32_t getStatus(SDataDeleterHandle* pDeleter) {
|
|||
static int32_t putDataBlock(SDataSinkHandle* pHandle, const SInputData* pInput, bool* pContinue) {
|
||||
SDataDeleterHandle* pDeleter = (SDataDeleterHandle*)pHandle;
|
||||
SDataDeleterBuf* pBuf = taosAllocateQitem(sizeof(SDataDeleterBuf), DEF_QITEM);
|
||||
if (NULL == pBuf || !allocBuf(pDeleter, pInput, pBuf)) {
|
||||
if (NULL == pBuf) {
|
||||
return TSDB_CODE_QRY_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
if (!allocBuf(pDeleter, pInput, pBuf)) {
|
||||
taosFreeQitem(pBuf);
|
||||
return TSDB_CODE_QRY_OUT_OF_MEMORY;
|
||||
}
|
||||
|
||||
toDataCacheEntry(pDeleter, pInput, pBuf);
|
||||
taosWriteQitem(pDeleter->pDataBlocks, pBuf);
|
||||
*pContinue = (DS_BUF_LOW == updateStatus(pDeleter) ? true : false);
|
||||
|
|
|
|||
|
|
@ -324,6 +324,7 @@ int32_t createDataInserter(SDataSinkManager* pManager, const SDataSinkNode* pDat
|
|||
tsdbGetTableSchema(inserter->pParam->readHandle->vnode, pInserterNode->tableId, &inserter->pSchema, &suid);
|
||||
if (code) {
|
||||
destroyDataSinker((SDataSinkHandle*)inserter);
|
||||
taosMemoryFree(inserter);
|
||||
return code;
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -26,19 +26,25 @@ static void *taosProcessSchedQueue(void *param);
|
|||
static void taosDumpSchedulerStatus(void *qhandle, void *tmrId);
|
||||
|
||||
void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *label, SSchedQueue *pSched) {
|
||||
bool schedMalloced = false;
|
||||
|
||||
if (NULL == pSched) {
|
||||
pSched = (SSchedQueue *)taosMemoryCalloc(sizeof(SSchedQueue), 1);
|
||||
if (pSched == NULL) {
|
||||
uError("%s: no enough memory for pSched", label);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
schedMalloced = true;
|
||||
}
|
||||
|
||||
pSched->queue = (SSchedMsg *)taosMemoryCalloc(sizeof(SSchedMsg), queueSize);
|
||||
if (pSched->queue == NULL) {
|
||||
uError("%s: no enough memory for queue", label);
|
||||
taosCleanUpScheduler(pSched);
|
||||
taosMemoryFree(pSched);
|
||||
if (schedMalloced) {
|
||||
taosMemoryFree(pSched);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
|
|
@ -46,6 +52,9 @@ void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *lab
|
|||
if (pSched->qthread == NULL) {
|
||||
uError("%s: no enough memory for qthread", label);
|
||||
taosCleanUpScheduler(pSched);
|
||||
if (schedMalloced) {
|
||||
taosMemoryFree(pSched);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
|
|
@ -58,18 +67,27 @@ void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *lab
|
|||
if (taosThreadMutexInit(&pSched->queueMutex, NULL) < 0) {
|
||||
uError("init %s:queueMutex failed(%s)", label, strerror(errno));
|
||||
taosCleanUpScheduler(pSched);
|
||||
if (schedMalloced) {
|
||||
taosMemoryFree(pSched);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (tsem_init(&pSched->emptySem, 0, (uint32_t)pSched->queueSize) != 0) {
|
||||
uError("init %s:empty semaphore failed(%s)", label, strerror(errno));
|
||||
taosCleanUpScheduler(pSched);
|
||||
if (schedMalloced) {
|
||||
taosMemoryFree(pSched);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
if (tsem_init(&pSched->fullSem, 0, 0) != 0) {
|
||||
uError("init %s:full semaphore failed(%s)", label, strerror(errno));
|
||||
taosCleanUpScheduler(pSched);
|
||||
if (schedMalloced) {
|
||||
taosMemoryFree(pSched);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
|
|
@ -83,6 +101,9 @@ void *taosInitScheduler(int32_t queueSize, int32_t numOfThreads, const char *lab
|
|||
if (code != 0) {
|
||||
uError("%s: failed to create rpc thread(%s)", label, strerror(errno));
|
||||
taosCleanUpScheduler(pSched);
|
||||
if (schedMalloced) {
|
||||
taosMemoryFree(pSched);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
++pSched->numOfThreads;
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@ sleep 50
|
|||
sql connect
|
||||
|
||||
print =============== create database
|
||||
sql create database d1 vgroups 1
|
||||
sql create database d1 keep 36500d vgroups 1
|
||||
sql use d1
|
||||
|
||||
print =============== create super table, include column type for count/sum/min/max/first
|
||||
|
|
@ -25,8 +25,8 @@ if $rows != 1 then
|
|||
endi
|
||||
|
||||
print =============== insert data, mode1: one row one table in sql
|
||||
sql insert into ct1 values(now+0s, 10, 2.0, 3.0)
|
||||
sql insert into ct1 values(now+1s, 11, 2.1, 3.1)(now+2s, -12, -2.2, -3.2)(now+3s, -13, -2.3, -3.3)
|
||||
sql insert into ct1 values('2022-10-19 09:55:45.682', 10, 2.0, 3.0)
|
||||
sql insert into ct1 values('2022-10-19 09:55:46.682', 11, 2.1, 3.1)('2022-10-19 09:55:47.682', -12, -2.2, -3.2)('2022-10-19 09:55:48.682', -13, -2.3, -3.3)
|
||||
|
||||
|
||||
print =============== create sma index from super table
|
||||
|
|
@ -34,7 +34,7 @@ sql create sma index sma_index_name1 on stb function(max(c1),max(c2),min(c1)) in
|
|||
print $data00 $data01 $data02 $data03
|
||||
|
||||
print =============== trigger stream to execute sma aggr task and insert sma data into sma store
|
||||
sql insert into ct1 values(now+5s, 20, 20.0, 30.0)
|
||||
sql insert into ct1 values('2022-10-19 09:55:50.682', 20, 20.0, 30.0)
|
||||
#===================================================================
|
||||
|
||||
print =============== show streams ================================
|
||||
|
|
|
|||
|
|
@ -232,7 +232,7 @@ python3 ./test.py -f 6-cluster/5dnode2mnode.py -N 5 -M 3
|
|||
python3 ./test.py -f 6-cluster/5dnode3mnodeStop.py -N 5 -M 3
|
||||
python3 ./test.py -f 6-cluster/5dnode3mnodeStop2Follower.py -N 5 -M 3
|
||||
python3 ./test.py -f 6-cluster/5dnode3mnodeStopLoop.py -N 5 -M 3
|
||||
python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopDnodeCreateDb.py -N 5 -M 3
|
||||
# python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopDnodeCreateDb.py -N 5 -M 3
|
||||
python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopMnodeCreateDb.py -N 5 -M 3
|
||||
python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopVnodeCreateDb.py -N 5 -M 3
|
||||
python3 ./test.py -f 6-cluster/5dnode3mnodeSep1VnodeStopMnodeCreateDbRep3.py -N 5 -M 3
|
||||
|
|
|
|||
Loading…
Reference in New Issue