/**
 * 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 java.util.ArrayList;
import java.util.Collection;
import java.util.Arrays;
import java.util.Iterator;
import java.io.*;
import java.net.URI;
import java.net.URL;
import org.apache.mrql.gen.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.flink.util.Collector;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.api.common.io.FileInputFormat;
import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.operators.*;
import org.apache.flink.api.java.*;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
import org.apache.flink.api.java.io.DiscardingOutputFormat;
import org.apache.flink.api.java.functions.*;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.client.program.ContextEnvironment;


/** Evaluates physical plans in Apache Flink mode */
public class FlinkEvaluator extends Evaluator implements Serializable {
    final static URL flink_jar = FlinkEvaluator.class.getProtectionDomain().getCodeSource().getLocation();
    public static ExecutionEnvironment flink_env;
    public static StreamExecutionEnvironment stream_env;
    // an HDFS tmp file used to hold the data source directory information in distributed mode
    static String data_source_dir_name;
    static String master_host = "";
    static int master_port = 6123;
    static String fs_default_name;

    /** initialize the Flink evaluator */
    final public void init ( Configuration conf ) {
        DataSource.parserDirectory = new DataSource.ParserDirectory();
        DataSource.parserDirectory.put("xml",FlinkXMLParser.class);
        DataSource.parserDirectory.put("json",FlinkJsonParser.class);
        DataSource.parserDirectory.put("line",FlinkLineParser.class);
        // Disable bag materialization (it uses the Hadoop HDFS library instead of Flink's)
        Config.max_materialized_bag = Long.MAX_VALUE;
        if (Config.local_mode) {
            flink_env = ExecutionEnvironment.createLocalEnvironment(Config.nodes);
            // curently, the compiler doesn't work in local mode
            Config.compile_functional_arguments = false;
            fs_default_name = "file:///";
	    Plan.conf.set("fs.defaultFS",fs_default_name);
        } else if (Config.distributed_mode) {
            String master_node = System.getenv("FLINK_MASTER");
            if (master_node == null)
                throw new Error("Need to run the Flink application master first: $FLINK_HOME/bin/yarn-session.sh");
            if (!master_node.equals("")) {
		String[] m = master_node.split(":");
		if (m.length != 2)
		    throw new Error("Need both host name and port number for the Flink application master: "+master_node);
		master_host = m[0];
		master_port = Integer.parseInt(m[1]);
	    };
            if (!System.getenv("FS_DEFAULT_NAME").equals("")) {
                fs_default_name = System.getenv("FS_DEFAULT_NAME");
                Plan.conf.set("fs.defaultFS",fs_default_name);
            } else fs_default_name = Plan.conf.get("fs.defaultFS");
            data_source_dir_name = absolute_path("tmp/data_source_dir.txt");
        }
    }

    /** shutdown the Flink evaluator */
    final public void shutdown ( Configuration conf ) {
    }

    final public void initialize_query () {
        try {
            if (!Config.local_mode) {
		if (Config.compile_functional_arguments) {
		    Plan.distribute_compiled_arguments(Plan.conf);
		    if (master_host.equals("")) {
			ContextEnvironment ce = (ContextEnvironment)ExecutionEnvironment.getExecutionEnvironment();
			List<URL> jars = Arrays.asList(flink_jar,new URL("file://"+Plan.conf.get("mrql.jar.path")));
			flink_env = new ContextEnvironment(ce.getClient(),jars,new ArrayList<URL>(),
                                                           FlinkEvaluator.class.getClassLoader(),
                                                           Plan.new_path(Plan.conf));
			flink_env.setParallelism(Config.nodes);
		    } else flink_env = ExecutionEnvironment.createRemoteEnvironment(master_host,master_port,
					     Config.nodes,flink_jar.toURI().getPath(),Plan.conf.get("mrql.jar.path"));
		} else if (master_host.equals(""))
		    flink_env = ExecutionEnvironment.getExecutionEnvironment();
                else flink_env = ExecutionEnvironment.createRemoteEnvironment(master_host,master_port,
					     Config.nodes,flink_jar.toURI().getPath());
	    } else {
		flink_env.setDefaultLocalParallelism(Config.nodes);
                if (Config.stream_window > 0)
                    stream_env = StreamExecutionEnvironment.createLocalEnvironment(Config.nodes)
			               .setBufferTimeout(Config.stream_window);
            };
            if (!Config.info)
                flink_env.getConfig().disableSysoutLogging();
        } catch (Exception ex) {
            throw new Error("Cannot initialize the Flink evaluator: "+ex);
        }
    }

    final public Configuration new_configuration () {
        return new Configuration();
    }

    /** return the FileInputFormat for parsed files (CSV, XML, JSON, etc) */
    final public Class<? extends MRQLFileInputFormat> parsedInputFormat () {
        return FlinkParsedInputFormat.class;
    }

    /** return the FileInputFormat for binary files */
    final public Class<? extends MRQLFileInputFormat> binaryInputFormat () {
        return FlinkSequenceInputFormat.class;
    }

    /** return the FileInputFormat for data generator files */
    final public Class<? extends MRQLFileInputFormat> generatorInputFormat () {
        return FlinkGeneratorInputFormat.class;
    }

    /** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
    final public void streaming ( Tree plan, Environment env, Function f ) {
        FlinkStreaming.evaluate(plan,env,f);
    }

    /** returns the absolute path relative to the directory that contains the MRQL executable */
    public static String absolute_path ( String path) {
        try {
            URI uri = new URI(path);
            if (uri.getScheme() == null)
                if (Config.hadoop_mode && !Config.local_mode)
                    if (uri.toString().startsWith("/"))
                        uri = new URI(fs_default_name+uri);
                    else uri = new URI(fs_default_name+"/user/"+System.getProperty("user.name")+"/"+uri);
                else if (uri.toString().startsWith("/"))
                    uri = new URI("file:"+uri);
                else uri = new URI("file:"+System.getProperty("user.dir")+"/"+uri);
            uri = uri.normalize();
            return uri.toString();
        } catch (Exception ex) {
            throw new Error("Wrong pathname: "+path);
        }
    }

    /** for dumped data to a file, return the MRQL type of the data */
    @Override
    public Tree get_type ( String file ) {
	return super.get_type(absolute_path(file));
    }

    /** dump MRQL data into a Hadoop Sequence file */
    @Override
    public void dump ( String file, Tree type, MRData data ) throws Exception {
	super.dump(absolute_path(file),type,data);
    }

    /** dump MRQL data into a text CVS file */
    @Override
    public void dump_text ( String file, Tree type, MRData data ) throws Exception {
        int ps = Config.max_bag_size_print;
        Config.max_bag_size_print = -1;
        Path path = new Path(absolute_path(file));
        FileSystem fs = path.getFileSystem();
        final PrintStream out = new PrintStream(fs.create(path,true));
        if (data instanceof MR_dataset)
            data = Plan.collect(((MR_dataset)data).dataset());
        if (Translator.collection_type(type)) {
            Tree tp = ((Node)type).children().head();
            if (tp instanceof Node && ((Node)tp).name().equals("tuple")) {
                Trees ts = ((Node)tp).children();
                for ( MRData x: (Bag)data ) {
                    Tuple t = (Tuple)x;
                    out.print(print(t.get((short)0),ts.nth(0)));
                    for ( short i = 1; i < t.size(); i++ )
                        out.print(","+print(t.get(i),ts.nth(i)));
                    out.println();
                }
            } else for ( MRData x: (Bag)data )
                       out.println(print(x,tp));
        } else out.println(print(data,query_type));
        Config.max_bag_size_print = ps;
        out.close();
    }

    private static MRData aggregate ( DataSet<FData> data_set, MRData zero, Tree acc_fnc, Tree merge_fnc, Tree type ) {
        DataSet<FData> d = data_set.flatMap(new FlinkDataSource.reduce_mapper(zero,acc_fnc.toString(),
                                                                              merge_fnc.toString()));
        //d.print();  // needs a datasink (prints the aggregation value)
        d.output(new DiscardingOutputFormat());
        try {
            // due to a Flink bug, we cannot return a custom object (FData) from an accumulator
            String value = FlinkEvaluator.flink_env.execute("MRQL aggregator").getAccumulatorResult("reducer");
            Tree tp = TopLevel.query_type;
            MRData res = MRQL.query(value+" as "+type);  // parse the value using the MRQL query parser (ugly)
            TopLevel.query_type = tp;
            return res;
        } catch (Exception ex) {
            throw new Error("Evaluation error during aggregation: "+ex);
        }
    }

    /* broadcast the loop value to all nodes */
    public static final class set_environment extends RichMapFunction<FData,FData> {
        String var;

        set_environment ( String v ) { var = v; }

        @Override
        public void open ( org.apache.flink.configuration.Configuration parameters ) throws Exception {
            final Bag value = bag(getRuntimeContext().getBroadcastVariable(var));
            Interpreter.new_global_binding(var.toString(),value);
        }

        @Override
        public FData map ( FData value ) throws Exception {
            return new FData(((Tuple)value.data()).first());
        }
    }

    private static Bag bag ( Collection<Object> values ) {
        Bag s = new Bag();
        for ( Object value: values )
            s.add(((FData)value).data());
        return s;
    }

    /* convert a Bag into a DataSet */
    private static DataSet<FData> dataset ( Bag s ) {
        ArrayList<FData> a = new ArrayList(s.size());
        for ( MRData x: s )
            a.add(new FData(x));
        return flink_env.fromCollection(a);
    }

    /** Coerce a persistent collection to a Bag */
    public Bag toBag ( MRData data ) {
        try {
            if (data instanceof MR_flink) {
                final Iterator<FData> i = ((MR_flink)data).flink().collect().iterator();
                return new Bag(new BagIterator() {
                        public MRData next () {
                            return i.next().data();
                        }
                        public boolean hasNext () {
                            return i.hasNext();
                        }
                    });
            };
            return (Bag)data;
        } catch (Exception ex) {
            throw new Error("Cannot coerce "+data+" to a Bag: "+ex);
        }
    }

    /** The Aggregate physical operator
     * @param merge  the accumulator function from (T,T) to T
     * @param zero  the zero element of type T
     * @param plan the plan that constructs the dataset that contains the bag of values {T}
     * @param env contains bindings fro variables to values (MRData)
     * @return the aggregation result of type T
     */
    final public MRData aggregate ( Tree merge,
                                    Tree zero,
                                    Tree plan,
                                    Environment env ) throws Exception {
        final MRData z = evalE(zero,env);
        Tree type = TypeInference.type_inference(zero);
        match plan {
        case repeat(lambda(`v,`b),`s,`n):
            // Special case for Flink: repetition where the loop result must be shared among nodes
            final int max_num = ((MR_int)evalE(n,null)).get();
            final Bag init = (Bag)evalE(s,env);
            Interpreter.new_global_binding(v.toString(),init);
            final DataSet<FData> ds = dataset(init);
            final IterativeDataSet<FData> startOfIteration = ds.iterate(max_num);
            final DataSet<FData> step = eval(b,env);
            final DataSet<FData> toBeFedBack = step.map(new set_environment(v.toString()))
                                                   .withBroadcastSet(startOfIteration,v.toString());
            // Flink doesn't allow termination criteria for kmeans-like iterations:
            //   (java.lang.IllegalStateException: Error: Iterative task without a single iterative input)
            //final DataSet<FData> terminationCriterion = step.filter(new repeat_filter());
            final DataSet<FData> res = startOfIteration.closeWith(toBeFedBack);//,terminationCriterion);
            String path = absolute_path(Plan.new_path(Plan.conf));
            res.write(new FlinkOutputFormat(),path);
            flink_env.execute("MRQL repeat");
            Bag bs = new Bag();
            for ( MRData x: new FlinkDataSource(res,path,false).take(-1) )
                bs.add(x);
            return bs;
        case AggregateMap(`m,`acc,_,`s):
            final DataSet<FData> ds = eval(s,env).mapPartition(new aggregate_partition(m,acc,z));
            return aggregate(ds,z,merge,merge,type);
        case MapAggregateReduce(`m,`r,`acc,_,`s,`o):
            if (acc.equals(#<null>))
                fail;
            final DataSet<FData> ds = eval(s,env).flatMap(cmap_fnc(m));
            final DataSet<FData> gs = ds.groupBy(new KeySelector<FData,FData>() {
                                                      @Override
                                                      public FData getKey ( FData value ) {
                                                          return new FData(((Tuple)value.data()).first());
                                                      }
                                             }).reduceGroup(new groupBy_combiner_reducer(r,acc,z));
            return aggregate(gs,z,merge,merge,type);
        case CrossAggregateProduct(`mx,`my,`r,`acc,_,`x,`y):
            if (acc.equals(#<null>))
                fail;
            final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
            final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
            final DataSet<FData> ds = xs.cross(ys).with(new cross_combiner_reducer(r,acc,z));
            return aggregate(ds,z,merge,merge,type);
        case MapAggregateReduce2(`mx,`my,`r,`acc,_,`x,`y,`o):
            if (acc.equals(#<null>))
                fail;
            final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
            final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
            final DataSet<FData> ds = xs.coGroup(ys)
                .where(new join_key())
                .equalTo(new join_key())
                .with(new join_combiner_reducer(r,acc,z));
            return aggregate(ds,z,merge,merge,type);
        case MapAggregateJoin(`mx,`my,`r,`acc,_,`x,`y):
            if (acc.equals(#<null>))
                fail;
            return aggregate(evalD(#<MapJoin(`mx,`my,`r,`x,`y)>,env),z,acc,merge,type);
        };
        throw new Error("Unrecognized aggregation: "+plan);
    }

    /** Evaluate a loop a fixed number of times. Doesn't use Flink iterations */
    final public Tuple loop ( Tree e, Environment env ) throws Exception {
        match e {
        case Loop(lambda(tuple(...vs),tuple(...bs)),tuple(...ss),`num):
            final int limit = ((MR_int)evalE(num,env)).get();
            MR_flink[] s = new MR_flink[vs.length()];
            for ( int i = 0; i < vs.length(); i++ )
                s[i] = new MR_flink(eval(ss.nth(i),env));
            for ( int n = 0; n < limit; n++ ) {
                Environment nenv = env;
                for ( int i = 0; i < vs.length(); i ++ )
                    nenv = new Environment(vs.nth(i).toString(),s[i],nenv);
                for ( int i = 0; i < vs.length(); i ++ )
                    s[i] = new MR_flink(eval(bs.nth(i),nenv));
            };
            return new Tuple(s);
        };
        throw new Error("Wrong Loop format: "+e);
    }

    private static boolean needs_sorting ( Tree e ) {
        match e {
        case MapReduce(`m,`r,`s,true): return true;
        case MapCombineReduce(`m,`c,`r,`s,true): return true;
        };
        return false;
    }

    /** Evaluate an MRQL physical plan and print tracing info
     * @param e the physical plan
     * @param env contains bindings fro variables to values (MRData)
     * @return an MRQL DataSet (stored in HDFS)
     */
    final public org.apache.mrql.DataSet eval ( final Tree e,
                                                final Environment env,
                                                final String counter ) {
        DataSet<FData> d = eval(e,env);
        try {
            String path = absolute_path(Plan.new_path(Plan.conf));
            d.write(new FlinkOutputFormat(),path);
            flink_env.execute("MRQL query");
            return new org.apache.mrql.DataSet(new FlinkDataSource(d,path,needs_sorting(e)),0L,0L);
        } catch (Exception ex) {
            System.err.println(ex.getMessage());
            throw new Error("Cannot evaluate the query: "+e);
        }
    }

    /** Evaluate an MRQL physical plan using Flink and print tracing info
     * @param e the physical plan
     * @param env contains bindings from variables to values (MRData)
     * @return a Flink DataSet
     */
    final public DataSet<FData> eval ( final Tree e, final Environment env ) {
        if (Config.trace_execution) {
            tab_count += 3;
            System.out.println(tabs(tab_count)+print_query(e));
        };
        final DataSet<FData> res = evalD(e,env);
        if (Config.trace_execution) 
            try {
                System.out.println(tabs(tab_count)+"->");
                res.first(Config.max_bag_size_print).print();
                tab_count -= 3;
            } catch (Exception ex) {
                System.err.println(ex.getMessage());
                throw new Error("Cannot collect the operator output: "+e);
            };
        return res;
    }

    /* convert an MRQL lambda to a Flink Function */
    private static RichFlatMapFunction<FData,FData> cmap_fnc ( final Tree fnc ) {
        final Function f = evalF(fnc,null);
        return new RichFlatMapFunction<FData,FData>() {
            @Override
            public void flatMap ( FData value, Collector<FData> out ) throws Exception {
                for ( MRData e: (Bag)f.eval(value.data()) )
                    out.collect(new FData(e));
            }
        };
    }

    /* convert the stream of pairs to a bag of their second element */
    private static Bag bag ( Iterable<FData> values ) {
        Bag s = new Bag();
        for ( FData value: values )
            s.add(((Tuple)value.data()).second());
        return s;
    }

    /* group-by s and then reduce by reduce_fnc (optional: use combine_fnc) */
    private static DataSet<FData> groupBy ( DataSet<FData> s, Tree combine_fnc, Tree reduce_fnc ) {
        final Function combiner = (combine_fnc == null) ? null : evalF(combine_fnc,null);
        final Function reducer = evalF(reduce_fnc,null);
        DataSet<FData> ds = s;
        if (combiner != null) {
            ds = ds.groupBy(new KeySelector<FData,FData>() {
                    @Override
                    public FData getKey ( FData value ) {
                        return new FData(((Tuple)value.data()).first());
                    }
            }).combineGroup(new RichGroupCombineFunction<FData,FData>() {
                @Override
                public void combine ( Iterable<FData> values, Collector<FData> out ) throws Exception {
                    Bag c = null;
                    MRData key = null;
                    for ( FData v: values )
                        if (c == null) {
                            c = new Bag(((Tuple)v.data()).second());
                            key = ((Tuple)v.data()).first();
                        } else
                            c = (Bag)combiner.eval(new Tuple(key,c.add_element(((Tuple)v.data()).second())));
                    for ( MRData x: c )
                        out.collect(new FData(new Tuple(key,x)));
                }
           });
        };
        return ds.groupBy(new KeySelector<FData,FData>() {
                @Override
                public FData getKey ( FData value ) {
                    return new FData(((Tuple)value.data()).first());
                }
            }).reduceGroup(new GroupReduceFunction<FData,FData>() {
                @Override
                public void reduce ( final Iterable<FData> values, Collector<FData> out ) {
                    Bag s = new Bag();
                    MRData key = null;
                    for ( FData value: values ) {
                        Tuple t = (Tuple)value.data();
                        key = t.first();
                        s.add(t.second());
                    };
                    for ( MRData v: (Bag)reducer.eval(new Tuple(key,s)) )
                        out.collect(new FData(v));
                }
           });
    }

    /* group-by and sort s and then reduce by reduce_fnc (optional: use combine_fnc) */
    private static DataSet<FData> sortBy ( DataSet<FData> s, Tree combine_fnc, Tree reduce_fnc ) {
        final Function combiner = (combine_fnc == null) ? null : evalF(combine_fnc,null);
        final Function reducer = evalF(reduce_fnc,null);
        DataSet<Tuple2<FData,FData>> ds = s.map(new RichMapFunction<FData,Tuple2<FData,FData>>() {
                @Override
                public Tuple2<FData,FData> map ( FData value ) {
                    Tuple t = (Tuple)value.data();
                    return new Tuple2<FData,FData>(new FData(t.first()),new FData(t.second()));
                }
            });
        if (combiner != null) {
            ds = ds.groupBy(0).combineGroup(new GroupCombineFunction<Tuple2<FData,FData>,Tuple2<FData,FData>>() {
                    @Override
                    public void combine ( Iterable<Tuple2<FData,FData>> values,
                                          Collector<Tuple2<FData,FData>> out ) throws Exception {
                        Bag c = null;
                        FData key = null;
                        for ( Tuple2<FData,FData> v: values )
                            if (c == null) {
                                c = new Bag(v.f1.data());
                                key = v.f0;
                            } else
                                c = (Bag)combiner.eval(new Tuple(key.data(),c.add_element(v.f0.data())));
                        for ( MRData x: c )
                            out.collect(new Tuple2<FData,FData>(key,new FData(x)));
                    }
                });
        };
        return ds.groupBy(0).sortGroup(0,Order.ASCENDING)
            .reduceGroup(new RichGroupReduceFunction<Tuple2<FData,FData>,FData>() {
                    @Override
                    public void reduce ( final Iterable<Tuple2<FData,FData>> values, Collector<FData> out ) {
                        Bag s = new Bag();
                        MRData key = null;
                        for ( Tuple2<FData,FData> value: values ) {
                            key = value.f0.data();
                            s.add(value.f1.data());
                        };
                        for ( MRData v: (Bag)reducer.eval(new Tuple(key,s)) )
                            out.collect(new FData(new Tuple(key,v)));
                    }
               });
    }

    public static final class join_key implements KeySelector<FData,FData> {
        @Override
        public FData getKey ( FData value ) {
            return new FData(((Tuple)value.data()).first());
        }
    }

    public static final class aggregate_partition extends RichMapPartitionFunction<FData,FData> {
        final Function m;
        final Function acc;
        final MRData zero;

        aggregate_partition ( Tree m, Tree acc, MRData zero ) {
            this.m = evalF(m,null);
            this.acc = evalF(acc,null);
            this.zero = zero;
        }

        @Override
        public void mapPartition ( Iterable<FData> values, Collector<FData> out ) throws Exception {
            MRData aggregated_value = zero;
            for ( FData value: values )
                for ( MRData v: (Bag)m.eval(value.data()) )
                    aggregated_value = acc.eval(new Tuple(aggregated_value,v));
            out.collect(new FData(aggregated_value));
        }
    }

    public static final class groupBy_combiner_reducer extends RichGroupReduceFunction<FData,FData> {
        final Function r;
        final Function acc;
        final MRData zero;

        groupBy_combiner_reducer ( Tree r, Tree acc, MRData zero ) {
            this.r = evalF(r,null);
            this.acc = evalF(acc,null);
            this.zero = zero;
        }

        @Override
        public void reduce ( final Iterable<FData> values, Collector<FData> out ) {
            MRData value = zero;
            for ( MRData v: (Bag)r.eval(new Tuple(bag(values))) )
                value = acc.eval(new Tuple(value,v));
            out.collect(new FData(value));
        }
    }

    public static final class join_reducer extends RichCoGroupFunction<FData,FData,FData> {
        final Function f;

        join_reducer ( Tree fnc ) {
            f = evalF(fnc,null);
        }

        @Override
        public void coGroup ( Iterable<FData> xs, Iterable<FData> ys, Collector<FData> out ) {
            for ( MRData v: (Bag)f.eval(new Tuple(bag(xs),bag(ys))) )
                out.collect(new FData(v));
        }
    }

    public static final class join_combiner_reducer extends RichCoGroupFunction<FData,FData,FData> {
        final Function r;
        final Function acc;
        final MRData zero;

        join_combiner_reducer ( Tree r, Tree acc, MRData zero ) {
            this.r = evalF(r,null);
            this.acc = evalF(acc,null);
            this.zero = zero;
        }

        @Override
        public void coGroup ( Iterable<FData> xs, Iterable<FData> ys, Collector<FData> out ) {
            MRData value = zero;
            for ( MRData v: (Bag)r.eval(new Tuple(bag(xs),bag(ys))) )
                value = acc.eval(new Tuple(value,v));
            out.collect(new FData(value));
        }
    }

    public static final class cross_reducer extends RichCrossFunction<FData,FData,FData> {
        final Function f;

        cross_reducer ( Tree fnc ) {
            f = evalF(fnc,null);
        }

        @Override
        public FData cross ( FData x, FData y ) throws Exception {
            return new FData(f.eval(new Tuple(x.data(),y.data())));
        }
    }

    public static final class cross_combiner_reducer extends RichCrossFunction<FData,FData,FData> {
        final Function r;
        final Function acc;
        final MRData zero;

        cross_combiner_reducer ( Tree r, Tree acc, MRData zero ) {
            this.r = evalF(r,null);
            this.acc = evalF(acc,null);
            this.zero = zero;
        }

        @Override
        public FData cross ( FData x, FData y ) throws Exception {
            return new FData(acc.eval(new Tuple(zero,r.eval(new Tuple(x.data(),y.data())))));
        }
    }

    public static final class mapjoin_reducer extends RichJoinFunction<FData,FData,FData> {
        final Function f;

        mapjoin_reducer ( Tree fnc ) {
            f = evalF(fnc,null);
        }

        @Override
        public FData join ( FData x, FData y ) throws Exception {
            return new FData(f.eval(new Tuple(x.data(),y.data())));
        }
    }

    /** restore the global bindings at the client node */
    public static final class restore_global_functions extends RichMapFunction<FData,FData> {
        Environment client_env;

        restore_global_functions () {
            client_env = Interpreter.global_env;
        }

        @Override
        public void open ( org.apache.flink.configuration.Configuration parameters ) throws Exception {
            Interpreter.set_global_bindings(client_env);
        }

        @Override
        public FData map ( FData value ) throws Exception { return value; }
    }

    public static final class generator_mapper extends RichMapFunction<Long,FData> {
        @Override
        public FData map ( Long value ) throws Exception {
            return new FData(new MR_long(value));
        }
    }

    public static final class parsed_mapper extends RichFlatMapFunction<String,FData> {
        final Parser parser;

        parsed_mapper ( Class<? extends Parser> parser_class ) throws Exception {
            parser = parser_class.newInstance();
        }

        @Override
        public void flatMap ( String value, Collector<FData> out ) throws Exception {
            for ( MRData e: parser.parse(value) )
                out.collect(new FData(e));
        }
    }

    public static final class group_join_left extends RichFlatMapFunction<FData,FData> {
        final MRData one = new MR_byte(1);
        final Function fgx;
        final int m, n;

        group_join_left ( Tree gx, int m, int n ) {
            this.m = m;
            this.n = n;
            fgx = evalF(gx,null);
        }

        @Override
        public void flatMap ( FData value, Collector<FData> out ) throws Exception {
            for ( int i = 0; i < n; i++ ) {
                final MRData key = new MR_int((fgx.eval(value.data()).hashCode() % m)+m*i);
                out.collect(new FData(new Tuple(key,new Tuple(one,value.data()))));
            }
        }
    }

    public static final class group_join_right extends RichFlatMapFunction<FData,FData> {
        final MRData two = new MR_byte(2);
        final Function fgy;
        final int m, n;

        group_join_right ( Tree gy, int m, int n ) {
            this.m = m;
            this.n = n;
            fgy = evalF(gy,null);
        }

        @Override
        public void flatMap ( FData value, Collector<FData> out ) throws Exception {
            for ( int j = 0; j < m; j++ ) {
                final MRData key = new MR_int((fgy.eval(value.data()).hashCode() % n)*m+j);
                out.collect(new FData(new Tuple(key,new Tuple(two,value.data()))));
            }
        }
    }

    public static final class group_join_reducer extends RichGroupReduceFunction<FData,FData> {
        final Function fkx, fky, fgx, fgy, fc, fr;
        final MRData z;

        group_join_reducer ( Tree kx, Tree ky, Tree gx, Tree gy, Tree c, Tree r, Tree zero ) {
            fkx = evalF(kx,null); fky = evalF(ky,null); fgx = evalF(gx,null); fgy = evalF(gy,null);
            fr = evalF(r,null); fc = evalF(c,null); z = evalE(zero,null);
        }

        @Override
        public void reduce ( final Iterable<FData> values, Collector<FData> out ) {
            Bag xb = new Bag();
            Bag yb = new Bag();
            for ( FData value: values ) {
                final Tuple t = (Tuple)value.data();
                final Tuple p = (Tuple)t.second();
                if (((MR_byte)p.first()).get() == 1)
                    xb.add(new Tuple(t.first(),p.second()));
                else yb.add(new Tuple(t.first(),p.second()));
            }
            final Bag b = MapReduceAlgebra.mergeGroupByJoin(fkx,fky,fgx,fgy,fc,z,fr,xb,yb);
            for ( MRData v: b )
                out.collect(new FData(v));
        };
    }

    public static final class repeat_values extends RichMapFunction<FData,FData> {
        @Override
        public FData map ( FData value ) throws Exception {
            return new FData(((Tuple)value.data()).first());
        }
    }

    public static final class repeat_filter extends RichFilterFunction<FData> {
        @Override
        public boolean filter ( FData value ) throws Exception {
            return ((MR_bool)(((Tuple)value.data()).second())).get();
        }
    }

    /** Evaluate a MRQL physical plan
     * @param e the physical plan
     * @param env contains bindings from variables to values (MRData)
     * @return a Flink DataSet
     */
    final private DataSet<FData> evalD ( final Tree e, final Environment env ) {
        try {
            match e {
            case MapAggregateReduce(`m,`r,null,_,`s,`o):
                return evalD(#<MapReduce(`m,`r,`s,`o)>,env);
            case CrossAggregateProduct(`mx,`my,`r,null,_,`x,`y):
                return evalD(#<CrossProduct(`mx,`my,`r,`x,`y)>,env);
            case MapAggregateReduce2(`mx,`my,`r,null,_,`x,`y,`o):
                return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env);
            case MapAggregateJoin(`mx,`my,`r,null,_,`x,`y):
                return evalD(#<MapJoin(`mx,`my,`r,`x,`y)>,env);
            case cMap(`f,`s):
                return eval(s,env).flatMap(cmap_fnc(f));
            case MapReduce(`m,`r,`s,true):
                return sortBy(eval(s,env).flatMap(cmap_fnc(m)),null,r);
            case MapReduce(`m,`r,`s,`o):
                return groupBy(eval(s,env).flatMap(cmap_fnc(m)),null,r);
            case MapCombineReduce(`m,`c,`r,`s,true):
                return sortBy(eval(s,env).flatMap(cmap_fnc(m)),c,r);
            case MapCombineReduce(`m,`c,`r,`s,`o):
                return groupBy(eval(s,env).flatMap(cmap_fnc(m)),c,r);
            case MapCombineReduce2(`mx,`my,`c,`r,`x,`y,`o):
                return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env);
            case MapReduce2(`mx,`my,`r,`x,`y,`o):
                final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
                final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
                final DataSet<FData> ds = xs.coGroup(ys)
                    .where(new join_key())
                    .equalTo(new join_key())
                    .with(new join_reducer(r));
                if (o.equals(#<true>))
                    return sortBy(ds,null,#<lambda(x,bag(x))>);
                else return ds;
            case MapJoin(`mx,`my,`r,`x,`y):
                final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
                final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
                return xs.joinWithTiny(ys)
                    .where(new join_key())
                    .equalTo(new join_key())
                    .with(new mapjoin_reducer(r));
            case CrossProduct(`mx,`my,`r,`x,`y):
                final DataSet<FData> xs = eval(x,env).flatMap(cmap_fnc(mx));
                final DataSet<FData> ys = eval(y,env).flatMap(cmap_fnc(my));
                return xs.cross(ys).with(new cross_reducer(r))
                         .flatMap(cmap_fnc(#<lambda(x,x)>));
            case GroupByJoin(`kx,`ky,`gx,`gy,`acc,`zero,`r,`x,`y,`o):
                final int n = Math.max(2,(int)Math.floor(Math.sqrt(Config.nodes)));
                final int m = n;
                // System.err.println("Using a groupBy join on a "+n+"*"+m+" grid of partitions");
                final DataSet<FData> xs = eval(x,env).flatMap(new group_join_left(gx,m,n));
                final DataSet<FData> ys = eval(y,env).flatMap(new group_join_right(gy,m,n));
                return xs.union(ys).groupBy(new join_key())
                         .reduceGroup(new group_join_reducer(kx,ky,gx,gy,acc,r,zero));
            case Merge(`x,`y):
                return eval(x,env).union(eval(y,env));
            case Generator(`min,`max,`size):
                final long from = ((MR_long)evalE(min,null)).get();
                final long to = ((MR_long)evalE(max,null)).get();
                return flink_env.generateSequence(from,to).map(new generator_mapper())
                                .map(new restore_global_functions());
            case BinarySource(`file,_):
                final String path = absolute_path(((MR_string)evalE(file,null)).get());
                new BinaryDataSource(path,Plan.conf);
                final FileInputFormat<FData> sf = new FlinkSequenceInputFormat().inputFormat(path);
                return flink_env.createInput(sf).map(new restore_global_functions());
            case ParsedSource(`parser,`file,...args):
                final String path = absolute_path(((MR_string)evalE(file,null)).get());
                final Class<? extends Parser> parser_class = DataSource.parserDirectory.get(parser.toString());
                if (parser_class == null)
                    throw new Error("Unknown parser: "+parser);
                final DataSource ds = new FlinkParsedDataSource(path,parser_class,args);
                return flink_env.readFile(new FlinkParsedInputFormat.ParsedInputFormat(path),path)
                                .map(new restore_global_functions());
            case Repeat(lambda(`v,`b),`s,`n):
                final int max_num = ((MR_int)evalE(n,null)).get();
                final IterativeDataSet<FData> startOfIteration = eval(s,env).iterate(max_num);
                final DataSet<FData> step = eval(b,new Environment(v.toString(),new MR_flink(startOfIteration),env));
                final DataSet<FData> toBeFedBack = step.map(new repeat_values());
                final DataSet<FData> terminationCriterion = step.filter(new repeat_filter());
                return startOfIteration.closeWith(toBeFedBack,terminationCriterion);
            case Closure(lambda(`v,`b),`s,`n):
                final int max_num = ((MR_int)evalE(n,null)).get();
                IterativeDataSet<FData> startOfIteration = eval(s,env).iterate(max_num);
                DataSet<FData> toBeFedBack = eval(b,new Environment(v.toString(),new MR_flink(startOfIteration),env));
                return startOfIteration.closeWith(toBeFedBack);
            case let(`v,DataSetCollect(`s),`body):
                MRData x = evalE(s,env);
                if (x instanceof MR_dataset)
                    x = new Bag(((MR_dataset)x).dataset().take(Integer.MAX_VALUE));
                else if (x instanceof Bag)
                    ((Bag)x).materialize();
                new_distributed_binding(v.toString(),x);
                return eval(body,new Environment(v.toString(),x,env));
            case let(`v,`u,`body):
                return eval(body,new Environment(v.toString(),evalE(u,null),env));
            case Let(`v,`u,`body):
                return eval(body,new Environment(v.toString(),new MR_flink(eval(u,env)),env));
            case If(`c,`x,`y):
                if (((MR_bool)evalE(c,env)).get())
                    return eval(x,env);
                else return eval(y,env);
           case nth(`u,`n):
                MRData x = ((Tuple)evalE(u,env)).get((int)n.longValue());
                if (x instanceof MR_flink)
                    return ((MR_flink)x).flink();
                throw new Error("Evaluation error in: "+print_query(e));
            case `v:
                if (!v.is_variable())
                    fail;
                MRData x = variable_lookup(v.toString(),env);
                if (x != null)
                    if (x instanceof MR_flink)
                        return ((MR_flink)x).flink();
                    else new Error("Variable "+v+" is of type: "+x);
                x = variable_lookup(v.toString(),global_env);
                if (x != null)
                    if (x instanceof MR_flink)
                        return ((MR_flink)x).flink();
                    else if (x instanceof MR_dataset)
                        return dataset(new Bag(((MR_dataset)x).dataset.take(Integer.MAX_VALUE)))
			    .map(new restore_global_functions());
                    else new Error("Variable "+v+" is of type: "+x);
                throw new Error("Variable "+v+" is not bound");
            };
            throw new Error("Cannot evaluate the Flink plan: "+e);
        } catch (Error msg) {
            if (!Config.trace)
                throw new Error(msg.getMessage());
            System.err.println(msg.getMessage());
            msg.printStackTrace();
            throw new Error("Evaluation error in: "+print_query(e));
        } catch (Exception ex) {
            System.err.println(ex.getMessage());
            ex.printStackTrace();
            throw new Error("Evaluation error in: "+print_query(e));
        }
    }
}