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

taos_stmt2_get_stb_fields support insert into ? using stb tags(?) values(?)

This commit is contained in:
Pengrongkun 2024-11-29 10:05:35 +08:00
parent a7f655a784
commit 653eff8c1a
7 changed files with 108 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
include/libs/parser/parser.h
View File

@ -27,7 +27,7 @@ extern "C" {
typedef struct SStmtCallback {
TAOS_STMT* pStmt;
int32_t (*getTbNameFn)(TAOS_STMT*, char**);
int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, SName*, bool, SHashObj*, SHashObj*, const char*);
int32_t (*setInfoFn)(TAOS_STMT*, STableMeta*, void*, SName*, bool, SHashObj*, SHashObj*, const char*, bool);
int32_t (*getExecInfoFn)(TAOS_STMT*, SHashObj**, SHashObj**);
} SStmtCallback;
@ -147,7 +147,8 @@ int32_t qBindStmtColsValue(void* pBlock, SArray* pCols, TAOS_MULTI_BIND* bind, c
int32_t qBindStmtSingleColValue(void* pBlock, SArray* pCols, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen,
int32_t colIdx, int32_t rowNum);
int32_t qBuildStmtColFields(void* pDataBlock, int32_t* fieldNum, TAOS_FIELD_E** fields);
int32_t qBuildStmtStbColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_STB** fields);
int32_t qBuildStmtStbColFields(void* pBlock, void* boundTags, bool hasCtbName, int32_t* fieldNum,
TAOS_FIELD_STB** fields);
int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD_E** fields);
int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, const char* sTableName, char* tName,
TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
@ -177,8 +178,8 @@ int32_t smlBindData(SQuery* handle, bool dataFormat, SArray* tags, SArray* colsS
STableMeta* pTableMeta, char* tableName, const char* sTableName, int32_t sTableNameLen, int32_t ttl,
char* msgBuf, int32_t msgBufLen);
int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash);
int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, void* fields,
int numFields, bool needChangeLength, char* errstr, int32_t errstrLen, bool raw);
int rawBlockBindData(SQuery* query, STableMeta* pTableMeta, void* data, SVCreateTbReq* pCreateTb, void* fields,
int numFields, bool needChangeLength, char* errstr, int32_t errstrLen, bool raw);
int32_t rewriteToVnodeModifyOpStmt(SQuery* pQuery, SArray* pBufArray);
int32_t serializeVgroupsCreateTableBatch(SHashObj* pVgroupHashmap, SArray** pOut);

4
source/client/inc/clientStmt.h
View File

@ -64,12 +64,12 @@ typedef struct SStmtBindInfo {
int32_t sBindLastIdx;
int8_t tbType;
bool tagsCached;
bool preCtbname;
void *boundTags;
char tbName[TSDB_TABLE_FNAME_LEN];
char tbFName[TSDB_TABLE_FNAME_LEN];
char stbFName[TSDB_TABLE_FNAME_LEN];
SName sname;
char statbName[TSDB_TABLE_FNAME_LEN];
} SStmtBindInfo;
@ -133,7 +133,6 @@ typedef struct SStmtQueue {
uint64_t qRemainNum;
} SStmtQueue;
typedef struct STscStmt {
STscObj *taos;
SCatalog *pCatalog;
@ -204,7 +203,6 @@ extern char *gStmtStatusStr[];
} \
} while (0)
#define STMT_FLOG(param, ...) qFatal("stmt:%p " param, pStmt, __VA_ARGS__)
#define STMT_ELOG(param, ...) qError("stmt:%p " param, pStmt, __VA_ARGS__)
#define STMT_DLOG(param, ...) qDebug("stmt:%p " param, pStmt, __VA_ARGS__)

2
source/client/src/clientStmt.c
View File

@ -260,7 +260,7 @@ int32_t stmtUpdateExecInfo(TAOS_STMT* stmt, SHashObj* pVgHash, SHashObj* pBlockH
}
int32_t stmtUpdateInfo(TAOS_STMT* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) {
SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName, bool preCtbname) {
STscStmt* pStmt = (STscStmt*)stmt;
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));

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

@ -178,7 +178,7 @@ static int32_t stmtGetTbName(TAOS_STMT2* stmt, char** tbName) {
}
static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName,
const char* sTableName, bool autoCreateTbl) {
const char* sTableName, bool autoCreateTbl, bool preCtbname) {
STscStmt2* pStmt = (STscStmt2*)stmt;
char tbFName[TSDB_TABLE_FNAME_LEN];
int32_t code = tNameExtractFullName(tbName, tbFName);
@ -202,6 +202,7 @@ static int32_t stmtUpdateBindInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void
pStmt->bInfo.boundTags = tags;
pStmt->bInfo.tagsCached = false;
pStmt->bInfo.preCtbname = preCtbname;
tstrncpy(pStmt->bInfo.stbFName, sTableName, sizeof(pStmt->bInfo.stbFName));
return TSDB_CODE_SUCCESS;
@ -217,10 +218,10 @@ static int32_t stmtUpdateExecInfo(TAOS_STMT2* stmt, SHashObj* pVgHash, SHashObj*
}
static int32_t stmtUpdateInfo(TAOS_STMT2* stmt, STableMeta* pTableMeta, void* tags, SName* tbName, bool autoCreateTbl,
SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName) {
SHashObj* pVgHash, SHashObj* pBlockHash, const char* sTableName, bool preCtbname) {
STscStmt2* pStmt = (STscStmt2*)stmt;
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl));
STMT_ERR_RET(stmtUpdateBindInfo(stmt, pTableMeta, tags, tbName, sTableName, autoCreateTbl, preCtbname));
STMT_ERR_RET(stmtUpdateExecInfo(stmt, pVgHash, pBlockHash));
pStmt->sql.autoCreateTbl = autoCreateTbl;
@ -1092,7 +1093,7 @@ static int stmtFetchStbColFields2(STscStmt2* pStmt, int32_t* fieldNum, TAOS_FIEL
}
}
STMT_ERR_RET(qBuildStmtStbColFields(*pDataBlock, fieldNum, fields));
STMT_ERR_RET(qBuildStmtStbColFields(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.preCtbname, fieldNum, fields));
return TSDB_CODE_SUCCESS;
}

11
source/libs/parser/src/parInsertSql.c
View File

@ -31,6 +31,7 @@ typedef struct SInsertParseContext {
bool needTableTagVal;
bool needRequest; // whether or not request server
bool isStmtBind; // whether is stmt bind
bool preCtbname;
} SInsertParseContext;
typedef int32_t (*_row_append_fn_t)(SMsgBuf* pMsgBuf, const void* value, int32_t len, void* param);
@ -1831,6 +1832,7 @@ static int32_t doGetStbRowValues(SInsertParseContext* pCxt, SVnodeModifyOpStmt*
if (TK_NK_QUESTION == pToken->type) {
pCxt->isStmtBind = true;
if (pCols->pColIndex[i] == tbnameIdx) {
pCxt->preCtbname = false;
*bFoundTbName = true;
}
if (NULL == pCxt->pComCxt->pStmtCb) {
@ -2571,10 +2573,11 @@ static int32_t checkTableClauseFirstToken(SInsertParseContext* pCxt, SVnodeModif
}
int32_t code = (*pCxt->pComCxt->pStmtCb->getTbNameFn)(pCxt->pComCxt->pStmtCb->pStmt, &tbName);
if (code != TSDB_CODE_SUCCESS) {
return code;
pCxt->preCtbname = true;
} else {
pTbName->z = tbName;
pTbName->n = strlen(tbName);
}
pTbName->z = tbName;
pTbName->n = strlen(tbName);
}
if (pCxt->isStmtBind) {
@ -2599,7 +2602,7 @@ static int32_t setStmtInfo(SInsertParseContext* pCxt, SVnodeModifyOpStmt* pStmt)
SStmtCallback* pStmtCb = pCxt->pComCxt->pStmtCb;
int32_t code = (*pStmtCb->setInfoFn)(pStmtCb->pStmt, pStmt->pTableMeta, tags, &pStmt->targetTableName,
pStmt->usingTableProcessing, pStmt->pVgroupsHashObj, pStmt->pTableBlockHashObj,
pStmt->usingTableName.tname);
pStmt->usingTableName.tname, pCxt->preCtbname);
memset(&pCxt->tags, 0, sizeof(pCxt->tags));
pStmt->pVgroupsHashObj = NULL;

79
source/libs/parser/src/parInsertStmt.c
View File

@ -939,39 +939,72 @@ int32_t buildBoundFields(int32_t numOfBound, int16_t* boundColumns, SSchema* pSc
}
int32_t buildStbBoundFields(SBoundColInfo boundColsInfo, SSchema* pSchema, int32_t* fieldNum, TAOS_FIELD_STB** fields,
STableMeta* pMeta) {
STableMeta* pMeta, void* boundTags, bool preCtbname) {
SBoundColInfo* tags = (SBoundColInfo*)boundTags;
int32_t numOfBound = boundColsInfo.numOfBound + tags->numOfBound + (preCtbname ? 1 : 0);
int32_t idx = 0;
if (fields != NULL) {
*fields = taosMemoryCalloc(boundColsInfo.numOfBound, sizeof(TAOS_FIELD_STB));
*fields = taosMemoryCalloc(numOfBound, sizeof(TAOS_FIELD_STB));
if (NULL == *fields) {
return terrno;
}
SSchema* schema = &pSchema[boundColsInfo.pColIndex[0]];
if (TSDB_DATA_TYPE_TIMESTAMP == schema->type) {
(*fields)[0].precision = pMeta->tableInfo.precision;
if (preCtbname && numOfBound != boundColsInfo.numOfBound) {
(*fields)[idx].field_type = TAOS_FIELD_TBNAME;
tstrncpy((*fields)[idx].name, "tbname", sizeof((*fields)[idx].name));
(*fields)[idx].type = TSDB_DATA_TYPE_VARCHAR;
(*fields)[idx].bytes = TSDB_TABLE_FNAME_LEN;
idx++;
}
for (int32_t i = 0; i < boundColsInfo.numOfBound; ++i) {
int16_t idx = boundColsInfo.pColIndex[i];
if (tags->numOfBound > 0) {
SSchema* pSchema = getTableTagSchema(pMeta);
if (idx == pMeta->tableInfo.numOfColumns + pMeta->tableInfo.numOfTags) {
(*fields)[i].field_type = TAOS_FIELD_TBNAME;
tstrncpy((*fields)[i].name, "tbname", sizeof((*fields)[i].name));
continue;
} else if (idx < pMeta->tableInfo.numOfColumns) {
(*fields)[i].field_type = TAOS_FIELD_COL;
} else {
(*fields)[i].field_type = TAOS_FIELD_TAG;
if (TSDB_DATA_TYPE_TIMESTAMP == pSchema->type) {
(*fields)[0].precision = pMeta->tableInfo.precision;
}
schema = &pSchema[idx];
tstrncpy((*fields)[i].name, schema->name, sizeof((*fields)[i].name));
(*fields)[i].type = schema->type;
(*fields)[i].bytes = schema->bytes;
for (int32_t i = 0; i < tags->numOfBound; ++i) {
(*fields)[idx].field_type = TAOS_FIELD_TAG;
SSchema* schema = &pSchema[tags->pColIndex[i]];
tstrncpy((*fields)[idx].name, schema->name, sizeof((*fields)[i].name));
(*fields)[idx].type = schema->type;
(*fields)[idx].bytes = schema->bytes;
idx++;
}
}
if (boundColsInfo.numOfBound > 0) {
SSchema* schema = &pSchema[boundColsInfo.pColIndex[0]];
if (TSDB_DATA_TYPE_TIMESTAMP == schema->type) {
(*fields)[0].precision = pMeta->tableInfo.precision;
}
for (int32_t i = 0; i < boundColsInfo.numOfBound; ++i) {
int16_t idxCol = boundColsInfo.pColIndex[i];
if (idxCol == pMeta->tableInfo.numOfColumns + pMeta->tableInfo.numOfTags) {
(*fields)[idx].field_type = TAOS_FIELD_TBNAME;
tstrncpy((*fields)[i].name, "tbname", sizeof((*fields)[idx].name));
continue;
} else if (idxCol < pMeta->tableInfo.numOfColumns) {
(*fields)[idx].field_type = TAOS_FIELD_COL;
} else {
(*fields)[idx].field_type = TAOS_FIELD_TAG;
}
schema = &pSchema[idxCol];
tstrncpy((*fields)[idx].name, schema->name, sizeof((*fields)[idx].name));
(*fields)[idx].type = schema->type;
(*fields)[idx].bytes = schema->bytes;
idx++;
}
}
}
*fieldNum = boundColsInfo.numOfBound;
*fieldNum = numOfBound;
return TSDB_CODE_SUCCESS;
}
@ -1018,7 +1051,8 @@ int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_E** fiel
return TSDB_CODE_SUCCESS;
}
int32_t qBuildStmtStbColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_STB** fields) {
int32_t qBuildStmtStbColFields(void* pBlock, void* boundTags, bool preCtbname, int32_t* fieldNum,
TAOS_FIELD_STB** fields) {
STableDataCxt* pDataBlock = (STableDataCxt*)pBlock;
SSchema* pSchema = getTableColumnSchema(pDataBlock->pMeta);
if (pDataBlock->boundColsInfo.numOfBound <= 0) {
@ -1030,7 +1064,8 @@ int32_t qBuildStmtStbColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD_STB**
return TSDB_CODE_SUCCESS;
}
CHECK_CODE(buildStbBoundFields(pDataBlock->boundColsInfo, pSchema, fieldNum, fields, pDataBlock->pMeta));
CHECK_CODE(buildStbBoundFields(pDataBlock->boundColsInfo, pSchema, fieldNum, fields, pDataBlock->pMeta, boundTags,
preCtbname));
return TSDB_CODE_SUCCESS;
}

34
tests/script/api/stmt2-get-fields.c Normal file → Executable file
View File

@ -21,7 +21,7 @@ void getFields(TAOS *taos, const char *sql) {
if (code != 0) {
printf("failed get col,ErrCode: 0x%x, ErrMessage: %s.\n", code, taos_stmt2_error(stmt));
} else {
printf("col nums:%d\n", fieldNum);
printf("bind nums:%d\n", fieldNum);
for (int i = 0; i < fieldNum; i++) {
printf("field[%d]: %s, data_type:%d, field_type:%d\n", i, pFields[i].name, pFields[i].type,
pFields[i].field_type);
@ -76,7 +76,7 @@ void do_stmt(TAOS *taos) {
// case 4 : INSERT INTO db.? using db.stb TAGS(?,?) VALUES(?,?)
// not support this clause
sql = "insert into db.? using db.stb tags(?, ?) values(?,?)";
printf("case 4 (not support): %s\n", sql);
printf("case 4 : %s\n", sql);
getFields(taos, sql);
// case 5 : INSERT INTO db.stb(t1,t2,ts,b) values(?,?,?,?)
@ -114,6 +114,36 @@ void do_stmt(TAOS *taos) {
sql = "insert into db.ntb(nts,ni) values(?,?)";
printf("case 10 : %s\n", sql);
getFields(taos, sql);
// case 11 : insert into db.? values(?,?)
// normal table must have tbnam
sql = "insert into db.? values(?,?)";
printf("case 11 (normal table must have tbname): %s\n", sql);
getFields(taos, sql);
// case 12 : insert into db.? using db.stb(t2,t1) tags(?, ?) (b,ts)values(?,?)
// disordered
sql = "insert into db.? using db.stb(t2,t1) tags(?, ?) (b,ts)values(?,?)";
printf("case 12 : %s\n", sql);
getFields(taos, sql);
// case 13 : insert into db.? using db.stb tags(?, ?) values(?,?)
// no field name
sql = "insert into db.? using db.stb tags(?, ?) values(?,?)";
printf("case 13 : %s\n", sql);
getFields(taos, sql);
// case 14 : insert into db.? using db.stb tags(?, ?) values(?,?)
// less para
sql = "insert into db.? using db.stb (t2)tags(?) (ts)values(?)";
printf("case 14 : %s\n", sql);
getFields(taos, sql);
// case 15 : insert into db.? using db.stb tags(?, ?) values(?,?)
// less para
sql = "insert into db.d0 (ts)values(?)";
printf("case 15 : %s\n", sql);
getFields(taos, sql);
}
int main() {