/********************************************************************************
*
* The optimizer for ODMG 2.0 OQL.
* The calculus and optimizer are described in the paper:
* L. Fegaras and D. Maier, "Optimizing Object Queries Using an Effective Calculus"
* (available at:  http://lambda.uta.edu/tods00.ps.gz)
*
* Copyright (c) 1999-2003 by Leonidas Fegaras, the University of Texas at
*     Arlington. All rights reserved.
* Programmer: Leonidas Fegaras
* Date: 4/9/97
*
*********************************************************************************/

#include <math.h>
#include "odl.h"
#include "typecheck.h"


/* set it to false if you don't want query unnesting */
bool query_unnesting = true;


/* set it to false if you don't want pointer joins (ie. no materialization of path expressions)	*/
bool pointer_joins = false;


/* set it to false if you don't want convertion of unnests to pointer joins)    */
bool unnests_to_joins = false;


/* set them to true for extensive tracing of normalization and unnesting */
const bool TRACE_NORMALIZATION = false;
const bool TRACE_UNNEST = false;


/* all global assignments */
static list<Expr>* global_variables = Nil;


/* does e generate a collection? -- defined in code_generator.gen */
bool bulk_operation ( Expr e );

/* Is attribute a key in table? -- defined in rule_based_opt.gen */
bool is_key ( Expr table, Expr attribute );

/* used in binding variables to values during normalization */
typedef binding<Expr>* Env;

Env empty_environment = new binding<Expr>;


double min ( double x, double y ) {
  if (x<y) return x;
  else return y;
};


/* catching exceptions */
#define DEBUG(exp) catch (odmg_error_class(m))		\
 { eout << "*** Found in " << exp << '\n';	\
   throw odmg_error_class(eout.str()); }


void warning ( string msg, Expr e ) {
  cerr << "\n*** Warning: " << msg << " (" << e << ")\n";
};


/* returns true if x is a subset of or equal to y */
bool subset ( list<Expr>* x, list<Expr>* y ) {
  for(list<Expr>* r = x; r->consp(); r=r->tl)
     if (!member(r->hd,y)) return false;
  return true;
};


/* check whether x and y are equal under variable renaming */
bool alpha_equivalent ( Expr x, Expr y, binding<Expr>* env ) {
  if (x->eq(y))
     return true;
  #case #<pair(`x,`y)>
  | pair(lambda(`vx,`bx),lambda(`vy,`by))
	=> return alpha_equivalent(bx,by,env->extend(vx->name(),vy));
  | pair(compr(`m,`hd1,...r),compr(`n,`hd2,...s))
	=> { for(; r->consp() && s->consp(); r=r->tl, s=s->tl)
		#case r->hd
		| iterate(`v,`e)
			=> #case s->hd
			   | iterate(`w,`u)
				=> if (alpha_equivalent(e,u,env))
				      env = env->extend(v->name(),w);
				   else return false;
			   | _ => return false;
			   #end;
		| _ => if (!alpha_equivalent(r->hd,s->hd,env))
			  return false;
		#end;
		// return (alpha_equivalent(m,n,env) && alpha_equivalent(hd1,hd2,env));
		return alpha_equivalent(hd1,hd2,env);
	    };
  | pair(get(_,`x,`v,...r),get(_,`y,`w,...s))
	=> alpha_equivalent(x,y,env)
	   && alpha_equivalent(#<f(...r)>,#<f(...s)>,env->extend(v->name(),w));
  | pair(unnest(_,`x,`v,...r),unnest(_,`y,`w,...s))
	=> alpha_equivalent(x,y,env)
	   && alpha_equivalent(#<f(...r)>,#<f(...s)>,env->extend(v->name(),w));
  | pair(nest(_,`x,`v,...r),nest(_,`y,`w,...s))
	=> alpha_equivalent(x,y,env)
	   && alpha_equivalent(#<f(...r)>,#<f(...s)>,env->extend(v->name(),w));
  | pair(`f(...r),`g(...s))
	=> { if (!alpha_equivalent(f,g,env))
		return false;
	     for(; r->consp() && s->consp(); r=r->tl, s=s->tl)
		if (!alpha_equivalent(r->hd,s->hd,env))
		   return false;
	     return r->nullp() && s->nullp();
	   };
  | _ : x->variablep()
      => if (env->in(x->name()))
	    return alpha_equivalent(env->find(x->name()),y,env);
	 else return x->eq(y);
  | _ => if (x->functionp() || y->functionp())
	    return false;
	 else return x->eq(y);
  #end;
};


/* is the collection monoid x at least as strong as the monoid y? */
bool no_weaker_monoid ( Expr x, Expr y) {
  #case x
  | list => return true;
  | bag => return !y->eq(#<list>);
  | set => return !y->eq(#<list>) && !y->eq(#<sum>) && !y->eq(#<prod>);
  | _ => return true;
  #end;
};


/* ... defined later */
Expr normalize ( Expr e, Env env );


/* collect and insert the predicates of a comprehension at its righttmost end */
Expr collect_predicates ( Expr e ) {
  #case e
  | compr(`m,`hd,...r)
	=> { list<Expr>* nr = Nil;
	     list<Expr>* pr = Nil;
	     m = collect_predicates(m);
	     for(list<Expr>* s = r; s->consp(); s=s->tl)
	     {  Expr u = collect_predicates(s->hd);
		#case u
		| iterate(_,_) => nr = nr->cons(u);
		| bind(_,_) => nr = nr->cons(u);
		| and(...l) => pr = pr->append(l);
		| _ => pr = pr->cons(u);
		#end;
	     };
	     nr = nr->reverse();
	     #case m
	     | some => #case collect_predicates(hd)
		       | and(...s)
			     => return #<compr(some,true,...nr,and(...pr,...s))>;
		      | false => return #<false>;
		       | `p  => return #<compr(some,true,...nr,and(...pr,`p))>;
		       #end;
	     | all => #case normalize(#<and(not(`(collect_predicates(hd))),...pr)>,empty_environment)
		      | and(...s)
			   => return #<compr(all,false,...nr,and(...s))>;
		      | false => return #<true>;
		      | `p => return #<compr(all,false,...nr,and(`p))>;
		      #end;
	     | _ => return #<compr(`m,`(collect_predicates(hd)),...nr,and(...pr))>;
	     #end;
	   };
  | `f(...r) => { list<Expr>* nr = Nil;
		  for(list<Expr>* s = r; s->consp(); s=s->tl)
		     nr = nr->cons(collect_predicates(s->hd));
		  nr = nr->reverse();
		  return #<`f(...nr)>;
		};
  | _ => return e;
  #end;
};


/* return the variables of a pattern */
list<Expr>* pattern_variables ( Expr pattern ) {
  #case pattern
  |  none => return Nil;
  |  pair(`a,`b) => return pattern_variables(a)->append(pattern_variables(b));
  |  `v => return Cons(v,Nil);
  #end;
};


/* return the variables in e that are not bound by some lambda binding or iterator */
list<Expr>* free_variables ( Expr e, list<Expr>* except ) {
  #case e
  |  project(`x,`a,...r) => return free_variables(x,except);
  |  method(`x,`a,...r) => return free_variables(x,except)->append(free_variables(#<f(...r)>,except));
  |  bind(`a,`x) => return free_variables(x,except);
  |  case(`x,...r) => return free_variables(x,except);
  |  lambda(`v,`e) => return free_variables(e,except->append(pattern_variables(v)));
  |  assign(...r,`e)
	=> { list<Expr>* res = Nil;
	     for(; r->consp(); r=r->tl)
		#case r->hd
		| bind(`v,`u,...s)
			=> { res = res->append(free_variables(u,except));
			     except = Cons(v,except);
			   };
		#end;
	     return res->append(free_variables(e,except));
	   };
  |  compr(`m,`hd,...r)
	=> { list<Expr>* l = except;
	     list<Expr>* s = Nil;
	     for(; r->consp(); r=r->tl)
		#case r->hd
		|  iterate(`v,`u) 
			=> { l = l->cons(v);
			     s = s->append(free_variables(u,l));
			   };
		|  bind(`v,`u)
			=> { l = l->cons(v);
			     s = s->append(free_variables(u,l));
			   };
		| `u => s = s->append(free_variables(u,l));
		#end;
	     return s->append(free_variables(hd,l));
	   };
  | `f(_,`x,`v,...r)
	: f->eq(#<get>) || f->eq(#<TABLE_SCAN>) || f->eq(#<INDEX_SCAN>)
	=> return free_variables(#<xx(`x,...r)>,Cons(v,except));
  | `f(_,`x,`v,...r)
	: f->eq(#<nest>) || f->eq(#<NEST>) || f->eq(#<reduce>) || f->eq(#<REDUCE>)
	  || f->eq(#<unnest>) || f->eq(#<UNNEST>) || f->eq(#<MAP>)
	=> return free_variables(#<xx(`x,...r)>,Cons(v,except));
  |  `f(...r) => { list<Expr>* s = Nil;
		   for(; r->consp(); r=r->tl)
		      s = s->append(free_variables(r->hd,except));
		   return s;
		 };
  |  `v => if (v->variablep()
	       && (v->name()->content()[0] == '_')
	       && !member(v,except)
	       && !member(v,global_variables)
	       && !extentp(v)
	       && find_scoped_variable(v->name())->eq(#<none>))
	      return Cons(v,Nil);
  #end;
  return Nil;
};


/*  returns true if pred refers to a variable in pattern */
bool refers_to ( Expr pred, Expr pattern ) {
  list<Expr>* fv = free_variables(pred,Nil);
  for(list<Expr>* r = pattern_variables(pattern); r->consp(); r=r->tl)
     if (member(r->hd,fv))
	return true;
  return false;
};


/* bind pattern variables to values; eg. pattern_binding of pair(pair(a,b),c)
*  to pair(pair(1,2),3) will bind a to 1, b to 2, and c to 3 */
binding<Expr>* pattern_bindings ( Expr pattern, Expr value ) {
  #case pattern
  | pair(`x,`y)
	=> #case value
	   | pair(`a,`b)
		=> return pattern_bindings(x,a)
				->extend(pattern_bindings(y,b));
	   | _ => return pattern_bindings(x,#<fst(`value)>)
				->extend(pattern_bindings(y,#<snd(`value)>));
	   #end;
  | _ => return (new binding<Expr>)->extend(pattern->name(),value);
  #end;
};


/* given a pattern, return a new pattern of the same shape but with fresh variables */
Expr new_patterns ( Expr pattern ) {
  #case pattern
  | pair(`x,`y) => return #<pair(`(new_patterns(x)),`(new_patterns(y)))>;
  | _ => return new_variable();
  #end;
};


/* ... defined below ... */
list<Expr>* rename_vars_list ( list<Expr>* terms, binding<Expr>* binds );


/* alpha coersion (renaming variables to avoid name conflicts) */
Expr rename_vars ( Expr term, binding<Expr>* binds ) {
  #case term
  | project(`e,`v,...r)
	=> return #<project(`(rename_vars(e,binds)),`v,...r)>;
  | bind(`v,`e)
	=> return #<bind(`v,`(rename_vars(e,binds)))>;
  | case(`e,...r)
	=> return #<case(`(rename_vars(e,binds)),...r)>;
  | lambda(`v,`e)
	=> { Expr nv = new_patterns(v);
	     return #<lambda(`nv,`(rename_vars(e,pattern_bindings(v,nv)->extend(binds))))>;
	   };
  | compr(`m,`hd,...r)
	=> { list<Expr>* q = Nil;
	     binding<Expr>* nbinds = binds;
	     for(list<Expr>* s = r; s->consp(); s=s->tl)
		#case s->hd
		| iterate(`v,`e)
			=> { Expr nv = new_patterns(v);
			     q = q->cons(#<iterate(`nv,`(rename_vars(e,nbinds)))>);
			     nbinds = pattern_bindings(v,nv)->extend(nbinds);
			   };
		| bind(`v,`e)
			=> { Expr nv = new_patterns(v);
			     q = q->cons(#<bind(`nv,`(rename_vars(e,nbinds)))>);
			     nbinds = pattern_bindings(v,nv)->extend(nbinds);
			   };
		| _ => q = q->cons(rename_vars(s->hd,nbinds));
		#end;
	     Expr nm = rename_vars(m,nbinds);
	     Expr nhd = rename_vars(hd,nbinds);
	     q = q->reverse();
	     return #<compr(`nm,`nhd,...q)>;
	   };
  | `f(...r)
	=> return function(rename_vars(f,binds),rename_vars_list(r,binds));
  | _ => if (term->variablep())
	    return binds->find(term->name(),term);
	 else return term;
  #end;
};


list<Expr>* rename_vars_list ( list<Expr>* terms, binding<Expr>* binds ) {
  list<Expr>* res = Nil;
  for(list<Expr>* r = terms; r->consp(); r=r->tl)
     res = res->cons(rename_vars(r->hd,binds));
  return res->reverse();
};


/* merge environments */
Env extend ( Env e1, Env e2 ) {
  return e2->extend(e1);
};


/********************************************************************************
* Program Normalization
********************************************************************************/


list<list<Expr>*>* normalize_qualifiers
	 ( Expr m, list<Expr>* initqs, Expr head, list<Expr>* qualifiers, Env env ) {
  list<Expr>* nquals = initqs;
  Env nenv = env;
  for(list<Expr>* r = qualifiers; r->consp(); r=r->tl)
     #case r->hd
     | iterate(`v,zero(`n))
	=> return new list<list<Expr>*>;
     | iterate(`v,unit(`n,`e))
	=> nenv = extend(pattern_bindings(v,normalize(e,nenv)),nenv);
     | iterate(`v,merge(`n,`x,`y))
	=> { if (!no_weaker_monoid(n,m))
	        warning("A weaker monoid in a comprehension generator may result into indeterminism",r->hd);
	     return normalize_qualifiers(m,nquals,head,r->tl->cons(#<iterate(`v,`x)>),nenv)
			 ->append(normalize_qualifiers(m,nquals,head,r->tl->cons(#<iterate(`v,`y)>),nenv));
	   };
     | iterate(`v,compr(`n,`hd,...s))
	=> r = Cons(r->hd,s->append(r->tl->cons(#<iterate(`v,unit(`n,`hd))>)));
     | bind(`v,`e)
	=> nenv = extend(pattern_bindings(v,normalize(e,nenv)),nenv);
     | true
	=> ;
     | false
	=> return new list<list<Expr>*>;
     | compr(some,`pred,...s)
	=> r = Cons(r->hd,s->append(r->tl->cons(pred)));
     | `e => nquals = nquals->cons(normalize(e,nenv));
     #end;
  return Cons(nquals->reverse()->cons(normalize(head,nenv))->cons(normalize(m,nenv)),
	      new list<list<Expr>*>);
};


Expr normalize_comprehension ( Expr e, Env env, bool &changed ) {
  #case e
  | compr(`m,`hd,...r)
	=> { Expr res;
	     list<list<Expr>*>* new_qualifiers = normalize_qualifiers(m,Nil,hd,r,env);
	     if (new_qualifiers->nullp())
		return #<zero(`m)>;
	     m = new_qualifiers->hd->hd;
	     if (new_qualifiers->hd->tl->tl->nullp())
		res = #<unit(`m,`(new_qualifiers->hd->tl->hd))>;
	     else res = #<compr(...(new_qualifiers->hd))>;
	     for(list<list<Expr>*>* s = new_qualifiers->tl; s->consp(); s=s->tl)
		if (s->hd->tl->tl->nullp())
		   res = #<merge(`m,`res,unit(...(s->hd)))>;
		else res = #<merge(`m,`res,compr(...(s->hd)))>;
	     changed = !e->eq(res);
	     return res;
	   };
  #end;
  return e;
};


/* collect the range variables of the qualifiers in a comprehension */
list<Expr>* range_variables ( list<Expr>* qualifiers ) {
  list<Expr>* res = Nil;
  for(list<Expr>* r = qualifiers; r->consp(); r=r->tl)
     #case r->hd
     | iterate(`v,`e) =>  res = res->cons(v);
     #end;
  return res;
};


/* Is e the name of a class with an extent? */
bool has_extent ( Expr e ) {
  return e->variablep() && !get_class_extent(e)->eq(#<none>);
};


/* materialization of path expression into pointer joins between extents */
Expr path_materialization ( Expr e, Env env, bool &changed ) {
  if (!pointer_joins)
     return e;
  #case e
  | compr(`m,`h,...r,`f[project(project(`x,`A,`Aclass),`B,`tp)],...s)
	: member(x,range_variables(r))		/* path expression in a qualifier */
	  && has_extent(Aclass)
	=> {  Expr extent = get_class_extent(Aclass);
	      Expr nv = new_variable();
	      list<Expr>* ns = subst_list(#<project(`x,`A,`Aclass)>,nv,
					  Cons(#<`f[project(`nv,`B,`tp)]>,s));
	      Expr nh = subst_expr(#<project(`x,`A,`Aclass)>,nv,h);
	      changed = true;
	      return #<compr(`m,`nh,...r,iterate(`nv,`extent),
				    eq(project(`x,`A,`Aclass),OID(`nv)),...ns)>;
	   };
  | compr(`m,`f[project(project(`x,`A,`Aclass),`B,`tp)],...r)
	: member(x,range_variables(r))		/* path expression in the comprehension head */
	  && has_extent(Aclass)
	=> {  Expr extent = get_class_extent(Aclass);
	      Expr nv = new_variable();
	      Expr nh = subst_expr(#<project(`x,`A,`Aclass)>,nv,
				   #<`f[project(`nv,`B,`tp)]>);
	      changed = true;
	      return #<compr(`m,`nh,...r,iterate(`nv,`extent),
				    eq(project(`x,`A,`Aclass),OID(`nv)))>;
	   };
  #end;
  return e;
};


bool inverse_relationship_is_variable ( Expr A, Expr Aclass ) {
  #case get_declaration_binding(A->name(),Aclass->name())
  |  relationship(`x,_,inverse(_,_))
	=> return x->variablep();
  |  _ => return false;
  #end;
};


/* can an unnest be a join? (see below) */
Expr unnest_can_be_join ( Expr A, Expr Aclass ) {
  if (!has_extent(Aclass))
     return #<none>;
  for(list< Ref<Catalog> >* r = get_declarations(A->name()); r->consp(); r=r->tl)
     #case get_declaration_binding(r->hd)
     |  relationship(`v,`rname,inverse(`inv_name,`inv_var))
	   : inv_name->eq(Aclass)
	     && inverse_relationship_is_variable(inv_var,inv_name)
	   => return inv_var;
     #end;
  return #<none>;
};


Expr class_of_inverse_relationship ( Expr A, Expr Aclass ) {
  #case get_declaration_binding(A->name(),Aclass->name())
  |  relationship(`x,_,inverse(_,_))
	=> return x;
  |  _ => return #<OID>;
  #end;
};


/* Some unnests can be translated into pointer joins: this is
*  possible when there is an one-to-many relationship for the unnesting
*  path and the class that corresponds to the 'many' part has an extent */
Expr unnest_to_join ( Expr e, Env env, bool &changed ) {
  if (!unnests_to_joins)
     return e;
  #case e
  | compr(`m,`h,...r,iterate(`v,project(`x,`A,`f(`Aclass))),...s)
	: !unnest_can_be_join(A,Aclass)->eq(#<none>)
	=> {  Expr rname = unnest_can_be_join(A,Aclass);
	      Expr extent = get_class_extent(Aclass);
	      Expr Bclass = class_of_inverse_relationship(rname,Aclass);
	      changed = true;
	      return #<compr(`m,`h,...r,iterate(`v,`extent),
				eq(project(`v,`rname,`Bclass),OID(`x)),...s)>;
	   };
  #end;
  return e;
};


bool pos_predicate ( Expr e, Expr nv ) {
  #case e
  | and(...r)
	=> for(; r->consp(); r=r->tl)
              if (pos_predicate(r->hd,nv))
                 return true;
  | `v : v->eq(nv) => return true;
  #end;
  return false;
};


bool is_key_of ( Expr e, Expr key ) {
  return e->variablep() && extentp(e) && is_key(extent_type(e->name()),key);
};


list<Expr>* place_bind_in_qualifiers ( Expr v, Expr e, list<Expr>* gl ) {
  list<list<Expr>*>* fv = new list<list<Expr>*>;
  list<Expr>* res = Nil;
  for(list<Expr>* r = gl; r->consp(); r=r->tl)
     #case r->hd
     | iterate(`v,`u) => fv = fv->cons(free_variables(u,Nil));
     | bind(`v,`u) => fv = fv->cons(free_variables(u,Nil));
     | `u => fv = fv->cons(free_variables(u,Nil));
     #end;
  fv = fv->reverse();
  bool found = false;
  for(list<Expr>* r = gl; r->consp(); r=r->tl, fv=fv->tl)
     if (!found && member(v,fv->hd))
     {  found = true;
        res = Cons(r->hd,Cons(#<bind(`v,`e)>,res));
     } else res = Cons(r->hd,res);
  if (!found)
     return Cons(#<bind(`v,`e)>,res->reverse());
  else return res->reverse();
};


/* Some programmers still use OQL like SQL! This function converts some joins into pointer
 * traversals using class relationships. This is a kind of inverse of unnest_to_join */
Expr relational_joins_to_relationships ( Expr e, Env env, bool &changed ) {
  static Expr nv = new_variable();
  #case e
  | compr(`m,`h,...r,iterate(`v1,`e1),...s,`f[eq(project(`e2,`a1,`t1),project(`v2,`a2,`t2))],...q)
	 :  a1->eq(a2) && v1->eq(v2) && pos_predicate(#<`f[`nv]>,nv) && is_key_of(e1,a1)
	=> { changed = true;
	     list<Expr>* n = place_bind_in_qualifiers(v1,e2,s);
	     return #<compr(`m,`h,...r,...n,`f[true],...q)>;
	   };
  | compr(`m,`h,...r,iterate(`v1,`e1),...s,`f[eq(project(`v2,`a2,`t2),project(`e2,`a1,`t1))],...q)
	 :  a1->eq(a2) && v1->eq(v2) && pos_predicate(#<`f[`nv]>,nv) && is_key_of(e1,a1)
	=> { changed = true;
	     list<Expr>* n = place_bind_in_qualifiers(v1,e2,s);
	     return #<compr(`m,`h,...r,...n,`f[true],...q)>;
	   };
  | compr(`m,`h,...r,iterate(`v1,`e1),...s,`f[eq(`e2,`v2)],...q)
	 :  v1->eq(v2) && e1->variablep() && pos_predicate(#<`f[`nv]>,nv) && extentp(e1)
	=> { changed = true;
	     list<Expr>* n = place_bind_in_qualifiers(v1,e2,s);
	     return #<compr(`m,`h,...r,...n,`f[true],...q)>;
	   };
  | compr(`m,`h,...r,iterate(`v1,`e1),...s,`f[eq(`v2,`e2)],...q)
	 :  v1->eq(v2) && e1->variablep() && pos_predicate(#<`f[`nv]>,nv) && extentp(e1)
	=> { changed = true;
	     list<Expr>* n = place_bind_in_qualifiers(v1,e2,s);
	     return #<compr(`m,`h,...r,...n,`f[true],...q)>;
	   };
  #end;
  return e;
};


bool without_assign ( Expr e ) {
  #case e
  | `f[assign(...r)] => return false;
  | _ => return true;
  #end;
};


/* Handling non-correlated nesting: pull out from a comprehension terms that are constant */
Expr eliminate_constant_subterms ( Expr e, Env env, bool &changed ) {
  static list<Expr>* vf = Nil->cons(#<compr>)->cons(#<call>);
  #case e
  | compr(...r,`f[`g(...s)],...t)
	: member(g,vf) && free_variables(#<`g(...s)>,Nil)->nullp() && without_assign(#<`f[x]>)
	=> {  changed = true;
	      Expr nv = new_variable();
	      global_variables = global_variables->cons(nv);
	      return #<assign(bind(`nv,`g(...s)),compr(...r,`f[`nv],...t))>;
	   };
  #end;
  return e;
};


Expr basic_reduction ( Expr e, Env env, bool &changed ) {
  bool b = changed;
  changed = true;
  #case e
  | unit(`m,`x)
	: !collectionp(m)
	=> return x;
  | zero(sum)
	=> return #<0>;
  | zero(max)
	=> return #<0>;
  | zero(all)
	=> return #<true>;
  | zero(some)
	=> return #<false>;
  | merge(sum,`x,`y)
	=> return #<plus(`x,`y)>;
  | merge(max,`x,`y)
	=> return #<max(`x,`y)>;
  | merge(all,`x,`y)
	=> return #<and(`x,`y)>;
  | merge(some,`x,`y)
	=> return #<or(`x,`y)>;
  | merge(_,zero(...r),`x)
	=> return x;
  | merge(_,`x,zero(...r))
	=> return x;
  | plus(`x,`y)
	: x->integerp() && y->integerp()
	=> return integer(x->info.Integer+y->info.Integer);
  | not(true) => return #<false>;
  | not(false) => return #<true>;
  | not(not(`x)) => return x;
  | not(compr(all,false,...r))
	=> return #<compr(some,true,...r)>;
  | not(compr(some,true,...r))
	=> return #<compr(all,false,...r)>;
  | not(and(...ps))
	=> {  list<Expr>* s = Nil;
	      for(list<Expr>* r = ps->reverse(); r->consp(); r=r->tl)
		 s = s->cons(#<not(`(r->hd))>);
	      return #<or(...s)>;
	   };
  | not(or(...ps))
	=> {  list<Expr>* s = Nil;
	      for(list<Expr>* r = ps->reverse(); r->consp(); r=r->tl)
		 s = s->cons(#<not(`(r->hd))>);
	      return #<and(...s)>;
	   };
  | not(`f(`x,`y))
	=> #case f
	   | eq => return #<neq(`x,`y)>;
	   | neq => return #<eq(`x,`y)>;
	   | lt => return #<geq(`x,`y)>;
	   | gt => return #<leq(`x,`y)>;
	   | leq => return #<gt(`x,`y)>;
	   | geq => return #<lt(`x,`y)>;
	   | _ => { changed = b;
		    return e;
		  };
	   #end;
  | and() => return #<true>;
  | or() => return #<false>;
  | and(...r,false,...s)
	=> return #<false>;
  | and(...r,true,...s)
	=> return #<and(...r,...s)>;
  | or(...r,true,...s)
	=> return #<true>;
  | or(...r,false,...s)
	=> return #<or(...r,...s)>;
  | if(true,`x,`y)
	=> return x;
  | if(false,`x,`y)
	=> return y;
  | compr(`m,`hd,`pred)
	: (!pred->functionp() || !pred->info.Function.Name->eq(#<iterate>))
	=> return #<if(`pred,unit(`m,`hd),zero(`m))>;
  | apply(`f,`e)
	=> #case normalize(f,env)
	   | lambda(`v,`b)
		=> return normalize(b,extend(pattern_bindings(v,e),env));
	   #end;
  | project(struct(...r),`a,...n)
	=> {  for(list<Expr>* s = r; s->consp(); s=s->tl)
	         #case s->hd
		 | bind(`b,`u)
			: a->eq(b)
			=> return u;
		 #end;
	      return #<null>;
	   };
  | project(union_construction(`f,`tag,_,`u,_,_),`a,...n)
	=> if (tag->eq(a))
	      return u;
	   else odmg_error("wrong union construction/projection",e);
  | project(union_construction(`f,_,`tag,_,`u,_,_),`a,...n)
	=> if (tag->eq(a))
	      return u;
	   else odmg_error("inconsistent union construction/projection",e);
  | cases(union_construction(`f,`tag,_,_,_,_),...cs)
 	=> { bool found = false;
	     for(list<Expr>* r=cs; !found && r->consp(); r=r->tl)
		#case r->hd
		| case(`e,...tags)
			: member(tag,tags) || member(#<default>,tags)
			=> return e;
		#end;
		if (!found)
		   odmg_error("Union case does not handle this attribute",e);
	   };
  | cases(union_construction(`f,_,`tag,_,_,_,_),...cs)
 	=> { bool found = false;
	     for(list<Expr>* r=cs; !found && r->consp(); r=r->tl)
		#case r->hd
		| case(`e,...tags)
			: member(tag,tags) || member(#<default>,tags)
			=> return e;
		#end;
		if (!found)
		   odmg_error("Union case does not handle this attribute",e);
	   };
  | fst(pair(`a,`b))
	=> return a;
  | snd(pair(`a,`b))
	=> return b;
  | compr(set,`hd,iterate(`v,`x))
	: hd->eq(v)
	=> return x;
  | `v : v->variablep()
	=> if (env->in(e->name()))
	      return env->find(e->name());
  #end;
  changed = b;
  return e;
};


/* normalize e using an environment; this is the rewrite engine of the optimizer */
Expr normalize1 ( Expr e, Env env ) {
try{
  if (TRACE_NORMALIZATION) { cout << "** normalize1: "; e->pretty_print(0,cout); cout << endl; };
  Expr res;
  #case e
  | `f(...r)
	=> { list<Expr>* nr = Nil;
	     for(list<Expr>* s = r; s->consp(); s=s->tl)
		nr = nr->cons(normalize1(s->hd,env));
	     nr = nr->reverse();
	     res = #<`f(...nr)>;
	   };
  | _ => res = e;
  #end;
  bool changed = false;
  res = basic_reduction(res,env,changed);
  res = normalize_comprehension(res,env,changed);
  res = relational_joins_to_relationships(res,env,changed);
  if (TRACE_NORMALIZATION) { cout << "--> "; res->pretty_print(0,cout); cout << endl; };
  if (changed)
     return normalize1(res,env);
  else return res;
} DEBUG(e);
};


Expr normalize2 ( Expr e, Env env ) {
try{
  if (TRACE_NORMALIZATION) { cout << "** normalize2: "; e->pretty_print(0,cout); cout << endl; };
  Expr res;
  #case e
  | `f(...r)
	=> { list<Expr>* nr = Nil;
	     for(list<Expr>* s = r; s->consp(); s=s->tl)
		nr = nr->cons(normalize2(s->hd,env));
	     nr = nr->reverse();
	     res = #<`f(...nr)>;
	   };
  | _ => res = e;
  #end;
  bool changed = false;
  res = path_materialization(res,env,changed);
  res = unnest_to_join(res,env,changed);
  res = eliminate_constant_subterms(res,env,changed);
  if (TRACE_NORMALIZATION) { cout << "--> "; res->pretty_print(0,cout); cout << endl; };
  if (changed)
     return normalize2(res,env);
  else return res;
} DEBUG(e);
};


Expr normalize ( Expr e, Env env ) {
  Expr res = normalize2(normalize1(e,env),env);
  return res;
};


/********************************************************************************
* Translation of comprehensions into algebraic operators
********************************************************************************/

typedef struct { Expr fst; Expr snd; Expr thrd; Expr frth; } qtuple;


/* splits the predicate pred into four parts: it returns the quadruple
* [and(...left),and(...right),and(...both),and(...rest)], where left holds all
* predicates that refer to the variables in the left_pattern
* exclusively; right holds all predicates that refer to the variables in
* the right_pattern exclusively; both holds all predicates that
* refer to the variables in both left and right patterns, and rest holds the rest */
qtuple split_predicate ( Expr pred, Expr left_pattern, Expr right_pattern ) {
  list<Expr>* lefts = Nil;
  list<Expr>* rights = Nil;
  list<Expr>* both = Nil;
  list<Expr>* rests = Nil;
  #case pred
  | and(...r)
      => for(; r->consp(); r=r->tl)
	 #case r->hd
	 | `f[compr(...s)] => rests = rests->cons(r->hd);
	 | and(...s)
		=> { qtuple t = split_predicate(r->hd,left_pattern,right_pattern);
		     lefts = lefts->append(t.fst->arguments());
		     rights = rights->append(t.snd->arguments());
		     both = both->append(t.thrd->arguments());
		     rests = rests->append(t.frth->arguments());
		   };
	 | _ => { list<Expr>* vars = free_variables(r->hd,Nil);
		  if (subset(vars,pattern_variables(left_pattern)))
		     lefts = lefts->cons(r->hd);
		  else if (subset(vars,pattern_variables(right_pattern)))
		     rights = rights->cons(r->hd);
		  else if (subset(vars,pattern_variables(left_pattern)
					->append(pattern_variables(right_pattern))))
		     both = both->cons(r->hd);
		  else rests = rests->cons(r->hd);
		};
	 #end;
	 qtuple x = { #<and(...lefts)>, #<and(...rights)>,
		      #<and(...both)>, #<and(...rests)> };
         return x;
  #end;
};


Expr bulk ( Expr m ) {
  if (m->eq(#<sum>) || m->eq(#<max>) || m->eq(#<min>) || m->eq(#<avg>))
     return #<bag>;
  if (m->eq(#<all>) || m->eq(#<some>))
     return #<set>;
  #case m
  | ordered(...r)
	=> return #<list>;
  | _ => return m;
  #end;
};


Expr promote_pred ( Expr e, Expr pat, Expr pred ) {
  #case e
  | get(`M,`table,`v,and(...s))
	=> return #<get(`M,`table,`v,and(...s,...(pred->arguments())))>;
  | join(`M,`x,`y,and(...s),`keep)
	=> #case pat
	   | pair(`px,`py)
		=> {  qtuple tp = split_predicate(pred,px,py);
		      Expr nx = promote_pred(x,px,tp.fst);
		      Expr ny = promote_pred(y,py,tp.snd);
		      list<Expr>* r = tp.thrd->arguments()->append(tp.frth->arguments());
		      return #<join(`M,`nx,`ny,and(...s,...r),`keep)>;
		   };
	   #end;
  | nest(`M,`x,`v,`hd,`vars,and(...s),`apred)
	=> #case pat
	   | pair(`pv,`px)
		=> {  qtuple tp = split_predicate(pred,pv,px);
		      Expr nx = promote_pred(x,px,tp.snd);
		      list<Expr>* r = tp.thrd->arguments()->append(tp.frth->arguments());
		      return #<nest(`M,`nx,`v,`hd,`vars,and(...s,...r),`apred)>;
		   };
	   #end;
  | reduce(`M,`x,`v,`hd,and(...s))
	=> #case pat
	   | pair(`pv,`px)
		=> {  qtuple tp = split_predicate(pred,pv,px);
		      Expr nx = promote_pred(x,px,tp.snd);
		      list<Expr>* r = tp.thrd->arguments()->append(tp.frth->arguments());
		      return #<reduce(`M,`nx,`v,`hd,and(...s,...r))>;
		   };
	   #end;
  | unnest(`M,`x,`v,`path,and(...s),`keep)
	=> #case pat
	   | pair(`pv,`px)
		=> {  qtuple tp = split_predicate(pred,pv,px);
		      Expr nx = promote_pred(x,px,tp.snd);
		      list<Expr>* r = tp.thrd->arguments()->append(tp.frth->arguments());
		      return #<unnest(`M,`nx,`v,`path,and(...s,...r),`keep)>;
		   };
	   #end;
  | _ => return #<filter(`e,`pred)>;		/* ??? */
  #end;
};


list<Expr>* project_vars ( Expr e ) {
  #case e
  | project(`x,_,_) => return project_vars(x);
  | OID(`x) => return project_vars(x);
  | `f(...r) => { list<Expr>* s = Nil;
		  for(list<Expr>* n = r->reverse(); n->consp(); n=n->tl)
		     s = s->append(project_vars(n->hd));
		  return s;
		};
  | _ => if (e->variablep())
	    return Cons(e,Nil);
	 else return Nil;
  #end;
};


binding<Expr>* group_bindings ( Expr e1, Expr e2, binding<Expr>* env ) {
  list<Expr>* r = project_vars(e1);
  list<Expr>* s = project_vars(e2);
  for(; r->consp() && s->consp(); r=r->tl, s=s->tl)
     if (!r->hd->eq(s->hd) && !env->in(r->hd->name()) && !env->in(s->hd->name()))
	env = env->extend(r->hd->name(),s->hd);
  return env;
};


/* The memo table is used for eliminating duplicate nested queries in a query */
list<pair<Expr,Expr>*>* memo_table = new list<pair<Expr,Expr>*>;


void memoize ( Expr v, Expr e ) {
  memo_table = (memo_table)->cons(new pair<Expr,Expr>(v,e));
};


Expr find_in_memo ( Expr e ) {
  for(list<pair<Expr,Expr>*>* r = memo_table; r->consp(); r=r->tl)
     if (alpha_equivalent(r->hd->second,e,new binding<Expr>))
	return r->hd->first;
  return #<none>;
};


/* alpha-renaming using an environment */
Expr env_subst ( Expr e, binding<Expr>* env ) {
  #case e
  | `f(...r)
	=> {  list<Expr>* s = Nil;
	      for(; r->consp(); r=r->tl)
		 s = s->cons(env_subst(r->hd,env));
	      s = s->reverse();
	      return #<`f(...s)>;
	   };
  | `v : v->variablep()
	=> if (env->in(v->name()))
	      return env->find(v->name());
	   else return e;
  | _ => return e;
  #end;
};


binding<Expr>* zip_range_vars ( list<Expr>* r, list<Expr>* s ) {
  binding<Expr>* env = new binding<Expr>;
  for(; r->consp() && s->consp(); r=r->tl, s=s->tl)
     #case #<pair(`(r->hd),`(s->hd))>
     | pair(iterate(`x,`a),iterate(`y,`b))
	   => if (alpha_equivalent(a,b,env))
		 env = env->extend(x->name(),y);
	      else return new binding<Expr>;
     | pair(iterate(`x,`a),_)
	=> return new binding<Expr>;
     | pair(_,iterate(`x,`a))
	=> return new binding<Expr>;
     #end;
  for(; r->consp(); r=r->tl)
     #case r->hd
     | iterate(_,_) => return new binding<Expr>;
     #end;
  for(; s->consp(); s=s->tl)
     #case s->hd
     | iterate(_,_) => return new binding<Expr>;
     #end;
  return env;
};


list<Expr>* candidate_for_group_by ( Expr pred, binding<Expr>* env ) {
  #case pred
  | and(...r)
	=> {  list<Expr>* s = Nil;
	      for(; r->consp(); r=r->tl)
		 #case r->hd
		 | eq(`e1,`e2)
			: !e1->variablep() && !e2->variablep()
			=> if (alpha_equivalent(e1,e2,group_bindings(e1,e2,env)))
			      s = Cons(e2,s);
			   else if (alpha_equivalent(e2,e1,group_bindings(e2,e1,env)))
			      s = Cons(e1,s);
		#end;
	      return s;
	   };
  | _ => return Nil;
  #end;
};


list<Expr>* rest_of_groupby ( list<Expr>* r, list<Expr>* groups, list<Expr>* preds, binding<Expr>* env ) {
  list<Expr>* res = Nil;
  for(; r->consp(); r=r->tl)
     #case r->hd
     | eq(`x,`y)
	 : member(x,groups) || member(y,groups)
	 => ;
     | `e : member(env_subst(e,env),preds)
	 => ;
     | `e => res = res->cons(env_subst(e,env));
     #end;
  return res->reverse();
};


Expr last ( list<Expr>* r ) {
  if (r->tl->nullp())
     return r->hd;
  else return last(r->tl);
};


Expr groupby_vars ( list<Expr>* r ) {
  if (r->tl->nullp())
     return r->hd;
  else return #<pair(`(groupby_vars(r->tl)),`(r->hd))>;
};


/* Translating and unnesting nested comprehensions */
Expr translate ( Expr e, Expr u, Expr w, Expr v, Expr E ) {
  #case e
  | compr(...r)
	: TRACE_UNNEST
	=> {  cout << "** unnest: "; e->pretty_print(11,cout);
	      cout << "\nu / w / v: " << u << " / " << w << " / " << v;
	      cout << "\n        E: "; E->pretty_print(11,cout); cout << endl;
	   };
  #end;
  Expr res = #<none>;
  static Expr nv = new_variable();
  #case e
  /* Shortcut to unnesting: the following two rules recognize calculus forms generated by OQL group-bys */
  | compr(`m,`hd1,...rr,`f[compr(`n,`hd2,...ss)])
	: query_unnesting
	  && zip_range_vars(ss,rr)->length() > 0
	  && candidate_for_group_by(last(ss),zip_range_vars(ss,rr))->consp()
	  && free_variables(#<`f[`nv]>,Cons(nv,range_variables(rr)))->nullp()
	=> {  binding<Expr>* env = zip_range_vars(ss,rr);
	      Expr pred2 = last(ss);
	      list<Expr>* gs = candidate_for_group_by(pred2,env);
	      Expr vars = groupby_vars(gs->append(pattern_variables(u)));
	      Expr nhd = env_subst(hd2,env);
	      list<Expr>* rest = rest_of_groupby(last(ss)->arguments(),gs,#<`f[X]>->arguments(),env);
	      Expr ne = #<group_by(`n,`nhd,`vars,...rest)>;
	      res = translate( #<compr(`m,`hd1,...rr,`f[`ne])>, u, w, v, E );
	   };
  | compr(`m,`f[compr(`n,`hd2,...ss)],...rr)
	: query_unnesting
	  && zip_range_vars(ss,rr)->length() > 0
	  && candidate_for_group_by(last(ss),zip_range_vars(ss,rr))->consp()
	  && free_variables(#<`f[`nv]>,Cons(nv,range_variables(rr)))->nullp()
	=> {  binding<Expr>* env = zip_range_vars(ss,rr);
	      Expr pred2 = last(ss);
	      list<Expr>* gs = candidate_for_group_by(pred2,env);
	      Expr vars = groupby_vars(gs->append(pattern_variables(u)));
	      Expr nhd = env_subst(hd2,env);
	      list<Expr>* rest = rest_of_groupby(last(ss)->arguments(),gs,last(rr)->arguments(),env);
	      Expr ne = #<group_by(`n,`nhd,`vars,...rest)>;
	      res = translate( #<compr(`m,`f[`ne],...rr)>, u, w, v, E );
	   };
  | compr(`m,`hd1,`pred)
	=> #case pred
 	   | `p[compr(`n,`hd2,...rr,and(...ss))]	/* nested comprehension in a predicate */
		=> if (query_unnesting)
		   { Expr pp = #<compr(`n,`hd2,...rr,and(...ss))>;
		     Expr nv = find_in_memo(pp);
		     if (!nv->eq(#<none>))
			res = translate( #<compr(`m,`hd1,`p[`nv])>, u, w, v, E );
		     else { nv = new_variable();
			    memoize(nv,pp);
			    res = translate( #<compr(`m,`hd1,`p[`nv])>, u, #<pair(`w,`nv)>, v,
				      translate( #<compr(`n,`hd2,...rr,and(...ss))>, w, w, nv, E ) );
			  };
		   } else {
		      Expr np = translate(#<compr(`n,`hd2,...rr,and(...ss))>,#<none>,#<none>,new_variable(),#<none>);
		      res = translate( #<compr(`m,`hd1,`p[`np])>, u, w, v, E );
		   };
	   | _ => #case hd1
		  | `f[compr(`n,`hd2,...r)]	/* nested comprehension in the head */
			=> if (query_unnesting)
			   { Expr pp = #<compr(`n,`hd2,...r)>;
			     Expr nv = find_in_memo(pp);
			     if (!nv->eq(#<none>))
				res = translate( #<compr(`m,`f[`nv],`pred)>, u, w, v, E );
			     else { nv = new_variable();
				    memoize(nv,pp);
				    res = translate( #<compr(`m,`f[`nv],`pred)>, u, #<pair(`w,`nv)>, v,
						     translate( #<compr(`n,`hd2,...r)>, w, w, nv, E ) );
				  };
			   } else {
			     Expr np = translate(#<compr(`n,`hd2,...r)>,#<none>,#<none>,new_variable(),#<none>);
			     res = translate( #<compr(`m,`f[`np],`pred)>, u, w, v, E );
			   };
		  | _ => #case m
			 | `f[compr(`n,`hd2,...r)]	/* nested comprehension in the accumulator (for order-by) */
				=> if (query_unnesting)
				   { Expr pp = #<compr(`n,`hd2,...r)>;
				     Expr nv = find_in_memo(pp);
				     if (!nv->eq(#<none>))
					res = translate( #<compr(`f[`nv],`hd1,`pred)>, u, w, v, E );
				     else { nv = new_variable();
					    memoize(nv,pp);
					    res = translate( #<compr(`f[`nv],`hd1,`pred)>, u, #<pair(`w,`nv)>, v,
							     translate( #<compr(`n,`hd2,...r)>, w, w, nv, E ) );
					  };
				   } else {
				     Expr np = translate(#<compr(`n,`hd2,...r)>,#<none>,#<none>,new_variable(),#<none>);
				     res = translate( #<compr(`f[`np],`hd1,`pred)>, u, w, v, E );
				   };
			 | _ => {  #case m
				   |  ordered(...r)
					=> {  m = #<list>;
					      E = #<sort(`E,order(...r))>;
					   };
				   #end;
				   if (u->eq(#<none>))
				      res = #<reduce(`m,`E,`v,`hd1,`pred)>;
				   else res = #<nest(`m,`E,`v,`hd1,`u,`pred,and())>;
				};
			 #end;
		  #end;
	   #end;
  | compr(`m,`hd1,iterate(`a,`f[compr(`n,`hd2,...r)]),...s,`p)
	/* nested comprehension in the generator domain */
	=> if (query_unnesting)
	   {  Expr pp = #<compr(`n,`hd2,...r)>;
	      Expr nv = find_in_memo(pp);
	      if (!nv->eq(#<none>))
		 res = translate( #<compr(`m,`hd1,iterate(`a,`f[`nv]),...s,`p)>, u, w, v, E );
	      else { nv = new_variable();
		     memoize(nv,pp);
		     res = translate( #<compr(`m,`hd1,iterate(`a,`f[`nv]),...s,`p)>, u, #<pair(`w,`nv)>, v,
				      translate( #<compr(`n,`hd2,...r)>, w, w, nv, E ) );
		   };
	   } else {
	      Expr nd = translate(#<compr(`n,`hd2,...r)>,#<none>,#<none>,new_variable(),#<none>);
	      res = translate( #<compr(`m,`hd1,iterate(`a,`f[`nd]),...s,`p)>, u, w, v, E );
	   };
  | compr(`m,`hd,iterate(`a,`X),...r,`p)
	=> if (u->eq(#<none>))
	   { qtuple tp = split_predicate(p,w,a);
             Expr newE = (tp.fst->arguments()->nullp()) ? E : promote_pred(E,w,tp.fst);
	     Expr np = #<and(...(tp.snd->arguments()),...(tp.thrd->arguments()))>;
	     if (w->eq(#<none>))
		if (true || X->variablep())
		   res = translate( #<compr(`m,`hd,...r,`(tp.frth))>, u, a, v,
			 	    #<get(`(bulk(m)),`X,`a,and(...(tp.fst->arguments()),...(tp.snd->arguments()),...(tp.thrd->arguments())))> );
	        else res = translate( #<compr(`m,`hd,...r,`(tp.frth))>, u, a, v,
			 	      #<unnest( `(bulk(m)), `newE, `a, `X, `np, none )> );
	     else if (X->variablep() && !member(X,pattern_variables(w)))
	     {  Expr x = #<get(`(bulk(m)),`X,`a,`(tp.snd))>;
	        res = translate( #<compr(`m,`hd,...r,`(tp.frth))>, u, #<pair(`w,`a)>, v,
				 #<join( `(bulk(m)), `newE, `x, `(tp.thrd), none )> );
	     }
	     else res = translate( #<compr(`m,`hd,...r,`(tp.frth))>, u, #<pair(`w,`a)>, v,
			 	   #<unnest( `(bulk(m)), `newE, `a, `X, `np, none )> );
	   }
	   else { qtuple tp = split_predicate(p,w,a);
		  Expr np = #<and(...(tp.snd->arguments()),...(tp.thrd->arguments()))>;
		  Expr newE = (tp.fst->arguments()->nullp()) ? E : promote_pred(E,w,tp.fst);
		  if (X->variablep())
		     res = translate( #<compr(`m,`hd,...r,`(tp.frth))>, u, #<pair(`w,`a)>, v,
					      #<join( `(bulk(m)), `newE, get(`(bulk(m)),`X,`a,`(tp.snd)),
						      `(tp.thrd), `u )> );
		  else res = translate( #<compr(`m,`hd,...r,`(tp.frth))>, u, #<pair(`w,`a)>, v,
					        #<unnest( `(bulk(m)), `newE, `a,`X, `np, `u)> );
	   };
  | merge(`m,`x,`y)
	=> { Expr nx = translate(x,u,w,v,E);
	     Expr ny = translate(y,u,w,v,E);
	     res = #<merge(`m,`nx,`ny)>;
	   };
  | unit(`m,`x)
	=> { Expr nx = translate(x,u,w,v,E);
	     res = #<unit(`m,`nx)>;
	   };
  | zero(`m)
	=> res = #<zero(`m)>;
  | bind(`n,`x)
	=> { Expr nx = translate(x,u,w,v,E);
	     res = #<bind(`n,`nx)>;
	   };
  | `c(`n,...r)
	: c->eq(#<construct>) || c->eq(#<persistent_construct>) || c->eq(#<union_construction>)
	=> {  list<Expr>* s = Nil;
	      for(; r->consp(); r=r->tl)
		 s = s->cons(translate(r->hd,u,w,v,E));
	      s = s->reverse();
	      res = #<`c(`n,...s)>;
	   };

  | _ : bulk_operation(e)
      => {  Expr v = new_variable();
	    Expr w = new_variable();
	    res = #<reduce(set,get(set,`e,`v,and()),`w,`v,and())>;
	 }
  | `f(...r)
	=> {  list<Expr>* s = Nil;
	      for(; r->consp(); r=r->tl)
		 s = s->cons(translate(r->hd,u,w,v,E));
	      s = s->reverse();
	      res = #<`f(...s)>;
	   };
  | _ => if (e->variablep() && extentp(e))
	 {  Expr v = new_variable();
	    Expr w = new_variable();
	    res = #<reduce(set,get(set,`e,`v,and()),`w,`v,and())>;
	 }
	 else if (e->variablep())
	    #case find_scoped_variable(e->name())
	    | none => res = e;
	    | `f(`tp) : collectionp(f)
		=> {  Expr v = new_variable();
		      Expr w = new_variable();
		      res = #<reduce(`f,get(`f,`e,`v,and()),`w,`v,and())>;
		   };
	    | _ => res = e;
	    #end
	 else res = e;
  #end;
  #case e
  | compr(...r)
	: TRACE_UNNEST
	=> { cout << "--> "; res->pretty_print(0,cout); cout << endl; };
  #end;
  return res;
};



/********************************************************************************
* Shortcut to group-by unnesting (remove groupby's generated by the unnesting algorithm)
********************************************************************************/


/* if e is a path, return the beginning of the path */
Expr project_variable ( Expr e ) {
  #case e
  | project(`x,_,_) => return project_variable(x);
  | OID(`x) => return project_variable(x);
  | _ => if (e->variablep())
	    return e;
	 else return #<none>;
  #end;
};


binding<Expr>* zip_vars ( list<Expr>* x, list<Expr>* y ) {
  binding<Expr>* env = new binding<Expr>;
  list<Expr>* r = x;
  list<Expr>* s = y;
  for(; r->consp() && s->consp(); r=r->tl, s=s->tl)
     env = env->extend(new binding<Expr>(r->hd->name(),s->hd));
  if (r->consp() || s->consp())
     return new binding<Expr>;
  return env;
};


bool alpha_nest_equiv ( Expr x, Expr y, binding<Expr>* env );


bool alpha_nest_equiv_list ( list<Expr>* xl, Expr y, binding<Expr>* env ) {
  for(list<Expr>* r = xl; r->consp(); r=r->tl)
     if (alpha_nest_equiv(r->hd,y,env))
        return true;
  return false;
};


bool alpha_nest_equiv ( Expr x, Expr y, binding<Expr>* env ) {
  if (x->eq(y))
     return true;
  #case #<pair(`x,`y)>
  | pair(lambda(`vx,`bx),lambda(`vy,`by))
	=> return alpha_nest_equiv(bx,by,env->extend(vx->name(),vy));
  | pair(get(_,`x,`v,...r),get(_,`y,`w,...s))
	=> alpha_nest_equiv(x,y,env)
	   && alpha_nest_equiv(#<f(...r)>,#<f(...s)>,env->extend(v->name(),w));
  | pair(unnest(_,`x,`v,...r),unnest(_,`y,`w,...s))
	=> alpha_nest_equiv(x,y,env)
	   && alpha_nest_equiv(#<f(...r)>,#<f(...s)>,env->extend(v->name(),w));
  | pair(nest(_,`x,`v,...r),nest(_,`y,`w,...s))
	=> alpha_nest_equiv(x,y,env)
	   && alpha_nest_equiv(#<f(...r)>,#<f(...s)>,env->extend(v->name(),w));
  | pair(`f(...r),`g(...s))
	=> { if (!alpha_nest_equiv(f,g,env))
		return false;
	     for(; s->consp(); s=s->tl)
		if (!alpha_nest_equiv_list(r,s->hd,env))
		   return false;
	     return true;
	   };
  | _ : x->variablep()
      => if (env->in(x->name()))
	    return alpha_nest_equiv(env->find(x->name()),y,env);
	 else return x->eq(y);
  | _ => if (x->functionp() || y->functionp())
	    return false;
	 else return x->eq(y);
  #end;
};


Expr pass_gvar ( Expr e, Expr var ) {
  #case e
  | nest(`m,`x,`v,`hd,`gv,...r)
	=> if (v->eq(var))
	      return e;
	   else if (member(var,pattern_variables(gv)))
	      return e;
	   else return #<nest(`m,`(pass_gvar(x,var)),`v,`hd,pair(`gv,`var),...r)>;
  | unnest(`m,`x,`v,`path,`pred,`keep)
	=> if (v->eq(var))
	      return e;
	   else if (member(var,pattern_variables(keep)))
	      return e;
	   else return #<unnest(`m,`(pass_gvar(x,var)),`v,`path,`pred,pair(`keep,`var))>;
  | join(`m,`x,`y,`pred,`keep)
	=> if (member(var,pattern_variables(keep)))
	      return e;
	   else return #<join(`m,`(pass_gvar(x,var)),`y,`pred,pair(`keep,`var))>;
  | _ => return e;
  #end;
};


list<Expr>* range_vars ( Expr e );


Expr convert_groupby ( Expr e ) {
  #case e
  | get(`m,`u,`v,`f[group_by(`n,`hd,`vars,...rest)])
	=> {  Expr nv = find_in_memo(#<group_by(`n,`hd,`vars,...rest)>);
	      if (!nv->eq(#<none>))
              {  u = pass_gvar(u,nv);
		 return convert_groupby(#<get(`m,`u,`v,`f[`nv])>);
	      };
	      nv = new_variable();
	      memoize(nv,#<group_by(`n,`hd,`vars,...rest)>);
	      Expr res = convert_groupby(#<nest(`n,get(`m,`u,`v,and()),`nv,`hd,pair(`vars,`v),and(...rest),`f[`nv])>);
	      return res;
	   };
  | nest(`m,`u,`v,`f[group_by(`n,`hd,`vars,...rest)],`vars1,`bpred,`apred)
	=> {  u = convert_groupby(u);
	      Expr nv = find_in_memo(#<group_by(`n,`hd,`vars,...rest)>);
	      if (!nv->eq(#<none>))
              {  u = pass_gvar(u,nv);
		 return convert_groupby(#<nest(`m,`u,`v,`f[`nv],`vars1,`bpred,`apred)>);
	      };
	      nv = new_variable();
	      memoize(nv,#<group_by(`n,`hd,`vars,...rest)>);
	      return convert_groupby(#<nest(`m,nest(`n,`u,`nv,`hd,`vars,and(...rest),and()),`v,`f[`nv],`vars1,`bpred,`apred)>);
	   };
  | nest(`m,`u,`v,`hd1,`vars1,`bpred,`f[group_by(`n,`hd,`vars,...rest)])
	=> {  u = convert_groupby(u);
	      Expr nv = find_in_memo(#<group_by(`n,`hd,`vars,...rest)>);
	      if (!nv->eq(#<none>))
              {  u = pass_gvar(u,nv);
		 return convert_groupby(#<nest(`m,`u,`v,`hd1,`vars1,`bpred,`f[`nv])>);
	      };
	      nv = new_variable();
	      memoize(nv,#<group_by(`n,`hd,`vars,...rest)>);
	      return convert_groupby(#<nest(`m,nest(`n,`u,`nv,`hd,`vars,and(...rest),and()),`v,`hd1,`vars1,`bpred,`f[`nv])>);
	   };
  | reduce(`m,`u,`v,`f[group_by(`n,`hd,`vars,...rest)],`pred)
	=> {  u = convert_groupby(u);
	      Expr nv = find_in_memo(#<group_by(`n,`hd,`vars,...rest)>);
	      if (!nv->eq(#<none>))
              {  u = pass_gvar(u,nv);
		 return convert_groupby(#<reduce(`m,`u,`v,`f[`nv],`pred)>);
	      };
	      nv = new_variable();
	      memoize(nv,#<group_by(`n,`hd,`vars,...rest)>);
	      return convert_groupby(#<reduce(`m,nest(`n,`u,`nv,`hd,`vars,and(...rest),and()),`v,`f[`nv],`pred)>);
	   };
  | reduce(`m,`u,`v,`hd1,`f[group_by(`n,`hd,`vars,...rest)])
	=> {  u = convert_groupby(u);
	      Expr nv = find_in_memo(#<group_by(`n,`hd,`vars,...rest)>);
	      if (!nv->eq(#<none>))
              {  u = pass_gvar(u,nv);
		 return convert_groupby(#<reduce(`m,`u,`v,`hd1,`f[`nv])>);
	      };
	      nv = new_variable();
	      memoize(nv,#<group_by(`n,`hd,`vars,...rest)>);
	      return convert_groupby(#<reduce(`m,nest(`n,`u,`nv,`hd,`vars,and(...rest),and()),`v,`hd1,`f[`nv])>);
	   };
  | nest(`m,join(`n,`x,`y,`pred,`v),`nv,`hd,`gv,`pb,`pa)
	: query_unnesting
	  && alpha_nest_equiv(x,y,new binding<Expr>)
	  && subset(pattern_variables(gv),range_vars(x))
	  && candidate_for_group_by(pred,zip_vars(range_vars(y),range_vars(x)))->consp()
	=> {  binding<Expr>* env = zip_vars(range_vars(y),range_vars(x));
	      list<Expr>* gs = candidate_for_group_by(pred,env);
	      Expr vars = #<pair(`(groupby_vars(gs)),`gv)>;
	      Expr nhd = env_subst(hd,env);
	      list<Expr>* rest = rest_of_groupby(pred->arguments(),gs,pb->arguments(),env);
	      Expr ne = #<nest(`m,`x,`nv,`nhd,`vars,and(...rest),`pa)>;
	      return convert_groupby(ne);
	   };
  | `f(...r)
	=> {  list<Expr>* s = Nil;
	      for(; r->consp(); r=r->tl)
		 s = s->cons(convert_groupby(r->hd));
	      s = s->reverse();
	      return #<`f(...s)>;
	   };
  | _ => return e;
  #end;
};


/********************************************************************************
* Putting everything together
********************************************************************************/

/* optimize an OQL query */
Expr optimize_query (Expr query) {
  memo_table = new list<pair<Expr,Expr>*>;
  global_variables = Nil;
  Expr e = rename_vars(query,new binding<Expr>);
  Expr u = collect_predicates(normalize(e,empty_environment));
  /* type_of(u); checking the consistency of the calculus form */
  Expr t = translate(rename_vars(u,new binding<Expr>),#<none>,#<none>,new_variable(),#<none>);
  t = convert_groupby(t);
  type_of(t); /* checking the consistency of the algebraic form */
  return t;
};