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

Merge pull request #985 from taosdata/feature/liaohj 

Feature/liaohj
This commit is contained in:
slguan 2019-12-21 15:18:32 +08:00 committed by GitHub
parent 9a0ef41bc2 f6d56d843b
commit 07b32ec20e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 154 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/client/inc/tscUtil.h
View File

@ -66,19 +66,24 @@ typedef struct SJoinSubquerySupporter {
char path[PATH_MAX]; // temporary file path
} SJoinSubquerySupporter;
void tscDestroyDataBlock(STableDataBlocks* pDataBlock);
STableDataBlocks* tscCreateDataBlock(size_t initialBufSize, int32_t rowSize, int32_t startOffset, const char* name);
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
STableDataBlocks** dataBlocks);
void tscAppendDataBlock(SDataBlockList* pList, STableDataBlocks* pBlocks);
void tscDestroyDataBlock(STableDataBlocks* pDataBlock);
SParamInfo* tscAddParamToDataBlock(STableDataBlocks* pDataBlock, char type, uint8_t timePrec, short bytes,
uint32_t offset);
SDataBlockList* tscCreateBlockArrayList();
void* tscDestroyBlockArrayList(SDataBlockList* pList);
int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock);
void tscFreeUnusedDataBlocks(SDataBlockList* pList);
int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SDataBlockList* pDataList);
STableDataBlocks* tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList, int64_t id, int32_t size,
int32_t startOffset, int32_t rowSize, const char* tableId);
int32_t tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList, int64_t id, int32_t size,
int32_t startOffset, int32_t rowSize, const char* tableId,
STableDataBlocks** dataBlocks);
SVnodeSidList* tscGetVnodeSidList(SMetricMeta* pMetricmeta, int32_t vnodeIdx);
SMeterSidExtInfo* tscGetMeterSidInfo(SVnodeSidList* pSidList, int32_t idx);

25
src/client/src/tscParseInsert.c
View File

@ -653,9 +653,12 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableHashList, char
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0);
SMeterMeta * pMeterMeta = pMeterMetaInfo->pMeterMeta;
STableDataBlocks *dataBuf =
tscGetDataBlockFromList(pTableHashList, pCmd->pDataBlocks, pMeterMeta->uid, TSDB_DEFAULT_PAYLOAD_SIZE,
sizeof(SShellSubmitBlock), pMeterMeta->rowSize, pMeterMetaInfo->name);
STableDataBlocks *dataBuf = NULL;
int32_t ret = tscGetDataBlockFromList(pTableHashList, pCmd->pDataBlocks, pMeterMeta->uid, TSDB_DEFAULT_PAYLOAD_SIZE,
sizeof(SShellSubmitBlock), pMeterMeta->rowSize, pMeterMetaInfo->name, &dataBuf);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
int32_t maxNumOfRows = tscAllocateMemIfNeed(dataBuf, pMeterMeta->rowSize);
if (0 == maxNumOfRows) {
@ -1072,8 +1075,12 @@ int doParserInsertSql(SSqlObj *pSql, char *str) {
strcpy(fname, full_path.we_wordv[0]);
wordfree(&full_path);
STableDataBlocks *pDataBlock = tscCreateDataBlock(PATH_MAX, pMeterMetaInfo->pMeterMeta->rowSize,
sizeof(SShellSubmitBlock), pMeterMetaInfo->name);
STableDataBlocks *pDataBlock = NULL;
int32_t ret = tscCreateDataBlock(PATH_MAX, pMeterMetaInfo->pMeterMeta->rowSize, sizeof(SShellSubmitBlock),
pMeterMetaInfo->name, &pDataBlock);
if (ret != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
tscAppendDataBlock(pCmd->pDataBlocks, pDataBlock);
strcpy(pDataBlock->filename, fname);
@ -1314,8 +1321,12 @@ static int tscInsertDataFromFile(SSqlObj *pSql, FILE *fp, char *tmpTokenBuf) {
int32_t rowSize = pMeterMeta->rowSize;
pCmd->pDataBlocks = tscCreateBlockArrayList();
STableDataBlocks *pTableDataBlock = tscCreateDataBlock(TSDB_PAYLOAD_SIZE, pMeterMeta->rowSize,
sizeof(SShellSubmitBlock), pMeterMetaInfo->name);
STableDataBlocks *pTableDataBlock = NULL;
int32_t ret = tscCreateDataBlock(TSDB_PAYLOAD_SIZE, pMeterMeta->rowSize, sizeof(SShellSubmitBlock),
pMeterMetaInfo->name, &pTableDataBlock);
if (ret != TSDB_CODE_SUCCESS) {
return -1;
}
tscAppendDataBlock(pCmd->pDataBlocks, pTableDataBlock);

117
src/client/src/tscUtil.c
View File

@ -22,8 +22,8 @@
#include "tscJoinProcess.h"
#include "tscProfile.h"
#include "tscSecondaryMerge.h"
#include "tscUtil.h"
#include "tschemautil.h"
#include "tscUtil.h"
#include "tsclient.h"
#include "tsqldef.h"
#include "ttimer.h"
@ -60,7 +60,7 @@ void tscGetMetricMetaCacheKey(SSqlCmd* pCmd, char* str, uint64_t uid) {
}
// estimate the buffer size
size_t tbnameCondLen = pTagCond->tbnameCond.cond != NULL? strlen(pTagCond->tbnameCond.cond):0;
size_t tbnameCondLen = pTagCond->tbnameCond.cond != NULL ? strlen(pTagCond->tbnameCond.cond) : 0;
size_t redundantLen = 20;
size_t bufSize = strlen(pMeterMetaInfo->name) + tbnameCondLen + strlen(join) + strlen(tagIdBuf);
@ -68,12 +68,11 @@ void tscGetMetricMetaCacheKey(SSqlCmd* pCmd, char* str, uint64_t uid) {
bufSize += strlen(cond->cond);
}
bufSize = (size_t) ((bufSize + redundantLen) * 1.5);
bufSize = (size_t)((bufSize + redundantLen) * 1.5);
char* tmp = calloc(1, bufSize);
int32_t keyLen = snprintf(tmp, bufSize, "%s,%s,%s,%d,%s,[%s],%d", pMeterMetaInfo->name,
(cond != NULL ? cond->cond : NULL),
(tbnameCondLen > 0 ? pTagCond->tbnameCond.cond : NULL),
(cond != NULL ? cond->cond : NULL), (tbnameCondLen > 0 ? pTagCond->tbnameCond.cond : NULL),
pTagCond->relType, join, tagIdBuf, pCmd->groupbyExpr.orderType);
assert(keyLen <= bufSize);
@ -84,7 +83,7 @@ void tscGetMetricMetaCacheKey(SSqlCmd* pCmd, char* str, uint64_t uid) {
MD5_CTX ctx;
MD5Init(&ctx);
MD5Update(&ctx, (uint8_t*) tmp, keyLen);
MD5Update(&ctx, (uint8_t*)tmp, keyLen);
char* pStr = base64_encode(ctx.digest, tListLen(ctx.digest));
strcpy(str, pStr);
}
@ -242,11 +241,11 @@ bool tscProjectionQueryOnMetric(SSqlCmd* pCmd) {
return false;
}
//for project query, only the following two function is allowed
// for project query, only the following two function is allowed
for (int32_t i = 0; i < pCmd->fieldsInfo.numOfOutputCols; ++i) {
int32_t functionId = tscSqlExprGet(pCmd, i)->functionId;
if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ &&
functionId != TSDB_FUNC_TAG && functionId != TSDB_FUNC_TS && functionId != TSDB_FUNC_ARITHM) {
if (functionId != TSDB_FUNC_PRJ && functionId != TSDB_FUNC_TAGPRJ && functionId != TSDB_FUNC_TAG &&
functionId != TSDB_FUNC_TS && functionId != TSDB_FUNC_ARITHM) {
return false;
}
}
@ -460,7 +459,7 @@ void tscDestroyDataBlock(STableDataBlocks* pDataBlock) {
tfree(pDataBlock->params);
// free the refcount for metermeta
taosRemoveDataFromCache(tscCacheHandle, (void**) &(pDataBlock->pMeterMeta), false);
taosRemoveDataFromCache(tscCacheHandle, (void**)&(pDataBlock->pMeterMeta), false);
tfree(pDataBlock);
}
@ -533,16 +532,16 @@ void* tscDestroyBlockArrayList(SDataBlockList* pList) {
}
int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock) {
SSqlCmd *pCmd = &pSql->cmd;
SSqlCmd* pCmd = &pSql->cmd;
assert(pDataBlock->pMeterMeta != NULL);
pCmd->count = pDataBlock->numOfMeters;
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0);
//set the correct metermeta object, the metermeta has been locked in pDataBlocks, so it must be in the cache
// set the correct metermeta object, the metermeta has been locked in pDataBlocks, so it must be in the cache
if (pMeterMetaInfo->pMeterMeta != pDataBlock->pMeterMeta) {
strcpy(pMeterMetaInfo->name, pDataBlock->meterId);
taosRemoveDataFromCache(tscCacheHandle, (void**) &(pMeterMetaInfo->pMeterMeta), false);
taosRemoveDataFromCache(tscCacheHandle, (void**)&(pMeterMetaInfo->pMeterMeta), false);
pMeterMetaInfo->pMeterMeta = pDataBlock->pMeterMeta;
pDataBlock->pMeterMeta = NULL; // delegate the ownership of metermeta to pMeterMetaInfo
@ -587,13 +586,18 @@ void tscFreeUnusedDataBlocks(SDataBlockList* pList) {
* @param rowSize
* @param startOffset
* @param name
* @param pMeterMeta the ownership of pMeterMeta should be transfer to STableDataBlocks
* @param dataBlocks
* @return
*/
STableDataBlocks* tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name) {
int32_t tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_t startOffset, const char* name,
STableDataBlocks** dataBlocks) {
STableDataBlocks* dataBuf = (STableDataBlocks*)calloc(1, sizeof(STableDataBlocks));
dataBuf->nAllocSize = (uint32_t) initialSize;
if (dataBuf == NULL) {
tscError("failed to allocated memory, reason:%s", strerror(errno));
return TSDB_CODE_CLI_OUT_OF_MEMORY;
}
dataBuf->nAllocSize = (uint32_t)initialSize;
dataBuf->pData = calloc(1, dataBuf->nAllocSize);
dataBuf->ordered = true;
dataBuf->prevTS = INT64_MIN;
@ -604,29 +608,43 @@ STableDataBlocks* tscCreateDataBlock(size_t initialSize, int32_t rowSize, int32_
strncpy(dataBuf->meterId, name, TSDB_METER_ID_LEN);
// sure that the metermeta must be in the local client cache
/*
* The metermeta may be released since the metermeta cache are completed clean by other thread
* due to operation such as drop database.
*/
dataBuf->pMeterMeta = taosGetDataFromCache(tscCacheHandle, dataBuf->meterId);
assert(dataBuf->pMeterMeta != NULL && initialSize > 0);
assert(initialSize > 0);
return dataBuf;
if (dataBuf->pMeterMeta == NULL) {
tfree(dataBuf);
return TSDB_CODE_QUERY_CACHE_ERASED;
} else {
*dataBlocks = dataBuf;
return TSDB_CODE_SUCCESS;
}
}
STableDataBlocks* tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList, int64_t id, int32_t size,
int32_t startOffset, int32_t rowSize, const char* tableId) {
STableDataBlocks* dataBuf = NULL;
int32_t tscGetDataBlockFromList(void* pHashList, SDataBlockList* pDataBlockList, int64_t id, int32_t size,
int32_t startOffset, int32_t rowSize, const char* tableId,
STableDataBlocks** dataBlocks) {
*dataBlocks = NULL;
STableDataBlocks** t1 = (STableDataBlocks**)taosGetIntHashData(pHashList, id);
STableDataBlocks** t1 = (STableDataBlocks**) taosGetIntHashData(pHashList, id);
if (t1 != NULL) {
dataBuf = *t1;
*dataBlocks = *t1;
}
if (dataBuf == NULL) {
dataBuf = tscCreateDataBlock((size_t)size, rowSize, startOffset, tableId);
dataBuf = *(STableDataBlocks**)taosAddIntHash(pHashList, id, (char*)&dataBuf);
tscAppendDataBlock(pDataBlockList, dataBuf);
if (*dataBlocks == NULL) {
int32_t ret = tscCreateDataBlock((size_t) size, rowSize, startOffset, tableId, dataBlocks);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
return dataBuf;
*dataBlocks = *(STableDataBlocks**)taosAddIntHash(pHashList, id, (char*)(*dataBlocks));
tscAppendDataBlock(pDataBlockList, *dataBlocks);
}
return TSDB_CODE_SUCCESS;
}
int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SDataBlockList* pTableDataBlockList) {
@ -638,9 +656,15 @@ int32_t tscMergeTableDataBlocks(SSqlObj* pSql, SDataBlockList* pTableDataBlockLi
for (int32_t i = 0; i < pTableDataBlockList->nSize; ++i) {
STableDataBlocks* pOneTableBlock = pTableDataBlockList->pData[i];
STableDataBlocks* dataBuf =
tscGetDataBlockFromList(pVnodeDataBlockHashList, pVnodeDataBlockList, pOneTableBlock->vgid, TSDB_PAYLOAD_SIZE,
tsInsertHeadSize, 0, pOneTableBlock->meterId);
STableDataBlocks* dataBuf = NULL;
int32_t ret = tscGetDataBlockFromList(pVnodeDataBlockHashList, pVnodeDataBlockList, pOneTableBlock->vgid,
TSDB_PAYLOAD_SIZE, tsInsertHeadSize, 0, pOneTableBlock->meterId, &dataBuf);
if (ret != TSDB_CODE_SUCCESS) {
tscError("%p failed to allocate the data buffer block for merging table data", pSql);
tscDestroyBlockArrayList(pTableDataBlockList);
return ret;
}
int64_t destSize = dataBuf->size + pOneTableBlock->size;
if (dataBuf->nAllocSize < destSize) {
@ -1155,7 +1179,7 @@ SColumnBase* tscColumnBaseInfoInsert(SSqlCmd* pCmd, SColumnIndex* pColIndex) {
}
void tscColumnFilterInfoCopy(SColumnFilterInfo* dst, const SColumnFilterInfo* src) {
assert (src != NULL && dst != NULL);
assert(src != NULL && dst != NULL);
assert(src->filterOnBinary == 0 || src->filterOnBinary == 1);
if (src->lowerRelOptr == TSDB_RELATION_INVALID && src->upperRelOptr == TSDB_RELATION_INVALID) {
@ -1164,15 +1188,15 @@ void tscColumnFilterInfoCopy(SColumnFilterInfo* dst, const SColumnFilterInfo* sr
*dst = *src;
if (dst->filterOnBinary) {
size_t len = (size_t) dst->len + 1;
size_t len = (size_t)dst->len + 1;
char* pTmp = calloc(1, len);
dst->pz = (int64_t) pTmp;
memcpy((char*) dst->pz, (char*) src->pz, (size_t) len);
dst->pz = (int64_t)pTmp;
memcpy((char*)dst->pz, (char*)src->pz, (size_t)len);
}
}
void tscColumnBaseCopy(SColumnBase* dst, const SColumnBase* src) {
assert (src != NULL && dst != NULL);
assert(src != NULL && dst != NULL);
*dst = *src;
@ -1230,7 +1254,7 @@ void tscColumnBaseInfoDestroy(SColumnBaseInfo* pColumnBaseInfo) {
assert(pColBase->filterInfo[j].filterOnBinary == 0 || pColBase->filterInfo[j].filterOnBinary == 1);
if (pColBase->filterInfo[j].filterOnBinary) {
free((char*) pColBase->filterInfo[j].pz);
free((char*)pColBase->filterInfo[j].pz);
pColBase->filterInfo[j].pz = 0;
}
}
@ -1780,13 +1804,15 @@ bool tscIsUpdateQuery(STscObj* pObj) {
SSqlCmd* pCmd = &pObj->pSql->cmd;
return ((pCmd->command >= TSDB_SQL_INSERT && pCmd->command <= TSDB_SQL_DROP_DNODE) ||
TSDB_SQL_USE_DB == pCmd->command) ? 1 : 0;
TSDB_SQL_USE_DB == pCmd->command)
? 1
: 0;
}
int32_t tscInvalidSQLErrMsg(char *msg, const char *additionalInfo, const char *sql) {
const char *msgFormat1 = "invalid SQL: %s";
const char *msgFormat2 = "invalid SQL: syntax error near \"%s\" (%s)";
const char *msgFormat3 = "invalid SQL: syntax error near \"%s\"";
int32_t tscInvalidSQLErrMsg(char* msg, const char* additionalInfo, const char* sql) {
const char* msgFormat1 = "invalid SQL: %s";
const char* msgFormat2 = "invalid SQL: syntax error near \"%s\" (%s)";
const char* msgFormat3 = "invalid SQL: syntax error near \"%s\"";
const int32_t BACKWARD_CHAR_STEP = 0;
@ -1816,4 +1842,3 @@ bool tscHasReachLimitation(SSqlObj* pSql) {
return (pCmd->globalLimit > 0 && pRes->numOfTotal >= pCmd->globalLimit);
}

3
src/inc/taoserror.h
View File

@ -137,8 +137,9 @@ extern "C" {
#define TSDB_CODE_INVALID_VNODE_STATUS 116
#define TSDB_CODE_FAILED_TO_LOCK_RESOURCES 117
#define TSDB_CODE_TABLE_ID_MISMATCH 118
#define TSDB_CODE_QUERY_CACHE_ERASED 119
#define TSDB_CODE_MAX_ERROR_CODE 119
#define TSDB_CODE_MAX_ERROR_CODE 120
#ifdef __cplusplus
}

3
src/rpc/src/tstring.c
View File

@ -242,5 +242,6 @@ char *tsError[] = {"success",
"invalid table id", // 115
"invalid vnode status",
"failed to lock resources",
"table id/uid mismatch", // 118
"table id/uid mismatch",
"client query cache erased", // 119
};

5
src/util/src/textbuffer.c
View File

@ -97,10 +97,11 @@ void tExtMemBufferDestroy(tExtMemBuffer **pMemBuffer) {
// close temp file
if ((*pMemBuffer)->dataFile != 0) {
int32_t ret = fclose((*pMemBuffer)->dataFile);
if (ret != 0) {
if (fclose((*pMemBuffer)->dataFile) != 0) {
pError("failed to close file:%s, reason:%s", (*pMemBuffer)->dataFilePath, strerror(errno));
}
pTrace("remove temp file:%s for external buffer", (*pMemBuffer)->dataFilePath);
unlink((*pMemBuffer)->dataFilePath);
}