diff --git a/src/client/inc/tsclient.h b/src/client/inc/tsclient.h index 29abff7685..471db9c99f 100644 --- a/src/client/inc/tsclient.h +++ b/src/client/inc/tsclient.h @@ -171,11 +171,7 @@ typedef struct STableDataBlocks { * to avoid it to be removed from cache */ STableMeta *pTableMeta; - - union { - char *filename; - char *pData; - }; + char *pData; // for parameter ('?') binding uint32_t numOfAllocedParams; @@ -398,7 +394,7 @@ void waitForQueryRsp(void *param, TAOS_RES *tres, int code) ; int doAsyncParseSql(SSqlObj* pSql); void doAsyncQuery(STscObj *pObj, SSqlObj *pSql, void (*fp)(), void *param, const char *sqlstr, size_t sqlLen); -void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql); +void tscProcessMultiVnodesImportFromFile(SSqlObj *pSql); void tscKillSTableQuery(SSqlObj *pSql); void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen); bool tscIsUpdateQuery(SSqlObj* pSql); diff --git a/src/client/src/tscParseInsert.c b/src/client/src/tscParseInsert.c index f2af4f1d32..df4ccca9bc 100644 --- a/src/client/src/tscParseInsert.c +++ b/src/client/src/tscParseInsert.c @@ -1153,29 +1153,19 @@ int tsParseInsertSql(SSqlObj *pSql) { goto _error; } - char fname[PATH_MAX] = {0}; - strncpy(fname, sToken.z, sToken.n); - strdequote(fname); + strncpy(pCmd->payload, sToken.z, sToken.n); + strdequote(pCmd->payload); + // todo refactor extract method wordexp_t full_path; - if (wordexp(fname, &full_path, 0) != 0) { + if (wordexp(pCmd->payload, &full_path, 0) != 0) { code = tscInvalidSQLErrMsg(pCmd->payload, "invalid filename", sToken.z); goto _error; } - strcpy(fname, full_path.we_wordv[0]); + + tstrncpy(pCmd->payload, full_path.we_wordv[0], pCmd->allocSize); wordfree(&full_path); - STableDataBlocks *pDataBlock = NULL; - STableMeta* pTableMeta = pTableMetaInfo->pTableMeta; - - int32_t ret = tscCreateDataBlock(PATH_MAX, tinfo.rowSize, sizeof(SSubmitBlk), pTableMetaInfo->name, - pTableMeta, &pDataBlock); - if (ret != TSDB_CODE_SUCCESS) { - goto _error; - } - - taosArrayPush(pCmd->pDataBlocks, &pDataBlock); - strcpy(pDataBlock->filename, fname); } else if (sToken.type == TK_LP) { /* insert into tablename(col1, col2,..., coln) values(v1, v2,... vn); */ STableMeta *pTableMeta = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0)->pTableMeta; @@ -1377,172 +1367,141 @@ static int doPackSendDataBlock(SSqlObj *pSql, int32_t numOfRows, STableDataBlock return code; } - if ((code = tscProcessSql(pSql)) != TSDB_CODE_SUCCESS) { - return code; - } - - return TSDB_CODE_SUCCESS; + return tscProcessSql(pSql); } -static int tscInsertDataFromFile(SSqlObj *pSql, FILE *fp, char *tmpTokenBuf) { - size_t readLen = 0; - char * line = NULL; - size_t n = 0; - int len = 0; - int32_t maxRows = 0; - SSqlCmd * pCmd = &pSql->cmd; - int numOfRows = 0; - int32_t code = 0; - int nrows = 0; - +typedef struct SImportFileSupport { + SSqlObj *pSql; + FILE *fp; +} SImportFileSupport; + +static void parseFileSendDataBlock(void *param, TAOS_RES *tres, int code) { + assert(param != NULL && tres != NULL); + + SSqlObj *pSql = tres; + SSqlCmd *pCmd = &pSql->cmd; + + SImportFileSupport *pSupporter = (SImportFileSupport *) param; + + SSqlObj *pParentSql = pSupporter->pSql; + FILE *fp = pSupporter->fp; + + if (taos_errno(pSql) != TSDB_CODE_SUCCESS) { // handle error + assert(taos_errno(pSql) == code); + + taos_free_result(pSql); + tfree(pSupporter); + fclose(fp); + + pParentSql->res.code = code; + return; + } + + // accumulate the total submit records + pParentSql->res.numOfRows += pSql->res.numOfRows; + STableMetaInfo *pTableMetaInfo = tscGetTableMetaInfoFromCmd(pCmd, pCmd->clauseIndex, 0); STableMeta * pTableMeta = pTableMetaInfo->pTableMeta; - STableComInfo tinfo = tscGetTableInfo(pTableMeta); - - assert(pCmd->numOfClause == 1); - - int32_t rowSize = tinfo.rowSize; + SSchema * pSchema = tscGetTableSchema(pTableMeta); + STableComInfo tinfo = tscGetTableInfo(pTableMeta); + + SParsedDataColInfo spd = {.numOfCols = tinfo.numOfColumns}; + tscSetAssignedColumnInfo(&spd, pSchema, tinfo.numOfColumns); + + size_t n = 0; + ssize_t readLen = 0; + char * line = NULL; + int32_t count = 0; + int32_t maxRows = 0; + + tscDestroyBlockArrayList(pSql->cmd.pDataBlocks); + pCmd->pDataBlocks = taosArrayInit(1, POINTER_BYTES); - pCmd->pDataBlocks = taosArrayInit(4, POINTER_BYTES); STableDataBlocks *pTableDataBlock = NULL; - - int32_t ret = tscCreateDataBlock(TSDB_PAYLOAD_SIZE, rowSize, sizeof(SSubmitBlk), pTableMetaInfo->name, pTableMeta, &pTableDataBlock); + int32_t ret = tscCreateDataBlock(TSDB_PAYLOAD_SIZE, tinfo.rowSize, sizeof(SSubmitBlk), pTableMetaInfo->name, pTableMeta, &pTableDataBlock); if (ret != TSDB_CODE_SUCCESS) { - return ret; +// return ret; } taosArrayPush(pCmd->pDataBlocks, &pTableDataBlock); + tscAllocateMemIfNeed(pTableDataBlock, tinfo.rowSize, &maxRows); - code = tscAllocateMemIfNeed(pTableDataBlock, rowSize, &maxRows); - if (TSDB_CODE_SUCCESS != code) return -1; - - int count = 0; - SParsedDataColInfo spd = {.numOfCols = tinfo.numOfColumns}; - SSchema * pSchema = tscGetTableSchema(pTableMeta); - - tscSetAssignedColumnInfo(&spd, pSchema, tinfo.numOfColumns); + char *tokenBuf = calloc(1, 4096); while ((readLen = getline(&line, &n, fp)) != -1) { - // line[--readLen] = '\0'; - if (('\r' == line[readLen - 1]) || ('\n' == line[readLen - 1])) line[--readLen] = 0; - if (readLen == 0) continue; // fang, <= to == + if (('\r' == line[readLen - 1]) || ('\n' == line[readLen - 1])) { + line[--readLen] = 0; + } + + if (readLen == 0) { + continue; + } char *lineptr = line; strtolower(line, line); - - len = tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd->payload, tinfo.precision, &code, tmpTokenBuf); + + int32_t len = + tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd->payload, tinfo.precision, &code, tokenBuf); if (len <= 0 || pTableDataBlock->numOfParams > 0) { pSql->res.code = code; - return (-code); + break; } pTableDataBlock->size += len; - count++; - nrows++; - if (count >= maxRows) { - if ((code = doPackSendDataBlock(pSql, count, pTableDataBlock)) != TSDB_CODE_SUCCESS) { - return -code; - } - - pTableDataBlock = taosArrayGetP(pCmd->pDataBlocks, 0); - pTableDataBlock->size = sizeof(SSubmitBlk); - pTableDataBlock->rowSize = tinfo.rowSize; - - numOfRows += pSql->res.numOfRows; - pSql->res.numOfRows = 0; - count = 0; + if (++count >= maxRows) { + break; } } + tfree(tokenBuf); + free(line); + if (count > 0) { if ((code = doPackSendDataBlock(pSql, count, pTableDataBlock)) != TSDB_CODE_SUCCESS) { - return -code; + pParentSql->res.code = code; } - numOfRows += pSql->res.numOfRows; - pSql->res.numOfRows = 0; + } else { + taos_free_result(pSql); + tfree(pSupporter); + fclose(fp); + + pParentSql->fp = pParentSql->fetchFp; + + // all data has been sent to vnode, call user function + int32_t v = (pParentSql->res.code != TSDB_CODE_SUCCESS) ? pParentSql->res.code : pParentSql->res.numOfRows; + (*pParentSql->fp)(pParentSql->param, pParentSql, v); } - - if (line) tfree(line); - - return numOfRows; } -void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql) { +void tscProcessMultiVnodesImportFromFile(SSqlObj *pSql) { SSqlCmd *pCmd = &pSql->cmd; if (pCmd->command != TSDB_SQL_INSERT) { return; } - SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, 0); - STableMetaInfo *pTableMetaInfo = tscGetMetaInfo(pQueryInfo, 0); + assert(pCmd->dataSourceType == DATA_FROM_DATA_FILE && strlen(pCmd->payload) != 0); - STableDataBlocks *pDataBlock = NULL; - int32_t affected_rows = 0; + SImportFileSupport *pSupporter = calloc(1, sizeof(SImportFileSupport)); + SSqlObj *pNew = createSubqueryObj(pSql, 0, parseFileSendDataBlock, pSupporter, TSDB_SQL_INSERT, NULL); - assert(pCmd->dataSourceType == DATA_FROM_DATA_FILE && pCmd->pDataBlocks != NULL); - SArray *pDataBlockList = pCmd->pDataBlocks; - pCmd->pDataBlocks = NULL; + pNew->cmd.pDataBlocks = taosArrayInit(4, POINTER_BYTES); + pCmd->count = 1; - char path[PATH_MAX] = {0}; + FILE *fp = fopen(pCmd->payload, "r"); + if (fp == NULL) { + pSql->res.code = TAOS_SYSTEM_ERROR(errno); + tscError("%p failed to open file %s to load data from file, code:%s", pSql, pCmd->payload, tstrerror(pSql->res.code)); - size_t size = taosArrayGetSize(pDataBlockList); - for (int32_t i = 0; i < size; ++i) { - pDataBlock = taosArrayGetP(pDataBlockList, i ); - if (pDataBlock == NULL) { - continue; - } + tfree(pSupporter) + tscQueueAsyncRes(pSql); - if (TSDB_CODE_SUCCESS != tscAllocPayload(pCmd, TSDB_PAYLOAD_SIZE)) { - tscError("%p failed to malloc when insert file", pSql); - continue; - } - pCmd->count = 1; - - tstrncpy(path, pDataBlock->filename, sizeof(path)); - - FILE *fp = fopen(path, "r"); - if (fp == NULL) { - tscError("%p failed to open file %s to load data from file, reason:%s", pSql, path, strerror(errno)); - continue; - } - - tstrncpy(pTableMetaInfo->name, pDataBlock->tableId, sizeof(pTableMetaInfo->name)); - memset(pDataBlock->pData, 0, pDataBlock->nAllocSize); - - int32_t ret = tscGetTableMeta(pSql, pTableMetaInfo); - if (ret != TSDB_CODE_SUCCESS) { - tscError("%p get meter meta failed, abort", pSql); - continue; - } - - char *tmpTokenBuf = calloc(1, 4096); // used for deleting Escape character: \\, \', \" - if (NULL == tmpTokenBuf) { - tscError("%p calloc failed", pSql); - continue; - } - - int nrows = tscInsertDataFromFile(pSql, fp, tmpTokenBuf); - free(tmpTokenBuf); - - pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); - - if (nrows < 0) { - fclose(fp); - tscTrace("%p no records(%d) in file %s", pSql, nrows, path); - continue; - } - - fclose(fp); - affected_rows += nrows; - - tscTrace("%p Insert data %d records from file %s", pSql, nrows, path); + return; } - pSql->res.numOfRows = affected_rows; + pSupporter->pSql = pSql; + pSupporter->fp = fp; - // all data have been submit to vnode, release data blocks - pCmd->pDataBlocks = tscDestroyBlockArrayList(pCmd->pDataBlocks); - tscDestroyBlockArrayList(pDataBlockList); + parseFileSendDataBlock(pSupporter, pNew, 0); } diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index 631a9b5eb3..5448e5433a 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -195,7 +195,11 @@ int tscSendMsgToServer(SSqlObj *pSql) { .code = 0 }; - pSql->pRpcCtx = rpcSendRequest(pObj->pDnodeConn, &pSql->ipList, &rpcMsg); + // NOTE: the rpc context should be acquired before sending data to server. + // Otherwise, the pSql object may have been released already during the response function, which is + // processMsgFromServer function. In the meanwhile, the assignment of the rpc context to sql object will absolutely + // cause crash. + /*pSql->pRpcCtx = */rpcSendRequest(pObj->pDnodeConn, &pSql->ipList, &rpcMsg); return TSDB_CODE_SUCCESS; } diff --git a/src/client/src/tscUtil.c b/src/client/src/tscUtil.c index 4d90caddcb..b4fe670df8 100644 --- a/src/client/src/tscUtil.c +++ b/src/client/src/tscUtil.c @@ -1880,7 +1880,7 @@ void tscDoQuery(SSqlObj* pSql) { } if (pCmd->dataSourceType == DATA_FROM_DATA_FILE) { - tscProcessMultiVnodesInsertFromFile(pSql); + tscProcessMultiVnodesImportFromFile(pSql); } else { SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, pCmd->clauseIndex); uint16_t type = pQueryInfo->type; diff --git a/src/inc/taosdef.h b/src/inc/taosdef.h index 8cc5e82590..eefd9f0c00 100644 --- a/src/inc/taosdef.h +++ b/src/inc/taosdef.h @@ -245,7 +245,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size); #define TSDB_DEFAULT_PKT_SIZE 65480 //same as RPC_MAX_UDP_SIZE #define TSDB_PAYLOAD_SIZE TSDB_DEFAULT_PKT_SIZE -#define TSDB_DEFAULT_PAYLOAD_SIZE 2048 // default payload size +#define TSDB_DEFAULT_PAYLOAD_SIZE 5120 // default payload size, greater than PATH_MAX value #define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth #define TSDB_CQ_SQL_SIZE 1024 #define TSDB_MAX_VNODES 256 diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index e4b84fc8e8..e673277a0b 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -470,11 +470,12 @@ int main(int argc, char *argv[]) { char command[BUFFER_SIZE] = "\0"; sprintf(command, "drop database %s;", db_name); - taos_query(taos, command); - + TAOS_RES* res = taos_query(taos, command); + taos_free_result(res); sprintf(command, "create database %s;", db_name); - taos_query(taos, command); + res = taos_query(taos, command); + taos_free_result(res); char cols[STRING_LEN] = "\0"; int colIndex = 0;