sortmerge.c

mini database sort-merge join

#include <string.h>
#include <assert.h>
#include "sortMerge.h"


#define ASSERTFIRST(cond, code) ; if(cond != OK) MINIBASE_FIRST_ERROR(JOINS, code)
#define ASSERT(cond, code) ; if(!(cond)) MINIBASE_CHAIN_ERROR(JOINS, code)

// Error Protocall:

//enum ErrCodes {};

//static const char* ErrMsgs[] = 	{};
struct _rec {
	int	key;
	char	filler[4];
};

enum ErrCodes { FAIL_TO_CREATE_HEAP_FILE };

static const char* ErrMsgs[] =  { "Error: fail to create heapfile" };

static error_string_table ErrTable( JOINS, ErrMsgs );

sortMerge::sortMerge(
    char*           filename1,      // Name of heapfile for relation R
    int             len_in1,        // # of columns in R.
    AttrType        in1[],          // Array containing field types of R.
    short           t1_str_sizes[], // Array containing size of columns in R
    int             join_col_in1,   // The join column of R 

    char*           filename2,      // Name of heapfile for relation S
    int             len_in2,        // # of columns in S.
    AttrType        in2[],          // Array containing field types of S.
    short           t2_str_sizes[], // Array containing size of columns in S
    int             join_col_in2,   // The join column of S

    char*           filename3,      // Name of heapfile for merged results
    int             amt_of_mem,     // Number of pages available
    TupleOrder      order,          // Sorting order: Ascending or Descending
    Status&         s               // Status of constructor
){

	tuple_len1 = 0;
	for(int i=0; i<len_in1; i++){
		tuple_len1 += t1_str_sizes[i];
	}
	
	tuple_len2 = 0;
	for(int i=0; i<len_in2; i++){
		tuple_len2 += t2_str_sizes[i];
	}
	
	filename1s = "sort_R";
	Sort(filename1, filename1s, len_in1, in1, t1_str_sizes, join_col_in1, order, amt_of_mem, s);
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	hf1 = new HeapFile(filename1s, s);
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	scan1 = hf1->openScan(s);
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	
	filename2s = "sort_S";
	Sort(filename2, filename2s, len_in2, in2, t2_str_sizes, join_col_in2, order, amt_of_mem, s);
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	hf2 = new HeapFile(filename2s, s);
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	scan2 = hf2->openScan(s);
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	
	hf3 = new HeapFile(filename3, s);
	if(s!= OK) MINIBASE_FIRST_ERROR(JOINS,s);

	RID rid1,rid2;
	char recPtr1[tuple_len1];
	char recPtr2[tuple_len2];
	
	
	S1 = scan1->getNext(rid1, recPtr1, tuple_len1);

	S2 = scan2->getNext(rid2, recPtr2, tuple_len2);
	
	RID out_rid;
	char out_recPtr[tuple_len1+tuple_len2];
	
	RID pos_rid;
	
	while((S1==OK) && (S2==OK)){
		while (tupleCmp(recPtr1, recPtr2)<0){	
			if (S1 != OK )break;
			S1 = scan1->getNext(rid1, recPtr1, tuple_len1);
		}
		
		while(tupleCmp(recPtr1, recPtr2)>0){
			if (S2!=OK)break;
			S2 = scan2->getNext(rid2, recPtr2, tuple_len2);
		}
	
		while(tupleCmp(recPtr1, recPtr2)==0){
			pos_rid = rid2;
			while (tupleCmp(recPtr1, recPtr2)==0){
				if (S2 != OK)break;
				memcpy((char *)(out_recPtr), recPtr1, tuple_len1);
				memcpy((char *)(out_recPtr+tuple_len1), recPtr2, tuple_len2);
				hf3->insertRecord(out_recPtr, tuple_len1+tuple_len2, out_rid );
				
				RID tmp = rid2;
				while(tmp == rid2){
					if (S2!= OK) break;
					S2 = scan2->getNext(rid2, recPtr2, tuple_len2);
				}
			}
			
			if (S1!=OK) break;
			S1 = scan1->getNext(rid1, recPtr1, tuple_len1);
			if (S1!=OK) break;
			
			rid2 = pos_rid;
			s = scan2->position(rid2);
			if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
			
			s = hf2->getRecord(rid2, recPtr2, tuple_len2);
			if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
		}
		
	}
	delete scan1;
	s=hf1->deleteFile();
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	delete hf1;
	
	delete scan2;
	s=hf2->deleteFile();
	if (s!= OK) MINIBASE_CHAIN_ERROR(JOINS,s);
	delete hf2;
}

sortMerge::~sortMerge()
{
	
}