database.cpp

实现内存数据库的源代码

		delete[] fileName;
	    } else { 
		char* newFileName = new char[strlen(backupFileName) + 5];
		sprintf(newFileName,"%s.new", backupFileName);
		backup(newFileName, false);
		::remove(backupFileName);
		::rename(newFileName, backupFileName);
		delete[] newFileName;
	    }
	} else { 
	    return;
	}
    }
}    


void dbDatabase::recovery()
{
    int curr = header->curr;
    header->size = file.getSize();
    header->root[1-curr].indexUsed = header->root[curr].indexUsed; 
    header->root[1-curr].freeList = header->root[curr].freeList; 
    header->root[1-curr].index = header->root[curr].shadowIndex;
    header->root[1-curr].indexSize = 
	header->root[curr].shadowIndexSize;
    header->root[1-curr].shadowIndex = header->root[curr].index;
    header->root[1-curr].shadowIndexSize = 
	header->root[curr].indexSize;
    
    offs_t* dst = (offs_t*)(baseAddr+header->root[1-curr].index);
    offs_t* src = (offs_t*)(baseAddr+header->root[curr].index);
    currIndex = dst;
    for (int n = header->root[curr].indexUsed; --n >= 0;) { 
	if (*dst != *src) { 
	    *dst = *src;
	}
	dst += 1;
	src += 1;
    }
    //
    // Restore consistency of table rows l2-list 
    //
    restoreTablesConsistency();
    memset(monitor->dirtyPagesMap, 0, dbDirtyPageBitmapSize);
}

void dbDatabase::restoreTablesConsistency()
{
    dbTable* table = (dbTable*)getRow(dbMetaTableId);
    oid_t lastId = table->lastRow;
    if (lastId != 0) { 
	dbRecord* record = getRow(lastId);
	if (record->next != 0) { 
	    record->next = 0;
	    file.markAsDirty(currIndex[lastId], sizeof(dbTable));
	}
    }
    oid_t tableId = table->firstRow;
    while (tableId != 0) { 
	table = (dbTable*)getRow(tableId);
	lastId = table->lastRow;
	if (lastId != 0) { 
	    dbRecord* record = getRow(lastId);
	    if (record->next != 0) { 
		record->next = 0;
		file.markAsDirty(currIndex[lastId], sizeof(dbTable));
	    }
	}
	tableId = table->next;
    }
}

void dbDatabase::setConcurrency(unsigned nThreads)
{
    if (nThreads == 0) { // autodetect number of processors
	nThreads = dbThread::numberOfProcessors();
    } 
    if (nThreads > dbMaxParallelSearchThreads) { 
	nThreads = dbMaxParallelSearchThreads;
    }
    parThreads = nThreads;
}


bool dbDatabase::loadScheme(bool alter) 
{
    if (!beginTransaction(accessType != dbReadOnly)) { 
	return false;
    }

    dbTableDescriptor *desc, *next;
    dbTable* metaTable = (dbTable*)getRow(dbMetaTableId);
    oid_t first = metaTable->firstRow;
    oid_t last = metaTable->lastRow;
    int nTables = metaTable->nRows;
    oid_t tableId = first;

    for (desc = dbTableDescriptor::chain; desc != NULL; desc = next) {
	next = desc->next;
	if (desc->db != NULL && desc->db != DETACHED_TABLE && desc->db != this) { 
	    continue;
	}
	if (desc->db == DETACHED_TABLE) {
	    desc = desc->clone();
	}
	dbFieldDescriptor* fd;
	for (fd = desc->firstField; fd != NULL; fd = fd->nextField) {
	    fd->tTree = 0;
	    fd->hashTable = 0;
	    fd->attr &= ~dbFieldDescriptor::Updated;
	}

	int n = nTables;
	while (--n >= 0) {
	    dbTable* table = (dbTable*)getRow(tableId);
	    oid_t next = table->next;
	    if (strcmp(desc->name, (char*)table + table->name.offs) == 0) {
		if (!desc->equal(table)) { 
		    if (!alter) { 
			handleError(DatabaseOpenError, "Incompatible class"
				    " definition in application");
			return false;
		    }
		    modified = true;
		    if (table->nRows == 0) { 
			TRACE_MSG(("Replace definition of table '%s'\n", 
				   desc->name));
			updateTableDescriptor(desc, tableId);
		    } else { 
			reformatTable(tableId, desc);
		    } 
		} else { 
		    linkTable(desc, tableId);
		}
		desc->setFlags();
		break;
	    }
	    if (tableId == last) {
		tableId = first;
	    } else { 
		tableId = next;
	    }
	}
	if (n < 0) { // no match found
	    if (accessType == dbReadOnly) { 
		handleError(DatabaseOpenError, "New table definition can not "
			    "be added to read only database");
		return false;
	    } else {     
		TRACE_MSG(("Create new table '%s' in database\n", desc->name));
		addNewTable(desc);
		modified = true;
	    }
	}
	if (accessType != dbReadOnly) { 
	    if (!addIndices(alter, desc)) { 
		handleError(DatabaseOpenError, "Failed to alter indices with"
			    " active applications");
		rollback();
		return false;
	    }
	}
    }	
    for (desc = tables; desc != NULL; desc = desc->nextDbTable) {
	if (desc->cloneOf != NULL) {
	    for (dbFieldDescriptor *fd = desc->firstField; fd != NULL; fd = fd->nextField)
	    {
		if (fd->refTable != NULL) {
		    fd->refTable = lookupTable(fd->refTable);
		}
	    }
	}
	desc->checkRelationship();
    }
    commit();
    return true;
} 


void dbDatabase::reformatTable(oid_t tableId, dbTableDescriptor* desc)
{
    dbTable* table = (dbTable*)putRow(tableId);

    if (desc->match(table)) { 
	TRACE_MSG(("New version of table '%s' is compatible with old one\n", 
		   desc->name));
	updateTableDescriptor(desc, tableId);
    } else { 
	TRACE_MSG(("Reformat table '%s'\n", desc->name));
	oid_t oid = table->firstRow;
	updateTableDescriptor(desc, tableId);
	while (oid != 0) { 
	    dbRecord* record = getRow(oid);
	    size_t size = 
		desc->columns->calculateNewRecordSize((byte*)record, 
						     desc->fixedSize);
	    offs_t offs = currIndex[oid];
	    record = putRow(oid, size);
	    byte* dst = (byte*)record;
	    byte* src = baseAddr + offs;
	    desc->columns->convertRecord(dst, src, desc->fixedSize);
	    oid = record->next;
	}
    }
}
 
void dbDatabase::deleteTable(dbTableDescriptor* desc)
{
    beginTransaction(true);
    modified = true;
    dbTable* table = (dbTable*)putRow(desc->tableId);
    oid_t rowId = table->firstRow;
    table->firstRow = table->lastRow = 0;
    table->nRows = 0;
	
    while (rowId != 0) {
	dbRecord* record = getRow(rowId);
	oid_t nextId = record->next;
	size_t size = record->size;
	
	removeInverseReferences(desc, rowId);

	if (rowId < committedIndexSize && index[0][rowId] == index[1][rowId]) {
	    cloneBitmap(currIndex[rowId], size);
	} else { 
	    deallocate(currIndex[rowId], size);
	}
	freeId(rowId);
	rowId = nextId;
    }
    dbFieldDescriptor* fd;
    for (fd = desc->hashedFields; fd != NULL; fd = fd->nextHashedField) { 
	dbHashTable::purge(this, fd->hashTable);
    } 
    for (fd = desc->indexedFields; fd != NULL; fd = fd->nextIndexedField) { 
	dbTtree::purge(this, fd->tTree);
    } 
} 

void dbDatabase::dropHashTable(dbFieldDescriptor* fd)
{
    beginTransaction(true);
    modified = true;
    dbHashTable::drop(this, fd->hashTable);
    fd->hashTable = 0;
    fd->indexType &= ~HASHED;

    dbFieldDescriptor** fpp = &fd->defTable->hashedFields;
    while (*fpp != fd) { 
	fpp = &(*fpp)->nextHashedField;
    }
    *fpp = fd->nextHashedField;

    dbTable* table = (dbTable*)putRow(fd->defTable->tableId);
    dbField* field = (dbField*)((char*)table + table->fields.offs);
    field[fd->fieldNo].hashTable = 0;
}

void dbDatabase::dropIndex(dbFieldDescriptor* fd)
{
    beginTransaction(true);
    modified = true;
    dbTtree::drop(this, fd->tTree);
    fd->tTree = 0;
    fd->indexType &= ~INDEXED;

    dbFieldDescriptor** fpp = &fd->defTable->indexedFields;
    while (*fpp != fd) { 
	fpp = &(*fpp)->nextIndexedField;
    }
    *fpp = fd->nextIndexedField;

    dbTable* table = (dbTable*)putRow(fd->defTable->tableId);
    dbField* field = (dbField*)((char*)table + table->fields.offs);
    field[fd->fieldNo].tTree = 0;
}

void dbDatabase::createHashTable(dbFieldDescriptor* fd)
{
    beginTransaction(true);
    modified = true;
    dbTable* table = (dbTable*)getRow(fd->defTable->tableId);
    int nRows = table->nRows;
    fd->hashTable = dbHashTable::allocate(this, 2*nRows);
    fd->attr &= ~dbFieldDescriptor::Updated;
    fd->nextHashedField = fd->defTable->hashedFields;
    fd->defTable->hashedFields = fd;
    fd->indexType |= HASHED;
    table = (dbTable*)putRow(fd->defTable->tableId);
    dbField* field = (dbField*)((char*)table + table->fields.offs);
    field[fd->fieldNo].hashTable = fd->hashTable;

    for (oid_t oid = table->firstRow; oid != 0; oid = getRow(oid)->next) {
	dbHashTable::insert(this, fd->hashTable, oid, fd->type, fd->dbsSize, fd->dbsOffs,
			    nRows);
    }
}


void dbDatabase::createIndex(dbFieldDescriptor* fd)
{
    beginTransaction(true);
    modified = true;
    fd->tTree = dbTtree::allocate(this); 
    fd->attr &= ~dbFieldDescriptor::Updated;
    fd->nextIndexedField = fd->defTable->indexedFields;
    fd->defTable->indexedFields = fd;
    fd->indexType |= INDEXED;
    dbTable* table = (dbTable*)putRow(fd->defTable->tableId);
    dbField* field = (dbField*)((char*)table + table->fields.offs);
    field[fd->fieldNo].tTree = fd->tTree;
    
    for (oid_t oid = table->firstRow; oid != 0; oid = getRow(oid)->next) {
	dbTtree::insert(this, fd->tTree, oid, fd->type, fd->dbsSize, fd->comparator, fd->dbsOffs);
    }
}

void dbDatabase::dropTable(dbTableDescriptor* desc)
{
    deleteTable(desc);
    freeRow(dbMetaTableId, desc->tableId);

    dbFieldDescriptor* fd;
    for (fd = desc->hashedFields; fd != NULL; fd = fd->nextHashedField) { 
	dbHashTable::drop(this, fd->hashTable);
    } 
    for (fd = desc->indexedFields; fd != NULL; fd = fd->nextIndexedField) { 
	dbTtree::drop(this, fd->tTree);
    } 
}


bool dbDatabase::addIndices(bool alter, dbTableDescriptor* desc)
{
    dbFieldDescriptor* fd;
    oid_t tableId = desc->tableId;
    dbTable* table = (dbTable*)getRow(tableId);
    int nRows = table->nRows;
    oid_t firstId = table->firstRow;
    int nNewIndices = 0;
    int nDelIndices = 0;
    for (fd = desc->firstField; fd != NULL; fd = fd->nextField) { 
	if ((fd->indexType & HASHED) && fd->type != dbField::tpStructure) { 
	    if (fd->hashTable == 0) { 
		if (!alter && tableId < committedIndexSize
		    && index[0][tableId] == index[1][tableId])
		{
		    // If there are some other active applications which 
		    // can use this table, then they will not know
		    // about newly created index, which leads to inconsistency
		    return false;
		}
		fd->hashTable = dbHashTable::allocate(this, nRows);
		nNewIndices += 1;
		TRACE_MSG(("Create hash table for field '%s'\n", fd->name));
	    }
	} else if (fd->hashTable != 0) { 
	    if (alter) { 
		TRACE_MSG(("Remove hash table for field '%s'\n", fd->name));
		nDelIndices += 1;
		fd->hashTable = 0;
	    } else { 
		return false;
	    }
	}
	if ((fd->indexType & INDEXED) && fd->type != dbField::tpStructure) { 
	    if (fd->tTree == 0) { 
		if (!alter && tableId < committedIndexSize
		    && index[0][tableId] == index[1][tableId])
		{
		    return false;
		}
		fd->tTree = dbTtree::allocate(this);
		nNewIndices += 1;
		TRACE_MSG(("Create index for field '%s'\n", fd->name));
	    }
	} else if (fd->tTree != 0) { 
	    if (alter) { 
		nDelIndices += 1;
		TRACE_MSG(("Remove index for field '%s'\n", fd->name));
		fd->tTree = 0;
	    } else { 
		return false;
	    }
	}
    }
    if (nNewIndices > 0) { 
	modified = true;
	for (oid_t rowId = firstId; rowId != 0; rowId = getRow(rowId)->next) {
	    for (fd = desc->hashedFields; fd != NULL; fd=fd->nextHashedField) {
		if (fd->hashTable >= committedIndexSize 
		    || index[0][fd->hashTable] != index[1][fd->hashTable]) 
		{ 
		    dbHashTable::insert(this, fd->hashTable, rowId, 
					fd->type, fd->dbsSize, fd->dbsOffs, 2*nRows);
		}
	    }
	    for (fd=desc->indexedFields; fd != NULL; fd=fd->nextIndexedField) {
		if (fd->tTree >= committedIndexSize 
		    || index[0][fd->tTree] != index[1][fd->tTree]) 
		{ 
		    dbTtree::insert(this, fd->tTree, rowId, 
				    fd->type, fd->dbsSize, fd->comparator, fd->dbsOffs);
		}
	    }
	} 
    }
    if (nNewIndices + nDelIndices != 0) { 
	table = (dbTable*)putRow(tableId);
	offs_t fieldOffs = currIndex[tableId] + table->fields.offs;
	for (fd = desc->firstField; fd != NULL; fd = fd->nextField) { 
	    dbField* field = (dbField*)(baseAddr + fieldOffs);    
	    if (field->hashTable != fd->hashTable) { 
		if (field->hashTable != 0) { 
		    assert(fd->hashTable == 0);
		    modified = true;
		    dbHashTable::drop(this, field->hashTable);
		    field = (dbField*)(baseAddr + fieldOffs);
		}
		field->hashTable = fd->hashTable;
	    }
	    if (field->tTree != fd->tTree) { 
		if (field->tTree != 0) { 
		    assert(fd->tTree == 0);