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

[TD-15] fix bug in sdbInsertRow

This commit is contained in:
slguan 2020-03-20 23:18:00 +08:00
parent eb6764be24
commit d0b0ac956e
2 changed files with 89 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
src/mnode/src/mgmtSdb.c
View File

@ -222,7 +222,6 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
int32_t real_size = 0; int32_t real_size = 0;
int32_t maxAutoIndex = 0; int32_t maxAutoIndex = 0;
oldId = pTable->id;
if (sdbOpenSdbFile(pTable) < 0) return -1; if (sdbOpenSdbFile(pTable) < 0) return -1;
total_size = sizeof(SRowHead) + pTable->maxRowSize + sizeof(TSCKSUM); total_size = sizeof(SRowHead) + pTable->maxRowSize + sizeof(TSCKSUM);
@ -234,7 +233,6 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
sdbTrace("open sdb file:%s for read", pTable->fn); sdbTrace("open sdb file:%s for read", pTable->fn);
// Loop to read sdb file row by row
while (1) { while (1) {
memset(rowHead, 0, total_size); memset(rowHead, 0, total_size);
@ -259,9 +257,6 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
continue; continue;
} }
// sdbTrace("%s id:%ld rowSize:%d", pTable->name, rowHead->id,
// rowHead->rowSize);
bytes = read(pTable->fd, rowHead->data, rowHead->rowSize + sizeof(TSCKSUM)); bytes = read(pTable->fd, rowHead->data, rowHead->rowSize + sizeof(TSCKSUM));
if (bytes < rowHead->rowSize + sizeof(TSCKSUM)) { if (bytes < rowHead->rowSize + sizeof(TSCKSUM)) {
// TODO: Here may cause pTable->size not end of the file // TODO: Here may cause pTable->size not end of the file
@ -276,40 +271,22 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
continue; continue;
} }
// Check if the the object exists already if (pTable->keyType == SDB_KEYTYPE_AUTO) {
maxAutoIndex = MAX(maxAutoIndex, *(int32_t *) rowHead->data);
}
pMetaRow = sdbGetRow(pTable, rowHead->data); pMetaRow = sdbGetRow(pTable, rowHead->data);
if (pMetaRow == NULL) { // New object if (pMetaRow == NULL) {
if (rowHead->id < 0) { if (rowHead->id < 0) {
/* assert(0); */
sdbError("error sdb negative id:%d, sdb:%s, skip", rowHead->id, pTable->name); sdbError("error sdb negative id:%d, sdb:%s, skip", rowHead->id, pTable->name);
} else { } else {
rowMeta.id = rowHead->id; sdbInsertRow(pTable, rowHead->data, SDB_OPER_DISK);
// TODO: Get rid of the rowMeta.offset and rowSize
rowMeta.offset = pTable->size;
rowMeta.rowSize = rowHead->rowSize;
rowMeta.row = (*pTable->decodeFp)(rowHead->data);
(*sdbAddIndexFp[pTable->keyType])(pTable->iHandle, rowMeta.row, &rowMeta);
if (pTable->keyType == SDB_KEYTYPE_AUTO) {
pTable->autoIndex++;
maxAutoIndex = MAX(maxAutoIndex, *(int32_t*)rowHead->data);
} }
pTable->numOfRows++; } else {
} if (rowHead->id < 0) {
} else { // already exists sdbDeleteRow(pTable, rowHead->data, SDB_OPER_DISK);
if (pTable->keyType == SDB_KEYTYPE_AUTO) { } else {
pTable->autoIndex++; sdbUpdateRow(pTable, rowHead->data, rowHead->rowSize, SDB_OPER_DISK);
maxAutoIndex = MAX(maxAutoIndex, *(int32_t *) rowHead->data);
}
if (rowHead->id < 0) { // Delete the object
(*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, rowHead->data);
(*pTable->destroyFp)(pMetaRow);
pTable->numOfRows--;
numOfDels++;
} else { // Reset the object TODO: is it possible to merge reset and
// update ??
//(*(pTable->appTool))(SDB_TYPE_RESET, pMetaRow, rowHead->data, rowHead->rowSize, NULL);
} }
numOfDels++; numOfDels++;
} }
@ -317,7 +294,6 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
pTable->size += real_size; pTable->size += real_size;
if (pTable->id < abs(rowHead->id)) pTable->id = abs(rowHead->id); if (pTable->id < abs(rowHead->id)) pTable->id = abs(rowHead->id);
//TODO: check this valid
pTable->size += 4; pTable->size += 4;
lseek(pTable->fd, 4, SEEK_CUR); lseek(pTable->fd, 4, SEEK_CUR);
} }
@ -326,8 +302,10 @@ static int32_t sdbInitTableByFile(SSdbTable *pTable) {
pTable->autoIndex = maxAutoIndex; pTable->autoIndex = maxAutoIndex;
} }
sdbVersion += (pTable->id - oldId); sdbVersion += pTable->id;
if (numOfDels > pTable->hashSessions / 4) sdbSaveSnapShot(pTable); if (numOfDels > pTable->hashSessions / 4) {
sdbSaveSnapShot(pTable);
}
tfree(rowHead); tfree(rowHead);
return 0; return 0;
@ -390,12 +368,13 @@ void *sdbGetRow(void *handle, void *key) {
pMeta = (*sdbGetIndexFp[pTable->keyType])(pTable->iHandle, key); pMeta = (*sdbGetIndexFp[pTable->keyType])(pTable->iHandle, key);
pthread_mutex_unlock(&pTable->mutex); pthread_mutex_unlock(&pTable->mutex);
if (pMeta == NULL) return NULL; if (pMeta == NULL) {
return NULL;
}
return pMeta->row; return pMeta->row;
} }
// row here must be encoded string (rowSize > 0) or the object it self (rowSize = 0)
int32_t sdbInsertRow(void *handle, void *row, ESdbOperType oper) { int32_t sdbInsertRow(void *handle, void *row, ESdbOperType oper) {
SSdbTable *pTable = (SSdbTable *)handle; SSdbTable *pTable = (SSdbTable *)handle;
SRowMeta rowMeta; SRowMeta rowMeta;
@ -448,7 +427,7 @@ int32_t sdbInsertRow(void *handle, void *row, ESdbOperType oper) {
} }
} }
if (oper == SDB_OPER_GLOBAL || oper == SDB_OPER_LOCAL) { if (oper != SDB_OPER_DISK) {
rowHead->rowSize = (*pTable->encodeFp)(pObj, rowHead->data, pTable->maxRowSize); rowHead->rowSize = (*pTable->encodeFp)(pObj, rowHead->data, pTable->maxRowSize);
assert(rowHead->rowSize > 0 && rowHead->rowSize <= pTable->maxRowSize); assert(rowHead->rowSize > 0 && rowHead->rowSize <= pTable->maxRowSize);
@ -462,6 +441,11 @@ int32_t sdbInsertRow(void *handle, void *row, ESdbOperType oper) {
return -1; return -1;
} }
twrite(pTable->fd, rowHead, real_size);
pTable->size += real_size;
sdbFinishCommit(pTable);
}
// update in SDB layer // update in SDB layer
rowMeta.id = pTable->id; rowMeta.id = pTable->id;
rowMeta.offset = pTable->size; rowMeta.offset = pTable->size;
@ -469,9 +453,18 @@ int32_t sdbInsertRow(void *handle, void *row, ESdbOperType oper) {
rowMeta.row = pObj; rowMeta.row = pObj;
(*sdbAddIndexFp[pTable->keyType])(pTable->iHandle, pObj, &rowMeta); (*sdbAddIndexFp[pTable->keyType])(pTable->iHandle, pObj, &rowMeta);
twrite(pTable->fd, rowHead, real_size); if (pTable->keyType == SDB_KEYTYPE_AUTO) {
pTable->size += real_size; *((uint32_t *)pObj) = ++pTable->autoIndex;
sdbFinishCommit(pTable); }
pTable->numOfRows++;
if (oper != SDB_OPER_DISK) {
pTable->id++;
sdbVersion++;
}
pthread_mutex_unlock(&pTable->mutex);
switch (pTable->keyType) { switch (pTable->keyType) {
case SDB_KEYTYPE_STRING: case SDB_KEYTYPE_STRING:
@ -487,17 +480,6 @@ int32_t sdbInsertRow(void *handle, void *row, ESdbOperType oper) {
pTable->name, sdbVersion, rowHead->id, rowHead->rowSize, pTable->numOfRows, pTable->size); pTable->name, sdbVersion, rowHead->id, rowHead->rowSize, pTable->numOfRows, pTable->size);
break; break;
} }
}
if (pTable->keyType == SDB_KEYTYPE_AUTO) {
*((uint32_t *)pObj) = ++pTable->autoIndex;
}
pTable->numOfRows++;
pTable->id++;
sdbVersion++;
pthread_mutex_unlock(&pTable->mutex);
(*pTable->insertFp)(pObj); (*pTable->insertFp)(pObj);
@ -556,6 +538,7 @@ int32_t sdbDeleteRow(void *handle, void *row, ESdbOperType oper) {
} }
} }
if (oper != SDB_OPER_DISK) {
rowHead->delimiter = SDB_DELIMITER; rowHead->delimiter = SDB_DELIMITER;
rowHead->rowSize = rowSize; rowHead->rowSize = rowSize;
rowHead->id = -(pTable->id); rowHead->id = -(pTable->id);
@ -570,6 +553,7 @@ int32_t sdbDeleteRow(void *handle, void *row, ESdbOperType oper) {
twrite(pTable->fd, rowHead, total_size); twrite(pTable->fd, rowHead, total_size);
pTable->size += total_size; pTable->size += total_size;
sdbFinishCommit(pTable); sdbFinishCommit(pTable);
}
switch (pTable->keyType) { switch (pTable->keyType) {
case SDB_KEYTYPE_STRING: case SDB_KEYTYPE_STRING:
@ -590,14 +574,18 @@ int32_t sdbDeleteRow(void *handle, void *row, ESdbOperType oper) {
(*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, row); (*sdbDeleteIndexFp[pTable->keyType])(pTable->iHandle, row);
pTable->numOfRows--; pTable->numOfRows--;
if (oper != SDB_OPER_DISK) {
pTable->id++; pTable->id++;
sdbVersion++; sdbVersion++;
}
pthread_mutex_unlock(&pTable->mutex); pthread_mutex_unlock(&pTable->mutex);
tfree(rowHead); tfree(rowHead);
(*pTable->deleteFp)(pMetaRow); (*pTable->deleteFp)(pMetaRow);
(*pTable->destroyFp)(pMetaRow);
return 0; return 0;
} }
@ -661,6 +649,7 @@ int32_t sdbUpdateRow(void *handle, void *row, int32_t updateSize, ESdbOperType o
real_size = sizeof(SRowHead) + rowHead->rowSize + sizeof(TSCKSUM); real_size = sizeof(SRowHead) + rowHead->rowSize + sizeof(TSCKSUM);
// write to the new position // write to the new position
if (oper != SDB_OPER_DISK) {
rowHead->delimiter = SDB_DELIMITER; rowHead->delimiter = SDB_DELIMITER;
rowHead->id = pTable->id; rowHead->id = pTable->id;
if (taosCalcChecksumAppend(0, (uint8_t *)rowHead, real_size) < 0) { if (taosCalcChecksumAppend(0, (uint8_t *)rowHead, real_size) < 0) {
@ -678,6 +667,7 @@ int32_t sdbUpdateRow(void *handle, void *row, int32_t updateSize, ESdbOperType o
pTable->size += real_size; pTable->size += real_size;
sdbFinishCommit(pTable); sdbFinishCommit(pTable);
}
switch (pTable->keyType) { switch (pTable->keyType) {
case SDB_KEYTYPE_STRING: case SDB_KEYTYPE_STRING:
@ -694,8 +684,10 @@ int32_t sdbUpdateRow(void *handle, void *row, int32_t updateSize, ESdbOperType o
break; break;
} }
if (oper != SDB_OPER_DISK) {
pTable->id++; pTable->id++;
sdbVersion++; sdbVersion++;
}
pthread_mutex_unlock(&pTable->mutex); pthread_mutex_unlock(&pTable->mutex);

3
src/mnode/src/mgmtUser.c
View File

@ -176,8 +176,7 @@ static int32_t mgmtCreateUser(SAcctObj *pAcct, char *name, char *pass) {
return code; return code;
} }
pUser = malloc(sizeof(SUserObj)); pUser = calloc(1, sizeof(SUserObj));
memset(pUser, 0, sizeof(SUserObj));
strcpy(pUser->user, name); strcpy(pUser->user, name);
taosEncryptPass((uint8_t*) pass, strlen(pass), pUser->pass); taosEncryptPass((uint8_t*) pass, strlen(pass), pUser->pass);
strcpy(pUser->acct, pAcct->user); strcpy(pUser->acct, pAcct->user);