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

fix bug issue #643.

This commit is contained in:
hjxilinx 2019-10-30 22:16:03 +08:00
parent 09e347045a
commit d0f62cfff1
6 changed files with 48 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
src/client/src/tscSecondaryMerge.c
View File

@ -279,8 +279,9 @@ void tscCreateLocalReducer(tExtMemBuffer **pMemBuffer, int32_t numOfBuffer, tOrd
pRes->pLocalReducer = pReducer; pRes->pLocalReducer = pReducer;
pRes->numOfGroups = 0; pRes->numOfGroups = 0;
int16_t prec = pCmd->pMeterMeta->precision;
int64_t stime = (pCmd->stime < pCmd->etime) ? pCmd->stime : pCmd->etime; int64_t stime = (pCmd->stime < pCmd->etime) ? pCmd->stime : pCmd->etime;
int64_t revisedSTime = taosGetIntervalStartTimestamp(stime, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit); int64_t revisedSTime = taosGetIntervalStartTimestamp(stime, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit, prec);
SInterpolationInfo *pInterpoInfo = &pReducer->interpolationInfo; SInterpolationInfo *pInterpoInfo = &pReducer->interpolationInfo;
taosInitInterpoInfo(pInterpoInfo, pCmd->order.order, revisedSTime, pCmd->groupbyExpr.numOfGroupbyCols, taosInitInterpoInfo(pInterpoInfo, pCmd->order.order, revisedSTime, pCmd->groupbyExpr.numOfGroupbyCols,
@ -664,12 +665,11 @@ void adjustLoserTreeFromNewData(SLocalReducer *pLocalReducer, SLocalDataSource *
} }
} }
void savePrevRecordAndSetupInterpoInfo( void savePrevRecordAndSetupInterpoInfo(SLocalReducer *pLocalReducer, SSqlCmd *pCmd, SInterpolationInfo *pInterpoInfo) {
SLocalReducer *pLocalReducer, SSqlCmd *pCmd, // discard following dataset in the same group and reset the interpolation information
SInterpolationInfo int16_t prec = pCmd->pMeterMeta->precision;
*pInterpoInfo) { // discard following dataset in the same group and reset the interpolation information
int64_t stime = (pCmd->stime < pCmd->etime) ? pCmd->stime : pCmd->etime; int64_t stime = (pCmd->stime < pCmd->etime) ? pCmd->stime : pCmd->etime;
int64_t revisedSTime = taosGetIntervalStartTimestamp(stime, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit); int64_t revisedSTime = taosGetIntervalStartTimestamp(stime, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit, prec);
taosInitInterpoInfo(pInterpoInfo, pCmd->order.order, revisedSTime, pCmd->groupbyExpr.numOfGroupbyCols, taosInitInterpoInfo(pInterpoInfo, pCmd->order.order, revisedSTime, pCmd->groupbyExpr.numOfGroupbyCols,
pLocalReducer->rowSize); pLocalReducer->rowSize);
@ -811,9 +811,12 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
functions[i] = tscSqlExprGet(pCmd, i)->sqlFuncId; functions[i] = tscSqlExprGet(pCmd, i)->sqlFuncId;
} }
int8_t precision = pCmd->pMeterMeta->precision;
while (1) { while (1) {
int32_t remains = taosNumOfRemainPoints(pInterpoInfo); int32_t remains = taosNumOfRemainPoints(pInterpoInfo);
TSKEY etime = taosGetRevisedEndKey(actualETime, pCmd->order.order, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit); TSKEY etime = taosGetRevisedEndKey(actualETime, pCmd->order.order, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit,
precision);
int32_t nrows = taosGetNumOfResultWithInterpo(pInterpoInfo, pPrimaryKeys, remains, pCmd->nAggTimeInterval, etime, int32_t nrows = taosGetNumOfResultWithInterpo(pInterpoInfo, pPrimaryKeys, remains, pCmd->nAggTimeInterval, etime,
pLocalReducer->resColModel->maxCapacity); pLocalReducer->resColModel->maxCapacity);
@ -1115,10 +1118,12 @@ static void resetEnvForNewResultset(SSqlRes *pRes, SSqlCmd *pCmd, SLocalReducer
pRes->numOfTotal = 0; pRes->numOfTotal = 0;
pCmd->limit.offset = pLocalReducer->offset; pCmd->limit.offset = pLocalReducer->offset;
int16_t precision = pCmd->pMeterMeta->precision;
if (pCmd->interpoType != TSDB_INTERPO_NONE) { if (pCmd->interpoType != TSDB_INTERPO_NONE) {
/* for group result interpolation, do not return if not data is generated */ /* for group result interpolation, do not return if not data is generated */
int64_t stime = (pCmd->stime < pCmd->etime) ? pCmd->stime : pCmd->etime; int64_t stime = (pCmd->stime < pCmd->etime) ? pCmd->stime : pCmd->etime;
int64_t newTime = taosGetIntervalStartTimestamp(stime, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit); int64_t newTime = taosGetIntervalStartTimestamp(stime, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit, precision);
taosInitInterpoInfo(&pLocalReducer->interpolationInfo, pCmd->order.order, newTime, taosInitInterpoInfo(&pLocalReducer->interpolationInfo, pCmd->order.order, newTime,
pCmd->groupbyExpr.numOfGroupbyCols, pLocalReducer->rowSize); pCmd->groupbyExpr.numOfGroupbyCols, pLocalReducer->rowSize);
@ -1135,6 +1140,7 @@ static bool doInterpolationForCurrentGroup(SSqlObj *pSql) {
SLocalReducer * pLocalReducer = pRes->pLocalReducer; SLocalReducer * pLocalReducer = pRes->pLocalReducer;
SInterpolationInfo *pInterpoInfo = &pLocalReducer->interpolationInfo; SInterpolationInfo *pInterpoInfo = &pLocalReducer->interpolationInfo;
int8_t p = pCmd->pMeterMeta->precision;
if (taosHasRemainsDataForInterpolation(pInterpoInfo)) { if (taosHasRemainsDataForInterpolation(pInterpoInfo)) {
assert(pCmd->interpoType != TSDB_INTERPO_NONE); assert(pCmd->interpoType != TSDB_INTERPO_NONE);
@ -1143,7 +1149,7 @@ static bool doInterpolationForCurrentGroup(SSqlObj *pSql) {
int64_t etime = *(int64_t *)(pFinalDataBuf->data + TSDB_KEYSIZE * (pInterpoInfo->numOfRawDataInRows - 1)); int64_t etime = *(int64_t *)(pFinalDataBuf->data + TSDB_KEYSIZE * (pInterpoInfo->numOfRawDataInRows - 1));
int32_t remain = taosNumOfRemainPoints(pInterpoInfo); int32_t remain = taosNumOfRemainPoints(pInterpoInfo);
TSKEY ekey = taosGetRevisedEndKey(etime, pCmd->order.order, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit); TSKEY ekey = taosGetRevisedEndKey(etime, pCmd->order.order, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit, p);
int32_t rows = taosGetNumOfResultWithInterpo(pInterpoInfo, (TSKEY *)pLocalReducer->pBufForInterpo, remain, int32_t rows = taosGetNumOfResultWithInterpo(pInterpoInfo, (TSKEY *)pLocalReducer->pBufForInterpo, remain,
pCmd->nAggTimeInterval, ekey, pLocalReducer->resColModel->maxCapacity); pCmd->nAggTimeInterval, ekey, pLocalReducer->resColModel->maxCapacity);
if (rows > 0) { // do interpo if (rows > 0) { // do interpo
@ -1164,6 +1170,7 @@ static bool doHandleLastRemainData(SSqlObj *pSql) {
SInterpolationInfo *pInterpoInfo = &pLocalReducer->interpolationInfo; SInterpolationInfo *pInterpoInfo = &pLocalReducer->interpolationInfo;
bool prevGroupCompleted = (!pLocalReducer->discard) && pLocalReducer->hasUnprocessedRow; bool prevGroupCompleted = (!pLocalReducer->discard) && pLocalReducer->hasUnprocessedRow;
int8_t precision = pCmd->pMeterMeta->precision;
if ((isAllSourcesCompleted(pLocalReducer) && !pLocalReducer->hasPrevRow) || pLocalReducer->pLocalDataSrc[0] == NULL || if ((isAllSourcesCompleted(pLocalReducer) && !pLocalReducer->hasPrevRow) || pLocalReducer->pLocalDataSrc[0] == NULL ||
prevGroupCompleted) { prevGroupCompleted) {
@ -1171,7 +1178,7 @@ static bool doHandleLastRemainData(SSqlObj *pSql) {
if (pCmd->interpoType != TSDB_INTERPO_NONE) { if (pCmd->interpoType != TSDB_INTERPO_NONE) {
int64_t etime = (pCmd->stime < pCmd->etime) ? pCmd->etime : pCmd->stime; int64_t etime = (pCmd->stime < pCmd->etime) ? pCmd->etime : pCmd->stime;
etime = taosGetRevisedEndKey(etime, pCmd->order.order, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit); etime = taosGetRevisedEndKey(etime, pCmd->order.order, pCmd->nAggTimeInterval, pCmd->intervalTimeUnit, precision);
int32_t rows = taosGetNumOfResultWithInterpo(pInterpoInfo, NULL, 0, pCmd->nAggTimeInterval, etime, int32_t rows = taosGetNumOfResultWithInterpo(pInterpoInfo, NULL, 0, pCmd->nAggTimeInterval, etime,
pLocalReducer->resColModel->maxCapacity); pLocalReducer->resColModel->maxCapacity);
if (rows > 0) { // do interpo if (rows > 0) { // do interpo

4
src/inc/tinterpolation.h
View File

@ -40,14 +40,14 @@ typedef struct SPoint {
typedef void (*__interpo_callback_fn_t)(void *param); typedef void (*__interpo_callback_fn_t)(void *param);
int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char intervalTimeUnit); int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char intervalTimeUnit, int16_t precision);
void taosInitInterpoInfo(SInterpolationInfo *pInterpoInfo, int32_t order, int64_t startTimeStamp, int32_t numOfTags, void taosInitInterpoInfo(SInterpolationInfo *pInterpoInfo, int32_t order, int64_t startTimeStamp, int32_t numOfTags,
int32_t rowSize); int32_t rowSize);
void taosInterpoSetStartInfo(SInterpolationInfo *pInterpoInfo, int32_t numOfRawDataInRows, int32_t type); void taosInterpoSetStartInfo(SInterpolationInfo *pInterpoInfo, int32_t numOfRawDataInRows, int32_t type);
TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int32_t timeInterval, int8_t intervalTimeUnit); TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int32_t timeInterval, int8_t intervalTimeUnit, int8_t precision);
/** /**
* *

14
src/system/inc/vnode.h
View File

@ -278,14 +278,14 @@ typedef struct {
TSKEY ekey; TSKEY ekey;
int64_t nAggTimeInterval; int64_t nAggTimeInterval;
char intervalTimeUnit; // interval data type, used for daytime revise char intervalTimeUnit; // interval data type, used for daytime revise
char precision;
int16_t numOfOutputCols;
int16_t interpoType;
int16_t checkBufferInLoop; // check if the buffer is full during scan each block
int16_t numOfOutputCols; SLimitVal limit;
int16_t interpoType; int32_t rowSize;
int16_t checkBufferInLoop; // check if the buffer is full during scan each block int32_t dataRowSize; // row size of each loaded data from disk, the value is
SLimitVal limit;
int32_t rowSize;
int32_t dataRowSize; // row size of each loaded data from disk, the value is
// used for prepare buffer // used for prepare buffer
SSqlGroupbyExpr * pGroupbyExpr; SSqlGroupbyExpr * pGroupbyExpr;
SSqlFunctionExpr *pSelectExpr; SSqlFunctionExpr *pSelectExpr;

20
src/system/src/vnodeQueryImpl.c
View File

@ -1105,7 +1105,8 @@ static int32_t applyAllFunctions(SQueryRuntimeEnv *pRuntimeEnv, int32_t forwardS
TSKEY ts = QUERY_IS_ASC_QUERY(pQuery) ? pQuery->skey : pQuery->ekey; TSKEY ts = QUERY_IS_ASC_QUERY(pQuery) ? pQuery->skey : pQuery->ekey;
int64_t alignedTimestamp = taosGetIntervalStartTimestamp(ts, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit); int64_t alignedTimestamp = taosGetIntervalStartTimestamp(ts, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit,
pQuery->precision);
setExecParams(pQuery, &pCtx[k], alignedTimestamp, dataBlock, (char *)primaryKeyCol, forwardStep, functionId, setExecParams(pQuery, &pCtx[k], alignedTimestamp, dataBlock, (char *)primaryKeyCol, forwardStep, functionId,
tpField, hasNull, pRuntimeEnv->blockStatus, &sas, pRuntimeEnv->scanFlag); tpField, hasNull, pRuntimeEnv->blockStatus, &sas, pRuntimeEnv->scanFlag);
@ -1190,7 +1191,8 @@ static int32_t applyAllFunctions_Filter(SQueryRuntimeEnv *pRuntimeEnv, int32_t *
char *dataBlock = getDataBlocks(pRuntimeEnv, data, &sasArray[k], k, isDiskFileBlock); char *dataBlock = getDataBlocks(pRuntimeEnv, data, &sasArray[k], k, isDiskFileBlock);
TSKEY ts = QUERY_IS_ASC_QUERY(pQuery) ? pQuery->skey : pQuery->ekey; TSKEY ts = QUERY_IS_ASC_QUERY(pQuery) ? pQuery->skey : pQuery->ekey;
int64_t alignedTimestamp = taosGetIntervalStartTimestamp(ts, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit); int64_t alignedTimestamp = taosGetIntervalStartTimestamp(ts, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit,
pQuery->precision);
setExecParams(pQuery, &pCtx[k], alignedTimestamp, dataBlock, (char *)primaryKeyCol, (*forwardStep), functionId, setExecParams(pQuery, &pCtx[k], alignedTimestamp, dataBlock, (char *)primaryKeyCol, (*forwardStep), functionId,
pFields, hasNull, pRuntimeEnv->blockStatus, &sasArray[k], pRuntimeEnv->scanFlag); pFields, hasNull, pRuntimeEnv->blockStatus, &sasArray[k], pRuntimeEnv->scanFlag);
@ -2096,7 +2098,8 @@ static void doGetAlignedIntervalQueryRangeImpl(SQuery *pQuery, int64_t qualified
return; return;
} }
*skey = taosGetIntervalStartTimestamp(qualifiedKey, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit); *skey = taosGetIntervalStartTimestamp(qualifiedKey, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit,
pQuery->precision);
int64_t endKey = *skey + pQuery->nAggTimeInterval - 1; int64_t endKey = *skey + pQuery->nAggTimeInterval - 1;
if (*skey < keyFirst) { if (*skey < keyFirst) {
@ -3276,7 +3279,8 @@ int32_t vnodeQuerySingleMeterPrepare(SQInfo *pQInfo, SMeterObj *pMeterObj, SMete
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
int64_t rs = taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit); int64_t rs = taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit,
pQuery->precision);
taosInitInterpoInfo(&pSupporter->runtimeEnv.interpoInfo, pQuery->order.order, rs, 0, 0); taosInitInterpoInfo(&pSupporter->runtimeEnv.interpoInfo, pQuery->order.order, rs, 0, 0);
allocMemForInterpo(pSupporter, pQuery, pMeterObj); allocMemForInterpo(pSupporter, pQuery, pMeterObj);
@ -3431,8 +3435,8 @@ int32_t vnodeMultiMeterQueryPrepare(SQInfo *pQInfo, SQuery *pQuery) {
pQuery->interpoType = TSDB_INTERPO_NONE; pQuery->interpoType = TSDB_INTERPO_NONE;
} }
TSKEY revisedStime = TSKEY revisedStime = taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->nAggTimeInterval,
taosGetIntervalStartTimestamp(pSupporter->rawSKey, pQuery->nAggTimeInterval, pQuery->intervalTimeUnit); pQuery->intervalTimeUnit, pQuery->precision);
taosInitInterpoInfo(&pSupporter->runtimeEnv.interpoInfo, pQuery->order.order, revisedStime, 0, 0); taosInitInterpoInfo(&pSupporter->runtimeEnv.interpoInfo, pQuery->order.order, revisedStime, 0, 0);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
@ -6117,7 +6121,7 @@ bool vnodeHasRemainResults(void *handle) {
if (Q_STATUS_EQUAL(pQuery->over, QUERY_COMPLETED | QUERY_NO_DATA_TO_CHECK)) { if (Q_STATUS_EQUAL(pQuery->over, QUERY_COMPLETED | QUERY_NO_DATA_TO_CHECK)) {
/* query has completed */ /* query has completed */
TSKEY ekey = taosGetRevisedEndKey(pSupporter->rawEKey, pQuery->order.order, pQuery->nAggTimeInterval, TSKEY ekey = taosGetRevisedEndKey(pSupporter->rawEKey, pQuery->order.order, pQuery->nAggTimeInterval,
pQuery->intervalTimeUnit); pQuery->intervalTimeUnit, pQuery->precision);
int32_t numOfTotal = taosGetNumOfResultWithInterpo(pInterpoInfo, (TSKEY *)pRuntimeEnv->pInterpoBuf[0]->data, int32_t numOfTotal = taosGetNumOfResultWithInterpo(pInterpoInfo, (TSKEY *)pRuntimeEnv->pInterpoBuf[0]->data,
remain, pQuery->nAggTimeInterval, ekey, pQuery->pointsToRead); remain, pQuery->nAggTimeInterval, ekey, pQuery->pointsToRead);
return numOfTotal > 0; return numOfTotal > 0;
@ -6201,7 +6205,7 @@ int32_t vnodeQueryResultInterpolate(SQInfo *pQInfo, tFilePage **pDst, tFilePage
numOfRows = taosNumOfRemainPoints(&pRuntimeEnv->interpoInfo); numOfRows = taosNumOfRemainPoints(&pRuntimeEnv->interpoInfo);
TSKEY ekey = taosGetRevisedEndKey(pSupporter->rawEKey, pQuery->order.order, pQuery->nAggTimeInterval, TSKEY ekey = taosGetRevisedEndKey(pSupporter->rawEKey, pQuery->order.order, pQuery->nAggTimeInterval,
pQuery->intervalTimeUnit); pQuery->intervalTimeUnit, pQuery->precision);
int32_t numOfFinalRows = int32_t numOfFinalRows =
taosGetNumOfResultWithInterpo(&pRuntimeEnv->interpoInfo, (TSKEY *)pDataSrc[0]->data, numOfRows, taosGetNumOfResultWithInterpo(&pRuntimeEnv->interpoInfo, (TSKEY *)pDataSrc[0]->data, numOfRows,
pQuery->nAggTimeInterval, ekey, pQuery->pointsToRead); pQuery->nAggTimeInterval, ekey, pQuery->pointsToRead);

2
src/system/src/vnodeRead.c
View File

@ -222,6 +222,8 @@ static SQInfo *vnodeAllocateQInfoCommon(SQueryMeterMsg *pQueryMsg, SMeterObj *pM
} }
vnodeUpdateFilterColumnIndex(pQuery); vnodeUpdateFilterColumnIndex(pQuery);
pQuery->precision = vnodeList[pMeterObj->vnode].cfg.precision;
return pQInfo; return pQInfo;
_clean_memory: _clean_memory:

15
src/util/src/tinterpolation.c
View File

@ -25,7 +25,7 @@
#define INTERPOL_IS_ASC_INTERPOL(interp) ((interp)->order == TSQL_SO_ASC) #define INTERPOL_IS_ASC_INTERPOL(interp) ((interp)->order == TSQL_SO_ASC)
int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char intervalTimeUnit) { int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char intervalTimeUnit, int16_t precision) {
if (timeRange == 0) { if (timeRange == 0) {
return startTime; return startTime;
} }
@ -39,16 +39,17 @@ int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t timeRange, char
* *
* TODO dynmaically decide the start time of a day * TODO dynmaically decide the start time of a day
*/ */
#ifdef _MSC_VER
#if _MSC_VER >= 1900 #if defined(WINDOWS) && _MSC_VER >= 1900
// see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019 // see https://docs.microsoft.com/en-us/cpp/c-runtime-library/daylight-dstbias-timezone-and-tzname?view=vs-2019
int64_t timezone = _timezone; int64_t timezone = _timezone;
int32_t daylight = _daylight; int32_t daylight = _daylight;
char** tzname = _tzname; char** tzname = _tzname;
#endif
#endif #endif
int64_t revStartime = (startTime / timeRange) * timeRange + timezone * MILLISECOND_PER_SECOND; int64_t t = (precision == TSDB_TIME_PRECISION_MILLI)?MILLISECOND_PER_SECOND:MILLISECOND_PER_SECOND*1000L;
int64_t revStartime = (startTime / timeRange) * timeRange + timezone * t;
int64_t revEndtime = revStartime + timeRange - 1; int64_t revEndtime = revStartime + timeRange - 1;
if (revEndtime < startTime) { if (revEndtime < startTime) {
revStartime += timeRange; revStartime += timeRange;
@ -85,11 +86,11 @@ void taosInterpoSetStartInfo(SInterpolationInfo* pInterpoInfo, int32_t numOfRawD
pInterpoInfo->numOfRawDataInRows = numOfRawDataInRows; pInterpoInfo->numOfRawDataInRows = numOfRawDataInRows;
} }
TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int32_t timeInterval, int8_t intervalTimeUnit) { TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int32_t timeInterval, int8_t intervalTimeUnit, int8_t precision) {
if (order == TSQL_SO_ASC) { if (order == TSQL_SO_ASC) {
return ekey; return ekey;
} else { } else {
return taosGetIntervalStartTimestamp(ekey, timeInterval, intervalTimeUnit); return taosGetIntervalStartTimestamp(ekey, timeInterval, intervalTimeUnit, precision);
} }
} }