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 2014/03/13 15:24:50 UTC
[25/26] MRQL-32: Refactoring directory structure for Eclipse
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/bsp/src/main/java/org/apache/mrql/BSPPlan.java
----------------------------------------------------------------------
diff --git a/bsp/src/main/java/org/apache/mrql/BSPPlan.java b/bsp/src/main/java/org/apache/mrql/BSPPlan.java
new file mode 100644
index 0000000..bca8ab9
--- /dev/null
+++ b/bsp/src/main/java/org/apache/mrql/BSPPlan.java
@@ -0,0 +1,507 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.mrql;
+
+import org.apache.mrql.gen.*;
+import java.io.*;
+import java.util.Arrays;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hama.bsp.*;
+import org.apache.hama.bsp.sync.SyncException;
+import org.apache.hama.HamaConfiguration;
+import org.apache.hadoop.conf.Configuration;
+
+
+/** Evaluate a BSP plan using Hama */
+final public class BSPPlan extends Plan {
+
+ final static Configuration getConfiguration ( BSPJob job ) {
+ return job.getConf(); // use job.getConfiguration() for Hama 0.6.0
+ }
+
+ /** The BSP evaluator */
+ final static class BSPop extends BSP<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> {
+ final static MRContainer null_key = new MRContainer(new MR_byte(0));
+ // special message for sub-sync()
+ final static MRData more_to_come = new MR_sync();
+ final static MRData more_supersteps = new MR_more_bsp_steps();
+
+ private int source_num;
+ private Function superstep_fnc; // superstep function
+ private MRData state; // BSP state
+ private boolean orderp; // will output be ordered?
+ private MRData source; // BSP input
+ private Function acc_fnc; // aggregator
+ private MRData acc_result; // aggregation result
+ private static String[] all_peer_names; // all BSP peer names
+ private static BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer>[] all_peers; // all BSP peers
+ // a master peer that coordinates and collects results of partial aggregations
+ private String masterTask;
+ // buffer for received messages -- regularly in a vector, but can be spilled in a local file
+ Bag msg_cache;
+ // the cache that holds all local data in memory
+ Tuple local_cache;
+
+ private static BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> getPeer ( String name ) {
+ for ( int i = 0; i < all_peer_names.length; i++ )
+ if (all_peer_names[i].equals(name))
+ return all_peers[i];
+ throw new Error("Unknown peer: "+name);
+ }
+
+ private static void setPeer ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer ) {
+ String name = peer.getPeerName();
+ for ( int i = 0; i < all_peer_names.length; i++ )
+ if (all_peer_names[i].equals(name))
+ all_peers[i] = peer;
+ }
+
+ /** shuffle values to BSP peers based on uniform hashing on key */
+ private static String shuffle ( MRData key ) {
+ return all_peer_names[Math.abs(key.hashCode()) % all_peer_names.length];
+ }
+
+ /** to exit a BSP loop, all peers must agree to exit (this is used in BSPTranslate.bspSimplify) */
+ public static MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
+ return synchronize(getPeer(peerName.get()),mr_exit);
+ }
+
+ /** to exit a BSP loop, all peers must agree to exit (this is used in BSPTranslate.bspSimplify) */
+ public static MR_bool synchronize ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer, MR_bool mr_exit ) {
+ if (!Config.hadoop_mode)
+ return mr_exit;
+ // shortcut: if we know for sure that all peers want to exit/continue, we don't have to poll
+ if (mr_exit == SystemFunctions.bsp_true_value // must be ==, not equals
+ || mr_exit == SystemFunctions.bsp_false_value)
+ return mr_exit;
+ try {
+ // this case is only used for checking the exit condition of repeat/closure
+ boolean exit = mr_exit.get();
+ if (all_peer_names.length <= 1)
+ return (exit) ? SystemFunctions.bsp_true_value : SystemFunctions.bsp_false_value;
+ if (!exit)
+ // this peer is not ready to exit, so no peer should exit
+ for ( String p: peer.getAllPeerNames() )
+ peer.send(p,new MRContainer(more_supersteps));
+ peer.sync();
+ // now exit is true if no peer sent a "more_supersteps" message
+ exit = peer.getNumCurrentMessages() == 0;
+ peer.clear();
+ return (exit) ? SystemFunctions.bsp_true_value : SystemFunctions.bsp_false_value;
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
+
+ /** collect a bag from all peers by distributing the local copy s */
+ public static Bag distribute ( MR_string peerName, Bag s ) {
+ BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer = getPeer(peerName.get());
+ if (!Config.hadoop_mode)
+ return s;
+ try {
+ for ( MRData e: s )
+ for ( String p: all_peer_names )
+ peer.send(p,new MRContainer(e));
+ peer.sync();
+ MRContainer msg;
+ Bag res = new Bag();
+ while ((msg = peer.getCurrentMessage()) != null)
+ if (!res.contains(msg.data()))
+ res.add(msg.data());
+ peer.clear();
+ return res;
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
+
+ private void readLocalSources ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
+ throws IOException {
+ MRContainer key = new MRContainer();
+ MRContainer value = new MRContainer();
+ while (peer.readNext(key,value)) {
+ Tuple p = (Tuple)(value.data());
+ ((Bag)local_cache.get(((MR_int)p.first()).get())).add(p.second());
+ }
+ }
+
+ private void writeLocalResult ( Bag result,
+ BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
+ throws IOException {
+ MRContainer key = new MRContainer();
+ MRContainer value = new MRContainer();
+ for ( MRData v: result )
+ if (orderp) { // prepare for sorting
+ Tuple t = (Tuple)v;
+ key.set(t.get(1));
+ value.set(t.get(0));
+ peer.write(key,value);
+ } else {
+ value.set(v);
+ peer.write(null_key,value);
+ }
+ }
+
+ /** receive messages from other peers */
+ private void receive_messages ( final BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
+ throws IOException, SyncException, InterruptedException {
+ if (Config.bsp_msg_size <= 0) { // no buffering
+ msg_cache = new Bag(new BagIterator() {
+ MRContainer msg;
+ public boolean hasNext () {
+ try {
+ return (msg = peer.getCurrentMessage()) != null;
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
+ public MRData next () {
+ return msg.data();
+ }
+ });
+ } else {
+ boolean expect_more = false; // are we expecting more incoming messages?
+ do {
+ // just in case this peer did a regular-sync() before the others did a sub-sync()
+ expect_more = false;
+ MRContainer msg;
+ // cache the received messages
+ while ((msg = peer.getCurrentMessage()) != null)
+ // if at least one peer sends a more_to_come message, then expect_more
+ if (msg.data().equals(more_to_come))
+ expect_more = true;
+ else msg_cache.add(msg.data());
+ if (expect_more)
+ peer.sync(); // sub-sync()
+ } while (expect_more);
+ }
+ }
+
+ /** send the messages produced by a superstep to peers and then receive the replies */
+ private void send_messages ( Bag msgs,
+ BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
+ throws IOException, SyncException, InterruptedException {
+ int size = 0;
+ if (Config.bsp_msg_size > 0)
+ msg_cache.clear();
+ for ( MRData m: msgs ) {
+ Tuple t = (Tuple)m;
+ // if there are too many messages to send, then sub-sync()
+ if ( Config.bsp_msg_size > 0 && size++ > Config.bsp_msg_size ) {
+ // tell all peers that there is more to come after sync
+ for ( String p: all_peer_names )
+ if (!peer.getPeerName().equals(p))
+ peer.send(p,new MRContainer(more_to_come));
+ peer.sync(); // sub-sync()
+ size = 0;
+ MRContainer msg;
+ // cache the received messages
+ while ((msg = peer.getCurrentMessage()) != null)
+ if (!msg.data().equals(more_to_come))
+ msg_cache.add(msg.data());
+ };
+ // suffle messages based on key; needs new MRContainer object
+ peer.send(shuffle(t.get(0)),new MRContainer(t.get(1)));
+ };
+ peer.sync(); // regular-sync()
+ receive_messages(peer);
+ }
+
+ @Override
+ public void bsp ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
+ throws IOException, SyncException, InterruptedException {
+ final Tuple stepin = new Tuple(4);
+ stepin.set(3,new MR_string(peer.getPeerName()));
+ Tuple result;
+ boolean skip = false;
+ String tabs = "";
+ int step = 0;
+ boolean exit;
+ if (Evaluator.evaluator == null)
+ try {
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.BSPEvaluator").newInstance();
+ } catch (Exception ex) {
+ throw new Error(ex);
+ };
+ readLocalSources(peer);
+ setPeer(peer);
+ do {
+ if (!skip)
+ step++;
+ if (!skip && Config.trace_execution) {
+ tabs = Interpreter.tabs(Interpreter.tab_count);
+ System.err.println(tabs+" Superstep "+step+" ["+peer.getPeerName()+"]:");
+ System.err.println(tabs+" messages ["+peer.getPeerName()+"]: "+msg_cache);
+ System.err.println(tabs+" state ["+peer.getPeerName()+"]: "+state);
+ for ( int i = 0; i < local_cache.size(); i++)
+ if (local_cache.get(i) instanceof Bag && ((Bag)local_cache.get(i)).size() > 0)
+ System.out.println(tabs+" cache ["+peer.getPeerName()+"] "+i+": "+local_cache.get(i));
+ };
+ stepin.set(0,local_cache);
+ stepin.set(1,msg_cache);
+ stepin.set(2,state);
+ // evaluate one superstep
+ result = (Tuple)superstep_fnc.eval(stepin);
+ Bag msgs = (Bag)result.get(0);
+ exit = ((MR_bool)result.get(2)).get();
+ state = result.get(1);
+ // shortcuts: if we know for sure that all peers want to exit/continue
+ if (result.get(2) == SystemFunctions.bsp_true_value) { // must be ==, not equals
+ peer.sync();
+ if (Config.trace_execution)
+ System.err.println(tabs+" result ["+peer.getPeerName()+"]: "+result);
+ break;
+ };
+ if (result.get(2) == SystemFunctions.bsp_false_value) {
+ if (Config.trace_execution)
+ System.err.println(tabs+" result ["+peer.getPeerName()+"]: "+result);
+ send_messages(msgs,peer);
+ skip = false;
+ continue;
+ };
+ // shortcut: skip is true when NONE of the peers sent any messages
+ skip = (msgs == SystemFunctions.bsp_empty_bag); // must be ==, not equals
+ if (skip)
+ continue;
+ if (Config.trace_execution)
+ System.err.println(tabs+" result ["+peer.getPeerName()+"]: "+result);
+ exit = synchronize(peer,(MR_bool)result.get(2)).get();
+ send_messages(msgs,peer);
+ } while (!exit);
+ if (acc_result == null) {
+ // the BSP result is a bag that needs to be dumped to the HDFS
+ writeLocalResult((Bag)(local_cache.get(source_num)),peer);
+ // if there more results, dump them to HDFS
+ final MR_long key = new MR_long(0);
+ final MRContainer key_container = new MRContainer(key);
+ final MRContainer data_container = new MRContainer(new MR_int(0));
+ int loc = 0;
+ while ( loc < all_peer_names.length && peer.getPeerName().equals(all_peer_names[loc]) )
+ loc++;
+ Configuration conf = peer.getConfiguration();
+ String[] out_paths = conf.get("mrql.output.paths").split(",");
+ for ( int i = 1; i < out_paths.length; i++ ) {
+ String[] s = out_paths[i].split(":");
+ int out_num = Integer.parseInt(s[0]);
+ Path path = new Path(s[1]+"/peer"+loc);
+ FileSystem fs = path.getFileSystem(conf);
+ SequenceFile.Writer writer
+ = new SequenceFile.Writer(fs,conf,path,
+ MRContainer.class,MRContainer.class);
+ long count = 0;
+ for ( MRData e: (Bag)(local_cache.get(out_num)) ) {
+ key.set(count++);
+ data_container.set(e);
+ writer.append(key_container,data_container);
+ };
+ writer.close();
+ };
+ } else {
+ // the BSP result is an aggregation:
+ // send the partial results to the master peer
+ peer.send(masterTask,new MRContainer(local_cache.get(source_num)));
+ peer.sync();
+ if (peer.getPeerName().equals(masterTask)) {
+ // only the master peer collects the partial aggregations
+ MRContainer msg;
+ while ((msg = peer.getCurrentMessage()) != null)
+ acc_result = acc_fnc.eval(new Tuple(acc_result,msg.data()));
+ // write the final aggregation result
+ peer.write(null_key,new MRContainer(acc_result));
+ }
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void setup ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer ) {
+ try {
+ super.setup(peer);
+ Configuration conf = peer.getConfiguration();
+ Config.read(conf);
+ if (Plan.conf == null)
+ Plan.conf = conf;
+ all_peer_names = peer.getAllPeerNames();
+ all_peers = new BSPPeerImpl[all_peer_names.length];
+ Arrays.sort(all_peer_names); // is this necessary?
+ source_num = conf.getInt("mrql.output.tag",0);
+ Tree code = Tree.parse(conf.get("mrql.superstep"));
+ superstep_fnc = functional_argument(conf,code);
+ code = Tree.parse(conf.get("mrql.initial.state"));
+ state = Interpreter.evalE(code);
+ if (conf.get("mrql.zero") != null && !conf.get("mrql.zero").equals("")) {
+ code = Tree.parse(conf.get("mrql.zero"));
+ acc_result = Interpreter.evalE(code);
+ code = Tree.parse(conf.get("mrql.accumulator"));
+ acc_fnc = functional_argument(conf,code);
+ } else acc_result = null;
+ orderp = conf.getBoolean("mrql.orderp",false);
+ masterTask = all_peer_names[peer.getNumPeers()/2];
+ msg_cache = new Bag(1000);
+ local_cache = new Tuple(max_input_files);
+ for ( int i = 0; i < max_input_files; i++ )
+ local_cache.set(i,new Bag());
+ } catch (Exception e) {
+ e.printStackTrace();
+ throw new Error("Cannot setup the Hama BSP job: "+e);
+ }
+ }
+
+ @Override
+ public void cleanup ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer ) throws IOException {
+ if (!Config.local_mode)
+ clean();
+ local_cache = null;
+ super.cleanup(peer);
+ }
+ }
+
+ /** set Hama's min split size and number of BSP tasks (doesn't work with Hama 0.6.0) */
+ public static void setupSplits ( BSPJob job, DataSet ds ) throws IOException {
+ long[] sizes = new long[ds.source.size()];
+ if (sizes.length > Config.nodes)
+ throw new Error("Cannot distribute "+sizes.length+" files over "+Config.nodes+" BSP tasks");
+ for ( int i = 0; i < sizes.length; i++ )
+ sizes[i] = ds.source.get(i).size(Plan.conf);
+ long total_size = 0;
+ for ( long size: sizes )
+ total_size += size;
+ long split_size = Math.max(total_size/Config.nodes,100000);
+ int tasks = 0;
+ do { // adjust split_size
+ tasks = 0;
+ for ( long size: sizes )
+ tasks += (int)Math.ceil(size/(double)split_size);
+ if (tasks > Config.nodes)
+ split_size = (long)Math.ceil((double)split_size*1.01);
+ } while (tasks > Config.nodes);
+ job.setNumBspTask(tasks);
+ System.err.println("*** Using "+tasks+" BSP tasks (out of a max "+Config.nodes+")."
+ +" Each task will handle about "+Math.min(total_size/Config.nodes,split_size)
+ +" bytes of input data.");
+ job.set("bsp.min.split.size",Long.toString(split_size));
+ }
+
+ /** Evaluate a BSP operation that returns a DataSet
+ * @param source_nums output tags
+ * @param superstep the superstep function
+ * @param init_state initial state
+ * @param orderp do we need to order the result?
+ * @param source input dataset
+ * @return a new data source that contains the result
+ */
+ public final static MRData BSP ( int[] source_nums, // output tags
+ Tree superstep, // superstep function
+ Tree init_state, // initial state
+ boolean orderp, // do we need to order the result?
+ DataSet source // input dataset
+ ) throws Exception {
+ String[] newpaths = new String[source_nums.length];
+ newpaths[0] = new_path(conf);
+ conf.set("mrql.output.paths",source_nums[0]+":"+newpaths[0]);
+ for ( int i = 1; i < source_nums.length; i++ ) {
+ newpaths[i] = new_path(conf);
+ Path path = new Path(newpaths[1]);
+ FileSystem fs = path.getFileSystem(conf);
+ fs.mkdirs(path);
+ conf.set("mrql.output.paths",conf.get("mrql.output.paths")+","+source_nums[i]+":"+newpaths[i]);
+ };
+ conf.set("mrql.superstep",superstep.toString());
+ conf.set("mrql.initial.state",init_state.toString());
+ conf.set("mrql.zero","");
+ conf.setInt("mrql.output.tag",source_nums[0]);
+ conf.setBoolean("mrql.orderp",orderp);
+ BSPJob job = new BSPJob((HamaConfiguration)conf,BSPop.class);
+ setupSplits(job,source);
+ job.setJobName(newpaths[0]);
+ distribute_compiled_arguments(getConfiguration(job));
+ job.setBspClass(BSPop.class);
+ Path outpath = new Path(newpaths[0]);
+ job.setOutputPath(outpath);
+ job.setOutputKeyClass(MRContainer.class);
+ job.setOutputValueClass(MRContainer.class);
+ job.setOutputFormat(SequenceFileOutputFormat.class);
+ job.setInputFormat(MultipleBSPInput.class);
+ FileInputFormat.setInputPaths(job,source.merge());
+ job.waitForCompletion(true);
+ if (source_nums.length == 1) {
+ BinaryDataSource ds = new BinaryDataSource(source_nums[0],newpaths[0],conf);
+ ds.to_be_merged = orderp;
+ return new MR_dataset(new DataSet(ds,0,3));
+ } else {
+ MRData[] s = new MRData[source_nums.length];
+ for ( int i = 0; i < source_nums.length; i++ ) {
+ BinaryDataSource ds = new BinaryDataSource(source_nums[i],newpaths[i],conf);
+ ds.to_be_merged = orderp;
+ s[i] = new MR_dataset(new DataSet(ds,0,3));
+ };
+ return new Tuple(s);
+ }
+ }
+
+ /** Evaluate a BSP operation that aggregates the results
+ * @param source_num output tag
+ * @param superstep the superstep function
+ * @param init_state initial state
+ * @param acc_fnc accumulator function
+ * @param zero zero value for the accumulator
+ * @param source input dataset
+ * @return the aggregation result
+ */
+ public final static MRData BSPaggregate ( int source_num, // output tag
+ Tree superstep, // superstep function
+ Tree init_state, // initial state
+ Tree acc_fnc, // accumulator function
+ Tree zero, // zero value for the accumulator
+ DataSet source // input dataset
+ ) throws Exception {
+ String newpath = new_path(conf);
+ conf.set("mrql.superstep",superstep.toString());
+ conf.set("mrql.initial.state",init_state.toString());
+ conf.set("mrql.accumulator",acc_fnc.toString());
+ conf.set("mrql.zero",zero.toString());
+ conf.setInt("mrql.output.tag",source_num);
+ conf.setBoolean("mrql.orderp",false);
+ BSPJob job = new BSPJob((HamaConfiguration)conf,BSPop.class);
+ setupSplits(job,source);
+ job.setJobName(newpath);
+ distribute_compiled_arguments(getConfiguration(job));
+ job.setBspClass(BSPop.class);
+ Path outpath = new Path(newpath);
+ job.setOutputPath(outpath);
+ job.setOutputKeyClass(MRContainer.class);
+ job.setOutputValueClass(MRContainer.class);
+ job.setOutputFormat(SequenceFileOutputFormat.class);
+ job.setInputFormat(MultipleBSPInput.class);
+ FileInputFormat.setInputPaths(job,source.merge());
+ job.waitForCompletion(true);
+ FileSystem fs = outpath.getFileSystem(conf);
+ FileStatus[] files = fs.listStatus(outpath);
+ for ( int i = 0; i < files.length; i++ ) {
+ SequenceFile.Reader sreader = new SequenceFile.Reader(fs,files[i].getPath(),conf);
+ MRContainer key = new MRContainer();
+ MRContainer value = new MRContainer();
+ sreader.next(key,value);
+ sreader.close();
+ if (value.data() != null)
+ return value.data();
+ };
+ return null;
+ }
+ }
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/bsp/src/main/java/org/apache/mrql/MultipleBSPInput.java
----------------------------------------------------------------------
diff --git a/bsp/src/main/java/org/apache/mrql/MultipleBSPInput.java b/bsp/src/main/java/org/apache/mrql/MultipleBSPInput.java
new file mode 100644
index 0000000..aa43423
--- /dev/null
+++ b/bsp/src/main/java/org/apache/mrql/MultipleBSPInput.java
@@ -0,0 +1,55 @@
+/**
+ * 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.Trees;
+import java.io.*;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hama.bsp.*;
+
+
+/** A FileInputFormat for multiple files, where each file may be associated with
+ * a different FileInputFormat */
+final public class MultipleBSPInput extends BSPMRQLFileInputFormat {
+ public RecordReader<MRContainer,MRContainer>
+ getRecordReader ( InputSplit split, BSPJob job ) throws IOException {
+ if (Evaluator.evaluator == null)
+ try {
+ Evaluator.evaluator = (Evaluator)Class.forName("org.apache.mrql.BSPEvaluator").newInstance();
+ } catch (Exception ex) {
+ throw new Error(ex);
+ };
+ String path = ((FileSplit)split).getPath().toString();
+ Configuration conf = BSPPlan.getConfiguration(job);
+ DataSource ds = DataSource.get(path,conf);
+ Plan.conf = conf;
+ if (ds instanceof ParsedDataSource)
+ return new BSPParsedInputFormat.ParsedRecordReader((FileSplit)split,
+ job,
+ ((ParsedDataSource)ds).parser,
+ ds.source_num,
+ (Trees)((ParsedDataSource)ds).args);
+ else if (ds instanceof BinaryDataSource)
+ return new BSPBinaryInputFormat.BinaryInputRecordReader((FileSplit)split,job,ds.source_num);
+ else if (ds instanceof GeneratorDataSource)
+ return new BSPGeneratorInputFormat.GeneratorRecordReader((FileSplit)split,ds.source_num,job);
+ else throw new Error("Unknown data source: "+ds+" for path "+path);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/pom.xml
----------------------------------------------------------------------
diff --git a/core/pom.xml b/core/pom.xml
index 097fe50..f9d447e 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -42,22 +42,6 @@
<build>
<plugins>
<plugin>
- <groupId>org.codehaus.mojo</groupId>
- <artifactId>build-helper-maven-plugin</artifactId>
- <version>1.8</version>
- <executions>
- <execution>
- <phase>generate-sources</phase>
- <goals><goal>add-source</goal></goals>
- <configuration>
- <sources>
- <source>../src/main/java/core</source>
- </sources>
- </configuration>
- </execution>
- </executions>
- </plugin>
- <plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>2.1</version>
@@ -101,8 +85,8 @@
</goals>
<configuration>
<lexDefinitions>
- <lexDefinition>../src/main/java/core/mrql.lex</lexDefinition>
- <lexDefinition>../src/main/java/core/JSON.lex</lexDefinition>
+ <lexDefinition>src/main/java/org/apache/mrql/mrql.lex</lexDefinition>
+ <lexDefinition>src/main/java/org/apache/mrql/JSON.lex</lexDefinition>
</lexDefinitions>
<outputDirectory>${project.build.directory}/generated-sources</outputDirectory>
<verbose>false</verbose>
@@ -125,7 +109,7 @@
<target>
<mkdir dir="${project.build.directory}/generated-sources/org/apache/mrql" />
<property name="compile_classpath" refid="maven.compile.classpath" />
- <fileset id="gen.path" dir="../src/main/java/core" includes="*.gen" />
+ <fileset id="gen.path" dir="src/main/java/org/apache/mrql" includes="*.gen" />
<pathconvert pathsep=" " property="gen.files" refid="gen.path" />
<java classname="org.apache.mrql.gen.Main" classpath="../lib/mrql-gen-${project.version}.jar:${compile_classpath}">
<arg line="${gen.files}" />
@@ -133,13 +117,13 @@
<arg file="${project.build.directory}/generated-sources/org/apache/mrql" />
</java>
<java classname="org.apache.mrql.gen.Main" classpath="../lib/mrql-gen-${project.version}.jar:${compile_classpath}">
- <arg file="../src/main/java/core/mrql.cgen" />
+ <arg file="src/main/java/org/apache/mrql/mrql.cgen" />
<arg line="-o" />
<arg file="${project.build.directory}/generated-sources/mrql.cup" />
</java>
<taskdef name="cup" classname="java_cup.anttask.CUPTask" classpathref="maven.compile.classpath" />
<cup srcfile="${project.build.directory}/generated-sources/mrql.cup" destdir="${project.build.directory}/generated-sources" nosummary="true" parser="MRQLParser" />
- <cup srcfile="src/main/java/core/JSON.cup" destdir="${project.build.directory}/generated-sources" nosummary="true" parser="JSONParser" symbols="jsym" />
+ <cup srcfile="${project.basedir}/src/main/java/org/apache/mrql/JSON.cup" destdir="${project.build.directory}/generated-sources" nosummary="true" parser="JSONParser" symbols="jsym" />
</target>
</configuration>
</execution>
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/AlgebraicOptimization.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/AlgebraicOptimization.gen b/core/src/main/java/org/apache/mrql/AlgebraicOptimization.gen
new file mode 100644
index 0000000..d574e20
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/AlgebraicOptimization.gen
@@ -0,0 +1,315 @@
+/**
+ * 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.*;
+
+/**
+ * Optimize MRQL algebra expressions using normalization rules and heuristic optimizations
+ */
+public class AlgebraicOptimization extends Simplification {
+
+ /** Is this a group-by operation?*/
+ private static boolean is_groupBy ( Tree e ) {
+ match e {
+ case groupBy(...): return true;
+ case orderBy(...): return true;
+ };
+ return false;
+ }
+
+ /**
+ * algebraic optimization (algebra to algebra)
+ * @param e the algebraic form to be optimized
+ * @return the optimized form
+ */
+ public static Tree translate ( Tree e ) {
+ match e {
+ // if the group-by key is the same as the join key, fuse the group-by into the join
+ case mapReduce2(lambda(`v,bag(tuple(nth(`v1,0),`mx))),`my,
+ lambda(`rv,`rb),
+ `groupBy(cmap(lambda(`w,`m),`X)),`Y,`o):
+ if (!v1.equals(v) || !Config.groupJoinOpt || ! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ Tree nv = new_var();
+ Tree nr = subst(#<nth(`rv,0)>,#<cmap(lambda(`v,bag(`mx)),groupBy(nth(`rv,0)))>,rb);
+ Tree res = #<mapReduce2(lambda(`w,cmap(lambda(`nv,bag(tuple(nth(`nv,0),`nv))),`m)),
+ `my,lambda(`rv,`nr),`X,`Y,`o)>;
+ res = simplify_all(rename(res));
+ TypeInference.type_inference(res);
+ return translate(res);
+ // same for the right join input
+ case mapReduce2(`mx,lambda(`v,bag(tuple(nth(`v1,0),`my))),
+ lambda(`rv,`rb),
+ `X,`groupBy(cmap(lambda(`w,`m),`Y)),`o):
+ if (!v1.equals(v) || !Config.groupJoinOpt || ! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ Tree nv = new_var();
+ Tree nr = subst(#<nth(`rv,1)>,#<cmap(lambda(`v,bag(`my)),groupBy(nth(`rv,1)))>,rb);
+ Tree res = #<mapReduce2(`mx,lambda(`w,cmap(lambda(`nv,bag(tuple(nth(`nv,0),`nv))),`m)),
+ lambda(`rv,`nr),`X,`Y,`o)>;
+ res = simplify_all(rename(res));
+ TypeInference.type_inference(res);
+ return translate(res);
+ // convert self-joins to single-source mapreduce; Hadoop doesn't like self-joins anyway
+ case mapReduce2(lambda(`vx,`bx),lambda(`vy,`by),lambda(`s,`f),`X,`Y,`o):
+ if (!alpha_equivalent(X,Y) || !Config.selfJoinOpt)
+ fail;
+ Tree ny = subst(vy,vx,by);
+ Tree tx = null;
+ Tree ty = null;
+ match TypeInference.type_inference(bx) {
+ case _(tuple(_,`t)): tx = t;
+ };
+ match TypeInference.type_inference(by) {
+ case _(tuple(_,`t)): ty = t;
+ };
+ Tree b = subst(s,#<tuple(cmap(lambda(tuple(n,v1,v2),
+ if(call(eq,n,1),bag(v1),bag())),s),
+ cmap(lambda(tuple(n,v1,v2),
+ if(call(eq,n,2),bag(v2),bag())),s))>,
+ f);
+ Tree res = #<mapReduce(lambda(`vx,call(plus,
+ cmap(lambda(x,bag(tuple(nth(x,0),tuple(1,nth(x,1),typed(null,`ty))))),`bx),
+ cmap(lambda(y,bag(tuple(nth(y,0),tuple(2,typed(null,`tx),nth(y,1))))),`ny))),
+ lambda(tuple(k,s),`b),
+ `X,`o)>;
+ res = simplify_all(rename(res));
+ TypeInference.type_inference(res);
+ return translate(res);
+ case mapReduce2(`mx,`my,`r,cmap(lambda(`v,`b),`X),`Y,`o):
+ match X {
+ case groupBy(...): ;
+ case _: return translate(#<mapReduce2(lambda(`v,cmap(`mx,`b)),`my,`r,`X,`Y,`o)>);
+ };
+ fail
+ case mapReduce2(`mx,`my,`r,`X,cmap(lambda(`v,`b),`Y),`o):
+ match Y {
+ case groupBy(...): ;
+ case _: return translate(#<mapReduce2(`mx,lambda(`v,cmap(`my,`b)),`r,`X,`Y,`o)>);
+ };
+ fail
+ case crossProduct(`mx,`my,`r,cmap(lambda(`vx,`bx),`X),cmap(lambda(`vy,`by),`Y)):
+ return translate(#<crossProduct(lambda(`vx,cmap(`mx,`bx)),
+ lambda(`vy,cmap(`my,`by)),`r,`X,`Y)>);
+ case crossProduct(`mx,`my,`r,cmap(lambda(`v,`b),`X),`Y):
+ return translate(#<crossProduct(lambda(`v,cmap(`mx,`b)),`my,`r,`X,`Y)>);
+ case crossProduct(`mx,`my,`r,`X,cmap(lambda(`v,`b),`Y)):
+ return translate(#<crossProduct(`mx,lambda(`v,cmap(`my,`b)),`r,`X,`Y)>);
+ case cmap(`m,crossProduct(`mx,`my,lambda(`v,`b),`X,`Y)):
+ return translate(#<crossProduct(`mx,`my,lambda(`v,cmap(`m,`b)),`X,`Y)>);
+ case cmap(`r,`groupBy1(cmap(`m,`groupBy2(`s)))):
+ if (! #[groupBy,orderBy].member(#<`groupBy1>)
+ && ! #[groupBy,orderBy].member(#<`groupBy2>))
+ fail;
+ return #<mapReduce(`(identity()),
+ `(translate(r)),
+ `(translate(#<cmap(`m,`groupBy2(`s))>)),
+ `((#<`groupBy1>.equals(#<orderBy>)) ? #<true> : #<false>))>;
+ case cmap(`r,`groupBy(cmap(`m,`s))):
+ if (! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ return #<mapReduce(`(translate(m)),
+ `(translate(r)),
+ `(translate(s)),
+ `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
+ case `groupBy(cmap(`m,groupBy(`s))):
+ if (! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ return #<mapReduce(`(identity()),
+ `(identity()),
+ `(translate(#<cmap(`m,groupBy(`s))>)),
+ `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
+ case `groupBy(cmap(`m,`s)):
+ if (! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ return #<mapReduce(`(translate(m)),
+ `(identity()),
+ `(translate(s)),
+ `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
+ case cmap(`r,`groupBy(`s)):
+ if (! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ return #<mapReduce(`(identity()),
+ `(translate(r)),
+ `(translate(s)),
+ `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
+ case `groupBy(`s):
+ if (! #[groupBy,orderBy].member(#<`groupBy>))
+ fail;
+ return #<mapReduce(`(identity()),
+ `(identity()),
+ `(translate(s)),
+ `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
+ case cmap(`m,`s):
+ return #<cmap(`(translate(m)),
+ `(translate(s)))>;
+ // convert self-joins to single-source mapreduce; Hadoop doesn't like self-joins anyway
+ case join(lambda(`vx,`bx),lambda(`vy,`by),lambda(`s,`f),`x,`y):
+ if (!x.equals(y) || !Config.selfJoinOpt)
+ fail;
+ Tree ny = subst(vy,vx,by);
+ Tree b = subst(s,
+ #<tuple(cmap(lambda(tuple(n,v),
+ if(call(eq,n,1),bag(v),bag())),s),
+ cmap(lambda(tuple(n,v),
+ if(call(eq,n,2),bag(v),bag())),s))>,
+ f);
+ Tree res = #<mapReduce(lambda(`vx,bag(tuple(`bx,tuple(1,`vx)),
+ tuple(`ny,tuple(2,`vx)))),
+ lambda(tuple(k,s),`b),`x,false)>;
+ res = simplify_all(rename(res));
+ TypeInference.type_inference(res);
+ return translate(res);
+ case join(lambda(`vx,`bx),lambda(`vy,`by),`f,`x,`y):
+ return translate(#<mapReduce2(lambda(`vx,bag(tuple(`bx,`vx))),
+ lambda(`vy,bag(tuple(`by,`vy))),
+ `f,`x,`y,false)>);
+ case nth(`x,`n):
+ match TypeInference.type_inference2(x) {
+ case `S(tuple(...bl)):
+ if (!is_collection(S))
+ fail;
+ Tree nv = new_var();
+ type_env.insert(nv.toString(),bl.nth((int)n.longValue()));
+ return translate(#<cmap(lambda(`nv,`S(nth(`nv,`n))),`x)>);
+ };
+ fail
+ case project(`x,`a):
+ match TypeInference.type_inference2(x) {
+ case `S(record(...bl)):
+ if (!is_collection(S))
+ fail;
+ for ( Tree b: bl )
+ match b {
+ case bind(`c,_):
+ if (!a.equals(c))
+ fail;
+ Tree nv = new_var();
+ type_env.insert(nv.toString(),#<record(...bl)>);
+ return translate(#<cmap(lambda(`nv,`S(project(`nv,`a))),`x)>);
+ };
+ };
+ fail
+ case `f(...al):
+ Trees bl = #[];
+ for ( Tree a: al )
+ bl = bl.append(translate(a));
+ return #<`f(...bl)>;
+ };
+ return e;
+ }
+
+ /** apply algebraic optimizations multiple times until no change */
+ public static Tree translate_all ( Tree e ) {
+ Tree ne = translate(e);
+ if (e.equals(ne))
+ return e;
+ else return translate_all(ne);
+ }
+
+ /**
+ * does a form contain a bulk plan that doesn't refer to a given variable?
+ * @param e the form
+ * @param v the given variable
+ * @return true if e contains a bulk plan that doesn't refer to v
+ */
+ private static boolean contains_plan ( Tree e, Tree v ) {
+ match e {
+ case lambda(`x,`u): return false;
+ case let(...): return false;
+ case Let(...): return false;
+ case `f(...as):
+ if (plan_names.member(#<`f>) && !free_variables(e,#[]).member(v))
+ return true;
+ for (Tree a: as)
+ if (contains_plan(a,v))
+ return true;
+ return false;
+ };
+ return false;
+ }
+
+ /**
+ * extract the common factors (common sub-expressions) from a form
+ * that do not refer to a given variable
+ * @param e the form
+ * @param v the given variable
+ * @return the list of common factors
+ */
+ private static Trees common_factors ( Tree e, Tree v ) {
+ match e {
+ case lambda(`x,`u): return #[];
+ case let(...): return #[];
+ case Let(...): return #[];
+ case `f(...as):
+ if (!contains_plan(e,v))
+ fail;
+ if (plan_names.member(#<`f>) && !free_variables(e,#[]).member(v))
+ return #[`e];
+ Trees bs = #[];
+ for ( Tree a: as )
+ bs = bs.append(common_factors(a,v));
+ return bs;
+ };
+ return #[];
+ }
+
+ /**
+ * if a term is used multiple times in a query, factor it out using let-expressions
+ * @param e the expression to be factored-out
+ * @return the factored-out expression
+ */
+ public static Tree common_factoring ( Tree e ) {
+ match e {
+ case `f(...as):
+ if (!plan_names.member(#<`f>))
+ fail;
+ Trees bs = #[];
+ Trees binds = #[];
+ for ( Tree a: as )
+ match a {
+ case lambda(`v,`u):
+ if (!contains_plan(u,v))
+ fail;
+ Trees gs = common_factors(u,v);
+ Tree nb = u;
+ for ( Tree g: gs) {
+ Tree nv = new_var();
+ nb = subst(g,nv,nb);
+ binds = binds.append(#<bind(`nv,`g)>);
+ };
+ bs = bs.append(#<lambda(`v,`(common_factoring(nb)))>);
+ case _: bs = bs.append(common_factoring(a));
+ };
+ Tree res = #<`f(...bs)>;
+ for ( Tree bind: binds )
+ match bind {
+ case bind(`v,`x):
+ res = #<let(`v,`x,`res)>;
+ };
+ return res;
+ case `f(...as):
+ Trees bs = #[];
+ for ( Tree a: as )
+ bs = bs.append(common_factoring(a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/core/src/main/java/org/apache/mrql/BSPTranslator.gen
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/BSPTranslator.gen b/core/src/main/java/org/apache/mrql/BSPTranslator.gen
new file mode 100644
index 0000000..a2af7a6
--- /dev/null
+++ b/core/src/main/java/org/apache/mrql/BSPTranslator.gen
@@ -0,0 +1,753 @@
+/**
+ * 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.*;
+
+
+/** Translates physical plans to BSP plans to be executed on Hama */
+public class BSPTranslator extends TypeInference {
+ final static int orM = ClassImporter.find_method_number("or",#[bool,bool]);
+ final static int andM = ClassImporter.find_method_number("and",#[bool,bool]);
+ final static int notM = ClassImporter.find_method_number("not",#[bool]);
+ final static int eqM = ClassImporter.find_method_number("eq",#[int,int]);
+ final static int plusM = ClassImporter.find_method_number("plus",#[int,int]);
+ final static int neqM = ClassImporter.find_method_number("neq",#[int,int]);
+ final static int geqM = ClassImporter.find_method_number("geq",#[long,long]);
+ final static int gtM = ClassImporter.find_method_number("gt",#[int,int]);
+ final static int unionM = ClassImporter.find_method_number("union",#[bag(any),bag(any)]);
+ final static int countM = ClassImporter.find_method_number("count",#[bag(any)]);
+ final static int coerceM = ClassImporter.find_method_number("coerce",#[any,int]);
+
+ final static Trees planNames = plans_with_distributed_lambdas
+ .append(#[ParsedSource,Generator,BinarySource,GroupByJoin,Repeat,Closure,repeat]);
+
+ private static int source_num = 0;
+
+ public static void reset () {
+ source_num = 0;
+ }
+
+ private static Tree new_source ( Tree e ) {
+ match e {
+ case `f(...):
+ if (planNames.member(#<`f>))
+ return preprocess(e);
+ };
+ return #<BSPSource(`(source_num++),`(preprocess(e)))>;
+ }
+
+ /** add a new source num to every BSP operation */
+ private static Tree preprocess ( Tree e ) {
+ match e {
+ case lambda(`v,`b):
+ return #<lambda(`v,`(preprocess(b)))>;
+ case Aggregate(`acc,`zero,`S):
+ return #<Aggregate(`acc,`zero,
+ `(new_source(S)))>;
+ case cMap(`m,`S):
+ return #<cMap(`m,
+ `(new_source(S)))>;
+ case AggregateMap(`m,`acc,`zero,`S):
+ return #<AggregateMap(`m,`acc,`zero,
+ `(new_source(S)))>;
+ case MapReduce(`m,`r,`S,`o):
+ return #<MapReduce(`m,`r,
+ `(new_source(S)),`o)>;
+ case MapAggregateReduce(`m,`r,`acc,`zero,`S,`o):
+ return #<MapAggregateReduce(`m,`r,`acc,`zero,
+ `(new_source(S)),`o)>;
+ case MapCombineReduce(`m,`c,`r,`S,`o):
+ return #<MapCombineReduce(`m,`c,`r,
+ `(new_source(S)),`o)>;
+ case MapReduce2(`mx,`my,`r,`x,`y,`o):
+ return #<MapReduce2(`mx,`my,`r,
+ `(new_source(x)),
+ `(new_source(y)),`o)>;
+ case MapReduce2(`mx,`my,`c,`r,`x,`y,`o):
+ return #<MapCombineReduce2(`mx,`my,`c,`r,
+ `(new_source(x)),
+ `(new_source(y)),`o)>;
+ case MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,`o):
+ return #<MapAggregateReduce2(`mx,`my,`r,`acc,`zero,
+ `(new_source(x)),
+ `(new_source(y)),`o)>;
+ case MapJoin(`mx,`my,`r,`x,`y):
+ return #<MapJoin(`mx,`my,`r,
+ `(new_source(x)),
+ `(new_source(y)))>;
+ case MapAggregateJoin(`mx,`my,`r,`acc,`zero,`x,`y):
+ return #<MapAggregateJoin(`mx,`my,`r,`acc,`zero,
+ `(new_source(x)),
+ `(new_source(y)))>;
+ case CrossProduct(`mx,`my,`r,`x,`y):
+ return #<CrossProduct(`mx,`my,`r,
+ `(new_source(x)),
+ `(new_source(y)))>;
+ case CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y):
+ return #<CrossAggregateProduct(`mx,`my,`r,`acc,`zero,
+ `(new_source(x)),
+ `(new_source(y)))>;
+ case GroupByJoin(`kx,`ky,`gx,`gy,`mr,`c,`r,`x,`y,`o):
+ return #<GroupByJoin(`kx,`ky,`gx,`gy,`mr,`c,`r,
+ `(new_source(x)),
+ `(new_source(y)),`o)>;
+ case Repeat(`f,`ds,`max):
+ return #<Repeat(`(preprocess(f)),`(preprocess(ds)),`max)>;
+ case repeat(`f,`ds,`max):
+ return #<repeat(`(preprocess(f)),`(preprocess(ds)),`max)>;
+ case Closure(`f,`ds,`max):
+ return #<Closure(`(preprocess(f)),`(preprocess(ds)),`max)>;
+ case `f(...):
+ if (! #[ParsedSource,Generator,BinarySource].member(#<`f>))
+ fail;
+ return #<BSPSource(`(source_num++),`e)>;
+ case `f(...as):
+ Trees bs = #[];
+ for (Tree a: as)
+ bs = bs.append(preprocess(a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ /** returns the source num of a BSP operation */
+ private static int source_num ( Tree e ) {
+ match e {
+ case BSP(`i,...):
+ return (int)((LongLeaf)i).value();
+ case BSPSource(`i,_):
+ return (int)((LongLeaf)i).value();
+ };
+ return -1;
+ }
+
+ private static Tree subst_getCache_num ( int n, int m, Tree e ) {
+ match e {
+ case getCache(`cache,`k):
+ if (!k.equals(#<`n>))
+ fail;
+ return #<getCache(`cache,`m)>;
+ case `f(...as):
+ Trees bs = #[];
+ for (Tree a: as)
+ bs = bs.append(subst_getCache_num(n,m,a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ private static Tree subst_setCache_num ( int n, int m, Tree e ) {
+ match e {
+ case setCache(`cache,`k,...r):
+ if (!k.equals(#<`n>))
+ fail;
+ return #<setCache(`cache,`m,...r)>;
+ case `f(...as):
+ Trees bs = #[];
+ for (Tree a: as)
+ bs = bs.append(subst_setCache_num(n,m,a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ private static Tree set_cache_num ( Tree e, int n ) {
+ match e {
+ case BSP(`m,...r):
+ return subst_setCache_num((int)((LongLeaf)m).value(),n,#<BSP(`n,...r)>);
+ case BSPSource(_,`x):
+ return #<BSPSource(`n,`x)>;
+ };
+ return e;
+ }
+
+ // return the source numbers of a BSP operation
+ private static Trees source_nums ( Tree e ) {
+ match e {
+ case BSP(`i,...):
+ return #[`i];
+ case Merge(`x,`y):
+ return source_nums(x).append(source_nums(y));
+ case BSPSource(`i,_):
+ return #[`i];
+ };
+ return #[];
+ }
+
+ private static Trees source_nums ( Trees s ) {
+ Trees res = #[];
+ for ( Tree x: s )
+ res = res.append(source_nums(x));
+ return res;
+ }
+
+ private static Tree getCache ( Tree var, Tree e, Tree body ) {
+ Trees ns = source_nums(e);
+ if (ns.length() > 0)
+ return subst(var,#<getCache(cache,...ns)>,body);
+ return body;
+ }
+
+ private static Tree getCache ( Tree var, Trees s, Tree body ) {
+ Trees ns = source_nums(s);
+ return subst(var,#<getCache(cache,...ns)>,body);
+ }
+
+ /** optimize a BSP plan after BSP fusion */
+ private static Tree post_simplify ( Tree e ) {
+ match e {
+ case cmap(lambda(`x,`S(`a)),`b):
+ if (is_collection(S))
+ return post_simplify(#<map(lambda(`x,`a),`b)>);
+ else fail
+ case `f(...as):
+ Trees bs = #[];
+ for (Tree a: as)
+ bs = bs.append(post_simplify(a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ private static Tree processBSP ( Tree e ) {
+ return post_simplify(Simplification.simplify_all(Simplification.rename(e)));
+ }
+
+ private static Tree mkBSP ( Tree superstep, Tree state, Tree orderp, Tree input ) {
+ int rn = source_num++;
+ superstep = processBSP(subst(#<o_>,#<`rn>,getCache(#<i_>,input,superstep)));
+ return #<BSP(`rn,`superstep,`state,`orderp,`input)>;
+ }
+
+ private static Tree mkBSPL ( Tree superstep, Tree state, Tree orderp, Trees input ) {
+ int rn = source_num++;
+ superstep = processBSP(subst(#<o_>,#<`rn>,getCache(#<i>,input,superstep)));
+ return #<BSP(`rn,`superstep,`state,`orderp,...input)>;
+ }
+
+ private static Tree mkBSPL ( int[] ns, Tree superstep, Tree state, Tree orderp, Trees input ) {
+ superstep = processBSP(subst(#<o_>,#<`(ns[0])>,getCache(#<i_>,input,superstep)));
+ Trees s = #[];
+ for ( int n: ns )
+ s = s.append(#<`n>);
+ return #<BSP(tuple(...s),`superstep,`state,`orderp,...input)>;
+ }
+
+ private static Tree mkBSPL ( int n, Tree superstep, Tree state, Tree orderp, Trees input ) {
+ superstep = processBSP(subst(#<o_>,#<`n>,getCache(#<i_>,input,superstep)));
+ return #<BSP(`n,`superstep,`state,`orderp,...input)>;
+ }
+
+ private static Tree mkBSP2 ( Tree superstep, Tree state, Tree orderp, Tree left, Tree right ) {
+ int rn = source_num++;
+ superstep = processBSP(subst(#<o_>,#<`rn>,getCache(#<i_>,left,
+ getCache(#<j_>,right,superstep))));
+ return #<BSP(`rn,`superstep,`state,`orderp,`left,`right)>;
+ }
+
+ /** construct a BSP plan from a physical plan
+ * @param e the physical plan
+ * @return the BSP plan
+ */
+ private static Tree mr2bsp ( Tree e ) {
+ match e {
+ case Aggregate(`acc,`zero,`S):
+ return #<Aggregate(`acc,`zero,`(mr2bsp(S)))>;
+ case cMap(`m,`S):
+ return mkBSP(#<lambda(tuple(cache,ms,k,peer),
+ setCache(cache,o_,cmap(`m,i_),
+ tuple(BAG(),tuple(),TRUE())))>,
+ #<tuple()>,
+ #<false>,
+ mr2bsp(S));
+ case AggregateMap(`m,`acc,`zero,`S):
+ return mkBSP(#<lambda(tuple(cache,ms,k,peer),
+ setCache(cache,o_,aggregate(`acc,`zero,cmap(`m,i_)),
+ tuple(BAG(),tuple(),TRUE())))>,
+ #<tuple()>,
+ #<false>,
+ mr2bsp(S));
+ case MapReduce(`m,`r,`S,`o):
+ return mkBSP(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(cmap(lambda(tuple(k,c),
+ bag(tuple(k,tuple(k,c)))),
+ cmap(`m,i_)),
+ false,
+ FALSE()),
+ setCache(cache,o_,`(o.equals(#<true>) // need to sort the result?
+ ? #<cmap(lambda(tuple(k,s),
+ cmap(lambda(x,bag(tuple(x,k))),
+ apply(`r,tuple(k,s)))),
+ groupBy(ms))>
+ : #<cmap(`r,groupBy(ms))>),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ o,
+ mr2bsp(S));
+ case MapAggregateReduce(`m,`r,`acc,`zero,`S,_):
+ return mkBSP(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(cmap(lambda(tuple(k,c),
+ bag(tuple(k,tuple(k,c)))),
+ cmap(`m,i_)),
+ false,
+ FALSE()),
+ setCache(cache,o_,aggregate(`acc,`zero,cmap(`r,groupBy(ms))),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ #<false>,
+ mr2bsp(S));
+ case MapCombineReduce(`m,`c,`r,`S,`o):
+ return mkBSP(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(cmap(lambda(tuple(k,s),
+ cmap(lambda(x,bag(tuple(k,tuple(k,x)))),
+ apply(`c,tuple(k,s)))),
+ groupBy(cmap(`m,i_))),
+ false,
+ FALSE()),
+ setCache(cache,o_,`(o.equals(#<true>) // need to sort the result?
+ ? #<cmap(lambda(tuple(k,s),
+ cmap(lambda(x,bag(tuple(x,k))),
+ apply(`r,tuple(k,s)))),
+ groupBy(ms))>
+ : #<cmap(`r,groupBy(ms))>),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ o,
+ mr2bsp(S));
+ case MapCombineReduce2(`mx,`my,`c,`r,`x,`y,`o):
+ return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(callM(union,`unionM,
+ cmap(lambda(tuple(kx,x),
+ bag(tuple(kx,tuple(kx,tuple(1,x))))),
+ cmap(`mx,i_)),
+ cmap(lambda(tuple(ky,y),
+ bag(tuple(ky,tuple(ky,tuple(2,y))))),
+ cmap(`my,j_))),
+ false,
+ FALSE()),
+ setCache(cache,o_,cmap(lambda(tuple(k,s),
+ cmap(lambda(tuple(kk,ss),
+ cmap(lambda(x,bag(tuple(kk,x))),
+ apply(`c,tuple(kk,ss)))),
+ groupBy(apply(`r,
+ tuple(cmap(lambda(tuple(kx,x),
+ if(callM(eq,`eqM,kx,1),
+ bag(x),
+ bag())),
+ s),
+ cmap(lambda(tuple(ky,y),
+ if(callM(eq,`eqM,ky,2),
+ bag(y),
+ bag())),
+ s)))))),
+ groupBy(ms)),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ o,
+ mr2bsp(x),
+ mr2bsp(y));
+ case MapReduce2(`mx,`my,`r,`x,`y,`o):
+ return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(callM(union,`unionM,
+ cmap(lambda(tuple(kx,x),
+ bag(tuple(kx,tuple(kx,tuple(1,x))))),
+ cmap(`mx,i_)),
+ cmap(lambda(tuple(ky,y),
+ bag(tuple(ky,tuple(ky,tuple(2,y))))),
+ cmap(`my,j_))),
+ false,
+ FALSE()),
+ setCache(cache,o_,cmap(lambda(tuple(k,s),
+ cmap(lambda(x,bag(`(o.equals(#<true>) ? #<tuple(x,k)> : #<x>))),
+ apply(`r,
+ tuple(cmap(lambda(tuple(kx,x),
+ if(callM(eq,`eqM,kx,1),
+ bag(x),
+ bag())),
+ s),
+ cmap(lambda(tuple(ky,y),
+ if(callM(eq,`eqM,ky,2),
+ bag(y),
+ bag())),
+ s))))),
+ groupBy(ms)),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ o,
+ mr2bsp(x),
+ mr2bsp(y));
+ case MapAggregateReduce2(`mx,`my,`r,`acc,`zero,`x,`y,_):
+ return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(callM(union,`unionM,
+ cmap(lambda(tuple(kx,x),
+ bag(tuple(kx,tuple(kx,tuple(1,x))))),
+ cmap(`mx,i_)),
+ cmap(lambda(tuple(ky,y),
+ bag(tuple(ky,tuple(ky,tuple(2,y))))),
+ cmap(`my,j_))),
+ false,
+ FALSE()),
+ setCache(cache,o_,aggregate(`acc,`zero,
+ cmap(lambda(tuple(k,s),
+ apply(`r,tuple(cmap(lambda(tuple(kx,x),
+ if(callM(eq,`eqM,kx,1),
+ bag(x),
+ bag())),
+ s),
+ cmap(lambda(tuple(ky,y),
+ if(callM(eq,`eqM,ky,2),
+ bag(y),
+ bag())),
+ s)))),
+ groupBy(ms))),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ #<false>,
+ mr2bsp(x),
+ mr2bsp(y));
+ case MapJoin(`mx,`my,`r,`x,`y):
+ return mr2bsp(#<MapReduce2(`mx,`my,
+ lambda(tuple(xs,ys),cmap(lambda(x,apply(`r,tuple(x,ys))),xs)),
+ `x,`y,false)>);
+ case MapAggregateJoin(`mx,`my,`r,`acc,`zero,`x,`y):
+ return mr2bsp(#<MapAggregateReduce2(`mx,`my,
+ lambda(tuple(xs,ys),cmap(lambda(x,apply(`r,tuple(x,ys))),xs)),
+ `acc,`zero,`x,`y)>);
+ case GroupByJoin(`kx,`ky,lambda(`vx,`gx),lambda(`vy,`gy),`mr,`c,`r,`x,`y,`o):
+ int n = 8;
+ int m = 8;
+ System.err.println("*** Using a groupBy join on a "+n+"*"+m+" grid of partitions");
+ Tree xkey = #<cmap(lambda(i,bag(tuple(call(plus,call(mod,call(hash_code,`gx),`m),call(times,`m,i)),
+ tuple(1,call(plus,call(mod,call(hash_code,`gx),`m),call(times,`m,i)),`vx)))),
+ range(0,`(n-1)))>;
+ Tree ykey = #<cmap(lambda(j,bag(tuple(call(plus,call(times,call(mod,call(hash_code,`gy),`n),`m),j),
+ tuple(2,call(plus,call(times,call(mod,call(hash_code,`gy),`n),`m),j),`vy)))),
+ range(0,`(m-1)))>;
+ type_inference(xkey);
+ type_inference(ykey);
+ xkey = PlanGeneration.makePlan(xkey);
+ ykey = PlanGeneration.makePlan(ykey);
+ return mkBSP2(#<lambda(tuple(cache,ms,map_step,peer),
+ if(map_step,
+ tuple(callM(union,`unionM,
+ cmap(lambda(`vx,`xkey),i_),
+ cmap(lambda(`vy,`ykey),j_)),
+ false,
+ FALSE()),
+ setCache(cache,o_,
+ mergeGroupByJoin(`kx,`ky,lambda(`vx,`gx),lambda(`vy,`gy),`mr,`c,`r,
+ cmap(lambda(tuple(kx,p,x),
+ if(callM(eq,`eqM,kx,1),bag(tuple(p,x)),bag())),ms),
+ cmap(lambda(tuple(ky,p,y),
+ if(callM(eq,`eqM,ky,2),bag(tuple(p,y)),bag())),ms),`o),
+ tuple(BAG(),false,TRUE()))))>,
+ #<true>,
+ o,
+ mr2bsp(x),
+ mr2bsp(y));
+ case CrossProduct(`mx,`my,`r,`x,`y):
+ return mkBSP(#<lambda(tuple(cache,ms,ys,peer),
+ tuple(BAG(),
+ setCache(cache,o_,
+ cmap(lambda(x,
+ cmap(lambda(y,apply(`r,tuple(x,y))),
+ cmap(`my,ys))),
+ cmap(`mx,i_)),
+ tuple()),
+ TRUE()))>,
+ #<Collect(`(mr2bsp(y)))>,
+ #<false>,
+ mr2bsp(x));
+ case CrossAggregateProduct(`mx,`my,`r,`acc,`zero,`x,`y):
+ return mkBSP(#<lambda(tuple(cache,ms,ys,peer),
+ tuple(BAG(),
+ setCache(cache,o_,
+ aggregate(`acc,`zero,
+ cmap(lambda(x,
+ cmap(lambda(y,apply(`r,tuple(x,y))),
+ cmap(`my,ys))),
+ cmap(`mx,i_)),
+ tuple())),
+ TRUE()))>,
+ #<Collect(`(mr2bsp(y)))>,
+ #<false>,
+ mr2bsp(x));
+ case Repeat(lambda(`v,`b),`ds,`max):
+ Tree step = bspSimplify(mr2bsp(b));
+ int step_cache_num = source_num(step);
+ match step {
+ case BSP(`n,`s,`k0,_,...as):
+ Tree ds_source = mr2bsp(ds);
+ int ds_cache_num = source_num(ds_source);
+ ds_source = set_cache_num(ds_source,step_cache_num);
+ // the initial values of all data sources
+ Trees sources = #[`ds_source];
+ Tree step_input = #<0>;
+ for ( Tree x: as )
+ match x {
+ case BSPSource(`j,`y):
+ if (y.equals(v))
+ step_input = j;
+ else sources = sources.append(x);
+ case _: sources = sources.append(x);
+ };
+ s = subst_getCache_num((int)((LongLeaf)step_input).value(),step_cache_num,s);
+ return mkBSPL(step_cache_num,
+ #<lambda(tuple(cache,ms,tuple(k,steps),peer),
+ let(tuple(ts,kk,step_end),
+ apply(`s,tuple(cache,ms,k,peer)),
+ if(step_end, // end of repeat step
+ setCache(cache,`step_cache_num,
+ map(lambda(tuple(x,bb),x),
+ getCache(cache,`step_cache_num)),
+ tuple(bag(),
+ tuple(`k0,callM(plus,`plusM,steps,1)),
+ if(callM(gt,`gtM,steps,`max),
+ TRUE(), // if # of steps <= limit, exit
+ callM(not,`notM, // ... else check the stopping condition
+ aggregate(lambda(tuple(x,y),callM(or,`orM,x,y)),
+ false,
+ map(lambda(tuple(x,bb),bb),
+ getCache(cache,`step_cache_num))))))),
+ tuple(ts,tuple(kk,steps),FALSE()))))>,
+ #<tuple(`k0,1)>,
+ #<false>,
+ sources);
+ case `x: throw new Error("Cannot compile the repeat function: "+x);
+ }
+ // when the repeat variable is in memory
+ case repeat(lambda(`v,`b),`ds,`max):
+ Tree step = bspSimplify(mr2bsp(b));
+ int step_cache_num = source_num(step);
+ step = subst(v,#<map(lambda(tuple(x,b),x),getCache(cache,`step_cache_num))>,step);
+ match step {
+ case BSP(`n,`s,`k0,_,...as):
+ // the initial values of all data sources
+ Trees sources = #[];
+ for ( Tree x: as )
+ if (!x.equals(v))
+ sources = sources.append(x);
+ Tree res = mkBSPL(#<lambda(tuple(cache,ms,tuple(k,steps,firstp,S),peer),
+ let(ignore,if(firstp,setCache(cache,`step_cache_num,S,0),0),
+ let(tuple(ts,kk,step_end),
+ apply(`s,tuple(cache,ms,k,peer)),
+ if(step_end, // end of repeat step
+ setCache(cache,o_,map(lambda(tuple(x,b),x),getCache(cache,`step_cache_num)),
+ setCache(cache,`step_cache_num,
+ distribute(peer,getCache(cache,`step_cache_num)),
+ tuple(bag(),
+ tuple(`k0,callM(plus,`plusM,steps,1),false,bag()),
+ if(callM(gt,`gtM,steps,`max),
+ TRUE(), // if # of steps > limit, exit
+ callM(not,`notM, // ... else check the stopping condition
+ aggregate(lambda(tuple(x,y),callM(or,`orM,x,y)),
+ false,
+ map(lambda(tuple(x,bb),bb),
+ getCache(cache,`step_cache_num)))))))),
+ tuple(ts,tuple(kk,steps,false,bag()),FALSE())))))>,
+ #<tuple(`k0,1,true,map(lambda(x,tuple(x,false)),`ds))>,
+ #<false>,
+ sources);
+ return #<Collect(`res)>; // must return a memory bag
+ case `x: throw new Error("Cannot compile the repeat function: "+x);
+ }
+ case Closure(lambda(`v,`b),`ds,`max):
+ Tree step = bspSimplify(mr2bsp(b));
+ int step_cache_num = source_num(step);
+ match step {
+ case BSP(`n,`s,`k0,_,...as):
+ Tree ds_source = mr2bsp(ds);
+ int ds_cache_num = source_num(ds_source);
+ ds_source = set_cache_num(ds_source,step_cache_num);
+ // the initial values of all data sources
+ Trees sources = #[`ds_source];
+ Tree step_input = #<0>;
+ for ( Tree x: as )
+ match x {
+ case BSPSource(`j,`y):
+ if (y.equals(v))
+ step_input = j;
+ else sources = sources.append(x);
+ case _: sources = sources.append(x);
+ };
+ s = subst_getCache_num((int)((LongLeaf)step_input).value(),step_cache_num,s);
+ return mkBSPL(step_cache_num,
+ #<lambda(tuple(cache,ms,tuple(k,steps,len),peer),
+ let(tuple(ts,kk,step_end),
+ apply(`s,tuple(cache,ms,k,peer)),
+ if(step_end, // end of repeat step
+ let(newLen,callM(count,`countM,getCache(cache,`step_cache_num)),
+ tuple(bag(),
+ tuple(`k0,callM(plus,`plusM,steps,1),newLen),
+ if(callM(gt,`gtM,steps,`max),
+ TRUE(), // if # of steps > limit, exit
+ // ... else check if the new size is the same as the old size
+ callM(geq,`geqM,len,newLen)))),
+ tuple(ts,tuple(kk,steps,len),FALSE()))))>,
+ #<tuple(`k0,1,callM(coerce,`coerceM,0,4))>,
+ #<false>,
+ sources);
+ case `x: throw new Error("Cannot compile the closure function: "+x);
+ }
+ case Loop(lambda(tuple(...vs),tuple(...bs)),tuple(...ss),`max):
+ Tree[] steps = new Tree[vs.length()];
+ Tree[] inits = new Tree[vs.length()];
+ int[] cache_num = new int[vs.length()];
+ Tree[] k = new Tree[vs.length()];
+ Trees sources = #[];
+ Trees all_cache = #[];
+ for ( int i = 0; i < vs.length(); i++ ) {
+ inits[i] = mr2bsp(ss.nth(i));
+ sources = sources.append(inits[i]);
+ cache_num[i] = source_num(inits[i]);
+ all_cache = all_cache.append(#<getCache(cache,`(cache_num[i]))>);
+ };
+ for ( int i = 0; i < vs.length(); i++ )
+ match bspSimplify(mr2bsp(bs.nth(i))) {
+ case BSP(`n,`s,`k0,_,...as):
+ steps[i] = subst_setCache_num((int)((LongLeaf)n).value(),cache_num[i],s);
+ k[i] = k0;
+ loop: for ( Tree x: as )
+ match x {
+ case BSPSource(`m,`w):
+ if (!vs.member(w))
+ fail;
+ for ( int j = 0; j < vs.length(); j++ )
+ if (w.equals(vs.nth(j)))
+ steps[i] = subst_getCache_num((int)((LongLeaf)m).value(),cache_num[j],steps[i]);
+ case BSPSource(`n1,`d1):
+ for ( Tree y: sources )
+ match y {
+ case BSPSource(`n2,`d2):
+ if (d1.equals(d2)) {
+ steps[i] = subst_getCache_num((int)((LongLeaf)n1).value(),
+ (int)((LongLeaf)n2).value(),steps[i]);
+ continue loop;
+ }
+ };
+ sources = sources.append(x);
+ case _: sources = sources.append(x);
+ };
+ };
+ Tree code = #<tuple(BAG(),tuple(`(k[0]),0,1),TRUE())>;
+ for ( int i = 0; i < vs.length(); i++ )
+ code = #<if(callM(eq,`eqM,i,`i),
+ let(tuple(ts,kk,step_end),
+ apply(`(steps[i]),tuple(cache,ms,k,peer)),
+ if(step_end, // end of repeat step
+ `((i+1 < vs.length())
+ ? #<tuple(bag(),tuple(`(k[i+1]),`(i+1),steps),FALSE())>
+ : #<tuple(bag(),
+ tuple(`(k[0]),0,callM(plus,`plusM,steps,1)),
+ if(callM(gt,`gtM,steps,`max), // if # of steps > limit, exit
+ TRUE(),
+ FALSE()))>),
+ tuple(ts,tuple(kk,i,steps),FALSE()))),
+ `code)>;
+ return mkBSPL(cache_num,
+ #<lambda(tuple(cache,ms,tuple(k,i,steps),peer),`code)>,
+ #<tuple(`(k[0]),0,2)>,
+ #<false>,
+ sources);
+ case `f(...as):
+ Trees bs = #[];
+ for ( Tree a: as )
+ bs = bs.append(mr2bsp(a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ /** simplify the BSP plan by fusing consequtive BSP plans */
+ private static Tree bspSimplify ( Tree e ) {
+ match e {
+ case BSP(`n,`s2,`k2,`o,...r,BSP(_,`s1,`k1,_,...s),...t):
+ Trees ys = r.append(t);
+ Trees ns = #[];
+ loop: for ( Tree x: s )
+ match x {
+ case BSPSource(`n1,`d1):
+ for ( Tree y: ys )
+ match y {
+ case BSPSource(`n2,`d2):
+ if (d1.equals(d2)) {
+ s1 = subst_getCache_num((int)((LongLeaf)n1).value(),
+ (int)((LongLeaf)n2).value(),s1);
+ continue loop;
+ }
+ };
+ ns = ns.append(x);
+ case _: ns = ns.append(x);
+ };
+ return bspSimplify(mkBSPL((int)((LongLeaf)n).value(),
+ #<lambda(tuple(cache,ms,tuple(first,k),peer),
+ if(first,
+ let(tuple(ts,kk,b),apply(`s1,tuple(cache,ms,k,peer)),
+ let(exit,
+ synchronize(peer,b), // poll all peers: do you want to exit?
+ // all peers must aggree to exit the inner BSP
+ // and proceed to the outer BSP
+ tuple(ts,tuple(callM(not,`notM,exit),
+ if(exit,`k2,kk)),FALSE()))),
+ let(tuple(ts,kk,bb),apply(`s2,tuple(cache,ms,k,peer)),
+ tuple(ts,tuple(false,kk),bb))))>,
+ #<tuple(true,`k1)>,
+ o,
+ #[...r,...ns,...t]));
+ case `f(...as):
+ Trees bs = #[];
+ for ( Tree a: as )
+ bs = bs.append(bspSimplify(a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ private static Tree post_simplify_plan ( Tree e ) {
+ match e {
+ case setCache(`cache,`a,`v,`ret):
+ return post_simplify_plan(#<setNth(`cache,`a,materialize(`v),`ret)>);
+ case getCache(`cache,`a,...as):
+ Tree z = #<nth(`cache,`a)>;
+ for ( Tree x: as )
+ z = #<callM(union,`unionM,`z,nth(`cache,`x))>;
+ return z;
+ case `f(...as):
+ Trees bs = #[];
+ for (Tree a: as)
+ bs = bs.append(post_simplify_plan(a));
+ return #<`f(...bs)>;
+ };
+ return e;
+ }
+
+ /** construct and simplify the BSP plan from a physical plan
+ * @param plan the physical plan
+ * @return the BSP plan
+ */
+ public static Tree constructBSPplan ( Tree plan ) {
+ return post_simplify_plan(bspSimplify(mr2bsp(preprocess(plan))));
+ }
+}