Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
提交 bd42e715 编写于 作者: S Shengliang Guan
浏览文件
操作

The table name stored in mnode is changed to a variable-length string

上级 53a0e401
隐藏空白更改
内联 并排
Showing with 164 addition and 109 deletion
src/inc/taosdef.h
浏览文件 @ bd42e715
...@@ -236,7 +236,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size); ...@@ -236,7 +236,7 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_PAYLOAD_SIZE (TSDB_DEFAULT_PKT_SIZE - 100) #define TSDB_PAYLOAD_SIZE (TSDB_DEFAULT_PKT_SIZE - 100)
#define TSDB_DEFAULT_PAYLOAD_SIZE 1024 // default payload size #define TSDB_DEFAULT_PAYLOAD_SIZE 1024 // default payload size
#define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth #define TSDB_EXTRA_PAYLOAD_SIZE 128 // extra bytes for auth
#define TSDB_SQLCMD_SIZE 1024 #define TSDB_CQ_SQL_SIZE 1024
#define TSDB_MAX_VNODES 256 #define TSDB_MAX_VNODES 256
#define TSDB_MIN_VNODES 50 #define TSDB_MIN_VNODES 50
#define TSDB_INVALID_VNODE_NUM 0 #define TSDB_INVALID_VNODE_NUM 0
......
src/mnode/inc/mgmtDef.h
浏览文件 @ bd42e715
...@@ -68,7 +68,7 @@ typedef struct SMnodeObj { ...@@ -68,7 +68,7 @@ typedef struct SMnodeObj {
// todo use dynamic length string // todo use dynamic length string
typedef struct { typedef struct {
char tableId[TSDB_TABLE_ID_LEN + 1]; char *tableId;
int8_t type; int8_t type;
} STableObj; } STableObj;
......
src/mnode/inc/mgmtSdb.h
浏览文件 @ bd42e715
...@@ -35,7 +35,8 @@ typedef enum { ...@@ -35,7 +35,8 @@ typedef enum {
typedef enum { typedef enum {
SDB_KEY_STRING, SDB_KEY_STRING,
SDB_KEY_INT, SDB_KEY_INT,
SDB_KEY_AUTO SDB_KEY_AUTO,
SDB_KEY_VAR_STRING,
} ESdbKey; } ESdbKey;
typedef enum { typedef enum {
......
src/mnode/src/mgmtSdb.c
浏览文件 @ bd42e715
...@@ -104,6 +104,14 @@ bool sdbIsServing() { ...@@ -104,6 +104,14 @@ bool sdbIsServing() {
return tsSdbObj.status == SDB_STATUS_SERVING; return tsSdbObj.status == SDB_STATUS_SERVING;
} }
static void *sdbGetObjKey(SSdbTable *pTable, void *key) {
if (pTable->keyType == SDB_KEY_VAR_STRING) {
return *(char **)key;
}
return key;
}
static char *sdbGetActionStr(int32_t action) { static char *sdbGetActionStr(int32_t action) {
switch (action) { switch (action) {
case SDB_ACTION_INSERT: case SDB_ACTION_INSERT:
...@@ -116,20 +124,25 @@ static char *sdbGetActionStr(int32_t action) { ...@@ -116,20 +124,25 @@ static char *sdbGetActionStr(int32_t action) {
return "invalid"; return "invalid";
} }
static char *sdbGetkeyStr(SSdbTable *pTable, void *row) { static char *sdbGetKeyStr(SSdbTable *pTable, void *key) {
static char str[16]; static char str[16];
switch (pTable->keyType) { switch (pTable->keyType) {
case SDB_KEY_STRING: case SDB_KEY_STRING:
return (char *)row; case SDB_KEY_VAR_STRING:
return (char *)key;
case SDB_KEY_INT: case SDB_KEY_INT:
case SDB_KEY_AUTO: case SDB_KEY_AUTO:
sprintf(str, "%d", *(int32_t *)row); sprintf(str, "%d", *(int32_t *)key);
return str; return str;
default: default:
return "invalid"; return "invalid";
} }
} }
static char *sdbGetKeyStrFromObj(SSdbTable *pTable, void *key) {
return sdbGetKeyStr(pTable, sdbGetObjKey(pTable, key));
}
static void *sdbGetTableFromId(int32_t tableId) { static void *sdbGetTableFromId(int32_t tableId) {
return tsSdbObj.tableList[tableId]; return tsSdbObj.tableList[tableId];
} }
...@@ -332,50 +345,48 @@ void sdbCleanUp() { ...@@ -332,50 +345,48 @@ void sdbCleanUp() {
pthread_mutex_destroy(&tsSdbObj.mutex); pthread_mutex_destroy(&tsSdbObj.mutex);
} }
void sdbIncRef(void *handle, void *pRow) { void sdbIncRef(void *handle, void *pObj) {
if (pRow) { if (pObj == NULL) return;
SSdbTable *pTable = handle;
int32_t * pRefCount = (int32_t *)(pRow + pTable->refCountPos); SSdbTable *pTable = handle;
atomic_add_fetch_32(pRefCount, 1); int32_t * pRefCount = (int32_t *)(pObj + pTable->refCountPos);
if (0 && (pTable->tableId == SDB_TABLE_MNODE || pTable->tableId == SDB_TABLE_DNODE)) { atomic_add_fetch_32(pRefCount, 1);
sdbTrace("table:%s, add ref to record:%s:%s:%d", pTable->tableName, pTable->tableName, sdbGetkeyStr(pTable, pRow), if (0 && (pTable->tableId == SDB_TABLE_MNODE || pTable->tableId == SDB_TABLE_DNODE)) {
*pRefCount); sdbTrace("table:%s, add ref to record:%s:%d", pTable->tableName, sdbGetKeyStrFromObj(pTable, pObj), *pRefCount);
}
} }
} }
void sdbDecRef(void *handle, void *pRow) { void sdbDecRef(void *handle, void *pObj) {
if (pRow) { if (pObj == NULL) return;
SSdbTable *pTable = handle;
int32_t * pRefCount = (int32_t *)(pRow + pTable->refCountPos); SSdbTable *pTable = handle;
int32_t refCount = atomic_sub_fetch_32(pRefCount, 1); int32_t * pRefCount = (int32_t *)(pObj + pTable->refCountPos);
if (0 && (pTable->tableId == SDB_TABLE_MNODE || pTable->tableId == SDB_TABLE_DNODE)) { int32_t refCount = atomic_sub_fetch_32(pRefCount, 1);
sdbTrace("table:%s, def ref of record:%s:%s:%d", pTable->tableName, pTable->tableName, sdbGetkeyStr(pTable, pRow), if (0 && (pTable->tableId == SDB_TABLE_MNODE || pTable->tableId == SDB_TABLE_DNODE)) {
*pRefCount); sdbTrace("table:%s, def ref of record:%s:%d", pTable->tableName, sdbGetKeyStrFromObj(pTable, pObj), *pRefCount);
}
int8_t *updateEnd = pRow + pTable->refCountPos - 1;
if (refCount <= 0 && *updateEnd) {
sdbTrace("table:%s, record:%s:%s:%d is destroyed", pTable->tableName, pTable->tableName,
sdbGetkeyStr(pTable, pRow), *pRefCount);
SSdbOper oper = {.pObj = pRow};
(*pTable->destroyFp)(&oper);
}
} }
}
static SSdbRow *sdbGetRowMeta(void *handle, void *key) { int8_t *updateEnd = pObj + pTable->refCountPos - 1;
SSdbTable *pTable = (SSdbTable *)handle; if (refCount <= 0 && *updateEnd) {
SSdbRow * pMeta; sdbTrace("table:%s, record:%s:%d is destroyed", pTable->tableName, sdbGetKeyStrFromObj(pTable, pObj), *pRefCount);
SSdbOper oper = {.pObj = pObj};
(*pTable->destroyFp)(&oper);
}
}
if (handle == NULL) return NULL; static SSdbRow *sdbGetRowMeta(SSdbTable *pTable, void *key) {
if (pTable == NULL) return NULL;
int32_t keySize = sizeof(int32_t); int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) { if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
keySize = strlen((char *)key); keySize = strlen((char *)key);
} }
pMeta = taosHashGet(pTable->iHandle, key, keySize);
return taosHashGet(pTable->iHandle, key, keySize);
}
return pMeta; static SSdbRow *sdbGetRowMetaFromObj(SSdbTable *pTable, void *key) {
return sdbGetRowMeta(pTable, sdbGetObjKey(pTable, key));
} }
void *sdbGetRow(void *handle, void *key) { void *sdbGetRow(void *handle, void *key) {
...@@ -387,7 +398,7 @@ void *sdbGetRow(void *handle, void *key) { ...@@ -387,7 +398,7 @@ void *sdbGetRow(void *handle, void *key) {
pthread_mutex_lock(&pTable->mutex); pthread_mutex_lock(&pTable->mutex);
int32_t keySize = sizeof(int32_t); int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) { if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
keySize = strlen((char *)key); keySize = strlen((char *)key);
} }
pMeta = taosHashGet(pTable->iHandle, key, keySize); pMeta = taosHashGet(pTable->iHandle, key, keySize);
...@@ -400,6 +411,10 @@ void *sdbGetRow(void *handle, void *key) { ...@@ -400,6 +411,10 @@ void *sdbGetRow(void *handle, void *key) {
return pMeta->row; return pMeta->row;
} }
static void *sdbGetRowFromObj(SSdbTable *pTable, void *key) {
return sdbGetRow(pTable, sdbGetObjKey(pTable, key));
}
static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) { static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
SSdbRow rowMeta; SSdbRow rowMeta;
rowMeta.rowSize = pOper->rowSize; rowMeta.rowSize = pOper->rowSize;
...@@ -407,11 +422,14 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) { ...@@ -407,11 +422,14 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
pthread_mutex_lock(&pTable->mutex); pthread_mutex_lock(&pTable->mutex);
void * key = sdbGetObjKey(pTable, pOper->pObj);
int32_t keySize = sizeof(int32_t); int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) {
keySize = strlen((char *)pOper->pObj); if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
keySize = strlen((char *)key);
} }
taosHashPut(pTable->iHandle, pOper->pObj, keySize, &rowMeta, sizeof(SSdbRow));
taosHashPut(pTable->iHandle, key, keySize, &rowMeta, sizeof(SSdbRow));
sdbIncRef(pTable, pOper->pObj); sdbIncRef(pTable, pOper->pObj);
pTable->numOfRows++; pTable->numOfRows++;
...@@ -425,7 +443,7 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) { ...@@ -425,7 +443,7 @@ static int32_t sdbInsertHash(SSdbTable *pTable, SSdbOper *pOper) {
pthread_mutex_unlock(&pTable->mutex); pthread_mutex_unlock(&pTable->mutex);
sdbTrace("table:%s, insert record:%s to hash, numOfRows:%d version:%" PRIu64, pTable->tableName, sdbTrace("table:%s, insert record:%s to hash, numOfRows:%d version:%" PRIu64, pTable->tableName,
sdbGetkeyStr(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion()); sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
(*pTable->insertFp)(pOper); (*pTable->insertFp)(pOper);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
...@@ -436,17 +454,20 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) { ...@@ -436,17 +454,20 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
pthread_mutex_lock(&pTable->mutex); pthread_mutex_lock(&pTable->mutex);
void * key = sdbGetObjKey(pTable, pOper->pObj);
int32_t keySize = sizeof(int32_t); int32_t keySize = sizeof(int32_t);
if (pTable->keyType == SDB_KEY_STRING) {
keySize = strlen((char *)pOper->pObj); if (pTable->keyType == SDB_KEY_STRING || pTable->keyType == SDB_KEY_VAR_STRING) {
keySize = strlen((char *)key);
} }
taosHashRemove(pTable->iHandle, pOper->pObj, keySize);
taosHashRemove(pTable->iHandle, key, keySize);
pTable->numOfRows--; pTable->numOfRows--;
pthread_mutex_unlock(&pTable->mutex); pthread_mutex_unlock(&pTable->mutex);
sdbTrace("table:%s, delete record:%s from hash, numOfRows:%d version:%" PRIu64, pTable->tableName, sdbTrace("table:%s, delete record:%s from hash, numOfRows:%d version:%" PRIu64, pTable->tableName,
sdbGetkeyStr(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion()); sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
int8_t *updateEnd = pOper->pObj + pTable->refCountPos - 1; int8_t *updateEnd = pOper->pObj + pTable->refCountPos - 1;
*updateEnd = 1; *updateEnd = 1;
...@@ -457,7 +478,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) { ...@@ -457,7 +478,7 @@ static int32_t sdbDeleteHash(SSdbTable *pTable, SSdbOper *pOper) {
static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbOper *pOper) { static int32_t sdbUpdateHash(SSdbTable *pTable, SSdbOper *pOper) {
sdbTrace("table:%s, update record:%s in hash, numOfRows:%d version:%" PRIu64, pTable->tableName, sdbTrace("table:%s, update record:%s in hash, numOfRows:%d version:%" PRIu64, pTable->tableName,
sdbGetkeyStr(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion()); sdbGetKeyStrFromObj(pTable, pOper->pObj), pTable->numOfRows, sdbGetVersion());
(*pTable->updateFp)(pOper); (*pTable->updateFp)(pOper);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
...@@ -488,7 +509,7 @@ static int sdbWrite(void *param, void *data, int type) { ...@@ -488,7 +509,7 @@ static int sdbWrite(void *param, void *data, int type) {
} else if (pHead->version != tsSdbObj.version + 1) { } else if (pHead->version != tsSdbObj.version + 1) {
pthread_mutex_unlock(&tsSdbObj.mutex); pthread_mutex_unlock(&tsSdbObj.mutex);
sdbError("table:%s, failed to restore %s record:%s from wal, version:%" PRId64 " too large, sdb version:%" PRId64, sdbError("table:%s, failed to restore %s record:%s from wal, version:%" PRId64 " too large, sdb version:%" PRId64,
pTable->tableName, sdbGetActionStr(action), sdbGetkeyStr(pTable, pHead->cont), pHead->version, pTable->tableName, sdbGetActionStr(action), sdbGetKeyStr(pTable, pHead->cont), pHead->version,
tsSdbObj.version); tsSdbObj.version);
return TSDB_CODE_OTHERS; return TSDB_CODE_OTHERS;
} else { } else {
...@@ -540,8 +561,8 @@ int32_t sdbInsertRow(SSdbOper *pOper) { ...@@ -540,8 +561,8 @@ int32_t sdbInsertRow(SSdbOper *pOper) {
SSdbTable *pTable = (SSdbTable *)pOper->table; SSdbTable *pTable = (SSdbTable *)pOper->table;
if (pTable == NULL) return -1; if (pTable == NULL) return -1;
if (sdbGetRow(pTable, pOper->pObj)) { if (sdbGetRowFromObj(pTable, pOper->pObj)) {
sdbError("table:%s, failed to insert record:%s, already exist", pTable->tableName, sdbGetkeyStr(pTable, pOper->pObj)); sdbError("table:%s, failed to insert record:%s, already exist", pTable->tableName, sdbGetKeyStrFromObj(pTable, pOper->pObj));
sdbDecRef(pTable, pOper->pObj); sdbDecRef(pTable, pOper->pObj);
return TSDB_CODE_ALREADY_THERE; return TSDB_CODE_ALREADY_THERE;
} }
...@@ -580,7 +601,7 @@ int32_t sdbDeleteRow(SSdbOper *pOper) { ...@@ -580,7 +601,7 @@ int32_t sdbDeleteRow(SSdbOper *pOper) {
SSdbTable *pTable = (SSdbTable *)pOper->table; SSdbTable *pTable = (SSdbTable *)pOper->table;
if (pTable == NULL) return -1; if (pTable == NULL) return -1;
SSdbRow *pMeta = sdbGetRowMeta(pTable, pOper->pObj); SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
if (pMeta == NULL) { if (pMeta == NULL) {
sdbTrace("table:%s, record is not there, delete failed", pTable->tableName); sdbTrace("table:%s, record is not there, delete failed", pTable->tableName);
return -1; return -1;
...@@ -590,25 +611,27 @@ int32_t sdbDeleteRow(SSdbOper *pOper) { ...@@ -590,25 +611,27 @@ int32_t sdbDeleteRow(SSdbOper *pOper) {
assert(pMetaRow != NULL); assert(pMetaRow != NULL);
if (pOper->type == SDB_OPER_GLOBAL) { if (pOper->type == SDB_OPER_GLOBAL) {
int32_t rowSize = 0; void * key = sdbGetObjKey(pTable, pOper->pObj);
int32_t keySize = 0;
switch (pTable->keyType) { switch (pTable->keyType) {
case SDB_KEY_STRING: case SDB_KEY_STRING:
rowSize = strlen((char *)pOper->pObj) + 1; case SDB_KEY_VAR_STRING:
keySize = strlen((char *)key) + 1;
break; break;
case SDB_KEY_INT: case SDB_KEY_INT:
case SDB_KEY_AUTO: case SDB_KEY_AUTO:
rowSize = sizeof(uint64_t); keySize = sizeof(uint32_t);
break; break;
default: default:
return -1; return -1;
} }
int32_t size = sizeof(SWalHead) + rowSize; int32_t size = sizeof(SWalHead) + keySize;
SWalHead *pHead = taosAllocateQitem(size); SWalHead *pHead = taosAllocateQitem(size);
pHead->version = 0; pHead->version = 0;
pHead->len = rowSize; pHead->len = keySize;
pHead->msgType = pTable->tableId * 10 + SDB_ACTION_DELETE; pHead->msgType = pTable->tableId * 10 + SDB_ACTION_DELETE;
memcpy(pHead->cont, pOper->pObj, rowSize); memcpy(pHead->cont, key, keySize);
int32_t code = sdbWrite(pOper, pHead, pHead->msgType); int32_t code = sdbWrite(pOper, pHead, pHead->msgType);
taosFreeQitem(pHead); taosFreeQitem(pHead);
...@@ -622,7 +645,7 @@ int32_t sdbUpdateRow(SSdbOper *pOper) { ...@@ -622,7 +645,7 @@ int32_t sdbUpdateRow(SSdbOper *pOper) {
SSdbTable *pTable = (SSdbTable *)pOper->table; SSdbTable *pTable = (SSdbTable *)pOper->table;
if (pTable == NULL) return -1; if (pTable == NULL) return -1;
SSdbRow *pMeta = sdbGetRowMeta(pTable, pOper->pObj); SSdbRow *pMeta = sdbGetRowMetaFromObj(pTable, pOper->pObj);
if (pMeta == NULL) { if (pMeta == NULL) {
sdbTrace("table:%s, record is not there, delete failed", pTable->tableName); sdbTrace("table:%s, record is not there, delete failed", pTable->tableName);
return -1; return -1;
......
src/mnode/src/mgmtTable.c
浏览文件 @ bd42e715
...@@ -84,6 +84,7 @@ static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg); ...@@ -84,6 +84,7 @@ static void mgmtProcessAlterTableRsp(SRpcMsg *rpcMsg);
static int32_t mgmtFindSuperTableColumnIndex(SSuperTableObj *pStable, char *colName); static int32_t mgmtFindSuperTableColumnIndex(SSuperTableObj *pStable, char *colName);
static void mgmtDestroyChildTable(SChildTableObj *pTable) { static void mgmtDestroyChildTable(SChildTableObj *pTable) {
tfree(pTable->info.tableId);
tfree(pTable->schema); tfree(pTable->schema);
tfree(pTable->sql); tfree(pTable->sql);
tfree(pTable); tfree(pTable);
...@@ -180,6 +181,7 @@ static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) { ...@@ -180,6 +181,7 @@ static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) {
SChildTableObj *pNew = pOper->pObj; SChildTableObj *pNew = pOper->pObj;
SChildTableObj *pTable = mgmtGetChildTable(pNew->info.tableId); SChildTableObj *pTable = mgmtGetChildTable(pNew->info.tableId);
if (pTable != pNew) { if (pTable != pNew) {
void *oldTableId = pTable->info.tableId;
void *oldSql = pTable->sql; void *oldSql = pTable->sql;
void *oldSchema = pTable->schema; void *oldSchema = pTable->schema;
memcpy(pTable, pNew, pOper->rowSize); memcpy(pTable, pNew, pOper->rowSize);
...@@ -188,6 +190,7 @@ static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) { ...@@ -188,6 +190,7 @@ static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) {
free(pNew); free(pNew);
free(oldSql); free(oldSql);
free(oldSchema); free(oldSchema);
free(oldTableId);
} }
mgmtDecTableRef(pTable); mgmtDecTableRef(pTable);
...@@ -195,51 +198,66 @@ static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) { ...@@ -195,51 +198,66 @@ static int32_t mgmtChildTableActionUpdate(SSdbOper *pOper) {
} }
static int32_t mgmtChildTableActionEncode(SSdbOper *pOper) { static int32_t mgmtChildTableActionEncode(SSdbOper *pOper) {
const int32_t maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16);
SChildTableObj *pTable = pOper->pObj; SChildTableObj *pTable = pOper->pObj;
assert(pTable != NULL && pOper->rowData != NULL); assert(pTable != NULL && pOper->rowData != NULL);
if (pTable->info.type == TSDB_CHILD_TABLE) { int32_t len = strlen(pTable->info.tableId);
memcpy(pOper->rowData, pTable, tsChildTableUpdateSize); if (len > TSDB_TABLE_ID_LEN) return TSDB_CODE_INVALID_TABLE_ID;
pOper->rowSize = tsChildTableUpdateSize;
} else { memcpy(pOper->rowData, pTable->info.tableId, len);
memset(pOper->rowData + len, 0, 1);
len++;
memcpy(pOper->rowData + len, (char*)pTable + sizeof(char *), tsChildTableUpdateSize);
len += tsChildTableUpdateSize;
if (pTable->info.type != TSDB_CHILD_TABLE) {
int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema); int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
if (maxRowSize < tsChildTableUpdateSize + schemaSize) { memcpy(pOper->rowData + len, pTable->schema, schemaSize);
return TSDB_CODE_INVALID_MSG_LEN; len += schemaSize;
if (pTable->sqlLen != 0) {
memcpy(pOper->rowData + len, pTable->sql, pTable->sqlLen);
len += pTable->sqlLen;
} }
memcpy(pOper->rowData, pTable, tsChildTableUpdateSize);
memcpy(pOper->rowData + tsChildTableUpdateSize, pTable->schema, schemaSize);
memcpy(pOper->rowData + tsChildTableUpdateSize + schemaSize, pTable->sql, pTable->sqlLen);
pOper->rowSize = tsChildTableUpdateSize + schemaSize + pTable->sqlLen;
} }
pOper->rowSize = len;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mgmtChildTableActionDecode(SSdbOper *pOper) { static int32_t mgmtChildTableActionDecode(SSdbOper *pOper) {
assert(pOper->rowData != NULL); assert(pOper->rowData != NULL);
SChildTableObj *pTable = calloc(1, sizeof(SChildTableObj)); SChildTableObj *pTable = calloc(1, sizeof(SChildTableObj));
if (pTable == NULL) { if (pTable == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
memcpy(pTable, pOper->rowData, tsChildTableUpdateSize); int32_t len = strlen(pOper->rowData);
if (len > TSDB_TABLE_ID_LEN) return TSDB_CODE_INVALID_TABLE_ID;
pTable->info.tableId = strdup(pOper->rowData);
len++;
memcpy((char*)pTable + sizeof(char *), pOper->rowData + len, tsChildTableUpdateSize);
len += tsChildTableUpdateSize;
if (pTable->info.type != TSDB_CHILD_TABLE) { if (pTable->info.type != TSDB_CHILD_TABLE) {
int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema); int32_t schemaSize = pTable->numOfColumns * sizeof(SSchema);
pTable->schema = (SSchema *)malloc(schemaSize); pTable->schema = (SSchema *)malloc(schemaSize);
if (pTable->schema == NULL) { if (pTable->schema == NULL) {
mgmtDestroyChildTable(pTable); mgmtDestroyChildTable(pTable);
return TSDB_CODE_SERV_OUT_OF_MEMORY; return TSDB_CODE_INVALID_TABLE_TYPE;
} }
memcpy(pTable->schema, pOper->rowData + tsChildTableUpdateSize, schemaSize); memcpy(pTable->schema, pOper->rowData + len, schemaSize);
len += schemaSize;
pTable->sql = (char *)malloc(pTable->sqlLen); if (pTable->sqlLen != 0) {
if (pTable->sql == NULL) { pTable->sql = malloc(pTable->sqlLen);
mgmtDestroyChildTable(pTable); if (pTable->sql == NULL) {
return TSDB_CODE_SERV_OUT_OF_MEMORY; mgmtDestroyChildTable(pTable);
return TSDB_CODE_SERV_OUT_OF_MEMORY;
}
memcpy(pTable->sql, pOper->rowData + len, pTable->sqlLen);
} }
memcpy(pTable->sql, pOper->rowData + tsChildTableUpdateSize + schemaSize, pTable->sqlLen);
} }
pOper->pObj = pTable; pOper->pObj = pTable;
...@@ -311,15 +329,15 @@ static int32_t mgmtChildTableActionRestored() { ...@@ -311,15 +329,15 @@ static int32_t mgmtChildTableActionRestored() {
static int32_t mgmtInitChildTables() { static int32_t mgmtInitChildTables() {
SChildTableObj tObj; SChildTableObj tObj;
tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsChildTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type;
SSdbTableDesc tableDesc = { SSdbTableDesc tableDesc = {
.tableId = SDB_TABLE_CTABLE, .tableId = SDB_TABLE_CTABLE,
.tableName = "ctables", .tableName = "ctables",
.hashSessions = tsMaxTables, .hashSessions = tsMaxTables,
.maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16), .maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16) + TSDB_TABLE_ID_LEN + TSDB_CQ_SQL_SIZE,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
.keyType = SDB_KEY_STRING, .keyType = SDB_KEY_VAR_STRING,
.insertFp = mgmtChildTableActionInsert, .insertFp = mgmtChildTableActionInsert,
.deleteFp = mgmtChildTableActionDelete, .deleteFp = mgmtChildTableActionDelete,
.updateFp = mgmtChildTableActionUpdate, .updateFp = mgmtChildTableActionUpdate,
...@@ -372,6 +390,7 @@ static void mgmtDestroySuperTable(SSuperTableObj *pStable) { ...@@ -372,6 +390,7 @@ static void mgmtDestroySuperTable(SSuperTableObj *pStable) {
taosHashCleanup(pStable->vgHash); taosHashCleanup(pStable->vgHash);
pStable->vgHash = NULL; pStable->vgHash = NULL;
} }
tfree(pStable->info.tableId);
tfree(pStable->schema); tfree(pStable->schema);
tfree(pStable); tfree(pStable);
} }
...@@ -408,11 +427,13 @@ static int32_t mgmtSuperTableActionUpdate(SSdbOper *pOper) { ...@@ -408,11 +427,13 @@ static int32_t mgmtSuperTableActionUpdate(SSdbOper *pOper) {
SSuperTableObj *pNew = pOper->pObj; SSuperTableObj *pNew = pOper->pObj;
SSuperTableObj *pTable = mgmtGetSuperTable(pNew->info.tableId); SSuperTableObj *pTable = mgmtGetSuperTable(pNew->info.tableId);
if (pTable != pNew) { if (pTable != pNew) {
void *oldTableId = pTable->info.tableId;
void *oldSchema = pTable->schema; void *oldSchema = pTable->schema;
memcpy(pTable, pNew, pOper->rowSize); memcpy(pTable, pNew, pOper->rowSize);
pTable->schema = pNew->schema; pTable->schema = pNew->schema;
free(pNew->vgHash); free(pNew->vgHash);
free(pNew); free(pNew);
free(oldTableId);
free(oldSchema); free(oldSchema);
} }
mgmtDecTableRef(pTable); mgmtDecTableRef(pTable);
...@@ -420,40 +441,50 @@ static int32_t mgmtSuperTableActionUpdate(SSdbOper *pOper) { ...@@ -420,40 +441,50 @@ static int32_t mgmtSuperTableActionUpdate(SSdbOper *pOper) {
} }
static int32_t mgmtSuperTableActionEncode(SSdbOper *pOper) { static int32_t mgmtSuperTableActionEncode(SSdbOper *pOper) {
const int32_t maxRowSize = sizeof(SChildTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16);
SSuperTableObj *pStable = pOper->pObj; SSuperTableObj *pStable = pOper->pObj;
assert(pOper->pObj != NULL && pOper->rowData != NULL); assert(pOper->pObj != NULL && pOper->rowData != NULL);
int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags); int32_t len = strlen(pStable->info.tableId);
if (len > TSDB_TABLE_ID_LEN) len = TSDB_CODE_INVALID_TABLE_ID;
if (maxRowSize < tsSuperTableUpdateSize + schemaSize) { memcpy(pOper->rowData, pStable->info.tableId, len);
return TSDB_CODE_INVALID_MSG_LEN; memset(pOper->rowData + len, 0, 1);
} len++;
memcpy(pOper->rowData + len, (char*)pStable + sizeof(char *), tsSuperTableUpdateSize);
len += tsSuperTableUpdateSize;
memcpy(pOper->rowData, pStable, tsSuperTableUpdateSize); int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
memcpy(pOper->rowData + tsSuperTableUpdateSize, pStable->schema, schemaSize); memcpy(pOper->rowData + len, pStable->schema, schemaSize);
pOper->rowSize = tsSuperTableUpdateSize + schemaSize; len += schemaSize;
pOper->rowSize = len;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
static int32_t mgmtSuperTableActionDecode(SSdbOper *pOper) { static int32_t mgmtSuperTableActionDecode(SSdbOper *pOper) {
assert(pOper->rowData != NULL); assert(pOper->rowData != NULL);
SSuperTableObj *pStable = (SSuperTableObj *) calloc(1, sizeof(SSuperTableObj)); SSuperTableObj *pStable = (SSuperTableObj *) calloc(1, sizeof(SSuperTableObj));
if (pStable == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY; if (pStable == NULL) return TSDB_CODE_SERV_OUT_OF_MEMORY;
memcpy(pStable, pOper->rowData, tsSuperTableUpdateSize); int32_t len = strlen(pOper->rowData);
if (len > TSDB_TABLE_ID_LEN) return TSDB_CODE_INVALID_TABLE_ID;
pStable->info.tableId = strdup(pOper->rowData);
len++;
memcpy((char*)pStable + sizeof(char *), pOper->rowData + len, tsSuperTableUpdateSize);
len += tsSuperTableUpdateSize;
int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags); int32_t schemaSize = sizeof(SSchema) * (pStable->numOfColumns + pStable->numOfTags);
pStable->schema = malloc(schemaSize); pStable->schema = malloc(schemaSize);
if (pStable->schema == NULL) { if (pStable->schema == NULL) {
mgmtDestroySuperTable(pStable); mgmtDestroySuperTable(pStable);
return -1; return TSDB_CODE_NOT_SUPER_TABLE;
} }
memcpy(pStable->schema, pOper->rowData + tsSuperTableUpdateSize, schemaSize); memcpy(pStable->schema, pOper->rowData + len, schemaSize);
pOper->pObj = pStable; pOper->pObj = pStable;
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
...@@ -465,15 +496,15 @@ static int32_t mgmtSuperTableActionRestored() { ...@@ -465,15 +496,15 @@ static int32_t mgmtSuperTableActionRestored() {
static int32_t mgmtInitSuperTables() { static int32_t mgmtInitSuperTables() {
SSuperTableObj tObj; SSuperTableObj tObj;
tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj; tsSuperTableUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj.info.type;
SSdbTableDesc tableDesc = { SSdbTableDesc tableDesc = {
.tableId = SDB_TABLE_STABLE, .tableId = SDB_TABLE_STABLE,
.tableName = "stables", .tableName = "stables",
.hashSessions = TSDB_MAX_SUPER_TABLES, .hashSessions = TSDB_MAX_SUPER_TABLES,
.maxRowSize = tsSuperTableUpdateSize + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16), .maxRowSize = sizeof(SSuperTableObj) + sizeof(SSchema) * (TSDB_MAX_TAGS + TSDB_MAX_COLUMNS + 16) + TSDB_TABLE_ID_LEN,
.refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj, .refCountPos = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
.keyType = SDB_KEY_STRING, .keyType = SDB_KEY_VAR_STRING,
.insertFp = mgmtSuperTableActionInsert, .insertFp = mgmtSuperTableActionInsert,
.deleteFp = mgmtSuperTableActionDelete, .deleteFp = mgmtSuperTableActionDelete,
.updateFp = mgmtSuperTableActionUpdate, .updateFp = mgmtSuperTableActionUpdate,
...@@ -720,14 +751,14 @@ static void mgmtProcessTableMetaMsg(SQueuedMsg *pMsg) { ...@@ -720,14 +751,14 @@ static void mgmtProcessTableMetaMsg(SQueuedMsg *pMsg) {
static void mgmtProcessCreateSuperTableMsg(SQueuedMsg *pMsg) { static void mgmtProcessCreateSuperTableMsg(SQueuedMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->pCont; SCMCreateTableMsg *pCreate = pMsg->pCont;
SSuperTableObj *pStable = (SSuperTableObj *)calloc(1, sizeof(SSuperTableObj)); SSuperTableObj *pStable = calloc(1, sizeof(SSuperTableObj));
if (pStable == NULL) { if (pStable == NULL) {
mError("table:%s, failed to create, no enough memory", pCreate->tableId); mError("table:%s, failed to create, no enough memory", pCreate->tableId);
mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY); mgmtSendSimpleResp(pMsg->thandle, TSDB_CODE_SERV_OUT_OF_MEMORY);
return; return;
} }
strcpy(pStable->info.tableId, pCreate->tableId); pStable->info.tableId = strdup(pCreate->tableId);
pStable->info.type = TSDB_SUPER_TABLE; pStable->info.type = TSDB_SUPER_TABLE;
pStable->createdTime = taosGetTimestampMs(); pStable->createdTime = taosGetTimestampMs();
pStable->uid = (((uint64_t) pStable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul)); pStable->uid = (((uint64_t) pStable->createdTime) << 16) + (sdbGetVersion() & ((1ul << 16) - 1ul));
...@@ -1358,7 +1389,7 @@ static void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableOb ...@@ -1358,7 +1389,7 @@ static void *mgmtBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SChildTableOb
} }
static SChildTableObj* mgmtDoCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) { static SChildTableObj* mgmtDoCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj *pVgroup, int32_t tid) {
SChildTableObj *pTable = (SChildTableObj *) calloc(1, sizeof(SChildTableObj)); SChildTableObj *pTable = calloc(1, sizeof(SChildTableObj));
if (pTable == NULL) { if (pTable == NULL) {
mError("table:%s, failed to alloc memory", pCreate->tableId); mError("table:%s, failed to alloc memory", pCreate->tableId);
terrno = TSDB_CODE_SERV_OUT_OF_MEMORY; terrno = TSDB_CODE_SERV_OUT_OF_MEMORY;
...@@ -1371,10 +1402,10 @@ static SChildTableObj* mgmtDoCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj ...@@ -1371,10 +1402,10 @@ static SChildTableObj* mgmtDoCreateChildTable(SCMCreateTableMsg *pCreate, SVgObj
pTable->info.type = TSDB_NORMAL_TABLE; pTable->info.type = TSDB_NORMAL_TABLE;
} }
strcpy(pTable->info.tableId, pCreate->tableId); pTable->info.tableId = strdup(pCreate->tableId);
pTable->createdTime = taosGetTimestampMs(); pTable->createdTime = taosGetTimestampMs();
pTable->sid = tid; pTable->sid = tid;
pTable->vgId = pVgroup->vgId; pTable->vgId = pVgroup->vgId;
if (pTable->info.type == TSDB_CHILD_TABLE) { if (pTable->info.type == TSDB_CHILD_TABLE) {
char *pTagData = (char *) pCreate->schema; // it is a tag key char *pTagData = (char *) pCreate->schema; // it is a tag key
......
src/tsdb/src/tsdbMeta.c
浏览文件 @ bd42e715
...@@ -80,7 +80,7 @@ STable *tsdbDecodeTable(void *cont, int contLen) { ...@@ -80,7 +80,7 @@ STable *tsdbDecodeTable(void *cont, int contLen) {
T_READ_MEMBER(ptr, int8_t, pTable->type); T_READ_MEMBER(ptr, int8_t, pTable->type);
int len = *(int *)ptr; int len = *(int *)ptr;
ptr = (char *)ptr + sizeof(int); ptr = (char *)ptr + sizeof(int);
pTable->name = calloc(1, len + VARSTR_HEADER_SIZE); pTable->name = calloc(1, len + VARSTR_HEADER_SIZE + 1);
if (pTable->name == NULL) return NULL; if (pTable->name == NULL) return NULL;
varDataSetLen(pTable->name, len); varDataSetLen(pTable->name, len);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或