engine.py

### Phase - 1 : Data formation and loader

import numpy as np
import csv

class Column:

    def __init__(self,name,data,dtype,valid=None):
        self.name=name
        self.data=np.array(data)    #stores the columns of data 
        self.dtype=dtype
    
    
    ###for checking if there any missing value 
    ###if True -> it's valid  
    ###if False -> it's a missing value


        if valid is None:
            self.valid=np.ones(len(self.data),dtype=bool)
        else:
            self.valid=np.array(valid,dtype=bool)
    
        if len(self.data) != len(self.valid):
            raise ValueError(f"data has {len(self.data)} rows but valid has {len(self.valid)}")
    
    def __len__(self):

        return len(self.data)
    
    
    def slice(self,start,end):

        return Column(self.name,
            self.data[start:end],
            self.dtype,
            self.valid[start:end])

#### till now we have created a colmn class for storing data column wise


class Batch:
    
    def __init__(self,columns_dict):
        
        self.columns_dict=columns_dict          ### here the keys are the column's name and 
                                                ### the values are the Column objects of the column class       
        a=len(list(columns_dict.values())[0])

        ### checking if all columns have same no. of rows

        for i in columns_dict.values():
            if len(i)==a:
                pass
            else:
                raise ValueError("not all have same rows")
    
    
    
    def num_rows(self):
        return len(list(self.columns_dict.values())[0])




    def __getitem__(self,key):
        return self.columns_dict[key]



    def select(self,keep_cols):
        new_columns_dict={}
        for i in keep_cols:
            old_column=self.columns_dict[i]
            new_columns_dict[i]=Column(old_column.name,
            old_column.data,
            old_column.dtype,
            old_column.valid)
        return Batch(new_columns_dict)




    def filter(self,mask):
        new_columns_dict={}
        for i in self.columns_dict:
            old_column=self.columns_dict[i]
            new_columns_dict[i]=Column(old_column.name,
            old_column.data[mask],
            old_column.dtype,
            old_column.valid[mask])
        return Batch(new_columns_dict)


### Now storing the batches in a new class table 

class Table:
    
    def __init__(self,name,schema,list_of_batches):
        self.name=name                                  ### table name
        self.schema=schema                              ### table schema dictionary of columns name and their datatype
        self.list_of_batches=list_of_batches            ### list of batches
        
        ### checking if the columns match the schema very important 
        ### for data integrity
        
        for i in range(len(self.list_of_batches)):
            if set(self.list_of_batches[i].columns_dict.keys()) != set(self.schema.keys()):
                raise ValueError("columns don't match schema")
                   



    def num_rows(self):
        a=len(self.list_of_batches)    
        s=0
        for i in range(a):
            s+=self.list_of_batches[i].num_rows()
        return s


class Catalog:
    def __init__(self,name,dic_of_tables):
        self.name=name
        self.dic_of_tables=dic_of_tables
    
    ### adding table objects into the catalog

    def add(self,table_object):
        self.dic_of_tables[table_object.name]=table_object


    def get(self,table_name):
        if table_name in self.dic_of_tables:
            return self.dic_of_tables[table_name]
        else:
            raise ValueError(f"Table {table_name} not found in catalog {self.name}")





# Creating a test CSV file

with open("test.csv", "w") as f:
    f.write("ID,Name,Score\n")
    f.write("1,Alice,85\n")
    f.write("2,Bob,\n")        # Empty score = NULL
    f.write("3,Carol,92\n")
    f.write("4,David,78\n")
    f.write("5,Eve,90\n")

print("Created test.csv")

my_schema={"ID":"INT","Name":"STRING","Score":"FLOAT"}

### Now making a CSV loader because it turns a .csv file on your disk 
### into a Table object that your engine can query. 


def loader(csv_path,schema):
    with open(csv_path,"r",encoding="utf-8") as f:
        reader=csv.reader(f)
        header=next(reader)

        ### verify if schema matches the  header of the file
        d={name:x for x,name in enumerate(header)}
        for i in schema:
            if i not in d:
                raise ValueError(f"Column {i} not found in header of the csv file")
        
        raw_data={}
        for i in schema:
            raw_data[i]=[]
        
### we are now loading the data from csv file with appropriate columns name using d dictionary

        for row in reader:
            for i in schema:
                raw_data[i].append(row[d[i]])



### type conversion for columns to appropriate data type by null masking

        raw_data_mask={}
        for i in raw_data:
            raw_data[i]=np.array(raw_data[i])
            raw_data_mask[i]=np.array((raw_data[i]!="") & (raw_data[i]!="NULL") & (raw_data[i]!="null"))  ### i was doing this wrong because we need
                                                                                                          ### where null values are there we need False but first i used | or which was wrong
                                                                                                          ### and got a wrong result , now the correct one is &
        ## for i in raw_data:
        ##    raw_data[i]=raw_data[i][raw_data_mask[i]]    this deletes the whole null value from the table,but we shouldnot remove this entirely from the table
        columns_dict={}     
        for i in schema:
            strings=raw_data[i]
            mask=raw_data_mask[i]
        
            data_list=[]
            for j,is_valid in zip(strings,mask):
                if is_valid:
                    if schema[i]=="INT":
                        data_list.append(int(j))
                    elif schema[i]=="FLOAT":    
                        data_list.append(float(j))
                    elif schema[i]=="STRING":
                        data_list.append(str(j))
                    else:
                        raise ValueError(f"Unsupported data type {schema[i]} for column {i}")
                else:
                    if schema[i]=="INT":
                        data_list.append(0)
                    elif schema[i]=="FLOAT":
                        data_list.append(0.0)
                    elif schema[i]=="STRING":
                        data_list.append("")
                    else:
                        raise ValueError(f"Unsupported data type {schema[i]} for column {i}")

            n_dtype={"INT":np.int64,"FLOAT":np.float64,"STRING":object}
            data_arr=np.array(data_list,dtype=n_dtype[schema[i]])
            columns_dict[i] = Column(i, data_arr, schema[i], mask)
        batch_of_students = Batch(columns_dict)
        return Table("students",schema, [batch_of_students])

### phase-2: expression trees


class Expression:
    
    def evaluate(self,batch):
        raise NotImplementedError("Subclasses must implement this method")

class ColumnRef(Expression):                        ### getting the a particular column from the batch
    
    def __init__(self,name):
        self.name=name

    
    def evaluate(self,batch):
        return batch[self.name].data

class Literal(Expression):                          ### expression with constant number 
    
    def __init__(self,value):
        self.value=value
    
    def evaluate(self,batch):
        return self.value

class GreaterThan(Expression):                      ### greater than comparison between two expressions
    
    def __init__(self,left,right):
        self.left=left
        self.right=right

    def evaluate(self,batch):
        left_result=self.left.evaluate(batch)      ### here evaluate is used as left is an expression object
        right_result=self.right.evaluate(batch)    ### here evaluate is used as right is an expression object
        return left_result > right_result

class LessThan(Expression):                         ### less than comparison between two expressions

    def __init__(self,left,right):
        self.left=left
        self.right=right
    
    def evaluate(self,batch):
        left_result=self.left.evaluate(batch)      ### here evaluate is used as left is an expression object
        right_result=self.right.evaluate(batch)
        return left_result < right_result

class Equal(Expression):

    def __init__(self,left,right):
        self.left=left
        self.right=right
    
    def evaluate(self,batch):
        left_result=self.left.evaluate(batch)
        right_result=self.right.evaluate(batch)
        return left_result == right_result

class And(Expression):

    def __init__(self,left,right):
        self.left=left
        self.right=right
    
    def evaluate(self,batch):
        left_result=self.left.evaluate(batch)
        right_result=self.right.evaluate(batch)
        return left_result & right_result

class Or(Expression):

    def __init__(self,left,right):
        self.left=left
        self.right=right
        
    def evaluate(self,batch):
        left_result=self.left.evaluate(batch)
        right_result=self.right.evaluate(batch)
        return left_result | right_result

class Not(Expression):
    
    def __init__(self,expression):
        self.expression=expression
    
    def evaluate(self,batch):
        result=self.expression.evaluate(batch)
        return ~result

### here all the above class gives the output as mask so we need filter .

class Filter:

    def __init__(self,expression):
        self.expression=expression
    
    def execute(self,batch):
        mask=self.expression.evaluate(batch)
        return batch.filter(mask)
    
### Phase-3 : join and aggregation

class Projector:
    def __init__(self,column_names):
        self.column_names=column_names

    def execute(self,batch):
        return batch.select(self.column_names)

class HashJoin:

    def __init__(self,left,right):
        self.left=left
        self.right=right
        

    def execute(self,left_batch,right_batch):
        left_result=left_batch[self.left].data   ### here are the data of the matched columns
        right_result=right_batch[self.right].data ### here are the data of the matched columns
        common_data=np.intersect1d(left_result,right_result)
        hash_table1={}
        hash_table2={}
        for i in common_data:
            hash_table1[i]=np.where(left_result==i)[0]
            hash_table2[i]=np.where(right_result==i)[0]
        left_idx=[]
        right_idx=[]
        for i in common_data:
            left_idx.extend(np.repeat(hash_table1[i],len(hash_table2[i])) )
            right_idx.extend(np.tile(hash_table2[i],len(hash_table1[i])) )
        
        joined_batch={}

        for i in left_batch.columns_dict.keys():
            joined_batch[i]=Column(
                i,
                left_batch[i].data[left_idx],
                left_batch[i].dtype,
                left_batch[i].valid[left_idx]   #  Correct: boolean array for matched rows
            )
        for i in right_batch.columns_dict.keys():
            new_name = f"{i}_right"
            joined_batch[new_name]=Column(
                new_name,
                right_batch[i].data[right_idx],
                right_batch[i].dtype,
                right_batch[i].valid[right_idx]   #  Correct: boolean array for matched rows
            )
        return Batch(joined_batch)