/********************************************************************************
*
* The rule-based optimizer for ODMG 2.0 OQL.
* It generates physical plans from algebraic forms using a rule-based rewriting engine
*
* 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"

Expr project_variable ( Expr e );
float selectivity ( Expr pred );
int table_cardinality ( Expr table );
int table_size ( Expr table );
list<Expr>* range_vars ( Expr e );
list<Expr>* pattern_variables ( Expr pattern );
Expr best_tree ( Expr e );
list<Expr>* free_variables ( Expr e, list<Expr>* except );


double ilog ( int i ) { return (i>0) ? log(i) : 0; };


/* all operator names */
list<Expr>* algebraic_operators
   = Nil->cons(#<get>)->cons(#<sort>)->cons(#<join>)->cons(#<nest>)->cons(#<unnest>)->cons(#<reduce>)->cons(#<merge>)->cons(#<unit>)->cons(#<zero>);


/* all_plans holds the names of all physical plans */
list<Expr>* all_plans
   = Nil->cons(#<TABLE_SCAN>)->cons(#<INDEX_SCAN>)->cons(#<SORT>)->cons(#<NESTED_LOOP>)->cons(#<BLOCK_NESTED_LOOP>)
        ->cons(#<MERGE_JOIN>)->cons(#<INDEXED_LOOP>)->cons(#<MAP>)->cons(#<REDUCE>)
	->cons(#<MERGE>)->cons(#<GROUP_BY>)->cons(#<NEST>)->cons(#<UNNEST>);

/* true if x is a plan */
bool is_plan ( Expr x ) {
  #case x
  |  `f(...r) => return member(f,all_plans);
  |  _ => return false;
  #end;
};


list<Expr>* set_union ( list<Expr>* x, list<Expr>* y ) {
  list<Expr>* res = x;
  for(list<Expr>* r = y; r->consp(); r=r->tl)
     if (!(member(r->hd,res))) res = res->cons(r->hd);
  return res;
};


/*  returns the attributes of the range variables in tables
*   that participate in the predicate pred */
list<Expr>* get_order ( Expr pred, list<Expr>* tables ) {
  #case pred
  |  OID(`x) => if (member(x,tables))
		   return Nil->cons(pred);
		else return get_order(x,tables);
  |  project(`x,`a,_)
	=> if (member(project_variable(x),tables))
	      return Nil->cons(pred);
	   else return Nil;
  |  `f(...r)
	=> {  list<Expr>* res = Nil;
	      for(; r->consp(); r=r->tl)
		  res = set_union(res,get_order(r->hd,tables));
	      return res;
	   };
  |  _ => if (pred->variablep() && member(pred,tables)) 
	     return Nil->cons(pred);
	  else return Nil;
  #end;
};


/* true if there is an equality predicate that binds an attribute of a table
*  in left_tables with an attribute of a table in right_tables.
*  This will imply that the join between these tables will be an equijoin */
bool equijoinp ( Expr pred, list<Expr>* left_tables, list<Expr>* right_tables ) {
  #case pred
  | and(...r,eq(`a,`b),...s)
	: ((member(project_variable(a),left_tables) && member(project_variable(b),right_tables))
	   || (member(project_variable(a),right_tables) && member(project_variable(b),left_tables)))
	=> return true;
  | _ => return false;
  #end;
};


/* finds an equality predicate from pred that associates attributes of the tables
*  in left_tables with attributes of tables in right_tables (if there is one).
*  Then it returns the attributes of the tables in the left_table that are involved
*  in this predicate. These attributes determine the order in which the left input
*  of a merge join must be sorted */
Expr required_order ( Expr pred, list<Expr>* left_tables, list<Expr>* right_tables ) {
  #case pred
  | and(...r,eq(`a,`b),...s)
	: member(project_variable(a),left_tables) && member(project_variable(b),right_tables)
	=> return #<order(`a)>; 
  | and(...r,eq(`a,`b),...s)
	: member(project_variable(a),right_tables) && member(project_variable(b),left_tables)
	=> return #<order(`b)>; 
  | _ => return #<order()>;
  #end;
};


Expr is_left_key ( Expr x, Expr y, Expr xorder, Expr yorder ) {
  #case xorder
  | order(...r,OID(_),...s)
	=> return #<true>;
  #end;
  #case yorder
  | order(...r,OID(_),...s)
	=> return #<false>;
  #end;
  /* need also to test if either xorder or yorder is a key for x or y */
  return #<none>;
};


/* true if attribute is a key of the class table */
bool is_key ( Expr table, Expr attribute ) {
  if (table->variablep() && classp(table->name()))
     return member(attribute,class_keys(get_declaration_binding(table->name())));
  else return false;
};


bool is_range_var ( Expr v, Expr x ) {
  #case x
  | `f[`h(_,_,`n,...r)]
	: v->eq(n) && (h->eq(#<TABLE_SCAN>) || h->eq(#<INDEX_SCAN>))
	=> return true;
  | _ => return false;
  #end;
};


/* true if the expected order is at least as strong as the expected order.
*  expected_order = computed_order plus something */
bool subsumes ( Expr computed_order, Expr expected_order ) {
  #case computed_order
  |  `f(...co)
	=> #case expected_order
	   |  `f(...eo)
		 => { list<Expr>* s = eo;
		      for(list<Expr>* r = co; r->consp(); r=r->tl, s=s->tl)
		         if (s->nullp())
			    return false;
		         else if (!r->hd->eq(s->hd) && !r->hd->eq(#<OID(`(s->hd))>)
						    && !s->hd->eq(#<OID(`(r->hd))>))
				 return false;
		      return true;
		    };
	   |  _ => return false;
	   #end;
  |  _ => return false;
  #end;
};


Expr index_order ( Expr class_name, Expr range_variable, list<Expr>* index_attributes ) {
  list<Expr>* ps = Nil;
  Schema sch = new binding<Expr>(range_variable->name(),class_name);
  for(list<Expr>* r = index_attributes->reverse(); r->consp(); r=r->tl)
  {  Expr e = #<project(`range_variable,`(r->hd))>;
     Expr tp = type_of(e,sch);
     ps = ps->cons(#<project(`range_variable,`(r->hd),`tp)>);
  };
  return #<order(...ps)>;
};


Expr find_index ( Expr table, Expr range_variable, Expr order ) {
  Expr cname = extent_type(table->name());
  #case get_declaration_binding(new string("_class_indexes"),cname->name());
  |  index(...s)
	=> for(; s->consp(); s=s->tl)
	      #case get_declaration_binding(s->hd->name())
	      | index(`index_name,`class_name,...attrs)
		   : subsumes(order,index_order(cname,range_variable,attrs))
		=> return #<index(`index_name,`(index_order(cname,range_variable,attrs)))>;
	      #end;
  #end;
  return #<none>;
};


/* the cost of accessing an index to satisfy a required order io and a predicate pred */
float index_access_cost ( Expr table, Expr name, Expr pred, list<Expr>* io ) {
  float size = table_size(table)+10;
  if (io->length() > 1)
  {  size = 3.0;
     for(list<Expr>* s = io; s->consp(); s=s->tl)
	#case pred
	| and(...m,eq(project(`x,`a,`c),`v),...n)
		: x->eq(name) && s->hd->eq(#<project(`x,`a,`c)>) && !member(name,free_variables(v,Nil))
		=> size = size*0.7;
	| and(...m,eq(`v,project(`x,`a,`c)),...n)
		: x->eq(name) && s->hd->eq(#<project(`x,`a,`c)>) && !member(name,free_variables(v,Nil))
		=> size = size*0.7;
	| _ => return table_size(table)+10;
	#end;
     return size;
  };
  for(list<Expr>* r = pred->arguments(); r->consp(); r=r->tl)
     #case r->hd
     |  eq(project(`x,`a,`c),`v)
	: x->eq(name) && member(#<project(`x,`a,`c)>,io)
	=> size = 3.0;
     |  eq(`v,project(`x,`a,`c))
	: x->eq(name) && member(#<project(`x,`a,`c)>,io)
	=> size = 3.0;
     |  `f(project(`x,`a,`c),`v)
	: x->eq(name) && member(#<project(`x,`a,`c)>,io) && (f->eq(#<lt>) || f->eq(#<gt>) || f->eq(#<leq>) || f->eq(#<geq>))
	=> size = size/2.0;
     |  `f(`v,project(`x,`a,`c))
	: x->eq(name) && member(#<project(`x,`a,`c)>,io) && (f->eq(#<lt>) || f->eq(#<gt>) || f->eq(#<leq>) || f->eq(#<geq>))
	=> size = size/2.0;
     | _ => size = size*1.5;
     #end;
  return size;
};


/* returns all indexes attached to table ranged by the range variable name */
list<Expr>* all_table_indexes ( Expr table, Expr name ) {
  list<Expr>* res = Nil;
  Expr cname = extent_type(table->name());
  #case get_declaration_binding(new string("_class_indexes"),cname->name())
  |  index(...s)
	=> for(; s->consp(); s=s->tl)
              #case get_declaration_binding(s->hd->name())
	      | index(`index_name,`class_name,...attrs)
		   => res = res->cons(#<index(`index_name,`(index_order(cname,name,attrs)))>);
	      #end;
  #end;
  return res;
};


pair<Expr,Expr>* find_index_preds ( Expr pred, list<Expr>* attrbs ) {
  Expr bottom = #<none>;
  Expr top = #<none>;
  if (attrbs->length() > 1)
  {  for(list<Expr>* s = attrbs; s->consp(); s=s->tl)
	#case pred
	| and(...m,eq(project(`w,`a,`tp),`value),...n)
		: s->hd->eq(#<project(`w,`a,`tp)>) && !member(s->hd->arguments()->hd,free_variables(value,Nil))
		=> top = (top->eq(#<none>)) ? value : #<pair(`top,`value)>;
	| and(...m,eq(`value,project(`w,`a,`tp)),...n)
		: s->hd->eq(#<project(`w,`a,`tp)>) && !member(s->hd->arguments()->hd,free_variables(value,Nil))
		=> top = (top->eq(#<none>)) ? value : #<pair(`top,`value)>;
	| _ => return Pair(bottom,top);
	#end;
     return Pair(top,top);
  };
  for(list<Expr>* r = pred->arguments(); r->consp(); r=r->tl)
     #case r->hd
     | eq(project(`w,`a,`tp),`value)
	   : member(#<project(`w,`a,`tp)>,attrbs) && !member(w,free_variables(value,Nil))
	   => top = bottom = value;
     | eq(`value,project(`w,`a,`tp))
	   : member(#<project(`w,`a,`tp)>,attrbs) && !member(w,free_variables(value,Nil))
	   => top = bottom = value;
     | `f(project(`w,`a,`tp),`value)
	   : member(#<project(`w,`a,`tp)>,attrbs) && (f->eq(#<lt>) || f->eq(#<gt>) || f->eq(#<leq>) || f->eq(#<geq>))
	     && !member(w,free_variables(value,Nil))
	   => if (f->eq(#<gt>) || f->eq(#<geq>))
		 bottom = value;
	      else top = value;
     | `f(`value,project(`w,`a,`tp))
	   : member(#<project(`w,`a,`tp)>,attrbs) && (f->eq(#<lt>) || f->eq(#<gt>) || f->eq(#<leq>) || f->eq(#<geq>))
	     && !member(w,free_variables(value,Nil))
	   => if (f->eq(#<gt>) || f->eq(#<geq>))
		 top = value;
	      else bottom = value;
     #end;
  return Pair(bottom,top);
};


Expr make_index_scan ( Expr index_name, list<Expr>* attrbs, Expr m, Expr table, Expr v, Expr pred ) {
  pair<Expr,Expr>* bt = find_index_preds(pred,attrbs);
  Expr bottom = bt->first;
  Expr top = bt->second;
  return #<INDEX_SCAN(`m,`table,`v,`pred,`index_name,`bottom,`top)>;
};


Expr applicable_index ( Expr index, Expr table, Expr v, Expr pred ) {
  Expr cname = extent_type(table->name());
  #case get_declaration_binding(index->name(),cname->name())
  | index(`index_name,`class_name,`attr)
	=> #case pred
	   | and(...r,eq(project(`w,`a,`tp),`path),...s)
		: w->eq(v) && a->eq(attr)
		=> return #<order(`path)>;
	   | and(...r,eq(`path,project(`w,`a,`tp)),...s)
		: w->eq(v) && a->eq(attr)
		=> return #<order(`path)>;
	   #end;
  | _ => cout << "******error\n";
  #end;
  return #<none>;
};


list<Expr>* nest_groups ( Expr e ) {
  #case e
  | `f(_,_,`v,...r)
	: f->eq(#<get>) || f->eq(#<TABLE_SCAN>) || f->eq(#<INDEX_SCAN>)
	=> return Cons(v,Nil);
  | `f(_,`x,`y,...r)
	: f->eq(#<join>) || f->eq(#<NESTED_LOOP>) || f->eq(#<INDEXED_LOOP>) || f->eq(#<MERGE_JOIN>)
	=> return nest_groups(x)->append(nest_groups(y));
  | `f(_,`x,`v,...r)
	: f->eq(#<unnest>) || f->eq(#<UNNEST>) || f->eq(#<MAP>)
	=> return nest_groups(x)->append(Cons(v,Nil));
  | `f(_,`x,`v,_,`gv,...r)
	: f->eq(#<nest>) || f->eq(#<NEST>)
	=> return pattern_variables(gv)->append(Cons(#<error>,Nil));
  | `f(`x,...r)
	: f->eq(#<sort>) || f->eq(#<SORT>)
	=> return Cons(#<error>,Nil);
  | _ => return Nil;
  #end;
};


Expr nest_order ( Expr gvars, Expr e ) {
  list<Expr>* pvars = pattern_variables(gvars);
  list<Expr>* r = pvars;
  for(list<Expr>* s = nest_groups(e); r->consp() && s->consp(); s=s->tl)
     if (r->hd->eq(s->hd))
	r = r->tl;
  if (r->consp())
     return #<order(...pvars)>;
  else return #<order()>;
};


/*defined below */
Expr best_evaluation_order ( Expr e );


%{

/* set this to true if you want to get a tracing of the rule system; lots of output */
#define trace_rules false

/* set this to true if you want to print the results of every optimization module */
#define trace_stages false

%trace { 0 }

/* inherited attributes */
%inherited order,	/* the expected order (for the physical stage) */
           groups	/* expected groups to be formed before nest (group-by) */


/* synthesized atttributes: */
%synthesized { int size;	/* the collection cardinality */
	       double cost;	/* the plan cost */
	       Expr order;	/* the computed (real) order */
	       Expr groups;	/* groups formed after nest (group-by) */
	     }
          = { 0, 1e100, #<order()>, #<groups()> }

%logical
none = none;

%physical

/* if no expected order, use TABLE_SCAN */
{ order=`ord; }
get(`M,`table,`name,`pred)
  : subsumes(ord,#<order(OID(`name))>)
  = TABLE_SCAN(`M,`table,`name,`pred)
     : { size = int(table_cardinality(table)*selectivity(pred));
	 cost = table_size(table);
	 order = #<order(OID(`name))>; };

/* an INDEX_SCAN that delivers the expected order  */
{ order=order(`o,...os); }
get(`M,`table,`name,`pred)
  : table->variablep()
  = #case find_index(table,name,#<order(`o,...os)>)
    |  index(`idx,order(...io))
	=> `(make_index_scan(idx,io,M,table,name,pred))
	   : { size = int(table_cardinality(table)*selectivity(pred));
	       cost = index_access_cost(table,name,pred,io);
	       order = #<order(...io)>; };
    #end;

/* if the first sort attribute is a table key, ignore the rest of the attributes and use index scan */
{ order=order(project(`t,`a,...s),`b,...os); }
get(`M,`table,`name,`pred)
  : table->variablep() && is_key(extent_type(table->name()),a)
  = #case find_index(table,name,#<order(project(`t,`a,...s))>)
    |  index(`idx,order(...io))
	=> `(make_index_scan(idx,io,M,table,name,pred))
	   : { size = int(table_cardinality(table)*selectivity(pred));
	       cost = index_access_cost(table,name,pred,io);
	       order = #<order(...io)>; };
    #end;

/* if expected order is empty, then use any index; so here we return every possible index */
{ order=order(); }
get(`M,`table,`name,`pred)
  : table->variablep()
  = #forall r in all_table_indexes(table,name) do
       #case r
       |  index(`idx,order(...io))
	     => `(make_index_scan(idx,io,M,table,name,pred))
	        : { size = int(table_cardinality(table)*selectivity(pred));
	            cost = index_access_cost(table,name,pred,io);
		    order = #<order(...io)>; };
       #end;
    #end;

/* if a non-empty order is expected for plan x, then SORT x and make		*/
/* the expected order for x empty so that this rule is not applied twice	*/
{ order=order(`o,...os); }
(`x<-{ order=order(); })
  : is_plan(x) && !subsumes(#<order(`o,...os)>,^x.order)
    && free_variables(#<order(`o,...os)>,range_vars(x))->nullp()
  = let Expr rwp = PHYSICAL_REWRITE(#<order(`o,...os)>,inherit(#<order()>,ATRBS.groups),LEVEL+1)->hd->expression
    in SORT(`x,`rwp)
    : { size = ^x.size;
	cost = ^x.cost+(^x.size*ilog(^x.size));
	order = #<order(`o,...os)>; };

{ order=order(`o,...os); }
(`x<-{ order=order(); })
  : is_plan(x) && subsumes(#<order(`o,...os)>,^x.order)
  = `x
    : { size = ^x.size;
	cost = ^x.cost;
	order =  ^x.order; };

{ order=order(); }
sort(`x<-{ order=`xord; },`xord)
  = `x
    : { size = ^x.size;
	cost = ^x.cost;
	order = ^x.order; };

{ order=order(`o,...os); }
sort(`x<-{ order=`xord; },`xord)
  = `x
    : { size = ^x.size;
	cost = ^x.cost;
	order = ^x.order; };

/* convert a join into a nested loop; the expected order propagates to the left input only */
{ order=`ord; }
join( `M, `x<-{ order=`ord; },
      `y<-{ order=order(); },
      `pred, `keep )
   : free_variables(ord,range_vars(x))->nullp()
   = NESTED_LOOP(`M,`x,`y,`pred,`keep)
     : { size = int(^x.size*^y.size*selectivity(pred));
	 cost = ^x.cost+^y.cost+(^x.size*^y.size);
	 order = ^x.order; };

{ order=order(); }
join( `M, `x<-{ order=order(); },
      `y<-{ order=order(); },
      `pred, `keep )
   = BLOCK_NESTED_LOOP(`M,`x,`y,`pred,`keep)
     : { size = int(^x.size*^y.size*selectivity(pred));
	 cost = ^x.cost+^y.cost+(^x.size*^y.size)/1000;
	 order = #<order()>; };

/* convert a join into an indexed nested loop; the expected order propagates to the left input only */
{ order=`ord; }
join( `M, `x<-{ order=`ord; },
      `y<-{ order=order(); },
      `pred, `keep )
   = #case y
     | INDEX_SCAN(`m,`Y,`v,`py,`index,`low,`high)
	   : !applicable_index(index,Y,v,pred)->eq(#<none>)
	   => INDEXED_LOOP(`M,`x,`y,`pred,`keep,`index,`(applicable_index(index,Y,v,pred)))
	      : { size = int(^x.size*^y.size*selectivity(pred));
		  cost = ^x.cost+4.0*^x.size;
		  order = ^x.order; };
     #end;

/* convert a pointer join into pointer access */
{ order=`ord; }
join( `M, `x<-{ order=`ord; },
      get( _, `table, `var, and(...q) ),
      and(...r,eq(`path,OID(`v)),...s), `keep )
   : v->eq(var)
   = MAP(`M,`x,`var,and(...q,...r,...s),`keep,`path)
     : { size = int(^x.size*selectivity(#<and(...q,...r,eq(`path,OID(`v)),...s)>));
	 cost = ^x.cost+4.0*^x.size;
	 order = ^x.order; };

/* convert a join into a merge join; both inputs are required to be sorted */
/* and at least one sort key to be a key (checked by is_left_key) */
{ order=`ord; }
join( `M, `x<-{ order=`(required_order(pred,range_vars(x),range_vars(y))); },
      `y<-{ order=`(required_order(pred,range_vars(y),range_vars(x))); },
      `pred, `keep )
   : keep->eq(#<none>)		/* currently I can't do an outer merge join */
     && equijoinp(pred,range_vars(x),range_vars(y))
     && subsumes(ord,^x.order)
   = let Expr xorder = required_order(pred,range_vars(x),range_vars(y)),
	 Expr yorder = required_order(pred,range_vars(y),range_vars(x))
     in #case is_left_key(x,y,xorder,yorder)
	| true => MERGE_JOIN(`M,`x,`y,`pred,true,`xorder,`yorder)
		  : { size = int(^x.size*^y.size*selectivity(pred));
		      cost = ^x.cost+^y.cost+(^x.size+^y.size);
		      order = ^x.order; };
	| false => MERGE_JOIN(`M,`x,`y,`pred,false,`xorder,`yorder)
		   : { size = int(^x.size*^y.size*selectivity(pred));
		       cost = ^x.cost+^y.cost+(^x.size+^y.size);
		       order = ^x.order; };
	#end;

{ order=`ord; }
nest( `M, `e<-{ order=`(nest_order(gv,e)); },
      `v, `hd, `gv, `bpred, `apred )
   : subsumes(ord,nest_order(gv,e))
   = NEST(`M,`e,`v,`hd,`gv,`bpred,`apred)
        : { size = ^e.size;
	    cost = ^e.cost+^e.size;
	    order = nest_order(gv,e); };

{ order=order(); }
reduce(`M,`e,`v,`head,`pred)
   = REDUCE(`M,`e,`v,`head,`pred)
     : { size = (collectionp(M)) ? ^e.size : 1;
	 cost = ^e.cost+^e.size;
	 order = ^e.order; };

{ order=`ord; }
unnest( `M, `e<-{ order=`ord; }, `v, `path, `pred, `keep )
  : subsumes(ord,^e.order) && !member(v,free_variables(ord,Nil))
  = UNNEST( `M, `e, `v, `path, `pred, `keep )
     : { size = ^e.size*10;
	 cost = ^e.cost+^e.size*10;
	 order = #<order(...(^e.order->arguments()),`v)>; };

{ order=order(...os,`o); }
unnest( `M, `e<-{ order=order(...os); }, `v, `path, `pred, `keep )
  : o->eq(v) && subsumes(#<order(...os)>,^e.order)
  = UNNEST( `M, `e, `v, `path, `pred, `keep )
     : { size = ^e.size*10;
	 cost = ^e.cost+^e.size*10;
	 order = #<order(...os,`o)>; };

{ order=order(); }
merge(`M,`x,`y)
  = MERGE(`M,`x,`y)
    : { size = ^x.size+^y.size;
	cost = ^x.cost+^y.cost+^x.size+^y.size;
	order = #<order()>; };

{ order=`ord; }
unit(`M,`x)
  = UNIT(`M,`x)
    : { size = 1;
	cost = 0;
	order = ord; };

{ order=`ord; }
zero(`M,`tp)
  = ZERO(`M,`tp)
    : { size = 0;
	cost = 0;
	order = ord; };

%}

/*--------------------------------------------------------------------------------*/


bool better_plan ( plan* x, plan* y ) {
  return !x->expression->eq(#<none>) && x->attributes.cost < y->attributes.cost;
};


/* returns the 10 best plans from a list of plans */
list<plan*>* ten_best_plans ( list<plan*>* plans ) {
  list<plan*>* res = new list<plan*>;
  short i = 0;
  for(list<plan*>* r = plans->sort(better_plan); i<10 && r->consp(); r=r->tl, i++)
     res = res->cons(r->hd);
  return res->reverse();
};


/* optimize an OQL query */
Expr best_evaluation_order ( Expr e ) {
  optimizer_init();
  /* initial inherited attributes */
  inherited init;
  init.order = #<order()>;
  init.groups = #<groups()>;
  plan* best_plan = new plan(#<none>,default_synthesized);
  best_plans = ten_best_plans;
  Expr be = best_tree(e);
  list<plan*>* plans = PHYSICAL_REWRITE(be,init,0);
  for(list<plan*>* r = plans; r->consp(); r = r->tl)
     if (r->hd->attributes.cost <= best_plan->attributes.cost)
	best_plan = r->hd;
  if (best_plan->expression->eq(#<none>))
     return plans->hd->expression;
  else return best_plan->expression;
};