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

fix:add logic for scan ins_columns

This commit is contained in:
wangmm0220 2023-01-10 20:32:31 +08:00
parent e1020b7967
commit 3b4ff4d094
7 changed files with 562 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
include/common/tmsg.h
View File

@ -382,6 +382,13 @@ static FORCE_INLINE void tDeleteSSchemaWrapper(SSchemaWrapper* pSchemaWrapper) {
}
}
static FORCE_INLINE void tDeleteSSchemaWrapperForHash(void* pSchemaWrapper) {
if (pSchemaWrapper) {
taosMemoryFree(((SSchemaWrapper*)pSchemaWrapper)->pSchema);
taosMemoryFree(pSchemaWrapper);
}
}
static FORCE_INLINE int32_t taosEncodeSSchema(void** buf, const SSchema* pSchema) {
int32_t tlen = 0;
tlen += taosEncodeFixedI8(buf, pSchema->type);

3
source/common/src/systable.c
View File

@ -178,13 +178,14 @@ static const SSysDbTableSchema userTagsSchema[] = {
static const SSysDbTableSchema userColsSchema[] = {
{.name = "table_name", .bytes = SYSTABLE_SCH_TABLE_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "table_type", .bytes = 21 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "col_name", .bytes = TSDB_COL_NAME_LEN - 1 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "col_type", .bytes = 32 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = false},
{.name = "col_length", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "col_precision", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "col_scale", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "col_nullable", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false},
{.name = "col_nullable", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = false}
};
static const SSysDbTableSchema userTblDistSchema[] = {

4
source/dnode/mnode/impl/src/mndShow.c
View File

@ -19,6 +19,7 @@
#include "systable.h"
#define SHOW_STEP_SIZE 100
#define SHOW_COLS_STEP_SIZE 4096
static SShowObj *mndCreateShowObj(SMnode *pMnode, SRetrieveTableReq *pReq);
static void mndFreeShowObj(SShowObj *pShow);
@ -228,6 +229,9 @@ static int32_t mndProcessRetrieveSysTableReq(SRpcMsg *pReq) {
}
}
if(pShow->type == TSDB_MGMT_TABLE_COL){ // expend capacity for ins_columns
rowsToRead = SHOW_COLS_STEP_SIZE;
}
ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type];
if (retrieveFp == NULL) {
mndReleaseShowObj(pShow, false);

449
source/dnode/mnode/impl/src/mndStb.c
View File

@ -43,6 +43,7 @@ static int32_t mndProcessAlterStbReq(SRpcMsg *pReq);
static int32_t mndProcessDropStbReq(SRpcMsg *pReq);
static int32_t mndProcessTableMetaReq(SRpcMsg *pReq);
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter);
static int32_t mndProcessTableCfgReq(SRpcMsg *pReq);
static int32_t mndAlterStbImp(SMnode *pMnode, SRpcMsg *pReq, SDbObj *pDb, SStbObj *pStb, bool needRsp,
@ -69,10 +70,14 @@ int32_t mndInitStb(SMnode *pMnode) {
mndSetMsgHandle(pMnode, TDMT_MND_TABLE_META, mndProcessTableMetaReq);
mndSetMsgHandle(pMnode, TDMT_MND_TTL_TIMER, mndProcessTtlTimer);
mndSetMsgHandle(pMnode, TDMT_MND_TABLE_CFG, mndProcessTableCfgReq);
// mndSetMsgHandle(pMnode, TDMT_MND_SYSTABLE_RETRIEVE, mndProcessRetrieveStbReq);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_STB, mndRetrieveStb);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_STB, mndCancelGetNextStb);
mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_COL, mndRetrieveStbCol);
mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_COL, mndCancelGetNextStb);
return sdbSetTable(pMnode->pSdb, table);
}
@ -2489,6 +2494,283 @@ void mndExtractTbNameFromStbFullName(const char *stbFullName, char *dst, int32_t
}
}
//static int32_t mndProcessRetrieveStbReq(SRpcMsg *pReq) {
// SMnode *pMnode = pReq->info.node;
// SShowMgmt *pMgmt = &pMnode->showMgmt;
// SShowObj *pShow = NULL;
// int32_t rowsToRead = SHOW_STEP_SIZE;
// int32_t rowsRead = 0;
//
// SRetrieveTableReq retrieveReq = {0};
// if (tDeserializeSRetrieveTableReq(pReq->pCont, pReq->contLen, &retrieveReq) != 0) {
// terrno = TSDB_CODE_INVALID_MSG;
// return -1;
// }
//
// SMnode *pMnode = pReq->info.node;
// SSdb *pSdb = pMnode->pSdb;
// int32_t numOfRows = 0;
// SDbObj *pDb = NULL;
// ESdbStatus objStatus = 0;
//
// SUserObj *pUser = mndAcquireUser(pMnode, pReq->info.conn.user);
// if (pUser == NULL) return 0;
// bool sysinfo = pUser->sysInfo;
//
// // Append the information_schema database into the result.
//// if (!pShow->sysDbRsp) {
//// SDbObj infoschemaDb = {0};
//// setInformationSchemaDbCfg(pMnode, &infoschemaDb);
//// size_t numOfTables = 0;
//// getVisibleInfosTablesNum(sysinfo, &numOfTables);
//// mndDumpDbInfoData(pMnode, pBlock, &infoschemaDb, pShow, numOfRows, numOfTables, true, 0, 1);
////
//// numOfRows += 1;
////
//// SDbObj perfschemaDb = {0};
//// setPerfSchemaDbCfg(pMnode, &perfschemaDb);
//// numOfTables = 0;
//// getPerfDbMeta(NULL, &numOfTables);
//// mndDumpDbInfoData(pMnode, pBlock, &perfschemaDb, pShow, numOfRows, numOfTables, true, 0, 1);
////
//// numOfRows += 1;
//// pShow->sysDbRsp = true;
//// }
//
// SSDataBlock* p = buildInfoSchemaTableMetaBlock(TSDB_INS_TABLE_COLS);
// blockDataEnsureCapacity(p, rowsToRead);
//
// size_t size = 0;
// const SSysTableMeta* pSysDbTableMeta = NULL;
//
// getInfosDbMeta(&pSysDbTableMeta, &size);
// p->info.rows = buildDbColsInfoBlock(sysinfo, p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB);
//
// getPerfDbMeta(&pSysDbTableMeta, &size);
// p->info.rows = buildDbColsInfoBlock(sysinfo, p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB);
//
// blockDataDestroy(p);
//
//
// while (numOfRows < rowsToRead) {
// pShow->pIter = sdbFetchAll(pSdb, SDB_DB, pShow->pIter, (void **)&pDb, &objStatus, true);
// if (pShow->pIter == NULL) break;
// if (strncmp(retrieveReq.db, pDb->name, strlen(retrieveReq.db)) != 0){
// continue;
// }
// if (mndCheckDbPrivilege(pMnode, pReq->info.conn.user, MND_OPER_READ_OR_WRITE_DB, pDb) != 0) {
// continue;
// }
//
// while (numOfRows < rowsToRead) {
// pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
// if (pShow->pIter == NULL) break;
//
// if (pDb != NULL && pStb->dbUid != pDb->uid) {
// sdbRelease(pSdb, pStb);
// continue;
// }
//
// cols = 0;
//
// SName name = {0};
// char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
// mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
// varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
//
// SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)stbName, false);
//
// char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
// tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB);
// tNameGetDbName(&name, varDataVal(db));
// varDataSetLen(db, strlen(varDataVal(db)));
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)db, false);
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)&pStb->createdTime, false);
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)&pStb->numOfColumns, false);
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)&pStb->numOfTags, false);
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)&pStb->updateTime, false); // number of tables
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// if (pStb->commentLen > 0) {
// char comment[TSDB_TB_COMMENT_LEN + VARSTR_HEADER_SIZE] = {0};
// STR_TO_VARSTR(comment, pStb->comment);
// colDataAppend(pColInfo, numOfRows, comment, false);
// } else if (pStb->commentLen == 0) {
// char comment[VARSTR_HEADER_SIZE + VARSTR_HEADER_SIZE] = {0};
// STR_TO_VARSTR(comment, "");
// colDataAppend(pColInfo, numOfRows, comment, false);
// } else {
// colDataAppendNULL(pColInfo, numOfRows);
// }
//
// char watermark[64 + VARSTR_HEADER_SIZE] = {0};
// sprintf(varDataVal(watermark), "%" PRId64 "a,%" PRId64 "a", pStb->watermark[0], pStb->watermark[1]);
// varDataSetLen(watermark, strlen(varDataVal(watermark)));
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)watermark, false);
//
// char maxDelay[64 + VARSTR_HEADER_SIZE] = {0};
// sprintf(varDataVal(maxDelay), "%" PRId64 "a,%" PRId64 "a", pStb->maxdelay[0], pStb->maxdelay[1]);
// varDataSetLen(maxDelay, strlen(varDataVal(maxDelay)));
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)maxDelay, false);
//
// char rollup[160 + VARSTR_HEADER_SIZE] = {0};
// int32_t rollupNum = (int32_t)taosArrayGetSize(pStb->pFuncs);
// char *sep = ", ";
// int32_t sepLen = strlen(sep);
// int32_t rollupLen = sizeof(rollup) - VARSTR_HEADER_SIZE - 2;
// for (int32_t i = 0; i < rollupNum; ++i) {
// char *funcName = taosArrayGet(pStb->pFuncs, i);
// if (i) {
// strncat(varDataVal(rollup), sep, rollupLen);
// rollupLen -= sepLen;
// }
// strncat(varDataVal(rollup), funcName, rollupLen);
// rollupLen -= strlen(funcName);
// }
// varDataSetLen(rollup, strlen(varDataVal(rollup)));
//
// pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
// colDataAppend(pColInfo, numOfRows, (const char *)rollup, false);
//
// numOfRows++;
// sdbRelease(pSdb, pStb);
// }
//
// if (pDb != NULL) {
// mndReleaseDb(pMnode, pDb);
// }
//
// sdbRelease(pSdb, pDb);
// }
//
// pShow->numOfRows += numOfRows;
// mndReleaseUser(pMnode, pUser);
//
//
//
//
//
//
//
//
// ShowRetrieveFp retrieveFp = pMgmt->retrieveFps[pShow->type];
// if (retrieveFp == NULL) {
// mndReleaseShowObj(pShow, false);
// terrno = TSDB_CODE_MSG_NOT_PROCESSED;
// mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr());
// return -1;
// }
//
// mDebug("show:0x%" PRIx64 ", start retrieve data, type:%d", pShow->id, pShow->type);
// if (retrieveReq.user[0] != 0) {
// memcpy(pReq->info.conn.user, retrieveReq.user, TSDB_USER_LEN);
// } else {
// memcpy(pReq->info.conn.user, TSDB_DEFAULT_USER, strlen(TSDB_DEFAULT_USER) + 1);
// }
// if (retrieveReq.db[0] && mndCheckShowPrivilege(pMnode, pReq->info.conn.user, pShow->type, retrieveReq.db) != 0) {
// return -1;
// }
//
// int32_t numOfCols = pShow->pMeta->numOfColumns;
//
// SSDataBlock *pBlock = createDataBlock();
// for (int32_t i = 0; i < numOfCols; ++i) {
// SColumnInfoData idata = {0};
//
// SSchema *p = &pShow->pMeta->pSchemas[i];
//
// idata.info.bytes = p->bytes;
// idata.info.type = p->type;
// idata.info.colId = p->colId;
// blockDataAppendColInfo(pBlock, &idata);
// }
//
// blockDataEnsureCapacity(pBlock, rowsToRead);
//
// if (mndCheckRetrieveFinished(pShow)) {
// mDebug("show:0x%" PRIx64 ", read finished, numOfRows:%d", pShow->id, pShow->numOfRows);
// rowsRead = 0;
// } else {
// rowsRead = (*retrieveFp)(pReq, pShow, pBlock, rowsToRead);
// if (rowsRead < 0) {
// terrno = rowsRead;
// mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id);
// mndReleaseShowObj(pShow, true);
// blockDataDestroy(pBlock);
// return -1;
// }
//
// pBlock->info.rows = rowsRead;
// mDebug("show:0x%" PRIx64 ", stop retrieve data, rowsRead:%d numOfRows:%d", pShow->id, rowsRead, pShow->numOfRows);
// }
//
// size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * pShow->pMeta->numOfColumns +
// blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(taosArrayGetSize(pBlock->pDataBlock));
//
// SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
// if (pRsp == NULL) {
// mndReleaseShowObj(pShow, false);
// terrno = TSDB_CODE_OUT_OF_MEMORY;
// mError("show:0x%" PRIx64 ", failed to retrieve data since %s", pShow->id, terrstr());
// blockDataDestroy(pBlock);
// return -1;
// }
//
// pRsp->handle = htobe64(pShow->id);
//
// if (rowsRead > 0) {
// char *pStart = pRsp->data;
// SSchema *ps = pShow->pMeta->pSchemas;
//
// *(int32_t *)pStart = htonl(pShow->pMeta->numOfColumns);
// pStart += sizeof(int32_t); // number of columns
//
// for (int32_t i = 0; i < pShow->pMeta->numOfColumns; ++i) {
// SSysTableSchema *pSchema = (SSysTableSchema *)pStart;
// pSchema->bytes = htonl(ps[i].bytes);
// pSchema->colId = htons(ps[i].colId);
// pSchema->type = ps[i].type;
//
// pStart += sizeof(SSysTableSchema);
// }
//
// int32_t len = blockEncode(pBlock, pStart, pShow->pMeta->numOfColumns);
// }
//
// pRsp->numOfRows = htonl(rowsRead);
// pRsp->precision = TSDB_TIME_PRECISION_MILLI; // millisecond time precision
// pReq->info.rsp = pRsp;
// pReq->info.rspLen = size;
//
// if (rowsRead == 0 || rowsRead < rowsToRead) {
// pRsp->completed = 1;
// mDebug("show:0x%" PRIx64 ", retrieve completed", pShow->id);
// mndReleaseShowObj(pShow, true);
// } else {
// mDebug("show:0x%" PRIx64 ", retrieve not completed yet", pShow->id);
// mndReleaseShowObj(pShow, false);
// }
//
// blockDataDestroy(pBlock);
// return TSDB_CODE_SUCCESS;
//}
static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
@ -2599,6 +2881,173 @@ static int32_t mndRetrieveStb(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBloc
return numOfRows;
}
static int32_t buildDbColsInfoBlock(const SSDataBlock* p, const SSysTableMeta* pSysDbTableMeta, size_t size,
const char* dbName) {
char tName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
char dName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
int32_t numOfRows = p->info.rows;
STR_TO_VARSTR(dName, dbName);
STR_TO_VARSTR(typeName, "SYSTEM_TABLE");
for (int32_t i = 0; i < size; ++i) {
const SSysTableMeta* pm = &pSysDbTableMeta[i];
// if (pm->sysInfo) {
// continue;
// }
STR_TO_VARSTR(tName, pm->name);
for(int32_t j = 0; j < pm->colNum; j++){
// table name
SColumnInfoData* pColInfoData = taosArrayGet(p->pDataBlock, 0);
colDataAppend(pColInfoData, numOfRows, tName, false);
pColInfoData = taosArrayGet(p->pDataBlock, 1);
colDataAppend(pColInfoData, numOfRows, typeName, false);
// database name
pColInfoData = taosArrayGet(p->pDataBlock, 2);
colDataAppend(pColInfoData, numOfRows, dName, false);
// col name
char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(colName, pm->schema[j].name);
pColInfoData = taosArrayGet(p->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, colName, false);
// col type
int8_t colType = pm->schema[j].type;
pColInfoData = taosArrayGet(p->pDataBlock, 4);
char colTypeStr[VARSTR_HEADER_SIZE + 32];
int colTypeLen = sprintf(varDataVal(colTypeStr), "%s", tDataTypes[colType].name);
if (colType == TSDB_DATA_TYPE_VARCHAR) {
colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d)",
(int32_t)(pm->schema[j].bytes - VARSTR_HEADER_SIZE));
} else if (colType == TSDB_DATA_TYPE_NCHAR) {
colTypeLen += sprintf(
varDataVal(colTypeStr) + colTypeLen, "(%d)",
(int32_t)((pm->schema[j].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(colTypeStr, colTypeLen);
colDataAppend(pColInfoData, numOfRows, (char*)colTypeStr, false);
pColInfoData = taosArrayGet(p->pDataBlock, 5);
colDataAppend(pColInfoData, numOfRows, (const char*)&pm->schema[j].bytes, false);
for (int32_t k = 6; k <= 8; ++k) {
pColInfoData = taosArrayGet(p->pDataBlock, k);
colDataAppendNULL(pColInfoData, numOfRows);
}
numOfRows += 1;
}
}
return numOfRows;
}
static int32_t buildSysDbColsInfo(SSDataBlock* p) {
size_t size = 0;
const SSysTableMeta* pSysDbTableMeta = NULL;
getInfosDbMeta(&pSysDbTableMeta, &size);
p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_INFORMATION_SCHEMA_DB);
getPerfDbMeta(&pSysDbTableMeta, &size);
p->info.rows = buildDbColsInfoBlock(p, pSysDbTableMeta, size, TSDB_PERFORMANCE_SCHEMA_DB);
return p->info.rows;
}
static int32_t mndRetrieveStbCol(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
SMnode *pMnode = pReq->info.node;
SSdb *pSdb = pMnode->pSdb;
SStbObj *pStb = NULL;
int32_t cols = 0;
int32_t numOfRows = buildSysDbColsInfo(pBlock);
SDbObj *pDb = NULL;
if (strlen(pShow->db) > 0) {
pDb = mndAcquireDb(pMnode, pShow->db);
if (pDb == NULL) return terrno;
}
char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(typeName, "SUPER_TABLE");
while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_STB, pShow->pIter, (void **)&pStb);
if (pShow->pIter == NULL) break;
if (pDb != NULL && pStb->dbUid != pDb->uid) {
sdbRelease(pSdb, pStb);
continue;
}
cols = 0;
SName name = {0};
char stbName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
mndExtractTbNameFromStbFullName(pStb->name, &stbName[VARSTR_HEADER_SIZE], TSDB_TABLE_NAME_LEN);
varDataSetLen(stbName, strlen(&stbName[VARSTR_HEADER_SIZE]));
char db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
tNameFromString(&name, pStb->db, T_NAME_ACCT | T_NAME_DB);
tNameGetDbName(&name, varDataVal(db));
varDataSetLen(db, strlen(varDataVal(db)));
for(int i = 0; i < pStb->numOfColumns; i++){
SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)stbName, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, typeName, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char *)db, false);
// col name
char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(colName, pStb->pColumns[i].name);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, colName, false);
// col type
int8_t colType = pStb->pColumns[i].type;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
char colTypeStr[VARSTR_HEADER_SIZE + 32];
int colTypeLen = sprintf(varDataVal(colTypeStr), "%s", tDataTypes[colType].name);
if (colType == TSDB_DATA_TYPE_VARCHAR) {
colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d)",
(int32_t)(pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE));
} else if (colType == TSDB_DATA_TYPE_NCHAR) {
colTypeLen += sprintf(
varDataVal(colTypeStr) + colTypeLen, "(%d)",
(int32_t)((pStb->pColumns[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(colTypeStr, colTypeLen);
colDataAppend(pColInfo, numOfRows, (char*)colTypeStr, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppend(pColInfo, numOfRows, (const char*)&pStb->pColumns[i].bytes, false);
while(cols < pShow->numOfColumns) {
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataAppendNULL(pColInfo, numOfRows);
}
}
numOfRows++;
sdbRelease(pSdb, pStb);
}
if (pDb != NULL) {
mndReleaseDb(pMnode, pDb);
}
pShow->numOfRows += numOfRows;
return numOfRows;
}
static void mndCancelGetNextStb(SMnode *pMnode, void *pIter) {
SSdb *pSdb = pMnode->pSdb;
sdbCancelFetch(pSdb, pIter);

2
source/dnode/vnode/inc/vnode.h
View File

@ -150,7 +150,7 @@ typedef struct SMTbCursor SMTbCursor;
SMTbCursor *metaOpenTbCursor(SMeta *pMeta);
void metaCloseTbCursor(SMTbCursor *pTbCur);
int32_t metaTbCursorNext(SMTbCursor *pTbCur);
int32_t metaTbCursorNext(SMTbCursor *pTbCur, ETableType jumpTableType);
#endif
// tsdb

4
source/dnode/vnode/src/meta/metaQuery.c
View File

@ -310,7 +310,7 @@ void metaCloseTbCursor(SMTbCursor *pTbCur) {
}
}
int metaTbCursorNext(SMTbCursor *pTbCur) {
int metaTbCursorNext(SMTbCursor *pTbCur, ETableType jumpTableType) {
int ret;
void *pBuf;
STbCfg tbCfg;
@ -324,7 +324,7 @@ int metaTbCursorNext(SMTbCursor *pTbCur) {
tDecoderClear(&pTbCur->mr.coder);
metaGetTableEntryByVersion(&pTbCur->mr, ((SUidIdxVal *)pTbCur->pVal)[0].version, *(tb_uid_t *)pTbCur->pKey);
if (pTbCur->mr.me.type == TSDB_SUPER_TABLE) {
if (pTbCur->mr.me.type == jumpTableType) {
continue;
}

135
source/libs/executor/src/sysscanoperator.c
View File

@ -67,6 +67,7 @@ typedef struct SSysTableScanInfo {
SLoadRemoteDataInfo loadInfo;
int32_t tbnameSlotId;
bool isGetStableCols;
} SSysTableScanInfo;
typedef struct {
@ -140,8 +141,9 @@ static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo,
SMetaReader* smrChildTable, const char* dbname, const char* tableName,
int32_t* pNumOfRows, const SSDataBlock* dataBlock);
static int32_t sysTableUserColsFillOneTableCols(const SSysTableScanInfo* pInfo, SMetaReader* smrTable, const char* dbname,
int32_t* pNumOfRows, const SSDataBlock* dataBlock);
static int32_t sysTableUserColsFillOneTableCols(const SSysTableScanInfo* pInfo, const char* dbname,
int32_t* pNumOfRows, const SSDataBlock* dataBlock,
char* tName, SSchemaWrapper* schemaRow, char* tableType);
static void relocateAndFilterSysTagsScanResult(SSysTableScanInfo* pInfo, int32_t numOfRows, SSDataBlock* dataBlock,
SFilterInfo* pFilterInfo);
@ -457,14 +459,38 @@ static SSDataBlock* sysTableScanUserCols(SOperatorInfo* pOperator) {
return NULL;
}
if (smrTable.me.type == TSDB_CHILD_TABLE) {
if (smrTable.me.type == TSDB_SUPER_TABLE) {
metaReaderClear(&smrTable);
blockDataDestroy(dataBlock);
pInfo->loadInfo.totalRows = 0;
return NULL;
}
sysTableUserColsFillOneTableCols(pInfo, &smrTable, dbname, &numOfRows, dataBlock);
if (smrTable.me.type == TSDB_CHILD_TABLE) {
int64_t suid = smrTable.me.ctbEntry.suid;
metaReaderClear(&smrTable);
metaReaderInit(&smrTable, pInfo->readHandle.meta, 0);
code = metaGetTableEntryByUid(&smrTable, suid);
if (code != TSDB_CODE_SUCCESS) {
// terrno has been set by metaGetTableEntryByName, therefore, return directly
metaReaderClear(&smrTable);
blockDataDestroy(dataBlock);
pInfo->loadInfo.totalRows = 0;
return NULL;
}
}
char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
SSchemaWrapper *schemaRow = NULL;
if(smrTable.me.type == TSDB_SUPER_TABLE){
schemaRow = &smrTable.me.stbEntry.schemaRow;
STR_TO_VARSTR(typeName, "CHILD_TABLE");
}else if(smrTable.me.type == TSDB_NORMAL_TABLE){
schemaRow = &smrTable.me.ntbEntry.schemaRow;
STR_TO_VARSTR(typeName, "NORMAL_TABLE");
}
sysTableUserColsFillOneTableCols(pInfo, dbname, &numOfRows, dataBlock, tableName, schemaRow, typeName);
metaReaderClear(&smrTable);
if (numOfRows > 0) {
@ -482,12 +508,50 @@ static SSDataBlock* sysTableScanUserCols(SOperatorInfo* pOperator) {
pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
}
while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
if (pInfo->pCur->mr.me.type == TSDB_CHILD_TABLE) {
SHashObj *stableSchema = taosHashInit(64, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), false, HASH_NO_LOCK);
taosHashSetFreeFp(stableSchema, tDeleteSSchemaWrapperForHash);
while ((ret = metaTbCursorNext(pInfo->pCur, TSDB_TABLE_MAX)) == 0) {
char typeName[TSDB_TABLE_FNAME_LEN + VARSTR_HEADER_SIZE] = {0};
char tableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
SSchemaWrapper *schemaRow = NULL;
if(pInfo->pCur->mr.me.type == TSDB_SUPER_TABLE){
void *schema = taosHashGet(stableSchema, &pInfo->pCur->mr.me.uid, sizeof(int64_t));
if(schema == NULL){
SSchemaWrapper *schemaWrapper = tCloneSSchemaWrapper(&pInfo->pCur->mr.me.stbEntry.schemaRow);
taosHashPut(stableSchema, &pInfo->pCur->mr.me.uid, sizeof(int64_t), &schemaWrapper, POINTER_BYTES);
}
continue;
}else if (pInfo->pCur->mr.me.type == TSDB_CHILD_TABLE) {
STR_TO_VARSTR(typeName, "CHILD_TABLE");
STR_TO_VARSTR(tableName, pInfo->pCur->mr.me.name);
int64_t suid = pInfo->pCur->mr.me.ctbEntry.suid;
void *schema = taosHashGet(stableSchema, &pInfo->pCur->mr.me.ctbEntry.suid, sizeof(int64_t));
if(schema != NULL){
schemaRow = *(SSchemaWrapper **)schema;
}else{
tDecoderClear(&pInfo->pCur->mr.coder);
int code = metaGetTableEntryByUid(&pInfo->pCur->mr, suid);
if (code != TSDB_CODE_SUCCESS) {
// terrno has been set by metaGetTableEntryByName, therefore, return directly
qError("sysTableScanUserCols get meta by suid:%"PRId64 " error, code:%d", suid, code);
blockDataDestroy(dataBlock);
pInfo->loadInfo.totalRows = 0;
return NULL;
}
schemaRow = &pInfo->pCur->mr.me.stbEntry.schemaRow;
}
}else if(pInfo->pCur->mr.me.type == TSDB_NORMAL_TABLE){
schemaRow = &pInfo->pCur->mr.me.ntbEntry.schemaRow;
STR_TO_VARSTR(typeName, "NORMAL_TABLE");
STR_TO_VARSTR(tableName, pInfo->pCur->mr.me.name);
}else{
continue;
}
sysTableUserColsFillOneTableCols(pInfo, &pInfo->pCur->mr, dbname, &numOfRows, dataBlock);
sysTableUserColsFillOneTableCols(pInfo, dbname, &numOfRows, dataBlock, tableName, schemaRow, typeName);
if (numOfRows >= pOperator->resultInfo.capacity) {
relocateAndFilterSysTagsScanResult(pInfo, numOfRows, dataBlock, pOperator->exprSupp.pFilterInfo);
@ -593,7 +657,7 @@ static SSDataBlock* sysTableScanUserTags(SOperatorInfo* pOperator) {
pInfo->pCur = metaOpenTbCursor(pInfo->readHandle.meta);
}
while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
while ((ret = metaTbCursorNext(pInfo->pCur, TSDB_SUPER_TABLE)) == 0) {
if (pInfo->pCur->mr.me.type != TSDB_CHILD_TABLE) {
continue;
}
@ -830,64 +894,59 @@ static int32_t sysTableUserTagsFillOneTableTags(const SSysTableScanInfo* pInfo,
return TSDB_CODE_SUCCESS;
}
static int32_t sysTableUserColsFillOneTableCols(const SSysTableScanInfo* pInfo, SMetaReader* smrTable, const char* dbname,
int32_t* pNumOfRows, const SSDataBlock* dataBlock) {
char tableName[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(tableName, smrTable->me.name);
SSchemaWrapper schemaRow = {0};
if(smrTable->me.type == TSDB_SUPER_TABLE){
schemaRow = smrTable->me.stbEntry.schemaRow;
}else if(smrTable->me.type == TSDB_NORMAL_TABLE){
schemaRow = smrTable->me.ntbEntry.schemaRow;
static int32_t sysTableUserColsFillOneTableCols(const SSysTableScanInfo* pInfo, const char* dbname,
int32_t* pNumOfRows, const SSDataBlock* dataBlock, char* tName,
SSchemaWrapper* schemaRow, char* tableType) {
if(schemaRow == NULL){
qError("sysTableUserColsFillOneTableCols schemaRow is NULL");
return TSDB_CODE_SUCCESS;
}
int32_t numOfRows = *pNumOfRows;
int32_t numOfCols = schemaRow.nCols;
int32_t numOfCols = schemaRow->nCols;
for (int32_t i = 0; i < numOfCols; ++i) {
SColumnInfoData* pColInfoData = NULL;
// table name
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 0);
colDataAppend(pColInfoData, numOfRows, tableName, tableName == NULL ? true : false);
colDataAppend(pColInfoData, numOfRows, tName, false);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 1);
colDataAppend(pColInfoData, numOfRows, tableType, false);
// database name
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 1);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 2);
colDataAppend(pColInfoData, numOfRows, dbname, false);
// col name
char colName[TSDB_COL_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
STR_TO_VARSTR(colName, schemaRow.pSchema[i].name);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 2);
STR_TO_VARSTR(colName, schemaRow->pSchema[i].name);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 3);
colDataAppend(pColInfoData, numOfRows, colName, false);
// col type
int8_t colType = schemaRow.pSchema[i].type;
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 3);
int8_t colType = schemaRow->pSchema[i].type;
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4);
char colTypeStr[VARSTR_HEADER_SIZE + 32];
int colTypeLen = sprintf(varDataVal(colTypeStr), "%s", tDataTypes[colType].name);
if (colType == TSDB_DATA_TYPE_VARCHAR) {
colTypeLen += sprintf(varDataVal(colTypeStr) + colTypeLen, "(%d)",
(int32_t)(schemaRow.pSchema[i].bytes - VARSTR_HEADER_SIZE));
(int32_t)(schemaRow->pSchema[i].bytes - VARSTR_HEADER_SIZE));
} else if (colType == TSDB_DATA_TYPE_NCHAR) {
colTypeLen += sprintf(
varDataVal(colTypeStr) + colTypeLen, "(%d)",
(int32_t)((schemaRow.pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
(int32_t)((schemaRow->pSchema[i].bytes - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE));
}
varDataSetLen(colTypeStr, colTypeLen);
colDataAppend(pColInfoData, numOfRows, (char*)colTypeStr, false);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 4);
colDataAppend(pColInfoData, numOfRows, (const char*)&schemaRow.pSchema[i].bytes, false);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 5);
colDataAppend(pColInfoData, numOfRows, NULL, true);
colDataAppend(pColInfoData, numOfRows, (const char*)&schemaRow->pSchema[i].bytes, false);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 6);
colDataAppend(pColInfoData, numOfRows, NULL, true);
pColInfoData = taosArrayGet(dataBlock->pDataBlock, 7);
colDataAppend(pColInfoData, numOfRows, NULL, true);
for (int32_t j = 6; j <= 8; ++j) {
pColInfoData = taosArrayGet(dataBlock->pDataBlock, j);
colDataAppendNULL(pColInfoData, numOfRows);
}
++numOfRows;
}
@ -1197,7 +1256,7 @@ static SSDataBlock* sysTableBuildUserTables(SOperatorInfo* pOperator) {
char n[TSDB_TABLE_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
int32_t ret = 0;
while ((ret = metaTbCursorNext(pInfo->pCur)) == 0) {
while ((ret = metaTbCursorNext(pInfo->pCur, TSDB_SUPER_TABLE)) == 0) {
STR_TO_VARSTR(n, pInfo->pCur->mr.me.name);
// table name
@ -1489,7 +1548,7 @@ static SSDataBlock* doSysTableScan(SOperatorInfo* pOperator) {
pBlock = sysTableScanUserTables(pOperator);
} else if (strncasecmp(name, TSDB_INS_TABLE_TAGS, TSDB_TABLE_FNAME_LEN) == 0) {
pBlock = sysTableScanUserTags(pOperator);
} else if (strncasecmp(name, TSDB_INS_TABLE_COLS, TSDB_TABLE_FNAME_LEN) == 0) {
} else if (strncasecmp(name, TSDB_INS_TABLE_COLS, TSDB_TABLE_FNAME_LEN) == 0 && pInfo->readHandle.mnd == NULL) {
pBlock = sysTableScanUserCols(pOperator);
} else if (strncasecmp(name, TSDB_INS_TABLE_STABLES, TSDB_TABLE_FNAME_LEN) == 0 && pInfo->showRewrite &&
IS_SYS_DBNAME(dbName)) {