/**
 * 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 org.apache.mrql.gen.*;
import java.io.*;
import java.util.*;


/** Evaluation of MRQL algebra expressions in memory */
final public class MapReduceAlgebra {

    /** eager concat-map (not used) */
    private static Bag cmap_eager ( final Function f, final Bag s ) {
        Bag res = new Bag();
        for ( MRData e: s )
            res.addAll((Bag)f.eval(e));
        return res;
    }

    /** lazy concat-map (stream-based)
     * @param f a function from a to {b}
     * @param s the input of type {a}
     * @return a value of type {b}
     */
    public static Bag cmap ( final Function f, final Bag s ) {
        final Iterator<MRData> si = s.iterator();
        return new Bag(new BagIterator() {
                Iterator<MRData> data = null;
                boolean more = false;
                public boolean hasNext () {
                    if (data == null) {
                        while (!more && si.hasNext()) {
                            data = ((Bag)f.eval(si.next())).iterator();
                            more = data.hasNext();
                        }
                    } else {
                        if (more) {
                            more = data.hasNext();
                            if (more)
                                return true;
                        };
                        while (!more && si.hasNext()) {
                            data = ((Bag)f.eval(si.next())).iterator();
                            more = data.hasNext();
                        }
                    };
                    return more;
                }
                public MRData next () {
                    return data.next();
                }
            });
    }

    /** lazy map
     * @param f a function from a to b
     * @param s the input of type {a}
     * @return a value of type {b}
     */
    public static Bag map ( final Function f, final Bag s ) {
        final Iterator<MRData> si = s.iterator();
        return new Bag(new BagIterator() {
                public boolean hasNext () { return si.hasNext(); }
                public MRData next () { return f.eval(si.next()); }
            });
    }

    /** lazy filter combined with a map
     * @param p a function from a to boolean
     * @param f a function from a to b
     * @param s the input of type {a}
     * @return a value of type {b}
     */
    public static Bag filter ( final Function p, final Function f, final Bag s ) {
        final Iterator<MRData> si = s.iterator();
        return new Bag(new BagIterator() {
                MRData data = null;
                public boolean hasNext () {
                    while (si.hasNext()) {
                        data = si.next();
                        if (((MR_bool)p.eval(data)).get())
                            return true;
                    };
                    return false;
                }
                public MRData next () { return f.eval(data); }
            });
    }

    /** general reduction using an accumulator function and a zero element
     * @param acc a function from (a,b) to b
     * @param zero a value of type b
     * @param s the input of type {a}
     * @return a value of type b
     */
    public static MRData fold ( final Function acc, final MRData zero, final Bag s ) {
        MRData result = zero;
        for ( MRData value: s )
            result = acc.eval(new Tuple(value,result));
        return result;
    }

    /** strict group-by
     * @param s the input of type {(a,b)}
     * @return a value of type {(a,{b})}
     */
    public static Bag groupBy ( Bag s ) {
        Bag res = new Bag();
        s.sort();
        MRData last = null;
        Bag group = new Bag();
        for ( MRData e: s) {
            final Tuple p = (Tuple)e;
            if (last != null && p.first().equals(last))
                group.add(p.second());
            else {
                if (last != null) {
                    group.trim();
                    res.add(new Tuple(last,group));
                };
                last = p.first();
                group = new Bag();
                group.add(p.second());
            }
        };
        if (last != null) {
            group.trim();
            res.add(new Tuple(last,group));
        };
        //res.trim();
        return res;
    }

    /** lazy group-by (not used) */
    private static Bag groupBy_lazy ( Bag s ) {
        s.sort();
        final Iterator<MRData> it = s.iterator();
        return new Bag(new BagIterator() {
                MRData last = null;
                MRData data = null;
                Bag group = new Bag();
                public boolean hasNext () {
                    while (it.hasNext()) {
                        final Tuple p = (Tuple)it.next();
                        if (last != null && p.first().equals(last))
                            group.add(p.second());
                        else if (last != null) {
                            group.trim();
                            data = new Tuple(last,group);
                            last = p.first();
                            group = new Bag();
                            group.add(p.second());
                            return true;
                        } else {
                            last = p.first();
                            group = new Bag();
                            group.add(p.second());
                        }
                    };
                    if (last != null) {
                        group.trim();
                        data = new Tuple(last,group);
                        last = null;
                        return true;
                    };
                    return false;
                }
                public MRData next () {
                    return data;
                }
            });
    }

    /** the MapReduce operation
     * @param m a map function from a to {(k,b)}
     * @param r a reduce function from (k,{b}) to {c}
     * @param s the input of type {a}
     * @return a value of type {c}
     */
    public static Bag mapReduce ( final Function m, final Function r, final Bag s ) {
        return cmap(r,groupBy(cmap(m,s)));
    }

    /** Not used: use mapReduce2 instead */
    private static Bag join ( final Function kx, final Function ky, final Function f,
                              final Bag X, final Bag Y ) {
        return cmap(new Function() {
                public Bag eval ( final MRData e ) {
                    final Tuple p = (Tuple)e;
                    return (Bag)f.eval(new Tuple(p.second(),
                                                 cmap(new Function() {
                                                         public Bag eval ( final MRData y ) {
                                                             return (ky.eval(y).equals(p.first()))
                                                                     ? new Bag(y)
                                                                     : new Bag();
                                                         } }, Y))); }
            },
            groupBy(cmap(new Function() {
                    public Bag eval ( final MRData x ) {
                        return new Bag(new Tuple(kx.eval(x),x));
                    } }, X)));
    }

    /** A hash-based equi-join
     * @param kx left key function from a to k
     * @param ky right key function from b to k
     * @param f reducer from (a,b) to c
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {c}
     */
    public static Bag hash_join ( final Function kx, final Function ky, final Function f,
                                  final Bag X, final Bag Y ) {
        Hashtable<MRData,Bag> hashTable = new Hashtable<MRData,Bag>(1000);
        for ( MRData x: X ) {
            MRData key = kx.eval(x);
            Bag old = hashTable.get(key);
            if (old == null)
                hashTable.put(key,new Bag(x));
            else old.add(x);
        };
        Bag res = new Bag();
        for ( MRData y: Y ) {
            MRData key = ky.eval(y);
            Bag match = hashTable.get(key);
            if (match != null)
                for ( MRData x: match )
                    res.add(f.eval(new Tuple(x,y)));
        };
        return res;
    }

    /** A cross-product
     * @param mx left map function from a to {a'}
     * @param my right key function from b to {b'}
     * @param r reducer from (a',b') to {c}
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {c}
     */
    public static Bag crossProduct ( final Function mx, final Function my, final Function r,
                                     final Bag X, final Bag Y ) {
        Bag a = new Bag();
        for ( MRData y: Y )
            for ( MRData v: (Bag)my.eval(y) )
                a.add(v);
        Bag b = new Bag();
        for ( MRData x: X )
            for ( MRData xx: (Bag)mx.eval(x) )
                for ( MRData y: a )
                    for ( MRData v: (Bag)r.eval(new Tuple(xx,y)) )
                        b.add(v);
        return b;
    }

    /** A map-reduce operation with two mappers (a join)
     * @param mx left map function from a to {(k,a')}
     * @param my right key function from b to {(k,b')}
     * @param r reducer from ({a'},{b'}) to {c}
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {c}
     */
    public static Bag mapReduce2 ( final Function mx,   // left mapper
                                   final Function my,   // right mapper
                                   final Function r,    // reducer
                                   final Bag X, final Bag Y ) {
        final Bag left = cmap(new Function() {
                public Bag eval ( final MRData x ) {
                    return cmap(new Function() {
                            public Bag eval ( final MRData e ) {
                                final Tuple p = (Tuple)e;
                                return new Bag(new Tuple(p.first(),
                                                         new Tuple(new MR_byte(1),p.second())));
                            } }, (Bag)mx.eval(x));
                } }, X);
        final Bag right = cmap(new Function() {
                public Bag eval ( final MRData y ) {
                    return cmap(new Function() {
                            public Bag eval ( final MRData e ) {
                                final Tuple p = (Tuple)e;
                                return new Bag(new Tuple(p.first(),
                                                         new Tuple(new MR_byte(2),p.second())));
                            } }, (Bag)my.eval(y));
                } }, Y);
        final Iterator<MRData> li = left.iterator();
        final Iterator<MRData> ri = right.iterator();
        final Bag mix = new Bag(new BagIterator () {
                MRData data;
                public boolean hasNext () {
                    if (li.hasNext()) {
                        data = li.next();
                        return true;
                    } else if (ri.hasNext()) {
                        data = ri.next();
                        return true;
                    } else return false;
                }
                public MRData next () {
                    return data;
                }
            });
        return cmap(new Function() {
                public Bag eval ( final MRData e ) {
                    final Tuple p = (Tuple)e;
                    final Bag xs = cmap(new Function() {
                            public Bag eval ( final MRData e ) {
                                final Tuple q = (Tuple)e;
                                return (((MR_byte)q.first()).get() == 1)
                                        ? new Bag(q.second())
                                        : new Bag();
                            } }, (Bag)p.second());
                    final Bag ys = cmap(new Function() {
                            public Bag eval ( final MRData e ) {
                                final Tuple q = (Tuple)e;
                                return (((MR_byte)q.first()).get() == 2)
                                        ? new Bag(q.second())
                                        : new Bag();
                            } }, (Bag)p.second());
                    xs.materialize();
                    ys.materialize();
                    return (Bag)r.eval(new Tuple(xs,ys));
                } }, groupBy(mix));
    }

    /** The fragment-replicate join (map-side join)
     * @param kx left key function from a to k
     * @param ky right key function from b to k
     * @param r reducer from (a,{b}) to {c}
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {c}
     */
    public static Bag mapJoin ( final Function kx, final Function ky, final Function r,
                                final Bag X, final Bag Y ) {
        X.materialize();
        Y.materialize();
        return cmap(new Function() {
                public Bag eval ( final MRData e ) {
                    final Tuple p = (Tuple)e;
                    return cmap(new Function() {
                                    public Bag eval ( final MRData x ) {
                                        return (kx.eval(x).equals(p.first()))
                                            ? (Bag)r.eval(new Tuple(x,p.second()))
                                            : new Bag();
                                    } }, X); }
            },
            groupBy(cmap(new Function() {
                    public Bag eval ( final MRData y ) {
                        return new Bag(new Tuple(ky.eval(y),y));
                    } }, Y)));
    }

    /** An equi-join combined with a group-by (see GroupByJoinPlan)
     * @param kx left key function from a to k
     * @param ky right key function from b to k
     * @param gx group-by key function from a to k1
     * @param gy group-by key function from b to k2
     * @param acc accumulator from (c,(a,b)) to c
     * @param zero of type c
     * @param r reducer from ((k1,k2),c) to d
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {d}
     */
    public static Bag groupByJoin ( final Function kx, final Function ky,
                                    final Function gx, final Function gy,
                                    final Function acc, MRData zero,
                                    final Function r,
                                    final Bag X, final Bag Y ) {
        Bag s = groupBy(hash_join(kx,ky,
                                  new Function() {
                                      public MRData eval ( final MRData e ) {
                                          Tuple t = (Tuple)e;
                                          return new Tuple(new Tuple(gx.eval(t.first()),gy.eval(t.second())),t);
                                      } },
                                  X,Y));
        Bag res = new Bag();
        for ( MRData z: s ) {
            Tuple t = (Tuple)z;
            MRData v = zero;
            for ( MRData x: (Bag)t.second() )
                v = acc.eval(new Tuple(v,x));
            res.add(r.eval(new Tuple(t.first(),v)));
        };
        return res;
    }

    private static void flush_table ( final Map<MRData,MRData> hashTable, final Function r, final Bag result ) {
        Tuple pair = new Tuple(2);
        for ( Map.Entry<MRData,MRData> entry: hashTable.entrySet() ) {
            pair.set(0,entry.getKey());
            pair.set(1,entry.getValue());
            result.add(r.eval(pair));
        };
        hashTable.clear();
    }

    /** An equi-join combined with a group-by implemented using a sort-merge join
        combined with hash-based groupby/aggregation
     * @param kx left key function from a to k
     * @param ky right key function from b to k
     * @param gx group-by key function from a to k1
     * @param gy group-by key function from b to k2
     * @param acc accumulator from (c,(a,b)) to c
     * @param zero of type c
     * @param r reducer from ((k1,k2),c) to d
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {d}
     */
    final public static Bag mergeGroupByJoin ( final Function kx, final Function ky,
                                               final Function gx, final Function gy,
                                               final Function acc, MRData zero,
                                               final Function r,
                                               Bag X, Bag Y ) {
        Bag tbag = new Bag(2);
        Tuple pair = new Tuple(2);
        pair.set(1,new Tuple(2));
        final Map<MRData,MRData> hashTable = new HashMap<MRData,MRData>(1000);
        Bag xs = map(new Function() {
                public MRData eval ( final MRData e ) {
                    Tuple t = (Tuple)e;
                    return new Tuple(new Tuple(t.first(),kx.eval(t.second())),t.second());
                } }, X);
	xs.sort();
	X = null;
        Bag ys = map(new Function() {
                public MRData eval ( final MRData e ) {
                    Tuple t = (Tuple)e;
                    return new Tuple(new Tuple(t.first(),ky.eval(t.second())),t.second());
                } }, Y);
	ys.sort();
	Y = null;
	Bag res = new Bag();
	MRData partition = null;
	int i = 0;
	int j = 0;
	while ( i < xs.size() && j < ys.size() ) {
	    Tuple x = (Tuple)xs.get(i);
	    Tuple y = (Tuple)ys.get(j);
            int cmp = x.first().compareTo(y.first());
	    if ( cmp > 0 ) j++;
	    else if ( cmp < 0 ) i++;
	    else { // when cmp == 0
		Tuple key = (Tuple)x.first();
                if (partition == null)
                    partition = key.first();
                else if (!partition.equals(key.first())) {
                    partition = key.first();
                    flush_table(hashTable,r,res);
                };
		int k = i;
		int l = j;
		for ( ; k < xs.size() && key.compareTo(((Tuple)xs.get(k)).first()) == 0; k++ ) {
		    MRData xx = ((Tuple)xs.get(k)).second();
		    for ( l = j; l < ys.size() && key.compareTo(((Tuple)ys.get(l)).first()) == 0; l++ ) {
			MRData yy = ((Tuple)ys.get(l)).second();
                        Tuple gkey = new Tuple(gx.eval(xx),gy.eval(yy));
                        MRData old = hashTable.get(gkey);
                        if (old == null)
                            old = zero;
                        pair.set(0,old);
                        ((Tuple)pair.get(1)).set(0,xx).set(1,yy);
                        hashTable.put(gkey,acc.eval(pair));
		    }
		};
		i = k;
		j = l;
	    }
	};
        flush_table(hashTable,r,res);
        return res;
    }

    /** An equi-join combined with a group-by implemented using hashing
     * @param kx left key function from a to k
     * @param ky right key function from b to k
     * @param gx group-by key function from a to k1
     * @param gy group-by key function from b to k2
     * @param acc accumulator from (c,(a,b)) to c
     * @param zero of type c
     * @param r reducer from ((k1,k2),c) to d
     * @param X left input of type {a}
     * @param Y right input of type {b}
     * @return a value of type {d}
     */
    final public static Bag mergeGroupByJoin2 ( final Function kx, final Function ky,
                                               final Function gx, final Function gy,
                                               final Function acc, MRData zero,
                                               final Function r,
                                               Bag X, Bag Y ) {
        Bag tbag = new Bag(2);
        Tuple pair = new Tuple(2);
        pair.set(1,new Tuple(2));
        final Map<MRData,MRData> hashTable = new HashMap<MRData,MRData>(1000);
        Bag xs = groupBy(map(new Function() {
                public MRData eval ( final MRData e ) {
                    Tuple t = (Tuple)e;
                    return new Tuple(new Tuple(t.first(),kx.eval(t.second())),t.second());
                } }, X));
        Bag ys = groupBy(map(new Function() {
                public MRData eval ( final MRData e ) {
                    Tuple t = (Tuple)e;
                    return new Tuple(new Tuple(t.first(),ky.eval(t.second())),t.second());
                } }, Y));
        X = null; Y = null;
        Bag res = new Bag();
        final Iterator<MRData> xi = xs.iterator();
        final Iterator<MRData> yi = ys.iterator();
        if ( !xi.hasNext() || !yi.hasNext() )
            return res;
        Tuple x = (Tuple)xi.next();
        Tuple y = (Tuple)yi.next();
        MRData partition = null;
        while ( xi.hasNext() && yi.hasNext() ) {
            int cmp = x.first().compareTo(y.first());
            if (cmp < 0) { x = (Tuple)xi.next(); continue; };
            if (cmp > 0) { y = (Tuple)yi.next(); continue; };
            if (partition == null)
                partition = ((Tuple)x.first()).first();
            else if (!partition.equals(((Tuple)x.first()).first())) {
                partition = ((Tuple)x.first()).first();
                flush_table(hashTable,r,res);
            };
            for ( MRData xx: (Bag)x.second() )
                for ( MRData yy: (Bag)y.second() ) {
                    Tuple key = new Tuple(gx.eval(xx),gy.eval(yy));
                    MRData old = hashTable.get(key);
                    if (old == null)
                        old = zero;
                    pair.set(0,old);
                    ((Tuple)pair.get(1)).set(0,xx).set(1,yy);
                    hashTable.put(key,acc.eval(pair));
                };
            if (xi.hasNext())
                x = (Tuple)xi.next();
            if (yi.hasNext())
                y = (Tuple)yi.next();
        };
        flush_table(hashTable,r,res);
        return res;
    }

    /** repeat the loop until all termination conditions are true or until we reach the max num of steps
     * @param loop a function from {a} to {(a,boolean)}
     * @param init the initial value of type {a}
     * @param max_num the maximum number of steps
     * @return a value of type {a}
     */
    public static Bag repeat ( final Function loop,
                               final Bag init,
                               final int max_num ) throws Exception {
        boolean cont;
        int i = 0;
        Bag s = init;
        s.materializeAll();
        do {
            MRData d = loop.eval(s);
            i++;
            cont = false;
            if (d instanceof Bag) {
                Bag bag = (Bag) d;
                bag.materialize();
                s.clear();
                for ( MRData x: bag ) {
                    Tuple t = (Tuple)x;
                    cont |= ((MR_bool)t.second()).get();
                    s.add(t.first());
                }
            } else if (d instanceof MR_dataset) {
                DataSet ds = ((MR_dataset)d).dataset();
                s.clear();
                int c = 0;
                for ( MRData x: Plan.collect(ds) ) {
                    Tuple t = (Tuple)x;
                    if (((MR_bool)t.second()).get()) {
                        c++;
                        cont = true;
                    };
                    s.add(t.first());
                };
                if (!Config.testing)
                    System.err.println("*** Repeat #"+i+": "+c+" true results");
            } else throw new Error("Wrong repeat");
        } while (cont && i < max_num);
        return s;
    }

    /** transitive closure: repeat the loop until the new set is equal to the previous set
     *    or until we reach the max num of steps
     * @param loop a function from {a} to {a}
     * @param init the initial value of type {a}
     * @param max_num the maximum number of steps
     * @return a value of type {a}
     */
    public static Bag closure ( final Function loop,
                                final Bag init,
                                final int max_num ) throws Exception {
        int i = 0;
        long n = 0;
        long old = 0;
        Bag s = init;
        s.materializeAll();
        do {
            MRData d = loop.eval(s);
            i++;
            if (d instanceof Bag) {
                s = (Bag)d;
                s.materialize();
                old = n;
                n = s.size();
            } else if (d instanceof MR_dataset) {
                DataSet ds = ((MR_dataset)d).dataset();
                if (!Config.testing)
                    System.err.println("*** Repeat #"+i+": "+(ds.records-n)+" new records");
                old = n;
                n = ds.records;
                s = Plan.collect(ds);
            } else throw new Error("Wrong repeat");
        } while (old < n && i < max_num);
        return s;
    }

    /** repetition: repeat the loop until we reach the num of steps
     * @param loop a function from ({a1},...,{ak}) to ({a1},...,{ak})
     * @param init the initial value of type ({a1},...,{ak})
     * @param num the number of steps
     * @return a value of type ({a1},...,{ak})
     */
    public static Tuple loop ( final Function loop,
                               final Tuple init,
                               final int num ) {
        try {
            Tuple s = init;
            s.materializeAll();
            for ( int i = 0; i < num; i++ ) {
                Tuple d = (Tuple)loop.eval(s);
                for ( int j = 0; j < d.size(); j++ )
                    if (d.get(j) instanceof Bag) {
                        s.set(j,d.get(i));
                        ((Bag)d.get(j)).materialize();
                    } else if (d.get(j) instanceof MR_dataset) {
                        DataSet ds = ((MR_dataset)d.get(j)).dataset();
                        s.set(j,Plan.collect(ds));
                    } else throw new Error("Wrong loop");
            };
            return s;
        } catch (Exception ex) {
            throw new Error(ex);
        }
    }

    /** parse a text document using a given parser
     * @param parser the parser
     * @param path the text document (local file or directory of files)
     * @param args the arguments to pass to the parser
     * @return a lazy bag that contains the parsed data
     */
    public static Bag parsedSource ( final Parser parser,
                                     final String path,
                                     Trees args ) {
            parser.initialize(args);
            File file = new File(path);
            final File[] files = (file.isDirectory())
                                 ? file.listFiles()
                                 : new File[]{file};
            final int dl = files.length;
            parser.open(files[0].toString());
            return new Bag(new BagIterator () {
                    Iterator<MRData> iter;
                    int i = 0;
                    String line;
                    public boolean hasNext () {
                        while (iter == null || !iter.hasNext()) {
                            line = parser.slice();
                            while (line == null)
                                if (++i < dl)
                                    parser.open(files[i].toString());
                                else return false;
                            iter = parser.parse(line).iterator();
                        };
                        return true;
                    }
                    public MRData next () {
                        return iter.next();
                    }
                });
    }

    /** parse a text document using a given parser
     * @param parser the name of the parser
     * @param file the text document (local file)
     * @param args the arguments to pass to the parser
     * @return a lazy bag that contains the parsed data
     */
    public static Bag parsedSource ( String parser, String file, Trees args ) {
        try {
            return parsedSource(DataSource.parserDirectory.get(parser).newInstance(),file,args);
        } catch (Exception e) {
            throw new Error(e);
        }
    }

    private static Bag add_source_num ( int source_num, Bag input ) {
        return new Bag(new Tuple(new MR_int(source_num),input));
    }

    /** parse a text document using a given parser and tag output data with a source num
     * @param source_num the source id
     * @param parser the parser
     * @param file the text document (local file)
     * @param args the arguments to pass to the parser
     * @return a lazy bag that contains the parsed data taged with the source id
     */
    public static Bag parsedSource ( int source_num,
                                     Parser parser,
                                     String file,
                                     Trees args ) {
        return add_source_num(source_num,parsedSource(parser,file,args));
    }

    /** parse a text document using a given parser and tag output data with a source num
     * @param source_num the source id
     * @param parser the name of the parser
     * @param file the text document (local file)
     * @param args the arguments to pass to the parser
     * @return a lazy bag that contains the parsed data taged with the source id
     */
    public static Bag parsedSource ( int source_num, String parser, String file, Trees args ) {
        try {
            return parsedSource(source_num,DataSource.parserDirectory.get(parser).newInstance(),file,args);
        } catch (Exception e) {
            throw new Error(e);
        }
    }

    /** aggregate the Bag elements
     * @param accumulator a function from (b,a) to b
     * @param zero a value of type b
     * @param s a Bag of type {a}
     * @return a value of type b
     */
    public static MRData aggregate ( final Function accumulator,
                                     final MRData zero,
                                     final Bag s ) {
        MRData result = zero;
        for ( MRData x: s )
            result = accumulator.eval(new Tuple(result,x));
        return result;
    }

    public static MRData materialize ( MRData x ) {
        if (x instanceof Bag)
            ((Bag)x).materialize();
        return x;
    }

    /** Dump the value of some type to a binary local file;
     *  The type is dumped to a separate file.type
     */
    public static void dump ( String file, Tree type, MRData value ) throws IOException {
        File f = new File(file);
        File parent = f.getParentFile();
        if (parent != null && !parent.exists())
            parent.mkdirs();
        PrintStream ftp = new PrintStream(file+".type");
        ftp.print("1@"+type.toString()+"\n");
        ftp.close();
        DataOutputStream out = new DataOutputStream(new FileOutputStream(new File(file)));
        value.write(out);
        out.close();
    }

    /** return the type of the dumped binary local file from file.type */
    public static Tree get_type ( String file ) {
        try {
            BufferedReader ftp = new BufferedReader(new FileReader(new File(file+".type")));
            String s[] = ftp.readLine().split("@");
            ftp.close();
            if (s.length != 2)
                return null;
            if (!s[0].equals("1"))
                throw new Error("The binary file has been created in hadoop mode and cannot be read in java mode");
            return Tree.parse(s[1]);
        } catch (Exception e) {
            return null;
        }
    }

    /** read the contents of a dumped local binary file */
    public static MRData read_binary ( String file ) {
        try {
            Tree type = get_type(file);
            DataInputStream in = new DataInputStream(new FileInputStream(new File(file)));
            return MRContainer.read(in);
        } catch (Exception e) {
            return null;
        } 
    }

    /** read the contents of a dumped local binary file and tag data with a source num */
    public static Bag read_binary ( int source_num, String file ) {
        return add_source_num(source_num,(Bag)read_binary(file));
    }

    /** generate a lazy bag of long numbers {min...max} */
    public static Bag generator ( final long min, final long max ) {
        if (min > max)
            throw new Error("Min value ("+min+") is larger than max ("+max+") in generator");
        return new Bag(new BagIterator() {
                long index = min;
                public boolean hasNext () {
                    return index <= max;
                }
                public MRData next () {
                    return new MR_long(index++);
                }
            });
    }

    /** generate a lazy bag of long numbers {min...max} and tag each lon number with a source num */
    public static Bag generator ( int source_num, final long min, final long max ) {
        return add_source_num(source_num,generator(min,max));
    }

    /** the cache that holds all local data in memory */
    private static Tuple cache;

    /** return the cache element at location loc */
    public static MRData getCache ( int loc ) {
        return cache.get(loc);
    }

    /** set the cache element at location loc to value and return ret */
    public static MRData setCache ( int loc, MRData value, MRData ret ) {
        if (value instanceof Bag)
            materialize((Bag)value);
        cache.set(loc,value);
        return ret;
    }

    /** The BSP operation
     * @param source the source ids of the input Bags
     * @param superstep the BSP superstep is a function from ({M},S) to ({M},S,boolean)
     * @param init_state is the initial state of type S
     * @param order do we need to order the result?
     * @param inputs the input Bags
     * @return return a Bag in cache[0]
     */
    public static MRData BSP ( final int[] source,
                               final Function superstep,
                               final MRData init_state,
                               boolean order,
                               final Bag[] inputs ) {
        Bag msgs = new Bag();
        MRData state = init_state;
        Tuple result;
        boolean exit;
        boolean skip = false;
        String tabs = "";
        int step = 0;
        cache = new Tuple(100);
        for ( int i = 0; i < 100; i++ )
            cache.set(i,new Bag());
        for ( Bag x: inputs ) {
            Tuple p = (Tuple)(x.get(0));
            cache.set(((MR_int)p.first()).get(),
                      materialize(p.second()));
        };
        do {
            if (!skip)
                step++;
            if (!skip && Config.trace_execution) {
                tabs = Interpreter.tabs(Interpreter.tab_count);
                System.out.println(tabs+"  Superstep "+step+":");
                System.out.println(tabs+"      messages: "+msgs);
                System.out.println(tabs+"      state: "+state);
                for ( int i = 0; i < cache.size(); i++)
                    if (cache.get(i) instanceof Bag && ((Bag)cache.get(i)).size() > 0)
                        System.out.println(tabs+"      cache "+i+": "+cache.get(i));
            };
            result = (Tuple)superstep.eval(new Tuple(cache,msgs,state,new MR_string("")));
            Bag new_msgs = (Bag)result.get(0);
            state = result.get(1);
            exit = ((MR_bool)result.get(2)).get();
            skip = new_msgs == SystemFunctions.bsp_empty_bag;
            if ((!skip || exit) && Config.trace_execution)
                System.out.println(tabs+"      result: "+result);
            final Iterator<MRData> iter = new_msgs.iterator();
            msgs = new Bag(new BagIterator() {
                    public boolean hasNext () {
                        return iter.hasNext();
                    }
                    public MRData next () {
                        return ((Tuple)iter.next()).get(1);
                    }
                });
        } while (!exit);
        MRData[] data = new MRData[source.length];
        for ( int i = 0; i < data.length; i++ )
            data[i] = getCache(source[i]);
        if (order && data[0] instanceof Bag) {
            final Iterator<MRData> iter = ((Bag)data[0]).iterator();
            return new Bag(new BagIterator() {
                    public boolean hasNext () {
                        return iter.hasNext();
                    }
                    public MRData next () {
                        return ((Tuple)iter.next()).get(0);
                    }
                });
        };
        if (data.length == 1)
            return data[0];
        else return new Tuple(data);
    }
}