core/src/main/java/org/apache/mrql/TopLevel.gen - incubator-retired-mrql - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.mrql;

 import java.util.List;
 import org.apache.mrql.gen.*;


 /** Provides the API for compilation/code-generation */
 final public class TopLevel extends Interpreter {
     static Tree xml_type;

     public TopLevel () {
         // XML and JSON are user-defined types:
         datadef("XML",#<union(Node(tuple(string,bag(tuple(string,string)),list(XML))),
                               CData(string))>);
         datadef("JSON",#<union(JObject(bag(tuple(string,JSON))),
                                JArray(list(JSON)),
                                Jstring(string),
                                Jlong(long),
                                Jdouble(double),
                                Jbool(bool),
                                Jnull(tuple()))>);
         constant(#<PI>,#<double>,new MR_double(Math.PI));
         xml_type = global_datatype_env.lookup("XML");
         DataSource.loadParsers();
     }

     private static boolean memory_parsed_source ( Tree e ) {
         match e {
         case ParsedSource(...): return true;
         case Merge(`x,`y): return memory_parsed_source(x) && memory_parsed_source(y);
         };
         return false;
     }

     /** translate and evaluate an MRQL expression into MRData
      * @param e MRQL query to be evaluated
      * @param print do we want to print the result?
      * @return the result of evaluation (MRData)
      */
     public static MRData expression ( Tree e, boolean print ) {
         try {
             Tree plan = translate_expression(e);
             Tree type = query_type;
             query_plan = plan;
             tab_count = -3;
             trace_count = 0;
             if (plan == null)
                 return null;
             if (Config.hadoop_mode)
                 Evaluator.evaluator.initialize_query();
             MRData res = evalE(plan,null);
             if (print) {
                 if (!Config.quiet_execution)
                     System.out.println("Result:");
                 if (!Config.hadoop_mode && Config.bsp_mode && memory_parsed_source(plan))
                     System.out.println(print(((Tuple)((Bag)res).get(0)).second(),type));
                 else System.out.println(print(res,type));
             } return res;
         } catch (Exception x) {
             if (x.getMessage() != null)
                 System.err.println("*** MRQL System Error at line "+Main.parser.line_pos()+": "+x);
             if (Config.trace)
                 x.printStackTrace(System.err);
             if (Config.testing)
                 throw new Error(x);
             return null;
         } catch (Error x) {
             if (x.getMessage() != null)
                 System.err.println("*** MRQL System Error at line "+Main.parser.line_pos()+": "+x);
             if (Config.trace)
                 x.printStackTrace(System.err);
             if (Config.testing)
                 throw new Error(x);
             return null;
         }
     }

     /** translate, evaluate, and print the results of an MRQL expression e
      * @param e MRQL query to be evaluated
      * @return the result of evaluation (MRData)
      */
     public final static MRData expression ( Tree e ) {
         reset();
         return expression(e,true);
     }

     /** handle the assignment v=e */
     public final static void assign ( String v, Tree e ) {
         if (variable_lookup(v,global_env) != null) {
             global_type_env.remove(v);
             remove_global_binding(v);
         };
         global_vars.insert(v,e);
     }

     private final static boolean is_function ( Tree e ) {
         match e {
         case function(...): return true;
         };
         return false;
     }

     /** handle the assignment v:=e */
     public final static Tree distributed_assign ( String v, Tree e ) {
         reset();
         if (global_vars.lookup(v) != null)
             global_vars.remove(v);
         MRData res = expression(e,false);
         global_type_env.insert(v,query_type);
         if (res instanceof Bag)
             ((Bag)res).materialize();
         new_distributed_binding(v,res);
         return query_plan;
     }

     /** bind v to the result of e */
     public final static Tree store ( String v, Tree e ) {
         reset();
         if (global_vars.lookup(v) != null)
             global_vars.remove(v);
         MRData res = expression(e,false);
         global_type_env.insert(v,query_type);
         if (res instanceof Bag)
             ((Bag)res).materialize();
         new_global_binding(v,res);
         return query_plan;
     }

     /** define an MRQL constant, such as PI */
     private final static void constant ( Tree v, Tree type, MRData value ) {
         String var = v.toString();
         if (global_vars.lookup(var) != null)
             global_vars.remove(var);
         global_type_env.insert(var,type);
         new_global_binding(var,value);
     }

     /** define a new function
      * @param fnc function name
      * @param params parameter list
      * @param out_type output type
      * @param body function body
      */
     public final static void functiondef ( String fnc, Trees params, Tree out_type, Tree body ) {
         reset();
         Trees as = #[];
         Trees ps = #[];
         for ( Tree param: params )
             match param {
             case bind(`v,`tp):
                 Tree ntp = normalize_type(tp);
                 as = as.append(ntp);
                 ps = ps.append(#<bind(`v,`ntp)>);
             case _: type_error(param,"Ill-formed function parameter: "+param);
             };
         out_type = normalize_type(out_type);
         // needed for recursive functions
         global_type_env.insert(fnc,#<arrow(tuple(...as),`out_type)>);
         Tree fname = #<`fnc>;
         if (!is_pure(body))
             impure_functions = impure_functions.append(fname);
         Tree plan = store(fnc,#<function(tuple(...ps),`out_type,`body)>);
         if (plan != null)
             Translator.global_functions.insert(fnc,plan);
         if (Config.hadoop_mode && plan != null)
             Plan.conf.set("mrql.global."+fnc,
                           closure(plan,global_env).toString());
     }

     /** dump the result of evaluating the MRQL query e to a binary file */
     private final static void dump ( String file, Tree e ) {
         MRData res = expression(e,false);
         try {
             query_type = make_persistent_type(query_type);
             if (res != null)
                 if (Config.hadoop_mode)
                     Evaluator.evaluator.dump(file,query_type,res);
                 else MapReduceAlgebra.dump(file,query_type,res);
         } catch (Exception x) {
             throw new Error(x);
         }
     }

     /** dump the result of evaluating the MRQL query e to a text CVS file */
     private final static void dump_text ( String file, Tree e ) {
         MRData res = expression(e,false);
         if (res != null)
             try {
 		Evaluator.evaluator.dump_text(file,query_type,res);
             } catch (Exception x) {
                 throw new Error(x);
             }
     }

     /** true, if e is a strem-processing expression */
     public final static boolean stream_expression ( Tree e ) {
         match e {
         case call(stream,...): return true;
         case `f(...as):
             for ( Tree a: as )
                 if (stream_expression(a))
                     return true;
         };
         return false;
     }

     /** define a new named type (typedef) */
     private final static void typedef ( String name, Tree type ) {
         type_names.insert(name,normalize_type(type));
     }

     /** define a new data type, such as XML and JSON */
     private final static void datadef ( String name, Tree type ) {
         int i = 0;
         Trees as = #[];
         match type {
         case union(...nl):
             // needed for recursive datatypes
             global_datatype_env.insert(name,#<union(...nl)>);
             for ( Tree n: nl )
                 match n {
                 case `c(`t):
                     if (data_constructors.lookup(c.toString()) == null)
                         data_constructors.insert(c.toString(),#<`name(`i,`t)>);
                     else type_error(type,"Data constructor "+c+" has already been defined");
                     as = as.append(#<`c(`(normalize_type(t)))>);
                     i++;
                 }
         };
         global_datatype_env.remove(name);
         global_datatype_env.insert(name,#<union(...as)>);
     }

     /** define a user aggregation */
     private static void aggregation ( String name, Tree type, Tree plus, Tree zero, Tree unit ) {
         reset();
         zero = Simplification.rename(zero);
         plus = Simplification.rename(plus);
         unit = Simplification.rename(unit);
         type = normalize_type(type);
         Tree ztp = TypeInference.type_inference2(zero);
         Tree v1 = new_var();
         type_env.insert(v1.toString(),ztp);
         TypeInference.type_inference2(Normalization.normalize_all(#<apply(`plus,tuple(`v1,`v1))>));
         Tree v2 = new_var();
         type_env.insert(v2.toString(),type);
         Tree utp = TypeInference.type_inference2(Normalization.normalize_all(#<apply(`unit,`v2)>));
         if (unify(utp,ztp) == null)
             type_error(unit,"Wrong type in unit result (expected "+ztp+" found "+utp);
         monoids = monoids.append(#<`name(`type,`plus,`zero,`unit)>);
     }

     /** bind the pattern variables to values */
     private static Environment bind_list ( Tree pattern, MRData value, Environment env ) {
         match pattern {
         case tuple(...ps):
             int i = 0;
             Tuple t = (Tuple)value;
             for ( Tree p: ps )
                 env = bind_list(p,t.get(i++),env);
             return env;
         };
         return new Environment(pattern.toString(),value,env);
     }

     /** bind the pattern variables to values */
     private static Environment bind_list ( Tree pattern, Tree src, Environment env ) {
         match pattern {
         case tuple(...ps):
             int i = 0;
             match src {
             case tuple(...es):
                 for ( Tree p: ps )
                     env = bind_list(p,es.nth(i++),env);
                 return env;
             }
         };
         MRData value = Evaluator.evaluator.evalE(src,env);
         env.replace(pattern.toString(),value);
         return env;
     }

     /** the MRQL top-level interfacse to evaluate a single MRQL expression or command */
     public final static void evaluate_top_level ( Tree expr ) {
         if (expr == null)
             return;
         match expr {
         case expression(`e):
             long t = System.currentTimeMillis();
             if (expression(e) != null && !Config.quiet_execution)
                 System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
         case assign(`v,`e): assign(v.toString(),e);
         case store(`v,`e):
             long t = System.currentTimeMillis();
             if (distributed_assign(v.toString(),e) != null && !Config.quiet_execution)
                 System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
         case dump(`s,`e):  // dump stream output
             if (!stream_expression(e))
                 fail;
             reset();
             Tree plan = translate_expression(e);
             if (plan == null)
                 return;
             Evaluator.evaluator.initialize_query();
             final Tree qt = query_type;
             final String file = s.stringValue();
             Evaluator.evaluator.streaming(plan,null,null,new Function(){
                     long i = 0;
                     public MRData eval ( final MRData data ) {
                         try {
                             Evaluator.evaluator.dump(file+"/f"+(i++),qt,data);
                             return data;
                         } catch (Exception ex) {
                             throw new Error("Cannot dump the streaming result to the directory "+file);
                         }
                     }
                 });
         case dump_text(`s,`e):  // dump stream output
             if (!stream_expression(e))
                 fail;
             reset();
             Tree plan = translate_expression(e);
             if (plan == null)
                 return;
             Evaluator.evaluator.initialize_query();
             final Tree qt = query_type;
             final String file = s.stringValue();
             Evaluator.evaluator.streaming(plan,null,null,new Function(){
                     long i = 0;
                     public MRData eval ( final MRData data ) {
                         try {
                             Evaluator.evaluator.dump_text(file+"/f"+(i++),qt,data);
                             return data;
                         } catch (Exception ex) {
                             throw new Error("Cannot dump the streaming result to the directory "+file);
                         }
                     }
                 });
         case incr(`e):  // incremental stream processing
             Config.incremental = true;
             reset();
             Tree plan = translate_expression(e);
             Config.incremental = false;
             if (plan == null)
                 return;
             Evaluator.evaluator.initialize_query();
             match plan {
             case incr(`zero,lambda(`state,`merge),lambda(_,`answer)):
                 final Tree qt = make_persistent_type(query_type);
                 MRData z = evalE(zero,null);
                 final Environment env =bind_list(state,z,null);
                 final Tree ans = answer;
                 merge = Streaming.streamify(merge);
                 Evaluator.evaluator.streaming(#<lambda(`state,`merge)>,env,null,
                    new Function() {
                         public MRData eval ( final MRData state ) {
                             MRData a = Evaluator.evaluator.evalE(ans,env);
                             System.out.println(print(a,qt));
                             return state;
                         }
                     });
             };
         case lineage(`e):
             Config.lineage = true;
             boolean quiet = Config.quiet_execution;
             Config.quiet_execution = true;
             long t = System.currentTimeMillis();
             if (expression(e,false) != null && !Config.quiet_execution)
                 System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
             Config.lineage = false;
             Config.quiet_execution = quiet;
         case debug(`path,`e):
             Config.debug = true;
             boolean quiet = Config.quiet_execution;
             Config.quiet_execution = true;
             long t = System.currentTimeMillis();
             dump(path.stringValue(),e);
             if (!Config.quiet_execution)
                 System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
             Config.debug = false;
             Config.quiet_execution = quiet;
         case dump(`s,`e):
             long t = System.currentTimeMillis();
             dump(s.stringValue(),e);
             if (!Config.quiet_execution)
                 System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
         case dump_text(`s,`e):
             long t = System.currentTimeMillis();
             dump_text(s.stringValue(),e);
             if (!Config.quiet_execution)
                 System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
         case typedef(`v,`t): typedef(v.toString(),t);
         case datadef(`v,`t): datadef(v.toString(),t);
         case functiondef(`f,params(...p),`tp,`e):
             functiondef(f.toString(),p,tp,e);
         case macrodef(`name,params(...p),`e):
             Translator.global_macros.insert(name.toString(),#<macro(params(...p),`e)>);
         case aggregation(`aggr,`type,`plus,`zero,`unit):
             aggregation(aggr.toString(),type,plus,zero,unit);
         case import(`c):
             ClassImporter.importClass(c.variableValue());
         case import(`c,...l):
             for (Tree m: l)
                 ClassImporter.importMethod(c.variableValue(),m.variableValue());
         case include(`file):
             Main.include_file(file.toString());
         case parser(`n,`p):
             try {
                 Class<? extends Parser> c = Class.forName(p.toString()).asSubclass(Parser.class);
                 DataSource.parserDirectory.put(n.toString(),c);
             } catch (ClassNotFoundException e) {
                 throw new Error("Class "+p.toString()+" not found");
             }
         case impure(`fn):    // not used
             impure_functions = impure_functions.append(fn);
         case _: throw new Error("Unknown statement: "+expr);
         }
     }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.mrql;

	import java.util.List;
	import org.apache.mrql.gen.*;


	/** Provides the API for compilation/code-generation */
	final public class TopLevel extends Interpreter {
	static Tree xml_type;

	public TopLevel () {
	// XML and JSON are user-defined types:
	datadef("XML",#<union(Node(tuple(string,bag(tuple(string,string)),list(XML))),
	CData(string))>);
	datadef("JSON",#<union(JObject(bag(tuple(string,JSON))),
	JArray(list(JSON)),
	Jstring(string),
	Jlong(long),
	Jdouble(double),
	Jbool(bool),
	Jnull(tuple()))>);
	constant(#<PI>,#<double>,new MR_double(Math.PI));
	xml_type = global_datatype_env.lookup("XML");
	DataSource.loadParsers();
	}

	private static boolean memory_parsed_source ( Tree e ) {
	match e {
	case ParsedSource(...): return true;
	case Merge(`x,`y): return memory_parsed_source(x) && memory_parsed_source(y);
	};
	return false;
	}

	/** translate and evaluate an MRQL expression into MRData
	* @param e MRQL query to be evaluated
	* @param print do we want to print the result?
	* @return the result of evaluation (MRData)
	*/
	public static MRData expression ( Tree e, boolean print ) {
	try {
	Tree plan = translate_expression(e);
	Tree type = query_type;
	query_plan = plan;
	tab_count = -3;
	trace_count = 0;
	if (plan == null)
	return null;
	if (Config.hadoop_mode)
	Evaluator.evaluator.initialize_query();
	MRData res = evalE(plan,null);
	if (print) {
	if (!Config.quiet_execution)
	System.out.println("Result:");
	if (!Config.hadoop_mode && Config.bsp_mode && memory_parsed_source(plan))
	System.out.println(print(((Tuple)((Bag)res).get(0)).second(),type));
	else System.out.println(print(res,type));
	} return res;
	} catch (Exception x) {
	if (x.getMessage() != null)
	System.err.println("*** MRQL System Error at line "+Main.parser.line_pos()+": "+x);
	if (Config.trace)
	x.printStackTrace(System.err);
	if (Config.testing)
	throw new Error(x);
	return null;
	} catch (Error x) {
	if (x.getMessage() != null)
	System.err.println("*** MRQL System Error at line "+Main.parser.line_pos()+": "+x);
	if (Config.trace)
	x.printStackTrace(System.err);
	if (Config.testing)
	throw new Error(x);
	return null;
	}
	}

	/** translate, evaluate, and print the results of an MRQL expression e
	* @param e MRQL query to be evaluated
	* @return the result of evaluation (MRData)
	*/
	public final static MRData expression ( Tree e ) {
	reset();
	return expression(e,true);
	}

	/** handle the assignment v=e */
	public final static void assign ( String v, Tree e ) {
	if (variable_lookup(v,global_env) != null) {
	global_type_env.remove(v);
	remove_global_binding(v);
	};
	global_vars.insert(v,e);
	}

	private final static boolean is_function ( Tree e ) {
	match e {
	case function(...): return true;
	};
	return false;
	}

	/** handle the assignment v:=e */
	public final static Tree distributed_assign ( String v, Tree e ) {
	reset();
	if (global_vars.lookup(v) != null)
	global_vars.remove(v);
	MRData res = expression(e,false);
	global_type_env.insert(v,query_type);
	if (res instanceof Bag)
	((Bag)res).materialize();
	new_distributed_binding(v,res);
	return query_plan;
	}

	/** bind v to the result of e */
	public final static Tree store ( String v, Tree e ) {
	reset();
	if (global_vars.lookup(v) != null)
	global_vars.remove(v);
	MRData res = expression(e,false);
	global_type_env.insert(v,query_type);
	if (res instanceof Bag)
	((Bag)res).materialize();
	new_global_binding(v,res);
	return query_plan;
	}

	/** define an MRQL constant, such as PI */
	private final static void constant ( Tree v, Tree type, MRData value ) {
	String var = v.toString();
	if (global_vars.lookup(var) != null)
	global_vars.remove(var);
	global_type_env.insert(var,type);
	new_global_binding(var,value);
	}

	/** define a new function
	* @param fnc function name
	* @param params parameter list
	* @param out_type output type
	* @param body function body
	*/
	public final static void functiondef ( String fnc, Trees params, Tree out_type, Tree body ) {
	reset();
	Trees as = #[];
	Trees ps = #[];
	for ( Tree param: params )
	match param {
	case bind(`v,`tp):
	Tree ntp = normalize_type(tp);
	as = as.append(ntp);
	ps = ps.append(#<bind(`v,`ntp)>);
	case _: type_error(param,"Ill-formed function parameter: "+param);
	};
	out_type = normalize_type(out_type);
	// needed for recursive functions
	global_type_env.insert(fnc,#<arrow(tuple(...as),`out_type)>);
	Tree fname = #<`fnc>;
	if (!is_pure(body))
	impure_functions = impure_functions.append(fname);
	Tree plan = store(fnc,#<function(tuple(...ps),`out_type,`body)>);
	if (plan != null)
	Translator.global_functions.insert(fnc,plan);
	if (Config.hadoop_mode && plan != null)
	Plan.conf.set("mrql.global."+fnc,
	closure(plan,global_env).toString());
	}

	/** dump the result of evaluating the MRQL query e to a binary file */
	private final static void dump ( String file, Tree e ) {
	MRData res = expression(e,false);
	try {
	query_type = make_persistent_type(query_type);
	if (res != null)
	if (Config.hadoop_mode)
	Evaluator.evaluator.dump(file,query_type,res);
	else MapReduceAlgebra.dump(file,query_type,res);
	} catch (Exception x) {
	throw new Error(x);
	}
	}

	/** dump the result of evaluating the MRQL query e to a text CVS file */
	private final static void dump_text ( String file, Tree e ) {
	MRData res = expression(e,false);
	if (res != null)
	try {
	Evaluator.evaluator.dump_text(file,query_type,res);
	} catch (Exception x) {
	throw new Error(x);
	}
	}

	/** true, if e is a strem-processing expression */
	public final static boolean stream_expression ( Tree e ) {
	match e {
	case call(stream,...): return true;
	case `f(...as):
	for ( Tree a: as )
	if (stream_expression(a))
	return true;
	};
	return false;
	}

	/** define a new named type (typedef) */
	private final static void typedef ( String name, Tree type ) {
	type_names.insert(name,normalize_type(type));
	}

	/** define a new data type, such as XML and JSON */
	private final static void datadef ( String name, Tree type ) {
	int i = 0;
	Trees as = #[];
	match type {
	case union(...nl):
	// needed for recursive datatypes
	global_datatype_env.insert(name,#<union(...nl)>);
	for ( Tree n: nl )
	match n {
	case `c(`t):
	if (data_constructors.lookup(c.toString()) == null)
	data_constructors.insert(c.toString(),#<`name(`i,`t)>);
	else type_error(type,"Data constructor "+c+" has already been defined");
	as = as.append(#<`c(`(normalize_type(t)))>);
	i++;
	}
	};
	global_datatype_env.remove(name);
	global_datatype_env.insert(name,#<union(...as)>);
	}

	/** define a user aggregation */
	private static void aggregation ( String name, Tree type, Tree plus, Tree zero, Tree unit ) {
	reset();
	zero = Simplification.rename(zero);
	plus = Simplification.rename(plus);
	unit = Simplification.rename(unit);
	type = normalize_type(type);
	Tree ztp = TypeInference.type_inference2(zero);
	Tree v1 = new_var();
	type_env.insert(v1.toString(),ztp);
	TypeInference.type_inference2(Normalization.normalize_all(#<apply(`plus,tuple(`v1,`v1))>));
	Tree v2 = new_var();
	type_env.insert(v2.toString(),type);
	Tree utp = TypeInference.type_inference2(Normalization.normalize_all(#<apply(`unit,`v2)>));
	if (unify(utp,ztp) == null)
	type_error(unit,"Wrong type in unit result (expected "+ztp+" found "+utp);
	monoids = monoids.append(#<`name(`type,`plus,`zero,`unit)>);
	}

	/** bind the pattern variables to values */
	private static Environment bind_list ( Tree pattern, MRData value, Environment env ) {
	match pattern {
	case tuple(...ps):
	int i = 0;
	Tuple t = (Tuple)value;
	for ( Tree p: ps )
	env = bind_list(p,t.get(i++),env);
	return env;
	};
	return new Environment(pattern.toString(),value,env);
	}

	/** bind the pattern variables to values */
	private static Environment bind_list ( Tree pattern, Tree src, Environment env ) {
	match pattern {
	case tuple(...ps):
	int i = 0;
	match src {
	case tuple(...es):
	for ( Tree p: ps )
	env = bind_list(p,es.nth(i++),env);
	return env;
	}
	};
	MRData value = Evaluator.evaluator.evalE(src,env);
	env.replace(pattern.toString(),value);
	return env;
	}

	/** the MRQL top-level interfacse to evaluate a single MRQL expression or command */
	public final static void evaluate_top_level ( Tree expr ) {
	if (expr == null)
	return;
	match expr {
	case expression(`e):
	long t = System.currentTimeMillis();
	if (expression(e) != null && !Config.quiet_execution)
	System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
	case assign(`v,`e): assign(v.toString(),e);
	case store(`v,`e):
	long t = System.currentTimeMillis();
	if (distributed_assign(v.toString(),e) != null && !Config.quiet_execution)
	System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
	case dump(`s,`e): // dump stream output
	if (!stream_expression(e))
	fail;
	reset();
	Tree plan = translate_expression(e);
	if (plan == null)
	return;
	Evaluator.evaluator.initialize_query();
	final Tree qt = query_type;
	final String file = s.stringValue();
	Evaluator.evaluator.streaming(plan,null,null,new Function(){
	long i = 0;
	public MRData eval ( final MRData data ) {
	try {
	Evaluator.evaluator.dump(file+"/f"+(i++),qt,data);
	return data;
	} catch (Exception ex) {
	throw new Error("Cannot dump the streaming result to the directory "+file);
	}
	}
	});
	case dump_text(`s,`e): // dump stream output
	if (!stream_expression(e))
	fail;
	reset();
	Tree plan = translate_expression(e);
	if (plan == null)
	return;
	Evaluator.evaluator.initialize_query();
	final Tree qt = query_type;
	final String file = s.stringValue();
	Evaluator.evaluator.streaming(plan,null,null,new Function(){
	long i = 0;
	public MRData eval ( final MRData data ) {
	try {
	Evaluator.evaluator.dump_text(file+"/f"+(i++),qt,data);
	return data;
	} catch (Exception ex) {
	throw new Error("Cannot dump the streaming result to the directory "+file);
	}
	}
	});
	case incr(`e): // incremental stream processing
	Config.incremental = true;
	reset();
	Tree plan = translate_expression(e);
	Config.incremental = false;
	if (plan == null)
	return;
	Evaluator.evaluator.initialize_query();
	match plan {
	case incr(`zero,lambda(`state,`merge),lambda(_,`answer)):
	final Tree qt = make_persistent_type(query_type);
	MRData z = evalE(zero,null);
	final Environment env =bind_list(state,z,null);
	final Tree ans = answer;
	merge = Streaming.streamify(merge);
	Evaluator.evaluator.streaming(#<lambda(`state,`merge)>,env,null,
	new Function() {
	public MRData eval ( final MRData state ) {
	MRData a = Evaluator.evaluator.evalE(ans,env);
	System.out.println(print(a,qt));
	return state;
	}
	});
	};
	case lineage(`e):
	Config.lineage = true;
	boolean quiet = Config.quiet_execution;
	Config.quiet_execution = true;
	long t = System.currentTimeMillis();
	if (expression(e,false) != null && !Config.quiet_execution)
	System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
	Config.lineage = false;
	Config.quiet_execution = quiet;
	case debug(`path,`e):
	Config.debug = true;
	boolean quiet = Config.quiet_execution;
	Config.quiet_execution = true;
	long t = System.currentTimeMillis();
	dump(path.stringValue(),e);
	if (!Config.quiet_execution)
	System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
	Config.debug = false;
	Config.quiet_execution = quiet;
	case dump(`s,`e):
	long t = System.currentTimeMillis();
	dump(s.stringValue(),e);
	if (!Config.quiet_execution)
	System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
	case dump_text(`s,`e):
	long t = System.currentTimeMillis();
	dump_text(s.stringValue(),e);
	if (!Config.quiet_execution)
	System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
	case typedef(`v,`t): typedef(v.toString(),t);
	case datadef(`v,`t): datadef(v.toString(),t);
	case functiondef(`f,params(...p),`tp,`e):
	functiondef(f.toString(),p,tp,e);
	case macrodef(`name,params(...p),`e):
	Translator.global_macros.insert(name.toString(),#<macro(params(...p),`e)>);
	case aggregation(`aggr,`type,`plus,`zero,`unit):
	aggregation(aggr.toString(),type,plus,zero,unit);
	case import(`c):
	ClassImporter.importClass(c.variableValue());
	case import(`c,...l):
	for (Tree m: l)
	ClassImporter.importMethod(c.variableValue(),m.variableValue());
	case include(`file):
	Main.include_file(file.toString());
	case parser(`n,`p):
	try {
	Class<? extends Parser> c = Class.forName(p.toString()).asSubclass(Parser.class);
	DataSource.parserDirectory.put(n.toString(),c);
	} catch (ClassNotFoundException e) {
	throw new Error("Class "+p.toString()+" not found");
	}
	case impure(`fn): // not used
	impure_functions = impure_functions.append(fn);
	case _: throw new Error("Unknown statement: "+expr);
	}
	}
	}