diff --git a/conf/mrql-env.sh b/conf/mrql-env.sh
index a9b16cb..77da529 100644
--- a/conf/mrql-env.sh
+++ b/conf/mrql-env.sh
@@ -72,12 +72,12 @@
 BSP_SPLIT_INPUT=
 
 
-# Optional: Spark configuration. Supports versions 1.0.0, 1.0.2, 1.1.0, 1.1.1, 1.2.0, 1.3.0, 1.3.1, and 1.6.0
+# Optional: Spark configuration. Supports versions 1.0.0, 1.0.2, 1.1.0, 1.1.1, 1.2.0, 1.3.0, 1.3.1, 1.6.0, and 1.6.2
 # (Spark versions 0.8.1, 0.9.0, and 0.9.1 are supported by MRQL 0.9.0)
 # You may use the Spark prebuilts bin-hadoop1 or bin-hadoop2 (Yarn)
 # For distributed mode, give write permission to /tmp: hadoop fs -chmod -R 777 /tmp
 # Tested in local, standalone deploy, and Yarn modes
-SPARK_HOME=${HOME}/spark-1.6.0-bin-hadoop2.6
+SPARK_HOME=${HOME}/spark-1.6.2-bin-hadoop2.6
 # URI of the Spark master node:
 #   to run Spark on Standalone Mode, set it to spark://`hostname`:7077
 #   to run Spark on a YARN cluster, set it to "yarn-client"
diff --git a/core/src/main/java/org/apache/mrql/Evaluator.java b/core/src/main/java/org/apache/mrql/Evaluator.java
index ac89f4e..a504087 100644
--- a/core/src/main/java/org/apache/mrql/Evaluator.java
+++ b/core/src/main/java/org/apache/mrql/Evaluator.java
@@ -159,7 +159,7 @@
     }
 
     /** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
-    public void streaming ( Tree plan, Environment env, Function f ) {
+    public void streaming ( Tree plan, Environment env, Environment dataset_env, Function f ) {
         throw new Error("MRQL Streaming is not supported in this evaluation mode yet");
     }
 
diff --git a/core/src/main/java/org/apache/mrql/Interpreter.gen b/core/src/main/java/org/apache/mrql/Interpreter.gen
index 054963c..600f5aa 100644
--- a/core/src/main/java/org/apache/mrql/Interpreter.gen
+++ b/core/src/main/java/org/apache/mrql/Interpreter.gen
@@ -606,7 +606,7 @@
             return ((Lambda)fnc).lambda().eval(t);
         case Stream(...):  // streaming
             final Tree qt = query_type;
-            Evaluator.evaluator.streaming(e,env,new Function(){
+            Evaluator.evaluator.streaming(e,env,null,new Function(){
                     public MRData eval ( final MRData data ) {
                         System.out.println(print(data,qt));
                         return data;
diff --git a/core/src/main/java/org/apache/mrql/Main.java b/core/src/main/java/org/apache/mrql/Main.java
index 087f439..cafc62f 100644
--- a/core/src/main/java/org/apache/mrql/Main.java
+++ b/core/src/main/java/org/apache/mrql/Main.java
@@ -28,7 +28,7 @@
 
 
 final public class Main {
-    public final static String version = "0.9.6";
+    public final static String version = "0.9.8";
 
     public static PrintStream print_stream;
     public static Configuration conf;
diff --git a/core/src/main/java/org/apache/mrql/PlanGeneration.gen b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
index 1b724da..a42380c 100644
--- a/core/src/main/java/org/apache/mrql/PlanGeneration.gen
+++ b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
@@ -418,6 +418,18 @@
                                      `(makePlan(x)),
                                      `(makePlan(y)))>;
            else fail
+       case outerMerge(`m,`x,`y):
+               if (is_dataset_expr(x) && is_dataset_expr(y))
+                   return #<OuterMerge(`(makePlan(m)),
+                                       `(makePlan(x)),
+                                       `(makePlan(y)))>;
+               else fail
+       case rightOuterMerge(`m,`x,`y):
+               if (is_dataset_expr(x) && is_dataset_expr(y))
+                   return #<RightOuterMerge(`(makePlan(m)),
+                                            `(makePlan(x)),
+                                            `(makePlan(y)))>;
+               else fail
        case cmap(`m,`s):
            if (is_dataset_expr(s))
                return #<cMap(`(makePlan(m)),
diff --git a/core/src/main/java/org/apache/mrql/Printer.gen b/core/src/main/java/org/apache/mrql/Printer.gen
index 18975c9..03fc28e 100644
--- a/core/src/main/java/org/apache/mrql/Printer.gen
+++ b/core/src/main/java/org/apache/mrql/Printer.gen
@@ -342,6 +342,12 @@
         case CrossProduct(`mx,`my,`r,`x,`y):
             return "CrossProduct:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
                    +tab(n+3)+"right: "+print_plan(y,n+10,true);
+        case OuterMerge(`merge,`x,`y):
+            return "OuterMerge:\n"+tab(n+3)+"state: "+print_plan(x,n+10,true)+"\n"
+                   +tab(n+3)+"delta: "+print_plan(y,n+10,true);
+        case RightOuterMerge(`merge,`x,`y):
+            return "RightOuterMerge:\n"+tab(n+3)+"state: "+print_plan(x,n+10,true)+"\n"
+                   +tab(n+3)+"delta: "+print_plan(y,n+10,true);
         case CrossAggregateProduct(`mx,`my,`r,`a,null,`x,`y):
             return "CrossProduct:\n"+tab(n+3)+"left: "+print_plan(x,n+9,true)+"\n"
                    +tab(n+3)+"right: "+print_plan(y,n+10,true);
diff --git a/core/src/main/java/org/apache/mrql/Streaming.gen b/core/src/main/java/org/apache/mrql/Streaming.gen
index 99ce205..f54cad8 100644
--- a/core/src/main/java/org/apache/mrql/Streaming.gen
+++ b/core/src/main/java/org/apache/mrql/Streaming.gen
@@ -932,17 +932,21 @@
     }
 
     /** Return the merge function (over X and Y) of the monoid m; type is used for key equality */
-    private static Tree merge ( Tree m, Tree type, Tree X, Tree Y ) {
+    private static Tree merge ( Tree m, Tree type, Tree X, Tree Y, boolean first ) {
         match m {
         case `gb(`n):
             if (! #[groupBy,orderBy].member(#<`gb>) )
                 fail;
             match type {
             case `T(tuple(`keytp,`tp)):
-                if (unique_key(keytp)) {
+                if (first) {
+                    Tree nv = new_var();
+                    Tree mb = merge(n,tp,#<nth(`nv,0)>,#<nth(`nv,1)>,false);
+                    return #<outerMerge(lambda(`nv,`mb),`X,`Y)>;
+                } else if (unique_key(keytp)) {
                     Tree vx = new_var();
                     Tree vy = new_var();
-                    Tree v = merge(n,tp,#<nth(call(elem,`vx),1)>,#<nth(call(elem,`vy),1)>);
+                    Tree v = merge(n,tp,#<nth(call(elem,`vx),1)>,#<nth(call(elem,`vy),1)>,false);
                     if (is_persistent_collection(T)) {
                         X = #<Collect(`X)>;
                         Y = #<Collect(`Y)>;
@@ -961,7 +965,7 @@
                 Tree vy = new_var();
                 Tree mx = new_var();
                 Tree my = new_var();
-                Tree mb = merge(n,tp,#<nth(`vx,1)>,#<nth(`vy,1)>);
+                Tree mb = merge(n,tp,#<nth(`vx,1)>,#<nth(`vy,1)>,false);
                 Tree b = #<cmap(lambda(`vx,cmap(lambda(`vy,bag(tuple(nth(`vx,0),`mb))),
                                                 `my)),
                                 `mx)>;
@@ -980,7 +984,7 @@
             Trees bs = #[ ];
             int i = 0;
             for ( Tree a: ms ) {
-                bs = bs.append(merge(a,tps.nth(i),#<nth(`X,`i)>,#<nth(`Y,`i)>));
+                bs = bs.append(merge(a,tps.nth(i),#<nth(`X,`i)>,#<nth(`Y,`i)>,first));
                 i++;
             };
             return #<tuple(...bs)>;
@@ -992,7 +996,82 @@
                 match b {
                 case bind(`n,`a):
                     cs = cs.append(#<bind(`n,`(merge(a,((Node)tps.nth(i++)).children().nth(1),
-                                                     #<project(`X,`n)>,#<project(`Y,`n)>)))>);
+                                                     #<project(`X,`n)>,#<project(`Y,`n)>,first)))>);
+                }
+            return #<record(...cs)>;
+        case union:
+            return #<call(plus,`X,`Y)>;
+        case fixed:
+            return Y;
+        case _:
+            if (!m.is_variable())
+                fail;
+            for ( Tree monoid: monoids )
+                match monoid {
+                case `aggr(`mtp,`plus,`zero,`unit):
+                    if (#<`aggr>.equals(m))
+                        return #<apply(`plus,tuple(`X,`Y))>;
+                };
+        };
+        throw new Error("Undefined monoid: "+m);
+    }
+
+    /** Return the merge function (over X and Y) of the monoid m; type is used for key equality */
+    private static Tree merge_left ( Tree m, Tree type, Tree X, Tree Y, boolean first ) {
+        match m {
+        case `gb(`n):
+            if (! #[groupBy,orderBy].member(#<`gb>) )
+                fail;
+            match type {
+            case `T(tuple(`keytp,`tp)):
+                if (first) {
+                    Tree nv = new_var();
+                    Tree mb = merge(n,tp,#<nth(`nv,0)>,#<nth(`nv,1)>,false);
+                    return #<rightOuterMerge(lambda(`nv,`mb),`X,`Y)>;
+                } else if (unique_key(keytp)) {
+                    Tree vx = new_var();
+                    Tree v = merge_left(n,tp,#<nth(call(elem,`vx),1)>,#<nth(call(elem,`Y),1)>,false);
+                    if (is_persistent_collection(T))
+                        X = #<Collect(`X)>;
+                    return #<let(`vx,`X,
+                                 if(call(exists,`vx),
+                                    bag(tuple(`keytp,`v)),
+                                    call(elem,`Y)))>;
+                };
+                // needs an outer-join in the reducer function
+                Tree v = new_var();
+                Tree vx = new_var();
+                Tree vy = new_var();
+                Tree mx = new_var();
+                Tree mb = merge_left(n,tp,#<nth(`vx,1)>,#<nth(`vy,1)>,false);
+                Tree b = #<cmap(lambda(`vx,cmap(lambda(`vy,bag(tuple(nth(`vx,0),`mb))),
+                                                nth(`v,1))),
+                                `mx)>;
+                return #<join(lambda(`vx,`(key_equality(keytp,#<nth(`vx,0)>))),
+                              lambda(`vy,`(key_equality(keytp,#<nth(`vy,0)>))),
+                              lambda(`v,let(`mx,nth(`v,0),
+                                            if(call(exists,`mx),`b,nth(`v,1)))),
+                              `X, `Y)>;
+            };
+            throw new Error("Unknown type for merge_left: "+type);
+        case product(...ms):
+            Trees tps = ((Node)type).children();
+            Trees bs = #[ ];
+            int i = 0;
+            for ( Tree a: ms ) {
+                bs = bs.append(merge_left(a,tps.nth(i),#<nth(`X,`i)>,#<nth(`Y,`i)>,first));
+                i++;
+            };
+            return #<tuple(...bs)>;
+        case record(...bs):
+            Trees tps = ((Node)type).children();
+            Trees cs = #[ ];
+            int i = 0;
+            for ( Tree b: bs )
+                match b {
+                case bind(`n,`a):
+                    cs = cs.append(#<bind(`n,`(merge_left(a,((Node)tps.nth(i++)).children().nth(1),
+                                                     #<project(`X,`n)>,#<project(`Y,`n)>,first)))>);
                 }
             return #<record(...cs)>;
         case union:
@@ -1431,7 +1510,7 @@
             if (Config.trace) {
                 System.out.println("Subterm type: "+tp);
                 System.out.println("Merge function over X and Y: ");
-                System.out.println(SimplifyTerm(merge(m,tp,#<X>,#<Y>)).pretty(0));
+                System.out.println(SimplifyTerm(merge(m,tp,#<X>,#<Y>,true)).pretty(0));
             };
             type_env.insert(nv.toString(),tp);
             Tree answer = answer(ne,nv,m);
@@ -1444,8 +1523,8 @@
             };
             Tree zero = zero(m);
             Tree my = new_var();
-            Tree merge = convert_to_algebra(SimplifyTerm(merge(m,tp,nv,my)));
-            Tree res = #<tuple(`zero,lambda(tuple(`nv,`my),`merge),`q,lambda(`nv,`answer))>;
+            Tree merge = convert_to_algebra(SimplifyTerm(merge(m,tp,nv,my,true)));
+            Tree res = #<tuple(`zero,lambda(tuple(`nv,`my),`merge),`q,lambda(`nv,`answer),`m)>;
             if (Config.trace) {
                 System.out.println("Incremental subterm:");
                 System.out.println(res.pretty(0));
@@ -1457,7 +1536,7 @@
 
     private static Tree generate_incremental_code ( Tree e, Environment env ) {
         if (!is_streaming(e))
-            return #<tuple(tuple(),lambda(tuple(tuple(),tuple()),tuple()),tuple(),lambda(tuple(),`e))>;
+            return #<tuple(tuple(),lambda(tuple(tuple(),tuple()),tuple()),tuple(),lambda(tuple(),`e),fixed)>;
         match e {
         case call(stream,...):
             Tree v = new_var();
@@ -1472,50 +1551,53 @@
                 };
             if (is_monoid)
                 fail;
-            Trees zs = #[], vs = #[], ws = #[], ms = #[], hs = #[], as = #[];
+            Trees zs = #[], vs = #[], ws = #[], ms = #[], hs = #[], as = #[], ns = #[];
             int i = 0;
             for ( Tree q: qs )
                 match generate_incremental_code(q,env) {
-                case tuple(`z,lambda(tuple(`v,`w),`m),`h,lambda(_,`a)):
+                case tuple(`z,lambda(tuple(`v,`w),`m),`h,lambda(_,`a),`n):
                     zs = zs.append(z); hs = hs.append(h); as = as.append(a);
                     vs = vs.append(v); ws = ws.append(w); ms = ms.append(m);
-                    i++;
+                    ns = ns.append(n); i++;
                 };
             return #<tuple(tuple(...zs),
                            lambda(tuple(tuple(...vs),tuple(...ws)),tuple(...ms)),
                            tuple(...hs),
-                           lambda(tuple(...vs),call(`f,...as)))>;
+                           lambda(tuple(...vs),call(`f,...as)),
+                           tuple(...ns))>;
         case record(...rs):
-            Trees zs = #[], vs = #[], ws = #[], ms = #[], hs = #[], as = #[];
+            Trees zs = #[], vs = #[], ws = #[], ms = #[], hs = #[], as = #[], ns = #[];
             int i = 0;
             for ( Tree r: rs )
                 match r {
                 case bind(`k,`q):
                     match generate_incremental_code(q,env) {
-                    case tuple(`z,lambda(tuple(`v,`w),`m),`h,lambda(_,`a)):
+                    case tuple(`z,lambda(tuple(`v,`w),`m),`h,lambda(_,`a),`n):
                         zs = zs.append(z); hs = hs.append(h); as = as.append(a);
                         vs = vs.append(v); ws = ws.append(w); ms = ms.append(m);
-                        i++;
+                        ns = ns.append(n); i++;
                     }
                 };
             return #<tuple(tuple(...zs),
                            lambda(tuple(tuple(...vs),tuple(...ws)),tuple(...ms)),
                            tuple(...hs),
-                           lambda(tuple(...vs),record(...as)))>;
+                           lambda(tuple(...vs),record(...as)),
+                           tuple(...ns))>;
         case tuple(...qs):
-            Trees zs = #[], vs = #[], ws = #[], ms = #[], hs = #[], as = #[];
+            Trees zs = #[], vs = #[], ws = #[], ms = #[], hs = #[], as = #[], ns = #[];
             int i = 0;
             for ( Tree q: qs )
                 match generate_incremental_code(q,env) {
-                case tuple(`z,lambda(tuple(`v,`w),`m),`h,lambda(_,`a)):
+                case tuple(`z,lambda(tuple(`v,`w),`m),`h,lambda(_,`a),`n):
                     zs = zs.append(z); hs = hs.append(h); as = as.append(a);
                     vs = vs.append(v); ws = ws.append(w); ms = ms.append(m);
-                    i++;
+                    ns = ns.append(n); i++;
                 };
             return #<tuple(tuple(...zs),
                            lambda(tuple(tuple(...vs),tuple(...ws)),tuple(...ms)),
                            tuple(...hs),
-                           lambda(tuple(...vs),tuple(...as)))>;
+                           lambda(tuple(...vs),tuple(...as)),
+                           tuple(...ns))>;
         };
         return generate_code(e,env);
     }
@@ -1554,7 +1636,50 @@
             TypeInference.type_inference(u);
             repeat_environment = new Environment(v.toString(),#<union>,repeat_environment);
             match generate_incremental_code(u,null) {
-            case tuple(`z,lambda(tuple(`v1,`v2),`m),`h,lambda(`s,`a)):
+            case tuple(`z,lambda(tuple(`v1,`v2),`m),`h,lambda(`s,`a),`monoid):
+                Tree deltaT = new_var();
+                Tree w = new_var();
+                Tree tp = TypeInference.type_inference(s);
+                TypeInference.type_env.insert(deltaT.toString(),tp);
+                match tp {
+                case `T(`etp):
+                    TypeInference.type_env.insert(T.toString(),etp);
+                };
+                Tree m_hat = convert_to_algebra(SimplifyTerm(merge_left(monoid,tp,v1,v2,true)));
+                if (Config.trace) {
+                    System.out.println("Left-merge function over X and Y: ");
+                    System.out.println(SimplifyTerm(merge_left(monoid,tp,#<X>,#<Y>,true)).pretty(0));
+                };
+                Tree nm = subst(v1,s,subst(v2,subst(v,subst(s,deltaT,a),h),m_hat));
+                Tree m0 = subst(v,x,h);//subst(v1,s,subst(v2,subst(v,x,h),m_hat));
+                Tree q = #<repeat(lambda(`deltaT,cmap(lambda(`w,bag(tuple(`w,true))),`nm)),
+                                  `m0,
+                                  `(nn-1))>;
+                Tree res = #<incr(`z,
+                                  lambda(`s,apply(lambda(tuple(`v1,`v2),`m),tuple(`s,`q))),
+                                  lambda(`s,`a))>;
+                res = SimplifyTerm(res);
+                if (Config.trace) {
+                    System.out.println("Incremental program:");
+                    System.out.println(res.pretty(0));
+                };
+                TypeInference.type_inference(res);
+                return res;
+            }
+        case repeat(lambda(`v,`u),`x,`n):
+            if (false && !is_streaming(x))
+                fail;
+            if (!(n instanceof LongLeaf))
+                throw new Error("The repeat must have a constant repetition: "+n);
+            int nn = (int)((LongLeaf)n).value();
+            if (nn < 1)
+                throw new Error("Wrong repeat number: "+n);
+            Tree nv = new_var();
+            u = #<cmap(lambda(`nv,bag(nth(`nv,0))),`u)>;
+            TypeInference.type_inference(u);
+            repeat_environment = new Environment(v.toString(),#<union>,repeat_environment);
+            match generate_incremental_code(u,null) {
+            case tuple(`z,lambda(tuple(`v1,`v2),`m),`h,lambda(`s,`a),_):
                 Tree deltaT = new_var();
                 Tree w = new_var();
                 TypeInference.type_env.insert(deltaT.toString(),TypeInference.type_inference(s));
@@ -1576,8 +1701,8 @@
                 return res;
             }
         };
-        match generate_incremental_code(e,null) {
-        case tuple(`z,lambda(tuple(`v1,`v2),`m),`h,lambda(_,`a)):
+    match generate_incremental_code(e,null) {
+    case tuple(`z,lambda(tuple(`v1,`v2),`m),`h,lambda(_,`a),_):
             Tree tp = TypeInference.type_inference(h);
             bind_pattern2type(v1,tp);
             Tree nm = SimplifyTerm(#<apply(lambda(`v2,`m),`h)>);
diff --git a/core/src/main/java/org/apache/mrql/TopLevel.gen b/core/src/main/java/org/apache/mrql/TopLevel.gen
index 13f4d68..9c5ecfa 100644
--- a/core/src/main/java/org/apache/mrql/TopLevel.gen
+++ b/core/src/main/java/org/apache/mrql/TopLevel.gen
@@ -321,7 +321,7 @@
             Evaluator.evaluator.initialize_query();
             final Tree qt = query_type;
             final String file = s.stringValue();
-            Evaluator.evaluator.streaming(plan,null,new Function(){
+            Evaluator.evaluator.streaming(plan,null,null,new Function(){
                     long i = 0;
                     public MRData eval ( final MRData data ) {
                         try {
@@ -342,7 +342,7 @@
             Evaluator.evaluator.initialize_query();
             final Tree qt = query_type;
             final String file = s.stringValue();
-            Evaluator.evaluator.streaming(plan,null,new Function(){
+            Evaluator.evaluator.streaming(plan,null,null,new Function(){
                     long i = 0;
                     public MRData eval ( final MRData data ) {
                         try {
@@ -365,27 +365,10 @@
             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 boolean incr_test = false;
-                if (incr_test) {
-                    // for code testing only
-                    long t = System.currentTimeMillis();
-                    Environment nenv = bind_list(state,Streaming.unstreamify(merge),env);
-                    for ( Environment ev = nenv; ev != null; ev = ev.next )
-                        if (ev.value instanceof MR_dataset) {
-                            DataSet ds = ((MR_dataset)ev.value).dataset();
-                            List<MRData> vals = ds.take(Config.max_bag_size_print);
-                            System.out.println(ev.name+" = "+vals.size()+ " "+vals);
-                        };
-                    MRData a = Evaluator.evaluator.evalE(answer,nenv);
-                    System.out.println(print(a,qt));
-                    if (!Config.quiet_execution)
-                        System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
-                    return;
-                };
+                final Environment env =bind_list(state,z,null);
                 final Tree ans = answer;
                 merge = Streaming.streamify(merge);
-                Evaluator.evaluator.streaming(#<lambda(`state,`merge)>,env,
+                Evaluator.evaluator.streaming(#<lambda(`state,`merge)>,env,null,
                    new Function() {
                         public MRData eval ( final MRData state ) {
                             MRData a = Evaluator.evaluator.evalE(ans,env);
diff --git a/core/src/main/java/org/apache/mrql/Translator.gen b/core/src/main/java/org/apache/mrql/Translator.gen
index 3be0d9f..2d9d2fe 100644
--- a/core/src/main/java/org/apache/mrql/Translator.gen
+++ b/core/src/main/java/org/apache/mrql/Translator.gen
@@ -178,7 +178,8 @@
     static Trees plans_with_distributed_lambdas
         = #[MapReduce,MapAggregateReduce,MapCombineReduce,FroupByJoin,Aggregate,
             MapReduce2,MapCombineReduce2,MapAggregateReduce2,MapJoin,MapAggregateJoin,
-            CrossProduct,CrossAggregateProduct,cMap,AggregateMap,BSP,GroupByJoin];
+            CrossProduct,CrossAggregateProduct,cMap,AggregateMap,BSP,GroupByJoin,
+            OuterMerge,RightOuterMerge];
 
     static Trees algebraic_operators
         = #[mapReduce,mapReduce2,cmap,join,groupBy,orderBy,aggregate,map,filter,repeat,closure];
diff --git a/core/src/main/java/org/apache/mrql/TypeInference.gen b/core/src/main/java/org/apache/mrql/TypeInference.gen
index 7c80240..f6769f4 100644
--- a/core/src/main/java/org/apache/mrql/TypeInference.gen
+++ b/core/src/main/java/org/apache/mrql/TypeInference.gen
@@ -32,12 +32,11 @@
 
     public static Tree make_persistent_type ( Tree tp ) {
         match tp {
-        case `f(...al):
-            Trees bs = #[];
-            for ( Tree a: al )
-                bs = bs.append(make_persistent_type(a));
-            String g = persistent_collection(f);
-            return #<`g(...bs)>;
+        case `T(`t):
+            if (!is_collection(T))
+                fail;
+            String S = persistent_collection(T);
+            return #<`S(`t)>;
         };
         return tp;
     }
@@ -1200,6 +1199,32 @@
             };
         case cstep(`x):  // used in QueryPlan for closure
             return type_inference(x);
+        case outerMerge(lambda(`v,`b),`s,`d):
+            Tree ts = type_inference(s);
+            if (unify(type_inference(d),ts) == null)
+                type_error(e,"Wrong outerMerge");
+            match ts {
+            case `S(tuple(`k,`tp)):
+                if (!is_collection(S))
+                    fail;
+                type_env.insert(v.toString(),#<tuple(`tp,`tp)>);
+                type_inference(b);
+                return ts;
+            };
+            type_error(e,"Wrong OuterMerge");
+        case rightOuterMerge(lambda(`v,`b),`s,`d):
+            Tree ts = type_inference(s);
+            if (unify(type_inference(d),ts) == null)
+                type_error(e,"Wrong RightOuterMerge");
+            match ts {
+            case `S(tuple(`k,`tp)):
+                if (!is_collection(S))
+                    fail;
+                type_env.insert(v.toString(),#<tuple(`tp,`tp)>);
+                type_inference(b);
+                return ts;
+            };
+            type_error(e,"Wrong rightOuterMerge");
         case if(`p,`x,`y):
             if (unify(type_inference2(p),#<bool>) == null)
                 type_error(e,"Expected a boolean predicate in if-then-else: "+print_query(p));
diff --git a/pom.xml b/pom.xml
index bd9079b..cb1ac62 100644
--- a/pom.xml
+++ b/pom.xml
@@ -47,7 +47,7 @@
     <hadoop.version>1.2.1</hadoop.version>
     <yarn.version>2.7.1</yarn.version>
     <hama.version>0.7.0</hama.version>
-    <spark.version>1.6.0</spark.version>
+    <spark.version>1.6.2</spark.version>
     <scala.version>2.10</scala.version>
     <flink.version>1.0.2</flink.version>
     <skipTests>true</skipTests>
diff --git a/queries/factorization3.mrql b/queries/factorization3.mrql
index de61b5d..3aca55b 100644
--- a/queries/factorization3.mrql
+++ b/queries/factorization3.mrql
@@ -44,7 +44,7 @@
 
 // cell-wise addition:
 macro Cadd ( X, Y ) {
-  select ( x*y, i, j )
+  select ( x+y, i, j )
     from (x,i,j) in X, (y,i,j) in Y
 };
 
diff --git a/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen b/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
index 60a9bb3..616da62 100644
--- a/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
+++ b/spark/src/main/java/org/apache/mrql/SparkEvaluator.gen
@@ -19,10 +19,7 @@
 
 import org.apache.mrql.gen.*;
 import java_cup.runtime.Scanner;
-import java.util.List;
-import java.util.ArrayList;
-import java.util.Hashtable;
-import java.util.Iterator;
+import java.util.*;
 import java.io.*;
 import java.util.Enumeration;
 import org.apache.log4j.*;
@@ -59,11 +56,14 @@
     final static Bag empty_bag = new Bag();
     // an HDFS tmp file used to hold the data source directory information in distributed mode
     final static String data_source_dir_name = "tmp/"+System.getenv("USER")+"_data_source_dir.txt";
-    private static boolean first_time = true; // true at the beginning of a query execution
+    // holds the cached RDDs that need to unpersist
+    private static ArrayList<JavaRDD<MRData>> cached_rdds = new ArrayList<JavaRDD<MRData>>();
+    static Environment global_rdds = null;
 
     /** initialize the Spark evaluator */
     final public void init ( Configuration conf ) {
         Config.spark_mode = true;
+        global_rdds = null;
         SparkConf spark_conf = new SparkConf();
         spark_conf.setAppName("MRQL");
         if (Config.hadoop_mode && Config.local_mode) {
@@ -100,7 +100,6 @@
     }
 
     final public void initialize_query () {
-        first_time = true;
         Plan.distribute_compiled_arguments(Plan.conf);
         if (spark_context != null && Config.compile_functional_arguments)
             spark_context.addJar(Plan.conf.get("mrql.jar.path"));
@@ -125,9 +124,15 @@
         return SparkGeneratorInputFormat.class;
     }
 
+    final public static void GC_cached_rdds () {
+	for ( JavaRDD<MRData> rdd: cached_rdds )
+	    rdd.unpersist();
+	cached_rdds = new ArrayList<JavaRDD<MRData>>();
+    }
+
     /** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
-    final public void streaming ( Tree plan, Environment env, org.apache.mrql.Function f ) {
-        SparkStreaming.evaluate(plan,env,f);
+    final public void streaming ( Tree plan, Environment env, Environment dataset_env, org.apache.mrql.Function f ) {
+        SparkStreaming.evaluate(plan,env,dataset_env,f);
     }
 
     /** used by the master to send parsing details (eg, record types) to workers */
@@ -179,6 +184,7 @@
                         return new Tuple2<MRContainer,MRContainer>(zero,new MRContainer(value));
                     }
                 }).saveAsHadoopFile(file,MRContainer.class,MRContainer.class,SequenceFileOutputFormat.class);
+            rdd.unpersist();
         } else super.dump(file,type,data);
     }
 
@@ -206,23 +212,14 @@
                 rdd.saveAsTextFile(file);
             };
             Config.max_bag_size_print = ps;
+            rdd.unpersist();
         } else super.dump_text(file,type,data);
     }
 
-    private static void set_global_env ( Environment env ) {
-        if (true || first_time) {  // Bug: it doesn't need to be executed more than once per worker
-            // pass the global bindings to workers
-            set_global_bindings(env);
-            first_time = false;
-        }
-    }
-
     private static Function2<MRData,MRData,MRData> accumulator ( final Tree acc_fnc, final Environment env ) {
-        final Environment master_env = global_env;
         final org.apache.mrql.Function f = evalF(acc_fnc,env);
         return new Function2<MRData,MRData,MRData>() {
             public MRData call ( MRData x, MRData y ) {
-                set_global_env(master_env);
                 return f.eval(new Tuple(x,y));
             }
         };
@@ -243,27 +240,27 @@
         MRData z = evalE(zero,env);
         match plan {
         case AggregateMap(`m,`acc,_,`s):
-            return evalD(#<cMap(`m,`s)>,env)
+            return evalD(#<cMap(`m,`s)>,env,null)
                      .aggregate(z,accumulator(acc,env),f2);
         case MapAggregateReduce(`m,`r,`acc,_,`s,`o):
             if (acc.equals(#<null>))
                 fail;
-            return evalD(#<MapReduce(`m,`r,`s,`o)>,env)
+            return evalD(#<MapReduce(`m,`r,`s,`o)>,env,null)
                      .aggregate(z,accumulator(acc,env),f2);
         case CrossAggregateProduct(`mx,`my,`r,`acc,_,`x,`y):
             if (acc.equals(#<null>))
                 fail;
-            return evalD(#<CrossProduct(`mx,`my,`r,`x,`y)>,env)
+            return evalD(#<CrossProduct(`mx,`my,`r,`x,`y)>,env,null)
                      .aggregate(z,accumulator(acc,env),f2);
         case MapAggregateReduce2(`mx,`my,`r,`acc,_,`x,`y,`o):
             if (acc.equals(#<null>))
                 fail;
-            return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env)
+            return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env,null)
                      .aggregate(z,accumulator(acc,env),f2);
         case MapAggregateJoin(`mx,`my,`r,`acc,_,`x,`y):
             if (acc.equals(#<null>))
                 fail;
-            return evalD(#<MapJoin(`mx,`my,`r,`x,`y)>,env)
+            return evalD(#<MapJoin(`mx,`my,`r,`x,`y)>,env,null)
                      .aggregate(z,accumulator(acc,env),f2);
         };
         throw new Error("Unrecognized aggregation: "+plan);
@@ -276,13 +273,15 @@
             int limit = ((MR_int)evalE(num,env)).get();
             MR_rdd[] s = new MR_rdd[vs.length()];
             for ( int i = 0; i < vs.length(); i++ )
-                s[i] = new MR_rdd(eval(ss.nth(i),env).cache());
+                s[i] = new MR_rdd(eval(ss.nth(i),env,(Environment)null).cache());
             for ( int n = 0; n < limit; n++ ) {
-                Environment nenv = env;
+                Environment nenv = null;
                 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_rdd(eval(bs.nth(i),nenv).cache());
+                for ( int i = 0; i < vs.length(); i ++ ) {
+                    s[i].rdd().unpersist();
+                    s[i] = new MR_rdd(eval(bs.nth(i),env,nenv).cache());
+                }
             };
             Tuple t = new Tuple(vs.length());
             for ( int i = 0; i < vs.length(); i++ )
@@ -304,57 +303,18 @@
             });
     }
 
-    // Return the RDD elements at the given position
-    private static Iterator<MRData> partition ( final JavaRDD<MRData> rdd, final int position ) {
-        /* Doesn't work (needs the right TaskContext)
-        TaskContext context = new TaskContext(0,0,(long)0,Config.local_mode,null);
-        List<Partition> ps = rdd.splits();
-        return rdd.iterator(ps.get(position),context);
-        */
-        return rdd.mapPartitionsWithIndex(new Function2<Integer,Iterator<MRData>,Iterator<MRData>>() {
-                public Iterator<MRData> call ( Integer partition, Iterator<MRData> values ) {
-                    if (partition == position)
-                        return values;
-                    else return new ArrayList<MRData>().iterator();
-                }
-            },true).collect().iterator();
-    }
-
-    final static int MAX_CACHE_SIZE = 1000;
-
     /** Convert a Spark RDD into a lazy bag
      * @param rdd the Spark RDD
      * @return a lazy bag that contains all RDD elements
      */
     public static Bag bag ( final JavaRDD<MRData> rdd ) throws IOException {
-        final JavaRDD<MRData> rd = rdd.cache();
-        if (rd.count() <= MAX_CACHE_SIZE) {  // small RDD
-            final Iterator<MRData> i = rd.collect().iterator();
-            return new Bag(new BagIterator() {
-                    public MRData next () {
-                        return i.next();
-                    }
-                    public boolean hasNext () {
-                        return i.hasNext();
-                    }
-                });
-        };
-        // return the RDD elements lazily, one partition at a time
-        final int splits = rd.splits().size();
+        final Iterator<MRData> i = rdd.toLocalIterator();
         return new Bag(new BagIterator() {
-                Iterator<MRData> i = null;
-                int c = 0;
                 public MRData next () {
                     return i.next();
                 }
                 public boolean hasNext () {
-                    do {
-                        if (i != null && i.hasNext()) 
-                            return true;
-                        if (c >= splits)
-                            return false;
-                        i = partition(rd,c++);
-                    } while (true);
+                    return i.hasNext();
                 }
             });
     }
@@ -370,15 +330,6 @@
         }
     }
 
-    private static JavaRDD<MRData> materialize ( JavaRDD<MRData> rdd ) {
-        return rdd.map(new Function<MRData,MRData>() {
-                public MRData call ( MRData value ) {
-                    value.materializeAll();
-                    return value;
-                };
-            }).cache();
-    }
-
     private final static Function<MRData,MRData> get_first
         = new Function<MRData,MRData>() {
                    public MRData call ( MRData value ) {
@@ -395,9 +346,10 @@
     final public DataSet eval ( final Tree e,
                                 final Environment env,
                                 final String counter ) {
-        JavaRDD<MRData> rd = eval(e,env);
+        JavaRDD<MRData> rd = eval(e,env,(Environment)null);
         long count = 0;
         rd = rd.cache();
+	cached_rdds.add(rd);
         if (!counter.equals("-")) {
             final Accumulator<Integer> c = spark_context.intAccumulator(0);
             rd.foreach(new VoidFunction<MRData>() {
@@ -415,14 +367,15 @@
     /** Evaluate an MRQL physical plan using Spark and print tracing info
      * @param e the physical plan
      * @param env contains bindings from variables to values (MRData)
+     * @param rdd_env contains bindings from variables to RDDs
      * @return a Spark RDD
      */
-    final public JavaRDD<MRData> eval ( final Tree e, final Environment env ) {
+    final public JavaRDD<MRData> eval ( final Tree e, final Environment env, final Environment rdd_env ) {
         if (Config.trace_execution) {
             tab_count += 3;
             System.out.println(tabs(tab_count)+print_query(e));
         };
-        final JavaRDD<MRData> res = evalD(e,env);
+        final JavaRDD<MRData> res = evalD(e,env,rdd_env);
         if (Config.trace_execution) 
             try {
                 System.out.println(tabs(tab_count)+"-> "+res.take(Config.max_bag_size_print));
@@ -435,11 +388,11 @@
 
     /* convert an MRQL lambda to a Spark Function */
     private static FlatMapFunction<MRData,MRData> cmap_fnc ( Tree fnc, Environment env ) {
-        final Environment master_env = global_env;
+        final Broadcast<Environment> broadcast_env = spark_context.broadcast(global_env);
         final org.apache.mrql.Function f = evalF(fnc,env);
         return new FlatMapFunction<MRData,MRData>() {
             public Iterable<MRData> call ( MRData value ) {
-                set_global_env(master_env);
+                global_env = broadcast_env.value();
                 return (Bag)f.eval(value);
             }
         };
@@ -449,17 +402,18 @@
     private static JavaRDD<MRData> groupBy ( JavaRDD<MRData> s, Tree fnc, Environment env, Tree o ) {
         match o {
         case true:   // the result must be sorted
-            return s.groupBy(get_first)
-                .sortByKey()
-                .map(new Function<Tuple2<MRData,Iterable<MRData>>,MRData>() {
-                        public MRData call ( Tuple2<MRData,Iterable<MRData>> value ) {
-                            return new Tuple(value._1,bag(value._2));
+            return s.mapToPair(new PairFunction<MRData,MRData,MRData>() {
+                    public Tuple2<MRData,MRData> call ( MRData value ) {
+                        Tuple t = (Tuple)value;
+                        return new Tuple2(t.first(),t.second());
+                    }
+                })
+                .sortByKey(true)
+                .map(new Function<Tuple2<MRData,MRData>,MRData>() {
+                        public MRData call ( Tuple2<MRData,MRData> value ) {
+                            return new Tuple(value._1,new Bag(value._2));
                         }})
-                .flatMap(cmap_fnc(fnc,env)).map(new Function<MRData,MRData>() {
-                        public MRData call ( MRData value ) {
-                            return ((Tuple)value).second();
-                        }
-                    });
+                .flatMap(cmap_fnc(fnc,env));
         };
         return s.groupBy(get_first)
             .map(new Function<Tuple2<MRData,Iterable<MRData>>,MRData> () {
@@ -479,10 +433,10 @@
     }
 
     static JavaRDD<MRData> containerData ( JavaPairRDD<MRContainer,MRContainer> rd ) {
-        final Environment master_env = global_env;
+        final Broadcast<Environment> broadcast_env = spark_context.broadcast(global_env);
         return rd.map(new Function<Tuple2<MRContainer,MRContainer>,MRData>() {
                 public MRData call ( Tuple2<MRContainer,MRContainer> value ) {
-                    set_global_env(master_env);
+                    global_env = broadcast_env.value();
                     return value._2.data();
                 }
             });
@@ -506,10 +460,8 @@
     }
 
     private static FlatMapFunction<Iterator<MRData>,MRData> combiner_fnc ( final org.apache.mrql.Function f ) {
-        final Environment master_env = global_env;
         return new FlatMapFunction<Iterator<MRData>,MRData>() {
                   public Iterable<MRData> call ( final Iterator<MRData> i ) {
-                      set_global_env(master_env);
                       return MapReduceAlgebra.cmap(new org.apache.mrql.Function() {
                               public MRData eval ( MRData x ) {
                                   final MRData key = ((Tuple)x).first();
@@ -535,77 +487,93 @@
         };
     }
 
-    private static Hashtable<MRData,Bag> make_built_table ( List<Tuple2<MRData,MRData>> values ) {
-        Hashtable<MRData,Bag> built_table = new Hashtable<MRData,Bag>(Config.map_cache_size);
-        for ( Tuple2<MRData,MRData> t: values ) {
-            Bag entries = built_table.get(t._1);
-            built_table.put(t._1,
-                            (entries == null)
-                            ? (new Bag(t._2))
-                            : entries.add_element(t._2));
-        };
-        return built_table;
-    }
-
     /** Evaluate MRQL physical operators using Spark
      * @param e the physical plan
-     * @param env contains bindings fro variables to values (MRData)
+     * @param env contains bindings from variables to values (MRData)
+     * @param rdd_env contains bindings from variables to RDDs
      * @return a Spark RDD
      */
-    final public JavaRDD<MRData> evalD ( final Tree e, final Environment env ) {
-        final Environment master_env = global_env;
+    final public JavaRDD<MRData> evalD ( final Tree e, final Environment env, final Environment rdd_env ) {
         try {
             match e {
             case MapAggregateReduce(`m,`r,null,_,`s,`o):
-                return evalD(#<MapReduce(`m,`r,`s,`o)>,env);
+                return evalD(#<MapReduce(`m,`r,`s,`o)>,env,rdd_env);
             case CrossAggregateProduct(`mx,`my,`r,null,_,`x,`y):
-                return evalD(#<CrossProduct(`mx,`my,`r,`x,`y)>,env);
+                return evalD(#<CrossProduct(`mx,`my,`r,`x,`y)>,env,rdd_env);
             case MapAggregateReduce2(`mx,`my,`r,null,_,`x,`y,`o):
-                return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env);
+                return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env,rdd_env);
             case MapAggregateJoin(`mx,`my,`r,null,_,`x,`y):
-                return evalD(#<MapJoin(`mx,`my,`r,`x,`y)>,env);
+                return evalD(#<MapJoin(`mx,`my,`r,`x,`y)>,env,rdd_env);
             case cMap(`f,`s):
-                return eval(s,env).flatMap(cmap_fnc(f,env));
+                return eval(s,env,rdd_env).flatMap(cmap_fnc(f,env));
             case MapReduce(`m,`r,`s,`o):
-                return groupBy(eval(s,env).flatMap(cmap_fnc(m,env)),r,env,o);
+                return groupBy(eval(s,env,rdd_env).flatMap(cmap_fnc(m,env)),r,env,o);
             case MapCombineReduce(`m,`c,`r,`s,`o):
-                return groupBy(eval(s,env).flatMap(cmap_fnc(m,env))
+                return groupBy(eval(s,env,rdd_env).flatMap(cmap_fnc(m,env))
                                .mapPartitions(combiner_fnc(evalF(c,env))),r,env,o);
             case MapCombineReduce2(`mx,`my,`c,`r,`x,`y,`o):
-                return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env).mapPartitions(combiner_fnc(evalF(c,env)));
+                return evalD(#<MapReduce2(`mx,`my,`r,`x,`y,`o)>,env,rdd_env).mapPartitions(combiner_fnc(evalF(c,env)));
             case CrossProduct(`mx,`my,`r,`x,`y):
                 final org.apache.mrql.Function fr = evalF(r,env);
-                return eval(x,env)
+                return eval(x,env,rdd_env)
                     .flatMap(cmap_fnc(mx,env))
-                    .cartesian(eval(y,env).flatMap(cmap_fnc(my,env)))
+                    .cartesian(eval(y,env,rdd_env).flatMap(cmap_fnc(my,env)))
                     .flatMap(new FlatMapFunction<Tuple2<MRData,MRData>,MRData>() {
                             public Iterable<MRData> call ( Tuple2<MRData,MRData> value ) {
-                                set_global_env(master_env);
                                 return (Bag)fr.eval(new Tuple(value._1,value._2));
                             }
                         });
             case MapReduce2(`mx,`my,`r,`x,`y,`o):
-                final org.apache.mrql.Function fx = evalF(mx,env);
-                final org.apache.mrql.Function fy = evalF(my,env);
                 final org.apache.mrql.Function fr = evalF(r,env);
+                boolean sfx = false;
+                match mx {
+                case compiled(`fmx,lambda(`vx,bag(`mex))):
+                    sfx = true;
+                    mx = #<lambda(`vx,`mex)>;
+                case lambda(`vx,bag(`mex)):
+                    sfx = true;
+                    mx = #<lambda(`vx,`mex)>;
+                };
+                final org.apache.mrql.Function fx = evalF(mx,env);
                 JavaPairRDD<MRData,MRData> xs
-                    = eval(x,env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
+                    = (sfx)
+                    ? eval(x,env,rdd_env).mapToPair(new PairFunction<MRData,MRData,MRData>() {
+                            public Tuple2<MRData,MRData> call ( MRData value ) {
+                                Tuple t = (Tuple)fx.eval(value);
+                                return new Tuple2(t.first(),t.second());
+                            }
+                        })
+                    : eval(x,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
                             public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
-                                set_global_env(master_env);
                                 return joinIterator(((Bag)fx.eval(value)).iterator());
                             }
                         });
+                boolean sfy = false;
+                match my {
+                case compiled(`fmy,lambda(`vy,bag(`mey))):
+                    sfy = true;
+                    my = #<lambda(`vy,`mey)>;
+                case lambda(`vy,bag(`mey)):
+                    sfy = true;
+                    my = #<lambda(`vy,`mey)>;
+                };
+                final org.apache.mrql.Function fy = evalF(my,env);
                 JavaPairRDD<MRData,MRData> ys
-                    = eval(y,env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
+                    = (sfy)
+                    ? eval(y,env,rdd_env).mapToPair(new PairFunction<MRData,MRData,MRData>() {
+                            public Tuple2<MRData,MRData> call ( MRData value ) {
+                                Tuple t = (Tuple)fy.eval(value);
+                                return new Tuple2(t.first(),t.second());
+                            }
+                        })
+                    : eval(y,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
                             public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
-                                set_global_env(master_env);
                                 return joinIterator(((Bag)fy.eval(value)).iterator());
                             }
                         });
 		return xs.cogroup(ys)
                     .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
                             public Iterable<MRData> call ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
-                                set_global_env(master_env);
                                 return (Bag)fr.eval(new Tuple(bag(value._2._1),bag(value._2._2)));
                             }
                         });
@@ -623,9 +591,8 @@
                 final MRData one = new MR_byte(1);
                 final MRData two = new MR_byte(2);
                 final JavaPairRDD<MRData,MRData> xs
-                    = eval(x,env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
+                    = eval(x,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
                             public Iterable<Tuple2<MRData,MRData>> call ( final MRData value ) {
-                                set_global_env(master_env);
                                 return new Iterable<Tuple2<MRData,MRData>>() {
                                     public Iterator<Tuple2<MRData,MRData>> iterator() {
                                         return new Iterator<Tuple2<MRData,MRData>>() {
@@ -645,9 +612,8 @@
                             }
                         });
                 final JavaPairRDD<MRData,MRData> ys
-                    = eval(y,env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
+                    = eval(y,env,rdd_env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
                             public Iterable<Tuple2<MRData,MRData>> call ( final MRData value ) {
-                                set_global_env(master_env);
                                 return new Iterable<Tuple2<MRData,MRData>>() {
                                     public Iterator<Tuple2<MRData,MRData>> iterator() {
                                         return new Iterator<Tuple2<MRData,MRData>>() {
@@ -666,7 +632,7 @@
                                 };
                             }
                         });
-                return xs.union(ys).groupByKey(Config.nodes)
+                return xs.union(ys).groupByKey()
                     .mapPartitions(new FlatMapFunction<Iterator<Tuple2<MRData,Iterable<MRData>>>,MRData>() {
                         public Iterable<MRData> call ( final Iterator<Tuple2<MRData,Iterable<MRData>>> value ) {
                             Bag xb = new Bag();
@@ -688,22 +654,116 @@
                                 }
                             };
                         }
+                        });                
+            case OuterMerge(`merge,`state,`data):
+                final org.apache.mrql.Function fm = evalF(merge,env);
+                JavaPairRDD<MRData,MRData> S = eval(state,env,rdd_env)
+                    .mapPartitionsToPair(new PairFlatMapFunction<Iterator<MRData>,MRData,MRData>() {
+                            public Iterable<Tuple2<MRData,MRData>> call ( final Iterator<MRData> i ) {
+                                return joinIterator(i);
+                            }
+                        },true);   // do not repartition the state S
+                JavaPairRDD<MRData,MRData> ds = eval(data,env,rdd_env)
+                    .mapToPair(new PairFunction<MRData,MRData,MRData>() {
+                            public Tuple2<MRData,MRData> call ( MRData value ) {
+                                Tuple t = (Tuple)value;
+                                return new Tuple2(t.first(),t.second());
+                            }
                         });
+                JavaRDD<MRData> res = S.cogroup(ds)
+                    .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
+                            public Iterable<MRData> call ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
+                                final Iterator<MRData> ix = value._2._1.iterator();
+                                final Iterator<MRData> iy = value._2._2.iterator();
+                                ArrayList<MRData> a = new ArrayList<MRData>();
+                                if (ix.hasNext())
+                                    if (iy.hasNext())
+                                        a.add(new Tuple(value._1,fm.eval(new Tuple(ix.next(),iy.next()))));
+                                    else a.add(new Tuple(value._1,ix.next()));
+                                else if (iy.hasNext())
+                                    a.add(new Tuple(value._1,iy.next()));
+                                return a;
+                            }
+                        }).cache();
+                cached_rdds.add(res);
+                return res;
+            case RightOuterMerge(`merge,`state,`data):
+                final org.apache.mrql.Function fm = evalF(merge,env);
+                JavaPairRDD<MRData,MRData> S = eval(state,env,rdd_env)
+                    .mapPartitionsToPair(new PairFlatMapFunction<Iterator<MRData>,MRData,MRData>() {
+                            public Iterable<Tuple2<MRData,MRData>> call ( final Iterator<MRData> i ) {
+                                return joinIterator(i);
+                            }
+                        },true);   // do not repartition the state S
+                JavaPairRDD<MRData,MRData> ds = eval(data,env,rdd_env)
+                    .mapToPair(new PairFunction<MRData,MRData,MRData>() {
+                            public Tuple2<MRData,MRData> call ( MRData value ) {
+                                Tuple t = (Tuple)value;
+                                return new Tuple2(t.first(),t.second());
+                            }
+                        });
+                JavaRDD<MRData> res = S.cogroup(ds)
+                    .flatMap(new FlatMapFunction<Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>>,MRData>() {
+                            public Iterable<MRData> call ( Tuple2<MRData,Tuple2<Iterable<MRData>,Iterable<MRData>>> value ) {
+                                final Iterator<MRData> ix = value._2._1.iterator();
+                                final Iterator<MRData> iy = value._2._2.iterator();
+                                return (ix.hasNext())
+                                    ? ((iy.hasNext())
+                                       ? new Bag(new Tuple(value._1,fm.eval(new Tuple(ix.next(),iy.next()))))
+                                       : new Bag())
+                                    : ((iy.hasNext())
+                                       ? new Bag(new Tuple(value._1,iy.next()))
+                                       : new Bag());
+                            }
+                        }).cache();
+                cached_rdds.add(res);
+                return res;
+            case MapJoin(lambda(`vx,bag(`mx)),lambda(`vy,bag(`my)),`r,`x,`y,`o):
+                final org.apache.mrql.Function fx = evalF(#<lambda(`vx,`mx)>,env);
+                final org.apache.mrql.Function fy = evalF(#<lambda(`vy,`my)>,env);
+                final org.apache.mrql.Function fr = evalF(r,env);
+                final Broadcast<Map<MRData,MRData>> ys
+                    = spark_context.broadcast(eval(y,env,rdd_env)
+                         .mapToPair(new PairFunction<MRData,MRData,MRData>() {
+                            public Tuple2<MRData,MRData> call ( MRData value ) {
+                                Tuple t = (Tuple)fy.eval(value);
+                                return new Tuple2(t.first(),t.second());
+                            }
+                        }).collectAsMap());
+                return eval(x,env,rdd_env).mapPartitions(new FlatMapFunction<Iterator<MRData>,MRData>() {
+                        public Iterable<MRData> call ( final Iterator<MRData> i ) {
+                            final Map<MRData,MRData> m = ys.value();
+                            return new Iterable<MRData>() {
+                                public Iterator<MRData> iterator() {
+                                    return new Iterator<MRData> () {
+                                        public MRData next () {
+                                            final Tuple p = (Tuple)fx.eval(i.next());
+                                            final MRData pd = m.get(p.first());
+                                            return ((Bag)fr.eval(new Tuple(p.second(),(pd == null) ? empty_bag : pd)));
+                                        }
+                                        public boolean hasNext () {
+                                            return i.hasNext();
+                                        }
+                                        public void remove () {}
+                                    };
+                                }
+                            };
+                        }
+                    },true);
             case MapJoin(`mx,`my,`r,`x,`y):
                 final org.apache.mrql.Function fx = evalF(mx,env);
                 final org.apache.mrql.Function fy = evalF(my,env);
                 final org.apache.mrql.Function fr = evalF(r,env);
-                final Broadcast<List<Tuple2<MRData,MRData>>> ys
-                    = spark_context.broadcast(eval(y,env).flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
-                                public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
-                                    set_global_env(master_env);
-                                    return joinIterator(((Bag)fy.eval(value)).iterator());
-                                }
-                            }).collect());
-                return eval(x,env).flatMap(new FlatMapFunction<MRData,MRData>() {
-                        final Hashtable<MRData,Bag> built_table = make_built_table(ys.value());
+                final Broadcast<Map<MRData,MRData>> ys
+                    = spark_context.broadcast(eval(y,env,rdd_env)
+                         .flatMapToPair(new PairFlatMapFunction<MRData,MRData,MRData>() {
+                            public Iterable<Tuple2<MRData,MRData>> call ( MRData value ) {
+                                return joinIterator(((Bag)fy.eval(value)).iterator());
+                            }
+                        }).collectAsMap());
+                return eval(x,env,rdd_env).flatMap(new FlatMapFunction<MRData,MRData>() {
                         public Iterable<MRData> call ( MRData value ) {
-                            set_global_env(master_env);
+                            final Map<MRData,MRData> m = ys.value();
                             final Iterator<MRData> i = ((Bag)fx.eval(value)).iterator();
                             return new Iterable<MRData>() {
                                 public Iterator<MRData> iterator() {
@@ -718,7 +778,7 @@
                                                 return true;
                                             while (i.hasNext()) {
                                                 p = (Tuple)i.next();
-                                                MRData pd = built_table.get(p.first());
+                                                MRData pd = m.get(p.first());
                                                 Bag bb = ((Bag)fr.eval(new Tuple(p.second(),(pd == null) ? empty_bag : pd)));
                                                 ix = bb.iterator();
                                                 if (ix.hasNext())
@@ -749,19 +809,18 @@
                                                               MRContainer.class,MRContainer.class,
                                                               Config.nodes));
             case Merge(`x,`y):
-                return eval(x,env).union(eval(y,env));
+                return eval(x,env,rdd_env).union(eval(y,env,rdd_env));
             case Repeat(lambda(`v,`b),`s,`n):
                 int max_num = ((MR_int)evalE(n,env)).get();
                 JavaRDD<MRData> rd;
-                JavaRDD<MRData> res = eval(s,env);
-                //res = materialize(res);
+                JavaRDD<MRData> res = eval(s,env,rdd_env).cache();
                 int i = 0;
                 boolean cont = true;
                 do {
-                    rd = eval(b,new Environment(v.toString(),new MR_rdd(res),env)).cache();
-                    res = rd.map(get_first);
-                    //res = materialize(res);
-		    long size = res.count();
+                    rd = eval(b,env,new Environment(v.toString(),new MR_rdd(res),rdd_env)).cache();
+                    res.unpersist();
+		    res = rd.map(get_first).cache();
+		    long size = rd.count();
                     Integer true_results
                         = rd.aggregate(new Integer(0),
                                        new Function2<Integer,MRData,Integer>() {
@@ -778,23 +837,28 @@
 			if (size == 0)
 			    System.err.println("Repeat #"+i);
 			else System.err.println("Repeat #"+i+": "+true_results+" true results out of "+size);
+                    rd.unpersist();
                 } while (cont);
+		cached_rdds.add(res);
                 return res;
             case Closure(lambda(`v,`b),`s,`m):
                 int max_num = ((MR_int)evalE(m,env)).get();
-                JavaRDD<MRData> res = eval(s,env).cache();
+                JavaRDD<MRData> res = eval(s,env,rdd_env).cache();
                 long n = 0;
                 long old = 0;
                 int i = 0;
                 boolean cont = true;
                 do {
-                    res = eval(b,new Environment(v.toString(),new MR_rdd(res),env)).cache();
+                    JavaRDD<MRData> rd = eval(b,env,new Environment(v.toString(),new MR_rdd(res),rdd_env)).cache();
+		    res.unpersist();
+		    res = rd;
                     old = n;
                     n = res.count();
                     i++;
                     if (!Config.testing)
                         System.err.println("Repeat #"+i+": "+(old-n)+" new records");
                 } while (old < n && i < max_num);
+		cached_rdds.add(res);
                 return res;
             case Generator(`min,`max,`size):
                 DataSet ds = Plan.generator(((MR_long)evalE(min,env)).get(),
@@ -815,20 +879,23 @@
                 else if (x instanceof Bag)
                     ((Bag)x).materialize();
                 new_distributed_binding(v.toString(),x);
-                return eval(body,new Environment(v.toString(),x,env));
+                return eval(body,new Environment(v.toString(),x,env),rdd_env);
             case let(`v,`u,`body):
                 MRData val = evalE(u,env);
                 Interpreter.new_global_binding(v.toString(),val);
-                return eval(body,new Environment(v.toString(),val,env));
+                return eval(body,new Environment(v.toString(),val,env),rdd_env);
             case Let(`v,`u,`body):
-                return eval(body,new Environment(v.toString(),new MR_rdd(eval(u,env).cache()),env));
+		JavaRDD<MRData> val = eval(u,env,rdd_env).cache();
+                JavaRDD<MRData> rd = eval(body,env,new Environment(v.toString(),new MR_rdd(val),rdd_env));
+		cached_rdds.add(val);
+                return rd;
             case If(`c,`x,`y):
                 if (((MR_bool)evalE(c,env)).get())
-                    return eval(x,env);
-                else return eval(y,env);
+                    return eval(x,env,rdd_env);
+                else return eval(y,env,rdd_env);
             case trace(`msg,`tp,`x):
                 long n = pre_trace(((MR_string)evalE(msg,env)).get());
-                JavaRDD<MRData> ds = evalD(x,env);
+                JavaRDD<MRData> ds = evalD(x,env,rdd_env);
                 trace(n,tp,new Bag(ds.take(Config.max_bag_size_print)));
                 return ds;
             case apply(`f,`arg):
@@ -839,7 +906,13 @@
             case `v:
                 if (!v.is_variable())
                     fail;
-                MRData x = variable_lookup(v.toString(),env);
+                MRData x = variable_lookup(v.toString(),rdd_env);
+                if (x != null && x instanceof MR_rdd)
+                    return ((MR_rdd)x).rdd();
+                x = variable_lookup(v.toString(),global_rdds);
+                if (x != null && x instanceof MR_rdd)
+                    return ((MR_rdd)x).rdd();
+                x = variable_lookup(v.toString(),global_env);
                 if (x != null)
                     if (x instanceof MR_dataset)
                         return ((RDDDataSource)((MR_dataset)x).dataset().source.get(0)).rdd;
@@ -855,12 +928,10 @@
                         l.add(x);
                         return spark_context.parallelize(l);
                     };
-                x = variable_lookup(v.toString(),global_env);
+                x = variable_lookup(v.toString(),env);
                 if (x != null)
                     if (x instanceof MR_dataset)
                         return ((RDDDataSource)((MR_dataset)x).dataset().source.get(0)).rdd;
-                    else if (x instanceof MR_rdd)
-                        return ((MR_rdd)x).rdd();
                     else if (x instanceof Bag) {
                         ArrayList<MRData> l = new ArrayList<MRData>();
                         for ( MRData a: (Bag)x )
diff --git a/spark/src/main/java/org/apache/mrql/SparkStreaming.gen b/spark/src/main/java/org/apache/mrql/SparkStreaming.gen
index d0ec379..1ae28b1 100644
--- a/spark/src/main/java/org/apache/mrql/SparkStreaming.gen
+++ b/spark/src/main/java/org/apache/mrql/SparkStreaming.gen
@@ -29,6 +29,7 @@
 import scala.collection.Seq;
 import scala.collection.JavaConversions;
 import scala.runtime.AbstractFunction2;
+import org.apache.spark.storage.RDDInfo;
 
 
 /** Evaluates physical plans in Apache Spark stream mode */
@@ -74,28 +75,29 @@
         return plan;
     }
 
-    private final static Evaluator ef = Evaluator.evaluator;
+    private final static SparkEvaluator ef = (SparkEvaluator)Evaluator.evaluator;
 
     /** bind the pattern variables to values */
-    private static Environment bind_list ( Tree pattern, Tree src, Environment env ) {
+    private static void bind_list ( Tree pattern, Tree src, Environment env, Environment rdd_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;
+                    bind_list(p,es.nth(i++),env,rdd_env);
             }
         };
-        MRData value = ef.evalE(src,env);
-        env.replace(pattern.toString(),value);
+        MRData value = new MR_rdd(ef.eval(src,env,rdd_env));
+        if (variable_lookup(pattern.toString(),global_rdds) == null)
+            global_rdds = new Environment(pattern.toString(),value,global_rdds);
+        else global_rdds.replace(pattern.toString(),value);
         if (repeat_variables.member(pattern))
             Interpreter.new_distributed_binding(pattern.toString(),value);
-        return env;
     }
 
-    private static void stream_processing ( final Tree plan, final Environment env, final Function f ) {
+    private static void stream_processing ( final Tree plan, final Environment env,
+                                            final Environment dataset_env, final Function f ) {
         if (streams.size() == 0)
             throw new Error("No input streams in query");
         ArrayList<DStream<?>> rdds = new ArrayList<DStream<?>>();
@@ -106,7 +108,7 @@
             new AbstractFunction2<Seq<RDD<?>>,Time,RDD<MRData>>() {
             public RDD<MRData> apply ( Seq<RDD<?>> rdds, Time tm ) {
                 long t = System.currentTimeMillis();
-                Environment new_env = env;
+                Environment rdd_env = dataset_env;
                 int i = 0;
                 for ( RDD<?> rdd: JavaConversions.seqAsJavaList(rdds) ) {
                     try {
@@ -115,17 +117,25 @@
                     } catch (Exception ex) {
                         return SparkEvaluator.spark_context.sc().emptyRDD(classtag);
                     };
-                    new_env = new Environment(vars.get(i++),new MR_rdd(new JavaRDD(rdd,classtag)),new_env);
+                    MRData d = new MR_rdd(new JavaRDD(rdd,classtag));
+                    global_rdds = new Environment(vars.get(i),d,global_rdds);
+                    rdd_env = new Environment(vars.get(i++),d,rdd_env);
                 };
                 match plan {
                 case lambda(`pat,`e):
-                    new_env = bind_list(pat,e,new_env);
+                    bind_list(pat,e,env,rdd_env);
                     f.eval(null);
                 case _: // non-incremental streaming
-                    f.eval(ef.evalE(plan,new_env));
+                    f.eval(ef.evalE(plan,env));
                 };
                 if (!Config.quiet_execution)
                     System.out.println("Run time: "+(System.currentTimeMillis()-t)/1000.0+" secs");
+                for ( RDD<?> rdd: JavaConversions.seqAsJavaList(rdds) )
+                    rdd.unpersist(true);
+		GC_cached_rdds();
+		System.gc();
+		for ( RDDInfo rdi: SparkEvaluator.spark_context.sc().getRDDStorageInfo() )
+                    System.err.println("Warning: cached RDD "+rdi);
 		if (++tries >= Config.stream_tries) {  // for performance experiments only
 		    tries = 0;
 		    SparkEvaluator.stream_context.stop(true,true);
@@ -138,16 +148,16 @@
     }
 
     /** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
-    final public static void evaluate ( Tree plan, Environment env, Function f ) {
+    final public static void evaluate ( Tree plan, Environment env, Environment dataset_env, Function f ) {
         streams = new ArrayList<JavaInputDStream<MRData>>();
         stream_names = new ArrayList<String>();
         match plan {
         case lambda(`p,`b):
             b = get_streams(b,env);
-            stream_processing(#<lambda(`p,`b)>,env,f);
+            stream_processing(#<lambda(`p,`b)>,env,null,f);
         case _:  // non-incremental streaming
             plan = get_streams(plan,env);
-            stream_processing(plan,env,f);
+            stream_processing(plan,env,dataset_env,f);
         };
         stream_context.start();
         stream_context.awaitTermination();