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:36 UTC
[11/26] MRQL-32: Refactoring directory structure for Eclipse
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/MapReduce/MapReducePlan.java
----------------------------------------------------------------------
diff --git a/src/main/java/MapReduce/MapReducePlan.java b/src/main/java/MapReduce/MapReducePlan.java
deleted file mode 100644
index ae243cd..0000000
--- a/src/main/java/MapReduce/MapReducePlan.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/**
- * 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 org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.mapreduce.*;
-
-
-/** A superclass for Hadoop MapReduce physical operators (files *Operator.java) */
-public class MapReducePlan extends Plan {
-
- MapReducePlan () {}
-
- /** find the number of records in the hadoop MapReduce job output */
- public final static long outputRecords ( Job job ) throws Exception {
- CounterGroup cg = job.getCounters().getGroup("org.apache.hadoop.mapred.Task$Counter");
- long rc = cg.findCounter("REDUCE_OUTPUT_RECORDS").getValue();
- if (rc == 0)
- return cg.findCounter("MAP_OUTPUT_RECORDS").getValue();
- return rc;
- }
-
- /** The Aggregate physical operator
- * @param acc_fnc the accumulator function from (T,T) to T
- * @param zero the zero element of type T
- * @param S the dataset that contains the bag of values {T}
- * @return the aggregation result of type T
- */
- public final static MRData aggregate ( final Tree acc_fnc,
- final Tree zero,
- final DataSet S ) throws Exception {
- MRData res = Interpreter.evalE(zero);
- Function accumulator = functional_argument(Plan.conf,acc_fnc);
- Tuple pair = new Tuple(2);
- for ( DataSource s: S.source )
- if (s.inputFormat != MapReduceBinaryInputFormat.class) {
- pair.set(0,res);
- pair.set(1,aggregate(acc_fnc,zero,
- MapOperation.cMap(Interpreter.identity_mapper,acc_fnc,zero,
- new DataSet(s,0,0),"-")));
- res = accumulator.eval(pair);
- } else {
- Path path = new Path(s.path);
- final FileSystem fs = path.getFileSystem(conf);
- final FileStatus[] ds
- = fs.listStatus(path,
- new PathFilter () {
- public boolean accept ( Path path ) {
- return !path.getName().startsWith("_");
- }
- });
- MRContainer key = new MRContainer(new MR_int(0));
- MRContainer value = new MRContainer(new MR_int(0));
- for ( int i = 0; i < ds.length; i++ ) {
- SequenceFile.Reader reader = new SequenceFile.Reader(fs,ds[i].getPath(),conf);
- while (reader.next(key,value)) {
- pair.set(0,res);
- pair.set(1,value.data());
- res = accumulator.eval(pair);
- };
- reader.close();
- }
- };
- return res;
- }
-
- /** The repeat physical operator. It repeats the loop until all values satisfy
- * the termination condition (when dataset.counter == 0)
- * @param loop the function from DataSet to DataSet to be repeated
- * @param init the initial input DataSet for the loop
- * @param max_num max number of repetitions
- * @return the resulting DataSet from the loop
- */
- public final static DataSet repeat ( Function loop, DataSet init, int max_num ) {
- MR_dataset s = new MR_dataset(init);
- int i = 0;
- do {
- s.dataset = ((MR_dataset)loop.eval(s)).dataset;
- i++;
- System.err.println("*** Repeat #"+i+": "+s.dataset.counter+" true results");
- } while (s.dataset.counter != 0 && i < max_num);
- return s.dataset;
- }
-
- /** The closure physical operator. It repeats the loop until the size
- * of the new DataSet is equal to that of the previous DataSet
- * @param loop the function from DataSet to DataSet to be repeated
- * @param init the initial input DataSet for the loop
- * @param max_num max number of repetitions
- * @return the resulting DataSet from the loop
- */
- public final static DataSet closure ( Function loop, DataSet init, int max_num ) {
- MR_dataset s = new MR_dataset(init);
- int i = 0;
- long n = 0;
- long old = 0;
- do {
- s.dataset = ((MR_dataset)loop.eval(s)).dataset;
- i++;
- System.err.println("*** Repeat #"+i+": "+(s.dataset.records-n)+" new records");
- old = n;
- n = s.dataset.records;
- } while (old < n && i < max_num);
- return s.dataset;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/MultipleInputs/DelegatingInputFormat.java
----------------------------------------------------------------------
diff --git a/src/main/java/MultipleInputs/DelegatingInputFormat.java b/src/main/java/MultipleInputs/DelegatingInputFormat.java
deleted file mode 100644
index 0d9b8f9..0000000
--- a/src/main/java/MultipleInputs/DelegatingInputFormat.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/**
- * 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.hadoop.mapreduce.lib.input;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.util.ReflectionUtils;
-
-/**
- * An {@link InputFormat} that delegates behavior of paths to multiple other
- * InputFormats.
- *
- * @see MultipleInputs#addInputPath(Job, Path, Class, Class)
- */
-public class DelegatingInputFormat<K, V> extends InputFormat<K, V> {
-
- @SuppressWarnings("unchecked")
- public List<InputSplit> getSplits(JobContext job)
- throws IOException, InterruptedException {
- Configuration conf = job.getConfiguration();
- Job jobCopy =new Job(conf);
- List<InputSplit> splits = new ArrayList<InputSplit>();
- Map<Path, InputFormat> formatMap =
- MultipleInputs.getInputFormatMap(job);
- Map<Path, Class<? extends Mapper>> mapperMap = MultipleInputs
- .getMapperTypeMap(job);
- Map<Class<? extends InputFormat>, List<Path>> formatPaths
- = new HashMap<Class<? extends InputFormat>, List<Path>>();
-
- // First, build a map of InputFormats to Paths
- for (Entry<Path, InputFormat> entry : formatMap.entrySet()) {
- if (!formatPaths.containsKey(entry.getValue().getClass())) {
- formatPaths.put(entry.getValue().getClass(), new LinkedList<Path>());
- }
-
- formatPaths.get(entry.getValue().getClass()).add(entry.getKey());
- }
-
- for (Entry<Class<? extends InputFormat>, List<Path>> formatEntry :
- formatPaths.entrySet()) {
- Class<? extends InputFormat> formatClass = formatEntry.getKey();
- InputFormat format = (InputFormat) ReflectionUtils.newInstance(
- formatClass, conf);
- List<Path> paths = formatEntry.getValue();
-
- Map<Class<? extends Mapper>, List<Path>> mapperPaths
- = new HashMap<Class<? extends Mapper>, List<Path>>();
-
- // Now, for each set of paths that have a common InputFormat, build
- // a map of Mappers to the paths they're used for
- for (Path path : paths) {
- Class<? extends Mapper> mapperClass = mapperMap.get(path);
- if (!mapperPaths.containsKey(mapperClass)) {
- mapperPaths.put(mapperClass, new LinkedList<Path>());
- }
-
- mapperPaths.get(mapperClass).add(path);
- }
-
- // Now each set of paths that has a common InputFormat and Mapper can
- // be added to the same job, and split together.
- for (Entry<Class<? extends Mapper>, List<Path>> mapEntry :
- mapperPaths.entrySet()) {
- paths = mapEntry.getValue();
- Class<? extends Mapper> mapperClass = mapEntry.getKey();
-
- if (mapperClass == null) {
- try {
- mapperClass = job.getMapperClass();
- } catch (ClassNotFoundException e) {
- throw new IOException("Mapper class is not found", e);
- }
- }
-
- FileInputFormat.setInputPaths(jobCopy, paths.toArray(new Path[paths
- .size()]));
-
- // Get splits for each input path and tag with InputFormat
- // and Mapper types by wrapping in a TaggedInputSplit.
- List<InputSplit> pathSplits = format.getSplits(jobCopy);
- for (InputSplit pathSplit : pathSplits) {
- splits.add(new TaggedInputSplit(pathSplit, conf, format.getClass(),
- mapperClass));
- }
- }
- }
-
- return splits;
- }
-
- @Override
- public RecordReader<K, V> createRecordReader(InputSplit split,
- TaskAttemptContext context) throws IOException, InterruptedException {
- return new DelegatingRecordReader<K, V>(split, context);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/MultipleInputs/DelegatingMapper.java
----------------------------------------------------------------------
diff --git a/src/main/java/MultipleInputs/DelegatingMapper.java b/src/main/java/MultipleInputs/DelegatingMapper.java
deleted file mode 100644
index d72d2b5..0000000
--- a/src/main/java/MultipleInputs/DelegatingMapper.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/**
- * 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.hadoop.mapreduce.lib.input;
-
-import java.io.IOException;
-
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.util.ReflectionUtils;
-
-/**
- * An {@link Mapper} that delegates behavior of paths to multiple other
- * mappers.
- *
- * @see MultipleInputs#addInputPath(Job, Path, Class, Class)
- */
-public class DelegatingMapper<K1, V1, K2, V2> extends Mapper<K1, V1, K2, V2> {
-
- private Mapper<K1, V1, K2, V2> mapper;
-
- @SuppressWarnings("unchecked")
- protected void setup(Context context)
- throws IOException, InterruptedException {
- // Find the Mapper from the TaggedInputSplit.
- TaggedInputSplit inputSplit = (TaggedInputSplit) context.getInputSplit();
- mapper = (Mapper<K1, V1, K2, V2>) ReflectionUtils.newInstance(inputSplit
- .getMapperClass(), context.getConfiguration());
-
- }
-
- @SuppressWarnings("unchecked")
- public void run(Context context)
- throws IOException, InterruptedException {
- setup(context);
- mapper.run(context);
- cleanup(context);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/MultipleInputs/DelegatingRecordReader.java
----------------------------------------------------------------------
diff --git a/src/main/java/MultipleInputs/DelegatingRecordReader.java b/src/main/java/MultipleInputs/DelegatingRecordReader.java
deleted file mode 100644
index f0d060e..0000000
--- a/src/main/java/MultipleInputs/DelegatingRecordReader.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/**
- * 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.hadoop.mapreduce.lib.input;
-
-import java.io.IOException;
-
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.util.ReflectionUtils;
-
-/**
- * This is a delegating RecordReader, which delegates the functionality to the
- * underlying record reader in {@link TaggedInputSplit}
- */
-public class DelegatingRecordReader<K, V> extends RecordReader<K, V> {
- RecordReader<K, V> originalRR;
-
- /**
- * Constructs the DelegatingRecordReader.
- *
- * @param split TaggegInputSplit object
- * @param context TaskAttemptContext object
- *
- * @throws IOException
- * @throws InterruptedException
- */
- @SuppressWarnings("unchecked")
- public DelegatingRecordReader(InputSplit split, TaskAttemptContext context)
- throws IOException, InterruptedException {
- // Find the InputFormat and then the RecordReader from the
- // TaggedInputSplit.
- TaggedInputSplit taggedInputSplit = (TaggedInputSplit) split;
- InputFormat<K, V> inputFormat = (InputFormat<K, V>) ReflectionUtils
- .newInstance(taggedInputSplit.getInputFormatClass(), context
- .getConfiguration());
- originalRR = inputFormat.createRecordReader(taggedInputSplit
- .getInputSplit(), context);
- }
-
- @Override
- public void close() throws IOException {
- originalRR.close();
- }
-
- @Override
- public K getCurrentKey() throws IOException, InterruptedException {
- return originalRR.getCurrentKey();
- }
-
- @Override
- public V getCurrentValue() throws IOException, InterruptedException {
- return originalRR.getCurrentValue();
- }
-
- @Override
- public float getProgress() throws IOException, InterruptedException {
- return originalRR.getProgress();
- }
-
- @Override
- public void initialize(InputSplit split, TaskAttemptContext context)
- throws IOException, InterruptedException {
- originalRR.initialize(((TaggedInputSplit) split).getInputSplit(), context);
- }
-
- @Override
- public boolean nextKeyValue() throws IOException, InterruptedException {
- return originalRR.nextKeyValue();
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/MultipleInputs/MultipleInputs.java
----------------------------------------------------------------------
diff --git a/src/main/java/MultipleInputs/MultipleInputs.java b/src/main/java/MultipleInputs/MultipleInputs.java
deleted file mode 100644
index 9140512..0000000
--- a/src/main/java/MultipleInputs/MultipleInputs.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/**
- * 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.hadoop.mapreduce.lib.input;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.util.ReflectionUtils;
-
-/**
- * This class supports MapReduce jobs that have multiple input paths with
- * a different {@link InputFormat} and {@link Mapper} for each path
- */
-public class MultipleInputs {
- /**
- * Add a {@link Path} with a custom {@link InputFormat} to the list of
- * inputs for the map-reduce job.
- *
- * @param job The {@link Job}
- * @param path {@link Path} to be added to the list of inputs for the job
- * @param inputFormatClass {@link InputFormat} class to use for this path
- */
- @SuppressWarnings("unchecked")
- public static void addInputPath(Job job, Path path,
- Class<? extends InputFormat> inputFormatClass) {
- String inputFormatMapping = path.toString() + ";"
- + inputFormatClass.getName();
- Configuration conf = job.getConfiguration();
- String inputFormats = conf.get("mapred.input.dir.formats");
- conf.set("mapred.input.dir.formats",
- inputFormats == null ? inputFormatMapping : inputFormats + ","
- + inputFormatMapping);
-
- job.setInputFormatClass(DelegatingInputFormat.class);
- }
-
- /**
- * Add a {@link Path} with a custom {@link InputFormat} and
- * {@link Mapper} to the list of inputs for the map-reduce job.
- *
- * @param job The {@link Job}
- * @param path {@link Path} to be added to the list of inputs for the job
- * @param inputFormatClass {@link InputFormat} class to use for this path
- * @param mapperClass {@link Mapper} class to use for this path
- */
- @SuppressWarnings("unchecked")
- public static void addInputPath(Job job, Path path,
- Class<? extends InputFormat> inputFormatClass,
- Class<? extends Mapper> mapperClass) {
-
- addInputPath(job, path, inputFormatClass);
- Configuration conf = job.getConfiguration();
- String mapperMapping = path.toString() + ";" + mapperClass.getName();
- String mappers = conf.get("mapred.input.dir.mappers");
- conf.set("mapred.input.dir.mappers", mappers == null ? mapperMapping
- : mappers + "," + mapperMapping);
-
- job.setMapperClass(DelegatingMapper.class);
- }
-
- /**
- * Retrieves a map of {@link Path}s to the {@link InputFormat} class
- * that should be used for them.
- *
- * @param job The {@link JobContext}
- * @see #addInputPath(JobConf, Path, Class)
- * @return A map of paths to inputformats for the job
- */
- @SuppressWarnings("unchecked")
- static Map<Path, InputFormat> getInputFormatMap(JobContext job) {
- Map<Path, InputFormat> m = new HashMap<Path, InputFormat>();
- Configuration conf = job.getConfiguration();
- String[] pathMappings = conf.get("mapred.input.dir.formats").split(",");
- for (String pathMapping : pathMappings) {
- String[] split = pathMapping.split(";");
- InputFormat inputFormat;
- try {
- inputFormat = (InputFormat) ReflectionUtils.newInstance(conf
- .getClassByName(split[1]), conf);
- } catch (ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
- m.put(new Path(split[0]), inputFormat);
- }
- return m;
- }
-
- /**
- * Retrieves a map of {@link Path}s to the {@link Mapper} class that
- * should be used for them.
- *
- * @param job The {@link JobContext}
- * @see #addInputPath(JobConf, Path, Class, Class)
- * @return A map of paths to mappers for the job
- */
- @SuppressWarnings("unchecked")
- static Map<Path, Class<? extends Mapper>>
- getMapperTypeMap(JobContext job) {
- Configuration conf = job.getConfiguration();
- if (conf.get("mapred.input.dir.mappers") == null) {
- return Collections.emptyMap();
- }
- Map<Path, Class<? extends Mapper>> m =
- new HashMap<Path, Class<? extends Mapper>>();
- String[] pathMappings = conf.get("mapred.input.dir.mappers").split(",");
- for (String pathMapping : pathMappings) {
- String[] split = pathMapping.split(";");
- Class<? extends Mapper> mapClass;
- try {
- mapClass =
- (Class<? extends Mapper>) conf.getClassByName(split[1]);
- } catch (ClassNotFoundException e) {
- throw new RuntimeException(e);
- }
- m.put(new Path(split[0]), mapClass);
- }
- return m;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/MultipleInputs/TaggedInputSplit.java
----------------------------------------------------------------------
diff --git a/src/main/java/MultipleInputs/TaggedInputSplit.java b/src/main/java/MultipleInputs/TaggedInputSplit.java
deleted file mode 100644
index 68bb789..0000000
--- a/src/main/java/MultipleInputs/TaggedInputSplit.java
+++ /dev/null
@@ -1,159 +0,0 @@
-/**
- * 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.hadoop.mapreduce.lib.input;
-
-import java.io.DataInput;
-import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-import org.apache.hadoop.conf.Configurable;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.Writable;
-import org.apache.hadoop.io.serializer.Deserializer;
-import org.apache.hadoop.io.serializer.SerializationFactory;
-import org.apache.hadoop.io.serializer.Serializer;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.util.ReflectionUtils;
-
-/**
- * An {@link InputSplit} that tags another InputSplit with extra data for use
- * by {@link DelegatingInputFormat}s and {@link DelegatingMapper}s.
- */
-class TaggedInputSplit extends InputSplit implements Configurable, Writable {
-
- private Class<? extends InputSplit> inputSplitClass;
-
- private InputSplit inputSplit;
-
- @SuppressWarnings("unchecked")
- private Class<? extends InputFormat> inputFormatClass;
-
- @SuppressWarnings("unchecked")
- private Class<? extends Mapper> mapperClass;
-
- private Configuration conf;
-
- public TaggedInputSplit() {
- // Default constructor.
- }
-
- /**
- * Creates a new TaggedInputSplit.
- *
- * @param inputSplit The InputSplit to be tagged
- * @param conf The configuration to use
- * @param inputFormatClass The InputFormat class to use for this job
- * @param mapperClass The Mapper class to use for this job
- */
- @SuppressWarnings("unchecked")
- public TaggedInputSplit(InputSplit inputSplit, Configuration conf,
- Class<? extends InputFormat> inputFormatClass,
- Class<? extends Mapper> mapperClass) {
- this.inputSplitClass = inputSplit.getClass();
- this.inputSplit = inputSplit;
- this.conf = conf;
- this.inputFormatClass = inputFormatClass;
- this.mapperClass = mapperClass;
- }
-
- /**
- * Retrieves the original InputSplit.
- *
- * @return The InputSplit that was tagged
- */
- public InputSplit getInputSplit() {
- return inputSplit;
- }
-
- /**
- * Retrieves the InputFormat class to use for this split.
- *
- * @return The InputFormat class to use
- */
- @SuppressWarnings("unchecked")
- public Class<? extends InputFormat> getInputFormatClass() {
- return inputFormatClass;
- }
-
- /**
- * Retrieves the Mapper class to use for this split.
- *
- * @return The Mapper class to use
- */
- @SuppressWarnings("unchecked")
- public Class<? extends Mapper> getMapperClass() {
- return mapperClass;
- }
-
- public long getLength() throws IOException, InterruptedException {
- return inputSplit.getLength();
- }
-
- public String[] getLocations() throws IOException, InterruptedException {
- return inputSplit.getLocations();
- }
-
- @SuppressWarnings("unchecked")
- public void readFields(DataInput in) throws IOException {
- inputSplitClass = (Class<? extends InputSplit>) readClass(in);
- inputFormatClass = (Class<? extends InputFormat<?, ?>>) readClass(in);
- mapperClass = (Class<? extends Mapper<?, ?, ?, ?>>) readClass(in);
- inputSplit = (InputSplit) ReflectionUtils
- .newInstance(inputSplitClass, conf);
- SerializationFactory factory = new SerializationFactory(conf);
- Deserializer deserializer = factory.getDeserializer(inputSplitClass);
- deserializer.open((DataInputStream)in);
- inputSplit = (InputSplit)deserializer.deserialize(inputSplit);
- }
-
- private Class<?> readClass(DataInput in) throws IOException {
- String className = Text.readString(in);
- try {
- return conf.getClassByName(className);
- } catch (ClassNotFoundException e) {
- throw new RuntimeException("readObject can't find class", e);
- }
- }
-
- @SuppressWarnings("unchecked")
- public void write(DataOutput out) throws IOException {
- Text.writeString(out, inputSplitClass.getName());
- Text.writeString(out, inputFormatClass.getName());
- Text.writeString(out, mapperClass.getName());
- SerializationFactory factory = new SerializationFactory(conf);
- Serializer serializer =
- factory.getSerializer(inputSplitClass);
- serializer.open((DataOutputStream)out);
- serializer.serialize(inputSplit);
- }
-
- public Configuration getConf() {
- return conf;
- }
-
- public void setConf(Configuration conf) {
- this.conf = conf;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/1adaa71c/src/main/java/core/AlgebraicOptimization.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/AlgebraicOptimization.gen b/src/main/java/core/AlgebraicOptimization.gen
deleted file mode 100644
index d574e20..0000000
--- a/src/main/java/core/AlgebraicOptimization.gen
+++ /dev/null
@@ -1,315 +0,0 @@
-/**
- * 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/src/main/java/core/BSPTranslator.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/BSPTranslator.gen b/src/main/java/core/BSPTranslator.gen
deleted file mode 100644
index a2af7a6..0000000
--- a/src/main/java/core/BSPTranslator.gen
+++ /dev/null
@@ -1,753 +0,0 @@
-/**
- * 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))));
- }
-}