sm_manager.cpp

斯坦福大学数据库实现的代码,非常好的.可以了解数据库系统的实现

#include "SM_Manager.h"

SM_Manager::SM_Manager (RM_Manager &rmManager) : rmm(rmManager) { 		 
	bOpen = false;
}

SM_Manager::~SM_Manager () { 
}

RC SM_Manager::CreateDb (const char *dbName) { 
	int ret;
	/* Store the current work directory */
	getcwd (cwd, 80);
	/* Create a directory for the database */
	ret = mkdir (dbName, 00700);
	if (ret == -1) { return PF_UNIX; }
	/* Directory created; chdir to directory */
	ret = chdir (dbName);
	if (ret == -1) { return PF_UNIX; }

	RM_FileHandle fileHandle;
	RID rid;
	SM_Tuple T;
	char *pData;
	RC rc;

	/* Create the relation catalog */
	rc = rmm.CreateFile (relName_relcat, MAXNAME+2*INT_SIZE);
	if (rc != SUCCESS) { return rc; }
	/* Open the relation catalog */
	rc = rmm.OpenFile (relName_relcat, fileHandle);
	if (rc != SUCCESS) { return rc; }
	/* Insert info about relcat into this catalog */
	T = SM_Tuple(dataAttrInfo_relcat, 3, "relcat", MAXNAME+2*INT_SIZE, 3);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	/* Insert info about attrcat into this catalog */
	T = SM_Tuple(dataAttrInfo_relcat, 3, "attrcat", 2*MAXNAME+3*INT_SIZE, 5);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	/* Write pages to disk */
	rc = fileHandle.ForceAllPages ();
	if (rc != SUCCESS) { return rc; }
	/* Close the relation catalog */
	rc = rmm.CloseFile (fileHandle);
	if (rc != SUCCESS) { return rc; }

	/* Create the attribute catalog */
	rc = rmm.CreateFile (relName_attrcat, 2*MAXNAME+3*INT_SIZE);
	if (rc != SUCCESS) { return rc; }
	/* Open the attribute catalog */
	rc = rmm.OpenFile (relName_attrcat, fileHandle);
	if (rc != SUCCESS) { return rc; }
	/* Insert info about relcat into this catalog */
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "relcat", "relName", 0,
			TYPE_STRING, MAXNAME);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "relcat", "tupleLength",
			MAXNAME, TYPE_INT, INT_SIZE);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "relcat", "attrCount",
			MAXNAME+INT_SIZE, TYPE_INT, INT_SIZE);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	/* Insert info about attrcat into this catalog */
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "attrcat", "relName", 0,
			TYPE_STRING, MAXNAME);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "attrcat", "attrName",
			MAXNAME, TYPE_STRING, MAXNAME);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "attrcat", "offset",
			2*MAXNAME, TYPE_INT, INT_SIZE);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "attrcat", "attrType",
			2*MAXNAME+INT_SIZE, TYPE_INT, INT_SIZE);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	T = SM_Tuple(dataAttrInfo_attrcat, 5, "attrcat", "attrLength",
			2*MAXNAME+2*INT_SIZE, TYPE_INT, INT_SIZE);
		T.GetData(pData);
		rc = fileHandle.InsertRec (pData, rid);
		if (rc != SUCCESS) { return rc; }
	/* Write pages to disk */
	rc = fileHandle.ForceAllPages ();
	if (rc != SUCCESS) { return rc; }
	/* Close the relation catalog */
	rc = rmm.CloseFile (fileHandle);
	if (rc != SUCCESS) { return rc; }

	/* Catalogs created; return to the original directory */
	chdir (cwd);

	return SUCCESS;
}

RC SM_Manager::DestroyDb (const char *dbName) { 
	char cmd[80];
	/* Build system command to delete directory */
	sprintf (cmd, "rm -r %s", dbName);
	/* Execute sys command */
	if (system (cmd) == -1) {
		return PF_UNIX;
	}
	return SUCCESS;
}

RC SM_Manager::OpenDb (const char *dbName) { 
	/* Check if we are already connected to a database */
	if (bOpen) { return SM_DBOPEN; }

	getcwd (cwd, 80);
	/* chdir to the DB directory */
	if (chdir(dbName) < 0) {
		fprintf (stderr, "could not change directory to %s\n", dbName);
		return PF_UNIX;
	}

	RC rc;
	/* Open a handle to the relation catalog */
	rc = rmm.OpenFile (relName_relcat, fileHandle_relcat);
	if (rc != SUCCESS) { return rc; }
	/* Open a handle to the attribute catalog */
	rc = rmm.OpenFile (relName_attrcat, fileHandle_attrcat);
	if (rc != SUCCESS) { return rc; }

	/* Successfully opened catalog files */
	bOpen = true;
	return SUCCESS;
}

RC SM_Manager::CloseDb () { 
	/* Check if we are not connected to a database */
	if (!bOpen) { return SM_DBCLOSED; }

	RC rc;
	/* Flush dirty catalog pages to disk */
	rc = fileHandle_relcat.ForceAllPages ();
	if (rc != SUCCESS) { return rc; }
	rc = fileHandle_attrcat.ForceAllPages ();
	if (rc != SUCCESS) { return rc; }
	/* Close handle to the relation catalog */
	rc = rmm.CloseFile (fileHandle_relcat);
	if (rc != SUCCESS) { return rc; }
	/* Close handle to the attribute catalog */
	rc = rmm.CloseFile (fileHandle_attrcat);
	if (rc != SUCCESS) { return rc; }

	/* Get back to the previous directory */
	chdir (cwd);

	/* Successfully closed catalog files */
	bOpen = false;
	return SUCCESS;
}

RC SM_Manager::CreateTable (const char *relName, 
				 int attrCount,
				 AttrInfo *attributes) {
	/* Check if we are not connected to a database */
	if (!bOpen) { return SM_DBCLOSED; }

	/* Check if relName is either "relcat" or "attrcat" */
	if (strcmp(relName, relName_relcat) == 0
		|| strcmp(relName, relName_attrcat) ==0 ) {
			return SM_RELNAMECAT;
	}

	DataAttrInfo *info; 
	int nAttr;
	/* Check if the relation already exits */
	if (GetDataAttrInfo(relName, info, nAttr) == SUCCESS) {
		free (info);
		return SM_RELEXISTS;
	}

	int recordSize = 0;
	/* Determine record size of the relation to be created */
	for (int i = 0; i < attrCount; i++) {
		recordSize += attributes[i].attrLength;
	}

	RC rc;
	/* Create a new file for this relation */
	rc = rmm.CreateFile (relName, recordSize);
	if (rc != SUCCESS) { 
		return rc; 
	}

	SM_Tuple T;
	RID rid_relcat;
	char *pData;
	/* Update the relation catalog */
	T = SM_Tuple(dataAttrInfo_relcat, 3, relName, recordSize, attrCount);
	T.GetData(pData);
	rc = fileHandle_relcat.InsertRec (pData, rid_relcat);
	/* Clean up if an error occurs */
	if (rc != SUCCESS) { 
		rmm.DestroyFile (relName);
		return rc; 
	}	

	RID rid_attrcat;
	/* Update the attribute catalog */
	int offset = 0;
	for (int i = 0; i < attrCount; i++) {
		/* Construct data ptr */
		T = SM_Tuple(dataAttrInfo_attrcat, 5, relName,
				attributes[i].attrName, offset,
				attributes[i].attrType,
				attributes[i].attrLength);
		T.GetData(pData);
		offset += attributes[i].attrLength;
		/* Insert record into the catalog */
		rc = fileHandle_attrcat.InsertRec (pData, rid_attrcat);
		/* Clean up and return if an error occurs */
		if (rc != SUCCESS) { 
			rmm.DestroyFile (relName);
			return rc; 
		}
	}

	/* Successfully created table */
	return SUCCESS;
}

RC SM_Manager::DropTable (const char *relName) { 
	/* Check if we are not connected to a database */
	if (!bOpen) { return SM_DBCLOSED; }

	/* Check if relName is either "relcat" or "attrcat" */
	if (strcmp(relName, relName_relcat) == 0
		|| strcmp(relName, relName_attrcat) ==0 ) {
			return SM_RELNAMECAT;
	}

	RC rc;
	/* Drop the file used by this relation */
	rc = rmm.DestroyFile (relName);
	if (rc != SUCCESS) { return rc; }

	RM_FileScan fileScan;
	RM_Record record;
	RID rid;

	char rName[MAXNAME+1];
	memset (rName, 0, MAXNAME+1);
	/* Make a copy of relName for use in file scans */
	strcpy (rName, relName);
	/* Update the relation catalog */
	rc = fileScan.OpenScan (fileHandle_relcat, TYPE_STRING,
			MAXNAME, 0, EQ_OP, rName, NO_HINT);
	if (rc != SUCCESS) { return rc; }
	/* Get the tuple corresponding to the deleted relation */
	rc = fileScan.GetNextRec (record);
	if (rc != SUCCESS) { return rc; }
	/* Get the tuple's rid */
	record.GetRid (rid);
	/* Delete the tuple */
	rc = fileHandle_relcat.DeleteRec (rid);
	if (rc != SUCCESS) { return rc; }
	/* Close the scan */
	rc = fileScan.CloseScan ();
	if (rc != SUCCESS) { return rc; }

	/* Update the attribute catalog */
	rc = fileScan.OpenScan (fileHandle_attrcat, TYPE_STRING,
			MAXNAME, 0, EQ_OP, rName, NO_HINT);
	if (rc != SUCCESS) { return rc; }
	/* Get the tuples corresponding to the deleted relation */
	while (fileScan.GetNextRec (record) == SUCCESS) {
		/* Get tuple's rid */
		rc = record.GetRid (rid);
		if (rc != SUCCESS) { 
			fileScan.CloseScan ();
			return rc;
		}
		/* Delete tuple */
		rc = fileHandle_attrcat.DeleteRec (rid);
		/* Clean up before returning from an error */
		if (rc != SUCCESS) { 
			fileScan.CloseScan ();
			return rc; 
		}
	}
	/* Close the scan */
	rc = fileScan.CloseScan ();
	if (rc != SUCCESS) { return rc; }

	/* Successfully deleted the relation and updated the catalogs */
	return SUCCESS;
}

RC SM_Manager::DeleteRecords (const char *relName) {
	/* Check if we are not connected to a database */
	if (!bOpen) { return SM_DBCLOSED; }
	/* Check if relName is either "relcat" or "attrcat" */
	if (strcmp(relName, relName_relcat) == 0
		|| strcmp(relName, relName_attrcat) ==0 ) {
			return SM_RELNAMECAT;
	}

	/* Step 1 */
	RC rc;
	/* Determine the structure of the relation */
	DataAttrInfo *info;
	int attrCount;
	rc = GetDataAttrInfo (relName, info, attrCount);
	if (rc != SUCCESS) { return rc; }
	/* We don't need the structure information for deleting records */
	/* Free up memory */
	free (info);

	/* Step 2 */
	/* Scan the relation and delete each record */
	RM_FileHandle fileHandle;
	/* Get a handle to the file first */
	rc = rmm.OpenFile (relName, fileHandle);
	if (rc != SUCCESS) { return rc; }

	RM_FileScan fileScan;
	/* Open a file scan using the file handle */
	rc = fileScan.OpenScan (fileHandle, 
			TYPE_STRING, MAXNAME, 0, /* <-- Doesn't matter what
						    these values are
						    because the scan is
						    based on NO_OP */
			NO_OP, NULL, NO_HINT);
	if (rc != SUCCESS) { return rc; }

	RM_Record rec;
	/* Get each record from the file */
	while (fileScan.GetNextRec(rec) == SUCCESS) {
		char *pData;
		/* Get the record's contents (data) */
		rc = rec.GetData (pData);
		/* Clean up before returning from an error */
		if (rc != SUCCESS) {
			fileScan.CloseScan ();
			return rc;
		}
		RID rid;
		/* Get the record's rid */
		rc = rec.GetRid (rid);
		if (rc != SUCCESS) { 
			fileScan.CloseScan ();
			return rc; 
		}
		/* Delete the record */
		rc = fileHandle.DeleteRec (rid);
		if (rc != SUCCESS) { 
			fileScan.CloseScan ();
			return rc; 
		}
	}

	/* Close the scan */
	rc = fileScan.CloseScan ();
	if (rc != SUCCESS) { return rc; }

	/* Flush dirty pages to disk */
	rc = fileHandle.ForceAllPages ();
	if (rc != SUCCESS) { return rc; }
	/* Close the file handle */
	rc = rmm.CloseFile (fileHandle);
	if (rc != SUCCESS) { return rc; }

	/* Successfully printed relation */
	return SUCCESS;
}

RC SM_Manager::Load (const char *relName, const char *fileName) {
	/* Check if we are not connected to a database */
	if (!bOpen) { return SM_DBCLOSED; }
	/* Check if relName is either "relcat" or "attrcat" */
	if (strcmp(relName, relName_relcat) == 0
		|| strcmp(relName, relName_attrcat) ==0 ) {
			return SM_RELNAMECAT;
	}

	DataAttrInfo *info;
	int attrCount;
	RC rc;
	/* Get the structure of the relation */
	rc = GetDataAttrInfo (relName, info, attrCount);
	if (rc != SUCCESS) { return rc; }

	int tupleSize = 0;
	/* Determine the tuple size of this relation */
	tupleSize = info[attrCount-1].offset +
		info[attrCount-1].attrLength;

	RM_FileHandle fileHandle;
	/* Open the table */
	rc = rmm.OpenFile (relName, fileHandle);
	if (rc != SUCCESS) { 
		free (info);
		return rc; 
	}

	/* Open the file for loading */
	FILE *ifp; // Input file pointer
	ifp = fopen (fileName, "r");
	if (ifp == NULL) {
		free (info);
		perror ("ERR");
		return PF_UNIX;
	}

	/* Store line number information */
	int lineNum = 1;

	/* Read in each line of the file */
	while (!feof(ifp)) {
		char *pData;
		/* Create an "empty" tuple */
		pData = (char *) malloc(tupleSize);
		memset (pData, 0, tupleSize);
		SM_Tuple T (info, attrCount, pData);	
				// Store each line of input in tuple T 
				// first. Then use this tuple to insert 
				// a record into the file.

		int ret = 0;
		/* Parse the input buffer */
		for (int i = 0; i < attrCount; i++) {
			ret = 0;
			switch (info[i].attrType) {
				case TYPE_INT:
					int j;
					ret = fscanf (ifp, " %d ", &j);