/********************************************************************************
*
* File: evaluation.gen
* Title: The evaluator for OQL.
* Programmer: Leonidas Fegaras, UTA
* Date: 4/24/97
*
*********************************************************************************/

#include "evaluation.h"
#include "typecheck.h"
#include "constant.h"


list<Object>* Onil = new list<Object>;


#define DEBUG(name,action) catch (object::evaluation_exception)\
 { cout << "********************************************************************************\n"\
        << #name << ": \n";\
   action;\
   throw object::evaluation_exception(); }


/* a type error is a fatal error: it causes the program to abort */
TypeError ( string msg, Expr e ) {
  cout << "\n*** Type Error: " << msg << " (" << e << ")\n";
  throw object::evaluation_exception();
};


/* the actual database (schema & data) */
database* DATABASE;


/* all_plans holds the names of all physical plans (defined at optimizer.gen) */
extern list<Expr>* all_plans;


/* all primitive methods (eg. plus, and, etc) (defined at constant.cc) */
extern Internals internal_methods;


Object TRUE;
Object FALSE;


/* defined in optimizer.gen: is computed_order stronger then expected_order? */
bool subsumes ( Expr computed_order, Expr expected_order );


template<class T>
list<T>* ntimes ( int n, T x ) {
  if (n) return Cons(x,ntimes(n-1,x));
  else return new list<T>;
};


/* an evaluation error is a fatal error: it causes the program to abort
*  after it dumps the calling stack (all the functions that use the DEBUG macro) */
evaluation_error ( string msg, Object o ) {
  cout << "\n*** Run-time Error: " << msg << " (" << o << ")\n";
  throw object::evaluation_exception();
};


Type schema_to_type ( Schema sch ) {
  list<Expr>* args = Nil;
  for(Schema n = sch; n->more(); n=n->next())
     args = args->cons(#<bind(`(variable(n->current()->first)),`(n->current()->second))>);
  args = args->reverse();
  return #<struct(...args)>;
};


Schema type_to_schema ( Type tp ) {
  Schema res = new binding<Type>;
  #case tp
  | struct(...args)
      => for(list<Expr>* r = args->reverse(); r->consp(); r=r->tl)
	    #case r->hd
	    | bind(`x,`t) => res = res->extend(x->name(),t);
	    | _ => evaluation_error("(in type_to_schema) Ill-formed record type",new object());
	    #end;
  | _ => evaluation_error("(in type_to_schema) Expected a record type",new object());
  #end;
  return res;
};


Environment object_to_env ( Object o, Type tp ) {
  Environment res = new binding<Object>; 
  #case tp
  | struct(...args)
      => if (o->aggregationp())
         {  list<Object>* s = o->info.Aggregation.Components;
            for(list<Expr>* r = args->reverse(); r->consp(); r=r->tl, s=s->tl)
	    #case r->hd
	    | bind(`x,`t) => res = res->extend(x->name(),s->hd);
	    | _ => evaluation_error("(in object_to_env) Ill-formed record type",new object());
	    #end;
         }
         else evaluation_error("(in object_to_env) Expected an aggregation",new object());
  | _ => evaluation_error("(in object_to_env) Expected a record type",new object());
  #end;
  return res;
};


/*********************************************************************************/


interface::interface () {
  name = new_name();
  primary_key = new list<String>;
  schema = new binding<Type>;
  methods = new binding<Method>;
  supertype = NULL;
  subtypes = new list<Interface>;
  data = NULL;
};


interface::interface ( String nm ) {
  name = nm;
  primary_key = new list<String>;
  schema = new binding<Type>;
  methods = new binding<Method>;
  supertype = NULL;
  subtypes = new list<Interface>;
  data = NULL;
};


interface::interface ( Interface intrf ) {
  name = intrf->name;
  primary_key = intrf->primary_key;
  schema = intrf->schema;
  methods = intrf->methods;
  supertype = intrf->supertype;
  subtypes = intrf->subtypes;
  data = intrf->data;
};


interface::interface ( Expr class_def ) {
/* compile an ODL class definition */
  name = new_name();
  primary_key = new list<String>;
  schema = new binding<Type>;
  methods = new binding<Method>;
  supertype = NULL;
  subtypes = new list<Interface>;
  data = NULL;
  #case class_def
  |  interface(`nm,`super,info(...keys),...attrbs)
	=> { name = nm->name();
	     schema = new binding<Type>;
	     for(list<Expr>* r = attrbs->reverse(); r->consp(); r=r->tl)
		#case r->hd
		| attribute(`v,`tp) => schema = schema->extend(v->name(),tp);
		| relationship(`v,`tp,_,_) => schema = schema->extend(v->name(),tp);
		| method(`f,`body,`tp,...params)
		    => methods = methods->extend(f->name(),
			  new pair<Signature,Expr>(new pair<Schema,Type>(type_to_schema(#<struct(...params)>),
								         tp),body));
		#end;
	     if (!(super->eq(#<NONE>)))
	     {  Interface SI = find_interface(super->name());
		supertype = SI;
		schema = SI->schema->extend(schema);
		SI->subtypes = SI->subtypes->cons(this);
	     };
	     for(list<Expr>* r = keys; r->consp(); r=r->tl)
		#case r->hd
		|  extent(`ext) => { data = new extent(ext->name(),schema);
				     DATABASE->content = DATABASE->content->extend(ext->name(),data);
				     DATABASE->extents = DATABASE->extents->extend(name,ext->name());
				   };
		|  key(...keys) => { primary_key = new list<String>;
				     for(list<Expr>* k = keys->reverse(); k->consp(); k=k->tl)
					primary_key = primary_key->cons(k->hd->name());
				   };
		#end;
	     DATABASE->schema = DATABASE->schema->extend(nm->name(),this);
	   };
  #end;
};


/* display an interface */
ostream& operator<< ( ostream& s, Interface itrf ) {
  s << "interface: " << itrf->name << " " << schema_to_type(itrf->schema)
    << "\n           ( key = " << itrf->primary_key;
  if (itrf->supertype != NULL)
     s << ", supertype = " << itrf->supertype->name;
  if (itrf->subtypes->consp())
  { Names n = new list<String>;
    for(list<Interface>* r = itrf->subtypes->reverse(); r->consp(); r=r->tl)
       n = n->cons(r->hd->name);
    s << ", subtypes = " << n;
  };
  s << " )\n";
  s << "methods: \n";
  for(binding<Method>* r = itrf->methods; r->more(); r=r->next())
     s << "   " << r->current()->first << r->current()->second->first << ": "
       << r->current()->second->second << endl;
  return s;
};


Expr names_to_expr ( Names n ) {
  list<Expr>* args = Nil;
  for(Names r = n->reverse(); r->consp(); r=r->tl)
     args = args->cons(variable(r->hd));
  return #<names(...args)>;
};


Names expr_to_names ( Expr e ) {
  Names args = new list<String>;
  for(list<Expr>* r = e->arguments()->reverse(); r->consp(); r=r->tl)
     args = args->cons(r->hd->name());
  return args;
};


Interface interface_read ( ifstream& from ) {
  Interface I = new interface();
  char c;
  int len;
  from.get(c);
  I->name = reads(from);
  I->primary_key = expr_to_names(read(from));
  I->schema = type_to_schema(read(from));
  from >> len;
  for(int i = 0; i<len; i++)
     I->methods = I->methods->extend(reads(from),
	new pair<Signature,Expr>(new pair<Schema,Type>(type_to_schema(read(from)),
		read(from)),read(from)));
  String name = reads(from);
  if (!name->eq(""))
     I->supertype = new interface(name);			/* will be set properly later */
  Names ns = new list<String>;
  from >> len;
  for(int i = 0; i<len; i++)
     ns = ns->cons(reads(from));
  for(Names r = ns->reverse(); r->consp(); r=r->tl)
     I->subtypes = I->subtypes->cons(new interface(r->hd));	/* will be set properly later */
  I->data = new extent(reads(from),new binding<Expr>);		/* will be set properly later */
  return I;
};


interface::interface_write ( ofstream& to ) {
  name->write(to);
  names_to_expr(primary_key)->write(to);
  schema_to_type(schema)->write(to);
  to << ' ' << methods->names()->length();
  for(binding<Method>* m = methods; m->more(); m=m->next())
  { schema_to_type(m->current()->second->first->first)->write(to);
    m->current()->second->first->second->write(to);
    m->current()->second->second->write(to);
  };
  if (supertype == NULL)
     (new string(""))->write(to);
  else supertype->name->write(to);
  to << ' ' << subtypes->length();
  for(list<Interface>* r = subtypes->reverse(); r->consp(); r=r->tl)
     r->hd->name->write(to);
  data->name->write(to);
};


/*********************************************************************************/


extent::extent ( String nm, Schema sch ) {
  name = nm;
  schema = sch;
  indexes = new binding<Names>;
  data = Onil;
  cardinality = 0;
  object_size = sch->names()->length()+1;
};


extent::extent ( Extent r ) {
  name = r->name;
  schema = r->schema;
  indexes = r->indexes;
  data = Onil;
  cardinality = 0;
  object_size = r->object_size;
};


/* insert a new object in the extent */
extent::insert ( Object x ) {
  data = data->append(Cons(x,Onil));
  cardinality++;
};


/* find an index for this interface that delivers this order */
Expr extent::find_index ( Names order ) {
  if (order->nullp())
     return #<none>;
  for(binding<Names>* n = indexes; n->more(); n=n->next())
  {  list<Expr>* idx = Nil;
     for(Names r = n->current()->second->reverse(); r->consp(); r=r->tl)
	idx = idx->cons(variable(r->hd));
     if (subsumes(#<order(...idx)>,#<none>))
	return #<index(`(variable(n->current()->first)),order(...idx))>;
  };
  return #<none>;
};


/* create a collection object that contains the extent */
Object extent::scan ( String var, Expr pred ) {
  Type element_type = schema_to_type(schema);
  Type etype = #<struct(bind(`(variable(var)),`element_type))>;
  list<Object>* el = Onil;
  Environment env = new binding<Object>;
  for(list<Object>* r = data->reverse(); r->consp(); r=r->tl)
  { Object o = new object(etype,Onil->cons(r->hd));
    if (o->predicate(pred,env))
       el = el->cons(o);
  };
  Type ctype = #<list(`etype)>;
  Object res = new object(ctype);
  res->info.Collection = el;
  return res;
};


/* print the data content of an extent */
ostream& operator<< ( ostream& s, Extent e ) {
  s << "Extent: " << e->name << " " << schema_to_type(e->schema)
    << "\n        ( size = " << e->object_size
    << ", cardinality = " << e->cardinality << " )\n"
    << e->data;
  return s;
};


Extent extent_read ( ifstream& from ) {
  int len;
  String s = reads(from);
  Extent E = new extent(s,type_to_schema(read(from)));
  from >> E->object_size >> E->cardinality >> len;
  for(int i = 0; i<len; i++)
     E->indexes = E->indexes->extend(reads(from),expr_to_names(read(from)));
  from >> len;
  for(int i = 0; i<len; i++)
     E->data = E->data->cons(object_read(from));
  return E;
};


extent::extent_write ( ofstream& to ) {
  name->write(to);
  schema_to_type(schema)->write(to);
  to << ' ' << object_size << ' ' << cardinality
     << ' ' << indexes->names()->length();
  for(binding<Names>* r = indexes; r->more(); r=r->next())
  {  r->current()->first->write(to);
     names_to_expr(r->current()->second)->write(to);
  };
  to << ' ' << data->length();
  for(list<Object>* r = data; r->consp(); r=r->tl)
     r->hd->object_write(to);
};


/*********************************************************************************/


int object::available_OID = 0;


object::object ( Type tp, list<Object>* r ) {
  tag = AGGREGATION;
  info.Aggregation.OID = available_OID;
  available_OID++;
  info.Aggregation.Components = r;
  otype = tp;
};


object::object ( Type tp ) {
  tag = COLLECTION;
  info.Collection = Onil;
  otype = tp;
};


/* shallow equality */
bool object::eq ( Object o ) {
  if (nullp() && o->nullp()) return true;
  if (integerp() && o->integerp())
     return info.Integer == o->info.Integer;
  if (boolp() && o->boolp())
     return info.Boolean == o->info.Boolean;
  if (stringp() && o->stringp())
     return info.Estring->eq(o->info.Estring);
  if (aggregationp() && o->aggregationp())
     return info.Aggregation.OID == o->info.Aggregation.OID;
  return (this == o);
};


/* evaluate a projection */
Object object::projection ( String attribute ) {
  if (nullp())
     return new object();
  if (!aggregationp())
     evaluation_error("(in method object::projection) Expected a record",this);
  Schema schema = type_to_schema(otype);
  list<Object>* r = info.Aggregation.Components;
  for(Schema n = schema; n->more(); n=n->next(), r=r->tl)
     if (n->current()->first->eq(attribute))
	return r->hd;
  evaluation_error(form("(in method object::projection) No such attribute %s",
			attribute->value()),this);
};


Object object::projection ( Names attributes ) {
  if (nullp())
     return new object();
  if (!aggregationp())
     evaluation_error("(in method object::projection) Expected a record",this);
  Schema schema = type_to_schema(otype);
  list<Object>* atrbs = Onil;
  Schema sch = new binding<Type>;
  list<Object>* r;
  for(Names n = attributes; n->consp(); n=n->tl)
  {  r = info.Aggregation.Components;
     for(Schema m = schema; m->more(); m=m->next(), r=r->tl)
	if (m->current()->first->eq(n->hd))
	{ sch = sch->extend(n->hd,m->current()->second);
	  atrbs = atrbs->cons(r->hd);
	};
  };
  return new object(schema_to_type(sch),atrbs->reverse());
};


/* evaluate a method */
Object object::method ( String fnc, list<Object>* arguments, Environment& env ) {
try {
  binding<Method>* methods = new binding<Method>;
  if (otype->variablep())
     methods = find_interface(otype->name())->methods;
  if (methods->in(fnc))
  {  Method m = methods->find(fnc);
     if (m->first->first->length() != arguments->length())
	evaluation_error("(in method object::method) Mismatched arguments in method",this);
     list<Object>* r = arguments->reverse();
     binding<Object>* new_env = env;
     for(Names s = m->first->first->names()->reverse(); s->consp(); s=s->tl, r=r->tl)
	new_env = new_env->extend(s->hd,r->hd);
     return expression(m->second,new_env);
   }
   else evaluation_error(form("(in method object::method) No such method: %s",fnc->value()),this);
} DEBUG(object::method,cout << this << endl << fnc << " " << arguments << endl << env << endl)
};


/* functions that correspond to various monoid accumulators */
int sum (int x,int y) { return x+y; };
int max (int x,int y) { return x<y ? y : x; };
int min (int x,int y) { return x<y ? x : y; };
bool and (bool x, bool y) { return x && y; };
bool or (bool x, bool y) { return x || y; };


Object zero ( Expr monoid ) {
  static binding<Object>* zeros = (new binding<Object>)
	->extend(new string("sum"),new object((int) 0))
	->extend(new string("max"),new object((int) 0))
	->extend(new string("min"),new object(32000))
	->extend(new string("all"),new object(true))
	->extend(new string("some"),new object(false));
  if (zeros->in(monoid->name()))
     return zeros->find(monoid->name());
  else return new object(#<`monoid(`(fresh_type_variable()))>);
};


/* evaluate an OQL expression */
Object object::expression ( Expr e, Environment& env ) {
try {
  if (false && trace_evaluation)
     cout << "*** EVAL: " << e << endl << "          " << this << endl;
  Object res;
  #case e
  |  nulll => res = new object();
  |  true => res = new object ((bool) true);
  |  false => res = new object ((bool) false);
  |  zero(`M) => res = zero(M);
  |  unit(`M,`e) => { Object o = expression(e,env);
		      if (M->name()->eq("sum") || M->name()->eq("min") || M->name()->eq("max")
			  || M->name()->eq("all") || M->name()->eq("some"))
			 res = o;
		      else {
			res = new object(#<`M(`(o->otype))>);
			res->info.Collection = Onil->cons(o);
		      };
		    };
  | MERGE(`M,`x,`y)
	=> { Object ox = evaluate_plan(x,env);
	     Object oy = evaluate_plan(y,env);
	     String accumulator = M->name();
	     if (accumulator->eq("sum") || accumulator->eq("min") || accumulator->eq("max"))
	     { int (*acc) (int,int) = accumulator->eq("sum") ? sum : (accumulator->eq("min") ? min : max);
	       res = new object((*acc)(ox->info.Integer,oy->info.Integer));
	     }
	     else if (accumulator->eq("all") || accumulator->eq("some"))
	     { bool (*acc) (bool,bool) = accumulator->eq("all") ? and : or;
	       res = new object((*acc)(ox->info.Boolean,oy->info.Boolean));
	     }
	     else {
	       res = new object(ox->otype);
	       res->info.Collection = ox->info.Collection->append(oy->info.Collection);
	     };
	   };
  |  project(`e,`atrb)
	=> res = expression(e,env)->projection(atrb->name());
  |  method(`f,...args)
	=> { list<Object>* vals = Onil;
	     for(list<Expr>* r = args->reverse(); r->consp(); r=r->tl)
		vals = vals->cons(expression(r->hd,env));
	     res = method(f->name(),vals,env);
	   };
  | struct(...r)
	=> { list<Object>* args = Onil;
	     Schema schema = new binding<Type>;
	     for(list<Expr>* s = r->reverse(); s->consp(); s=s->tl)
		#case s->hd
		| bind(`x,`v) => args = args->cons(expression(v,env));
				 schema = schema->extend(x->name(),fresh_type_variable());
		#end;
	     res = new object(schema_to_type(schema),args);
	    };
  | if(`pred,`x,`y)
	=> if (predicate(pred,env))
	      res = expression(x,env);
	   else res = expression(y,env);
  | eq(`x,`y) => res = new object(expression(x,env)->eq(expression(y,env)));
  | neq(`x,`y) => res = new object(!expression(x,env)->eq(expression(y,env)));
  | and(...args)
	=> { bool b = true;
	     for(; b && args->consp(); args=args->tl)
		b = predicate(args->hd,env);
	     res = new object(b);
	   };
  | or(...args)
	=> { bool b = false;
	     for(; !b && args->consp(); args=args->tl)
		b = predicate(args->hd,env);
	     res = new object(b);
	   };
  | OID(`x)
	 => { Object o = expression(x,env);
	      if (o->nullp())
		 res = new object(-1);
	      else if (o->aggregationp())
		 res = new object(o->info.Aggregation.OID);
	      else evaluation_error("(in method object::expression) OID must be applied on a structure",o);
	    };
  |  null => return new object();
  |  `f(...args)
        => if (member(f,all_interfaces()))		/* method is a constructor */
	   { list<Object>* vals = Onil;
	     for(list<Expr>* r = args->reverse(); r->consp(); r=r->tl)
		vals = vals->cons(expression(r->hd,env));
	     Interface I = find_interface(f->name());
	     res = new object(schema_to_type(I->schema),vals);
	     I->data->insert(res);
	     for(Interface r = I->supertype; r; r=r->supertype)
	        r->data->insert(new object(schema_to_type(r->schema),vals));
	   }
	   else if (member(f,Nil->cons(#<set>)->cons(#<bag>)->cons(#<list>)))
	   { list<Object>* vals = Onil;
	     for(list<Expr>* r = args; r->consp(); r=r->tl)
		vals = vals->cons(expression(r->hd,env));
	     res = new object(#<list(`(fresh_type_variable()))>);
	     res->info.Collection = vals->reverse();
	   }
	   else if (member(f,all_plans))
		   return evaluate_plan(e,env);
	   else if (internal_methods->in(f->name()))
		   { list<Object>* params = Onil;
		     pair<Signature,Function>* p = internal_methods->find(f->name());
		     bool null = false;
		     for(list<Expr>* r = args; r->consp(); r=r->tl)
		     {  Object o = expression(r->hd,env);
			if (o->nullp()) null = true;
			params = params->cons(o);
		     };
		     if (null)
			res = new object();
		     else res = (*p->second)(params->reverse());
		   }
	   else evaluation_error(form("(in method object::expression) No such function: %s",f->name()->value()),this);
  | _ => if (e->integerp())
	    res = new object(e->info.Integer);
	 else if (e->stringp())
	    res = new object(e->info.Estring);
	 else if (e->variablep())
		 if (env->in(e->name()))
		    res = env->find(e->name());
		 else if (type_to_schema(otype)->in(e->name()))
			 res = projection(e->name());
		 else TypeError("Invalid expression",e);
	 else TypeError("Invalid expression",e);
  #end;
  if (false && trace_evaluation)
     cout << "--> " << res << endl;
  return res;
} DEBUG(object::expression,cout << this << endl << e << " " << env << endl)
};


/* evaluate an OQL predicate */
bool object::predicate ( Expr pred, Environment& env ) {
try {
  Object o = expression(pred,env);
  if (o->boolp()) return o->info.Boolean;
  if (o->nullp()) return false;
  evaluation_error("(in method object::predicate) Expected a boolean value",o);
} DEBUG(object::predicate,cout << this << endl << pred << " " << env << endl)
};


/* C style comparison: it returns attribute-x.attribute */
int object::compare ( Object x, String attribute ) {
try {
  Object rx = projection(attribute);
  Object ry = x->projection(attribute);
  if (rx->integerp() && ry->integerp())
     return rx->info.Integer - ry->info.Integer;
  if (rx->boolp() && ry->boolp())
     return rx->info.Boolean - ry->info.Boolean;
  else if (rx->stringp() && ry->stringp())
     return strcmp(rx->info.Estring->value(),ry->info.Estring->value());
  else if (rx->aggregationp() && ry->aggregationp())
     return rx->info.Aggregation.OID - ry->info.Aggregation.OID;
  else if (rx->nullp() && ry->nullp())
     return 0;
  else if (rx->nullp())
    return 1;
  else if (ry->nullp())
    return -1;
  else evaluation_error("Can't do the comparison",this);
} DEBUG(object::compare,cout << this << endl << x << endl << attribute << endl)
};


/* true if for all attributes A in attributes: A=x.A */
bool object::equal_to ( Object x, Names attributes ) {
try {
  for(Names r = attributes; r->consp(); r=r->tl)
  {  int c = compare(x,r->hd);
     if (c != 0) return false;
  };
  return true;
} DEBUG(object::equal_to,cout << this << endl << x << endl << attributes << endl)
};


/* true if for all attributes A in attributes: A<x.A */
bool object::less_than ( Object x, Names attributes ) {
try {
  for(Names r = attributes; r->consp(); r=r->tl)
  {  int c = compare(x,r->hd);
     if (c>0) return false;
     if (c<0) return true;
  };
  return false;
} DEBUG(object::less_than,cout << this << endl << x << endl << attributes << endl)
};


/* the NF2 unnest operator */
Object object::unnest ( String var, Expr path, Expr pred, bool keep, Environment& env ) {
try {
  Schema schema;
  list<Expr>* args = Nil;
  #case otype
  | list(struct(...n)) => { schema = type_to_schema(#<struct(...n)>);
			    args = n;
			  };
  | _ => TypeError("(in method object::unnest) Expected a collection of records",otype);
  #end;
  Type tp;
  #case type_of(path,schema)
  | `f(`t) => tp = t;
  | `t => TypeError("(in method object::unnest) Unnesting a non-collection type",t);
  #end;
  Type rtp = #<struct(bind(`(variable(var)),`tp),...args)>;
  Type itp = #<struct(bind(`(variable(var)),`tp))>;
  Object res = new object(#<list(`rtp)>);
  for(list<Object>* s = info.Collection->reverse(); s->consp(); s=s->tl)
  { Object inner = s->hd->expression(path,env);
    if (inner->nullp())
    {  Object o = new object(rtp,s->hd->info.Aggregation.Components->cons(new object()));
       if (keep)
	  res->info.Collection = res->info.Collection->cons(o);
    }
    else if (inner->collectionp())
    {  list<Object>* filtered_inner = Onil;
       Environment new_env = env->extend(object_to_env(s->hd,#<struct(...args)>));
       for(list<Object>* r = inner->info.Collection; r->consp(); r=r->tl)
          if ((new object(itp,Onil->cons(r->hd)))->predicate(pred,new_env))
	     filtered_inner = filtered_inner->cons(r->hd);
       if (keep && filtered_inner->nullp())
          res->info.Collection = res->info.Collection->cons(
		new object(rtp,s->hd->info.Aggregation.Components->cons(new object())));
       for(list<Object>* k = filtered_inner; k->consp(); k=k->tl)
          res->info.Collection = res->info.Collection->cons(
		new object(rtp,s->hd->info.Aggregation.Components->cons(k->hd)));
     }
    else evaluation_error("(in method object::unnest) Expected a collection",inner);
  };
  return res;
} DEBUG(object::unnest,cout << this << endl << var << " " << path << " " << pred << " " << keep << endl << env << endl)
};



/* map the function f (of parameter v) over the collection object */
Object object::map ( String var, Expr f, Environment &env ) {
try {
  Schema schema;
  #case otype
  | list(`s) => schema = type_to_schema(s);
  | _ => evaluation_error("(in method object::map) Expected a collection",this);
  #end;
  Type tp = schema_to_type(schema->extend(var,type_of(f,schema)));
  Object res = new object(#<list(`tp)>);
  for(list<Object>* s = info.Collection->reverse(); s->consp(); s=s->tl)
     if (s->hd->nullp());
     else { Object v = s->hd->expression(f,env);
	    if (!v->nullp())
	       res->info.Collection = res->info.Collection->cons(
			new object(tp,s->hd->info.Aggregation.Components->cons(v)));
          };
  return res;
} DEBUG(object::map,cout << this << endl << var << " " << f << endl)
};


/* reduces a collection to a value or collection */
Object object::reduce ( String accumulator, String var, Expr head, Expr pred, Environment& env ) {
try {
  Schema schema;
  #case otype
  | list(`s) => schema = type_to_schema(s);
  | _ => evaluation_error("(in method object::reduce) Expected a collection",this);
  #end;
  Type tp = type_of(head,schema);
  if (accumulator->eq("list") || accumulator->eq("bag") || accumulator->eq("set"))
  { Object res = new object(#<list(`tp)>);
    for(list<Object>* s = info.Collection->reverse(); s->consp(); s=s->tl)
       if (s->hd->nullp());
       else if (s->hd->predicate(pred,env))
       { Object v = s->hd->expression(head,env);
	 if (!v->nullp())
	    res->info.Collection = res->info.Collection->cons(v);
       };
    return res;
  }
  else if (accumulator->eq("sum") || accumulator->eq("min") || accumulator->eq("max"))
  { int (*acc) (int,int) = accumulator->eq("sum") ? sum : (accumulator->eq("min") ? min : max);
    int zero = accumulator->eq("sum") ? 0 : (accumulator->eq("min") ? 32000 : 0);
    Object res = new object(zero);
    for(list<Object>* s = info.Collection->reverse(); s->consp(); s=s->tl)
       if (s->hd->nullp());
       else if (s->hd->predicate(pred,env))
       { Object v = s->hd->expression(head,env);
	 if (v->integerp())
	    res->info.Integer = (*acc)(res->info.Integer,v->info.Integer);
	 else if (!v->nullp())
	    evaluation_error("(in method object::reduce) Expected an integer",v);
       };
    return res;
  }
  else if (accumulator->eq("all") || accumulator->eq("some"))
  { bool (*acc) (bool,bool) = accumulator->eq("all") ? and : or;
    bool zero = accumulator->eq("all") ? true : false;
    Object res = new object((bool) zero);
    for(list<Object>* s = info.Collection->reverse(); s->consp(); s=s->tl)
       if (s->hd->nullp());
       else if (!s->hd->nullp() && s->hd->predicate(pred,env))
       { Object v = s->hd->expression(head,env);
	 if (v->boolp())
	    res->info.Boolean = (*acc)(res->info.Boolean,v->info.Boolean);
	 else if (!v->nullp())
	    evaluation_error("(in method object::reduce) Expected a boolean",v);
       };
    return res;
  };
  TypeError("(in method object::reduce) Invalid monoid",variable(accumulator));
} DEBUG(object::reduce,cout << this << endl << accumulator << " " << var << " " << head << " " << pred << endl)
};


/* group by groupby_vars; the input collection object must be sorted by groupby_vars;
   keep also the other_nestvars (from other nestings) */
Object object::group_by ( String accumulator, String var, Expr head, Names groupby_vars, Names other_nestvars, Expr pred, Environment& env ) {
  return nest(accumulator,var,head,groupby_vars,other_nestvars,pred,env);
};


/* the NF2 nest operator */
Object object::nest ( String accumulator, String var, Expr head, Names groupby_vars, Names other_nestvars, Expr pred, Environment& env ) {
try {
  Names gvars = groupby_vars->append(other_nestvars);
  if (!collectionp())
     evaluation_error("(in method object::group_by) Expected a collection",this);
  Schema schema;
  #case otype
  | `f(`x) => schema = type_to_schema(x);
  | _ => TypeError("(in method object::group_by) Expected a collection",otype);
  #end;
  Schema sch = new binding<Type>;
  for(Names r = gvars->reverse(); r->consp(); r=r->tl)
     if (schema->in(r->hd))
	sch = sch->extend(r->hd,schema->find(r->hd));
     else TypeError("(in method object::group_by) The nesting attribute is not in the scope",
		    variable(r->hd));
  Type nesting_type = type_of(head,schema);
  Type tp;
  if (accumulator->eq("list") || accumulator->eq("bag") || accumulator->eq("set"))
     tp = #<list(`nesting_type)>;
  else tp = nesting_type;
  Type rtp = schema_to_type(sch->extend(var,tp));
  Object res = new object(#<list(`rtp)>);
  for(list<Object>* r = info.Collection->reverse(); r->consp(); r=r->tl)
  {  bool found = false;
     for(list<Object>* s = res->info.Collection; s->consp() && !found; s=s->tl)
     {  if (r->hd->equal_to(s->hd,groupby_vars))
	{  found = true;
	   s->hd->info.Aggregation.Components->hd->info.Collection
		 = s->hd->info.Aggregation.Components->hd->info.Collection->append(Cons(r->hd,Onil));
        };
     };
     if (!found)
     {  Object o = new object(otype);
	o->info.Collection = o->info.Collection->cons(r->hd);
	Object co = new object(rtp,r->hd->projection(gvars)->info.Aggregation.Components->cons(o));
	res->info.Collection = res->info.Collection->cons(co);
     };
  };
  for(list<Object>* r = res->info.Collection; r->consp(); r=r->tl)
  {  r->hd->info.Aggregation.Components->hd				/* destructive */
		= r->hd->info.Aggregation.Components->hd->reduce(accumulator,var,head,#<and()>,env);
     r->hd->info.Aggregation.Components->hd->otype = tp;		/* destructive */
  };
  return res;
} DEBUG(object::nest,cout << this << endl << accumulator << " " << var << " " << head << " " << groupby_vars << " " << other_nestvars << " " << pred << endl << env << endl)
};


/* nested-loop join */
Object object::nested_loop ( Object r, Expr pred, Expr keep, Environment& env ) {
try {
  if (!collectionp() || !r->collectionp())
     evaluation_error("(in method object::nested_loop) Expected collections",this);
  bool left_outerp = keep->eq(#<left>);		/* is it a left outer-join? */
  bool right_outerp = keep->eq(#<right>);	/* is it a right outer-join? */
  list<Object>* left_nulls;			/* nulls for the left tuple */
  list<Object>* right_nulls;			/* nulls for the right tuple */
  list<Expr>* args;
  #case otype
  | list(struct(...m))
	=> #case r->otype
	   | list(struct(...n))
		 => { args = m->append(n);
		      left_nulls = ntimes(m->length(),new object());
		      right_nulls = ntimes(n->length(),new object());
		    };
	   | _ => TypeError("(in method object::nested_loop) Expected a collection type",r->otype);
	   #end;
  | _ => TypeError("(in method object::nested_loop) Expected a collection type",otype);
  #end;
  Type rtp = #<struct(...args)>;
  Object res = new object(#<list(`rtp)>);
  if (right_outerp)		/* it's a right outer-join */
     for(list<Object>* right = r->info.Collection->reverse(); right->consp(); right=right->tl)
     {  int c = 0;
	for(list<Object>* left = info.Collection->reverse(); left->consp(); left=left->tl)
	{  Object o = new object(rtp,left->hd->info.Aggregation.Components->append(right->hd->info.Aggregation.Components));
	   if (o->predicate(pred,env))
	   {  res->info.Collection = res->info.Collection->cons(o);
	      c++;
	   };
	};
	if (c == 0)
	   res->info.Collection = res->info.Collection->cons(
					new object(rtp,left_nulls->append(right->hd->info.Aggregation.Components)));
     }
  else  for(list<Object>* left = info.Collection->reverse(); left->consp(); left=left->tl)
	{  int c = 0;
	   for(list<Object>* right = r->info.Collection->reverse(); right->consp(); right=right->tl)
	   {  Object o = new object(rtp,left->hd->info.Aggregation.Components->append(right->hd->info.Aggregation.Components));
	      if (o->predicate(pred,env))
	      {  res->info.Collection = res->info.Collection->cons(o);
		c++;
	      };
	   };
	   if (c == 0 && left_outerp)		/* it's a left outer-join */
	      res->info.Collection = res->info.Collection->cons(
					new object(rtp,left->hd->info.Aggregation.Components->append(right_nulls)));
	};
  return res;
} DEBUG(object::nested_loop,cout << this << endl << r << endl << pred << endl << env << endl)
};


/* construct the join predicate for sorting and merging; use the requested orders
*  of the two inputs for the equality tests */
Expr make_pred_for_merge ( const char* comparison, list<Expr>* left_order, list<Expr>* right_order, Schema sch ) {
  list<Expr>* res = Nil;
  list<Expr>* right = right_order;
  for(list<Expr>* left = left_order; left->consp() && right->consp(); left=left->tl, right=right->tl)
  {  /* due to the lack of polymorphism, I had to check types to select the right comparison function */
     Expr cmp = (type_of(left->hd,sch)->eq(#<String>)) ? variable(new string(form("s%s",comparison)))
						       : variable(new string(comparison));
     res = res->cons(#<`cmp(`(left->hd),`(right->hd))>);
  };
  return #<and(...res)>;
};


/* sort a collection object by the attributes */
Object object::sort ( list<Expr>* attributes, Environment& env ) {
try {
  if (!collectionp())
     evaluation_error("(in method object::sort) Expected a collection",this);
  Names ns;
  Type tp;
  #case otype
  | list(`x) => { tp = x;
		  ns = type_to_schema(x)->names();
		};
  | _ => TypeError("(in method object::sort) Expected a collection type",otype);
  #end;
  String X = new_name();
  String Y = new_name();
  list<Expr>* left = attributes;
  list<Expr>* right = attributes;
  for(Names r = ns; r->consp(); r=r->tl) 
     left = subst_list(variable(r->hd),#<project(`(variable(X)),`(variable(r->hd)))>,left);
  for(Names r = ns; r->consp(); r=r->tl) 
     right = subst_list(variable(r->hd),#<project(`(variable(Y)),`(variable(r->hd)))>,right);
  Expr pred = make_pred_for_merge("lt",left,right,(new binding<Type>)->extend(X,tp)->extend(Y,tp));
  Object res = new object(otype);
  int len = info.Collection->length();
  Object A[len];
  int k = 0;
  for(list<Object>* r = info.Collection; r->consp() && k<len; r=r->tl)
     A[k++] = r->hd;
  /* bubble sort */
  for(int j = len-1; j>=1; j--)
     for(int i = 1; i<=j; i++)
     {  Environment new_env = env->extend(X,A[i])->extend(Y,A[i-1]);
	if (predicate(pred,new_env))
           { Object e = A[i-1];
      	     A[i-1] = A[i];
	     A[i] = e;
      	   };
     };
  for(int i = len-1; i>=0; i--)
     res->info.Collection = res->info.Collection->cons(A[i]);
  return res;
} DEBUG(object::sort,cout << this << endl << attributes << endl << env << endl);
};


/* sort-merge join: it scans the sorted left and right inputs simultaneously */
Object object::merge_join ( Object r, Expr pred, list<Expr>* left_order, list<Expr>* right_order, Expr keep, Environment& env ) {
try {
  if (!collectionp() || !r->collectionp())
     evaluation_error("(in method object::merge_join) Expected collections",this);
  list<Expr>* args = Nil;
  bool left_outerp = keep->eq(#<left>);		/* is it a left outer-join? */
  bool right_outerp = keep->eq(#<right>);	/* is it a right outer-join? */
  list<Object>* left_nulls;			/* nulls for the left tuple */
  list<Object>* right_nulls;			/* nulls for the right tuple */
  #case otype
  | list(struct(...m))
	=> #case r->otype
	   | list(struct(...n))
		 => { args = m->append(n);
		      left_nulls = ntimes(m->length(),new object());
		      right_nulls = ntimes(n->length(),new object());
		    };
	   | _ => TypeError("(in method object::merge_join) Expected a collection type",r->otype);
	   #end;
  | _ => TypeError("(in method object::merge_join) Expected a collection type",otype);
  #end;
  Type rtp = #<struct(...args)>;
  Object res = new object(#<list(`rtp)>);
  Schema sch = type_to_schema(rtp);
  Expr greater_than = make_pred_for_merge("gt",left_order,right_order,sch);
  Expr less_than = make_pred_for_merge("lt",left_order,right_order,sch);
  list<Object>* left = info.Collection;
  list<Object>* right = r->info.Collection;
  list<Object>* rd;
  while(left->consp() && right->consp())
  { Object o = new object(rtp,left->hd->info.Aggregation.Components->append(right->hd->info.Aggregation.Components));
    if (o->predicate(greater_than,env))
    {  if (right_outerp)
       {  Object on = new object(rtp,left_nulls->append(right->hd->info.Aggregation.Components));
	  res->info.Collection = res->info.Collection->cons(on);
       };
       right = right->tl;
    }
    else if (o->predicate(less_than,env))
	 {  if (left_outerp)
	    {  Object on = new object(rtp,left->hd->info.Aggregation.Components->append(right_nulls));
	       res->info.Collection = res->info.Collection->cons(on);
	    };
            left = left->tl;
	 }
    else { do { rd = right;
		do { o = new object(rtp,left->hd->info.Aggregation.Components->append(rd->hd->info.Aggregation.Components));
		     if (o->predicate(less_than,env)) break;
		     if (o->predicate(pred,env))
			res->info.Collection = res->info.Collection->cons(o);
		     else if (right_outerp)
			res->info.Collection = res->info.Collection->cons(new object(rtp,
							left_nulls->append(left->hd->info.Aggregation.Components)));
		     else if (left_outerp)
		        res->info.Collection = res->info.Collection->cons(new object(rtp,
							left->hd->info.Aggregation.Components->append(right_nulls)));
		     rd = rd->tl;
		} while (rd->consp());
		left = left->tl;
	   } while (left->consp() && rd->consp() && o->predicate(greater_than,env));
	   right = rd;
	 };
  };
  if (left_outerp)
     while(left->consp())
     { Object o = new object(rtp,left->hd->info.Aggregation.Components->append(right_nulls));
       res->info.Collection = res->info.Collection->cons(o);
       left = left->tl;
     };
  if (right_outerp)
     while(right->consp())
     { Object o = new object(rtp,left_nulls->append(right->hd->info.Aggregation.Components));
       res->info.Collection = res->info.Collection->cons(o);
       right = right->tl;
     };
  res->info.Collection = res->info.Collection->reverse();
  return res;
} DEBUG(object::merge_join,cout << this << endl << r << endl << pred << endl << left_order << " " << right_order << endl << env << endl)
};


Names attributes ( Expr e ) {
  Names res = new list<String>;
  for(list<Expr>* r = e->arguments()->reverse(); r->consp(); r=r->tl)
     res = res->cons(r->hd->name());
  return res;
};

String monoid_name ( Expr M ) {
  #case M
  | ALPHA(`x,`m) => return m->name();
  | _ => return M->name();
  #end;
};


/* evaluate a plan */
Object evaluate_plan ( Expr plan, Environment& env ) {
try {
  Object res;
  #case plan
  | TABLE_SCAN(`extent,`v,`pred)
	=> res = find_extent(extent->name())->scan(v->name(),pred);
  | REDUCE(`M,`e,`v,`head,`pred)
	=> res = evaluate_plan(e,env)->reduce(monoid_name(M),v->name(),head,pred,env);
  | NESTED_LOOP(`x,`y,`pred,`keep)
	=> res = evaluate_plan(x,env)->nested_loop(evaluate_plan(y,env),pred,keep,env);
  | MERGE_JOIN(`x,`y,`pred,`keep,`left_order,`right_order)
	=> res = evaluate_plan(x,env)->merge_join(evaluate_plan(y,env),pred,left_order->arguments(),
					          right_order->arguments(),keep,env);
  | SORT(`e,`order)
	=> res = evaluate_plan(e,env)->sort(order->arguments(),env);
  | UNNEST(`e,`v,`path,`pred,`keep)
	=> res = evaluate_plan(e,env)->unnest(v->name(),path,pred,keep->eq(#<true>),env);
  | NEST(`M,`e,`v,`head,`group,`nests,`pred)
	=> res = evaluate_plan(e,env)->nest(monoid_name(M),v->name(),head,attributes(group),attributes(nests),pred,env);
  | GROUP_BY(`M,`e,`v,`head,`group,`nests,`pred)
	=> res = evaluate_plan(e,env)->group_by(monoid_name(M),v->name(),head,attributes(group),attributes(nests),pred,env);
  | MAP( `e, `v, `f )
	=> res = evaluate_plan(e,env)->map(v->name(),f,env);
  | define(`tp,`x,`u)
	 => { if (env->in(x->name()))
	         *(env->find(x->name())) = *((new object())->expression(u,env));  /* destructive */
	      else env = env->extend(x->name(),(new object())->expression(u,env));
	      res = new object();
	    };
  | _ => res = (new object())->expression(plan,env);
  #end;
  if (trace_evaluation)
  {  cout << "The result of: ";
     plan->pretty_print(15);
     cout << "\nHas type: ";
     res->otype->pretty_print(10);
     cout << "\nResult: ";
     res->reify(0)->pretty_print(8);
     cout << "\n--------------------------------------------------------------------------------\n";
  };
  return res;
} DEBUG(evaluate_plan,cout << plan << endl << env << endl)
};


/* convert an object to an expr */
Expr object::reify ( int level ) {
  if (level > max_level) return variable(new string("..."));
  else if (nullp()) return #<NULL>;
  else if (integerp()) return integer(info.Integer);
  else if (boolp()) return (info.Boolean ? #<true> : #<false>);
  else if (stringp()) return estring(info.Estring);
  else if (collectionp())
  { list<Expr>* res = Nil;
    for(list<Object>* r = info.Collection; r->consp(); r=r->tl)
       res = res->cons(r->hd->reify(level+1));
    return #<collection(...(res->reverse()))>;
  }
  else if (aggregationp())
  { list<Expr>* res = Nil;
    Names n = type_to_schema(otype)->names();
    for(list<Object>* r = info.Aggregation.Components; r->consp() && n->consp(); r=r->tl, n=n->tl)
       res = res->cons(#<bind(`(variable(n->hd)),`(r->hd->reify(level+1)))>);
    return #<struct(`(integer(info.Aggregation.OID)),...(res->reverse()))>;
  };
};


/* display an object */
ostream& operator<< ( ostream& s, Object o ) {
  return s << o->reify(0);
};


Object object_read ( ifstream& from ) {
  Object res = new object();
  short tag;
  char c;
  from >> tag;
  (int) res->tag = tag;
  from.get(c);
  res->otype = read(from);
  if (res->integerp())
     from >> res->info.Integer;
  else if (res->boolp())
     from >> res->info.Boolean;
  else if (res->stringp())
     res->info.Estring = reads(from);
  else if (res->collectionp())
  { int len;
    from >> len;
    res->info.Collection = Onil;
    for(int i=0; i<len; i++)
       res->info.Collection = res->info.Collection->cons(object_read(from));
    res->info.Collection = res->info.Collection->reverse();
  }
  else if (res->aggregationp())
  { int len;
    from >> res->info.Aggregation.OID >> len;
    res->info.Aggregation.Components = Onil;
    for(int i=0; i<len; i++)
       res->info.Aggregation.Components = res->info.Aggregation.Components->cons(object_read(from));
    res->info.Aggregation.Components = res->info.Aggregation.Components->reverse();
  };
  return res;
};


object::object_write ( ofstream& to ) {
  to << " " << tag;
  otype->write(to);
  if (integerp())
     to << " " << info.Integer;
  else if (boolp())
     to << " " << info.Boolean;
  else if (stringp())
     info.Estring->write(to);
  else if (collectionp())
  { to << " " << info.Collection->length();
    for(list<Object>* r = info.Collection; r->consp(); r=r->tl)
       r->hd->object_write(to);
  }
  else if (aggregationp())
  { to << " " << info.Aggregation.OID << " " << info.Aggregation.Components->length();
    for(list<Object>* r = info.Aggregation.Components; r->consp(); r=r->tl)
       r->hd->object_write(to);
  };
};


/*********************************************************************************/


database::database () {
  content = new binding<Extent>;
  schema = new binding<Interface>;
  extents = new binding<String>;
  varspace = new binding<Object>;
  vartypes = new binding<Type>;
};


/* read a database from a file */
database::database ( String file_name ) {
  ifstream from (file_name->value());
  content = new binding<Extent>;
  int exn;
  from >> exn;
  for(int i = 0; i<exn; i++)
  { Extent ext;
    ext = extent_read(from);
    content = content->extend(ext->name,ext);
  };
  schema = new binding<Interface>;
  int schemas;
  from >> schemas;
  for(int i = 0; i<schemas; i++)
  {  Interface inf = interface_read(from);
     schema = schema->extend(inf->name,inf);
  };
  extents = new binding<String>;
  from >> exn;
  for(int i = 0; i<exn; i++)
     extents = extents->extend(reads(from),reads(from));
  for(Names r = schema->names(); r->consp(); r=r->tl)
  {  Interface I = schema->find(r->hd);
     if (I->supertype)
	I->supertype = schema->find(I->supertype->name);
     for(list<Interface>* s = I->subtypes->reverse(); s->consp(); s=s->tl)
	s->hd = schema->find(s->hd->name);
     I->data = content->find(I->data->name);
  };
  vartypes = type_to_schema(read(from));
  int names;
  from >> names;
  varspace = new binding<Object>;
  for(int i = 0; i<names; i++)
  {  String s = reads(from);
     varspace = varspace->extend(s,object_read(from)); 
  };
  from.close();
};


/* write a database to a file */
database::close ( String file_name ) {
  ofstream to (file_name->value());
  to << " " << content->names()->length();
  for(Names r = content->names(); r->consp(); r=r->tl)
     content->find(r->hd)->extent_write(to);
  to << " " << schema->names()->length();
  for(Names r = schema->names(); r->consp(); r=r->tl)
     schema->find(r->hd)->interface_write(to);
  to << " " << extents->names()->length();
  for(Names r = extents->names(); r->consp(); r=r->tl)
  {  r->hd->write(to);
     extents->find(r->hd)->write(to);
  };
  schema_to_type(vartypes)->write(to);
  to << " " << varspace->names()->length();
  for(Names r = varspace->names(); r->consp(); r=r->tl)
  {  r->hd->write(to);
     varspace->find(r->hd)->object_write(to);
  };
  to.close();
};


/* print the entire database schema and content; long output */
ostream& operator<< ( ostream& s, database* db ) {
  for(Names r = db->schema->names(); r->consp(); r=r->tl)
     s << db->schema->find(r->hd) << endl;
  for(Names r = db->content->names(); r->consp(); r=r->tl)
     s << db->content->find(r->hd) << endl;
  return s;
};


/* find an extent in the database */
Extent find_extent ( String name ) {
  if (!DATABASE->content->in(name))
     TypeError("No such extent",variable(name));
  return DATABASE->content->find(name);
};


/* find an interface in the database */
Interface find_interface ( String name ) {
  if (!DATABASE->schema->in(name))
     TypeError("No such interface",variable(name));
  return DATABASE->schema->find(name);
};


Names all_extents () {
  return DATABASE->content->names();
};


binding<String>* extent_types () {
  return DATABASE->extents;
};


Names all_interfaces () {
  return DATABASE->schema->names();
};


class_def ( Expr e ) { Interface I = new interface(e); };