From b420bf4c3d8d17fe725ddda6ccbb04180db11434 Mon Sep 17 00:00:00 2001 From: Xiaoyu Wang Date: Mon, 15 Aug 2022 14:45:59 +0800 Subject: [PATCH] enh: insert client optimize --- source/libs/parser/src/parInsert.c | 27 ++- source/libs/parser/test/parInsertTest.cpp | 270 +++------------------- source/libs/parser/test/parTestUtil.cpp | 64 +++-- 3 files changed, 96 insertions(+), 265 deletions(-) diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c index 4e3d18ca93..44ad68356c 100644 --- a/source/libs/parser/src/parInsert.c +++ b/source/libs/parser/src/parInsert.c @@ -1193,7 +1193,7 @@ static int parseOneRow(SInsertParseContext* pCxt, STableDataBlocks* pDataBlocks, tdSRowEnd(pBuilder); *gotRow = true; - + #ifdef TD_DEBUG_PRINT_ROW STSchema* pSTSchema = tdGetSTSChemaFromSSChema(schema, spd->numOfCols, 1); tdSRowPrint(row, pSTSchema, __func__); @@ -1365,6 +1365,20 @@ static void destroyInsertParseContext(SInsertParseContext* pCxt) { destroyBlockArrayList(pCxt->pVgDataBlocks); } +static int32_t parseTableName(SInsertParseContext* pCxt, SToken* pTbnameToken, SName* pName, char* pDbFName, + char* pTbFName) { + int32_t code = createSName(pName, pTbnameToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg); + if (TSDB_CODE_SUCCESS == code) { + tNameExtractFullName(pName, pTbFName); + code = taosHashPut(pCxt->pTableNameHashObj, pTbFName, strlen(pTbFName), pName, sizeof(SName)); + } + if (TSDB_CODE_SUCCESS == code) { + tNameGetFullDbName(pName, pDbFName); + code = taosHashPut(pCxt->pDbFNameHashObj, pDbFName, strlen(pDbFName), pDbFName, TSDB_DB_FNAME_LEN); + } + return code; +} + // tb_name // [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)] // [(field1_name, ...)] @@ -1418,13 +1432,7 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { NEXT_TOKEN(pCxt->pSql, sToken); if (!pCxt->pComCxt->async || TK_USING == sToken.type) { - CHECK_CODE(createSName(&name, &tbnameToken, pCxt->pComCxt->acctId, pCxt->pComCxt->db, &pCxt->msg)); - - tNameExtractFullName(&name, tbFName); - CHECK_CODE(taosHashPut(pCxt->pTableNameHashObj, tbFName, strlen(tbFName), &name, sizeof(SName))); - - tNameGetFullDbName(&name, dbFName); - CHECK_CODE(taosHashPut(pCxt->pDbFNameHashObj, dbFName, strlen(dbFName), dbFName, sizeof(dbFName))); + CHECK_CODE(parseTableName(pCxt, &tbnameToken, &name, dbFName, tbFName)); } bool existedUsing = false; @@ -1445,6 +1453,9 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) { } if (TK_USING == sToken.type) { + if (pCxt->pComCxt->async) { + CHECK_CODE(parseTableName(pCxt, &tbnameToken, &name, dbFName, tbFName)); + } CHECK_CODE(parseUsingClause(pCxt, tbNum, &name, tbFName)); NEXT_TOKEN(pCxt->pSql, sToken); autoCreateTbl = true; diff --git a/source/libs/parser/test/parInsertTest.cpp b/source/libs/parser/test/parInsertTest.cpp index a15fc6e8fa..ddf15ec67b 100644 --- a/source/libs/parser/test/parInsertTest.cpp +++ b/source/libs/parser/test/parInsertTest.cpp @@ -13,21 +13,13 @@ * along with this program. If not, see . */ -#include - #include -#include "mockCatalogService.h" -#include "os.h" -#include "parInt.h" +#include "parTestUtil.h" using namespace std; -using namespace std::placeholders; -using namespace testing; -namespace { -string toString(int32_t code) { return tstrerror(code); } -} // namespace +namespace ParserTest { // syntax: // INSERT INTO @@ -36,260 +28,60 @@ string toString(int32_t code) { return tstrerror(code); } // [(field1_name, ...)] // VALUES (field1_value, ...) [(field1_value2, ...) ...] | FILE csv_file_path // [...]; -class InsertTest : public Test { - protected: - InsertTest() : res_(nullptr) {} - ~InsertTest() { reset(); } - - void setDatabase(const string& acctId, const string& db) { - acctId_ = acctId; - db_ = db; - } - - void bind(const char* sql) { - reset(); - cxt_.acctId = atoi(acctId_.c_str()); - cxt_.db = (char*)db_.c_str(); - strcpy(sqlBuf_, sql); - cxt_.sqlLen = strlen(sql); - sqlBuf_[cxt_.sqlLen] = '\0'; - cxt_.pSql = sqlBuf_; - cxt_.pUser = "root"; - } - - int32_t run() { - code_ = parseInsertSql(&cxt_, &res_, nullptr); - if (code_ != TSDB_CODE_SUCCESS) { - cout << "code:" << toString(code_) << ", msg:" << errMagBuf_ << endl; - } - return code_; - } - - int32_t runAsync() { - cxt_.async = true; - bool request = true; - unique_ptr > metaCache( - new SParseMetaCache(), std::bind(_destoryParseMetaCache, _1, cref(request))); - code_ = parseInsertSyntax(&cxt_, &res_, metaCache.get()); - if (code_ != TSDB_CODE_SUCCESS) { - cout << "parseInsertSyntax code:" << toString(code_) << ", msg:" << errMagBuf_ << endl; - return code_; - } - - unique_ptr catalogReq(new SCatalogReq(), - MockCatalogService::destoryCatalogReq); - code_ = buildCatalogReq(&cxt_, metaCache.get(), catalogReq.get()); - if (code_ != TSDB_CODE_SUCCESS) { - cout << "buildCatalogReq code:" << toString(code_) << ", msg:" << errMagBuf_ << endl; - return code_; - } - - unique_ptr metaData(new SMetaData(), MockCatalogService::destoryMetaData); - g_mockCatalogService->catalogGetAllMeta(catalogReq.get(), metaData.get()); - - metaCache.reset(new SParseMetaCache()); - request = false; - code_ = putMetaDataToCache(catalogReq.get(), metaData.get(), metaCache.get(), true); - if (code_ != TSDB_CODE_SUCCESS) { - cout << "putMetaDataToCache code:" << toString(code_) << ", msg:" << errMagBuf_ << endl; - return code_; - } - - code_ = parseInsertSql(&cxt_, &res_, metaCache.get()); - if (code_ != TSDB_CODE_SUCCESS) { - cout << "parseInsertSql code:" << toString(code_) << ", msg:" << errMagBuf_ << endl; - return code_; - } - - return code_; - } - - void dumpReslut() { - SVnodeModifOpStmt* pStmt = getVnodeModifStmt(res_); - size_t num = taosArrayGetSize(pStmt->pDataBlocks); - cout << "payloadType:" << (int32_t)pStmt->payloadType << ", insertType:" << pStmt->insertType - << ", numOfVgs:" << num << endl; - for (size_t i = 0; i < num; ++i) { - SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(pStmt->pDataBlocks, i); - cout << "vgId:" << vg->vg.vgId << ", numOfTables:" << vg->numOfTables << ", dataSize:" << vg->size << endl; - SSubmitReq* submit = (SSubmitReq*)vg->pData; - cout << "length:" << ntohl(submit->length) << ", numOfBlocks:" << ntohl(submit->numOfBlocks) << endl; - int32_t numOfBlocks = ntohl(submit->numOfBlocks); - SSubmitBlk* blk = (SSubmitBlk*)(submit + 1); - for (int32_t i = 0; i < numOfBlocks; ++i) { - cout << "Block:" << i << endl; - cout << "\tuid:" << be64toh(blk->uid) << ", tid:" << be64toh(blk->suid) << ", sversion:" << ntohl(blk->sversion) - << ", dataLen:" << ntohl(blk->dataLen) << ", schemaLen:" << ntohl(blk->schemaLen) - << ", numOfRows:" << ntohl(blk->numOfRows) << endl; - blk = (SSubmitBlk*)(blk->data + ntohl(blk->dataLen)); - } - } - } - - void checkReslut(int32_t numOfTables, int32_t numOfRows1, int32_t numOfRows2 = -1) { - SVnodeModifOpStmt* pStmt = getVnodeModifStmt(res_); - ASSERT_EQ(pStmt->payloadType, PAYLOAD_TYPE_KV); - ASSERT_EQ(pStmt->insertType, TSDB_QUERY_TYPE_INSERT); - size_t num = taosArrayGetSize(pStmt->pDataBlocks); - ASSERT_GE(num, 0); - for (size_t i = 0; i < num; ++i) { - SVgDataBlocks* vg = (SVgDataBlocks*)taosArrayGetP(pStmt->pDataBlocks, i); - ASSERT_EQ(vg->numOfTables, numOfTables); - ASSERT_GE(vg->size, 0); - SSubmitReq* submit = (SSubmitReq*)vg->pData; - ASSERT_GE(ntohl(submit->length), 0); - ASSERT_GE(ntohl(submit->numOfBlocks), 0); - int32_t numOfBlocks = ntohl(submit->numOfBlocks); - SSubmitBlk* blk = (SSubmitBlk*)(submit + 1); - for (int32_t i = 0; i < numOfBlocks; ++i) { - ASSERT_EQ(ntohl(blk->numOfRows), (0 == i ? numOfRows1 : (numOfRows2 > 0 ? numOfRows2 : numOfRows1))); - blk = (SSubmitBlk*)(blk->data + ntohl(blk->dataLen)); - } - } - } - - private: - static const int max_err_len = 1024; - static const int max_sql_len = 1024 * 1024; - - static void _destoryParseMetaCache(SParseMetaCache* pMetaCache, bool request) { - destoryParseMetaCache(pMetaCache, request); - delete pMetaCache; - } - - void reset() { - memset(&cxt_, 0, sizeof(cxt_)); - memset(errMagBuf_, 0, max_err_len); - cxt_.pMsg = errMagBuf_; - cxt_.msgLen = max_err_len; - code_ = TSDB_CODE_SUCCESS; - qDestroyQuery(res_); - res_ = nullptr; - } - - SVnodeModifOpStmt* getVnodeModifStmt(SQuery* pQuery) { return (SVnodeModifOpStmt*)pQuery->pRoot; } - - string acctId_; - string db_; - char errMagBuf_[max_err_len]; - char sqlBuf_[max_sql_len]; - SParseContext cxt_; - int32_t code_; - SQuery* res_; -}; +class ParserInsertTest : public ParserTestBase {}; // INSERT INTO tb_name [(field1_name, ...)] VALUES (field1_value, ...) -TEST_F(InsertTest, singleTableSingleRowTest) { - setDatabase("root", "test"); +TEST_F(ParserInsertTest, singleTableSingleRowTest) { + useDb("root", "test"); - bind("insert into t1 values (now, 1, 'beijing', 3, 4, 5)"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); - dumpReslut(); - checkReslut(1, 1); + run("INSERT INTO t1 VALUES (now, 1, 'beijing', 3, 4, 5)"); - bind("insert into t1 (ts, c1, c2, c3, c4, c5) values (now, 1, 'beijing', 3, 4, 5)"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); - - bind("insert into t1 values (now, 1, 'beijing', 3, 4, 5)"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); - dumpReslut(); - checkReslut(1, 1); - - bind("insert into t1 (ts, c1, c2, c3, c4, c5) values (now, 1, 'beijing', 3, 4, 5)"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); + run("INSERT INTO t1 (ts, c1, c2, c3, c4, c5) VALUES (now, 1, 'beijing', 3, 4, 5)"); } // INSERT INTO tb_name VALUES (field1_value, ...)(field1_value, ...) -TEST_F(InsertTest, singleTableMultiRowTest) { - setDatabase("root", "test"); +TEST_F(ParserInsertTest, singleTableMultiRowTest) { + useDb("root", "test"); - bind( - "insert into t1 values (now, 1, 'beijing', 3, 4, 5)(now+1s, 2, 'shanghai', 6, 7, 8)" + run("INSERT INTO t1 VALUES (now, 1, 'beijing', 3, 4, 5)" + "(now+1s, 2, 'shanghai', 6, 7, 8)" "(now+2s, 3, 'guangzhou', 9, 10, 11)"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); - dumpReslut(); - checkReslut(1, 3); - - bind( - "insert into t1 values (now, 1, 'beijing', 3, 4, 5)(now+1s, 2, 'shanghai', 6, 7, 8)" - "(now+2s, 3, 'guangzhou', 9, 10, 11)"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); } // INSERT INTO tb1_name VALUES (field1_value, ...) tb2_name VALUES (field1_value, ...) -TEST_F(InsertTest, multiTableSingleRowTest) { - setDatabase("root", "test"); +TEST_F(ParserInsertTest, multiTableSingleRowTest) { + useDb("root", "test"); - bind("insert into st1s1 values (now, 1, \"beijing\") st1s2 values (now, 10, \"131028\")"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); - dumpReslut(); - checkReslut(2, 1); - - bind("insert into st1s1 values (now, 1, \"beijing\") st1s2 values (now, 10, \"131028\")"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); + run("INSERT INTO st1s1 VALUES (now, 1, 'beijing') st1s2 VALUES (now, 10, '131028')"); } // INSERT INTO tb1_name VALUES (field1_value, ...) tb2_name VALUES (field1_value, ...) -TEST_F(InsertTest, multiTableMultiRowTest) { - setDatabase("root", "test"); +TEST_F(ParserInsertTest, multiTableMultiRowTest) { + useDb("root", "test"); - bind( - "insert into st1s1 values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")" - " st1s2 values (now, 10, \"131028\")(now+1s, 20, \"132028\")"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); - dumpReslut(); - checkReslut(2, 3, 2); - - bind( - "insert into st1s1 values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")" - " st1s2 values (now, 10, \"131028\")(now+1s, 20, \"132028\")"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); + run("INSERT INTO " + "st1s1 VALUES (now, 1, 'beijing')(now+1s, 2, 'shanghai')(now+2s, 3, 'guangzhou') " + "st1s2 VALUES (now, 10, '131028')(now+1s, 20, '132028')"); } // INSERT INTO // tb1_name USING st1_name [(tag1_name, ...)] TAGS (tag1_value, ...) VALUES (field1_value, ...) // tb2_name USING st2_name [(tag1_name, ...)] TAGS (tag1_value, ...) VALUES (field1_value, ...) -TEST_F(InsertTest, autoCreateTableTest) { - setDatabase("root", "test"); +TEST_F(ParserInsertTest, autoCreateTableTest) { + useDb("root", "test"); - bind( - "insert into st1s1 using st1 tags(1, 'wxy', now) " - "values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); - dumpReslut(); - checkReslut(1, 3); + run("INSERT INTO st1s1 USING st1 TAGS(1, 'wxy', now) " + "VALUES (now, 1, 'beijing')(now+1s, 2, 'shanghai')(now+2s, 3, 'guangzhou')"); - bind( - "insert into st1s1 using st1 (tag1, tag2) tags(1, 'wxy') values (now, 1, \"beijing\")" - "(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); + run("INSERT INTO st1s1 USING st1 (tag1, tag2) TAGS(1, 'wxy') (ts, c1, c2) " + "VALUES (now, 1, 'beijing')(now+1s, 2, 'shanghai')(now+2s, 3, 'guangzhou')"); - bind( - "insert into st1s1 using st1 tags(1, 'wxy', now) " - "values (now, 1, \"beijing\")(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); + run("INSERT INTO st1s1 (ts, c1, c2) USING st1 (tag1, tag2) TAGS(1, 'wxy') " + "VALUES (now, 1, 'beijing')(now+1s, 2, 'shanghai')(now+2s, 3, 'guangzhou')"); - bind( - "insert into st1s1 using st1 (tag1, tag2) tags(1, 'wxy') values (now, 1, \"beijing\")" - "(now+1s, 2, \"shanghai\")(now+2s, 3, \"guangzhou\")"); - ASSERT_EQ(runAsync(), TSDB_CODE_SUCCESS); - - bind( - "insert into st1s1 using st1 tags(1, 'wxy', now) values (now, 1, \"beijing\")" - "st1s1 using st1 tags(1, 'wxy', now) values (now+1s, 2, \"shanghai\")"); - ASSERT_EQ(run(), TSDB_CODE_SUCCESS); + run("INSERT INTO " + "st1s1 USING st1 (tag1, tag2) TAGS(1, 'wxy') (ts, c1, c2) VALUES (now, 1, 'beijing') " + "st1s2 (ts, c1, c2) USING st1 TAGS(2, 'abc', now) VALUES (now+1s, 2, 'shanghai')"); } -TEST_F(InsertTest, toleranceTest) { - setDatabase("root", "test"); - - bind("insert into"); - ASSERT_NE(run(), TSDB_CODE_SUCCESS); - bind("insert into t"); - ASSERT_NE(run(), TSDB_CODE_SUCCESS); - - bind("insert into"); - ASSERT_NE(runAsync(), TSDB_CODE_SUCCESS); - bind("insert into t"); - ASSERT_NE(runAsync(), TSDB_CODE_SUCCESS); -} +} // namespace ParserTest diff --git a/source/libs/parser/test/parTestUtil.cpp b/source/libs/parser/test/parTestUtil.cpp index 10a9561300..98281b7bf0 100644 --- a/source/libs/parser/test/parTestUtil.cpp +++ b/source/libs/parser/test/parTestUtil.cpp @@ -233,8 +233,9 @@ class ParserTestBaseImpl { DO_WITH_THROW(g_mockCatalogService->catalogGetAllMeta, pCatalogReq, pMetaData); } - void doPutMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMetaData, SParseMetaCache* pMetaCache) { - DO_WITH_THROW(putMetaDataToCache, pCatalogReq, pMetaData, pMetaCache, false); + void doPutMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMetaData, SParseMetaCache* pMetaCache, + bool isInsertValues) { + DO_WITH_THROW(putMetaDataToCache, pCatalogReq, pMetaData, pMetaCache, isInsertValues); } void doAuthenticate(SParseContext* pCxt, SQuery* pQuery, SParseMetaCache* pMetaCache) { @@ -261,7 +262,9 @@ class ParserTestBaseImpl { void doParseSqlSyntax(SParseContext* pCxt, SQuery** pQuery, SCatalogReq* pCatalogReq) { DO_WITH_THROW(qParseSqlSyntax, pCxt, pQuery, pCatalogReq); ASSERT_NE(*pQuery, nullptr); - res_.parsedAst_ = toString((*pQuery)->pRoot); + if (nullptr != (*pQuery)->pRoot) { + res_.parsedAst_ = toString((*pQuery)->pRoot); + } } void doAnalyseSqlSemantic(SParseContext* pCxt, const SCatalogReq* pCatalogReq, const SMetaData* pMetaData, @@ -270,6 +273,17 @@ class ParserTestBaseImpl { res_.calcConstAst_ = toString(pQuery->pRoot); } + void doParseInsertSql(SParseContext* pCxt, SQuery** pQuery, SParseMetaCache* pMetaCache) { + DO_WITH_THROW(parseInsertSql, pCxt, pQuery, pMetaCache); + ASSERT_NE(*pQuery, nullptr); + res_.parsedAst_ = toString((*pQuery)->pRoot); + } + + void doParseInsertSyntax(SParseContext* pCxt, SQuery** pQuery, SParseMetaCache* pMetaCache) { + DO_WITH_THROW(parseInsertSyntax, pCxt, pQuery, pMetaCache); + ASSERT_NE(*pQuery, nullptr); + } + string toString(const SNode* pRoot) { char* pStr = NULL; int32_t len = 0; @@ -287,15 +301,20 @@ class ParserTestBaseImpl { SParseContext cxt = {0}; setParseContext(sql, &cxt); - unique_ptr query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), _destroyQuery); - doParse(&cxt, query.get()); - SQuery* pQuery = *(query.get()); + if (qIsInsertValuesSql(cxt.pSql, cxt.sqlLen)) { + unique_ptr query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), _destroyQuery); + doParseInsertSql(&cxt, query.get(), nullptr); + } else { + unique_ptr query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), _destroyQuery); + doParse(&cxt, query.get()); + SQuery* pQuery = *(query.get()); - doAuthenticate(&cxt, pQuery, nullptr); + doAuthenticate(&cxt, pQuery, nullptr); - doTranslate(&cxt, pQuery, nullptr); + doTranslate(&cxt, pQuery, nullptr); - doCalculateConstant(&cxt, pQuery); + doCalculateConstant(&cxt, pQuery); + } if (g_dump) { dump(); @@ -338,13 +357,18 @@ class ParserTestBaseImpl { setParseContext(sql, &cxt, true); unique_ptr query((SQuery**)taosMemoryCalloc(1, sizeof(SQuery*)), _destroyQuery); - doParse(&cxt, query.get()); - SQuery* pQuery = *(query.get()); - - bool request = true; + bool request = true; unique_ptr > metaCache( new SParseMetaCache(), bind(_destoryParseMetaCache, _1, cref(request))); - doCollectMetaKey(&cxt, pQuery, metaCache.get()); + bool isInsertValues = qIsInsertValuesSql(cxt.pSql, cxt.sqlLen); + if (isInsertValues) { + doParseInsertSyntax(&cxt, query.get(), metaCache.get()); + } else { + doParse(&cxt, query.get()); + doCollectMetaKey(&cxt, *(query.get()), metaCache.get()); + } + + SQuery* pQuery = *(query.get()); unique_ptr catalogReq(new SCatalogReq(), MockCatalogService::destoryCatalogReq); @@ -358,13 +382,17 @@ class ParserTestBaseImpl { metaCache.reset(new SParseMetaCache()); request = false; - doPutMetaDataToCache(catalogReq.get(), metaData.get(), metaCache.get()); + doPutMetaDataToCache(catalogReq.get(), metaData.get(), metaCache.get(), isInsertValues); - doAuthenticate(&cxt, pQuery, metaCache.get()); + if (isInsertValues) { + doParseInsertSql(&cxt, query.get(), metaCache.get()); + } else { + doAuthenticate(&cxt, pQuery, metaCache.get()); - doTranslate(&cxt, pQuery, metaCache.get()); + doTranslate(&cxt, pQuery, metaCache.get()); - doCalculateConstant(&cxt, pQuery); + doCalculateConstant(&cxt, pQuery); + } } catch (const TerminateFlag& e) { // success and terminate } catch (const runtime_error& e) {