/**
 * 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.*;

/** Generates code for streaming queries */
public class Streaming extends AlgebraicOptimization {
    private static int istate = 0; 
    // set it to true to debug the monoid inference
    private static boolean inference_tracing = false;
    // set it to true to debug the split function
    private static boolean split_tracing = false;

    /** An environment that binds variables to monoids */
    final static class Environment {
        public String name;
        public Tree monoid;
        public Environment next;

        Environment ( String n, Tree m, Environment next ) {
            name = n;
            monoid = m;
            this.next = next;
        }

        public String toString () {
            return name+": "+monoid+((next==null)?"":", "+next.toString());
        }
    }

    /** Check if name is defined in the environment
     * @param name a variable name
     * @param env an environment that binds variables to monoids
     * @return true if the name is defined in env
     */
    private static boolean member ( String name, Environment env ) {
        for ( Environment en = env; en != null; en = en.next )
            if (en.name.equals(name))
                return true;
        return false;
    }

    /** if name is defined in the environment, return its binding
     * @param name a variable name
     * @param env an environment that binds variables to monoids
     * @return the monoid associated with the variable
     */
    private static Tree find ( String name, Environment env ) {
        for ( Environment en = env; en != null; en = en.next )
            if (en.name.equals(name))
                return en.monoid;
        return #<none>;
    }

    private static Environment repeat_environment = null;

    /* true if e is a variable in a repeat loop */
    private static boolean repeat_var ( Tree e ) {
        return e.is_variable() && member(e.toString(),repeat_environment);
    }

    private static int tab_count = -3;

    /** If e is a homomorphism, return the associated monoid;
     * if it returns none, then it fails to find a monoid;
     * if it returns fixed, then its invariant.
     * @param e an MRQL algebraic term
     * @param env binds variables to monoids
     * @return the monoid, or none if it doesn't exist
     */
    static Tree inference ( Tree e, Environment env ) {
        if (inference_tracing) {
            tab_count += 3;
            System.out.println(Interpreter.tabs(tab_count)+e);
            if (env != null)
                System.out.println(Interpreter.tabs(tab_count)+env.toString());

        };
        Tree res = inference_trace(e,env);
        if (inference_tracing) {
            System.out.println(Interpreter.tabs(tab_count)+"-> "+res);
            tab_count -= 3;
        };
        return res;
    }

    private static Tree inference_trace ( Tree e, Environment env ) {
        match e {
        case groupBy(`x):
            match inference(x,env) {
            case none: return #<none>;
            case fixed: return #<fixed>;
            case `m:
                return #<groupBy(`m)>;
            }
        case orderBy(`x):
            match inference(x,env) {
            case none: return #<none>;
            case fixed: return #<fixed>;
            case `m:
                return #<orderBy(`m)>;
            }
        case coGroup(`x,`y):
            Tree mx = inference(x,env);
            Tree my = inference(y,env);
            if (mx.equals(#<none>) || my.equals(#<none>))
                return #<none>;
            else if (mx.equals(#<fixed>) && my.equals(#<fixed>))
                return #<fixed>;
            else return #<groupBy(product(`mx,`my))>;
        case cmap(lambda(`v,bag(`w)),`x):
            if (!w.equals(v))
                fail;
            return inference(x,env);
        case cmap(lambda(`v,bag(tuple(`k,`u))),`x):
            match inference(x,env) {
            case `gb(`m):
                if (! #[groupBy,orderBy].member(#<`gb>) ) fail;
                match inference(k,new Environment(v.toString(),#<product(fixed,`m)>,env)) {
                case fixed:
                    Tree b = inference(u,new Environment(v.toString(),#<product(fixed,`m)>,env));
                    if (b.equals(#<none>))
                        fail;
                    return #<`gb(`b)>;
                }
            };
            fail
        case cmap(lambda(`v,`b),`x):
            match inference(x,env) {
            case union:
                match inference(b,new Environment(v.toString(),#<fixed>,env)) {
                case `gb(`m):
                    return #<`gb(`m)>;
                };
                return #<union>;
            case `gb(`m):
                if (! #[groupBy,orderBy].member(#<`gb>) )
                    fail;
                match inference(b,new Environment(v.toString(),#<product(fixed,`m)>,env)) {
                case `gb2(`m2):
                    if (! #[groupBy,orderBy].member(#<`gb2>) )
                        fail;
                    return #<`gb2(`m2)>;
                case union:
                    return #<union>;
                case fixed:
                    return #<fixed>;
                }
            case fixed:
                match inference(b,new Environment(v.toString(),#<fixed>,env)) {
                case union:
                    return #<union>;
                case fixed:
                    return #<fixed>;
                case `gb(`m):
                    if (! #[groupBy,orderBy].member(#<`gb>) )
                        fail;
                    return #<`gb(`m)>;
                }
            };
            fail
        case reduce(groupBy(count),`u):
            if (inference(u,env).equals(#<none>))
                fail;
            return #<groupBy(count)>;
        case reduce(groupBy(`m),`u):
            match inference(u,env) {
            case groupBy(union):
                return #<groupBy(`m)>;
            };
            return #<none>;
        case reduce(count,`u):
            if (inference(u,env).equals(#<none>))
                fail;
            return #<count>;
        case reduce(`m,`u):
            match inference(u,env) {
            case union:
                return m;
            case fixed:
                return #<fixed>;
            };
            return #<none>;
        case tuple(...s):
            Trees ms = #[ ];
            for ( Tree x: s )
                match inference(x,env) {
                case none:
                    return #<none>;
                case `m:
                    ms = ms.append(m);
                };
            boolean fixed = true;
            for (Tree mm: ms)
                fixed &= mm.equals(#<fixed>);
            if (fixed)
                return #<fixed>;
            else return #<product(...ms)>;
        case record(...s):
           Trees ms = #[ ];
            for ( Tree b: s )
                match b {
                case bind(`n,`x):
                    match inference(x,env) {
                    case none:
                        return #<none>;
                    case `m:
                        ms = ms.append(#<bind(`n,`m)>);
                    }
                };
            boolean fixed = true;
            for (Tree mm: ms)
                match mm {
                case bind(_,`m):
                    fixed &= m.equals(#<fixed>);
                };
            if (fixed)
                return #<fixed>;
            else return #<record(...ms)>;
        case nth(`u,`i):
            int n = (int)i.longValue();
            match inference(u,env) {
            case fixed:
                return #<fixed>;
            case product(...ms):
                return ms.nth(n);
            };
            return #<none>;
        case project(`u,`a):
            match inference(u,env) {
            case fixed:
                return #<fixed>;
            };
            return #<none>;
        case if(`p,`e1,`e2):
            return inference(e1,env);
        case bag(`u):
            match inference(u,env) {
            case fixed:
                return #<fixed>;
            };
            return #<none>;
        case call(stream,...):
            return #<union>;
        case call(source,...):
            return #<fixed>;
        case call(`f,...s):
            Trees ms = #[ ];
            for ( Tree x: s )
                match inference(x,env) {
                case none:
                    return #<none>;
                case `m:
                    ms = ms.append(m);
                };
            boolean fixed = true;
            for (Tree mm: ms)
                fixed &= mm.equals(#<fixed>);
            if (fixed)
                return #<fixed>;
            return #<none>;
        case bind(_,`u):
            return inference(u,env);
        case typed(`u,_):
            return inference(u,env);
        case _:
            if (e.is_string() || e.is_long() || e.is_double())
                return #<fixed>;
            else if (e.is_variable())
                if (repeat_var(e))
                    return find(e.toString(),repeat_environment);
                else if (member(e.toString(),env))
                    return find(e.toString(),env);
                else return #<fixed>;
        };
        return #<none>;
    }

    public static Tree inference ( Tree e ) {
        return inference(e,null);
    }

    private static Tree smap1 ( Tree f, Tree e ) {
        Tree nv = new_var();
        Tree b = new_var();
        return #<cmap(lambda(`nv,cmap(lambda(`b,bag(tuple(nth(`nv,0),`b))),
                                      apply(`f,tuple(nth(nth(`nv,0),0),nth(`nv,1))))),
                      `e)>;
    }

    private static Tree smap2 ( Tree f, Tree e ) {
        Tree nv = new_var();
        Tree nw = new_var();
        return #<cmap(lambda(`nv,cmap(lambda(`nw,bag(tuple(nth(`nw,0),
                                                           tuple(nth(`nv,0),nth(`nw,1))))),
                                      apply(`f,tuple(nth(nth(`nv,0),0),nth(`nv,1))))),
                      `e)>;
    }

    private static Tree smap3 ( Tree f, Tree e ) {
        Tree nv = new_var();
        Tree nw = new_var();
        return #<cmap(lambda(`nv,cmap(lambda(`nw,bag(tuple(nth(`nw,0),
                                                           tuple(tuple(),nth(`nw,1))))),
                                      apply(`f,`nv))),
                      `e)>;
    }

    private static Tree swap ( Tree e ) {
        Tree nv = new_var();
        return #<cmap(lambda(`nv,bag(tuple(tuple(nth(`nv,0),nth(nth(`nv,1),0)),
                                           nth(nth(`nv,1),1)))),
                      `e)>;
    }

    private static Tree mix ( Tree e ) {
        Tree nv = new_var();
        Tree x = new_var();
        Tree y = new_var();
        Tree b = #<bag(tuple(tuple(nth(`nv,0),
                                   tuple(nth(`x,0),nth(`y,0))),
                             tuple(nth(`x,1),nth(`y,1))))>;
        return #<cmap(lambda(`nv,cmap(lambda(`x,cmap(lambda(`y,`b),
                                                     groupBy(nth(nth(`nv,1),1)))),
                                      groupBy(nth(nth(`nv,1),0)))),
                      `e)>;
    }

    /** inject lineage (all groupBy/join keys) to a query e
     * @param e an MRQL query
     * @return an algebraic term that associates lineage to every query result
     */
    public static Tree inject_q ( Tree e ) {
        match e {
        case reduce(`m,cmap(`f,call(stream,...s))):
            return #<bag(tuple(tuple(),`e))>;
        case reduce(`m,cmap(`f,`v)):
            if (!repeat_var(v))
                fail;
            return #<bag(tuple(tuple(),`e))>;
        case reduce(`m,cmap(`f,`u)):
            return #<reduce(groupBy(`m),`(smap1(f,inject_e(u))))>;
        case reduce(`m,call(stream,...s)):
            return #<bag(tuple(tuple(),`e))>;
        case reduce(`m,`v):
            if (!repeat_var(v))
                fail;
            return #<bag(tuple(tuple(),`e))>;
        case cmap(`f,call(stream,...s)):
            Tree a = new_var();
            Tree b = new_var();
            return #<cmap(lambda(`a,cmap(lambda(`b,bag(tuple(tuple(),`b))),
                                         apply(`f,`a))),
                          call(stream,...s))>;
        case cmap(`f,`v):
            if (!repeat_var(v))
                fail;
            Tree a = new_var();
            Tree b = new_var();
            return #<cmap(lambda(`a,cmap(lambda(`b,bag(tuple(tuple(),`b))),
                                         apply(`f,`a))),
                          `v)>;
        case cmap(`f,`u):
            return smap1(f,inject_e(u));
        };
        // otherwise, it's a constant
        return #<bag(tuple(tuple(),`e))>;
    }

    private static boolean stream_source ( Tree e ) {
        match e {
        case call(stream,...):
            return true;
        };
        return repeat_var(e);
    }

    /** inject lineage (all groupBy/join keys) to a groupBy/join term e
     * @param e a groupBy or coGroup (join) term
     * @return an algebraic term that associates lineage to every query result
     */
    static Tree inject_e ( Tree e ) {
        match e {
        case `gb(`c):
            if (! #[groupBy,orderBy].member(#<`gb>) )
                fail;
            return #<`gb(`(swap(inject_c(c))))>;
        case coGroup(`c1,`c2):
            match c1 {
            case cmap(_,`u):
                if (!stream_source(u))
                    fail;
                match c2 {
                case cmap(_,`s):
                    if (!stream_source(s))
                        fail;
                    Tree nv = new_var();
                    return #<cmap(lambda(`nv,bag(tuple(tuple(nth(`nv,0),tuple(tuple(),tuple())),
                                                       nth(`nv,1)))),
                                  coGroup(`c1,`c2))>;
                };
                Tree nv = new_var();
                Tree y = new_var();
                return #<cmap(lambda(`nv,cmap(lambda(`y,bag(tuple(tuple(nth(`nv,0),
                                                                        tuple(tuple(),nth(`y,0))),
                                                                  tuple(nth(nth(`nv,1),0),
                                                                        nth(`y,1))))),
                                              groupBy(nth(nth(`nv,1),1)))),
                              coGroup(`c1,`(inject_c(c2))))>;
            };
            match c2 {
            case cmap(_,`u):
                if (!stream_source(u))
                    fail;
                Tree nv = new_var();
                Tree x = new_var();
                return #<cmap(lambda(`nv,cmap(lambda(`x,bag(tuple(tuple(nth(`nv,0),
                                                                        tuple(nth(`x,0),tuple())),
                                                                  tuple(nth(`x,1),
                                                                        nth(nth(`nv,1),1))))),
                                              groupBy(nth(nth(`nv,1),0)))),
                              coGroup(`(inject_c(c1)),`c2))>;
            };
            return mix(#<coGroup(`(inject_c(c1)),`(inject_c(c2)))>);
        };
        return inject_c(e);
    }

    /** Inject lineage (all groupBy/join keys) to a cMap e
     * @param e a cMap term
     * @return an algebraic term that associates lineage to every query result
     */
    static Tree inject_c ( Tree e ) {
        match e {
        case cmap(`f,call(stream,...s)):
            return smap3(f,#<call(stream,...s)>);
        case cmap(`f,`v):
            if (!repeat_var(v))
                fail;
            return smap3(f,v);
        case cmap(`f,`u):
            return smap2(f,inject_e(u));
        case call(stream,...s):
            return smap3(#<lambda(x,bag(x))>,#<call(stream,...s)>);
        case `v:
            if (!repeat_var(v))
                fail;
            return smap3(#<lambda(x,bag(x))>,v);
        };
        return e;
    }

    /** find all homomorphic terms in the algebraic term e
     * @param e an algebraic term
     * @param env binds variables to monoids
     * @return a list of homomorphic terms
     */
    static Trees find_homomorphisms ( Tree e, Environment env ) {
        if (! #[none,fixed].member(inference(e,env)))
            return #[ `e ];
        match e {
        case cmap(lambda(`v,bag(`b)),`x):
            match inference(x,env) {
            case union:
                return #[ `e ];
            case `gb(`m):
                if (! #[groupBy,orderBy].member(#<`gb>) ) fail;
                match inference(b,new Environment(v.toString(),#<product(fixed,`m)>,env)) {
                case product(...):
                    return #[ `e ];
                }
            case fixed:
                match inference(b,new Environment(v.toString(),#<fixed>,env)) {
                case union:
                    return #[ `e ];
                case fixed: 
                    return #[ `e ];
                }
            };
            fail
        case record(...ds):
            Trees bs = #[ ];
            for ( Tree d: ds )
                match d {
                case bind(`n,`a):
                    bs = union(bs,find_homomorphisms(a,env));
                };
            return bs;
        case call(`f,...as):
            Trees bs = #[ ];
            for ( Tree a: as )
                bs = union(bs,find_homomorphisms(a,env));
            return bs;
        case nth(`u,`n):
            Trees bs = #[ ];
            for ( Tree x: find_homomorphisms(u,env) )
                bs = union(bs,#[nth(`x,`n)]);
            return bs;
        case project(`u,`a):
            Trees bs = #[ ];
            for ( Tree x: find_homomorphisms(u,env) )
                bs = union(bs,#[project(`x,`a)]);
            return bs;
        case `f(...as):
            Trees bs = #[ ];
            for ( Tree a: as )
                bs = union(bs,find_homomorphisms(a,env));
            return bs;
        };
        return #[ ];
    }

    /** Split the term e into a homomorhism and an answer function
     * @param e an algebraic term
     * @param var the input variable of the answer function
     * @param env binds variables to monoids
     * @return a pair (answer,homomorphism)
     */
    static Tree split ( Tree e, Tree var, Environment env ) {
        if (split_tracing) {
            tab_count += 3;
            System.out.println(Interpreter.tabs(tab_count)+var+"\n"
                               +Interpreter.tabs(tab_count)+e.pretty(tab_count));

        };
        Tree res = split_trace(e,var,env);
        if (split_tracing) {
            System.out.println(Interpreter.tabs(tab_count)+"-> "+res.pretty(tab_count+3));
            tab_count -= 3;
        };
        return res;
    }

    /** Split the term e into a homomorhism and an answer function
     * @param e an algebraic term
     * @param var the input variable of the answer function
     * @param env binds variables to monoids
     * @return a pair (answer,homomorphism)
     */
    static Tree split_trace ( Tree e, Tree var, Environment env ) {
        match e {
        case cmap(lambda(`v,bag(tuple(`k,`b))),`u):
            Environment nenv = env;
            Tree nv = new_var();
            match inference(u,env) {
            case union:
                nv = #<nth(`nv,1)>;
                nenv = new Environment(v.toString(),#<union>,nenv);
            case `gb(`m):
                if (! #[groupBy,orderBy].member(#<`gb>) )
                    fail;
                nenv = new Environment(v.toString(),#<product(fixed,`m)>,nenv);
            };
            Trees hs = find_homomorphisms(b,nenv);
            if (hs.length() == 1 && hs.nth(0).equals(b))
                return #<pair(`var,`e)>;
            Tree ne = b;
            int i = 0;
            for ( Tree h: hs ) {
                ne = subst(h,#<nth(nth(`nv,1),`i)>,ne);
                i++;
            };
            return #<pair(cmap(lambda(`nv,bag(tuple(nth(`nv,0),`ne))),`var),
                          cmap(lambda(`v,bag(tuple(`k,tuple(...hs)))),`u))>;
        case cmap(lambda(`v,`b),`u):
            match inference(u,env) {
            case `gb(`m):
                if (! #[groupBy,orderBy].member(#<`gb>) )
                    fail;
                Environment nenv = new Environment(v.toString(),#<product(fixed,`m)>,env);
                match inference(b,nenv) {
                case `gb2(`m2):
                    if (! #[groupBy,orderBy].member(#<`gb2>) )
                        fail;
                    return #<pair(`var,`e)>;
                case _:
                    Tree nv = new_var();
                    match split(b,nv,nenv) {
                    case pair(`a,`h):
                        Tree nw = new_var();
                        if (occurences(v,a) == 0) {
                            a = subst(nv,var,a);
                            return #<pair(`a,cmap(lambda(`v,`h),`u))>;
                        };
                        // The answer function needs v; need to piggyback v along with the state
                        a = subst(nv,#<bag(tuple(nth(`nw,0),nth(nth(`nw,1),1)))>,
                                  subst(v,#<nth(nth(`nw,1),0)>,a));
                        return #<pair(cmap(lambda(`nw,`a),`var),
                                      cmap(lambda(`v,cmap(lambda(`nv,bag(tuple(nth(`nv,0),
                                                                               tuple(`v,nth(`nv,1))))),
                                                          `h)),
                                           `u))>;
                    }
                }
            case `m:
                if (! #[union,fixed].member(m))
                    fail;
                Tree nv = new_var();
                match split(b,nv,new Environment(v.toString(),m,env)) {
                case pair(`a,`h):
                    Tree nw = new_var();
                    if (occurences(v,a) == 0) {
                        a = subst(nv,var,a);
                        return #<pair(`a,cmap(lambda(`v,`h),`u))>;
                    };
                    // The answer function needs v; need to piggyback v along with the state
                    a = subst(nv,#<bag(tuple(nth(`nw,0),nth(nth(`nw,1),1)))>,
                              subst(v,#<nth(nth(`nw,1),0)>,a));
                    return #<pair(cmap(lambda(`nw,`a),`var),
                                  cmap(lambda(`v,cmap(lambda(`nv,bag(tuple(nth(`nv,0),
                                                                           tuple(`v,nth(`nv,1))))),
                                                      `h)),
                                       `u))>;
                }
            }
        case call(stream,...):
            return #<pair(`var,`e)>;
        case reduce(`m,`u):
            match inference(u,env) {
            case union:
                return #<pair(`var,`e)>;
            case _:
                match split(u,var,env) {
                case pair(`b,`h):
                    return #<pair(reduce(`m,`b),`h)>;
                }
            };
        case tuple(...as):
            Trees bs = #[ ];
            Trees hs = #[ ];
            int i = 0;
            for ( Tree a: as )
                match split(a,#<nth(`var,`(i++))>,env) {
                case pair(`b,`h):
                    bs = bs.append(b);
                    hs = hs.append(h);
                };
            return #<pair(tuple(...bs),tuple(...hs))>;
        case record(...as):
            Trees bs = #[ ];
            Trees hs = #[ ];
            for ( Tree a: as )
                match a {
                case bind(`v,`u):
                    match split(u,#<project(`var,`a)>,env) {
                    case pair(`b,`h):
                        bs = bs.append(#<bind(`v,`b)>);
                        hs = hs.append(#<bind(`v,`h)>);
                    }
                };
            return #<pair(record(...bs),record(...hs))>;
        case call(`f,...as):
            if (! #[fixed].member(inference(e,env)))
                fail;
            return #<pair(`e,tuple())>;
        case call(`f,`a):
            match split(a,var,env) {
            case pair(`b,`h):
                return #<pair(call(`f,`b),`h)>;
            };
        case call(`f,...as):
            Tree nv = new_var();
            Trees bs = #[ ];
            Trees hs = #[ ];
            int i = 0;
            for ( Tree a: as )
                match split(a,#<nth(`nv,`(i++))>,env) {
                case pair(`b,`h):
                    bs = bs.append(b);
                    hs = hs.append(h);
                };
            return #<pair(apply(lambda(`nv,call(`f,...bs)),`var),tuple(...hs))>;
        case nth(`u,`n):
            match split(u,var,env) {
            case pair(`b,`h):
                return #<pair(nth(`b,`n),`h)>;
            };
        case project(`u,`a):
            match split(u,var,env) {
            case pair(`b,`h):
                return #<pair(project(`b,`a),`h)>;
            };
        case bag(tuple(`k,`u)):
            Tree ne = #<nth(call(elem,`var),1)>;
            match TypeInference.type_inference(u) {
            case `S(_):
                if (!is_persistent_collection(S))
                    fail;
                ne = #<nth(call(elem,Collect(`var)),1)>;
            };
            match split(u,ne,env) {
            case pair(`b,`h):
                return #<pair(`b,bag(tuple(`k,`h)))>;
            };
        case if(`p,`u,bag()):
            match split(u,var,env) {
            case pair(`b,`h):
                return #<pair(`b,if(`p,`h,bag()))>;
            };
        case `f(...as):
            Tree nv = new_var();
            Trees bs = #[ ];
            Trees hs = #[ ];
            int i = 0;
            for ( Tree a: as )
                match split(a,#<nth(`nv,`(i++))>,env) {
                case pair(`b,`h):
                    bs = bs.append(b);
                    hs = hs.append(h);
                };
            return #<pair(apply(lambda(`nv,`f(...bs)),`var),tuple(...hs))>;
        };
        return #<pair(`var,`e)>;
    }

    /** Return the monoid of a query e */
    private static Tree get_monoid ( Tree e, Environment env ) {
        match e {
        case reduce(`m,`u):
            return #<groupBy(`m)>;
        case cmap(`f,call(stream,...s)):
            return #<union>;
        case cmap(`f,`v):
            if (!repeat_var(v))
                fail;
            return #<union>;
        case cmap(lambda(`v,`b),`u):
            match inference(u,env) {
            case groupBy(`m):
                return get_monoid(b,new Environment(v.toString(),#<product(fixed,`m)>,env));
            case `m:
                return get_monoid(b,env);
            }
        case bag(tuple(`k,`v)):
            Tree m = inference(v,env);
            return #<groupBy(`m)>;
        case if(`p,`e1,bag()):
            return get_monoid(e1,env);
        };
        throw new Error("Unable to find the merge function for: "+e);
        //return #<groupBy(fixed)>;
    }

    private static Tree convert_reduce_all ( Tree m, Tree e ) {
        match m {
        case fixed:
            return #<call(elem,`e)>;
        case union:
            Tree v = new_var();
            return #<cmap(lambda(`v,`v),`e)>;
        case product(...ps):
            int i = 0;
            Trees as = #[ ];
            for ( Tree p: ps ) {
                Tree nv = new_var();
                as = as.append(convert_reduce_all(p,#<cmap(lambda(`nv,bag(nth(`nv,`(i++)))),`e)>));
            };
            return #<tuple(...as)>;
        case record(...bs):
            Trees as = #[ ];
            for ( Tree b: bs )
                match b {
                case bind(`n,`p):
                    Tree nv = new_var();
                    Tree nb = convert_reduce_all(p,#<cmap(lambda(`nv,bag(project(`nv,`n))),`e)>);
                    as = as.append(#<bind(`n,`nb)>);
                };
            return #<record(...as)>;
        case `gb(`gm):
            if (! #[groupBy,orderBy].member(#<`gb>) )
                fail;
            Tree v = new_var();
            Tree me = convert_reduce_all(gm,#<nth(`v,1)>);
            return #<cmap(lambda(`v,bag(tuple(nth(`v,0),`me))),
                          `gb(`e))>;
        case _:
            if (!m.is_variable())
                fail;
            for ( Tree monoid: monoids )
                match monoid {
                case `aggr(`mtp,`plus,`zero,`unit):
                    if (#<`aggr>.equals(m)) {
                        plus = Normalization.rename(plus);
                        return #<aggregate(`plus,`zero,`e)>;
                    }
                };
        };
        return #<call(elem,`e)>;
    }

    /** Convert a reduce on the monoid m to an aggregation */
    private static Tree convert_reduce ( Tree m, Tree e ) {
        match m {
        case fixed:
            return e;
        case union:
            return e;
        case product(...ps):
            int i = 0;
            Trees as = #[ ];
            for ( Tree p: ps )
                as = as.append(convert_reduce(p,#<nth(`e,`(i++))>));
            return #<tuple(...as)>;
        case record(...bs):
            Trees as = #[ ];
            for ( Tree b: bs )
                match b {
                case bind(`n,`p):
                    as = as.append(#<bind(`n,`(convert_reduce(p,#<project(`e,`n)>)))>);
                };
            return #<record(...as)>;
        case `gb(`gm):
            if (! #[groupBy,orderBy].member(#<`gb>) )
                fail;
            Tree v = new_var();
            Tree me = convert_reduce_all(gm,#<nth(`v,1)>);
            return #<cmap(lambda(`v,bag(tuple(nth(`v,0),`me))),
                          `gb(`e))>;
        case _:
            if (!m.is_variable())
                fail;
            for ( Tree monoid: monoids )
                match monoid {
                case `aggr(`mtp,`plus,`zero,`unit):
                    if (#<`aggr>.equals(m))
                        return #<call(`aggr,`e)>;
                };
        };
        return e;
    }

    /** Convert coGroups back to joins and reduces to aggregations */
    static Tree convert_to_algebra ( Tree e ) {
        match e {
        case coGroup(`x,`y):
            Tree v = new_var();
            Tree vx = new_var();
            Tree vy = new_var();
            Tree cx = convert_to_algebra(x);
            Tree cy = convert_to_algebra(y);
            Tree mx = new_var();
            Tree my = new_var();
            return #<join(lambda(`vx,nth(`vx,0)),lambda(`vy,nth(`vy,0)),
                          lambda(`v,bag(tuple(call(join_key,nth(`v,0),nth(`v,1)),
                                              tuple(cmap(lambda(`vx,bag(nth(`vx,1))),nth(`v,0)),
                                                    cmap(lambda(`vy,bag(nth(`vy,1))),nth(`v,1)))))),
                          `cx, `cy)>;
      case reduce(`aggr,`s):
            return convert_reduce(aggr,convert_to_algebra(s));
        case `f(...as):
            Trees bs = #[ ];
            for ( Tree a: as )
                bs = bs.append(convert_to_algebra(a));
            return #<`f(...bs)>;
        };
        return e;
    }

    /** If the key of the join for merging states contains a float/double number,
     *  use approximing equality */
    private static Tree key_equality ( Tree type, Tree e ) {
        if (occurences(#<float>,type) == 0 && occurences(#<double>,type) == 0)
            return e;
        match type {
        case tuple(...as):
            Trees bs = #[ ];
            int i = 0;
            for ( Tree a: as )
                bs = bs.append(key_equality(a,#<nth(`e,`(i++))>));
            return #<tuple(...bs)>;
        case record(...bs):
            Trees cs = #[ ];
            for ( Tree b: bs )
                match b {
                case bind(`n,`a):
                    cs = cs.append(#<bind(`n,`(key_equality(a,#<project(`e,`n)>)))>);
                }
            return #<record(...cs)>;
        case `T(`tp):
            if (!is_collection(T))
                fail;
            Tree nv = new_var();
            return #<cmap(lambda(`nv,`(key_equality(tp,nv))),`e)>;
        case float:
            return #<call(round,`e)>;
        case double:
            return #<call(round,`e)>;
        };
        return e;
    }

    /** true if there is only one instance value of type tp */
    private static boolean unique_key ( Tree tp ) {
        match tp {
        case tuple():
            return true;
        case `T(`t):
            if (!is_collection(T))
                fail;
            return unique_key(t);
        case tuple(...as):
            for ( Tree a: as )
                if (!unique_key(a))
                    return false;
            return true;
        case record(...as):
            for ( Tree a: as )
                match a {
                case bind(_,`b):
                    if (!unique_key(b))
                        return false;
                }
            return true;
        };
        return false;
    }

    /** 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, 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 #<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)>,false);
                    if (is_persistent_collection(T)) {
                        X = #<Collect(`X)>;
                        Y = #<Collect(`Y)>;
                    };
                    return #<let(`vx,`X,
                                 let(`vy,`Y,
                                     if(call(exists,`vx),
                                        if(call(exists,`vy),
                                           bag(tuple(`keytp,`v)),
                                           `vx),
                                        `vy)))>;
                };
                // 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 my = new_var();
                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)>;
                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),
                                            let(`my,nth(`v,1),
                                                if(call(exists,`mx),
                                                   if(call(exists,`my),`b,`mx),
                                                   `my)))),
                              `X, `Y)>;
            };
            throw new Error("Unknown type for merge: "+type);
        case product(...ms):
            Trees tps = ((Node)type).children();
            Trees bs = #[ ];
            int i = 0;
            for ( Tree a: ms ) {
                bs = bs.append(merge(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(a,((Node)tps.nth(i++)).children().nth(1),
                                                     #<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:
            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 zero element of the monoid m */
    private static Tree zero ( Tree m ) {
        match m {
        case `gb(`n):
            if (! #[groupBy,orderBy].member(#<`gb>) )
                fail;
            return #<bag()>;
        case product(...ms):
            Trees bs = #[ ];
            for ( Tree a: ms )
                bs = bs.append(zero(a));
            return #<tuple(...bs)>;
        case record(...bs):
            Trees cs = #[ ];
            for ( Tree b: bs )
                match b {
                case bind(`n,`a):
                    cs = cs.append(#<bind(`n,`(zero(a)))>);
                }
            return #<record(...cs)>;
        case fixed:
            return #<bag()>;
        case union:
            return #<bag()>;
        case _:
            if (!m.is_variable())
                fail;
            for ( Tree monoid: monoids )
                match monoid {
                case `aggr(`mtp,`plus,`zero,`unit):
                    if (#<`aggr>.equals(m))
                        return zero;
                };
        };
        throw new Error("Undefined monoid: "+m);
    }

    /** Convert joins to coGroups, plus other transformations */
    static Tree normalize_term ( Tree e ) {
        match e {
        case join(`kx,`ky,`r,cmap(`f,`x),cmap(`g,`y)):
            if (!x.equals(y))
                fail;
            Tree v = new_var(), w = new_var(),
                wx = new_var(), wy = new_var(),
                vx = new_var(), vy = new_var();
            Tree xtp = TypeInference.type_inference(((Node)kx).children.head);
            Tree ytp = TypeInference.type_inference(((Node)ky).children.head);
            Tree T = #<union(inL(`xtp),inR(`ytp))>;
            Tree X = #<cmap(lambda(`w,call(plus,cmap(lambda(`wx,bag(tuple(apply(`kx,`wx),
                                                                          typed(tagged_union(0,`wx),`T)))),
                                                     apply(`f,`w)),
                                           cmap(lambda(`wy,bag(tuple(apply(`ky,`wy),
                                                                     typed(tagged_union(1,`wy),`T)))),
                                                apply(`g,`w)))),
                            `x)>;
            Tree sx = #<cmap(lambda(`vx,if(call(eq,union_tag(`vx),0),bag(typed(union_value(`vx),`xtp)),bag())),
                             nth(`v,1))>;
            Tree sy = #<cmap(lambda(`vy,if(call(eq,union_tag(`vy),1),bag(typed(union_value(`vy),`ytp)),bag())),
                             nth(`v,1))>;
            return normalize_term(#<cmap(lambda(`v,apply(`r,tuple(`sx,`sy))),
                                         groupBy(`X))>);
        case join(`kx,`ky,`r,`x,cmap(`g,`y)):
            if (!x.equals(y))
                fail;
            Tree v = new_var(), w = new_var(), wy = new_var(),
                vx = new_var(), vy = new_var();
            Tree xtp = TypeInference.type_inference(((Node)kx).children.head);
            Tree ytp = TypeInference.type_inference(((Node)ky).children.head);
            Tree T = #<union(inL(`xtp),inR(`ytp))>;
            Tree X = #<cmap(lambda(`w,call(plus,bag(tuple(apply(`kx,`w),typed(tagged_union(0,`w),`T))),
                                           cmap(lambda(`wy,bag(tuple(apply(`ky,`wy),
                                                                     typed(tagged_union(1,`wy),`T)))),
                                                apply(`g,`w)))),
                            `x)>;
            Tree sx = #<cmap(lambda(`vx,if(call(eq,union_tag(`vx),0),bag(typed(union_value(`vx),`xtp)),bag())),
                             nth(`v,1))>;
            Tree sy = #<cmap(lambda(`vy,if(call(eq,union_tag(`vy),1),bag(typed(union_value(`vy),`ytp)),bag())),
                             nth(`v,1))>;
            return normalize_term(#<cmap(lambda(`v,apply(`r,tuple(`sx,`sy))),
                                         groupBy(`X))>);
        case join(`kx,`ky,`r,cmap(`f,`x),`y):
            if (!x.equals(y))
                fail;
            Tree v = new_var(), w = new_var(), wx = new_var(),
                vx = new_var(), vy = new_var();
            Tree xtp = TypeInference.type_inference(((Node)kx).children.head);
            Tree ytp = TypeInference.type_inference(((Node)ky).children.head);
            Tree T = #<union(inL(`xtp),inR(`ytp))>;
            Tree X = #<cmap(lambda(`w,call(plus,cmap(lambda(`wx,bag(tuple(apply(`kx,`wx),
                                                                          typed(tagged_union(0,`wx),`T)))),
                                                     apply(`f,`w)),
                                           bag(tuple(apply(`ky,`w),typed(tagged_union(1,`w),`T))))),
                            `x)>;
            Tree sx = #<cmap(lambda(`vx,if(call(eq,union_tag(`vx),0),bag(typed(union_value(`vx),`xtp)),bag())),
                             nth(`v,1))>;
            Tree sy = #<cmap(lambda(`vy,if(call(eq,union_tag(`vy),1),bag(typed(union_value(`vy),`ytp)),bag())),
                             nth(`v,1))>;
            return normalize_term(#<cmap(lambda(`v,apply(`r,tuple(`sx,`sy))),
                                         groupBy(`X))>);
        case join(`kx,`ky,`r,`x,`y):
            Tree v = new_var();
            Tree vx = new_var();
            Tree vy = new_var();
            Tree X = #<cmap(lambda(`vx,bag(tuple(apply(`kx,`vx),`vx))),`x)>;
            Tree Y = #<cmap(lambda(`vy,bag(tuple(apply(`ky,`vy),`vy))),`y)>;
            return normalize_term(#<cmap(lambda(`v,apply(`r,tuple(nth(nth(`v,1),0),
                                                                  nth(nth(`v,1),1)))),
                                         coGroup(`X,`Y))>);
        case `gb(`u):
            if (! #[groupBy,orderBy].member(#<`gb>) )
                fail;
            match u {
            case cmap(...):
                return #<`gb(`(normalize_term(u)))>;
            };
            Tree nv = new_var();
            return #<`gb(cmap(lambda(`nv,bag(`nv)),`(normalize_term(u))))>;
        case coGroup(`e1,`e2):
            match e1 {
            case cmap(...):
            case _:
                Tree v1 = new_var();
                e1 = #<cmap(lambda(`v1,bag(`v1)),`(normalize_term(e1)))>;
            };
            match e2 {
            case cmap(...):
            case _:
                Tree v2 = new_var();
                e2 = #<cmap(lambda(`v2,bag(`v2)),`(normalize_term(e2)))>;
            };
            return #<coGroup(`(normalize_term(e1)),`(normalize_term(e2)))>;
        case reduce(`m,`u):
            match u {
            case cmap(...):
                return #<reduce(`m,`(normalize_term(u)))>;
            };
            Tree nv = new_var();
            return #<reduce(`m,cmap(lambda(`nv,bag(`nv)),`(normalize_term(u))))>;
        case call(`f,`s):
            for ( Tree monoid: monoids )
                match monoid {
                case `aggr(`mtp,`plus,`zero,`unit):
                    if (#<`aggr>.equals(f)) {
                        match s {
                        case cmap(...):
                            return #<reduce(`aggr,`(normalize_term(s)))>;
                        };
                        Tree nv = new_var();
                        return #<reduce(`aggr,cmap(lambda(`nv,bag(`nv)),`(normalize_term(s))))>;
                    }
                };
            fail
        case `f(...as):
            Trees bs = #[ ];
            for ( Tree a: as )
                bs = bs.append(normalize_term(a));
            return #<`f(...bs)>;
        };
        return e;
    }

    /** Embed missing cmaps */
    static Tree embed_missing_cmaps ( Tree e ) {
        match e {
        case project(`x,`a):
            match TypeInference.type_inference(x) {
            case `T(`tp):
                if (!is_collection(T))
                    return #<project(`(embed_missing_cmaps(x)),`a)>;
                Tree v = new_var();
                type_env.insert(v.toString(),tp);
                return embed_missing_cmaps(#<cmap(lambda(`v,bag(project(`v,`a))),`x)>);
            };
            return #<project(`(embed_missing_cmaps(x)),`a)>;
        case nth(`x,`n):
            match TypeInference.type_inference(x) {
            case `T(`tp):
                if (!is_collection(T))
                    fail;
                Tree v = new_var();
                type_env.insert(v.toString(),tp);
                return embed_missing_cmaps(#<cmap(lambda(`v,bag(nth(`v,`n))),`x)>);
            };
            fail
        case `f(...as):
            Trees bs = #[ ];
            for ( Tree a: as )
                bs = bs.append(embed_missing_cmaps(a));
            return #<`f(...bs)>;
        };
        return e;
    }

    private static Tree subst_pattern ( Tree pattern, Tree src, Tree dst ) {
        match pattern {
        case tuple(...as):
            Tree res = dst;
            int i = 0;
            for ( Tree a: as )
                res = subst_pattern(a,#<nth(`src,`(i++))>,res);
            return res;
        };
        return subst_var(pattern,src,dst);
    }

    /** Simplify the term e using rewrite rules */
    static Tree simplify_term ( Tree e ) {
        match e {
        case cmap(`f,cmap(lambda(`v,`u),`x)):
            return simplify_term(#<cmap(lambda(`v,cmap(`f,`u)),`x)>);
        case cmap(lambda(`x,`b),bag(`a)):
            return simplify_term(subst_var(x,a,b));
        case cmap(lambda(`x,`b),bag()):
            return #<bag()>;
        case cmap(`f,if(`p,`e1,`e2)):
            return simplify_term(#<if(`p,cmap(`f,`e1),cmap(`f,`e2))>);
        case cmap(`f,call(plus,`x,`y)):
            return simplify_term(#<call(plus,cmap(`f,`x),cmap(`f,`y))>);
        case groupBy(bag()):
            return #<bag()>;
        case cmap(`f,join(`kx,`ky,lambda(`v,`r),`X,`Y)):
            return simplify_term(#<join(`kx,`ky,lambda(`v,cmap(`f,`r)),`X,`Y)>);
        case apply(lambda(`v,`b),`u):
            if (!v.is_variable())
                fail;
            return simplify_term(subst_var(v,u,b));
        case apply(lambda(`pattern,`b),`u):
            return simplify_term(subst_pattern(pattern,u,b));
        case call(elem,bag(`u)):
            return simplify_term(u);
        case nth(tuple(...al),`n):
            if (!n.is_long())
                fail;
            int i = (int)n.longValue();
            if (i >= 0 && i < al.length())
                return simplify_term(al.nth(i));
        case project(record(...bl),`a):
            for ( Tree b: bl )
                match b {
                case bind(`v,`u):
                    if (v.equals(a))
                        return simplify_term(u);
                };
        case nth(let(`v,`w,tuple(...al)),`n):
            if (!n.is_long())
                fail;
            int i = (int)n.longValue();
            if (i >= 0 && i < al.length())
                return #<let(`v,`w,`(simplify_term(al.nth(i))))>;
        case let(`v,`w,`x):
            if (occurences(v,x) <= 1)
                return subst_var(v,w,x);
            fail
        case `f(...as):
            Trees bs = #[ ];
            for ( Tree a: as )
                bs = bs.append(simplify_term(a));
            return #<`f(...bs)>;
        };
        return e;
    }

    /** Simplify the term e using rewrite rules */
    public static Tree SimplifyTerm ( Tree e ) {
        Tree ne = simplify_term(e);
        if (e.equals(ne))
            return e;
        else return SimplifyTerm(ne);
    }

    /** Does this term contain a stream source? */
    public static boolean is_streaming ( Tree e ) {
        match e {
        case call(stream,...):
            return true;
        case `f(...al):
            for ( Tree a: al )
                if (is_streaming(a))
                    return true;
        };
        if (repeat_var(e))
            return true;
        else return false;
    }

    /** Collect all stream sources */
    private static Trees stream_bindings ( Tree e ) {
        match e {
        case call(stream,...):
            return #[`e];
        case `f(...al):
            if (#[BinaryStream,ParsedStream,SocketStream].member(#<`f>))
                return #[`e];
            Trees rs = #[];
            for ( Tree a: al )
                for ( Tree b: stream_bindings(a) )
                    if (!rs.member(b))
                        rs = rs.cons(b);
            return rs;
        };
        if (false && repeat_var(e))
            return #[`e];
        else return #[];
    }

    /** Convert an algebraic expression into a stream-based by pulling out all the stream sources
     * @param e algebraic expression
     * @return a stream-based expression
     */
    public static Tree streamify ( Tree e ) {
        Trees sbs = stream_bindings(e);
        Tree ne = e;
        for ( Tree sb: sbs ) {
            Tree v = new_var();
            TypeInference.type_env.insert(v.toString(),TypeInference.type_inference(sb));
            ne = #<Stream(lambda(`v,`(subst(sb,v,ne))),`sb)>;
        };
        return ne;
    }

    public static Tree unstreamify ( Tree e ) {
        match e {
        case call(stream,...r):
            return #<call(source,...r)>;
        case BinaryStream(...r):
            return #<BinarySource(...r)>;
        case ParsedStream(...r):
            return #<ParsedSource(...r)>;
        case `f(...al):
            Trees bs = #[ ];
            for ( Tree a: al )
                bs = bs.append(unstreamify(a));
            return #<`f(...bs)>;
        };
        return e;
    }

    /** Return the answer function of a query e
     * @param e a query
     * @param var the input of the answer function
     * @param merge the monoid associated with the input
     * @return the answer function
     */
    static Tree answer ( Tree e, Tree var, Tree merge ) {
        match e {
        case cmap(`f,call(stream,...s)):
            return var;
        case cmap(`f,`v):
            if (!repeat_var(v))
                fail;
            return var;
        case cmap(lambda(`v,`b),`u):
            match merge {
            case `gb(`m):
                if (! #[groupBy,orderBy].member(#<`gb>) )
                    fail;
                Tree nv = new_var();
                return #<reduce(`gb(`m),
                                cmap(lambda(`nv,bag(tuple(nth(nth(`nv,0),0),nth(`nv,1)))),
                                     `var))>;
            }
        };
        return var;
    }

    /** Remove the state from the query output */
    private static Tree strip_state ( Tree e, Tree var ) {
        match e {
        case reduce(`m,call(stream,...s)):
            return var;
        case reduce(`m,`v):
            if (!repeat_var(v))
                fail;
            return var;
        case reduce(`m,`u):
            return var;
        case cmap(`f,`u):
            Tree nn = new_var();
            return #<cmap(lambda(`nn,bag(nth(`nn,1))),`var)>;
        };
        return var;
    }

    /** Convert a stream-based query to an incremental stream-based program.
     *    It returns the quadruple (zero,merge,homomorphism,answer),
     *    where (zero,merge) is the monoid for the homomorphism
     *    and answer(homomorphism) is equal to the original query
     * @param e a stream-based query
     * @param env binds variables to monoids
     * @return a quadruple that represents the incremental stream-based program
     */
    static Tree generate_code ( Tree e, Environment env ) {
        if (Config.trace) {
            System.out.println("Subterm:");
            System.out.println(e.pretty(0));
        };
        Tree ne = SimplifyTerm(normalize_term(e));
        TypeInference.type_inference(ne);
        ne = SimplifyTerm(embed_missing_cmaps(ne));
        if (Config.trace) {
            System.out.println("Subterm normalized:");
            System.out.println(ne.pretty(0));
        };
        Tree qe = SimplifyTerm(inject_q(ne));
        if (Config.trace) {
            System.out.println("After lineage injection:");
            System.out.println(qe.pretty(0));
        };
        Tree nv = new_var();
        match split(qe,nv,env) {
        case pair(`a,`h):
            if (Config.trace) {
                System.out.println("After split:");
                System.out.println(a.pretty(0)+"\n"+h.pretty(0));
            };
            h = SimplifyTerm(h);
            if (Config.trace) {
                System.out.println("After split (simplified):");
                System.out.println(SimplifyTerm(a).pretty(0)+"\n"+h.pretty(0));
            };
            Tree m = get_monoid(h,env);
            if (Config.trace)
                System.out.println("Subterm monoid: "+m.pretty(0));
            Tree q = convert_to_algebra(h);
            Tree tp = TypeInference.type_inference(q);
            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>,true)).pretty(0));
            };
            type_env.insert(nv.toString(),tp);
            Tree answer = answer(ne,nv,m);
            answer = subst(nv,answer,a);
            answer = strip_state(ne,answer);
            answer = SimplifyTerm(convert_to_algebra(SimplifyTerm(answer)));
            if (Config.trace) {
                System.out.println("Answer function:");
                System.out.println(answer.pretty(0));
            };
            Tree zero = zero(m);
            Tree my = new_var();
            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));
            };
            return res;
        };
        throw new Error("Cannot generate incremental code: "+e);
    }

    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),fixed)>;
        match e {
        case call(stream,...):
            Tree v = new_var();
            return generate_code(#<cmap(lambda(`v,bag(`v)),`e)>,env);
        case call(`f,...qs):
            boolean is_monoid = false;
            for ( Tree monoid: monoids )
                match monoid {
                case `aggr(`mtp,`plus,`zero,`unit):
                    if (#<`aggr>.equals(f))
                        is_monoid = true;
                };
            if (is_monoid)
                fail;
            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),`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);
                    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)),
                           tuple(...ns))>;
        case record(...rs):
            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),`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);
                        ns = ns.append(n); i++;
                    }
                };
            return #<tuple(tuple(...zs),
                           lambda(tuple(tuple(...vs),tuple(...ws)),tuple(...ms)),
                           tuple(...hs),
                           lambda(tuple(...vs),record(...as)),
                           tuple(...ns))>;
        case tuple(...qs):
            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),`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);
                    ns = ns.append(n); i++;
                };
            return #<tuple(tuple(...zs),
                           lambda(tuple(tuple(...vs),tuple(...ws)),tuple(...ms)),
                           tuple(...hs),
                           lambda(tuple(...vs),tuple(...as)),
                           tuple(...ns))>;
        };
        return generate_code(e,env);
    }

    /** bind the pattern variables to types */
    private static void bind_pattern2type ( Tree pattern, Tree tp ) {
        match pattern {
        case tuple(...pl):
            match tp {
            case tuple(...tl):
                int i = 0;
                for ( Tree p: pl )
                    bind_pattern2type(p,tl.nth(i++));
                return;
            }
        };
        TypeInference.type_env.insert(pattern.toString(),tp);
    }

    /** Convert a stream-based query to an incremental stream-based program.
     * @param e a stream-based query
     * @return the incremental stream-based program
     */
    public static Tree generate_incremental_code ( Tree e ) {
        match e {
        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),`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));
                TypeInference.type_env.insert(w.toString(),TypeInference.type_inference(s));
                Tree nm = subst(v1,deltaT,subst(v2,h,m));
                Tree m0 = subst(v,x,h);
                Tree q = #<repeat(lambda(`deltaT,cmap(lambda(`w,bag(tuple(`w,true))),`nm)),
                                  `m0,
                                  `(nn-1))>;
                Tree res = #<incr(tuple(`z,`(unstreamify(x))),
                                  lambda(tuple(`s,`v),tuple(apply(lambda(tuple(`v1,`v2),`m),tuple(`s,`q)),`a)),
                                  lambda(tuple(`s,`v),`v))>;
                res = SimplifyTerm(res);
                if (Config.trace) {
                    System.out.println("Incremental program:");
                    System.out.println(res.pretty(0));
                };
                TypeInference.type_inference(res);
                return res;
            }
        };
    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)>);
            a = SimplifyTerm(a);
            Tree res = #<incr(`z,lambda(`v1,`nm),lambda(`v1,`a))>;
            if (Config.trace) {
                System.out.println("Incremental program:");
                System.out.println(res.pretty(0));
            };
            TypeInference.type_inference(res);
            return res;
        };
        throw new Error("Cannot generate incremental code: "+e);
    }
}