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

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

This commit is contained in:
Shuduo Sang 2020-05-26 09:57:45 +08:00
parent 961ce678bb 7c31b143b2
commit 53e5f84d64
70 changed files with 3254 additions and 2664 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/client/inc/tsclient.h
View File

@ -210,7 +210,7 @@ typedef struct SQueryInfo {
int16_t numOfTables; int16_t numOfTables;
STableMetaInfo **pTableMetaInfo; STableMetaInfo **pTableMetaInfo;
struct STSBuf * tsBuf; struct STSBuf * tsBuf;
int64_t * defaultVal; // default value for interpolation int64_t * fillVal; // default value for interpolation
char * msg; // pointer to the pCmd->payload to keep error message temporarily char * msg; // pointer to the pCmd->payload to keep error message temporarily
int64_t clauseLimit; // limit for current sub clause int64_t clauseLimit; // limit for current sub clause

31
src/client/src/tscFunctionImpl.c
View File

@ -390,8 +390,12 @@ static void function_finalizer(SQLFunctionCtx *pCtx) {
if (pResInfo->hasResult != DATA_SET_FLAG) { if (pResInfo->hasResult != DATA_SET_FLAG) {
tscTrace("no result generated, result is set to NULL"); tscTrace("no result generated, result is set to NULL");
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes); setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
} }
}
doFinalizer(pCtx); doFinalizer(pCtx);
} }
@ -1864,12 +1868,22 @@ static void last_row_finalizer(SQLFunctionCtx *pCtx) {
SResultInfo *pResInfo = GET_RES_INFO(pCtx); SResultInfo *pResInfo = GET_RES_INFO(pCtx);
if (pCtx->currentStage == SECONDARY_STAGE_MERGE) { if (pCtx->currentStage == SECONDARY_STAGE_MERGE) {
if (pResInfo->hasResult != DATA_SET_FLAG) { if (pResInfo->hasResult != DATA_SET_FLAG) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes); setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
return; return;
} }
} else { } else {
if (pResInfo->hasResult != DATA_SET_FLAG) { if (pResInfo->hasResult != DATA_SET_FLAG) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes); setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
return; return;
} }
} }
@ -2885,7 +2899,12 @@ static void leastsquares_finalizer(SQLFunctionCtx *pCtx) {
SLeastsquareInfo *pInfo = pResInfo->interResultBuf; SLeastsquareInfo *pInfo = pResInfo->interResultBuf;
if (pInfo->num == 0) { if (pInfo->num == 0) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes); setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
return; return;
} }
@ -3139,7 +3158,7 @@ static void diff_function(SQLFunctionCtx *pCtx) {
pOutput += 1; pOutput += 1;
pTimestamp += 1; pTimestamp += 1;
} else { } else {
*pOutput = pData[i] - pCtx->param[1].i64Key; *pOutput = pData[i] - pCtx->param[1].dKey;
*pTimestamp = pCtx->ptsList[i]; *pTimestamp = pCtx->ptsList[i];
pOutput += 1; pOutput += 1;
pTimestamp += 1; pTimestamp += 1;
@ -3170,7 +3189,7 @@ static void diff_function(SQLFunctionCtx *pCtx) {
pOutput += 1; pOutput += 1;
pTimestamp += 1; pTimestamp += 1;
} else { } else {
*pOutput = pData[i] - pCtx->param[1].i64Key; *pOutput = pData[i] - pCtx->param[1].dKey;
*pTimestamp = pCtx->ptsList[i]; *pTimestamp = pCtx->ptsList[i];
pOutput += 1; pOutput += 1;
@ -3862,7 +3881,11 @@ static void interp_function(SQLFunctionCtx *pCtx) {
*(TSKEY *)pCtx->aOutputBuf = pInfoDetail->ts; *(TSKEY *)pCtx->aOutputBuf = pInfoDetail->ts;
} else { } else {
if (pInfoDetail->type == TSDB_FILL_NULL) { if (pInfoDetail->type == TSDB_FILL_NULL) {
if (pCtx->outputType == TSDB_DATA_TYPE_BINARY || pCtx->outputType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->outputType);
} else {
setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes); setNull(pCtx->aOutputBuf, pCtx->outputType, pCtx->outputBytes);
}
} else if (pInfoDetail->type == TSDB_FILL_SET_VALUE) { } else if (pInfoDetail->type == TSDB_FILL_SET_VALUE) {
tVariantDump(&pCtx->param[1], pCtx->aOutputBuf, pCtx->inputType); tVariantDump(&pCtx->param[1], pCtx->aOutputBuf, pCtx->inputType);
} else if (pInfoDetail->type == TSDB_FILL_PREV) { } else if (pInfoDetail->type == TSDB_FILL_PREV) {
@ -3913,11 +3936,15 @@ static void interp_function(SQLFunctionCtx *pCtx) {
taosDoLinearInterpolation(pCtx->outputType, &point1, &point2, &point); taosDoLinearInterpolation(pCtx->outputType, &point1, &point2, &point);
} }
} else {
if (srcType == TSDB_DATA_TYPE_BINARY || srcType == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(pCtx->aOutputBuf, pCtx->inputBytes);
} else { } else {
setNull(pCtx->aOutputBuf, srcType, pCtx->inputBytes); setNull(pCtx->aOutputBuf, srcType, pCtx->inputBytes);
} }
} }
} }
}
free(interpInfo.pInterpDetail); free(interpInfo.pInterpDetail);
} }

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

@ -312,8 +312,7 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SSQLToken *pToken, char *payload,
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
// binary data cannot be null-terminated char string, otherwise the last char of the string is lost // binary data cannot be null-terminated char string, otherwise the last char of the string is lost
if (pToken->type == TK_NULL) { if (pToken->type == TK_NULL) {
varDataSetLen(payload, sizeof(int8_t)); setVardataNull(payload, TSDB_DATA_TYPE_BINARY);
*(uint8_t*) varDataVal(payload) = TSDB_DATA_BINARY_NULL;
} else { // too long values will return invalid sql, not be truncated automatically } else { // too long values will return invalid sql, not be truncated automatically
if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) { //todo refactor if (pToken->n + VARSTR_HEADER_SIZE > pSchema->bytes) { //todo refactor
return tscInvalidSQLErrMsg(msg, "string data overflow", pToken->z); return tscInvalidSQLErrMsg(msg, "string data overflow", pToken->z);
@ -326,8 +325,7 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SSQLToken *pToken, char *payload,
case TSDB_DATA_TYPE_NCHAR: case TSDB_DATA_TYPE_NCHAR:
if (pToken->type == TK_NULL) { if (pToken->type == TK_NULL) {
varDataSetLen(payload, sizeof(int32_t)); setVardataNull(payload, TSDB_DATA_TYPE_NCHAR);
*(uint32_t*) varDataVal(payload) = TSDB_DATA_NCHAR_NULL;
} else { } else {
// if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long' // if the converted output len is over than pColumnModel->bytes, return error: 'Argument list too long'
size_t output = 0; size_t output = 0;

23
src/client/src/tscSQLParser.c
View File

@ -13,7 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#define _XOPEN_SOURCE #define _BSD_SOURCE
#define _XOPEN_SOURCE 500
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
#include "os.h" #include "os.h"
@ -4015,9 +4016,9 @@ int32_t parseFillClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySQL) {
size_t size = tscSqlExprNumOfExprs(pQueryInfo); size_t size = tscSqlExprNumOfExprs(pQueryInfo);
if (pQueryInfo->defaultVal == NULL) { if (pQueryInfo->fillVal == NULL) {
pQueryInfo->defaultVal = calloc(size, sizeof(int64_t)); pQueryInfo->fillVal = calloc(size, sizeof(int64_t));
if (pQueryInfo->defaultVal == NULL) { if (pQueryInfo->fillVal == NULL) {
return TSDB_CODE_CLI_OUT_OF_MEMORY; return TSDB_CODE_CLI_OUT_OF_MEMORY;
} }
} }
@ -4028,7 +4029,11 @@ int32_t parseFillClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySQL) {
pQueryInfo->fillType = TSDB_FILL_NULL; pQueryInfo->fillType = TSDB_FILL_NULL;
for (int32_t i = START_INTERPO_COL_IDX; i < size; ++i) { for (int32_t i = START_INTERPO_COL_IDX; i < size; ++i) {
TAOS_FIELD* pFields = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i); TAOS_FIELD* pFields = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
setNull((char*)&pQueryInfo->defaultVal[i], pFields->type, pFields->bytes); if (pFields->type == TSDB_DATA_TYPE_BINARY || pFields->type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull((char*) &pQueryInfo->fillVal[i], pFields->type);
} else {
setNull((char*)&pQueryInfo->fillVal[i], pFields->type, pFields->bytes);
};
} }
} else if (strncasecmp(pItem->pVar.pz, "prev", 4) == 0 && pItem->pVar.nLen == 4) { } else if (strncasecmp(pItem->pVar.pz, "prev", 4) == 0 && pItem->pVar.nLen == 4) {
pQueryInfo->fillType = TSDB_FILL_PREV; pQueryInfo->fillType = TSDB_FILL_PREV;
@ -4061,11 +4066,11 @@ int32_t parseFillClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySQL) {
TAOS_FIELD* pFields = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i); TAOS_FIELD* pFields = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
if (pFields->type == TSDB_DATA_TYPE_BINARY || pFields->type == TSDB_DATA_TYPE_NCHAR) { if (pFields->type == TSDB_DATA_TYPE_BINARY || pFields->type == TSDB_DATA_TYPE_NCHAR) {
setNull((char*)(&pQueryInfo->defaultVal[i]), pFields->type, pFields->bytes); setVardataNull((char*) &pQueryInfo->fillVal[i], pFields->type);
continue; continue;
} }
int32_t ret = tVariantDump(&pFillToken->a[j].pVar, (char*)&pQueryInfo->defaultVal[i], pFields->type); int32_t ret = tVariantDump(&pFillToken->a[j].pVar, (char*)&pQueryInfo->fillVal[i], pFields->type);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(pQueryInfo->msg, msg); return invalidSqlErrMsg(pQueryInfo->msg, msg);
} }
@ -4079,9 +4084,9 @@ int32_t parseFillClause(SQueryInfo* pQueryInfo, SQuerySQL* pQuerySQL) {
TAOS_FIELD* pFields = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i); TAOS_FIELD* pFields = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
if (pFields->type == TSDB_DATA_TYPE_BINARY || pFields->type == TSDB_DATA_TYPE_NCHAR) { if (pFields->type == TSDB_DATA_TYPE_BINARY || pFields->type == TSDB_DATA_TYPE_NCHAR) {
setNull((char*)(&pQueryInfo->defaultVal[i]), pFields->type, pFields->bytes); setVardataNull((char*) &pQueryInfo->fillVal[i], pFields->type);
} else { } else {
tVariantDump(&lastItem->pVar, (char*)&pQueryInfo->defaultVal[i], pFields->type); tVariantDump(&lastItem->pVar, (char*)&pQueryInfo->fillVal[i], pFields->type);
} }
} }
} }

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

@ -145,7 +145,7 @@ static SFillColInfo* createFillColInfo(SQueryInfo* pQueryInfo) {
pFillCol[i].flag = pExpr->colInfo.flag; pFillCol[i].flag = pExpr->colInfo.flag;
pFillCol[i].col.offset = offset; pFillCol[i].col.offset = offset;
pFillCol[i].functionId = pExpr->functionId; pFillCol[i].functionId = pExpr->functionId;
pFillCol[i].defaultVal.i = pQueryInfo->defaultVal[i]; pFillCol[i].fillVal.i = pQueryInfo->fillVal[i];
offset += pExpr->resBytes; offset += pExpr->resBytes;
} }
@ -946,8 +946,7 @@ static void doInterpolateResult(SSqlObj *pSql, SLocalReducer *pLocalReducer, boo
} }
while (1) { while (1) {
int64_t newRows = -1; int64_t newRows = taosGenerateDataBlock(pFillInfo, pResPages, pLocalReducer->resColModel->capacity);
taosGenerateDataBlock(pFillInfo, pResPages, &newRows, pLocalReducer->resColModel->capacity);
if (pQueryInfo->limit.offset < newRows) { if (pQueryInfo->limit.offset < newRows) {
newRows -= pQueryInfo->limit.offset; newRows -= pQueryInfo->limit.offset;

4
src/client/src/tscServer.c
View File

@ -781,8 +781,8 @@ int tscBuildQueryMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
if (pQueryInfo->fillType != TSDB_FILL_NONE) { if (pQueryInfo->fillType != TSDB_FILL_NONE) {
for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) { for (int32_t i = 0; i < pQueryInfo->fieldsInfo.numOfOutput; ++i) {
*((int64_t *)pMsg) = htobe64(pQueryInfo->defaultVal[i]); *((int64_t *)pMsg) = htobe64(pQueryInfo->fillVal[i]);
pMsg += sizeof(pQueryInfo->defaultVal[0]); pMsg += sizeof(pQueryInfo->fillVal[0]);
} }
} }

2
src/client/src/tscStream.c
View File

@ -223,7 +223,7 @@ static void tscProcessStreamRetrieveResult(void *param, TAOS_RES *res, int numOf
int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i); int16_t offset = tscFieldInfoGetOffset(pQueryInfo, i);
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i); TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->defaultVal[i]), pField->bytes, pField->type); assignVal(pSql->res.data + offset, (char *)(&pQueryInfo->fillVal[i]), pField->bytes, pField->type);
row[i] = pSql->res.data + offset; row[i] = pSql->res.data + offset;
} }

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

@ -281,7 +281,7 @@ void tscClearInterpInfo(SQueryInfo* pQueryInfo) {
} }
pQueryInfo->fillType = TSDB_FILL_NONE; pQueryInfo->fillType = TSDB_FILL_NONE;
tfree(pQueryInfo->defaultVal); tfree(pQueryInfo->fillVal);
} }
int32_t tscCreateResPointerInfo(SSqlRes* pRes, SQueryInfo* pQueryInfo) { int32_t tscCreateResPointerInfo(SSqlRes* pRes, SQueryInfo* pQueryInfo) {
@ -1616,7 +1616,7 @@ static void freeQueryInfoImpl(SQueryInfo* pQueryInfo) {
pQueryInfo->tsBuf = tsBufDestory(pQueryInfo->tsBuf); pQueryInfo->tsBuf = tsBufDestory(pQueryInfo->tsBuf);
tfree(pQueryInfo->defaultVal); tfree(pQueryInfo->fillVal);
} }
void tscClearSubqueryInfo(SSqlCmd* pCmd) { void tscClearSubqueryInfo(SSqlCmd* pCmd) {
@ -1768,7 +1768,7 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
pNewQueryInfo->order = pQueryInfo->order; pNewQueryInfo->order = pQueryInfo->order;
pNewQueryInfo->clauseLimit = pQueryInfo->clauseLimit; pNewQueryInfo->clauseLimit = pQueryInfo->clauseLimit;
pNewQueryInfo->pTableMetaInfo = NULL; pNewQueryInfo->pTableMetaInfo = NULL;
pNewQueryInfo->defaultVal = NULL; pNewQueryInfo->fillVal = NULL;
pNewQueryInfo->numOfTables = 0; pNewQueryInfo->numOfTables = 0;
pNewQueryInfo->tsBuf = NULL; pNewQueryInfo->tsBuf = NULL;
@ -1780,8 +1780,8 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
tscTagCondCopy(&pNewQueryInfo->tagCond, &pQueryInfo->tagCond); tscTagCondCopy(&pNewQueryInfo->tagCond, &pQueryInfo->tagCond);
if (pQueryInfo->fillType != TSDB_FILL_NONE) { if (pQueryInfo->fillType != TSDB_FILL_NONE) {
pNewQueryInfo->defaultVal = malloc(pQueryInfo->fieldsInfo.numOfOutput * sizeof(int64_t)); pNewQueryInfo->fillVal = malloc(pQueryInfo->fieldsInfo.numOfOutput * sizeof(int64_t));
memcpy(pNewQueryInfo->defaultVal, pQueryInfo->defaultVal, pQueryInfo->fieldsInfo.numOfOutput * sizeof(int64_t)); memcpy(pNewQueryInfo->fillVal, pQueryInfo->fillVal, pQueryInfo->fieldsInfo.numOfOutput * sizeof(int64_t));
} }
if (tscAllocPayload(pnCmd, TSDB_DEFAULT_PAYLOAD_SIZE) != TSDB_CODE_SUCCESS) { if (tscAllocPayload(pnCmd, TSDB_DEFAULT_PAYLOAD_SIZE) != TSDB_CODE_SUCCESS) {

12
src/common/src/tglobal.c
View File

@ -202,6 +202,8 @@ char tsTimezone[64] = {0};
char tsLocale[TSDB_LOCALE_LEN] = {0}; char tsLocale[TSDB_LOCALE_LEN] = {0};
char tsCharset[TSDB_LOCALE_LEN] = {0}; // default encode string char tsCharset[TSDB_LOCALE_LEN] = {0}; // default encode string
int32_t tsMaxBinaryDisplayWidth = 30;
static pthread_once_t tsInitGlobalCfgOnce = PTHREAD_ONCE_INIT; static pthread_once_t tsInitGlobalCfgOnce = PTHREAD_ONCE_INIT;
void taosSetAllDebugFlag() { void taosSetAllDebugFlag() {
@ -1227,6 +1229,16 @@ static void doInitGlobalConfig() {
cfg.ptrLength = 0; cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE; cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg); taosInitConfigOption(cfg);
cfg.option = "maxBinaryDisplayWidth";
cfg.ptr = &tsMaxBinaryDisplayWidth;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 1;
cfg.maxValue = 0x7fffffff;
cfg.ptrLength = 0;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
} }
void taosInitGlobalCfg() { void taosInitGlobalCfg() {

14
src/common/src/ttypes.c
View File

@ -381,6 +381,18 @@ bool isNull(const char *val, int32_t type) {
}; };
} }
void setVardataNull(char* val, int32_t type) {
if (type == TSDB_DATA_TYPE_BINARY) {
varDataSetLen(val, sizeof(int8_t));
*(uint8_t*) varDataVal(val) = TSDB_DATA_BINARY_NULL;
} else if (type == TSDB_DATA_TYPE_NCHAR) {
varDataSetLen(val, sizeof(int32_t));
*(uint32_t*) varDataVal(val) = TSDB_DATA_NCHAR_NULL;
} else {
assert(0);
}
}
void setNull(char *val, int32_t type, int32_t bytes) { setNullN(val, type, bytes, 1); } void setNull(char *val, int32_t type, int32_t bytes) { setNullN(val, type, bytes, 1); }
void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems) { void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems) {
@ -483,7 +495,7 @@ void assignVal(char *val, const char *src, int32_t len, int32_t type) {
break; break;
}; };
case TSDB_DATA_TYPE_NCHAR: { case TSDB_DATA_TYPE_NCHAR: {
wcsncpy((wchar_t*)val, (wchar_t*)src, len / TSDB_NCHAR_SIZE); varDataCopy(val, src);
break; break;
}; };
default: { default: {

2
src/connector/jdbc/src/main/java/com/taosdata/jdbc/TSDBDriver.java
View File

@ -188,7 +188,7 @@ public class TSDBDriver implements java.sql.Driver {
} }
public boolean acceptsURL(String url) throws SQLException { public boolean acceptsURL(String url) throws SQLException {
return true; return StringUtils.isNotBlank(url) && url.startsWith(URL_PREFIX);
} }
public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) throws SQLException { public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) throws SQLException {

1
src/inc/taosdef.h
View File

@ -157,6 +157,7 @@ extern tDataTypeDescriptor tDataTypeDesc[11];
bool isValidDataType(int32_t type, int32_t length); bool isValidDataType(int32_t type, int32_t length);
bool isNull(const char *val, int32_t type); bool isNull(const char *val, int32_t type);
void setVardataNull(char* val, int32_t type);
void setNull(char *val, int32_t type, int32_t bytes); void setNull(char *val, int32_t type, int32_t bytes);
void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems); void setNullN(char *val, int32_t type, int32_t bytes, int32_t numOfElems);

2
src/inc/taosmsg.h
View File

@ -443,7 +443,7 @@ typedef struct {
int16_t numOfOutput; // final output columns numbers int16_t numOfOutput; // final output columns numbers
int16_t tagNameRelType; // relation of tag criteria and tbname criteria int16_t tagNameRelType; // relation of tag criteria and tbname criteria
int16_t fillType; // interpolate type int16_t fillType; // interpolate type
uint64_t defaultVal; // default value array list uint64_t fillVal; // default value array list
int32_t tsOffset; // offset value in current msg body, NOTE: ts list is compressed int32_t tsOffset; // offset value in current msg body, NOTE: ts list is compressed
int32_t tsLen; // total length of ts comp block int32_t tsLen; // total length of ts comp block
int32_t tsNumOfBlocks; // ts comp block numbers int32_t tsNumOfBlocks; // ts comp block numbers

14
src/kit/shell/inc/shell.h
View File

@ -29,18 +29,6 @@
#define MAX_COMMAND_SIZE 65536 #define MAX_COMMAND_SIZE 65536
#define HISTORY_FILE ".taos_history" #define HISTORY_FILE ".taos_history"
#define BOOL_OUTPUT_LENGTH 6
#define TINYINT_OUTPUT_LENGTH 6
#define SMALLINT_OUTPUT_LENGTH 7
#define INT_OUTPUT_LENGTH 11
#define BIGINT_OUTPUT_LENGTH 21
#define FLOAT_OUTPUT_LENGTH 20
#define DOUBLE_OUTPUT_LENGTH 25
#define BINARY_OUTPUT_LENGTH 20
// dynamic config timestamp width according to maximum time precision
extern int32_t TIMESTAMP_OUTPUT_LENGTH;
typedef struct SShellHistory { typedef struct SShellHistory {
char* hist[MAX_HISTORY_SIZE]; char* hist[MAX_HISTORY_SIZE];
int hstart; int hstart;
@ -80,7 +68,7 @@ void get_history_path(char* history);
void cleanup_handler(void* arg); void cleanup_handler(void* arg);
void exitShell(); void exitShell();
int shellDumpResult(TAOS* con, char* fname, int* error_no, bool printMode); int shellDumpResult(TAOS* con, char* fname, int* error_no, bool printMode);
void shellPrintNChar(char* str, int width, bool printMode); void shellPrintNChar(const char* str, int length, int width);
void shellGetGrantInfo(void *con); void shellGetGrantInfo(void *con);
int isCommentLine(char *line); int isCommentLine(char *line);

40
src/kit/shell/src/shellDarwin.c
View File

@ -352,36 +352,30 @@ void *shellLoopQuery(void *arg) {
return NULL; return NULL;
} }
void shellPrintNChar(char *str, int width, bool printMode) { void shellPrintNChar(const char *str, int length, int width) {
int col_left = width; int pos = 0, cols = 0;
while (pos < length) {
wchar_t wc; wchar_t wc;
while (col_left > 0) { pos += mbtowc(&wc, str + pos, MB_CUR_MAX);
if (*str == '\0') break; if (pos > length) {
char *tstr = str; break;
int byte_width = mbtowc(&wc, tstr, MB_CUR_MAX); }
if (byte_width <= 0) break;
int col_width = wcwidth(wc); int w = wcwidth(wc);
if (col_width <= 0) { if (w > 0) {
str += byte_width; if (width > 0 && cols + w > width) {
continue; break;
} }
if (col_left < col_width) break;
printf("%lc", wc); printf("%lc", wc);
str += byte_width; cols += w;
col_left -= col_width; }
} }
while (col_left > 0) { for (; cols < width; cols++) {
printf(" "); putchar(' ');
col_left--; }
} }
if (!printMode) {
printf("|");
} else {
printf("\n");
}
}
int get_old_terminal_mode(struct termios *tio) { int get_old_terminal_mode(struct termios *tio) {
/* Make sure stdin is a terminal. */ /* Make sure stdin is a terminal. */

604
src/kit/shell/src/shellEngine.c
View File

@ -13,6 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#define _BSD_SOURCE
#define _GNU_SOURCE
#define _XOPEN_SOURCE #define _XOPEN_SOURCE
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE
@ -35,6 +37,9 @@ int prompt_size = 6;
TAOS_RES *result = NULL; TAOS_RES *result = NULL;
SShellHistory history; SShellHistory history;
#define DEFAULT_MAX_BINARY_DISPLAY_WIDTH 30
extern int32_t tsMaxBinaryDisplayWidth;
/* /*
* FUNCTION: Initialize the shell. * FUNCTION: Initialize the shell.
*/ */
@ -195,7 +200,15 @@ int32_t shellRunCommand(TAOS *con, char *command) {
} else if (regex_match(command, "^[\t ]*clear[ \t;]*$", REG_EXTENDED | REG_ICASE)) { } else if (regex_match(command, "^[\t ]*clear[ \t;]*$", REG_EXTENDED | REG_ICASE)) {
// If clear the screen. // If clear the screen.
system("clear"); system("clear");
return 0; } else if (regex_match(command, "^[\t ]*set[ \t]+max_binary_display_width[ \t]+(default|[1-9][0-9]*)[ \t;]*$", REG_EXTENDED | REG_ICASE)) {
strtok(command, " \t");
strtok(NULL, " \t");
char* p = strtok(NULL, " \t");
if (strcasecmp(p, "default") == 0) {
tsMaxBinaryDisplayWidth = DEFAULT_MAX_BINARY_DISPLAY_WIDTH;
} else {
tsMaxBinaryDisplayWidth = atoi(p);
}
} else if (regex_match(command, "^[ \t]*source[\t ]+[^ ]+[ \t;]*$", REG_EXTENDED | REG_ICASE)) { } else if (regex_match(command, "^[ \t]*source[\t ]+[^ ]+[ \t;]*$", REG_EXTENDED | REG_ICASE)) {
/* If source file. */ /* If source file. */
char *c_ptr = strtok(command, " ;"); char *c_ptr = strtok(command, " ;");
@ -310,32 +323,91 @@ int regex_match(const char *s, const char *reg, int cflags) {
return 0; return 0;
} }
int shellDumpResult(TAOS *con, char *fname, int *error_no, bool printMode) {
TAOS_ROW row = NULL;
int numOfRows = 0;
time_t tt;
char buf[25] = "\0";
struct tm *ptm;
int output_bytes = 0;
FILE * fp = NULL;
int num_fields = taos_field_count(con);
wordexp_t full_path;
assert(num_fields != 0); static char* formatTimestamp(char* buf, int64_t val, int precision) {
if (args.is_raw_time) {
result = taos_use_result(con); sprintf(buf, "%" PRId64, val);
if (result == NULL) { return buf;
taos_error(con);
return -1;
} }
if (fname != NULL) { time_t tt;
if (precision == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)(val / 1000000);
} else {
tt = (time_t)(val / 1000);
}
struct tm* ptm = localtime(&tt);
size_t pos = strftime(buf, 32, "%Y-%m-%d %H:%M:%S", ptm);
if (precision == TSDB_TIME_PRECISION_MICRO) {
sprintf(buf + pos, ".%06d", (int)(val % 1000000));
} else {
sprintf(buf + pos, ".%03d", (int)(val % 1000));
}
return buf;
}
static void dumpFieldToFile(FILE* fp, const char* val, TAOS_FIELD* field, int32_t length, int precision) {
if (val == NULL) {
fprintf(fp, "%s", TSDB_DATA_NULL_STR);
return;
}
char buf[TSDB_MAX_BYTES_PER_ROW];
switch (field->type) {
case TSDB_DATA_TYPE_BOOL:
fprintf(fp, "%d", ((((int)(*((char *)val))) == 1) ? 1 : 0));
break;
case TSDB_DATA_TYPE_TINYINT:
fprintf(fp, "%d", (int)(*((char *)val)));
break;
case TSDB_DATA_TYPE_SMALLINT:
fprintf(fp, "%d", (int)(*((short *)val)));
break;
case TSDB_DATA_TYPE_INT:
fprintf(fp, "%d", *((int *)val));
break;
case TSDB_DATA_TYPE_BIGINT:
fprintf(fp, "%" PRId64, *((int64_t *)val));
break;
case TSDB_DATA_TYPE_FLOAT:
fprintf(fp, "%.5f", GET_FLOAT_VAL(val));
break;
case TSDB_DATA_TYPE_DOUBLE:
fprintf(fp, "%.9f", GET_DOUBLE_VAL(val));
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
memcpy(buf, val, length);
buf[length] = 0;
fprintf(fp, "\'%s\'", buf);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
formatTimestamp(buf, *(int64_t*)val, precision);
fprintf(fp, "'%s'", buf);
break;
default:
break;
}
}
static int dumpResultToFile(const char* fname, TAOS_RES* result) {
TAOS_ROW row = taos_fetch_row(result);
if (row == NULL) {
return 0;
}
wordexp_t full_path;
if (wordexp(fname, &full_path, 0) != 0) { if (wordexp(fname, &full_path, 0) != 0) {
fprintf(stderr, "ERROR: invalid file name: %s\n", fname); fprintf(stderr, "ERROR: invalid file name: %s\n", fname);
return -1; return -1;
} }
fp = fopen(full_path.we_wordv[0], "w"); FILE* fp = fopen(full_path.we_wordv[0], "w");
if (fp == NULL) { if (fp == NULL) {
fprintf(stderr, "ERROR: failed to open file: %s\n", full_path.we_wordv[0]); fprintf(stderr, "ERROR: failed to open file: %s\n", full_path.we_wordv[0]);
wordfree(&full_path); wordfree(&full_path);
@ -343,327 +415,255 @@ int shellDumpResult(TAOS *con, char *fname, int *error_no, bool printMode) {
} }
wordfree(&full_path); wordfree(&full_path);
int num_fields = taos_num_fields(result);
TAOS_FIELD *fields = taos_fetch_fields(result);
int32_t* length = taos_fetch_lengths(result);
int precision = taos_result_precision(result);
for (int col = 0; col < num_fields; col++) {
if (col > 0) {
fprintf(fp, ",");
}
fprintf(fp, "%s", fields[col].name);
}
fputc('\n', fp);
int numOfRows = 0;
do {
for (int i = 0; i < num_fields; i++) {
if (i > 0) {
fputc(',', fp);
}
dumpFieldToFile(fp, row[i], fields +i, length[i], precision);
}
fputc('\n', fp);
numOfRows++;
row = taos_fetch_row(result);
} while( row != NULL);
fclose(fp);
return numOfRows;
} }
TAOS_FIELD *fields = taos_fetch_fields(result);
row = taos_fetch_row(result); static void printField(const char* val, TAOS_FIELD* field, int width, int32_t length, int precision) {
int32_t* length = taos_fetch_lengths(result); if (val == NULL) {
int w = width;
if (field->type < TSDB_DATA_TYPE_TINYINT || field->type > TSDB_DATA_TYPE_DOUBLE) {
w = 0;
}
w = printf("%*s", w, TSDB_DATA_NULL_STR);
for (; w < width; w++) {
putchar(' ');
}
return;
}
char t_str[TSDB_MAX_BYTES_PER_ROW] = "\0"; char buf[TSDB_MAX_BYTES_PER_ROW];
int l[TSDB_MAX_COLUMNS] = {0}; switch (field->type) {
int maxLenColumnName = 0;
if (row) {
// Print the header indicator
if (fname == NULL) { // print to standard output
if (!printMode) {
for (int col = 0; col < num_fields; col++) {
switch (fields[col].type) {
case TSDB_DATA_TYPE_BOOL: case TSDB_DATA_TYPE_BOOL:
l[col] = MAX(BOOL_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*s", width, ((((int)(*((char *)val))) == 1) ? "true" : "false"));
break; break;
case TSDB_DATA_TYPE_TINYINT: case TSDB_DATA_TYPE_TINYINT:
l[col] = MAX(TINYINT_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*d", width, (int)(*((char *)val)));
break; break;
case TSDB_DATA_TYPE_SMALLINT: case TSDB_DATA_TYPE_SMALLINT:
l[col] = MAX(SMALLINT_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*d", width, (int)(*((short *)val)));
break; break;
case TSDB_DATA_TYPE_INT: case TSDB_DATA_TYPE_INT:
l[col] = MAX(INT_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*d", width, *((int *)val));
break; break;
case TSDB_DATA_TYPE_BIGINT: case TSDB_DATA_TYPE_BIGINT:
l[col] = MAX(BIGINT_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*" PRId64, width, *((int64_t *)val));
break; break;
case TSDB_DATA_TYPE_FLOAT: case TSDB_DATA_TYPE_FLOAT:
l[col] = MAX(FLOAT_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*.5f", width, GET_FLOAT_VAL(val));
break; break;
case TSDB_DATA_TYPE_DOUBLE: case TSDB_DATA_TYPE_DOUBLE:
l[col] = MAX(DOUBLE_OUTPUT_LENGTH, strlen(fields[col].name)); printf("%*.9f", width, GET_DOUBLE_VAL(val));
break; break;
case TSDB_DATA_TYPE_BINARY: case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR: case TSDB_DATA_TYPE_NCHAR:
l[col] = MAX(fields[col].bytes, strlen(fields[col].name)); shellPrintNChar(val, length, width);
/* l[col] = max(BINARY_OUTPUT_LENGTH, strlen(fields[col].name)); */
break;
case TSDB_DATA_TYPE_TIMESTAMP: {
int32_t defaultWidth = TIMESTAMP_OUTPUT_LENGTH;
if (args.is_raw_time) {
defaultWidth = 14;
}
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
defaultWidth += 3;
}
l[col] = MAX(defaultWidth, strlen(fields[col].name));
break;
}
default:
break;
}
int spaces = (int)(l[col] - strlen(fields[col].name));
int left_space = spaces / 2;
int right_space = (spaces % 2 ? left_space + 1 : left_space);
printf("%*.s%s%*.s|", left_space, " ", fields[col].name, right_space, " ");
output_bytes += (l[col] + 1);
}
printf("\n");
for (int k = 0; k < output_bytes; k++) printf("=");
printf("\n");
} else {
for (int col = 0; col < num_fields; col++) {
if (strlen(fields[col].name) > maxLenColumnName) maxLenColumnName = strlen(fields[col].name);
}
}
// print the elements
do {
if (!printMode) {
for (int i = 0; i < num_fields; i++) {
if (row[i] == NULL) {
printf("%*s|", l[i], TSDB_DATA_NULL_STR);
continue;
}
switch (fields[i].type) {
case TSDB_DATA_TYPE_BOOL:
printf("%*s|", l[i], ((((int)(*((char *)row[i]))) == 1) ? "true" : "false"));
break;
case TSDB_DATA_TYPE_TINYINT:
printf("%*d|", l[i], (int)(*((char *)row[i])));
break;
case TSDB_DATA_TYPE_SMALLINT:
printf("%*d|", l[i], (int)(*((short *)row[i])));
break;
case TSDB_DATA_TYPE_INT:
printf("%*d|", l[i], *((int *)row[i]));
break;
case TSDB_DATA_TYPE_BIGINT:
printf("%*" PRId64 "|", l[i], *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_FLOAT: {
float fv = 0;
fv = GET_FLOAT_VAL(row[i]);
printf("%*.5f|", l[i], fv);
}
break;
case TSDB_DATA_TYPE_DOUBLE: {
double dv = 0;
dv = GET_DOUBLE_VAL(row[i]);
printf("%*.9f|", l[i], dv);
}
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
memset(t_str, 0, TSDB_MAX_BYTES_PER_ROW);
memcpy(t_str, row[i], length[i]);
/* printf("%-*s|",max(fields[i].bytes, strlen(fields[i].name)),
* t_str); */
/* printf("%-*s|", l[i], t_str); */
shellPrintNChar(t_str, l[i], printMode);
break; break;
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
if (args.is_raw_time) { formatTimestamp(buf, *(int64_t*)val, precision);
printf(" %" PRId64 "|", *(int64_t *)row[i]); printf("%s", buf);
} else {
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)((*(int64_t *)row[i]) / 1000000);
} else {
tt = (time_t)((*(int64_t *)row[i]) / 1000);
}
ptm = localtime(&tt);
strftime(buf, 64, "%y-%m-%d %H:%M:%S", ptm);
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
printf(" %s.%06d|", buf, (int)(*(int64_t *)row[i] % 1000000));
} else {
printf(" %s.%03d|", buf, (int)(*(int64_t *)row[i] % 1000));
}
}
break; break;
default: default:
break; break;
} }
} }
printf("\n");
} else {
static int verticalPrintResult(TAOS_RES* result) {
TAOS_ROW row = taos_fetch_row(result);
if (row == NULL) {
return 0;
}
int num_fields = taos_num_fields(result);
TAOS_FIELD *fields = taos_fetch_fields(result);
int32_t* length = taos_fetch_lengths(result);
int precision = taos_result_precision(result);
int maxColNameLen = 0;
for (int col = 0; col < num_fields; col++) {
int len = strlen(fields[col].name);
if (len > maxColNameLen) {
maxColNameLen = len;
}
}
int numOfRows = 0;
do {
printf("*************************** %d.row ***************************\n", numOfRows + 1); printf("*************************** %d.row ***************************\n", numOfRows + 1);
for (int i = 0; i < num_fields; i++) { for (int i = 0; i < num_fields; i++) {
// 1. print column name TAOS_FIELD* field = fields + i;
int left_space = (int)(maxLenColumnName - strlen(fields[i].name));
printf("%*.s%s: ", left_space, " ", fields[i].name);
// 2. print column value int padding = (int)(maxColNameLen - strlen(field->name));
if (row[i] == NULL) { printf("%*.s%s: ", padding, " ", field->name);
printf("%s\n", TSDB_DATA_NULL_STR);
continue;
}
switch (fields[i].type) { printField(row[i], field, 0, length[i], precision);
case TSDB_DATA_TYPE_BOOL: putchar('\n');
printf("%s\n", ((((int)(*((char *)row[i]))) == 1) ? "true" : "false"));
break;
case TSDB_DATA_TYPE_TINYINT:
printf("%d\n", (int)(*((char *)row[i])));
break;
case TSDB_DATA_TYPE_SMALLINT:
printf("%d\n", (int)(*((short *)row[i])));
break;
case TSDB_DATA_TYPE_INT:
printf("%d\n", *((int *)row[i]));
break;
case TSDB_DATA_TYPE_BIGINT:
printf("%" PRId64 "\n", *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_FLOAT: {
float fv = 0;
fv = GET_FLOAT_VAL(row[i]);
printf("%.5f\n", fv);
}
break;
case TSDB_DATA_TYPE_DOUBLE: {
double dv = 0;
dv = GET_DOUBLE_VAL(row[i]);
printf("%.9f\n", dv);
}
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
memset(t_str, 0, TSDB_MAX_BYTES_PER_ROW);
memcpy(t_str, row[i], length[i]);
l[i] = MAX(fields[i].bytes, strlen(fields[i].name));
shellPrintNChar(t_str, l[i], printMode);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
if (args.is_raw_time) {
printf("%" PRId64 "\n", *(int64_t *)row[i]);
} else {
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)((*(int64_t *)row[i]) / 1000000);
} else {
tt = (time_t)((*(int64_t *)row[i]) / 1000);
}
ptm = localtime(&tt);
strftime(buf, 64, "%y-%m-%d %H:%M:%S", ptm);
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
printf("%s.%06d\n", buf, (int)(*(int64_t *)row[i] % 1000000));
} else {
printf("%s.%03d\n", buf, (int)(*(int64_t *)row[i] % 1000));
}
}
break;
default:
break;
}
}
} }
numOfRows++; numOfRows++;
} while ((row = taos_fetch_row(result))); row = taos_fetch_row(result);
} while(row != NULL);
} else { // dump to file
// first write column
for (int col = 0; col < num_fields; col++) {
fprintf(fp, "%s", fields[col].name);
if (col < num_fields - 1) {
fprintf(fp, ",");
} else {
fprintf(fp, "\n");
}
}
do {
for (int i = 0; i < num_fields; i++) {
if (row[i]) {
switch (fields[i].type) {
case TSDB_DATA_TYPE_BOOL:
fprintf(fp, "%d", ((((int)(*((char *)row[i]))) == 1) ? 1 : 0));
break;
case TSDB_DATA_TYPE_TINYINT:
fprintf(fp, "%d", (int)(*((char *)row[i])));
break;
case TSDB_DATA_TYPE_SMALLINT:
fprintf(fp, "%d", (int)(*((short *)row[i])));
break;
case TSDB_DATA_TYPE_INT:
fprintf(fp, "%d", *((int *)row[i]));
break;
case TSDB_DATA_TYPE_BIGINT:
fprintf(fp, "%" PRId64, *((int64_t *)row[i]));
break;
case TSDB_DATA_TYPE_FLOAT: {
float fv = 0;
fv = GET_FLOAT_VAL(row[i]);
fprintf(fp, "%.5f", fv);
}
break;
case TSDB_DATA_TYPE_DOUBLE: {
double dv = 0;
dv = GET_DOUBLE_VAL(row[i]);
fprintf(fp, "%.9f", dv);
}
break;
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
memset(t_str, 0, TSDB_MAX_BYTES_PER_ROW);
memcpy(t_str, row[i], length[i]);
fprintf(fp, "\'%s\'", t_str);
break;
case TSDB_DATA_TYPE_TIMESTAMP:
if (args.is_raw_time) {
fprintf(fp, "%" PRId64, *(int64_t *)row[i]);
} else {
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
tt = (time_t)((*(int64_t *)row[i]) / 1000000);
} else {
tt = (time_t)((*(int64_t *)row[i]) / 1000);
}
ptm = localtime(&tt);
strftime(buf, 64, "%Y-%m-%d %H:%M:%S", ptm);
if (taos_result_precision(result) == TSDB_TIME_PRECISION_MICRO) {
fprintf(fp, "\'%s.%06d\'", buf, (int)(*(int64_t *)row[i] % 1000000));
} else {
fprintf(fp, "\'%s.%03d\'", buf, (int)(*(int64_t *)row[i] % 1000));
}
}
break;
default:
break;
}
} else {
fprintf(fp, "%s", TSDB_DATA_NULL_STR);
}
if (i < num_fields - 1) {
fprintf(fp, ",");
} else {
fprintf(fp, "\n");
}
}
numOfRows++;
} while ((row = taos_fetch_row(result)));
}
}
*error_no = taos_errno(con);
taos_free_result(result);
result = NULL;
if (fname != NULL) {
fclose(fp);
}
return numOfRows; return numOfRows;
} }
static int calcColWidth(TAOS_FIELD* field, int precision) {
int width = strlen(field->name);
switch (field->type) {
case TSDB_DATA_TYPE_BOOL:
return MAX(5, width); // 'false'
case TSDB_DATA_TYPE_TINYINT:
return MAX(4, width); // '-127'
case TSDB_DATA_TYPE_SMALLINT:
return MAX(6, width); // '-32767'
case TSDB_DATA_TYPE_INT:
return MAX(11, width); // '-2147483648'
case TSDB_DATA_TYPE_BIGINT:
return MAX(21, width); // '-9223372036854775807'
case TSDB_DATA_TYPE_FLOAT:
return MAX(20, width);
case TSDB_DATA_TYPE_DOUBLE:
return MAX(25, width);
case TSDB_DATA_TYPE_BINARY:
case TSDB_DATA_TYPE_NCHAR:
if (field->bytes > tsMaxBinaryDisplayWidth) {
return MAX(tsMaxBinaryDisplayWidth, width);
} else {
return MAX(field->bytes, width);
}
case TSDB_DATA_TYPE_TIMESTAMP:
if (args.is_raw_time) {
return MAX(14, width);
} else if (precision == TSDB_TIME_PRECISION_MICRO) {
return MAX(26, width); // '2020-01-01 00:00:00.000000'
} else {
return MAX(23, width); // '2020-01-01 00:00:00.000'
}
default:
assert(false);
}
return 0;
}
static void printHeader(TAOS_FIELD* fields, int* width, int num_fields) {
int rowWidth = 0;
for (int col = 0; col < num_fields; col++) {
TAOS_FIELD* field = fields + col;
int padding = (int)(width[col] - strlen(field->name));
int left = padding / 2;
printf(" %*.s%s%*.s |", left, " ", field->name, padding - left, " ");
rowWidth += width[col] + 3;
}
putchar('\n');
for (int i = 0; i < rowWidth; i++) {
putchar('=');
}
putchar('\n');
}
static int horizontalPrintResult(TAOS_RES* result) {
TAOS_ROW row = taos_fetch_row(result);
if (row == NULL) {
return 0;
}
int num_fields = taos_num_fields(result);
TAOS_FIELD *fields = taos_fetch_fields(result);
int32_t* length = taos_fetch_lengths(result);
int precision = taos_result_precision(result);
int width[TSDB_MAX_COLUMNS];
for (int col = 0; col < num_fields; col++) {
width[col] = calcColWidth(fields + col, precision);
}
printHeader(fields, width, num_fields);
int numOfRows = 0;
do {
for (int i = 0; i < num_fields; i++) {
putchar(' ');
printField(row[i], fields + i, width[i], length[i], precision);
putchar(' ');
putchar('|');
}
putchar('\n');
numOfRows++;
row = taos_fetch_row(result);
} while(row != NULL);
return numOfRows;
}
int shellDumpResult(TAOS *con, char *fname, int *error_no, bool vertical) {
int numOfRows = 0;
TAOS_RES* result = taos_use_result(con);
if (result == NULL) {
taos_error(con);
return -1;
}
if (fname != NULL) {
numOfRows = dumpResultToFile(fname, result);
} else if(vertical) {
numOfRows = verticalPrintResult(result);
} else {
numOfRows = horizontalPrintResult(result);
}
*error_no = taos_errno(con);
taos_free_result(result);
return numOfRows;
}
void read_history() { void read_history() {
// Initialize history // Initialize history
memset(history.hist, 0, sizeof(char *) * MAX_HISTORY_SIZE); memset(history.hist, 0, sizeof(char *) * MAX_HISTORY_SIZE);

1
src/kit/shell/src/shellImport.c
View File

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#define _GNU_SOURCE
#define _XOPEN_SOURCE #define _XOPEN_SOURCE
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE

39
src/kit/shell/src/shellLinux.c
View File

@ -329,34 +329,27 @@ void *shellLoopQuery(void *arg) {
return NULL; return NULL;
} }
void shellPrintNChar(char *str, int width, bool printMode) { void shellPrintNChar(const char *str, int length, int width) {
int col_left = width; int pos = 0, cols = 0;
while (pos < length) {
wchar_t wc; wchar_t wc;
while (col_left > 0) { pos += mbtowc(&wc, str + pos, MB_CUR_MAX);
if (*str == '\0') break; if (pos > length) {
char *tstr = str; break;
int byte_width = mbtowc(&wc, tstr, MB_CUR_MAX); }
if (byte_width <= 0) break;
int col_width = wcwidth(wc); int w = wcwidth(wc);
if (col_width <= 0) { if (w > 0) {
str += byte_width; if (width > 0 && cols + w > width) {
continue; break;
} }
if (col_left < col_width) break;
printf("%lc", wc); printf("%lc", wc);
str += byte_width; cols += w;
col_left -= col_width; }
} }
while (col_left > 0) { for (; cols < width; cols++) {
printf(" "); putchar(' ');
col_left--;
}
if (!printMode) {
printf("|");
} else {
printf("\n");
} }
} }

1
src/kit/shell/src/shellMain.c
View File

@ -20,7 +20,6 @@
TAOS* con; TAOS* con;
pthread_t pid; pthread_t pid;
int32_t TIMESTAMP_OUTPUT_LENGTH = 22;
// TODO: IMPLEMENT INTERRUPT HANDLER. // TODO: IMPLEMENT INTERRUPT HANDLER.
void interruptHandler(int signum) { void interruptHandler(int signum) {

38
src/kit/shell/src/shellWindows.c
View File

@ -217,31 +217,31 @@ void *shellLoopQuery(void *arg) {
return NULL; return NULL;
} }
void shellPrintNChar(char *str, int width, bool printMode) { void shellPrintNChar(const char *str, int length, int width) {
int col_left = width; int pos = 0, cols = 0;
while (pos < length) {
wchar_t wc; wchar_t wc;
while (col_left > 0) { int bytes = mbtowc(&wc, str + pos, MB_CUR_MAX);
if (*str == '\0') break; pos += bytes;
char *tstr = str; if (pos > length) {
int byte_width = mbtowc(&wc, tstr, MB_CUR_MAX); break;
int col_width = byte_width; }
if (col_left < col_width) break;
int w = bytes;
if (w > 0) {
if (width > 0 && cols + w > width) {
break;
}
printf("%lc", wc); printf("%lc", wc);
str += byte_width; cols += w;
col_left -= col_width; }
} }
while (col_left > 0) { for (; cols < width; cols++) {
printf(" "); putchar(' ');
col_left--; }
} }
if (!printMode) {
printf("|");
} else {
printf("\n");
}
}
void get_history_path(char *history) { sprintf(history, "%s/%s", ".", HISTORY_FILE); } void get_history_path(char *history) { sprintf(history, "%s/%s", ".", HISTORY_FILE); }

2
src/query/inc/qExecutor.h
View File

@ -138,7 +138,7 @@ typedef struct SQuery {
SColumnInfo* colList; SColumnInfo* colList;
SColumnInfo* tagColList; SColumnInfo* tagColList;
int32_t numOfFilterCols; int32_t numOfFilterCols;
int64_t* defaultVal; int64_t* fillVal;
uint32_t status; // query status uint32_t status; // query status
SResultRec rec; SResultRec rec;
int32_t pos; int32_t pos;

8
src/query/inc/qfill.h
View File

@ -28,7 +28,7 @@ typedef struct {
STColumn col; // column info STColumn col; // column info
int16_t functionId; // sql function id int16_t functionId; // sql function id
int16_t flag; // column flag: TAG COLUMN|NORMAL COLUMN int16_t flag; // column flag: TAG COLUMN|NORMAL COLUMN
union {int64_t i; double d;} defaultVal; union {int64_t i; double d;} fillVal;
} SFillColInfo; } SFillColInfo;
typedef struct SFillInfo { typedef struct SFillInfo {
@ -75,15 +75,13 @@ void taosFillCopyInputDataFromOneFilePage(SFillInfo* pFillInfo, tFilePage* pInpu
TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int64_t timeInterval, int8_t slidingTimeUnit, int8_t precision); TSKEY taosGetRevisedEndKey(TSKEY ekey, int32_t order, int64_t timeInterval, int8_t slidingTimeUnit, int8_t precision);
int32_t taosGetNumOfResultWithFill(SFillInfo* pFillInfo, int32_t numOfRows, int64_t ekey, int32_t maxNumOfRows); int64_t taosGetNumOfResultWithFill(SFillInfo* pFillInfo, int32_t numOfRows, int64_t ekey, int32_t maxNumOfRows);
int32_t taosNumOfRemainRows(SFillInfo *pFillInfo); int32_t taosNumOfRemainRows(SFillInfo *pFillInfo);
int32_t taosDoInterpoResult(SFillInfo* pFillInfo, tFilePage** data, int32_t numOfRows, int32_t outputRows, char** srcData);
int taosDoLinearInterpolation(int32_t type, SPoint *point1, SPoint *point2, SPoint *point); int taosDoLinearInterpolation(int32_t type, SPoint *point1, SPoint *point2, SPoint *point);
void taosGenerateDataBlock(SFillInfo* pFillInfo, tFilePage** output, int64_t* outputRows, int32_t capacity); int64_t taosGenerateDataBlock(SFillInfo* pFillInfo, tFilePage** output, int32_t capacity);
#ifdef __cplusplus #ifdef __cplusplus
} }

136
src/query/src/qExecutor.c
View File

@ -376,11 +376,16 @@ bool isSelectivityWithTagsQuery(SQuery *pQuery) {
bool isTSCompQuery(SQuery *pQuery) { return pQuery->pSelectExpr[0].base.functionId == TSDB_FUNC_TS_COMP; } bool isTSCompQuery(SQuery *pQuery) { return pQuery->pSelectExpr[0].base.functionId == TSDB_FUNC_TS_COMP; }
static bool limitResults(SQuery *pQuery) { static bool limitResults(SQueryRuntimeEnv* pRuntimeEnv) {
SQInfo* pQInfo = GET_QINFO_ADDR(pRuntimeEnv);
SQuery* pQuery = pRuntimeEnv->pQuery;
if ((pQuery->limit.limit > 0) && (pQuery->rec.total + pQuery->rec.rows > pQuery->limit.limit)) { if ((pQuery->limit.limit > 0) && (pQuery->rec.total + pQuery->rec.rows > pQuery->limit.limit)) {
pQuery->rec.rows = pQuery->limit.limit - pQuery->rec.total; pQuery->rec.rows = pQuery->limit.limit - pQuery->rec.total;
assert(pQuery->rec.rows > 0);
qTrace("QInfo:%p discard remain data due to result limitation, limit:%"PRId64", current return:%d, total:%"PRId64,
pQInfo, pQuery->limit.limit, pQuery->rec.rows, pQuery->rec.total + pQuery->rec.rows);
assert(pQuery->rec.rows >= 0);
setQueryStatus(pQuery, QUERY_COMPLETED); setQueryStatus(pQuery, QUERY_COMPLETED);
return true; return true;
} }
@ -624,15 +629,17 @@ static int32_t getForwardStepsInBlock(int32_t numOfRows, __block_search_fn_t sea
/** /**
* NOTE: the query status only set for the first scan of master scan. * NOTE: the query status only set for the first scan of master scan.
*/ */
static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey, SWindowResInfo *pWindowResInfo) { static int32_t doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey, SWindowResInfo *pWindowResInfo) {
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
if (pRuntimeEnv->scanFlag != MASTER_SCAN || (!isIntervalQuery(pQuery))) { if (pRuntimeEnv->scanFlag != MASTER_SCAN || (!isIntervalQuery(pQuery))) {
return; return pWindowResInfo->size;
} }
// no qualified results exist, abort check // no qualified results exist, abort check
int32_t numOfClosed = 0;
if (pWindowResInfo->size == 0) { if (pWindowResInfo->size == 0) {
return; return pWindowResInfo->size;
} }
// query completed // query completed
@ -646,10 +653,10 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
int32_t i = 0; int32_t i = 0;
int64_t skey = TSKEY_INITIAL_VAL; int64_t skey = TSKEY_INITIAL_VAL;
// TODO opt performance: get the closed time window here
for (i = 0; i < pWindowResInfo->size; ++i) { for (i = 0; i < pWindowResInfo->size; ++i) {
SWindowResult *pResult = &pWindowResInfo->pResult[i]; SWindowResult *pResult = &pWindowResInfo->pResult[i];
if (pResult->status.closed) { if (pResult->status.closed) {
numOfClosed += 1;
continue; continue;
} }
@ -672,16 +679,26 @@ static void doCheckQueryCompleted(SQueryRuntimeEnv *pRuntimeEnv, TSKEY lastKey,
pWindowResInfo->prevSKey = pWindowResInfo->pResult[pWindowResInfo->curIndex].window.skey; pWindowResInfo->prevSKey = pWindowResInfo->pResult[pWindowResInfo->curIndex].window.skey;
// the number of completed slots are larger than the threshold, dump to client immediately. // the number of completed slots are larger than the threshold, return current generated results to client.
int32_t n = numOfClosedTimeWindow(pWindowResInfo); if (numOfClosed > pWindowResInfo->threshold) {
if (n > pWindowResInfo->threshold) { qTrace("QInfo:%p total result window:%d closed:%d, reached the output threshold %d, return",
GET_QINFO_ADDR(pRuntimeEnv), pWindowResInfo->size, numOfClosed, pQuery->rec.threshold);
setQueryStatus(pQuery, QUERY_RESBUF_FULL); setQueryStatus(pQuery, QUERY_RESBUF_FULL);
} else {
qTrace("QInfo:%p total result window:%d already closed:%d", GET_QINFO_ADDR(pRuntimeEnv), pWindowResInfo->size,
numOfClosed);
}
} }
qTrace("QInfo:%p total window:%d, closed:%d", GET_QINFO_ADDR(pRuntimeEnv), pWindowResInfo->size, n); // output has reached the limitation, set query completed
if (pQuery->limit.limit > 0 && (pQuery->limit.limit + pQuery->limit.offset) <= numOfClosed &&
pRuntimeEnv->scanFlag == MASTER_SCAN) {
setQueryStatus(pQuery, QUERY_COMPLETED);
} }
assert(pWindowResInfo->prevSKey != TSKEY_INITIAL_VAL); assert(pWindowResInfo->prevSKey != TSKEY_INITIAL_VAL);
return numOfClosed;
} }
static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlockInfo, TSKEY *pPrimaryColumn, static int32_t getNumOfRowsInTimeWindow(SQuery *pQuery, SDataBlockInfo *pDataBlockInfo, TSKEY *pPrimaryColumn,
@ -1309,16 +1326,14 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
TSKEY lastKey = QUERY_IS_ASC_QUERY(pQuery) ? pDataBlockInfo->window.ekey : pDataBlockInfo->window.skey; TSKEY lastKey = QUERY_IS_ASC_QUERY(pQuery) ? pDataBlockInfo->window.ekey : pDataBlockInfo->window.skey;
pTableQInfo->lastKey = lastKey + GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order); pTableQInfo->lastKey = lastKey + GET_FORWARD_DIRECTION_FACTOR(pQuery->order.order);
doCheckQueryCompleted(pRuntimeEnv, lastKey, pWindowResInfo);
// interval query with limit applied // interval query with limit applied
if (isIntervalQuery(pQuery) && pQuery->limit.limit > 0 && int32_t numOfRes = 0;
(pQuery->limit.limit + pQuery->limit.offset) <= numOfClosedTimeWindow(pWindowResInfo) &&
pRuntimeEnv->scanFlag == MASTER_SCAN) {
setQueryStatus(pQuery, QUERY_COMPLETED);
}
int32_t numOfRes = getNumOfResult(pRuntimeEnv); if (isIntervalQuery(pQuery)) {
numOfRes = doCheckQueryCompleted(pRuntimeEnv, lastKey, pWindowResInfo);
} else {
numOfRes = getNumOfResult(pRuntimeEnv);
// update the number of output result // update the number of output result
if (numOfRes > 0 && pQuery->checkBuffer == 1) { if (numOfRes > 0 && pQuery->checkBuffer == 1) {
@ -1329,10 +1344,11 @@ static int32_t tableApplyFunctionsOnBlock(SQueryRuntimeEnv *pRuntimeEnv, SDataBl
setQueryStatus(pQuery, QUERY_RESBUF_FULL); setQueryStatus(pQuery, QUERY_RESBUF_FULL);
} }
if ((pQuery->limit.limit >= 0) && numOfRes >= (pQuery->limit.limit + pQuery->limit.offset)) { if ((pQuery->limit.limit >= 0) && (pQuery->limit.limit + pQuery->limit.offset) <= numOfRes) {
setQueryStatus(pQuery, QUERY_COMPLETED); setQueryStatus(pQuery, QUERY_COMPLETED);
} }
} }
}
return numOfRes; return numOfRes;
} }
@ -2026,10 +2042,10 @@ void pointInterpSupporterSetData(SQInfo *pQInfo, SPointInterpoSupporter *pPointI
tVariantCreateFromBinary(&pCtx->param[3], (char *)&count, sizeof(count), TSDB_DATA_TYPE_INT); tVariantCreateFromBinary(&pCtx->param[3], (char *)&count, sizeof(count), TSDB_DATA_TYPE_INT);
if (isNull((char *)&pQuery->defaultVal[i], pCtx->inputType)) { if (isNull((char *)&pQuery->fillVal[i], pCtx->inputType)) {
pCtx->param[1].nType = TSDB_DATA_TYPE_NULL; pCtx->param[1].nType = TSDB_DATA_TYPE_NULL;
} else { } else {
tVariantCreateFromBinary(&pCtx->param[1], (char *)&pQuery->defaultVal[i], pCtx->inputBytes, pCtx->inputType); tVariantCreateFromBinary(&pCtx->param[1], (char *)&pQuery->fillVal[i], pCtx->inputBytes, pCtx->inputType);
} }
pInterpDetail->ts = pQuery->window.skey; pInterpDetail->ts = pQuery->window.skey;
@ -2471,8 +2487,8 @@ static int64_t doScanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
SArray *pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis); SArray *pDataBlock = loadDataBlockOnDemand(pRuntimeEnv, pQueryHandle, &blockInfo, &pStatis);
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey, pDataBlock); int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, pStatis, binarySearchForKey, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, numOfRes:%d", GET_QINFO_ADDR(pRuntimeEnv), qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", numOfRows:%d, numOfRes:%d, lastKey:%"PRId64, GET_QINFO_ADDR(pRuntimeEnv),
blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows, numOfRes); blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows, numOfRes, pQuery->current->lastKey);
// while the output buffer is full or limit/offset is applied, query may be paused here // while the output buffer is full or limit/offset is applied, query may be paused here
if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL | QUERY_COMPLETED)) { if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL | QUERY_COMPLETED)) {
@ -3374,7 +3390,9 @@ static void clearEnvAfterReverseScan(SQueryRuntimeEnv *pRuntimeEnv, SQueryStatus
// during reverse scan // during reverse scan
pTableQueryInfo->lastKey = pStatus->lastKey; pTableQueryInfo->lastKey = pStatus->lastKey;
pQuery->status = pStatus->status; pQuery->status = pStatus->status;
pTableQueryInfo->win = pStatus->w; pTableQueryInfo->win = pStatus->w;
pQuery->window = pTableQueryInfo->win;
} }
void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) { void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) {
@ -3396,6 +3414,7 @@ void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) {
if (pRuntimeEnv->scanFlag == MASTER_SCAN) { if (pRuntimeEnv->scanFlag == MASTER_SCAN) {
qstatus.status = pQuery->status; qstatus.status = pQuery->status;
qstatus.curWindow.ekey = pTableQueryInfo->lastKey - step; qstatus.curWindow.ekey = pTableQueryInfo->lastKey - step;
qstatus.lastKey = pTableQueryInfo->lastKey;
} }
if (!needScanDataBlocksAgain(pRuntimeEnv)) { if (!needScanDataBlocksAgain(pRuntimeEnv)) {
@ -3424,6 +3443,9 @@ void scanAllDataBlocks(SQueryRuntimeEnv *pRuntimeEnv, TSKEY start) {
setQueryStatus(pQuery, QUERY_NOT_COMPLETED); setQueryStatus(pQuery, QUERY_NOT_COMPLETED);
pRuntimeEnv->scanFlag = REPEAT_SCAN; pRuntimeEnv->scanFlag = REPEAT_SCAN;
qTrace("QInfo:%p start to repeat scan data blocks due to query func required, qrange:%"PRId64"-%"PRId64, pQInfo,
cond.twindow.skey, cond.twindow.ekey);
// check if query is killed or not // check if query is killed or not
if (isQueryKilled(pQInfo)) { if (isQueryKilled(pQInfo)) {
return; return;
@ -3707,7 +3729,7 @@ static int32_t doCopyToSData(SQInfo *pQInfo, SWindowResult *result, int32_t orde
int32_t startIdx = 0; int32_t startIdx = 0;
int32_t step = -1; int32_t step = -1;
qTrace("QInfo:%p start to copy data from windowResInfo to query buf", GET_QINFO_ADDR(pQuery)); qTrace("QInfo:%p start to copy data from windowResInfo to query buf", pQInfo);
int32_t totalSubset = getNumOfSubset(pQInfo); int32_t totalSubset = getNumOfSubset(pQInfo);
if (orderType == TSDB_ORDER_ASC) { if (orderType == TSDB_ORDER_ASC) {
@ -3836,7 +3858,7 @@ bool queryHasRemainResults(SQueryRuntimeEnv* pRuntimeEnv) {
} }
/* /*
* There are no results returned to client now. * While the code reaches here, there are no results returned to client now.
* If query is not completed yet, the gaps between two results blocks need to be handled after next data block * If query is not completed yet, the gaps between two results blocks need to be handled after next data block
* is retrieved from TSDB. * is retrieved from TSDB.
* *
@ -3890,18 +3912,24 @@ static void doCopyQueryResultToMsg(SQInfo *pQInfo, int32_t numOfRows, char *data
} }
int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int32_t numOfRows, int32_t *numOfInterpo) { int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int32_t numOfRows, int32_t *numOfInterpo) {
SQInfo* pQInfo = GET_QINFO_ADDR(pRuntimeEnv);
SQuery *pQuery = pRuntimeEnv->pQuery; SQuery *pQuery = pRuntimeEnv->pQuery;
SFillInfo* pFillInfo = pRuntimeEnv->pFillInfo;
while (1) { while (1) {
taosGenerateDataBlock(pRuntimeEnv->pFillInfo, (tFilePage**) pQuery->sdata, &pQuery->rec.rows, pQuery->rec.capacity); int32_t ret = taosGenerateDataBlock(pFillInfo, (tFilePage**) pQuery->sdata, pQuery->rec.capacity);
int32_t ret = pQuery->rec.rows;
// todo apply limit output function // todo apply limit output function
/* reached the start position of according to offset value, return immediately */ /* reached the start position of according to offset value, return immediately */
if (pQuery->limit.offset == 0) { if (pQuery->limit.offset == 0) {
qTrace("QInfo:%p initial numOfRows:%d, generate filled result:%d rows", pQInfo, pFillInfo->numOfRows, ret);
return ret; return ret;
} }
if (pQuery->limit.offset < ret) { if (pQuery->limit.offset < ret) {
qTrace("QInfo:%p initial numOfRows:%d, generate filled result:%d rows, offset:%d. Discard due to offset, remain:%d, new offset:%d",
pQInfo, pFillInfo->numOfRows, ret, pQuery->limit.offset, ret - pQuery->limit.offset, 0);
ret -= pQuery->limit.offset; ret -= pQuery->limit.offset;
// todo !!!!there exactly number of interpo is not valid. // todo !!!!there exactly number of interpo is not valid.
// todo refactor move to the beginning of buffer // todo refactor move to the beginning of buffer
@ -3909,10 +3937,16 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
memmove(pDst[i]->data, pDst[i]->data + pQuery->pSelectExpr[i].bytes * pQuery->limit.offset, memmove(pDst[i]->data, pDst[i]->data + pQuery->pSelectExpr[i].bytes * pQuery->limit.offset,
ret * pQuery->pSelectExpr[i].bytes); ret * pQuery->pSelectExpr[i].bytes);
} }
pQuery->limit.offset = 0; pQuery->limit.offset = 0;
return ret; return ret;
} else { } else {
qTrace("QInfo:%p initial numOfRows:%d, generate filled result:%d rows, offset:%d. Discard due to offset, "
"remain:%d, new offset:%d", pQInfo, pFillInfo->numOfRows, ret, pQuery->limit.offset, 0,
pQuery->limit.offset - ret);
pQuery->limit.offset -= ret; pQuery->limit.offset -= ret;
pQuery->rec.rows = 0;
ret = 0; ret = 0;
} }
@ -3920,8 +3954,6 @@ int32_t doFillGapsInResults(SQueryRuntimeEnv* pRuntimeEnv, tFilePage **pDst, int
return ret; return ret;
} }
} }
return 0;
} }
void vnodePrintQueryStatistics(SQInfo *pQInfo) { void vnodePrintQueryStatistics(SQInfo *pQInfo) {
@ -4002,8 +4034,8 @@ static void updateOffsetVal(SQueryRuntimeEnv *pRuntimeEnv, SDataBlockInfo *pBloc
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock); int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, pBlockInfo, NULL, binarySearchForKey, pDataBlock);
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, numOfRes:%d", GET_QINFO_ADDR(pRuntimeEnv), qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", numOfRows:%d, numOfRes:%d, lastKey:%"PRId64, GET_QINFO_ADDR(pRuntimeEnv),
pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes); pBlockInfo->window.skey, pBlockInfo->window.ekey, pBlockInfo->rows, numOfRes, pQuery->current->lastKey);
} }
void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) { void skipBlocks(SQueryRuntimeEnv *pRuntimeEnv) {
@ -4120,8 +4152,9 @@ static bool skipTimeInterval(SQueryRuntimeEnv *pRuntimeEnv, TSKEY* start) {
int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, NULL, binarySearchForKey, pDataBlock); int32_t numOfRes = tableApplyFunctionsOnBlock(pRuntimeEnv, &blockInfo, NULL, binarySearchForKey, pDataBlock);
pRuntimeEnv->windowResInfo.curIndex = index; // restore the window index pRuntimeEnv->windowResInfo.curIndex = index; // restore the window index
qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", rows:%d, numOfRes:%d", qTrace("QInfo:%p check data block, brange:%" PRId64 "-%" PRId64 ", numOfRows:%d, numOfRes:%d, lastKey:%"PRId64,
GET_QINFO_ADDR(pRuntimeEnv), blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows, numOfRes); GET_QINFO_ADDR(pRuntimeEnv), blockInfo.window.skey, blockInfo.window.ekey, blockInfo.rows, numOfRes, pQuery->current->lastKey);
return true; return true;
} else { // do nothing } else { // do nothing
*start = tw.skey; *start = tw.skey;
@ -4215,7 +4248,7 @@ static SFillColInfo* taosCreateFillColInfo(SQuery* pQuery) {
pFillCol[i].col.offset = offset; pFillCol[i].col.offset = offset;
pFillCol[i].flag = TSDB_COL_NORMAL; // always be ta normal column for table query pFillCol[i].flag = TSDB_COL_NORMAL; // always be ta normal column for table query
pFillCol[i].functionId = pExprInfo->base.functionId; pFillCol[i].functionId = pExprInfo->base.functionId;
pFillCol[i].defaultVal.i = pQuery->defaultVal[i]; pFillCol[i].fillVal.i = pQuery->fillVal[i];
offset += pExprInfo->bytes; offset += pExprInfo->bytes;
} }
@ -4591,7 +4624,7 @@ static void sequentialTableProcess(SQInfo *pQInfo) {
skipResults(pRuntimeEnv); skipResults(pRuntimeEnv);
// the limitation of output result is reached, set the query completed // the limitation of output result is reached, set the query completed
if (limitResults(pQuery)) { if (limitResults(pRuntimeEnv)) {
pQInfo->tableIndex = pQInfo->groupInfo.numOfTables; pQInfo->tableIndex = pQInfo->groupInfo.numOfTables;
break; break;
} }
@ -4846,7 +4879,7 @@ static void tableFixedOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
pQuery->rec.rows = getNumOfResult(pRuntimeEnv); pQuery->rec.rows = getNumOfResult(pRuntimeEnv);
skipResults(pRuntimeEnv); skipResults(pRuntimeEnv);
limitResults(pQuery); limitResults(pRuntimeEnv);
} }
static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) { static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
@ -4894,7 +4927,7 @@ static void tableMultiOutputProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo)
resetCtxOutputBuf(pRuntimeEnv); resetCtxOutputBuf(pRuntimeEnv);
} }
limitResults(pQuery); limitResults(pRuntimeEnv);
if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)) { if (Q_STATUS_EQUAL(pQuery->status, QUERY_RESBUF_FULL)) {
qTrace("QInfo:%p query paused due to output limitation, next qrange:%" PRId64 "-%" PRId64, pQInfo, qTrace("QInfo:%p query paused due to output limitation, next qrange:%" PRId64 "-%" PRId64, pQInfo,
pQuery->current->lastKey, pQuery->window.ekey); pQuery->current->lastKey, pQuery->window.ekey);
@ -4972,7 +5005,7 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
// the offset is handled at prepare stage if no interpolation involved // the offset is handled at prepare stage if no interpolation involved
if (pQuery->fillType == TSDB_FILL_NONE || pQuery->rec.rows == 0) { if (pQuery->fillType == TSDB_FILL_NONE || pQuery->rec.rows == 0) {
limitResults(pQuery); limitResults(pRuntimeEnv);
break; break;
} else { } else {
TSKEY ekey = taosGetRevisedEndKey(pQuery->window.ekey, pQuery->order.order, pQuery->slidingTime, TSKEY ekey = taosGetRevisedEndKey(pQuery->window.ekey, pQuery->order.order, pQuery->slidingTime,
@ -4980,11 +5013,10 @@ static void tableIntervalProcess(SQInfo *pQInfo, STableQueryInfo* pTableInfo) {
taosFillSetStartInfo(pRuntimeEnv->pFillInfo, pQuery->rec.rows, ekey); taosFillSetStartInfo(pRuntimeEnv->pFillInfo, pQuery->rec.rows, ekey);
taosFillCopyInputDataFromFilePage(pRuntimeEnv->pFillInfo, (tFilePage**) pQuery->sdata); taosFillCopyInputDataFromFilePage(pRuntimeEnv->pFillInfo, (tFilePage**) pQuery->sdata);
numOfInterpo = 0; numOfInterpo = 0;
pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, pQuery->rec.rows, &numOfInterpo);
qTrace("QInfo: %p fill results completed, final:%d", pQInfo, pQuery->rec.rows); pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, pQuery->rec.rows, &numOfInterpo);
if (pQuery->rec.rows > 0 || Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) { if (pQuery->rec.rows > 0 || Q_STATUS_EQUAL(pQuery->status, QUERY_COMPLETED)) {
limitResults(pQuery); limitResults(pRuntimeEnv);
break; break;
} }
@ -5017,9 +5049,8 @@ static void tableQueryImpl(SQInfo *pQInfo) {
int32_t remain = taosNumOfRemainRows(pRuntimeEnv->pFillInfo); int32_t remain = taosNumOfRemainRows(pRuntimeEnv->pFillInfo);
pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, remain, &numOfInterpo); pQuery->rec.rows = doFillGapsInResults(pRuntimeEnv, (tFilePage **)pQuery->sdata, remain, &numOfInterpo);
qTrace("QInfo: %p fill results completed, final:%d", pQInfo, pQuery->rec.rows);
if (pQuery->rec.rows > 0) { if (pQuery->rec.rows > 0) {
limitResults(pQuery); limitResults(pRuntimeEnv);
} }
qTrace("QInfo:%p current:%d returned, total:%d", pQInfo, pQuery->rec.rows, pQuery->rec.total); qTrace("QInfo:%p current:%d returned, total:%d", pQInfo, pQuery->rec.rows, pQuery->rec.total);
@ -5351,7 +5382,7 @@ static int32_t convertQueryMsg(SQueryTableMsg *pQueryMsg, SArray **pTableIdList,
pQueryMsg->fillType = htons(pQueryMsg->fillType); pQueryMsg->fillType = htons(pQueryMsg->fillType);
if (pQueryMsg->fillType != TSDB_FILL_NONE) { if (pQueryMsg->fillType != TSDB_FILL_NONE) {
pQueryMsg->defaultVal = (uint64_t)(pMsg); pQueryMsg->fillVal = (uint64_t)(pMsg);
int64_t *v = (int64_t *)pMsg; int64_t *v = (int64_t *)pMsg;
for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) { for (int32_t i = 0; i < pQueryMsg->numOfOutput; ++i) {
@ -5722,13 +5753,13 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
} }
if (pQuery->fillType != TSDB_FILL_NONE) { if (pQuery->fillType != TSDB_FILL_NONE) {
pQuery->defaultVal = malloc(sizeof(int64_t) * pQuery->numOfOutput); pQuery->fillVal = malloc(sizeof(int64_t) * pQuery->numOfOutput);
if (pQuery->defaultVal == NULL) { if (pQuery->fillVal == NULL) {
goto _cleanup; goto _cleanup;
} }
// the first column is the timestamp // the first column is the timestamp
memcpy(pQuery->defaultVal, (char *)pQueryMsg->defaultVal, pQuery->numOfOutput * sizeof(int64_t)); memcpy(pQuery->fillVal, (char *)pQueryMsg->fillVal, pQuery->numOfOutput * sizeof(int64_t));
} }
// to make sure third party won't overwrite this structure // to make sure third party won't overwrite this structure
@ -5785,7 +5816,7 @@ static SQInfo *createQInfoImpl(SQueryTableMsg *pQueryMsg, SArray* pTableIdList,
return pQInfo; return pQInfo;
_cleanup: _cleanup:
tfree(pQuery->defaultVal); tfree(pQuery->fillVal);
if (pQuery->sdata != NULL) { if (pQuery->sdata != NULL) {
for (int16_t col = 0; col < pQuery->numOfOutput; ++col) { for (int16_t col = 0; col < pQuery->numOfOutput; ++col) {
@ -5893,8 +5924,8 @@ static void freeQInfo(SQInfo *pQInfo) {
tfree(pQuery->pSelectExpr); tfree(pQuery->pSelectExpr);
} }
if (pQuery->defaultVal != NULL) { if (pQuery->fillVal != NULL) {
tfree(pQuery->defaultVal); tfree(pQuery->fillVal);
} }
// todo refactor, extract method to destroytableDataInfo // todo refactor, extract method to destroytableDataInfo
@ -5997,7 +6028,12 @@ static int32_t doDumpQueryResult(SQInfo *pQInfo, char *data) {
} }
pQuery->rec.total += pQuery->rec.rows; pQuery->rec.total += pQuery->rec.rows;
qTrace("QInfo:%p current:%d, total:%d", pQInfo, pQuery->rec.rows, pQuery->rec.total); qTrace("QInfo:%p current numOfRes rows:%d, total:%d", pQInfo, pQuery->rec.rows, pQuery->rec.total);
if (pQuery->limit.limit > 0 && pQuery->limit.limit == pQuery->rec.total) {
qTrace("QInfo:%p results limitation reached, limitation:%"PRId64, pQInfo, pQuery->limit.limit);
setQueryStatus(pQuery, QUERY_OVER);
}
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;

90
src/query/src/qfill.c
View File

@ -34,7 +34,7 @@ int64_t taosGetIntervalStartTimestamp(int64_t startTime, int64_t slidingTime, ch
* here we revised the start time of day according to the local time zone, * here we revised the start time of day according to the local time zone,
* but in case of DST, the start time of one day need to be dynamically decided. * but in case of DST, the start time of one day need to be dynamically decided.
* *
* TODO dynamically decide the start time of a day * TODO dynamically decide the start time of a day, move to common module
*/ */
// todo refactor to extract function that is available for Linux/Windows/Mac platform // todo refactor to extract function that is available for Linux/Windows/Mac platform
@ -117,9 +117,8 @@ void taosFillSetStartInfo(SFillInfo* pFillInfo, int32_t numOfRows, TSKEY endKey)
} }
pFillInfo->rowIdx = 0; pFillInfo->rowIdx = 0;
pFillInfo->numOfRows = numOfRows;
pFillInfo->endKey = endKey; pFillInfo->endKey = endKey;
pFillInfo->numOfRows = numOfRows;
} }
void taosFillCopyInputDataFromFilePage(SFillInfo* pFillInfo, tFilePage** pInput) { void taosFillCopyInputDataFromFilePage(SFillInfo* pFillInfo, tFilePage** pInput) {
@ -131,6 +130,8 @@ void taosFillCopyInputDataFromFilePage(SFillInfo* pFillInfo, tFilePage** pInput)
void taosFillCopyInputDataFromOneFilePage(SFillInfo* pFillInfo, tFilePage* pInput) { void taosFillCopyInputDataFromOneFilePage(SFillInfo* pFillInfo, tFilePage* pInput) {
assert(pFillInfo->numOfRows == pInput->num); assert(pFillInfo->numOfRows == pInput->num);
int32_t t = 0;
for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) { for(int32_t i = 0; i < pFillInfo->numOfCols; ++i) {
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
@ -138,7 +139,7 @@ void taosFillCopyInputDataFromOneFilePage(SFillInfo* pFillInfo, tFilePage* pInpu
memcpy(pFillInfo->pData[i], s, pInput->num * pCol->col.bytes); memcpy(pFillInfo->pData[i], s, pInput->num * pCol->col.bytes);
if (pCol->flag == TSDB_COL_TAG) { // copy the tag value if (pCol->flag == TSDB_COL_TAG) { // copy the tag value
memcpy(pFillInfo->pTags[i], pFillInfo->pData[i], pCol->col.bytes); memcpy(pFillInfo->pTags[t++], pFillInfo->pData[i], pCol->col.bytes);
} }
} }
} }
@ -170,7 +171,7 @@ static int32_t taosGetTotalNumOfFilledRes(SFillInfo* pFillInfo, const TSKEY* tsA
} }
} }
int32_t taosGetNumOfResultWithFill(SFillInfo* pFillInfo, int32_t numOfRows, int64_t ekey, int32_t maxNumOfRows) { int64_t taosGetNumOfResultWithFill(SFillInfo* pFillInfo, int32_t numOfRows, int64_t ekey, int32_t maxNumOfRows) {
int32_t numOfRes = taosGetTotalNumOfFilledRes(pFillInfo, (int64_t*) pFillInfo->pData[0], numOfRows, int32_t numOfRes = taosGetTotalNumOfFilledRes(pFillInfo, (int64_t*) pFillInfo->pData[0], numOfRows,
pFillInfo->slidingTime, ekey); pFillInfo->slidingTime, ekey);
return (numOfRes > maxNumOfRows) ? maxNumOfRows : numOfRes; return (numOfRes > maxNumOfRows) ? maxNumOfRows : numOfRes;
@ -193,7 +194,7 @@ static double linearInterpolationImpl(double v1, double v2, double k1, double k2
int taosDoLinearInterpolation(int32_t type, SPoint* point1, SPoint* point2, SPoint* point) { int taosDoLinearInterpolation(int32_t type, SPoint* point1, SPoint* point2, SPoint* point) {
switch (type) { switch (type) {
case TSDB_DATA_TYPE_INT: { case TSDB_DATA_TYPE_INT: {
*(int32_t*)point->val = linearInterpolationImpl(*(int32_t*)point1->val, *(int32_t*)point2->val, point1->key, *(int32_t*)point->val = (int32_t) linearInterpolationImpl(*(int32_t*)point1->val, *(int32_t*)point2->val, point1->key,
point2->key, point->key); point2->key, point->key);
break; break;
} }
@ -209,17 +210,17 @@ int taosDoLinearInterpolation(int32_t type, SPoint* point1, SPoint* point2, SPoi
}; };
case TSDB_DATA_TYPE_TIMESTAMP: case TSDB_DATA_TYPE_TIMESTAMP:
case TSDB_DATA_TYPE_BIGINT: { case TSDB_DATA_TYPE_BIGINT: {
*(int64_t*)point->val = linearInterpolationImpl(*(int64_t*)point1->val, *(int64_t*)point2->val, point1->key, *(int64_t*)point->val = (int64_t) linearInterpolationImpl(*(int64_t*)point1->val, *(int64_t*)point2->val, point1->key,
point2->key, point->key); point2->key, point->key);
break; break;
}; };
case TSDB_DATA_TYPE_SMALLINT: { case TSDB_DATA_TYPE_SMALLINT: {
*(int16_t*)point->val = linearInterpolationImpl(*(int16_t*)point1->val, *(int16_t*)point2->val, point1->key, *(int16_t*)point->val = (int16_t) linearInterpolationImpl(*(int16_t*)point1->val, *(int16_t*)point2->val, point1->key,
point2->key, point->key); point2->key, point->key);
break; break;
}; };
case TSDB_DATA_TYPE_TINYINT: { case TSDB_DATA_TYPE_TINYINT: {
*(int8_t*)point->val = *(int8_t*) point->val = (int8_t)
linearInterpolationImpl(*(int8_t*)point1->val, *(int8_t*)point2->val, point1->key, point2->key, point->key); linearInterpolationImpl(*(int8_t*)point1->val, *(int8_t*)point2->val, point1->key, point2->key, point->key);
break; break;
}; };
@ -243,8 +244,8 @@ static void setTagsValue(SFillInfo* pColInfo, tFilePage** data, char** pTags, in
static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t* num, char** srcData, static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t* num, char** srcData,
int64_t ts, char** pTags, bool outOfBound) { int64_t ts, char** pTags, bool outOfBound) {
char** prevValues = &pFillInfo->prevValues; char* prevValues = pFillInfo->prevValues;
char** nextValues = &pFillInfo->nextValues; char* nextValues = pFillInfo->nextValues;
SPoint point1, point2, point; SPoint point1, point2, point;
@ -257,16 +258,21 @@ static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t*
// set the other values // set the other values
if (pFillInfo->fillType == TSDB_FILL_PREV) { if (pFillInfo->fillType == TSDB_FILL_PREV) {
char* pInterpolationData = FILL_IS_ASC_FILL(pFillInfo) ? *prevValues : *nextValues; char* p = FILL_IS_ASC_FILL(pFillInfo) ? prevValues : nextValues;
if (pInterpolationData != NULL) {
if (p != NULL) {
for (int32_t i = 1; i < numOfValCols; ++i) { for (int32_t i = 1; i < numOfValCols; ++i) {
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num); char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num);
if (isNull(pInterpolationData + pCol->col.offset, pCol->col.type)) { if (isNull(p + pCol->col.offset, pCol->col.type)) {
setNull(val1, pCol->col.type, pCol->col.bytes); if (pCol->col.type == TSDB_DATA_TYPE_BINARY || pCol->col.type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(val1, pCol->col.type);
} else { } else {
assignVal(val1, pInterpolationData + pCol->col.offset, pCol->col.bytes, pCol->col.type); setNull(val1, pCol->col.type, pCol->col.bytes);
}
} else {
assignVal(val1, p + pCol->col.offset, pCol->col.bytes, pCol->col.type);
} }
} }
} else { // no prev value yet, set the value for NULL } else { // no prev value yet, set the value for NULL
@ -274,14 +280,18 @@ static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t*
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num); char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num);
if (pCol->col.type == TSDB_DATA_TYPE_BINARY||pCol->col.type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(val1, pCol->col.type);
} else {
setNull(val1, pCol->col.type, pCol->col.bytes); setNull(val1, pCol->col.type, pCol->col.bytes);
} }
} }
}
setTagsValue(pFillInfo, data, pTags, numOfValCols, *num); setTagsValue(pFillInfo, data, pTags, numOfValCols, *num);
} else if (pFillInfo->fillType == TSDB_FILL_LINEAR) { } else if (pFillInfo->fillType == TSDB_FILL_LINEAR) {
// TODO : linear interpolation supports NULL value // TODO : linear interpolation supports NULL value
if (*prevValues != NULL && !outOfBound) { if (prevValues != NULL && !outOfBound) {
for (int32_t i = 1; i < numOfValCols; ++i) { for (int32_t i = 1; i < numOfValCols; ++i) {
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
@ -289,12 +299,15 @@ static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t*
int16_t bytes = pCol->col.bytes; int16_t bytes = pCol->col.bytes;
char *val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num); char *val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num);
if (type == TSDB_DATA_TYPE_BINARY || type == TSDB_DATA_TYPE_NCHAR || type == TSDB_DATA_TYPE_BOOL) { if (type == TSDB_DATA_TYPE_BINARY|| type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(val1, pCol->col.type);
continue;
} else if (type == TSDB_DATA_TYPE_BOOL) {
setNull(val1, pCol->col.type, bytes); setNull(val1, pCol->col.type, bytes);
continue; continue;
} }
point1 = (SPoint){.key = *(TSKEY*)(*prevValues), .val = *prevValues + pCol->col.offset}; point1 = (SPoint){.key = *(TSKEY*)(prevValues), .val = prevValues + pCol->col.offset};
point2 = (SPoint){.key = ts, .val = srcData[i] + pFillInfo->rowIdx * bytes}; point2 = (SPoint){.key = ts, .val = srcData[i] + pFillInfo->rowIdx * bytes};
point = (SPoint){.key = pFillInfo->start, .val = val1}; point = (SPoint){.key = pFillInfo->start, .val = val1};
taosDoLinearInterpolation(type, &point1, &point2, &point); taosDoLinearInterpolation(type, &point1, &point2, &point);
@ -307,8 +320,13 @@ static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t*
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num); char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num);
if (pCol->col.type == TSDB_DATA_TYPE_BINARY || pCol->col.type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(val1, pCol->col.type);
} else {
setNull(val1, pCol->col.type, pCol->col.bytes); setNull(val1, pCol->col.type, pCol->col.bytes);
} }
}
setTagsValue(pFillInfo, data, pTags, numOfValCols, *num); setTagsValue(pFillInfo, data, pTags, numOfValCols, *num);
@ -318,7 +336,7 @@ static void doInterpoResultImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t*
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num); char* val1 = elePtrAt(data[i]->data, pCol->col.bytes, *num);
assignVal(val1, (char*)&pCol->defaultVal.i, pCol->col.bytes, pCol->col.type); assignVal(val1, (char*)&pCol->fillVal.i, pCol->col.bytes, pCol->col.type);
} }
setTagsValue(pFillInfo, data, pTags, numOfValCols, *num); setTagsValue(pFillInfo, data, pTags, numOfValCols, *num);
@ -338,11 +356,16 @@ static void initBeforeAfterDataBuf(SFillInfo* pFillInfo, char** nextValues) {
*nextValues = calloc(1, pFillInfo->rowSize); *nextValues = calloc(1, pFillInfo->rowSize);
for (int i = 1; i < pFillInfo->numOfCols; i++) { for (int i = 1; i < pFillInfo->numOfCols; i++) {
SFillColInfo* pCol = &pFillInfo->pFillCol[i]; SFillColInfo* pCol = &pFillInfo->pFillCol[i];
if (pCol->col.type == TSDB_DATA_TYPE_BINARY||pCol->col.type == TSDB_DATA_TYPE_NCHAR) {
setVardataNull(*nextValues + pCol->col.offset, pCol->col.type);
} else {
setNull(*nextValues + pCol->col.offset, pCol->col.type, pCol->col.bytes); setNull(*nextValues + pCol->col.offset, pCol->col.type, pCol->col.bytes);
} }
} }
}
int32_t taosDoInterpoResult(SFillInfo* pFillInfo, tFilePage** data, int32_t numOfRows, int32_t outputRows, char** srcData) { int32_t generateDataBlockImpl(SFillInfo* pFillInfo, tFilePage** data, int32_t numOfRows, int32_t outputRows, char** srcData) {
int32_t num = 0; int32_t num = 0;
pFillInfo->numOfCurrent = 0; pFillInfo->numOfCurrent = 0;
@ -356,8 +379,8 @@ int32_t taosDoInterpoResult(SFillInfo* pFillInfo, tFilePage** data, int32_t numO
if (numOfRows == 0) { if (numOfRows == 0) {
/* /*
* we need to rebuild whole result set * These data are generated according to fill strategy, since the current timestamp is out of time window of
* NOTE:we need to keep the last saved data, to generated the filled data * real result set. Note that we need to keep the direct previous result rows, to generated the filled data.
*/ */
while (num < outputRows) { while (num < outputRows) {
doInterpoResultImpl(pFillInfo, data, &num, srcData, pFillInfo->start, pTags, true); doInterpoResultImpl(pFillInfo, data, &num, srcData, pFillInfo->start, pTags, true);
@ -387,7 +410,7 @@ int32_t taosDoInterpoResult(SFillInfo* pFillInfo, tFilePage** data, int32_t numO
while (((pFillInfo->start < ts && FILL_IS_ASC_FILL(pFillInfo)) || while (((pFillInfo->start < ts && FILL_IS_ASC_FILL(pFillInfo)) ||
(pFillInfo->start > ts && !FILL_IS_ASC_FILL(pFillInfo))) && num < outputRows) { (pFillInfo->start > ts && !FILL_IS_ASC_FILL(pFillInfo))) && num < outputRows) {
doInterpoResultImpl(pFillInfo, data, &num, srcData, pFillInfo->start, pTags, false); doInterpoResultImpl(pFillInfo, data, &num, srcData, ts, pTags, false);
} }
/* output buffer is full, abort */ /* output buffer is full, abort */
@ -420,7 +443,7 @@ int32_t taosDoInterpoResult(SFillInfo* pFillInfo, tFilePage** data, int32_t numO
assignVal(val1, src, pCol->col.bytes, pCol->col.type); assignVal(val1, src, pCol->col.bytes, pCol->col.type);
memcpy(*prevValues + pCol->col.offset, src, pCol->col.bytes); memcpy(*prevValues + pCol->col.offset, src, pCol->col.bytes);
} else { } else {
assignVal(val1, (char*) &pCol->defaultVal.i, pCol->col.bytes, pCol->col.type); assignVal(val1, (char*) &pCol->fillVal.i, pCol->col.bytes, pCol->col.type);
} }
} }
} }
@ -450,21 +473,12 @@ int32_t taosDoInterpoResult(SFillInfo* pFillInfo, tFilePage** data, int32_t numO
} }
} }
void taosGenerateDataBlock(SFillInfo* pFillInfo, tFilePage** output, int64_t* outputRows, int32_t capacity) { int64_t taosGenerateDataBlock(SFillInfo* pFillInfo, tFilePage** output, int32_t capacity) {
int32_t remain = taosNumOfRemainRows(pFillInfo); // todo use iterator? int32_t remain = taosNumOfRemainRows(pFillInfo); // todo use iterator?
// TSKEY ekey = taosGetRevisedEndKey(pQuery->window.ekey, pQuery->order.order, pQuery->slidingTime,
// pQuery->slidingTimeUnit, pQuery->precision);
// if (QUERY_IS_ASC_QUERY(pQuery)) {
// assert(ekey >= pQuery->window.ekey);
// } else {
// assert(ekey <= pQuery->window.ekey);
// }
int32_t rows = taosGetNumOfResultWithFill(pFillInfo, remain, pFillInfo->endKey, capacity); int32_t rows = taosGetNumOfResultWithFill(pFillInfo, remain, pFillInfo->endKey, capacity);
int32_t numOfRes = taosDoInterpoResult(pFillInfo, output, remain, rows, pFillInfo->pData); int32_t numOfRes = generateDataBlockImpl(pFillInfo, output, remain, rows, pFillInfo->pData);
*outputRows = rows;
assert(numOfRes == rows); assert(numOfRes == rows);
return numOfRes;
} }

9
src/query/src/tvariant.c
View File

@ -101,11 +101,12 @@ void tVariantCreateFromBinary(tVariant *pVar, const char *pz, size_t len, uint32
break; break;
} }
case TSDB_DATA_TYPE_NCHAR: { // here we get the nchar length from raw binary bits length case TSDB_DATA_TYPE_NCHAR: { // here we get the nchar length from raw binary bits length
pVar->nLen = len / TSDB_NCHAR_SIZE; int32_t lenInwchar = len / TSDB_NCHAR_SIZE;
pVar->wpz = calloc(1, (pVar->nLen + 1) * TSDB_NCHAR_SIZE); pVar->wpz = calloc(1, (lenInwchar + 1) * TSDB_NCHAR_SIZE);
wcsncpy(pVar->wpz, (wchar_t *)pz, pVar->nLen); wcsncpy(pVar->wpz, (wchar_t *)pz, lenInwchar);
pVar->wpz[pVar->nLen] = 0; pVar->wpz[lenInwchar] = 0;
pVar->nLen = len;
break; break;
} }

11
src/tsdb/src/tsdbRead.c
View File

@ -1539,14 +1539,21 @@ int32_t tsdbRetrieveDataBlockStatisInfo(TsdbQueryHandleT* pQueryHandle, SDataSta
tsdbLoadCompData(&pHandle->rhelper, pBlockInfo->compBlock, NULL); tsdbLoadCompData(&pHandle->rhelper, pBlockInfo->compBlock, NULL);
size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle); size_t numOfCols = QH_GET_NUM_OF_COLS(pHandle);
memset(pHandle->statis, 0, sizeof(SDataStatis) * numOfCols); for(int32_t i = 0; i < numOfCols; ++i) {
SDataStatis* st = &pHandle->statis[i];
int32_t colId = st->colId;
memset(st, 0, sizeof(SDataStatis));
st->colId = colId;
}
tsdbGetDataStatis(&pHandle->rhelper, pHandle->statis, numOfCols); tsdbGetDataStatis(&pHandle->rhelper, pHandle->statis, numOfCols);
*pBlockStatis = pHandle->statis; *pBlockStatis = pHandle->statis;
//update the number of NULL data rows //update the number of NULL data rows
for(int32_t i = 0; i < numOfCols; ++i) { for(int32_t i = 0; i < numOfCols; ++i) {
if (pHandle->statis[i].numOfNull == -1) { if (pHandle->statis[i].numOfNull == -1) { // set the column data are all NULL
pHandle->statis[i].numOfNull = pBlockInfo->compBlock->numOfRows; pHandle->statis[i].numOfNull = pBlockInfo->compBlock->numOfRows;
} }
} }

1
src/util/src/ttime.c
View File

@ -13,6 +13,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>. * along with this program. If not, see <http://www.gnu.org/licenses/>.
*/ */
#define _BSD_SOURCE
#define _XOPEN_SOURCE #define _XOPEN_SOURCE
#define _DEFAULT_SOURCE #define _DEFAULT_SOURCE

4
tests/pytest/fulltest.sh
View File

@ -129,4 +129,6 @@ python3 ./test.py -f user/pass_len.py
#query #query
python3 ./test.py -f query/filter.py python3 ./test.py -f query/filter.py
python3 ./test.py $1 -f query/filterCombo.py
python3 ./test.py $1 -f query/queryNormal.py
python3 ./test.py $1 -f query/queryError.py

10
tests/pytest/insert/randomNullCommit.py
View File

@ -29,7 +29,8 @@ class TDTestCase:
tdSql.prepare() tdSql.prepare()
tdLog.info("=============== step1") tdLog.info("=============== step1")
tdSql.execute('create table tb (ts timestamp, speed int, temp float, note binary(5), flag bool)') tdSql.execute(
'create table tb (ts timestamp, speed int, temp float, note binary(5), flag bool)')
numOfRecords = 0 numOfRecords = 0
randomList = [10, 50, 100, 500, 1000, 5000] randomList = [10, 50, 100, 500, 1000, 5000]
@ -38,9 +39,11 @@ class TDTestCase:
tdLog.info("will insert %d records" % num) tdLog.info("will insert %d records" % num)
for x in range(0, num): for x in range(0, num):
tdLog.info( tdLog.info(
'insert into tb values (now + %da, NULL, NULL, NULL, TRUE)' % x) 'insert into tb values (now + %da, NULL, NULL, NULL, TRUE)' %
x)
tdSql.execute( tdSql.execute(
'insert into tb values (now + %da, NULL, NULL, NULL, TRUE)' % x) 'insert into tb values (now + %da, NULL, NULL, NULL, TRUE)' %
x)
numOfRecords = numOfRecords + num numOfRecords = numOfRecords + num
@ -54,7 +57,6 @@ class TDTestCase:
tdDnodes.start(1) tdDnodes.start(1)
tdLog.sleep(5) tdLog.sleep(5)
def stop(self): def stop(self):
tdSql.close() tdSql.close()
tdLog.success("%s successfully executed" % __file__) tdLog.success("%s successfully executed" % __file__)

67
tests/pytest/insert/sparseData.py Normal file
View File

@ -0,0 +1,67 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
tdLog.info("=============== step1")
tdSql.execute(
'create table tb (ts timestamp, speed int, temp float, note binary(4000), flag bool)')
numOfRecords = 1000000
dividend = 1000
tdLog.info("will insert %d records" % numOfRecords)
ts = 1500000000000
for i in range(0, numOfRecords):
if (i % dividend):
print(".", end="")
tdSql.execute(
'insert into tb values (%d + %da, NULL, NULL, NULL, TRUE)' %
(ts, i))
else:
print("a", end="")
tdSql.execute(
'insert into tb values (%d + %da, NULL, NULL, "a", FALSE)' %
(ts, i))
tdSql.query("select * from tb")
tdSql.checkRows(numOfRecords)
tdSql.checkData(numOfRecords - dividend, 3, 'a')
tdSql.checkData(numOfRecords - dividend - 1, 3, None)
tdLog.info("stop dnode to commit data to disk")
tdDnodes.stop(1)
tdLog.info("dnodes:%d size is %d" % (1, tdDnodes.getDataSize(1)))
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

60
tests/pytest/query/filterCombo.py Normal file
View File

@ -0,0 +1,60 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==============step1")
tdSql.execute(
"create table if not exists st (ts timestamp, tagtype int, name nchar(16)) tags(dev nchar(50))")
tdSql.execute(
'CREATE TABLE if not exists dev_001 using st tags("dev_01")')
tdSql.execute(
'CREATE TABLE if not exists dev_002 using st tags("dev_02")')
print("==============step2")
tdSql.execute(
"""INSERT INTO dev_001(ts, tagtype, name) VALUES('2020-05-13 10:00:00.000', 1, 'first'),('2020-05-13 10:00:00.001', 2, 'second'),
('2020-05-13 10:00:00.002', 3, 'third') dev_002 VALUES('2020-05-13 10:00:00.003', 1, 'first'), ('2020-05-13 10:00:00.004', 2, 'second'),
('2020-05-13 10:00:00.005', 3, 'third')""")
# query with filter condition A OR condition B
tdSql.query("select * from db.st where ts > '2020-05-13 10:00:00.002' AND tagtype < 2")
tdSql.checkRows(1)
# query with filter condition A OR condition B, error expected
tdSql.error("select * from db.st where ts > '2020-05-13 10:00:00.002' OR tagtype < 2")
# illegal condition
tdSql.error("select * from db.st where ts != '2020-05-13 10:00:00.002' OR tagtype < 2")
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

67
tests/pytest/query/queryError.py Normal file
View File

@ -0,0 +1,67 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==============step1")
tdSql.execute(
"create table if not exists st (ts timestamp, tagtype int, name nchar(16)) tags(dev nchar(50))")
tdSql.execute(
'CREATE TABLE if not exists dev_001 using st tags("dev_01")')
tdSql.execute(
'CREATE TABLE if not exists dev_002 using st tags("dev_02")')
print("==============step2")
tdSql.execute(
"""INSERT INTO dev_001(ts, tagtype, name) VALUES('2020-05-13 10:00:00.000', 1, 'first'),('2020-05-13 10:00:00.001', 2, 'second'),
('2020-05-13 10:00:00.002', 3, 'third') dev_002 VALUES('2020-05-13 10:00:00.003', 1, 'first'), ('2020-05-13 10:00:00.004', 2, 'second'),
('2020-05-13 10:00:00.005', 3, 'third')""")
"""Error expected here, but no errors
# query first .. as ..
tdSql.error("select first(*) as one from st")
# query last .. as ..
tdSql.error("select last(*) as latest from st")
"""
# query last row .. as ..
tdSql.error("select last_row as latest from st")
# query distinct on normal colnum
tdSql.error("select distinct tagtype from st")
# query .. order by non-time field
tdSql.error("select * from st order by name")
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

84
tests/pytest/query/queryNormal.py Normal file
View File

@ -0,0 +1,84 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import taos
from util.log import *
from util.cases import *
from util.sql import *
class TDTestCase:
def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__)
tdSql.init(conn.cursor(), logSql)
def run(self):
tdSql.prepare()
print("==============step1")
tdSql.execute("create table stb1 (ts timestamp, c1 int, c2 float) tags(t1 int, t2 binary(10), t3 nchar(10))")
tdSql.execute("insert into tb1 using stb1 tags(1,'tb1', '表1') values ('2020-04-18 15:00:00.000', 1, 0.1), ('2020-04-18 15:00:01.000', 2, 0.1)")
tdSql.execute("insert into tb2 using stb1 tags(2,'tb2', '表2') values ('2020-04-18 15:00:02.000', 3, 2.1), ('2020-04-18 15:00:03.000', 4, 2.2)")
# join 2 tables -- bug exists
# tdSql.query("select * from tb1 a, tb2 b where a.ts = b.ts")
# tdSql.checkRows(1)
# join 3 tables -- bug exists
# tdSql.query("select stb_t.ts, stb_t.dscrption, stb_t.temperature, stb_p.id, stb_p.dscrption, stb_p.pressure,stb_v.velocity from stb_p, stb_t, stb_v where stb_p.ts=stb_t.ts and stb_p.ts=stb_v.ts and stb_p.id = stb_t.id")
# query count
tdSql.query("select count(*) from stb1")
tdSql.checkData(0, 0, 4)
# query first
tdSql.query("select first(*) from stb1")
tdSql.checkData(0, 1, 1)
# query last
tdSql.query("select last(*) from stb1")
tdSql.checkData(0, 1, 4)
# query as
tdSql.query("select t2 as number from stb1")
tdSql.checkRows(2)
# query first ... as
tdSql.query("select first(*) as begin from stb1")
tdSql.checkData(0, 1, 1)
# query last ... as
tdSql.query("select last(*) as end from stb1")
tdSql.checkData(0, 1, 4)
# query group .. by
tdSql.query("select sum(c1), t2 from stb1 group by t2")
tdSql.checkRows(2)
# query ... limit
tdSql.query("select * from stb1 limit 2")
tdSql.checkRows(2)
# query ... limit offset
tdSql.query("select * from stb1 limit 2 offset 3")
tdSql.checkRows(1)
def stop(self):
tdSql.close()
tdLog.success("%s successfully executed" % __file__)
tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())

1
tests/pytest/random-test/random-test.py
View File

@ -121,6 +121,7 @@ class Test:
tdDnodes.start(1) tdDnodes.start(1)
tdSql.prepare() tdSql.prepare()
class TDTestCase: class TDTestCase:
def init(self, conn, logSql): def init(self, conn, logSql):
tdLog.debug("start to execute %s" % __file__) tdLog.debug("start to execute %s" % __file__)

6
tests/pytest/smoketest.sh
View File

@ -56,4 +56,10 @@ python3 ./test.py $1 -s && sleep 1
#query #query
python3 ./test.py $1 -f query/filter.py python3 ./test.py $1 -f query/filter.py
python3 ./test.py $1 -s && sleep 1 python3 ./test.py $1 -s && sleep 1
python3 ./test.py $1 -f query/filterCombo.py
python3 ./test.py $1 -s && sleep 1
python3 ./test.py $1 -f query/queryNormal.py
python3 ./test.py $1 -s && sleep 1
python3 ./test.py $1 -f query/queryError.py
python3 ./test.py $1 -s && sleep 1

8
tests/pytest/table/boundary.py
View File

@ -98,12 +98,16 @@ class TDTestCase:
# create a super table with name exceed max length # create a super table with name exceed max length
sname = self.generateString(maxTableNameLen + 1) sname = self.generateString(maxTableNameLen + 1)
tdLog.info("create a super table with length %d" % len(sname)) tdLog.info("create a super table with length %d" % len(sname))
tdSql.error("create table %s (ts timestamp, value int) tags(id int)" % sname) tdSql.error(
"create table %s (ts timestamp, value int) tags(id int)" %
sname)
# create a super table with name of max length # create a super table with name of max length
sname = self.generateString(maxTableNameLen) sname = self.generateString(maxTableNameLen)
tdLog.info("create a super table with length %d" % len(sname)) tdLog.info("create a super table with length %d" % len(sname))
tdSql.execute("create table %s (ts timestamp, value int) tags(id int)" % sname) tdSql.execute(
"create table %s (ts timestamp, value int) tags(id int)" %
sname)
tdLog.info("check table count, should be one") tdLog.info("check table count, should be one")
tdSql.query('show stables') tdSql.query('show stables')
tdSql.checkRows(1) tdSql.checkRows(1)

17
tests/pytest/util/dnodes.py
View File

@ -99,6 +99,19 @@ class TDDnode:
def setValgrind(self, value): def setValgrind(self, value):
self.valgrind = value self.valgrind = value
def getDataSize(self):
totalSize = 0
if (self.deployed == 1):
for dirpath, dirnames, filenames in os.walk(self.dataDir):
for f in filenames:
fp = os.path.join(dirpath, f)
if not os.path.islink(fp):
totalSize = totalSize + os.path.getsize(fp)
return totalSize
def deploy(self): def deploy(self):
self.logDir = "%s/pysim/dnode%d/log" % (self.path, self.index) self.logDir = "%s/pysim/dnode%d/log" % (self.path, self.index)
self.dataDir = "%s/pysim/dnode%d/data" % (self.path, self.index) self.dataDir = "%s/pysim/dnode%d/data" % (self.path, self.index)
@ -384,6 +397,10 @@ class TDDnodes:
self.check(index) self.check(index)
self.dnodes[index - 1].stop() self.dnodes[index - 1].stop()
def getDataSize(self, index):
self.check(index)
return self.dnodes[index - 1].getDataSize()
def forcestop(self, index): def forcestop(self, index):
self.check(index) self.check(index)
self.dnodes[index - 1].forcestop() self.dnodes[index - 1].forcestop()

1
tests/script/general/parser/limit2_query.sim
View File

@ -36,6 +36,7 @@ if $data01 != 2 then
return -1 return -1
endi endi
if $data02 != tb2 then if $data02 != tb2 then
print expect tb2, actual: $data02
return -1 return -1
endi endi
if $data03 != tb2 then if $data03 != tb2 then

1
tests/script/sh/deploy.sh
View File

@ -96,6 +96,7 @@ echo "second ${HOSTNAME}:7200" >> $TAOS_CFG
echo "serverPort ${NODE}" >> $TAOS_CFG echo "serverPort ${NODE}" >> $TAOS_CFG
echo "dataDir $DATA_DIR" >> $TAOS_CFG echo "dataDir $DATA_DIR" >> $TAOS_CFG
echo "logDir $LOG_DIR" >> $TAOS_CFG echo "logDir $LOG_DIR" >> $TAOS_CFG
echo "debugFlag 135" >> $TAOS_CFG
echo "dDebugFlag 135" >> $TAOS_CFG echo "dDebugFlag 135" >> $TAOS_CFG
echo "mDebugFlag 135" >> $TAOS_CFG echo "mDebugFlag 135" >> $TAOS_CFG
echo "sdbDebugFlag 135" >> $TAOS_CFG echo "sdbDebugFlag 135" >> $TAOS_CFG

64
tests/script/sh/exec_tarbitrator.sh Executable file
View File

@ -0,0 +1,64 @@
#!/bin/sh
# if [ $# != 2 || $# != 3 ]; then
# echo "argument list need input : "
# echo " -s start/stop"
# exit 1
# fi
EXEC_OPTON=
while getopts "n:s:u:x:ct" arg
do
case $arg in
n)
NODE_NAME=$OPTARG
;;
s)
EXEC_OPTON=$OPTARG
;;
c)
CLEAR_OPTION="clear"
;;
t)
SHELL_OPTION="true"
;;
u)
USERS=$OPTARG
;;
x)
SIGNAL=$OPTARG
;;
?)
echo "unkown argument"
;;
esac
done
SCRIPT_DIR=`dirname $0`
cd $SCRIPT_DIR/../
SCRIPT_DIR=`pwd`
cd ../../
TAOS_DIR=`pwd`
BUILD_DIR=$TAOS_DIR/../debug/build
SIM_DIR=$TAOS_DIR/sim
NODE_DIR=$SIM_DIR/arbitrator
EXE_DIR=$BUILD_DIR/bin
LOG_DIR=$NODE_DIR/log
echo "------------ $EXEC_OPTON tarbitrator"
if [ "$EXEC_OPTON" = "start" ]; then
echo "------------ log path: $LOG_DIR"
nohup $EXE_DIR/tarbitrator -p 8000 -d 135 -g $LOG_DIR > /dev/null 2>&1 &
else
#relative path
PID=`ps -ef|grep tarbitrator | grep -v grep | awk '{print $2}'`
if [ -n "$PID" ]; then
sudo kill -9 $PID
sudo pkill -9 tarbitrator
fi
fi

9
tests/script/sh/stop_dnodes.sh
View File

@ -13,3 +13,12 @@ while [ -n "$PID" ]; do
fuser -k -n tcp 6030 fuser -k -n tcp 6030
PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'` PID=`ps -ef|grep -w taosd | grep -v grep | awk '{print $2}'`
done done
PID=`ps -ef|grep -w tarbitrator | grep -v grep | awk '{print $2}'`
while [ -n "$PID" ]; do
echo kill -9 $PID
pkill -9 tarbitrator
fuser -k -n tcp 6040
PID=`ps -ef|grep -w tarbitrator | grep -v grep | awk '{print $2}'`
done

8
tests/script/test.sh
View File

@ -62,6 +62,11 @@ CFG_DIR=$PRG_DIR/cfg
LOG_DIR=$PRG_DIR/log LOG_DIR=$PRG_DIR/log
DATA_DIR=$PRG_DIR/data DATA_DIR=$PRG_DIR/data
ARBITRATOR_PRG_DIR=$SIM_DIR/arbitrator
ARBITRATOR_LOG_DIR=$ARBITRATOR_PRG_DIR/log
chmod -R 777 $PRG_DIR chmod -R 777 $PRG_DIR
echo "------------------------------------------------------------------------" echo "------------------------------------------------------------------------"
echo "Start TDengine Testing Case ..." echo "Start TDengine Testing Case ..."
@ -72,9 +77,12 @@ echo "CFG_DIR : $CFG_DIR"
rm -rf $LOG_DIR rm -rf $LOG_DIR
rm -rf $CFG_DIR rm -rf $CFG_DIR
rm -rf $ARBITRATOR_LOG_DIR
mkdir -p $PRG_DIR mkdir -p $PRG_DIR
mkdir -p $LOG_DIR mkdir -p $LOG_DIR
mkdir -p $CFG_DIR mkdir -p $CFG_DIR
mkdir -p $ARBITRATOR_LOG_DIR
TAOS_CFG=$PRG_DIR/cfg/taos.cfg TAOS_CFG=$PRG_DIR/cfg/taos.cfg
touch -f $TAOS_CFG touch -f $TAOS_CFG

100
tests/script/unique/arbitrator/replica_changeWithArbitrator.sim Normal file
View File

@ -0,0 +1,100 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2
system sh/deploy.sh -n dnode3 -i 3
system sh/cfg.sh -n dnode1 -c numOfMPeers -v 2
system sh/cfg.sh -n dnode2 -c numOfMPeers -v 2
system sh/cfg.sh -n dnode3 -c numOfMPeers -v 2
system sh/cfg.sh -n dnode1 -c walLevel -v 1
system sh/cfg.sh -n dnode2 -c walLevel -v 1
system sh/cfg.sh -n dnode3 -c walLevel -v 1
system sh/cfg.sh -n dnode1 -c balanceInterval -v 10
system sh/cfg.sh -n dnode2 -c balanceInterval -v 10
system sh/cfg.sh -n dnode3 -c balanceInterval -v 10
system sh/cfg.sh -n dnode1 -c arbitrator -v $arbitrator
system sh/cfg.sh -n dnode2 -c arbitrator -v $arbitrator
system sh/cfg.sh -n dnode3 -c arbitrator -v $arbitrator
print ============== step0: start tarbitrator
system sh/exec_tarbitrator.sh -s start
print ============== step1: replica is 1, and start 1 dnode
system sh/exec_up.sh -n dnode1 -s start
sleep 3000
sql connect
$db = replica_db1
sql create database $db replica 1 maxTables 4
sql use $db
# create table , insert data
$stb = repl_stb
sql create table $stb (ts timestamp, c1 int) tags(t1 int)
$rowNum = 10
$tblNum = 12
$ts0 = 1420041600000
$ts = $ts0
$delta = 1
$i = 0
while $i < $tblNum
$tb = tb . $i
sql create table $tb using $stb tags( $i )
$x = 0
while $x < $rowNum
$xs = $x * $delta
$ts = $ts0 + $xs
sql insert into $tb values ( $ts , $x )
$x = $x + 1
endw
$i = $i + 1
endw
print ============== step2: add 1 new dnode, expect balanced
system sh/exec_up.sh -n dnode2 -s start
sql create dnode $hostname2
# expect after balanced, 2 vondes in dnode1, 1 vonde in dnode2
$x = 0
show2:
$x = $x + 1
sleep 2000
if $x == 10 then
return -1
endi
sql show dnodes
print dnode1 openVnodes $data2_1
print dnode2 openVnodes $data2_2
if $data2_1 != 2 then
goto show2
endi
if $data2_2 != 1 then
goto show2
endi
print ============== step4: stop dnode1, and wait dnode2 master
system sh/exec_up.sh -n dnode1 -s stop
$x = 0
loop_wait:
$x = $x + 1
sleep 2000
if $x == 10 then
print ERROR: after dnode1 stop, dnode2 didn't become a master!
return -1
endi
sql show mnodes
$dnodeRole = $data2_1
print dnodeRole ==> $dnodeRole
if $dnodeRole != master then
goto loop_wait
endi

5
tests/tsim/src/simExe.c
View File

@ -36,12 +36,17 @@ void simLogSql(char *sql) {
fflush(fp); fflush(fp);
} }
char *simParseArbitratorName(char *varName);
char *simParseHostName(char *varName); char *simParseHostName(char *varName);
char *simGetVariable(SScript *script, char *varName, int varLen) { char *simGetVariable(SScript *script, char *varName, int varLen) {
if (strncmp(varName, "hostname", 8) == 0) { if (strncmp(varName, "hostname", 8) == 0) {
return simParseHostName(varName); return simParseHostName(varName);
} }
if (strncmp(varName, "arbitrator", 10) == 0) {
return simParseArbitratorName(varName);
}
if (strncmp(varName, "error", varLen) == 0) return script->error; if (strncmp(varName, "error", varLen) == 0) return script->error;
if (strncmp(varName, "rows", varLen) == 0) return script->rows; if (strncmp(varName, "rows", varLen) == 0) return script->rows;

6
tests/tsim/src/simSystem.c
View File

@ -29,6 +29,12 @@ int simDebugFlag = 135;
void simCloseTaosdConnect(SScript *script); void simCloseTaosdConnect(SScript *script);
char simHostName[128]; char simHostName[128];
char *simParseArbitratorName(char *varName) {
static char hostName[140];
sprintf(hostName, "%s:%d", simHostName, 8000);
return hostName;
}
char *simParseHostName(char *varName) { char *simParseHostName(char *varName) {
static char hostName[140]; static char hostName[140];