You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrql.apache.org by fe...@apache.org on 2015/11/06 21:23:33 UTC
[2/2] incubator-mrql git commit: [MRQL-79] Add support for
incremental query processing
[MRQL-79] Add support for incremental query processing
Project: http://git-wip-us.apache.org/repos/asf/incubator-mrql/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-mrql/commit/1cf73e2c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-mrql/tree/1cf73e2c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-mrql/diff/1cf73e2c
Branch: refs/heads/master
Commit: 1cf73e2c8aa332a536548221aacc878e93174189
Parents: a5330f2
Author: fegaras <fe...@cse.uta.edu>
Authored: Tue Nov 3 11:03:35 2015 -0600
Committer: fegaras <fe...@cse.uta.edu>
Committed: Tue Nov 3 11:03:35 2015 -0600
----------------------------------------------------------------------
conf/mrql-env.sh | 4 +-
core/src/main/java/org/apache/mrql/Compiler.gen | 5 +
core/src/main/java/org/apache/mrql/Config.java | 12 +-
.../main/java/org/apache/mrql/Environment.java | 16 +
.../main/java/org/apache/mrql/Evaluator.java | 2 +-
.../main/java/org/apache/mrql/Interpreter.gen | 23 +-
.../java/org/apache/mrql/MapReduceAlgebra.java | 13 +
.../java/org/apache/mrql/Materialization.gen | 3 +
core/src/main/java/org/apache/mrql/Plan.java | 4 +-
.../java/org/apache/mrql/PlanGeneration.gen | 47 +-
core/src/main/java/org/apache/mrql/Printer.gen | 10 +
.../src/main/java/org/apache/mrql/Streaming.gen | 1215 +++++++++++++++++-
.../java/org/apache/mrql/SystemFunctions.java | 41 +
core/src/main/java/org/apache/mrql/TopLevel.gen | 56 +-
.../main/java/org/apache/mrql/Translator.gen | 2 +-
.../main/java/org/apache/mrql/TypeInference.gen | 55 +-
core/src/main/java/org/apache/mrql/mrql.cgen | 7 +-
core/src/main/java/org/apache/mrql/mrql.lex | 1 +
.../java/org/apache/mrql/FlinkEvaluator.gen | 2 +-
.../java/org/apache/mrql/FlinkStreaming.gen | 4 +-
.../java/org/apache/mrql/GroupByJoinPlan.java | 11 +-
.../org/apache/mrql/MapReduceOperation.java | 12 +-
pom.xml | 2 +-
queries/RMAT.mrql | 6 +-
queries/incremental-kmeans.mrql | 33 +
queries/incremental-pagerank.mrql | 40 +
queries/incremental-streaming.mrql | 25 +
queries/streaming2.mrql | 22 -
.../java/org/apache/mrql/SparkEvaluator.gen | 58 +-
.../java/org/apache/mrql/SparkStreaming.gen | 90 +-
30 files changed, 1693 insertions(+), 128 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/conf/mrql-env.sh
----------------------------------------------------------------------
diff --git a/conf/mrql-env.sh b/conf/mrql-env.sh
index 57505c5..cb5b4be 100644
--- a/conf/mrql-env.sh
+++ b/conf/mrql-env.sh
@@ -77,7 +77,7 @@ BSP_SPLIT_INPUT=
# 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.2.0-bin-hadoop2.3
+SPARK_HOME=${HOME}/spark-1.3.0-bin-hadoop2.4
# 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"
@@ -96,7 +96,7 @@ SPARK_WORKER_MEMORY=1G
# Optional: Flink configuration. Supports version 0.6-incubating
# Note: for yarn, set yarn.nodemanager.vmem-check-enabled to false in yarn-site.xml
-FLINK_VERSION=0.9.0
+FLINK_VERSION=0.9.1
# Flink installation directory
FLINK_HOME=${HOME}/flink-${FLINK_VERSION}
# number of slots per TaskManager (typically, the number of CPUs per machine)
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Compiler.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Compiler.gen b/core/src/main/java/org/apache/mrql/Compiler.gen
index 8dec3d6..531205a 100644
--- a/core/src/main/java/org/apache/mrql/Compiler.gen
+++ b/core/src/main/java/org/apache/mrql/Compiler.gen
@@ -405,6 +405,8 @@ final public class Compiler extends Translator {
+",(Bag)"+compileE(s)+")";
case map(`m,`s):
return "MapReduceAlgebra.map("+compileF(m)+",(Bag)"+compileE(s)+")";
+ case fold(`acc,`zero,`s):
+ return "MapReduceAlgebra.fold("+compileF(acc)+","+compileE(zero)+",(Bag)"+compileE(s)+")";
case range(`min,`max):
return "MapReduceAlgebra.generator(((MR_long)"+compileE(min)+").get(),"
+"((MR_long)"+compileE(max)+").get())";
@@ -621,6 +623,9 @@ final public class Compiler extends Translator {
return "MapReduceAlgebra.BSP("+((LongLeaf)n).value()+","
+compileF(superstep)+","+compileE(state)+","+o+","
+"new Bag[]{"+ds.substring(1)+"})";
+ case DataSetCollect(`s):
+ return "(new Bag(((MR_dataset)(Interpreter.lookup_global_binding(\""+s.toString()
+ +"\"))).dataset().take(Integer.MAX_VALUE)))";
case `v:
if (v.is_variable())
return v.toString();
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Config.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Config.java b/core/src/main/java/org/apache/mrql/Config.java
index 7e1d291..53db404 100644
--- a/core/src/main/java/org/apache/mrql/Config.java
+++ b/core/src/main/java/org/apache/mrql/Config.java
@@ -63,7 +63,7 @@ final public class Config {
// max number of incoming messages before a sub-sync()
public static int bsp_msg_size = Integer.MAX_VALUE;
// number of elements per mapper to process the range min...max
- public static long range_split_size = 100000;
+ public static long range_split_size = -1;
// max number of streams to merge simultaneously
public static int max_merged_streams = 100;
// the directory for temporary files and spilled bags
@@ -86,6 +86,9 @@ final public class Config {
public static boolean info = false;
// for streaming, stream_window > 0 is the stream window duration in milliseconds
public static int stream_window = 0;
+ public static int stream_tries = 100;
+ // if true and stream_window > 0, then incremental streaming
+ public static boolean incremental = false;
/** store the configuration parameters */
public static void write ( Configuration conf ) {
@@ -117,6 +120,7 @@ final public class Config {
conf.setBoolean("mrql.testing",testing);
conf.setBoolean("mrql.info",info);
conf.setInt("mrql.stream.window",stream_window);
+ conf.setBoolean("mrql.incremental",incremental);
}
/** load the configuration parameters */
@@ -149,6 +153,7 @@ final public class Config {
testing = conf.getBoolean("mrql.testing",testing);
info = conf.getBoolean("mrql.info",info);
stream_window = conf.getInt("mrql.stream.window",stream_window);
+ incremental = conf.getBoolean("mrql.incremental",incremental);
}
public static ArrayList<String> extra_args = new ArrayList<String>();
@@ -267,6 +272,11 @@ final public class Config {
throw new Error("Expected a stream window duration");
stream_window = Integer.parseInt(args[i]);
i++;
+ } else if (args[i].equals("-stream_tries")) {
+ if (++i >= args.length)
+ throw new Error("Expected a stream window tries");
+ stream_tries = Integer.parseInt(args[i]);
+ i++;
} else if (args[i].charAt(0) == '-')
throw new Error("Unknown MRQL parameter: "+args[i]);
else {
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Environment.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Environment.java b/core/src/main/java/org/apache/mrql/Environment.java
index da8dd84..d848e7c 100644
--- a/core/src/main/java/org/apache/mrql/Environment.java
+++ b/core/src/main/java/org/apache/mrql/Environment.java
@@ -33,6 +33,22 @@ final public class Environment implements Serializable {
this.next = next;
}
+ public void replace ( String n, MRData v ) {
+ for ( Environment e = this; e != null; e = e.next )
+ if (e.name.equals(n)) {
+ e.value = v;
+ return;
+ };
+ throw new Error("Cannot find the name "+n+" in the environment "+this);
+ }
+
+ public String toString () {
+ String s = "[";
+ for ( Environment e = this; e != null; e = e.next )
+ s += " " + e.name + ": " + e.value;
+ return s+" ]";
+ }
+
private void writeObject(ObjectOutputStream out) throws IOException {
out.defaultWriteObject();
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Evaluator.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Evaluator.java b/core/src/main/java/org/apache/mrql/Evaluator.java
index 39333ae..76152d4 100644
--- a/core/src/main/java/org/apache/mrql/Evaluator.java
+++ b/core/src/main/java/org/apache/mrql/Evaluator.java
@@ -156,7 +156,7 @@ abstract public class Evaluator extends Interpreter {
}
/** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
- public void streaming ( Tree plan, Environment env, DataSetFunction f ) {
+ public void streaming ( Tree plan, Environment env, Function f ) {
throw new Error("MRQL Streaming is not supported in this evaluation mode yet");
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Interpreter.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Interpreter.gen b/core/src/main/java/org/apache/mrql/Interpreter.gen
index 5714a54..6627a12 100644
--- a/core/src/main/java/org/apache/mrql/Interpreter.gen
+++ b/core/src/main/java/org/apache/mrql/Interpreter.gen
@@ -326,7 +326,14 @@ public class Interpreter extends TypeInference {
new_distributed_binding(nm,s);
return evalE(body,new Environment(nm,s,env));
}; };
- return MapReduceAlgebra.repeat(loop_fnc,(Bag)evalE(s,env),((MR_int)evalE(n,env)).get());
+ MRData init = evalE(s,env);
+ if (init instanceof MR_dataset) {
+ Bag vs = new Bag();
+ for ( MRData dv: ((MR_dataset)init).dataset().take(Integer.MAX_VALUE))
+ vs.add(dv);
+ init = vs;
+ };
+ return MapReduceAlgebra.repeat(loop_fnc,(Bag)init,((MR_int)evalE(n,env)).get());
case closure(lambda(`v,`b),`s,`n):
final String nm = v.toString();
final Tree body = b;
@@ -406,6 +413,9 @@ public class Interpreter extends TypeInference {
b.materialize();
return b;
} catch (Exception ex) { throw new Error(ex); }
+ case DataSetCollect(`s):
+ DataSet ds = Evaluator.evaluator.eval(s,env,"-");
+ return new Bag(ds.take(Integer.MAX_VALUE));
case dataset_size(`x):
return new MR_long(Plan.size(Evaluator.evaluator.eval(x,env,"-")) / (1024*1024));
case synchronize(`peer,`b):
@@ -577,9 +587,10 @@ public class Interpreter extends TypeInference {
return ((Lambda)fnc).lambda().eval(t);
case Stream(...): // streaming
final Tree qt = query_type;
- Evaluator.evaluator.streaming(e,env,new DataSetFunction(){
- public void eval ( final DataSet ds ) {
- System.out.println(print(new MR_dataset(ds),qt));
+ Evaluator.evaluator.streaming(e,env,new Function(){
+ public MRData eval ( final MRData data ) {
+ System.out.println(print(data,qt));
+ return data;
}
});
return new Bag();
@@ -857,6 +868,8 @@ public class Interpreter extends TypeInference {
Tree pt = type_inference(ne);
if (Config.trace)
System.out.println("Algebraic type: "+print_type(pt));
+ if (Config.stream_window > 0 && Config.incremental)
+ ne = Streaming.generate_incremental_code(ne);
ne = AlgebraicOptimization.translate_all(ne);
if (Config.trace)
System.out.println("Translated expression:\n"+ne.pretty(0));
@@ -866,7 +879,7 @@ public class Interpreter extends TypeInference {
System.out.println("Physical plan type: "+print_type(et));
repeat_variables = #[];
ne = Simplification.simplify_all(ne);
- if (Streaming.is_streaming(ne))
+ if (Streaming.is_streaming(ne) && !Config.incremental)
ne = Streaming.streamify(ne);
Tree plan = PlanGeneration.makePlan(ne);
if (Config.bsp_mode) {
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java b/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
index 68eccda..d323fbb 100644
--- a/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
+++ b/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
@@ -103,6 +103,19 @@ final public class MapReduceAlgebra {
});
}
+ /** general reduction using an accumulator function and a zero element
+ * @param acc a function from (a,b) to b
+ * @param zero a value of type b
+ * @param s the input of type {a}
+ * @return a value of type b
+ */
+ public static MRData fold ( final Function acc, final MRData zero, final Bag s ) {
+ MRData result = zero;
+ for ( MRData value: s )
+ result = acc.eval(new Tuple(value,result));
+ return result;
+ }
+
/** strict group-by
* @param s the input of type {(a,b)}
* @return a value of type {(a,{b})}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Materialization.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Materialization.gen b/core/src/main/java/org/apache/mrql/Materialization.gen
index 5683699..4869d2e 100644
--- a/core/src/main/java/org/apache/mrql/Materialization.gen
+++ b/core/src/main/java/org/apache/mrql/Materialization.gen
@@ -89,6 +89,9 @@ final public class Materialization extends Translator {
return new_domain(vars,s,d);
case orderBy(`s):
return new_domain(vars,s,d);
+ case mapReduce(lambda(`mv,`m),lambda(`v,`b),`s,`o):
+ return materialize(#[`v],b,materialize(#[`mv],m,
+ new_domain(vars,s,d)));
case mapReduce2(lambda(`v1,`b1),lambda(`v2,`b2),lambda(`v,`b),`x,`y,`o):
return materialize(#[`v],b,materialize(#[`v1],b1,materialize(#[`v2],b2,
new_domain(vars,x,new_domain(vars,y,d)))));
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Plan.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Plan.java b/core/src/main/java/org/apache/mrql/Plan.java
index 94fd011..94ee9b9 100644
--- a/core/src/main/java/org/apache/mrql/Plan.java
+++ b/core/src/main/java/org/apache/mrql/Plan.java
@@ -163,9 +163,7 @@ public class Plan {
if (min > max)
throw new Error("Wrong range: "+min+"..."+max);
if (split_length < 1)
- if (Config.bsp_mode)
- split_length = (max-min)/Config.nodes+1;
- else split_length = Config.range_split_size;
+ split_length = (max-min)/Config.nodes+1;
DataSet ds = new DataSet(0,0);
long i = min;
while (i+split_length <= max) {
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/PlanGeneration.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/PlanGeneration.gen b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
index c127b39..e3901a8 100644
--- a/core/src/main/java/org/apache/mrql/PlanGeneration.gen
+++ b/core/src/main/java/org/apache/mrql/PlanGeneration.gen
@@ -222,6 +222,25 @@ final public class PlanGeneration extends AlgebraicOptimization {
return res;
}
+ private static Tree get_nested_query ( Tree var, Tree e ) {
+ match e {
+ case MapReduce(lambda(`v,`b),`r,`s,...):
+ if (!repeat_variables.member(s))
+ fail;
+ if (!free_variables(b,#[`var,`v]).is_empty())
+ fail;
+ return e;
+ case `f(...r):
+ for ( Tree a: r )
+ match get_nested_query(var,a) {
+ case none: ;
+ case `c:
+ return c;
+ }
+ };
+ return #<none>;
+ }
+
/** compile an algebraic form to a algebraic plan
* @param e the algebraic form
* @return the algebraic plan
@@ -230,7 +249,7 @@ final public class PlanGeneration extends AlgebraicOptimization {
match e {
// combine groupBy with Join into a groupByJoin (generalized matrix multiplication)
case mapReduce(lambda(`vm,bag(`bm)),lambda(`vr,bag(`br)),`s,`o):
- if (!bm.equals(vm))
+ if (!bm.equals(vm) || !is_dataset_expr(s))
fail;
match s {
case mapReduce2(lambda(`mvx,bag(tuple(`jx,`mx))),
@@ -289,6 +308,24 @@ final public class PlanGeneration extends AlgebraicOptimization {
`(makePlan(Y)),`o)>;
};
fail
+ case mapReduce(lambda(`vm,`bm),`nr,`s,`o):
+ if (!is_dataset_expr(s))
+ fail;
+ Tree pm = makePlan(bm);
+ match get_nested_query(vm,pm) {
+ case MapReduce(`km,`kr,`ks,`ko):
+ Tree nv = new_var();
+ pm = subst(#<MapReduce(`km,`kr,`ks,`ko)>,
+ #<mapReduce(`km,`kr,`nv,`ko)>,
+ pm); // must be done in memory
+ TypeInference.global_type_env.insert(nv.toString(),TypeInference.type_inference(ks));
+ return #<let(`nv,DataSetCollect(`ks),
+ MapReduce(lambda(`vm,`pm),
+ `(makePlan(nr)),
+ `(makePlan(s)),
+ `o))>;
+ };
+ fail
// extract the mapReduce combiner
case mapReduce(lambda(`vm,`bm),lambda(`vr,`br),`s,`o):
if (!Config.use_combiner || !is_dataset_expr(s))
@@ -336,7 +373,7 @@ final public class PlanGeneration extends AlgebraicOptimization {
`(makePlan(s)),`o)>;
else fail
case mapReduce2(`mx,`my,`r,`x,`y,`o):
- if (is_dataset_expr(x) && is_dataset_expr(y) && streamed_MapReduce2_reducer(r))
+ if (is_dataset_expr(x) && is_dataset_expr(y) && streamed_MapReduce2_reducer(r))
return #<MapReduce2(`(makePlan(mx)),
`(makePlan(my)),
`(makePlan(r)),
@@ -429,8 +466,10 @@ final public class PlanGeneration extends AlgebraicOptimization {
repeat_variables = repeat_variables.cons(v);
return #<Closure(lambda(`v,`(makePlan(b))),`(makePlan(s)),`(Integer.MAX_VALUE))>;
case loop(lambda(tuple(...vs),`b),tuple(...s),`n):
- if (!is_dataset_expr(s.nth(0)))
- fail;
+ int i = 0;
+ for ( Tree v: vs )
+ if (!is_dataset_expr(s.nth(i)))
+ fail;
repeat_variables = repeat_variables.append(vs);
return #<Loop(lambda(tuple(...vs),`(makePlan(b))),`(makePlan(#<tuple(...s)>)),`(makePlan(n)))>;
case record(...bl):
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Printer.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Printer.gen b/core/src/main/java/org/apache/mrql/Printer.gen
index 66cbec6..6942a2f 100644
--- a/core/src/main/java/org/apache/mrql/Printer.gen
+++ b/core/src/main/java/org/apache/mrql/Printer.gen
@@ -360,6 +360,16 @@ public class Printer {
else fail
case ParsedSource(`parser,`file,...args):
return "Source ("+parser+"): "+file;
+ case BinaryStream(`k,`file,_):
+ return "Stream (binary): "+file;
+ case BinaryStream(`file,_):
+ return "Stream (binary): "+file;
+ case ParsedStream(`m,`parser,`file,...args):
+ if (m instanceof LongLeaf)
+ return "Stream ("+parser+"): "+file;
+ else fail
+ case ParsedStream(`parser,`file,...args):
+ return "Stream ("+parser+"): "+file;
case Generator(...):
return "Generator";
case Merge(`x,`y):
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Streaming.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Streaming.gen b/core/src/main/java/org/apache/mrql/Streaming.gen
index 973e323..01e754c 100644
--- a/core/src/main/java/org/apache/mrql/Streaming.gen
+++ b/core/src/main/java/org/apache/mrql/Streaming.gen
@@ -19,43 +19,1226 @@ package org.apache.mrql;
import org.apache.mrql.gen.*;
-
-/** Generate a physical plan from an algebraic expression */
+/** Generates code for streaming queries */
final public class Streaming extends AlgebraicOptimization {
+ // 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 if (my.equals(#<fixed>))
+ return #<groupBy(product(`mx,`my))>;
+ else return #<groupBy(product(fixed,`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);
+ };
+ fail
+ case project(`u,`a):
+ match inference(u,env) {
+ case fixed:
+ return #<fixed>;
+ };
+ fail
+ case bag(`u):
+ match inference(u,env) {
+ case fixed:
+ return #<fixed>;
+ };
+ fail
+ case call(stream,...):
+ return #<union>;
+ case call(source,...):
+ return #<fixed>;
+ case call(`f,`s):
+ for ( Tree monoid: monoids )
+ match monoid {
+ case `aggr(`mtp,`plus,`zero,`unit):
+ if (aggr.equals(f.toString()))
+ return #<monoid(`plus,`zero,`unit)>;
+ };
+ fail
+ 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>;
+ fail
+ case bind(_,`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 return find(e.toString(),env);
+ };
+ return #<none>;
+ }
+
+ public static Tree inference ( Tree e ) {
+ return inference(e,null);
+ }
+
+ private static Tree cmap1 ( 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 cmap2 ( Tree f, Tree e ) {
+ Tree nv = new_var();
+ Tree nw = new_var();
+ return #<cmap(lambda(`nv,cmap(lambda(`nw,bag(tuple(tuple(nth(`nw,0),nth(`nv,0)),nth(`nw,1)))),
+ apply(`f,tuple(nth(nth(`nv,0),0),nth(`nv,1))))),
+ `e)>;
+ }
+
+ private static Tree cmap3 ( Tree f, Tree e ) {
+ Tree nv = new_var();
+ Tree nw = new_var();
+ return #<cmap(lambda(`nv,cmap(lambda(`nw,bag(tuple(tuple(nth(`nw,0),tuple()),nth(`nw,1)))),
+ apply(`f,`nv))),
+ `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 tuple(...as):
+ Trees bs = #[ ];
+ for ( Tree a: as )
+ bs = bs.append(inject_q(a));
+ return #<tuple(...bs)>;
+ case reduce(`m,cmap(`f,call(stream,...s))):
+ return e;
+ case reduce(`m,cmap(`f,`v)):
+ if (!repeat_var(v))
+ fail;
+ return e;
+ case reduce(`m,cmap(`f,`u)):
+ return #<reduce(groupBy(`m),`(cmap1(f,inject_e(u))))>;
+ case reduce(`m,call(stream,...s)):
+ return e;
+ case reduce(`m,`v):
+ if (!repeat_var(v))
+ fail;
+ return 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 cmap1(f,inject_e(u));
+ case `f(...as):
+ Trees bs = #[ ];
+ for ( Tree a: as )
+ bs = bs.append(inject_q(a));
+ return #<`f(...bs)>;
+ };
+ return 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(`(inject_c(c)))>;
+ case coGroup(`c1,`c2):
+ return #<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 cmap3(f,#<call(stream,...s)>);
+ case cmap(`f,`v):
+ if (!repeat_var(v))
+ fail;
+ return cmap3(f,v);
+ case cmap(`f,`u):
+ return cmap2(f,inject_e(u));
+ case call(stream,...s):
+ return cmap3(#<lambda(x,bag(x))>,#<call(stream,...s)>);
+ case `v:
+ if (!repeat_var(v))
+ fail;
+ return cmap3(#<lambda(x,bag(x))>,v);
+ };
+ 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 tuple(...as):
+ match merge {
+ case product(...ms):
+ Trees bs = #[ ];
+ int i = 0;
+ for ( Tree a: as )
+ bs = bs.append(answer(a,#<nth(`var,`i)>,ms.nth(i++)));
+ return #<tuple(...bs)>;
+ }
+ case record(...as):
+ match merge {
+ case record(...ms):
+ Trees bs = #[ ];
+ int i = 0;
+ for ( Tree a: as )
+ match a {
+ case bind(`v,`u):
+ match ms.nth(i++) {
+ case bind(_,`m):
+ bs = bs.append(#<bind(`v,`(answer(u,#<project(`var,`v)>,m)))>);
+ }
+ }
+ return #<record(...bs)>;
+ }
+ case reduce(`m,cmap(`f,`u)):
+ match merge {
+ case `gb(_):
+ if (! #[groupBy,orderBy].member(#<`gb>) )
+ fail;
+ Tree nv = new_var();
+ return #<reduce(`m,cmap(lambda(`nv,bag(nth(`nv,1))),`var))>;
+ case _: // total aggregation
+ return var;
+ }
+ case reduce(`m,call(stream,...s)):
+ return var;
+ case reduce(`m,`v):
+ if (!repeat_var(v))
+ fail;
+ return var;
+ case cmap(`f,call(stream,...s)):
+ Tree nv = new_var();
+ return #<cmap(lambda(`nv,bag(nth(`nv,1))),`var)>;
+ case cmap(`f,`v):
+ if (!repeat_var(v))
+ fail;
+ Tree nv = new_var();
+ return #<cmap(lambda(`nv,bag(nth(`nv,1))),`var)>;
+ case cmap(`f,`u):
+ match merge {
+ case `gb(`m):
+ if (! #[groupBy,orderBy].member(#<`gb>) )
+ fail;
+ Tree nv = new_var();
+ match TypeInference.type_inference(var) {
+ case `T(tuple(tuple(_,tuple()),_)):
+ // only one key => don't group-by
+ return #<cmap(lambda(`nv,bag(nth(`nv,1))),`var)>;
+ };
+ Tree nw = new_var();
+ Tree nn = new_var();
+ return #<cmap(lambda(`nv,cmap(lambda(`nn,bag(reduce(`m,`nn))),
+ nth(`nv,1))),
+ `gb(cmap(lambda(`nw,bag(tuple(nth(nth(`nw,0),0),nth(`nw,1)))),
+ `var)))>;
+ }
+ };
+ return var;
+ }
- static boolean is_streaming ( Tree 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].member(inference(e,env)))
+ return #[ `e ];
match e {
- case call(stream,...): return true;
+ 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 = bs.append(find_homomorphisms(a,env));
+ };
+ return bs;
+ case call(`f,...as):
+ Trees bs = #[ ];
+ for ( Tree a: as )
+ bs = bs.append(find_homomorphisms(a,env));
+ return bs;
+ case `f(...as):
+ Trees bs = #[ ];
+ for ( Tree a: as )
+ bs = bs.append(find_homomorphisms(a,env));
+ return bs;
+ };
+ return #[ ];
+ }
+
+ /** Split a cMap term e into a homomorhism and an answer function
+ * @param e a cMap term
+ * @param var the input variable of the answer function
+ * @param cmap_var the cmap variable
+ * @param env binds variables to monoids
+ * @return a pair (answer,homomorphism)
+ */
+ static Tree split ( Tree e, Tree var, Tree cmap_var, Environment env ) {
+ if (split_tracing) {
+ tab_count += 3;
+ System.out.println(Interpreter.tabs(tab_count)+var+" "+cmap_var+"\n"+e.pretty(0));
+
+ };
+ Tree res = split_trace(e,var,cmap_var,env);
+ if (split_tracing) {
+ System.out.println(Interpreter.tabs(tab_count)+"-> "+res.pretty(0));
+ tab_count -= 3;
+ };
+ return res;
+ }
+
+ private static Tree split_trace ( Tree e, Tree var, Tree cmap_var, Environment env ) {
+ match e {
+ case cmap(lambda(`v,bag(tuple(nth(`k,0),`b))),`u):
+ if (!cmap_var.equals(k))
+ fail;
+ match inference(u,env) {
+ case `gb(`m):
+ if (! #[groupBy,orderBy].member(#<`gb>) )
+ fail;
+ Trees hs = find_homomorphisms(b,new Environment(v.toString(),#<product(fixed,`m)>,env));
+ if (hs.length() == 1 && hs.nth(0).equals(b))
+ return #<pair(`var,`e)>;
+ Tree nv = new_var();
+ Tree ne = b;
+ int i = 0;
+ for ( Tree h: hs ) {
+ ne = subst(h,#<nth(`nv,`i)>,ne);
+ i++;
+ };
+ return #<pair(cmap(lambda(`nv,bag(`ne)),`var),
+ cmap(lambda(`v,bag(tuple(nth(`k,0),tuple(...hs)))),`u))>;
+ case fixed:
+ Trees hs = find_homomorphisms(b,new Environment(v.toString(),#<fixed>,env));
+ if (hs.length() == 1 && hs.nth(0).equals(b))
+ return #<pair(`var,`e)>;
+ Tree nv = new_var();
+ Tree ne = b;
+ int i = 0;
+ for ( Tree h: hs ) {
+ ne = subst(h,#<nth(`nv,`i)>,ne);
+ i++;
+ };
+ return #<pair(cmap(lambda(`nv,bag(`ne)),`var),
+ cmap(lambda(`v,bag(tuple(nth(`k,0),tuple(...hs)))),`u))>;
+ }
+ case cmap(lambda(`v,bag(tuple(`k,`b))),`u):
+ match inference(u,env) {
+ case `gb(`m):
+ if (! #[groupBy,orderBy].member(#<`gb>) )
+ fail;
+ Trees hs = find_homomorphisms(b,new Environment(v.toString(),#<product(fixed,`m)>,env));
+ if (hs.length() == 1 && hs.nth(0).equals(b))
+ return #<pair(`var,`e)>;
+ Tree nv = new_var();
+ Tree ne = b;
+ int i = 0;
+ for ( Tree h: hs ) {
+ ne = subst(h,#<nth(`nv,`i)>,ne);
+ i++;
+ };
+ return #<pair(cmap(lambda(`nv,bag(`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):
+ a = subst(nv,var,a);
+ return #<pair(`a,cmap(lambda(`v,`h),`u))>;
+ }
+ }
+ case fixed:
+ Tree nv = new_var();
+ match split(b,nv,new Environment(v.toString(),#<fixed>,env)) {
+ case pair(`a,`h):
+ a = subst(nv,var,a);
+ a = subst(cmap_var,nv,a);
+ return #<pair(`a,cmap(lambda(`v,`h),`u))>;
+ }
+ }
+ };
+ return #<fail>;
+ }
+
+ /** 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 ) {
+ match e {
+ case cmap(lambda(`v,`b),`u):
+ match split(e,var,v,env) {
+ case fail:
+ return #<pair(cmap(lambda(`v,`b),`var),`u)>;
+ case `t:
+ return t;
+ }
+ 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 reduce(`m,`u):
+ return #<pair(`var,`e)>;
+ 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 `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(`e,tuple())>;
+ }
+
+ /** 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 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(gm,#<nth(`v,1)>);
+ return #<cmap(lambda(`v,bag(tuple(nth(`v,0),`me))),`gb(`e))>;
+ };
+ match e {
+ // FIX: may have more cases
+ case cmap(...):
+ return #<call(`m,`e)>;
+ };
+ return e;
+ }
+
+ /** Convert coGroups back to joins and reduces to aggregations */
+ private 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);
+ 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;
+ }
+
+ /** 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 ) {
+ match m {
+ case `gb(`n):
+ // needs outer-join in function r
+ if (! #[groupBy,orderBy].member(#<`gb>) )
+ fail;
+ Tree v = new_var();
+ Tree vx = new_var();
+ Tree vy = new_var();
+ Tree mx = new_var();
+ Tree my = new_var();
+ match type {
+ case `T(tuple(`keytp,`tp)):
+ 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),
+ cmap(lambda(`vx,cmap(lambda(`vy,bag(tuple(nth(`vx,0),
+ `(merge(n,tp,#<nth(`vx,1)>,#<nth(`vy,1)>))))),
+ `my)),
+ `mx),
+ `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)>));
+ 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)>)))>);
+ }
+ 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.toString()))
+ 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.toString()))
+ return zero;
+ };
+ };
+ throw new Error("Undefined monoid: "+m);
+ }
+
+ /** Convert joins to coGroups, plus other transformations */
+ private static Tree normalize_term ( Tree e ) {
+ match e {
+ case join(`kx,`ky,`r,`x,`y):
+ match TypeInference.type_inference(x) {
+ case `T(`xtp):
+ match TypeInference.type_inference(x) {
+ case `S(`ytp):
+ Tree v = new_var();
+ Tree vx = new_var();
+ Tree vy = new_var();
+ type_env.insert(vx.toString(),xtp);
+ type_env.insert(vy.toString(),ytp);
+ type_env.insert(v.toString(),#<tuple(none,tuple(`T(`xtp),`S(`ytp)))>);
+ return normalize_term(#<cmap(lambda(`v,apply(`r,tuple(nth(nth(`v,1),0),
+ nth(nth(`v,1),1)))),
+ coGroup(cmap(lambda(`vx,bag(tuple(apply(`kx,`vx),`vx))),`x),
+ cmap(lambda(`vy,bag(tuple(apply(`ky,`vy),`vy))),`y)))>);
+ }
+ };
+ fail
+ 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 project(`x,`a):
+ match TypeInference.type_inference(x) {
+ case `T(`tp):
+ if (!is_collection(T))
+ fail;
+ Tree v = new_var();
+ type_env.insert(v.toString(),tp);
+ return normalize_term(#<cmap(lambda(`v,bag(project(`v,`a))),`x)>);
+ };
+ fail
+ 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 normalize_term(#<cmap(lambda(`v,bag(nth(`v,`n))),`x)>);
+ };
+ fail
+ case call(`f,`s):
+ for ( Tree monoid: monoids )
+ match monoid {
+ case `aggr(`mtp,`plus,`zero,`unit):
+ if (aggr.equals(f.toString())) {
+ 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;
+ }
+
+ /** 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 groupBy(bag()):
+ return #<bag()>;
+ case apply(lambda(`v,`b),`u):
+ if (!v.is_variable())
+ fail;
+ return simplify_term(subst_var(v,u,b));
+ 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 `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;
};
- return false;
+ if (repeat_var(e))
+ return true;
+ else return false;
}
- static Trees stream_bindings ( Tree e ) {
+ /** Collect all stream sources */
+ private static Trees stream_bindings ( Tree e ) {
match e {
case call(stream,...):
- return #[bind(`(new_var()),`e)];
+ return #[`e];
case `f(...al):
+ if (#[BinaryStream,ParsedStream,SocketStream].member(#<`f>))
+ return #[`e];
Trees rs = #[];
for ( Tree a: al )
- rs = rs.append(stream_bindings(a));
+ for ( Tree b: stream_bindings(a) )
+ if (!rs.member(b))
+ rs = rs.cons(b);
return rs;
};
- return #[];
+ if (repeat_var(e))
+ return #[`e];
+ else return #[];
}
- static Tree streamify ( Tree e ) {
+ /** 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 )
- match sb {
- case bind(`v,`u):
- ne = #<stream(lambda(`v,`(subst(u,v,ne))),`u)>;
- TypeInference.type_inference(ne);
- };
+ 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;
}
+
+ /** 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 (!is_streaming(e)) {
+ Tree x = new_var();
+ Tree s = new_var();
+ TypeInference.type_env.insert(x.toString(),#<tuple(tuple(),tuple())>);
+ TypeInference.type_env.insert(s.toString(),#<tuple()>);
+ return #<tuple(tuple(),lambda(`x,tuple()),tuple(),lambda(`s,`e))>;
+ }
+ Tree ne = SimplifyTerm(normalize_term(e));
+ if (Config.trace) {
+ System.out.println("After converting joins to coGroup and simplifying:");
+ 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,`q):
+ if (Config.trace) {
+ System.out.println("After split:");
+ System.out.println(a.pretty(0)+"\n"+q.pretty(0));
+ System.out.println("After split (simplified):");
+ System.out.println(SimplifyTerm(a).pretty(0)+"\n"+SimplifyTerm(q).pretty(0));
+ };
+ Tree m = inference(q,env);
+ if (Config.trace)
+ System.out.println("monoid: "+m.pretty(0));
+ q = convert_to_algebra(SimplifyTerm(q));
+ Tree tp = TypeInference.type_inference(q);
+ if (Config.trace) {
+ System.out.println("Query type: "+tp);
+ System.out.println("Merge function over X and Y: ");
+ System.out.println(SimplifyTerm(merge(m,tp,#<X>,#<Y>)).pretty(0));
+ };
+ type_env.insert(nv.toString(),tp);
+ Tree answer = answer(ne,nv,m);
+ answer = subst(nv,answer,a);
+ 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 mv = new_var();
+ Tree merge = convert_to_algebra(SimplifyTerm(merge(m,tp,#<nth(`mv,0)>,#<nth(`mv,1)>)));
+ Tree res = #<tuple(`zero,lambda(`mv,`merge),`q,lambda(`nv,`answer))>;
+ return res;
+ };
+ throw new Error("Cannot generate incremental code: "+e);
+ }
+
+ /** 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 (!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_code(u,null) {
+ case tuple(`z,`m,`h,lambda(`s,`a)):
+ match generate_code(x,null) {
+ case tuple(`xz,`xm,`xh,lambda(`xs,`xa)):
+ Tree state = new_var();
+ Tree w = new_var();
+ TypeInference.type_env.insert(state.toString(),TypeInference.type_inference(s));
+ TypeInference.type_env.insert(w.toString(),TypeInference.type_inference(s));
+ Tree hh = subst(v,xa,h);
+ h = subst(v,a,h);
+ Tree q = #<repeat(lambda(`s,cmap(lambda(`w,bag(tuple(`w,true))),
+ apply(`m,tuple(`state,`h)))),
+ apply(`m,tuple(`state,`hh)),
+ `(nn-1))>;
+ Tree xq = #<apply(`xm,tuple(`xs,`xh))>;
+ a = subst(s,state,a);
+ TypeInference.type_inference(a);
+ Tree res = #<tuple(tuple(`xz,`z),
+ lambda(tuple(`xs,`state),tuple(`xq,`q)),
+ lambda(tuple(`xs,`state),`a))>;
+ res = SimplifyTerm(res);
+ Tree tp = TypeInference.type_inference(res);
+ return res;
+ }
+ }
+ case repeat(lambda(`v,`u),`x,`n):
+ // when x is invariant (doesn't contain a stream source)
+ if (!(n instanceof LongLeaf))
+ throw new Error("wrong repeat: "+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_code(u,null) {
+ case tuple(`z,`m,`h,lambda(`s,`a)):
+ Tree state = new_var();
+ Tree w = new_var();
+ TypeInference.type_env.insert(state.toString(),TypeInference.type_inference(s));
+ TypeInference.type_env.insert(w.toString(),TypeInference.type_inference(s));
+ Tree hh = subst(v,x,h);
+ h = subst(v,a,h);
+ Tree q = #<repeat(lambda(`s,cmap(lambda(`w,bag(tuple(`w,true))),
+ apply(`m,tuple(`state,`h)))),
+ apply(`m,tuple(`state,`hh)),
+ `(nn-1))>;
+ a = subst(s,state,a);
+ TypeInference.type_inference(a);
+ Tree res = #<tuple(`z,
+ lambda(`state,`q),
+ lambda(`state,`a))>;
+ res = SimplifyTerm(res);
+ Tree tp = TypeInference.type_inference(res);
+ return res;
+ }
+ };
+ match generate_code(e,null) {
+ case tuple(`z,`m,`h,lambda(`v,`a)):
+ Tree q = SimplifyTerm(#<apply(`m,tuple(`v,`h))>);
+ Tree res = #<tuple(`z,lambda(`v,`q),lambda(`v,`a))>;
+ TypeInference.type_inference(res);
+ return res;
+ };
+ throw new Error("Cannot generate incremental code: "+e);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/SystemFunctions.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/SystemFunctions.java b/core/src/main/java/org/apache/mrql/SystemFunctions.java
index b7385fb..c19e344 100644
--- a/core/src/main/java/org/apache/mrql/SystemFunctions.java
+++ b/core/src/main/java/org/apache/mrql/SystemFunctions.java
@@ -380,6 +380,47 @@ final public class SystemFunctions {
return new MR_double(sum/count);
}
+ /** used in avg */
+ public static MRData avg_aggr ( Bag s ) {
+ double sum = 0.0;
+ long count = 0;
+ MRData e = null;
+ for ( MRData value: s ) {
+ Tuple t = (Tuple)value;
+ e = t.first();
+ if (t.first() instanceof MR_double)
+ sum += ((MR_double)t.first()).get();
+ else if (t.first() instanceof MR_int)
+ sum += ((MR_int)t.first()).get();
+ else if (t.first() instanceof MR_long)
+ sum += ((MR_long)t.first()).get();
+ else if (t.first() instanceof MR_float)
+ sum += ((MR_float)t.first()).get();
+ if (t.second() instanceof MR_long)
+ count += ((MR_long)t.second()).get();
+ else if (t.second() instanceof MR_int)
+ count += ((MR_int)t.second()).get();
+ };
+ if (e instanceof MR_double)
+ e = new MR_double(sum);
+ else if (e instanceof MR_int)
+ e = new MR_int((int)sum);
+ else if (e instanceof MR_long)
+ e = new MR_long((long)sum);
+ else if (e instanceof MR_float)
+ e = new MR_float((float)sum);
+ return new Tuple(e,new MR_long(count));
+ }
+
+ /** return the join key from a matching pair in the join reducer */
+ public static MRData join_key ( Bag xs, Bag ys ) {
+ for ( MRData x: xs )
+ return ((Tuple)x).get(0);
+ for ( MRData y: ys )
+ return ((Tuple)y).get(0);
+ return null;
+ }
+
public static MR_string text ( Union node ) {
if (node.tag() == 1)
return (MR_string)(node.value());
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/TopLevel.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/TopLevel.gen b/core/src/main/java/org/apache/mrql/TopLevel.gen
index 645cc86..7e4f38e 100644
--- a/core/src/main/java/org/apache/mrql/TopLevel.gen
+++ b/core/src/main/java/org/apache/mrql/TopLevel.gen
@@ -289,14 +289,15 @@ final public class TopLevel extends Interpreter {
Evaluator.evaluator.initialize_query();
final Tree qt = query_type;
final String file = s.stringValue();
- Evaluator.evaluator.streaming(plan,null,new DataSetFunction(){
+ Evaluator.evaluator.streaming(plan,null,new Function(){
long i = 0;
- public void eval ( final DataSet ds ) {
+ public MRData eval ( final MRData data ) {
try {
- Evaluator.evaluator.dump(file+"/f"+(i++),qt,new MR_dataset(ds));
+ Evaluator.evaluator.dump(file+"/f"+(i++),qt,data);
+ return data;
} catch (Exception ex) {
throw new Error("Cannot dump the streaming result to the directory "+file);
- };
+ }
}
});
case dump_text(`s,`e): // dump stream output
@@ -309,16 +310,57 @@ final public class TopLevel extends Interpreter {
Evaluator.evaluator.initialize_query();
final Tree qt = query_type;
final String file = s.stringValue();
- Evaluator.evaluator.streaming(plan,null,new DataSetFunction(){
+ Evaluator.evaluator.streaming(plan,null,new Function(){
long i = 0;
- public void eval ( final DataSet ds ) {
+ public MRData eval ( final MRData data ) {
try {
- Evaluator.evaluator.dump_text(file+"/f"+(i++),qt,new MR_dataset(ds));
+ Evaluator.evaluator.dump_text(file+"/f"+(i++),qt,data);
+ return data;
} catch (Exception ex) {
throw new Error("Cannot dump the streaming result to the directory "+file);
}
}
});
+ case incr(`e): // incremental stream processing
+ Config.incremental = true;
+ reset();
+ Tree plan = translate_expression(e);
+ Config.incremental = false;
+ if (plan == null)
+ return;
+ Evaluator.evaluator.initialize_query();
+ match plan {
+ case tuple(`zero,lambda(tuple(`state1,`state2),`merge),lambda(_,`answer)):
+ final Tree qt = make_persistent_type(query_type);
+ Tuple z = (Tuple)evalE(zero,null);
+ final Environment env =
+ new Environment(state1.toString(),z.get(0),
+ new Environment(state2.toString(),z.get(1),null));
+ final Tree ans = answer;
+ merge = Streaming.streamify(merge);
+ Evaluator.evaluator.streaming(#<lambda(tuple(`state1,`state2),`merge)>,env,
+ new Function(){
+ public MRData eval ( final MRData state ) {
+ MRData a = Evaluator.evaluator.evalE(ans,env);
+ System.out.println(print(a,qt));
+ return state;
+ }
+ });
+ case tuple(`zero,lambda(`state,`merge),lambda(_,`answer)):
+ final Tree qt = make_persistent_type(query_type);
+ MRData z = evalE(zero,null);
+ final Environment env = new Environment(state.toString(),z,null);
+ final Tree ans = answer;
+ merge = Streaming.streamify(merge);
+ Evaluator.evaluator.streaming(#<lambda(`state,`merge)>,env,
+ new Function(){
+ public MRData eval ( final MRData state ) {
+ MRData a = Evaluator.evaluator.evalE(ans,env);
+ System.out.println(print(a,qt));
+ return state;
+ }
+ });
+ };
case dump(`s,`e):
long t = System.currentTimeMillis();
dump(s.stringValue(),e);
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/Translator.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Translator.gen b/core/src/main/java/org/apache/mrql/Translator.gen
index 0a00eb1..3be0d9f 100644
--- a/core/src/main/java/org/apache/mrql/Translator.gen
+++ b/core/src/main/java/org/apache/mrql/Translator.gen
@@ -184,7 +184,7 @@ public class Translator extends Printer {
= #[mapReduce,mapReduce2,cmap,join,groupBy,orderBy,aggregate,map,filter,repeat,closure];
static Trees plan_names = plans_with_distributed_lambdas.append(algebraic_operators)
- .append(#[Repeat,Closure,Generator,Let,If]);
+ .append(#[Repeat,Closure,Generator,Let,If,Stream]);
/** generates new variable names */
public static Tree new_var () {
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/TypeInference.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/TypeInference.gen b/core/src/main/java/org/apache/mrql/TypeInference.gen
index d12084c..7fe483d 100644
--- a/core/src/main/java/org/apache/mrql/TypeInference.gen
+++ b/core/src/main/java/org/apache/mrql/TypeInference.gen
@@ -564,9 +564,18 @@ public class TypeInference extends Translator {
};
return maxt;
case lambda(`v,`b):
- if(!v.is_variable())
- fail;
- if (type_env.lookup(v.toString()) == null)
+ if (!v.is_variable()) {
+ Tree nv = new_var();
+ type_env.insert(nv.toString(),type_inference(v));
+ st.begin_scope();
+ if (!Normalization.bind_pattern(v,nv).is_empty())
+ error("Lambda patterns must be irrefutable: "+print_query(e));
+ for ( String nm: st )
+ if (st.is_local(nm))
+ b = Simplification.subst(#<`nm>,st.lookup(nm),b);
+ st.end_scope();
+ return type_inference(#<lambda(`nv,`b)>);
+ } else if (type_env.lookup(v.toString()) == null)
type_env.insert(v.toString(),#<any>);
return #<arrow(`(type_env.lookup(v.toString())),`(type_inference(b)))>;
case function(tuple(...params),`outp,`body):
@@ -622,6 +631,8 @@ public class TypeInference extends Translator {
((Node)e).children = ((Node)e).children.append(tp); // destructive
return tp;
case call(stream,`parser,`f,...args):
+ if (parser.equals(#<binary>))
+ fail;
if (Config.stream_window == 0)
type_error(e,"Not in stream processing mode");
if (!parser.is_variable())
@@ -644,9 +655,16 @@ public class TypeInference extends Translator {
if (Config.stream_window == 0)
type_error(e,"Not in stream processing mode");
return type_inference(#<call(source,...r)>);
+ case BinaryStream(`path,`tp):
+ return tp;
+ case ParsedStream(`parser,...r):
+ return type_inference(#<call(stream,`parser,...r)>);
case stream(lambda(`v,`b),`u):
bind_pattern_type(v,type_inference2(u));
return type_inference2(b);
+ case Stream(lambda(`v,`b),`u):
+ bind_pattern_type(v,type_inference2(u));
+ return type_inference2(b);
case typed(null,`tp):
return tp;
case typed(`f(...),`tp):
@@ -902,6 +920,25 @@ public class TypeInference extends Translator {
: #<bag(tuple(`k,list(`v)))>;
};
type_error(e,"Wrong groupBy: "+print_query(e)+" of type "+print_type(tp));
+ case coGroup(`x,`y):
+ Tree xtp = type_inference2(x);
+ Tree ytp = type_inference2(y);
+ match xtp {
+ case `T(tuple(`kx,`a)):
+ if (!is_collection(T))
+ type_error(e,"Wrong coGroup left operand: "+print_query(x));
+ match xtp {
+ case `S(tuple(`ky,`b)):
+ if (!is_collection(T))
+ type_error(e,"Wrong coGroup right operand: "+print_query(y));
+ if (unify(kx,ky) == null)
+ type_error(e,"incompatible keys in coGroup: "+print_query(e));
+ return (is_persistent_collection(T) && is_persistent_collection(S))
+ ? #<Bag(tuple(`kx,tuple(`T(`a),`S(`b))))>
+ : #<bag(tuple(`kx,tuple(`T(`a),`S(`b))))>;
+ }
+ };
+ type_error(e,"Wrong coGroup: "+print_query(e));
case orderBy(`u):
Tree tp = type_inference2(u);
match tp {
@@ -1253,6 +1290,18 @@ public class TypeInference extends Translator {
fail
case call(avg,`s):
return type_inference(#<call(div,typed(call(sum,`s),double),call(count,`s))>);
+ case call(join_key,`s1,`s2):
+ match type_inference2(s1) {
+ case `T1(tuple(`tp1,_)):
+ match type_inference2(s2) {
+ case `T2(tuple(`tp2,_)):
+ if (is_collection(T1) && is_collection(T2) && unify(tp1,tp2) != null)
+ return tp1;
+ }
+ };
+ type_error(e,"Expected two collections with the same key in join key: "+e);
+ case reduce(`m,`s):
+ return type_inference2(#<call(`m,`s)>);
case trace(`msg,`t,`x):
if (unify(type_inference2(msg),#<string>) == null)
type_error(e,"Expected a string in the trace message");
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/mrql.cgen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/mrql.cgen b/core/src/main/java/org/apache/mrql/mrql.cgen
index a723fbd..87245a0 100644
--- a/core/src/main/java/org/apache/mrql/mrql.cgen
+++ b/core/src/main/java/org/apache/mrql/mrql.cgen
@@ -34,7 +34,7 @@ parser code {:
sym.SOME, sym.ALL, sym.GTR, sym.SEP, sym.STORE, sym.DUMP, sym.TYPE, sym.DATA, sym.REPEAT,
sym.STEP, sym.LIMIT, sym.LET, sym.ATSYM, sym.EXCLAMATION,
sym.Variable, sym.Integer, sym.Double, sym.String, sym.Decimal,
- sym.START_TEMPLATE, sym.END_TEMPLATE, sym.TEXT, sym.TRACE
+ sym.START_TEMPLATE, sym.END_TEMPLATE, sym.TEXT, sym.TRACE, sym.INCR
};
static String[] token_names = {
@@ -49,7 +49,7 @@ parser code {:
"some", "all", ">", "|", "store", "dump", "type", "data", "repeat",
"step", "limit", "let", "@", "!",
"Variable", "Integer", "Double", "String", "Decimal",
- "[|", "|]", "Text", "trace"
+ "[|", "|]", "Text", "trace", "incr"
};
public static String print ( Symbol s ) {
@@ -98,7 +98,7 @@ terminal IF, THEN, ELSE, SELECT, FROM, HAVING, LB, RB, LP, RP, LSB, RSB, LDOT, S
UNION, INTERSECT, EXCEPT, EXISTS, IN, COMMA, DOT, COLON, ASSIGN, SEMI, WHERE,
ORDER, GROUP, BY, ASCENDING, DESCENDING, UMINUS, FUNCTION, DISTINCT, BSLASH,
SOME, ALL, GTR, SEP, STORE, TYPE, DATA, CASE, ATSYM, XPATH, REPEAT, STEP, LIMIT,
- LET, IMPORT, PARSER, AGGREGATION, INCLUDE, EXCLAMATION, MACRO, DUMP, TRACE;
+ LET, IMPORT, PARSER, AGGREGATION, INCLUDE, EXCLAMATION, MACRO, DUMP, TRACE, INCR;
terminal String Variable;
terminal Long Integer;
@@ -142,6 +142,7 @@ item ::= expr:e {: RESULT = #<expression
| STORE var:v ASSIGN expr:e {: RESULT = #<store(`v,`e)>; :}
| STORE String:s FROM expr:e {: RESULT = #<dump(`(new StringLeaf(s)),`e)>; :}
| DUMP String:s FROM expr:e {: RESULT = #<dump_text(`(new StringLeaf(s)),`e)>; :}
+ | INCR expr:e {: RESULT = #<incr(`e)>; :}
| TYPE var:v EQ type:t {: RESULT = #<typedef(`v,`t)>; :}
| DATA var:v EQ data_binds:nl {: RESULT = #<datadef(`v,union(...nl))>; :}
| FUNCTION var:f LP
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/core/src/main/java/org/apache/mrql/mrql.lex
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/mrql.lex b/core/src/main/java/org/apache/mrql/mrql.lex
index 643201d..2f6cb54 100644
--- a/core/src/main/java/org/apache/mrql/mrql.lex
+++ b/core/src/main/java/org/apache/mrql/mrql.lex
@@ -171,6 +171,7 @@ DOUBLE = [0-9]+([\.][0-9]+)?([eE][+-]?[0-9]+)?
<YYINITIAL> "all" { return symbol(sym.ALL); }
<YYINITIAL> "store" { return symbol(sym.STORE); }
<YYINITIAL> "dump" { return symbol(sym.DUMP); }
+<YYINITIAL> "incr" { return symbol(sym.INCR); }
<YYINITIAL> "type" { return symbol(sym.TYPE); }
<YYINITIAL> "data" { return symbol(sym.DATA); }
<YYINITIAL> "case" { return symbol(sym.CASE); }
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen b/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
index 24cac23..6ccc8a5 100644
--- a/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
+++ b/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
@@ -139,7 +139,7 @@ public class FlinkEvaluator extends Evaluator implements Serializable {
}
/** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
- final public void streaming ( Tree plan, Environment env, DataSetFunction f ) {
+ final public void streaming ( Tree plan, Environment env, Function f ) {
FlinkStreaming.evaluate(plan,env,f);
}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1cf73e2c/flink/src/main/java/org/apache/mrql/FlinkStreaming.gen
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkStreaming.gen b/flink/src/main/java/org/apache/mrql/FlinkStreaming.gen
index 232c99d..12ec317 100644
--- a/flink/src/main/java/org/apache/mrql/FlinkStreaming.gen
+++ b/flink/src/main/java/org/apache/mrql/FlinkStreaming.gen
@@ -143,7 +143,7 @@ public class FlinkStreaming extends FlinkEvaluator {
}
private static void stream_processing ( final Tree plan, final Environment env,
- final DataSetFunction f, final DataStream<FData> stream ) {
+ final Function f, final DataStream<FData> stream ) {
final Evaluator ef = Evaluator.evaluator;
match plan {
case Stream(lambda(`v,`b),`source):
@@ -163,7 +163,7 @@ public class FlinkStreaming extends FlinkEvaluator {
}
/** evaluate plan in stream mode: evaluate each batch of data and apply the function f */
- final public static void evaluate ( Tree plan, final Environment env, final DataSetFunction f ) {
+ final public static void evaluate ( Tree plan, final Environment env, final Function f ) {
try {
if (true) // needs more work
throw new Error("MRQL Streaming is not supported in this evaluation mode yet");