/********************************************************************************
*
* File: evaluation.gen
* Title: The evaluator for the relational database system SDB.
* Programmer: Leonidas Fegaras, UTA
* Date: 2/21/97
*
* SDB is actually an in-memory database system: the database state is read
* from the file "database" when SDB starts, manipulated in memory, and
* then written back to the same file when SDB exits. Even though these
* evaluation operations are in-memory, they have the same semantics as
* their disk counterparts. Currently, only sorting, merging, nest-loops,
* and aggregation are supported.
*
* The following are the physical plan operators
* (indicated as plans) supported by SDB:
* 
* 1. TABLE_SCAN( table_name, range_variable, predicate )
*    implements the SQL query:
*    select * from table_name range_variable where predicate
* 2. INDEX_SCAN( table_name, range_variable, predicate, index_name, sort_order )
*    like TABLE_SCAN, but it uses the index index_name to
*    deliver the tuples order in sort_order
* 3. PROJECTION( plan, ... list_of_projections )
*    implements relational projection
* 4. FILTER( plan, predicate )
*    implements relational selection
* 5. SORT( plan, sort_order )
*    sorts a plan into the sort_order (a list of projections)
* 6. GROUP_BY( plan, ... list_of_projections )
*    groups a plan by the list_of_projections; plan should be
*    ordered by the list_of_projections
* 7. NESTED_LOOP( plan, plan, predicate )
* 8. MERGE_JOIN( plan, plan, predicate, left_sort_order, right_sort_order )
*    merge join requires the left input be sorted by left_sort_order
*    and the right by right_sort_order
* 9. UNION( plan, plan )
*
*********************************************************************************/


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


/* Lists of operator names; defined in typecheck.gen */
extern Names arithmetic;
extern Names comparisons;
extern Names aggregates;


/* DATABASE holds both the database schema and the database state */
extern database* DATABASE;


/* Maps an SQL query into an algebraic form and generates an optimized
   execution plan; defined in optimize.gen */
Expr optimize_query (Expr);


/* Testing ... */
test_foreign_keys () {
  Names all_table_names = DATABASE->content->names();
  for(Names tables = all_table_names; tables->consp(); tables=tables->tl)
  { Relation R = DATABASE->content->find(tables->hd);
    list<Expr>* foreign_keys = R->foreign_keys;
    for(list<Expr>* keys = foreign_keys; keys->consp(); keys=keys->tl)
    {  #case keys->hd
       | foreign(from(`a),to(`tbl,`b))
           => cout << "foreign key: from " << tables->hd << "." << a
		   << " to " << tbl << "." << b << "\n";
       #end;
    };
  };
};

/* Returns true if computed_order is a prefix of expected_order;
   defined in optimize.gen */
bool subsumes ( Expr computed_order, Expr expected_order );


/* These two functions appends the table name to the attribute name of a projection */
String projection_name ( String table, String atrb ) {
   string s = *table + "_" + *atrb;
   return new string(s);
};

String projection_name ( Expr e ) {
  #case e
  |  project(`table,`atrb)
	=> string s = *table->name() + "_" + *atrb->name();
	   return new string(s);
  #end;
};


/* convert a schema to an Expr and vice versa */
Expr schema_to_expr ( Schema schema ) {
  list<Expr>* sch = Nil;
  for(Names n = schema->names(); n->consp(); n=n->tl)
     sch = sch->cons(#<bind(`(variable(n->hd)),`(schema->find(n->hd)))>);
  return #<schema(...sch)>;
};

Schema expr_to_schema ( Expr e ) {
  Schema schema = new binding<Type>;
  for(list<Expr>* r = e->arguments(); r->consp(); r=r->tl)
  { list<Expr>* args = r->hd->arguments();
    schema = schema->extend(args->hd->name(),args->tl->hd);
  };
  return schema;
};


/* evaluate a projection */
Expr evaluator::projection ( Expr column, Schema schema ) {
  #case column
  |  project(`table,`atrb)
	=> { list<Expr>* v = values;
	     for(Names r = schema->names(); r->consp(); r=r->tl, v = v->tl)
                if (projection_name(column)->eq(r->hd)) return v->hd;
	   };
  #end;
  cout << schema << endl;
  type_error("Invalid projection",column);
};


/* evaluate a grouping */
Expr evaluator::grouping ( Expr fnc ) {
  #case fnc
  |  count => return integer(values->length());
  |  sum => int s = 0;
	    for(list<Expr>* r = values; r->consp(); r=r->tl)
	       if (r->hd->integerp()) s = s+r->hd->info.Integer;
	    return integer(s);
  |  _ => type_error("This aggregation is not implemented yet",fnc);
  #end;
};


/* evaluate an sql expression */
Expr evaluator::expression ( Expr e, Schema schema ) {
  #case e
  |  project(`table,`atrb) => return this->projection(e,schema);
  |  count(ALL)
	 => if (schema->in(new string("Grouping")))
	       return integer(values->hd->arguments()->length());
	    else type_error("count(*) without grouping has not been implemented yet",e);
  |  neg(`x) => return integer(-this->expression(x,schema)->info.Integer);
  |  `f(project(`table,`atrb))		/* must be a grouping before aggregation */
	 => if (member(f,aggregates) && schema->in(new string("Grouping")))
	    { Schema group_schema = expr_to_schema(schema->find(new string("Grouping")));
	      list<Expr>* res = Nil;
	      /* create a list of values, where each value is the table.atrb of each grouped tuple */
	      for(list<Expr>* r = values->hd->arguments(); r->consp(); r=r->tl)
	         res = res->cons(tuple(r->hd)->projection(#<project(`table,`atrb)>,group_schema));
	      /* perform the grouping */
	      return tuple(res)->grouping(f);
	     };
  |  `f(`x,`y) => if (member(f,arithmetic))
		     { Expr ex = this->expression(x,schema);
		       Expr ey = this->expression(y,schema);
		       #case f
		       |  plus => return integer(ex->info.Integer + ey->info.Integer);
		       |  minus => return integer(ex->info.Integer - ey->info.Integer);
		       |  times => return integer(ex->info.Integer * ey->info.Integer);
		       |  divide => return integer(ex->info.Integer / ey->info.Integer);
		       #end;
		     };
  #end;
  if (e->integerp() || e->stringp()) return e;
  type_error("Invalid expression",e);
};


/* evaluate an sql predicate */
bool evaluator::predicate ( Expr pred, Schema schema ) {
  #case pred
  |  true => return true;
  |  false => return false;
  |  and(...preds)
	=> for(list<Expr>* r = preds; r->consp(); r=r->tl)
	      if (!(this->predicate(r->hd,schema))) return false;
	   return true;
  |  or(...preds)
	=> for(list<Expr>* r = preds; r->consp(); r=r->tl)
	      if (this->predicate(r->hd,schema)) return true;
	   return false;
  |  not(`x) => return !this->predicate(x,schema);
  |  exists(`x) => return (evaluate_plan(x)->is_empty());
	/* otherwise must be a comparison */
  |  `f(`x,`y) => { Expr ex = this->expression(x,schema);
		    Expr ey = this->expression(y,schema);
		    #case f
		    |  eq => return ex->eq(ey);
		    |  neq => return !(ex->eq(ey));
		    |  _ => if (ex->eq(#<null>) || ey->eq(#<null>))
			       return false;	/* this is not accurate;		*/
						/* needs to handle nulls corectly	*/
			    else { int cmp = 0;
				   if (ex->integerp() && ey->integerp())
				      cmp = ex->info.Integer - ey->info.Integer;
				   else if (ex->stringp() && ey->stringp())
				      cmp = (int) strcpy(ex->info.Estring->value(),
							 ey->info.Estring->value());
				   else type_error("Can't compare these values",#<`f(`ex,`ey)>);
				   #case f
				   |  gt => return cmp > 0;
				   |  lt => return cmp < 0;
				   |  geq => return cmp >= 0;
				   |  leq => return cmp <= 0;
				   #end;
				 };
		    #end;
		  };
  #end;
  type_error("Invalid predicate",pred);
};


/* evaluate a plan produced by the optimizer */
Relation evaluate_plan ( Expr plan ) {
  Relation outrel;
  #case plan
  |  UNION(`x,`y)
	=> outrel = evaluate_plan(x)->append(evaluate_plan(y));
  |  TABLE_SCAN(`table,`name,`pred)
	=> outrel = DATABASE->content->find(table->name())->append_name(name->name())->select(pred);
  |  INDEX_SCAN(`table,`name,`pred,`idx,`order)
	=> outrel = DATABASE->content->find(table->name())->append_name(name->name())->sort(order)->select(pred);
  |  FILTER(`table,`pred)
	=> outrel = evaluate_plan(table)->select(pred);
  |  SORT(`table,`order)
	=> outrel = evaluate_plan(table)->sort(order);
  |  PROJECTION(`table,...attributes)
	=> outrel = evaluate_plan(table)->project(attributes);
  |  NESTED_LOOP(`x,`y,`pred)
	=> outrel = evaluate_plan(x)->nested_loop(evaluate_plan(y),pred);
  |  MERGE_JOIN(`x,`y,`pred,`x_order,`y_order)
	=> outrel = evaluate_plan(x)->merge_join(evaluate_plan(y),pred,x_order,y_order);
  |  GROUP_BY(`x,...groups)
	=> outrel = evaluate_plan(x)->group_by(groups);
  |  rename(`e,`name)
	=> outrel = evaluate_plan(e)->append_name(name->name());
  |  _ => type_error("Invalid plan",plan);
  #end;
  if (trace_evaluation)
  { cout << "The result of: " << plan << "\n";
    outrel->print();
  };
  return outrel;
};


/* compile and execute an sql DDL command (eg. CREATE TABLE) */
do_DDL ( Expr e, list<Expr>* constraints ) {
  #case e
  |  create_table(`table,...al)
	=> { e->pretty_print(0);
	     cout << "\n"; 
	     DATABASE->content = DATABASE->content->extend(table->name(),new relation(e,constraints));
	   };
  |  create_index(`name,`table,...al)
	=> Relation rel = DATABASE->find_relation(table->name());
	   rel->indexes = rel->indexes->extend(name->name(),#<order(...al)>);
  | _ => type_error("not supported yet",e);
  #end;
};


/* compile and execute an sql DML command (eg. an sql query) */
do_DML ( Expr e ) {
  #case e
  |  insert(`table,`atrbs,`value)
	=> DATABASE->find_relation(table->name())->insert(value);
  |  _ => { cout << "********************************************************************************\nQuery:\n";
	    if ((new typecheck())->query(e,#<NONE>))
	       { e->pretty_print(0);
		 cout << "\n";
		 Expr plan = optimize_query(e);
		 cout << "Plan:\n";
		 plan->pretty_print(0);
		 cout << "\n";
		 Relation outrel = evaluate_plan(plan);
		 outrel->print();
		};
	  };
  #end;
};



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


/* print schema and data content of a relation */
database::print () {
  cout << "\nDatabase contents:\n";
  for(Names r = content->names(); r->consp(); r=r->tl)
  {  cout << r->hd << ":\n";
     content->find(r->hd)->print();
  };
};


/* read a database from a file */
database::database ( String file_name ) {
  content = new binding<Relation>;
  ifstream from (file_name->value());
  int relations;
  from >> relations;
  for(int i = 0; i<relations; i++)
  { Relation rel;
    rel = relation_read(from);
    content = content->extend(rel->name,rel);
  };
  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)->relation_write(to);
  to.close();
};


/* find a relation in the database */
Relation database::find_relation ( String name ) {
  Relation r = content->find(name);
  if (r) return r;
  else type_error("No such table",variable(name));
};


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


/* create a relation useng the specs from the sql create table command */
relation::relation ( Expr e, list<Expr>* constraints ) {
  foreign_keys = constraints;
  primary_key = #<none>;
  indexes = new binding<Expr>;
  #case e
  |  create_table(`table,...al)
	=> { name = table->name();
	     Schema cols = new binding<Type>;
	     for(list<Expr>* r = al->reverse(); r->consp(); r=r->tl)
		#case r->hd
		|  attribute(`name,`tp,...l)
			=> cols = cols->extend(name->name(),tp);
		#end;
	     /* find the primary key and make it a primary (clustering) index */
	     for(list<Expr>* r = constraints; r->consp(); r=r->tl)
		#case r->hd
		|  key(...keys)
			=> { primary_key = r->hd;
			     indexes = indexes->extend(new string("Primary_idx"),#<order(...keys)>);
			   };
		#end;
	     schema = cols;
	   };
  |  _ => schema = new binding<Expr>;
  #end;
  data = Nil;
  record_size = schema->names()->length();
  cardinality = 0;
  size = 0;
};


relation::relation ( String n, Schema s, Expr key ) {
  /* new */
  foreign_keys = Nil;
  name = n;
  schema = s;
  primary_key = key;
  data = Nil;
  indexes = new binding<Expr>;
  record_size = s->names()->length();
  cardinality = 0;
  size = 0;
};


relation::relation ( Relation r ) {
  /* new */
  foreign_keys = r->foreign_keys;
  name = r->name;
  schema = r->schema;
  primary_key = r->primary_key;
  data = r->data;
  indexes = r->indexes;
  record_size = r->record_size;
  cardinality = r->cardinality;
  size = r->size;
};


Relation relation::copy () {
  Relation r = new relation(name,schema,primary_key);
  r->data = data;
  r->indexes = indexes;
  r->record_size = record_size;
  r->cardinality = cardinality;
  r->size = size;
  return r;
};


relation::drop_table () {
  delete schema;
};


/* insert a new tuple */
relation::insert ( Expr e ) {
  data = data->cons(e);
  cardinality++;
  size = (cardinality*record_size)/block_size+1;
};


/* remove all tuples that satisfy a predicate */
relation::remove ( Expr predicate ) {
  list<Expr>* res = Nil;
  for(list<Expr>* r = data; r->consp(); r=r->tl)
     if (!(tuple(r->hd)->predicate(predicate,schema)))
	res = res->cons(r->hd);
  data = res;
};


/* update all tuples that satisfy a predicate using the binding from attributes to values */
relation::update ( Expr predicate, binding<Expr>* values ) {
  for(list<Expr>* d = data; d->consp(); d=d->tl)
  {  Evaluator tuple = new evaluator(d->hd);
     if (tuple->predicate(predicate,schema))
	for(Names atrbs = values->names(); atrbs->consp(); atrbs=atrbs->tl)
	{  list<Expr>* v = d->hd->arguments();
	   for(Names r = schema->names(); r->consp(); r=r->tl, v = v->tl)
              if (atrbs->hd->eq(r->hd))
		 v->hd = tuple->expression(values->find(atrbs->hd),schema);
	};
   };
};


/* relational select */
Relation relation::select ( Expr predicate ) {
  Relation outrel = new relation(name,schema,primary_key);
  for(list<Expr>* r = data; r->consp(); r=r->tl)
     if (tuple(r->hd)->predicate(predicate,schema))
	outrel->insert(r->hd);
  outrel->data = outrel->data->reverse();
  return outrel;
};


/* relational project; doesn't remove duplicates */
Relation relation::project ( list<Expr>* attributes ) {
  /* find the new schema first */
  Schema s = new binding<Type>;
  for(list<Expr>* atrbs = attributes->reverse(); atrbs->consp(); atrbs=atrbs->tl)
     #case atrbs->hd
     |  as(project(`table,`atrb),`a)
	 => { String nm = projection_name(table->name(),atrb->name());
	      s = s->extend(nm,schema->find(nm));
	    };
     |  as(`e,`a) => s = s->extend(a->name(),#<UNKNOWN>);
     #end;
  /* then project each tuple */
  Relation outrel = new relation(name,s,primary_key);
  for(list<Expr>* d = data; d->consp(); d=d->tl)
  {  list<Expr>* v = d->hd->arguments();
     list<Expr>* o = Nil;
     for(list<Expr>* atrbs = attributes; atrbs->consp(); atrbs=atrbs->tl)
        #case atrbs->hd
        |  as(`e,`a) => o = o->cons(tuple(d->hd)->expression(e,schema));
        #end;
     outrel->insert(#<tuple(...(o->reverse()))>);
  };
  outrel->data = outrel->data->reverse();
  return outrel;
};


/* returns x.attribute-y.attribute (x and y have the same schema) */
int relation::compare ( Expr x, Expr y, Expr attribute ) {
  Expr rx = tuple(x)->expression(attribute,schema);
  Expr ry = tuple(y)->expression(attribute,schema);
  if (rx->integerp() && ry->integerp())
     return rx->info.Integer - ry->info.Integer;
  else if (rx->stringp() && ry->stringp())
     return strcmp(rx->info.Estring->value(),ry->info.Estring->value());
  else if (rx->eq(#<null>) || ry->eq(#<null>)) return 0;
  else type_error("Can't do the comparison",#<lt(`rx,`ry)>);
};


/* true if for all attributes A in attributes x.A=y.A (x and y have the same schema) */
bool relation::equal_to ( Expr x, Expr y, list<Expr>* attributes ) {
  for(list<Expr>* r = attributes; r->consp(); r=r->tl)
  {  int c = this->compare(x,y,r->hd);
     if (c != 0) return false;
  };
  return true;
};


/* true if for all attributes A in attributes x.A<y.A (x and y have the same schema) */
bool relation::less_than ( Expr x, Expr y, list<Expr>* attributes ) {
  for(list<Expr>* r = attributes; r->consp(); r=r->tl)
  {  int c = this->compare(x,y,r->hd);
     if (c>0) return false;
     if (c<0) return true;
  };
  return false;
};


/* sorts a relation; dumps a relation into a vector and uses bubble-sort */
Relation relation::sort ( Expr attributes ) {
  #case attributes
  |  order(...atrbs) =>
        Relation outrel = new relation(this);
        Expr A[cardinality];
        int k = 0;
        for(list<Expr>* r = data; r->consp() && k<cardinality; r=r->tl) A[k++] = r->hd;
	/* bubble sort */
        for(int j = cardinality-1; j>=1; j--)
           for(int i = 1; i<=j; i++)
	      	if (this->less_than(A[i],A[i-1],atrbs))
                { Expr e = A[i-1];
      	  	  A[i-1] = A[i];
	      	  A[i] = e;
      		};
        outrel->data = Nil;
        for(int i = cardinality-1; i>=0; i--)
	   outrel->data = outrel->data->cons(A[i]);
        return outrel;
  #end;
  return this;
};


/* nested loop join */
Relation relation::nested_loop ( Relation r, Expr predicate ) {
  Schema s = schema->extend(r->schema);
  Relation outrel = new relation(name,s,primary_key);
  for(list<Expr>* left = data; left->consp(); left=left->tl)
     for(list<Expr>* right = r->data; right->consp(); right=right->tl)
     {  list<Expr>* tp = left->hd->arguments()->append(right->hd->arguments());
	if (tuple(tp)->predicate(predicate,s))
	   outrel->insert(#<tuple(...tp)>);
     };
  return outrel;
};


/* construct the join predicate for the merge join; use the requested orders
*  of the two inputs for the equality tests */
Expr make_pred_for_merge ( Expr comparison, Expr left_order, Expr right_order ) {
  Expr res = #<true>;
  list<Expr>* right = right_order->arguments();
  for(list<Expr>* left = left_order->arguments();
      left->consp() && right->consp(); left=left->tl, right=right->tl)
    res = #<and(`comparison(`(left->hd),`(right->hd)),`res)>;
  return res;
};


/* concatenate the right to the left tuple */
Evaluator tuple ( Expr left, Expr right ) {
  return tuple(left->arguments()->append(right->arguments()));
};
    

/* sort-merge join: it scans the sorted left and right inputs simultaneously */
Relation relation::merge_join ( Relation r, Expr predicate, Expr order, Expr r_order ) {
  Schema s = schema->extend(r->schema);
  Relation outrel = new relation(name,s,primary_key);
  Expr greater_than = make_pred_for_merge(#<gt>,order,r_order);
  Expr less_than = make_pred_for_merge(#<lt>,order,r_order);
  list<Expr>* left = data;
  list<Expr>* right = r->data;
  list<Expr>* rd;
  while(left->consp() && right->consp())
  { if (tuple(left->hd,right->hd)->predicate(greater_than,outrel->schema))
       right = right->tl;
    else if (tuple(left->hd,right->hd)->predicate(less_than,outrel->schema))
       left = left->tl;
    else { do { for(rd = right;
		    rd->consp ()
		    && !(tuple(left->hd,rd->hd)->predicate(less_than,outrel->schema));
		    rd=rd->tl)
  		  if (tuple(left->hd,rd->hd)->predicate(predicate,outrel->schema))
		     outrel->insert(#<tuple(...(left->hd->arguments()->append(rd->hd->arguments())))>);
		left = left->tl;
	   } while (left->consp() && rd->consp()
		    && tuple(left->hd,right->hd)->predicate(greater_than,outrel->schema));
	   right = rd;
	 };
  };
  outrel->data = outrel->data->reverse();
  return outrel;
};


/* group-by expects the input to sorted by the groups attributes;
*  then it scans the input and it groups all items with equal values on
*  groups attributes into one big set; the resulting relation becomes a nested set */
Relation relation::group_by ( list<Expr>* groups ) {
  Relation outrel = new relation(new string("GROUPING"),new binding<Expr>,#<order(...groups)>);
  for(list<Expr>* r = groups; r->consp(); r=r->tl)
     outrel->schema = outrel->schema->extend(projection_name(r->hd),#<UNKNOWN>);
  outrel->schema = outrel->schema->extend(new string("Grouping"),schema_to_expr(schema));
  list<Expr>* s;
  for(list<Expr>* r = data; r->consp();)
  {  list<Expr>* group = Nil;
     for(s = r; s->consp() && this->equal_to(r->hd,s->hd,groups); s=s->tl)
	group = group->cons(s->hd);
     list<Expr>* tp = Nil;
     for(list<Expr>* a = groups->reverse(); a->consp(); a=a->tl)
	tp = tp->cons(tuple(r->hd)->projection(a->hd,schema));
     tp = tp->cons(#<group(...group)>);
     outrel->insert(#<tuple(...tp)>);
     r = s;
  };
  outrel->data = outrel->data->reverse();
  return outrel;
};


/* append a relation r at the end of a relation */
Relation relation::append ( Relation r ) {
  Relation res (this);
  res->data = res->data->append(r->data);
  res->cardinality += r->cardinality;
  res->size += r->size;
  return res;
};


/* find an index for this relation that delivers the order `index' */
Expr relation::find_index ( Expr index ) {
  if (index->arguments()->nullp()) return #<none>;
  if (index->arguments()->hd->arguments()->nullp()) return #<none>;
  Expr nm = index->arguments()->hd->arguments()->hd;
  for(Names n = indexes->names(); n->consp(); n=n->tl)
  {  list<Expr>* idx = Nil;
     for(list<Expr>* r = indexes->find(n->hd)->arguments()->reverse(); r->consp(); r=r->tl)
	idx = idx->cons(#<project(`nm,`(r->hd))>);
     if (subsumes(#<order(...idx)>,index))
	return #<index(`(variable(n->hd)),order(...idx))>;
  };
  return #<none>;
};


/* print info about a relation */
relation::print () {
  cout << "Relation: " << name << ", Primary key: " << primary_key
       << ", Record size: " << record_size
       << ", Cardinality: " << cardinality << ", Size: " << size << "\n";
  Names names = indexes->names();
  if (names->consp())
  { cout << "Indexes: [ " << names->hd << ":" << indexes->find(names->hd);
    for(Names n = names->tl; n->consp(); n=n->tl)
	cout << " " << n->hd << ":" << indexes->find(n->hd);
    cout << " ]\n";
  };
  names = schema->names();
  cout << "Schema: { " << names->hd << ":" << schema->find(names->hd);
  for(Names n = names->tl; n->consp(); n=n->tl)
	cout << ", " << n->hd << ":" << schema->find(n->hd);
  cout << " }\nValues:\n";
  for(list<Expr>* d = data; d->consp(); d=d->tl)
     cout << "   " << d->hd <<"\n";
};


/* put the name as a prefix to the attribute names */
Relation relation::append_name ( String name ) {
  Relation r = new relation(this);
  r->schema = new binding<Expr>;
  for(Names n = schema->names()->reverse(); n->consp(); n=n->tl)
     r->schema = r->schema->extend(projection_name(name,n->hd),schema->find(n->hd));
  return r;
};


Relation relation_read ( ifstream& from ) {
  Relation res = new relation(new string(""),new binding<Expr>,#<none>);
  char c;
  int len;
  from.get(c);
  res->name = reads(from);
  from >> res->record_size >> res->cardinality >> res->size;
  res->primary_key = read(from);
  from >> len;
  for(int i = 0; i<len; i++)
  { list<Expr>* r = read(from)->arguments();
    res->indexes = res->indexes->extend(r->hd->name(),r->tl->hd);
  };
  from >> len;
  for(int i = 0; i<len; i++)
  { list<Expr>* r = read(from)->arguments();
    res->schema = res->schema->extend(r->hd->name(),r->tl->hd);
  };
  /* new */
  res->foreign_keys = read(from)->arguments();
  res->data = read(from)->arguments();
  return res;
};


relation::relation_write ( ofstream& to ) {
  name->write(to);
  to << ' ' << record_size << ' ' << cardinality << ' ' << size;
  primary_key->write(to);
  to << ' ' << indexes->names()->length();
  for(Names r = indexes->names(); r->consp(); r=r->tl)
     #<attribute(`(variable(r->hd)),`(indexes->find(r->hd)))>->write(to);
  to << ' ' << schema->names()->length();
  for(Names r = schema->names()->reverse(); r->consp(); r=r->tl)
     #<attribute(`(variable(r->hd)),`(schema->find(r->hd)))>->write(to);
  /* new */
  #<data(...foreign_keys)>->write(to);
  #<data(...data)>->write(to);
};
