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

refactor stmt2 stb interlace mode

This commit is contained in:
pengrongkun94@qq.com 2024-12-30 17:43:36 +08:00
parent eafbe63683
commit 8d3c0a8d91
2 changed files with 124 additions and 153 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

208
source/client/src/clientStmt2.c
View File

@ -58,13 +58,13 @@ static bool stmtDequeue(STscStmt2* pStmt, SStmtQNode** param) {
return true;
}
static void stmtEnqueue(STscStmt2* pStmt, SStmtQNode* param) {
pStmt->queue.tail->next = param;
pStmt->queue.tail = param;
// static void stmtEnqueue(STscStmt2* pStmt, SStmtQNode* param) {
// pStmt->queue.tail->next = param;
// pStmt->queue.tail = param;
pStmt->stat.bindDataNum++;
(void)atomic_add_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
}
// pStmt->stat.bindDataNum++;
// (void)atomic_add_fetch_64((int64_t*)&pStmt->queue.qRemainNum, 1);
// }
static int32_t stmtCreateRequest(STscStmt2* pStmt) {
int32_t code = 0;
@ -76,7 +76,7 @@ static int32_t stmtCreateRequest(STscStmt2* pStmt) {
pStmt->reqid++;
}
if (pStmt->db != NULL) {
taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
taosMemoryFreeClear(pStmt->exec.pRequest->pDb);
pStmt->exec.pRequest->pDb = taosStrdup(pStmt->db);
}
if (TSDB_CODE_SUCCESS == code) {
@ -325,10 +325,10 @@ static int32_t stmtCleanBindInfo(STscStmt2* pStmt) {
return TSDB_CODE_SUCCESS;
}
static void stmtFreeTableBlkList(STableColsData* pTb) {
(void)qResetStmtColumns(pTb->aCol, true);
taosArrayDestroy(pTb->aCol);
}
// static void stmtFreeTableBlkList(STableColsData* pTb) {
// (void)qResetStmtColumns(pTb->aCol, true);
// taosArrayDestroy(pTb->aCol);
// }
static void stmtResetQueueTableBuf(STableBufInfo* pTblBuf, SStmtQueue* pQueue) {
pTblBuf->pCurBuff = taosArrayGetP(pTblBuf->pBufList, 0);
@ -895,32 +895,32 @@ int stmtPrepare2(TAOS_STMT2* stmt, const char* sql, unsigned long length) {
return TSDB_CODE_SUCCESS;
}
static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (!pSrc) {
return terrno;
}
STableDataCxt* pDst = NULL;
// static int32_t stmtInitStbInterlaceTableInfo(STscStmt2* pStmt) {
// STableDataCxt** pSrc = taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
// if (!pSrc) {
// return terrno;
// }
// STableDataCxt* pDst = NULL;
STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
pStmt->sql.siInfo.pDataCtx = pDst;
// STMT_ERR_RET(qCloneStmtDataBlock(&pDst, *pSrc, true));
// pStmt->sql.siInfo.pDataCtx = pDst;
SArray* pTblCols = NULL;
for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
pTblCols = taosArrayInit(20, POINTER_BYTES);
if (NULL == pTblCols) {
return terrno;
}
// SArray* pTblCols = NULL;
// for (int32_t i = 0; i < STMT_TABLE_COLS_NUM; i++) {
// pTblCols = taosArrayInit(20, POINTER_BYTES);
// if (NULL == pTblCols) {
// return terrno;
// }
if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
return terrno;
}
}
// if (taosArrayPush(pStmt->sql.siInfo.pTableCols, &pTblCols) == NULL) {
// return terrno;
// }
// }
pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
// pStmt->sql.siInfo.boundTags = pStmt->bInfo.boundTags;
return TSDB_CODE_SUCCESS;
}
// return TSDB_CODE_SUCCESS;
// }
int stmtIsInsert2(TAOS_STMT2* stmt, int* insert) {
STscStmt2* pStmt = (STscStmt2*)stmt;
@ -979,7 +979,7 @@ int stmtSetTbName2(TAOS_STMT2* stmt, const char* tbName) {
}
if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
// STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
}
int64_t startUs2 = taosGetTimestampUs();
@ -1009,7 +1009,7 @@ int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
// STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
}
SBoundColInfo* tags_info = (SBoundColInfo*)pStmt->bInfo.boundTags;
@ -1037,33 +1037,33 @@ int stmtSetTbTags2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* tags) {
return TSDB_CODE_SUCCESS;
}
static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
if (pStmt->errCode != TSDB_CODE_SUCCESS) {
return pStmt->errCode;
}
// static int stmtFetchColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_E** fields) {
// if (pStmt->errCode != TSDB_CODE_SUCCESS) {
// return pStmt->errCode;
// }
if (STMT_TYPE_QUERY == pStmt->sql.type) {
tscError("invalid operation to get query column fileds");
STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
}
// if (STMT_TYPE_QUERY == pStmt->sql.type) {
// tscError("invalid operation to get query column fileds");
// STMT_ERR_RET(TSDB_CODE_TSC_STMT_API_ERROR);
// }
STableDataCxt** pDataBlock = NULL;
// STableDataCxt** pDataBlock = NULL;
if (pStmt->sql.stbInterlaceMode) {
pDataBlock = &pStmt->sql.siInfo.pDataCtx;
} else {
pDataBlock =
(STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
if (NULL == pDataBlock) {
tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
STMT_ERR_RET(TSDB_CODE_APP_ERROR);
}
}
// if (pStmt->sql.stbInterlaceMode) {
// pDataBlock = &pStmt->sql.siInfo.pDataCtx;
// } else {
// pDataBlock =
// (STableDataCxt**)taosHashGet(pStmt->exec.pBlockHash, pStmt->bInfo.tbFName, strlen(pStmt->bInfo.tbFName));
// if (NULL == pDataBlock) {
// tscError("table %s not found in exec blockHash", pStmt->bInfo.tbFName);
// STMT_ERR_RET(TSDB_CODE_APP_ERROR);
// }
// }
STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
// STMT_ERR_RET(qBuildStmtColFields(*pDataBlock, fieldNum, fields));
return TSDB_CODE_SUCCESS;
}
// return TSDB_CODE_SUCCESS;
// }
static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIELD_ALL** fields) {
if (pStmt->errCode != TSDB_CODE_SUCCESS) {
@ -1117,46 +1117,46 @@ SArray* stmtGetFreeCol(STscStmt2* pStmt, int32_t* idx) {
}
}
*/
static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
if (NULL == pStmt->sql.siInfo.pVgroupHash) {
pStmt->sql.siInfo.pVgroupHash =
taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
}
if (NULL == pStmt->sql.siInfo.pVgroupList) {
pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
}
// static int32_t stmtAppendTablePostHandle(STscStmt2* pStmt, SStmtQNode* param) {
// if (NULL == pStmt->sql.siInfo.pVgroupHash) {
// pStmt->sql.siInfo.pVgroupHash =
// taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
// }
// if (NULL == pStmt->sql.siInfo.pVgroupList) {
// pStmt->sql.siInfo.pVgroupList = taosArrayInit(64, POINTER_BYTES);
// }
if (NULL == pStmt->sql.siInfo.pRequest) {
STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
(SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
// if (NULL == pStmt->sql.siInfo.pRequest) {
// STMT_ERR_RET(buildRequest(pStmt->taos->id, pStmt->sql.sqlStr, pStmt->sql.sqlLen, NULL, false,
// (SRequestObj**)&pStmt->sql.siInfo.pRequest, pStmt->reqid));
if (pStmt->reqid != 0) {
pStmt->reqid++;
}
pStmt->exec.pRequest->syncQuery = true;
// if (pStmt->reqid != 0) {
// pStmt->reqid++;
// }
// pStmt->exec.pRequest->syncQuery = true;
pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
}
// pStmt->sql.siInfo.requestId = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->requestId;
// pStmt->sql.siInfo.requestSelf = ((SRequestObj*)pStmt->sql.siInfo.pRequest)->self;
// }
if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
pStmt->sql.siInfo.tbFromHash = true;
}
// if (!pStmt->sql.siInfo.tbFromHash && pStmt->sql.siInfo.firstName[0] &&
// 0 == strcmp(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName)) {
// pStmt->sql.siInfo.tbFromHash = true;
// }
if (0 == pStmt->sql.siInfo.firstName[0]) {
tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
}
// if (0 == pStmt->sql.siInfo.firstName[0]) {
// tstrncpy(pStmt->sql.siInfo.firstName, pStmt->bInfo.tbName, TSDB_TABLE_NAME_LEN);
// }
param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
param->next = NULL;
// param->tblData.getFromHash = pStmt->sql.siInfo.tbFromHash;
// param->next = NULL;
(void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
// (void)atomic_add_fetch_64(&pStmt->sql.siInfo.tbRemainNum, 1);
stmtEnqueue(pStmt, param);
// stmtEnqueue(pStmt, param);
return TSDB_CODE_SUCCESS;
}
// return TSDB_CODE_SUCCESS;
// }
static FORCE_INLINE int32_t stmtGetTableColsFromCache(STscStmt2* pStmt, SArray** pTableCols) {
while (true) {
@ -1233,19 +1233,20 @@ static int stmtAddBatch2(TAOS_STMT2* stmt) {
STMT_ERR_RET(stmtSwitchStatus(pStmt, STMT_ADD_BATCH));
if (pStmt->sql.stbInterlaceMode) {
int64_t startUs2 = taosGetTimestampUs();
pStmt->stat.addBatchUs += startUs2 - startUs;
// todo:add stb interlace mode
// int64_t startUs2 = taosGetTimestampUs();
// pStmt->stat.addBatchUs += startUs2 - startUs;
pStmt->sql.siInfo.tableColsReady = false;
// pStmt->sql.siInfo.tableColsReady = false;
SStmtQNode* param = NULL;
STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
param->restoreTbCols = true;
param->next = NULL;
// SStmtQNode* param = NULL;
// STMT_ERR_RET(stmtAllocQNodeFromBuf(&pStmt->sql.siInfo.tbBuf, (void**)&param));
// param->restoreTbCols = true;
// param->next = NULL;
stmtEnqueue(pStmt, param);
// stmtEnqueue(pStmt, param);
return TSDB_CODE_SUCCESS;
// return TSDB_CODE_SUCCESS;
}
STMT_ERR_RET(stmtCacheBlock(pStmt));
@ -1366,7 +1367,7 @@ int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx) {
}
if (pStmt->sql.stbInterlaceMode && NULL == pStmt->sql.siInfo.pDataCtx) {
STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
// STMT_ERR_RET(stmtInitStbInterlaceTableInfo(pStmt));
}
STableDataCxt** pDataBlock = NULL;
@ -1450,7 +1451,8 @@ int stmtBindBatch2(TAOS_STMT2* stmt, TAOS_STMT2_BIND* bind, int32_t colIdx) {
pStmt->stat.bindDataUs3 += startUs4 - startUs3;
if (pStmt->sql.stbInterlaceMode) {
STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
// to do
// STMT_ERR_RET(stmtAppendTablePostHandle(pStmt, param));
} else {
STMT_ERR_RET(stmtAddBatch2(pStmt));
}
@ -1876,11 +1878,11 @@ int stmtGetParamNum2(TAOS_STMT2* stmt, int* nums) {
STMT_ERR_RET(stmtParseSql(pStmt));
}
if (STMT_TYPE_QUERY == pStmt->sql.type) {
*nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
} else {
STMT_ERR_RET(stmtFetchColFields2(stmt, nums, NULL));
}
// if (STMT_TYPE_QUERY == pStmt->sql.type) {
*nums = taosArrayGetSize(pStmt->sql.pQuery->pPlaceholderValues);
// } else {
// // STMT_ERR_RET(stmtFetchColFields2(stmt, nums, NULL));
// }
return TSDB_CODE_SUCCESS;
}

69
source/client/test/stmt2Test.cpp
View File

@ -880,75 +880,44 @@ TEST(stmt2Case, stmt2_query) {
taos_close(taos);
}
TEST(stmt2Case, stmt2_interlace) {
TEST(stmt2Case, stmt2_ntb_insert) {
TAOS* taos = taos_connect("localhost", "root", "taosdata", "", 0);
ASSERT_NE(taos, nullptr);
TAOS_STMT2_OPTION option = {0, true, true, NULL, NULL};
do_query(taos, "drop database if exists db1");
do_query(taos, "create database db1");
do_query(taos, "create table db1.ntb(ts timestamp, b binary(10))");
do_query(taos, "use db1");
do_query(taos, "drop database if exists db");
do_query(taos, "create database db");
do_query(taos, "create table db.ntb(ts timestamp, b binary(10))");
do_query(taos, "use db");
TAOS_STMT2* stmt = taos_stmt2_init(taos, &option);
ASSERT_NE(stmt, nullptr);
const char* sql = "insert into db1.ntb values(?,?)";
const char* sql = "insert into db.ntb values(?,?)";
int code = taos_stmt2_prepare(stmt, sql, 0);
ASSERT_EQ(code, 0);
ASSERT_EQ(terrno, 0);
for (int i = 0; i < 3; i++) {
int64_t ts[3] = {1591060628000 + i * 3, 1591060628001 + i * 3, 1591060628002 + i * 3};
int t64_len[3] = {sizeof(int64_t), sizeof(int64_t), sizeof(int64_t)};
int b_len[3] = {5,5,5};
TAOS_STMT2_BIND params1 = {TSDB_DATA_TYPE_TIMESTAMP, &ts[0], &t64_len[0], NULL, 3};
TAOS_STMT2_BIND params2 = {TSDB_DATA_TYPE_BINARY, (void*)"abcdefghijklmnopqrstuvwxyz", &b_len[0], NULL, 3};
TAOS_STMT2_BIND* paramv1 = &params1;
TAOS_STMT2_BIND* paramv2 = &params2;
for (int r = 0; r < 3; r++) {
// col params
int64_t** ts = (int64_t**)taosMemoryMalloc(3 * sizeof(int64_t*));
char** b = (char**)taosMemoryMalloc(3 * sizeof(char*));
int* ts_len = (int*)taosMemoryMalloc(3 * sizeof(int));
int* b_len = (int*)taosMemoryMalloc(3 * sizeof(int));
for (int i = 0; i < 3; i++) {
ts_len[i] = sizeof(int64_t);
b_len[i] = 1;
}
for (int i = 0; i < 3; i++) {
ts[i] = (int64_t*)taosMemoryMalloc(3 * sizeof(int64_t));
b[i] = (char*)taosMemoryMalloc(3 * sizeof(char));
for (int j = 0; j < 3; j++) {
ts[i][j] = 1591060628000 + r * 100000 + j;
b[i][j] = 'a' + j;
}
}
TAOS_STMT2_BINDV bindv1 = {1, NULL, NULL, &paramv1};
TAOS_STMT2_BINDV bindv2 = {1, NULL, NULL, &paramv2};
// bind params
TAOS_STMT2_BIND** paramv = (TAOS_STMT2_BIND**)taosMemoryMalloc(3 * sizeof(TAOS_STMT2_BIND*));
for (int i = 0; i < 3; i++) {
// create col params
paramv[i] = (TAOS_STMT2_BIND*)taosMemoryMalloc(2 * sizeof(TAOS_STMT2_BIND));
paramv[i][0] = {TSDB_DATA_TYPE_TIMESTAMP, &ts[i][0], &ts_len[0], NULL, 3};
paramv[i][1] = {TSDB_DATA_TYPE_BINARY, &b[i][0], &b_len[0], NULL, 3};
}
// bind
TAOS_STMT2_BINDV bindv = {1, NULL, NULL, paramv};
code = taos_stmt2_bind_param(stmt, &bindv, -1);
taos_stmt2_bind_param(stmt, &bindv1, 0);
taos_stmt2_bind_param(stmt, &bindv2, 1);
ASSERT_EQ(code, 0);
ASSERT_EQ(errno, 0);
// exec
code = taos_stmt2_exec(stmt, NULL);
ASSERT_EQ(code, 0);
ASSERT_EQ(errno, 0);
for (int i = 0; i < 3; i++) {
taosMemoryFree(paramv[i]);
taosMemoryFree(ts[i]);
taosMemoryFree(b[i]);
}
taosMemoryFree(ts);
taosMemoryFree(b);
taosMemoryFree(ts_len);
taosMemoryFree(b_len);
taosMemoryFree(paramv);
}
checkRows(taos, "select * from db1.ntb", 9);
checkRows(taos, "select * from db.ntb", 9);
taos_stmt2_close(stmt);
taos_close(taos);