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 2013/08/27 14:27:45 UTC
[07/21] MRQL-2: Mavenize MRQL
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MRData.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MRData.java b/src/main/java/core/MRData.java
new file mode 100644
index 0000000..d4343ed
--- /dev/null
+++ b/src/main/java/core/MRData.java
@@ -0,0 +1,38 @@
+/**
+ * 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.hadoop.io.WritableComparable;
+import java.io.*;
+
+
+/** All MRQL data are encoded as MRData (similar to AVRO form) */
+public abstract class MRData implements WritableComparable<MRData>, Serializable {
+ abstract public void materializeAll ();
+
+ /* doesn't work
+ private void writeObject ( ObjectOutputStream out ) throws IOException {
+ write(out);
+ }
+
+ private void readObject ( ObjectInputStream in ) throws IOException, ClassNotFoundException {
+ byte n = in.readByte();
+ readFields(in);
+ }
+ */
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MRQL.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MRQL.java b/src/main/java/core/MRQL.java
new file mode 100644
index 0000000..fa39db5
--- /dev/null
+++ b/src/main/java/core/MRQL.java
@@ -0,0 +1,58 @@
+/**
+ * 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 java.io.*;
+
+
+/** API for dynamic MRQL queries */
+final public class MRQL extends Interpreter {
+
+ /** evaluate an MRQL query in a string
+ * @param query a string that contains an MRQL query
+ * @return the evaluation result
+ */
+ public static MRData query ( String query ) {
+ evaluate("store tt := "+query+";");
+ return variable_lookup("tt",global_env);
+ }
+
+ /** evaluate MRQL statments in a string
+ * @param command a string that contains MRQL commands separated by ;
+ */
+ public static void evaluate ( String command ) {
+ try {
+ MRQLParser parser = new MRQLParser();
+ parser.setScanner(new MRQLLex(new StringReader(command)));
+ MRQLLex.reset();
+ parser.parse();
+ } catch (Exception x) {
+ x.printStackTrace();
+ throw new Error(x);
+ }
+ }
+
+ /** clean up the MRQL workspace */
+ public static void clean () {
+ try {
+ Plan.clean();
+ } catch (IOException ex) {
+ throw new Error("Failed to clean-up temporary files");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_bool.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_bool.java b/src/main/java/core/MR_bool.java
new file mode 100644
index 0000000..7f14d47
--- /dev/null
+++ b/src/main/java/core/MR_bool.java
@@ -0,0 +1,72 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+/** a container for boolean values */
+final public class MR_bool extends MRData {
+ private boolean value;
+
+ public MR_bool ( boolean i ) { value = i; }
+
+ public void materializeAll () {};
+
+ public boolean get () { return value; }
+
+ public void set ( boolean v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.BOOLEAN);
+ out.writeBoolean(value);
+ }
+
+ final public static MR_bool read ( DataInput in ) throws IOException {
+ return new MR_bool(in.readBoolean());
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = in.readBoolean();
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_bool);
+ return (value == ((MR_bool) x).value) ? 0 : (value ? 1 : -1);
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ size[0] = 2;
+ boolean bx = x[xs] > 0;
+ boolean by = y[ys] > 0;
+ return (bx == by) ? 0 : (bx ? 1 : -1);
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_bool && ((MR_bool)x).value==value;
+ }
+
+ public int hashCode () { return value ? 0 : 1; }
+
+ public String toString () {
+ return Boolean.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_byte.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_byte.java b/src/main/java/core/MR_byte.java
new file mode 100644
index 0000000..38262e9
--- /dev/null
+++ b/src/main/java/core/MR_byte.java
@@ -0,0 +1,73 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for byte values */
+final public class MR_byte extends MRData {
+ private byte value;
+
+ public MR_byte ( byte i ) { value = i; }
+ public MR_byte ( int i ) { value = (byte)i; }
+
+ public void materializeAll () {};
+
+ public byte get () { return value; }
+
+ public void set ( byte v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.BYTE);
+ out.writeByte(value);
+ }
+
+ final public static MR_byte read ( DataInput in ) throws IOException {
+ return new MR_byte(in.readByte());
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = in.readByte();
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_byte);
+ byte v = ((MR_byte) x).value;
+ return (value == v) ? 0 : ((value < v) ? -1 : 1);
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ size[0] = 2;
+ return x[xs]-y[ys];
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_byte && ((MR_byte)x).value==value;
+ }
+
+ public int hashCode () { return Math.abs(value); }
+
+ public String toString () {
+ return Integer.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_char.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_char.java b/src/main/java/core/MR_char.java
new file mode 100644
index 0000000..1bf5487
--- /dev/null
+++ b/src/main/java/core/MR_char.java
@@ -0,0 +1,71 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for a char */
+final public class MR_char extends MRData {
+ private char value;
+
+ public MR_char ( char i ) { value = i; }
+
+ public void materializeAll () {};
+
+ public char get () { return value; }
+
+ public void set ( char v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.CHAR);
+ out.writeChar(value);
+ }
+
+ final public static MR_char read ( DataInput in ) throws IOException {
+ return new MR_char(in.readChar());
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = in.readChar();
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_char);
+ return value-((MR_char) x).value;
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ size[0] = 2;
+ return x[xs]-y[ys];
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_char && ((MR_char)x).value==value;
+ }
+
+ public int hashCode () { return value; }
+
+ public String toString () {
+ return Character.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_dataset.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_dataset.java b/src/main/java/core/MR_dataset.java
new file mode 100644
index 0000000..c020628
--- /dev/null
+++ b/src/main/java/core/MR_dataset.java
@@ -0,0 +1,56 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a wrapper of a DataSet (stored in HDFS) as an MRData */
+final public class MR_dataset extends MRData {
+ public DataSet dataset;
+
+ public MR_dataset ( DataSet d ) { dataset = d; }
+
+ public void materializeAll () {};
+
+ public DataSet dataset () { return dataset; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ try {
+ Plan.collect(dataset).write(out);
+ } catch (Exception ex) {
+ throw new Error(ex);
+ }
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ throw new Error("DataSets are not serializable");
+ }
+
+ public int compareTo ( MRData x ) {
+ throw new Error("DataSets cannot be compared");
+ }
+
+ public boolean equals ( Object x ) {
+ throw new Error("DataSets cannot be compared");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_double.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_double.java b/src/main/java/core/MR_double.java
new file mode 100644
index 0000000..4d7ad59
--- /dev/null
+++ b/src/main/java/core/MR_double.java
@@ -0,0 +1,75 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for double values */
+final public class MR_double extends MRData {
+ private double value;
+
+ public MR_double ( double x ) { value = x; }
+
+ public void materializeAll () {};
+
+ public double get () { return value; }
+
+ public void set ( double v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.DOUBLE);
+ out.writeDouble(value);
+ }
+
+ final public static MR_double read ( DataInput in ) throws IOException {
+ return new MR_double(in.readDouble());
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = in.readDouble();
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_double);
+ double v = ((MR_double) x).value;
+ return (value == v) ? 0 : ((value > v) ? 1 : -1);
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ size[0] = (Double.SIZE >> 3)+1;
+ double v = WritableComparator.readDouble(x,xs) - WritableComparator.readDouble(y,ys);
+ return (v == 0) ? 0 : ((v > 0) ? 1 : -1);
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_double && ((MR_double)x).value==value;
+ }
+
+ public int hashCode () {
+ return Math.abs((int)Double.doubleToLongBits(value));
+ }
+
+ public String toString () {
+ return Double.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_float.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_float.java b/src/main/java/core/MR_float.java
new file mode 100644
index 0000000..d359013
--- /dev/null
+++ b/src/main/java/core/MR_float.java
@@ -0,0 +1,77 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for float values */
+final public class MR_float extends MRData {
+ private float value;
+
+ public MR_float ( float x ) { value = x; }
+
+ public MR_float ( double x ) { value = (float)x; }
+
+ public void materializeAll () {};
+
+ public float get () { return value; }
+
+ public void set ( float v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.FLOAT);
+ out.writeFloat(value);
+ }
+
+ final public static MR_float read ( DataInput in ) throws IOException {
+ return new MR_float(in.readFloat());
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = in.readFloat();
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_float);
+ float v = ((MR_float) x).value;
+ return (value == v) ? 0 : ((value > v) ? 1 : -1);
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ size[0] = (Float.SIZE >> 3)+1;
+ float v = WritableComparator.readFloat(x,xs) - WritableComparator.readFloat(y,ys);
+ return (v == 0) ? 0 : ((v > 0) ? 1 : -1);
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_float && ((MR_float)x).value==value;
+ }
+
+ public int hashCode () {
+ return Math.abs(Float.floatToIntBits(value));
+ }
+
+ public String toString () {
+ return Float.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_int.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_int.java b/src/main/java/core/MR_int.java
new file mode 100644
index 0000000..4c226a3
--- /dev/null
+++ b/src/main/java/core/MR_int.java
@@ -0,0 +1,74 @@
+/**
+ * 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 java.io.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for int values */
+final public class MR_int extends MRData {
+ private int value;
+
+ public MR_int ( int i ) { value = i; }
+
+ public void materializeAll () {};
+
+ public int get () { return value; }
+
+ public void set ( int v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.INT);
+ WritableUtils.writeVInt(out,value);
+ }
+
+ final public static MR_int read ( DataInput in ) throws IOException {
+ return new MR_int(WritableUtils.readVInt(in));
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = WritableUtils.readVInt(in);
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_int);
+ return value - ((MR_int) x).value;
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ try {
+ size[0] = 1+WritableUtils.decodeVIntSize(x[xs]);
+ int v = WritableComparator.readVInt(x,xs)-WritableComparator.readVInt(y,ys);
+ return (v == 0) ? 0 : ((v > 0) ? 1 : -1);
+ } catch (IOException e) {
+ throw new Error(e);
+ }
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_int && ((MR_int)x).value==value;
+ }
+
+ public int hashCode () { return Math.abs(value); }
+
+ public String toString () {
+ return Integer.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_long.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_long.java b/src/main/java/core/MR_long.java
new file mode 100644
index 0000000..008de25
--- /dev/null
+++ b/src/main/java/core/MR_long.java
@@ -0,0 +1,77 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for long values */
+final public class MR_long extends MRData {
+ private long value;
+
+ public MR_long ( long i ) { value = i; }
+
+ public void materializeAll () {};
+
+ public long get () { return value; }
+
+ public void set ( long v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.LONG);
+ WritableUtils.writeVLong(out,value);
+ }
+
+ final public static MR_long read ( DataInput in ) throws IOException {
+ return new MR_long(WritableUtils.readVLong(in));
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = WritableUtils.readVLong(in);
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_long);
+ long v = ((MR_long) x).value;
+ return (value == v) ? 0 : ((value > v) ? 1 : -1);
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ try {
+ size[0] = 1+WritableUtils.decodeVIntSize(x[xs]);
+ long v = WritableComparator.readVLong(x,xs)-WritableComparator.readVLong(y,ys);
+ return (v == 0) ? 0 : ((v > 0) ? 1 : -1);
+ } catch (IOException e) {
+ throw new Error(e);
+ }
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_long && ((MR_long)x).value==value;
+ }
+
+ public int hashCode () { return (int)Math.abs(value); }
+
+ public String toString () {
+ return Long.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_more_bsp_steps.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_more_bsp_steps.java b/src/main/java/core/MR_more_bsp_steps.java
new file mode 100644
index 0000000..58c7075
--- /dev/null
+++ b/src/main/java/core/MR_more_bsp_steps.java
@@ -0,0 +1,44 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+
+
+/** used for BSP synchronization when a peer needs to do more steps */
+final public class MR_more_bsp_steps extends MRData {
+ MR_more_bsp_steps () {}
+
+ public void materializeAll () {};
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.MORE_BSP_STEPS);
+ }
+
+ public void readFields ( DataInput in ) throws IOException {}
+
+ public int compareTo ( MRData x ) { return 0; }
+
+ public boolean equals ( Object x ) { return x instanceof MR_more_bsp_steps; }
+
+ public int hashCode () { return 0; }
+
+ public String toString () { return "more_bsp_steps"; }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_short.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_short.java b/src/main/java/core/MR_short.java
new file mode 100644
index 0000000..ced6be8
--- /dev/null
+++ b/src/main/java/core/MR_short.java
@@ -0,0 +1,71 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for short values */
+final public class MR_short extends MRData {
+ private short value;
+
+ public MR_short ( short i ) { value = i; }
+
+ public void materializeAll () {};
+
+ public short get () { return value; }
+
+ public void set ( short v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.SHORT);
+ out.writeShort(value);
+ }
+
+ final public static MR_short read ( DataInput in ) throws IOException {
+ return new MR_short(in.readShort());
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = in.readShort();
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_short);
+ return value - ((MR_short) x).value;
+ }
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ size[0] = (Short.SIZE >> 3)+1;
+ return WritableComparator.readUnsignedShort(x,xs) - WritableComparator.readUnsignedShort(y,ys);
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_short && ((MR_short)x).value==value;
+ }
+
+ public int hashCode () { return Math.abs((int)value); }
+
+ public String toString () {
+ return Short.toString(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_string.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_string.java b/src/main/java/core/MR_string.java
new file mode 100644
index 0000000..58a3e5f
--- /dev/null
+++ b/src/main/java/core/MR_string.java
@@ -0,0 +1,77 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.fs.*;
+
+
+/** a container for strings */
+final public class MR_string extends MRData {
+ private String value;
+
+ public MR_string ( String s ) { value = s; }
+
+ public void materializeAll () {};
+
+ public String get () { return value; }
+
+ public void set ( String v ) { value = v; }
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.STRING);
+ Text.writeString(out,value);
+ }
+
+ final public static MR_string read ( DataInput in ) throws IOException {
+ return new MR_string(Text.readString(in));
+ }
+
+ public void readFields ( DataInput in ) throws IOException {
+ value = Text.readString(in);
+ }
+
+ public int compareTo ( MRData x ) {
+ assert(x instanceof MR_string);
+ return value.compareTo(((MR_string) x).value);
+ }
+
+ final static Text.Comparator comparator = new Text.Comparator();
+
+ final public static int compare ( byte[] x, int xs, int xl, byte[] y, int ys, int yl, int[] size ) {
+ try {
+ size[0] = 1+WritableComparator.readVInt(x,xs)+WritableUtils.decodeVIntSize(x[xs]);
+ return comparator.compare(x,xs,xl,y,ys,yl);
+ } catch (IOException e) {
+ throw new Error(e);
+ }
+ }
+
+ public boolean equals ( Object x ) {
+ return x instanceof MR_string && value.equals(((MR_string) x).value);
+ }
+
+ public int hashCode () { return value.hashCode(); }
+
+ public String toString () {
+ return "\""+value+"\"";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_sync.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_sync.java b/src/main/java/core/MR_sync.java
new file mode 100644
index 0000000..ba6ca27
--- /dev/null
+++ b/src/main/java/core/MR_sync.java
@@ -0,0 +1,44 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+
+
+/** used for BSP synchronization */
+final public class MR_sync extends MRData {
+ MR_sync () {}
+
+ public void materializeAll () {};
+
+ final public void write ( DataOutput out ) throws IOException {
+ out.writeByte(MRContainer.SYNC);
+ }
+
+ public void readFields ( DataInput in ) throws IOException {}
+
+ public int compareTo ( MRData x ) { return 0; }
+
+ public boolean equals ( Object x ) { return x instanceof MR_sync; }
+
+ public int hashCode () { return 0; }
+
+ public String toString () { return "sync"; }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MR_variable.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MR_variable.java b/src/main/java/core/MR_variable.java
new file mode 100644
index 0000000..469884b
--- /dev/null
+++ b/src/main/java/core/MR_variable.java
@@ -0,0 +1,46 @@
+/**
+ * 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 java.io.IOException;
+import java.io.DataInput;
+import java.io.DataOutput;
+
+
+/** a template variable; should appear only in a template */
+final public class MR_variable extends MRData {
+ public int var_num;
+
+ MR_variable ( int n ) { var_num = n; }
+
+ public void materializeAll () {};
+
+ final public void write ( DataOutput out ) throws IOException {}
+
+ public void readFields ( DataInput in ) throws IOException {}
+
+ public int compareTo ( MRData x ) { return 0; }
+
+ public boolean equals ( Object x ) { return false; }
+
+ public int hashCode () { return 0; }
+
+ public String toString () {
+ return "variable("+var_num+")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/Main.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/Main.java b/src/main/java/core/Main.java
new file mode 100644
index 0000000..9703043
--- /dev/null
+++ b/src/main/java/core/Main.java
@@ -0,0 +1,163 @@
+/**
+ * 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 java.io.*;
+import java.util.Enumeration;
+import org.apache.hadoop.util.*;
+import org.apache.hadoop.conf.*;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.log4j.*;
+import jline.*;
+
+
+final public class Main {
+ public final static String version = "0.9.0";
+
+ public static PrintStream print_stream;
+ public static Configuration conf;
+ static MRQLParser parser = new MRQLParser();
+ public static String query_file = "";
+
+ public static void include_file ( String file ) {
+ try {
+ MRQLParser old_parser = parser;
+ boolean old_interactive = Config.interactive;
+ try {
+ parser = new MRQLParser(new MRQLLex(new FileInputStream(file)));
+ } catch (Exception e) {
+ Path path = new Path(file);
+ FileSystem fs = path.getFileSystem(conf);
+ parser = new MRQLParser(new MRQLLex(fs.open(path)));
+ };
+ Config.interactive = false;
+ parser.parse();
+ Config.interactive = old_interactive;
+ parser = old_parser;
+ } catch (Exception ex) {
+ ex.printStackTrace(System.err);
+ throw new Error(ex);
+ }
+ }
+
+ public static void main ( String[] args ) throws Exception {
+ boolean hadoop = false;
+ for ( String arg: args )
+ hadoop |= arg.equals("-local") || arg.equals("-dist");
+ if (hadoop) {
+ conf = Evaluator.new_configuration();
+ GenericOptionsParser gop = new GenericOptionsParser(conf,args);
+ conf = gop.getConfiguration();
+ args = gop.getRemainingArgs();
+ };
+ Config.parse_args(args,conf);
+ Config.hadoop_mode = Config.local_mode || Config.distributed_mode;
+ if (!Config.info) {
+ for ( Enumeration en = LogManager.getCurrentLoggers(); en.hasMoreElements(); )
+ ((Logger)en.nextElement()).setLevel(Level.WARN);
+ LogManager.getRootLogger().setLevel(Level.WARN);
+ };
+ Evaluator.init(conf);
+ new TopLevel();
+ System.out.print("Apache MRQL version "+version+" (");
+ if (Config.compile_functional_arguments)
+ System.out.print("compiled ");
+ else System.out.print("interpreted ");
+ if (hadoop) {
+ if (Config.local_mode)
+ System.out.print("local ");
+ else if (Config.distributed_mode)
+ System.out.print("distributed ");
+ if (Config.spark_mode)
+ System.out.println("Spark mode using "+Config.nodes+" workers)");
+ else if (Config.bsp_mode)
+ System.out.println("Hama BSP mode over "+Config.nodes+" BSP tasks)");
+ else if (Config.nodes > 0)
+ System.out.println("Hadoop MapReduce mode with "+Config.nodes+" reducers)");
+ else if (!Config.local_mode)
+ System.out.println("Hadoop MapReduce mode with 1 reducer, use -nodes to change it)");
+ else System.out.println("Hadoop MapReduce mode)");
+ } else if (Config.bsp_mode)
+ System.out.println("in-memory BSP mode)");
+ else System.out.println("in-memory MapReduce mode)");
+ if (Config.interactive) {
+ System.out.println("Type quit to exit");
+ ConsoleReader reader = new ConsoleReader();
+ reader.setBellEnabled(false);
+ History history = new History(new File(System.getProperty("user.home")+"/.mrqlhistory"));
+ reader.setHistory(history);
+ reader.setUseHistory(false);
+ try {
+ loop: while (true) {
+ String line = "";
+ String s = "";
+ try {
+ if (hadoop && Config.bsp_mode)
+ Config.write(Plan.conf);
+ do {
+ s = reader.readLine("> ");
+ if (s != null && (s.equals("quit") || s.equals("exit")))
+ break loop;
+ if (s != null)
+ line += " "+s;
+ } while (s == null || s.indexOf(";") <= 0);
+ line = line.substring(1);
+ history.addToHistory(line);
+ parser = new MRQLParser(new MRQLLex(new StringReader(line)));
+ MRQLLex.reset();
+ parser.parse();
+ } catch (EOFException x) {
+ break;
+ } catch (Exception x) {
+ if (x.getMessage() != null)
+ System.out.println(x);
+ } catch (Error x) {
+ System.out.println(x);
+ }
+ }
+ } finally {
+ if (hadoop) {
+ Plan.clean();
+ Evaluator.shutdown(Plan.conf);
+ };
+ if (Config.compile_functional_arguments)
+ Compiler.clean();
+ }
+ } else try {
+ if (hadoop && Config.bsp_mode)
+ Config.write(Plan.conf);
+ try {
+ parser = new MRQLParser(new MRQLLex(new FileInputStream(query_file)));
+ } catch (Exception e) {
+ // when the query file is in HDFS
+ Path path = new Path(query_file);
+ FileSystem fs = path.getFileSystem(conf);
+ parser = new MRQLParser(new MRQLLex(fs.open(path)));
+ };
+ parser.parse();
+ } finally {
+ if (hadoop) {
+ Plan.clean();
+ Evaluator.shutdown(Plan.conf);
+ };
+ if (Config.compile_functional_arguments)
+ Compiler.clean();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MapReduceAlgebra.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MapReduceAlgebra.java b/src/main/java/core/MapReduceAlgebra.java
new file mode 100644
index 0000000..9c35efa
--- /dev/null
+++ b/src/main/java/core/MapReduceAlgebra.java
@@ -0,0 +1,815 @@
+/**
+ * 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.*;
+
+
+/** Evaluation of MRQL algebra expressions in memory */
+final public class MapReduceAlgebra {
+
+ /** eager concat-map (not used) */
+ private static Bag cmap_eager ( final Function f, final Bag s ) {
+ Bag res = new Bag();
+ for ( MRData e: s )
+ res.addAll((Bag)f.eval(e));
+ return res;
+ }
+
+ /** lazy concat-map (stream-based)
+ * @param f a function from a to {b}
+ * @param s the input of type {a}
+ * @return a value of type {b}
+ */
+ public static Bag cmap ( final Function f, final Bag s ) {
+ final Iterator<MRData> si = s.iterator();
+ return new Bag(new BagIterator() {
+ Iterator<MRData> data = null;
+ boolean more = false;
+ public boolean hasNext () {
+ if (data == null) {
+ while (!more && si.hasNext()) {
+ data = ((Bag)f.eval(si.next())).iterator();
+ more = data.hasNext();
+ }
+ } else {
+ if (more) {
+ more = data.hasNext();
+ if (more)
+ return true;
+ };
+ while (!more && si.hasNext()) {
+ data = ((Bag)f.eval(si.next())).iterator();
+ more = data.hasNext();
+ }
+ };
+ return more;
+ }
+ public MRData next () {
+ return data.next();
+ }
+ });
+ }
+
+ /** lazy map
+ * @param f a function from a to b
+ * @param s the input of type {a}
+ * @return a value of type {b}
+ */
+ public static Bag map ( final Function f, final Bag s ) {
+ final Iterator<MRData> si = s.iterator();
+ return new Bag(new BagIterator() {
+ public boolean hasNext () { return si.hasNext(); }
+ public MRData next () { return f.eval(si.next()); }
+ });
+ }
+
+ /** lazy filter combined with a map
+ * @param p a function from a to boolean
+ * @param f a function from a to b
+ * @param s the input of type {a}
+ * @return a value of type {b}
+ */
+ public static Bag filter ( final Function p, final Function f, final Bag s ) {
+ final Iterator<MRData> si = s.iterator();
+ return new Bag(new BagIterator() {
+ MRData data = null;
+ public boolean hasNext () {
+ while (si.hasNext()) {
+ data = si.next();
+ if (((MR_bool)p.eval(data)).get())
+ return true;
+ };
+ return false;
+ }
+ public MRData next () { return f.eval(data); }
+ });
+ }
+
+ /** strict group-by
+ * @param s the input of type {(a,b)}
+ * @return a value of type {(a,{b})}
+ */
+ public static Bag groupBy ( Bag s ) {
+ Bag res = new Bag();
+ s.sort();
+ MRData last = null;
+ Bag group = new Bag();
+ for ( MRData e: s) {
+ final Tuple p = (Tuple)e;
+ if (last != null && p.first().equals(last))
+ group.add(p.second());
+ else {
+ if (last != null) {
+ group.trim();
+ res.add(new Tuple(last,group));
+ };
+ last = p.first();
+ group = new Bag();
+ group.add(p.second());
+ }
+ };
+ if (last != null) {
+ group.trim();
+ res.add(new Tuple(last,group));
+ };
+ //res.trim();
+ return res;
+ }
+
+ /** lazy group-by (not used) */
+ private static Bag groupBy_lazy ( Bag s ) {
+ s.sort();
+ final Iterator<MRData> it = s.iterator();
+ return new Bag(new BagIterator() {
+ MRData last = null;
+ MRData data = null;
+ Bag group = new Bag();
+ public boolean hasNext () {
+ while (it.hasNext()) {
+ final Tuple p = (Tuple)it.next();
+ if (last != null && p.first().equals(last))
+ group.add(p.second());
+ else if (last != null) {
+ group.trim();
+ data = new Tuple(last,group);
+ last = p.first();
+ group = new Bag();
+ group.add(p.second());
+ return true;
+ } else {
+ last = p.first();
+ group = new Bag();
+ group.add(p.second());
+ }
+ };
+ if (last != null) {
+ group.trim();
+ data = new Tuple(last,group);
+ last = null;
+ return true;
+ };
+ return false;
+ }
+ public MRData next () {
+ return data;
+ }
+ });
+ }
+
+ /** the MapReduce operation
+ * @param m a map function from a to {(k,b)}
+ * @param r a reduce function from (k,{b}) to {c}
+ * @param s the input of type {a}
+ * @return a value of type {c}
+ */
+ public static Bag mapReduce ( final Function m, final Function r, final Bag s ) {
+ return cmap(r,groupBy(cmap(m,s)));
+ }
+
+ /** Not used: use mapReduce2 instead */
+ private static Bag join ( final Function kx, final Function ky, final Function f,
+ final Bag X, final Bag Y ) {
+ return cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple p = (Tuple)e;
+ return (Bag)f.eval(new Tuple(p.second(),
+ cmap(new Function() {
+ public Bag eval ( final MRData y ) {
+ return (ky.eval(y).equals(p.first()))
+ ? new Bag(y)
+ : new Bag();
+ } }, Y))); }
+ },
+ groupBy(cmap(new Function() {
+ public Bag eval ( final MRData x ) {
+ return new Bag(new Tuple(kx.eval(x),x));
+ } }, X)));
+ }
+
+ /** A hash-based equi-join
+ * @param kx left key function from a to k
+ * @param ky right key function from b to k
+ * @param f reducer from ({a},{b}) to {c}
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {c}
+ */
+ public static Bag hash_join ( final Function kx, final Function ky, final Function f,
+ final Bag X, final Bag Y ) {
+ Hashtable<MRData,Bag> hashTable = new Hashtable<MRData,Bag>(1000);
+ for ( MRData x: X ) {
+ MRData key = kx.eval(x);
+ Bag old = hashTable.get(key);
+ if (old == null)
+ hashTable.put(key,new Bag(x));
+ else old.add(x);
+ };
+ Bag res = new Bag();
+ for ( MRData y: Y ) {
+ MRData key = ky.eval(y);
+ Bag match = hashTable.get(key);
+ if (match != null)
+ for ( MRData x: match )
+ res.add(f.eval(new Tuple(x,y)));
+ };
+ return res;
+ }
+
+ /** A cross-product
+ * @param mx left map function from a to {a'}
+ * @param my right key function from b to {b'}
+ * @param r reducer from (a',b') to {c}
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {c}
+ */
+ public static Bag crossProduct ( final Function mx, final Function my, final Function r,
+ final Bag X, final Bag Y ) {
+ Bag a = new Bag();
+ for ( MRData y: Y )
+ for ( MRData v: (Bag)my.eval(y) )
+ a.add(v);
+ Bag b = new Bag();
+ for ( MRData x: X )
+ for ( MRData xx: (Bag)mx.eval(x) )
+ for ( MRData y: a )
+ for ( MRData v: (Bag)r.eval(new Tuple(xx,y)) )
+ b.add(v);
+ return b;
+ }
+
+ /** A map-reduce operation with two mappers (a join)
+ * @param mx left map function from a to {(k,a')}
+ * @param my right key function from b to {(k,b')}
+ * @param r reducer from ({a'},{b'}) to {c}
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {c}
+ */
+ public static Bag mapReduce2 ( final Function mx, // left mapper
+ final Function my, // right mapper
+ final Function r, // reducer
+ final Bag X, final Bag Y ) {
+ final Bag left = cmap(new Function() {
+ public Bag eval ( final MRData x ) {
+ return cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple p = (Tuple)e;
+ return new Bag(new Tuple(p.first(),
+ new Tuple(new MR_byte(1),p.second())));
+ } }, (Bag)mx.eval(x));
+ } }, X);
+ final Bag right = cmap(new Function() {
+ public Bag eval ( final MRData y ) {
+ return cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple p = (Tuple)e;
+ return new Bag(new Tuple(p.first(),
+ new Tuple(new MR_byte(2),p.second())));
+ } }, (Bag)my.eval(y));
+ } }, Y);
+ final Iterator<MRData> li = left.iterator();
+ final Iterator<MRData> ri = right.iterator();
+ final Bag mix = new Bag(new BagIterator () {
+ MRData data;
+ public boolean hasNext () {
+ if (li.hasNext()) {
+ data = li.next();
+ return true;
+ } else if (ri.hasNext()) {
+ data = ri.next();
+ return true;
+ } else return false;
+ }
+ public MRData next () {
+ return data;
+ }
+ });
+ return cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple p = (Tuple)e;
+ final Bag xs = cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple q = (Tuple)e;
+ return (((MR_byte)q.first()).get() == 1)
+ ? new Bag(q.second())
+ : new Bag();
+ } }, (Bag)p.second());
+ final Bag ys = cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple q = (Tuple)e;
+ return (((MR_byte)q.first()).get() == 2)
+ ? new Bag(q.second())
+ : new Bag();
+ } }, (Bag)p.second());
+ xs.materialize();
+ ys.materialize();
+ return (Bag)r.eval(new Tuple(xs,ys));
+ } }, groupBy(mix));
+ }
+
+ /** The fragment-replicate join (map-side join)
+ * @param kx left key function from a to k
+ * @param ky right key function from b to k
+ * @param r reducer from ({a},{b}) to {c}
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {c}
+ */
+ public static Bag mapJoin ( final Function kx, final Function ky, final Function r,
+ final Bag X, final Bag Y ) {
+ X.materialize();
+ Y.materialize();
+ return cmap(new Function() {
+ public Bag eval ( final MRData e ) {
+ final Tuple p = (Tuple)e;
+ return cmap(new Function() {
+ public Bag eval ( final MRData x ) {
+ return (kx.eval(x).equals(p.first()))
+ ? (Bag)r.eval(new Tuple(x,p.second()))
+ : new Bag();
+ } }, X); }
+ },
+ groupBy(cmap(new Function() {
+ public Bag eval ( final MRData y ) {
+ return new Bag(new Tuple(ky.eval(y),y));
+ } }, Y)));
+ }
+
+ /** An equi-join combined with a group-by (see GroupByJoinPlan)
+ * @param kx left key function from a to k
+ * @param ky right key function from b to k
+ * @param gx group-by key function from a to k1
+ * @param gy group-by key function from b to k2
+ * @param m mapper from (a,b) to {c}
+ * @param c combiner from ((k1,k2),{c}) to d
+ * @param r reducer from ((k1,k2),d) to {e}
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {e}
+ */
+ public static Bag groupByJoin ( final Function kx, final Function ky,
+ final Function gx, final Function gy,
+ final Function m, final Function c, final Function r,
+ final Bag X, final Bag Y ) {
+ Bag s = groupBy(hash_join(kx,ky,
+ new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(gx.eval(t.first()),gy.eval(t.second())),t);
+ } },
+ X,Y));
+ Bag res = new Bag();
+ for ( MRData z: s ) {
+ Tuple t = (Tuple)z;
+ for ( MRData n: (Bag)r.eval(new Tuple(t.first(),c.eval(new Tuple(t.first(),cmap(m,(Bag)t.second()))))) )
+ res.add(n);
+ };
+ return res;
+ }
+
+ private static void flush_table ( Hashtable<MRData,MRData> hashTable, Function r, Bag result ) throws IOException {
+ Bag tbag = new Bag(2);
+ Tuple pair = new Tuple(2);
+ Enumeration<MRData> en = hashTable.keys();
+ while (en.hasMoreElements()) {
+ MRData key = en.nextElement();
+ MRData value = hashTable.get(key);
+ pair.set(0,key);
+ tbag.clear();
+ tbag.add_element(value);
+ pair.set(1,tbag);
+ for ( MRData e: (Bag)r.eval(pair) )
+ result.add(e);
+ };
+ hashTable.clear();
+ }
+
+ /** An equi-join combined with a group-by implemented using hashing
+ * @param kx left key function from a to k
+ * @param ky right key function from b to k
+ * @param gx group-by key function from a to k1
+ * @param gy group-by key function from b to k2
+ * @param m mapper from (a,b) to {c}
+ * @param c combiner from ((k1,k2),{c}) to d
+ * @param r reducer from ((k1,k2),d) to {e}
+ * @param X left input of type {a}
+ * @param Y right input of type {b}
+ * @return a value of type {e}
+ */
+ public static Bag mergeGroupByJoin ( final Function kx, final Function ky,
+ final Function gx, final Function gy,
+ final Function m, final Function c, final Function r,
+ final Bag X, final Bag Y ) {
+ try {
+ Bag tbag = new Bag(2);
+ Tuple pair = new Tuple(2);
+ Hashtable<MRData,MRData> hashTable = new Hashtable<MRData,MRData>(1000);
+ Bag xs = groupBy(map(new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(t.first(),kx.eval(t.second())),t.second());
+ } }, X));
+ Bag ys = groupBy(map(new Function() {
+ public MRData eval ( final MRData e ) {
+ Tuple t = (Tuple)e;
+ return new Tuple(new Tuple(t.first(),ky.eval(t.second())),t.second());
+ } }, Y));
+ Bag res = new Bag();
+ Iterator<MRData> xi = xs.iterator();
+ Iterator<MRData> yi = ys.iterator();
+ if ( !xi.hasNext() || !yi.hasNext() )
+ return res;
+ Tuple x = (Tuple)xi.next();
+ Tuple y = (Tuple)yi.next();
+ MRData partition = null;
+ while ( xi.hasNext() && yi.hasNext() ) {
+ int cmp = x.first().compareTo(y.first());
+ if (cmp < 0) { x = (Tuple)xi.next(); continue; };
+ if (cmp > 0) { y = (Tuple)yi.next(); continue; };
+ if (partition == null)
+ partition = ((Tuple)x.first()).first();
+ else if (!partition.equals(((Tuple)x.first()).first())) {
+ partition = ((Tuple)x.first()).first();
+ flush_table(hashTable,r,res);
+ };
+ for ( MRData xx: (Bag)x.second() )
+ for ( MRData yy: (Bag)y.second() ) {
+ Tuple key = new Tuple(gx.eval(xx),gy.eval(yy));
+ Tuple value = new Tuple(xx,yy);
+ MRData old = hashTable.get(key);
+ pair.set(0,key);
+ for ( MRData e: (Bag)m.eval(value) )
+ if (old == null)
+ hashTable.put(key,e);
+ else {
+ tbag.clear();
+ tbag.add_element(e).add_element(old);
+ pair.set(1,tbag);
+ for ( MRData z: (Bag)c.eval(pair) )
+ hashTable.put(key,z); // normally, done once
+ }
+ };
+ if (xi.hasNext())
+ x = (Tuple)xi.next();
+ if (yi.hasNext())
+ y = (Tuple)yi.next();
+ };
+ flush_table(hashTable,r,res);
+ return res;
+ } catch (IOException ex) {
+ throw new Error(ex);
+ }
+ }
+
+ /** repeat the loop until all termination conditions are true or until we reach the max num of steps
+ * @param loop a function from {a} to {(a,boolean)}
+ * @param init the initial value of type {a}
+ * @param max_num the maximum number of steps
+ * @return a value of type {a}
+ */
+ public static Bag repeat ( final Function loop,
+ final Bag init,
+ final int max_num ) throws Exception {
+ boolean cont;
+ int i = 0;
+ Bag s = init;
+ s.materializeAll();
+ do {
+ MRData d = loop.eval(s);
+ i++;
+ cont = false;
+ if (d instanceof Bag) {
+ Bag bag = (Bag) d;
+ bag.materialize();
+ s.clear();
+ for ( MRData x: bag ) {
+ Tuple t = (Tuple)x;
+ cont |= ((MR_bool)t.second()).get();
+ s.add(t.first());
+ }
+ } else if (d instanceof MR_dataset) {
+ DataSet ds = ((MR_dataset)d).dataset();
+ if (ds.counter != 0)
+ cont = true;
+ System.err.println("*** Repeat #"+i+": "+ds.counter+" true results");
+ s = Plan.collect(ds);
+ } else throw new Error("Wrong repeat");
+ } while (cont && i <= max_num);
+ return s;
+ }
+
+ /** transitive closure: repeat the loop until the new set is equal to the previous set
+ * or until we reach the max num of steps
+ * @param loop a function from {a} to {a}
+ * @param init the initial value of type {a}
+ * @param max_num the maximum number of steps
+ * @return a value of type {a}
+ */
+ public static Bag closure ( final Function loop,
+ final Bag init,
+ final int max_num ) throws Exception {
+ int i = 0;
+ long n = 0;
+ long old = 0;
+ Bag s = init;
+ s.materializeAll();
+ do {
+ MRData d = loop.eval(s);
+ i++;
+ if (d instanceof Bag) {
+ s = (Bag)d;
+ s.materialize();
+ old = n;
+ n = s.size();
+ } else if (d instanceof MR_dataset) {
+ DataSet ds = ((MR_dataset)d).dataset();
+ System.err.println("*** Repeat #"+i+": "+(ds.records-n)+" new records");
+ old = n;
+ n = ds.records;
+ s = Plan.collect(ds);
+ } else throw new Error("Wrong repeat");
+ } while (old < n && i <= max_num);
+ return s;
+ }
+
+ /** parse a text document using a given parser
+ * @param parser the parser
+ * @param file the text document (local file)
+ * @param args the arguments to pass to the parser
+ * @return a lazy bag that contains the parsed data
+ */
+ public static Bag parsedSource ( final Parser parser,
+ final String file,
+ Trees args ) {
+ try {
+ parser.initialize(args);
+ parser.open(file);
+ return new Bag(new BagIterator() {
+ Iterator<MRData> result = null;
+ MRData data;
+ public boolean hasNext () {
+ try {
+ while (result == null || !result.hasNext()) {
+ String s = parser.slice();
+ if (s == null)
+ return false;
+ result = parser.parse(s).iterator();
+ };
+ data = (MRData)result.next();
+ return true;
+ } catch (Exception e) {
+ throw new Error(e);
+ }
+ }
+ public MRData next () {
+ return data;
+ }
+ });
+ } catch (Exception e) {
+ throw new Error(e);
+ }
+ }
+
+ /** parse a text document using a given parser
+ * @param parser the name of the parser
+ * @param file the text document (local file)
+ * @param args the arguments to pass to the parser
+ * @return a lazy bag that contains the parsed data
+ */
+ public static Bag parsedSource ( String parser, String file, Trees args ) {
+ try {
+ return parsedSource(DataSource.parserDirectory.get(parser).newInstance(),file,args);
+ } catch (Exception e) {
+ throw new Error(e);
+ }
+ }
+
+ private static Bag add_source_num ( int source_num, Bag input ) {
+ return new Bag(new Tuple(new MR_int(source_num),input));
+ }
+
+ /** parse a text document using a given parser and tag output data with a source num
+ * @param source_num the source id
+ * @param parser the parser
+ * @param file the text document (local file)
+ * @param args the arguments to pass to the parser
+ * @return a lazy bag that contains the parsed data taged with the source id
+ */
+ public static Bag parsedSource ( int source_num,
+ Parser parser,
+ String file,
+ Trees args ) {
+ return add_source_num(source_num,parsedSource(parser,file,args));
+ }
+
+ /** parse a text document using a given parser and tag output data with a source num
+ * @param source_num the source id
+ * @param parser the name of the parser
+ * @param file the text document (local file)
+ * @param args the arguments to pass to the parser
+ * @return a lazy bag that contains the parsed data taged with the source id
+ */
+ public static Bag parsedSource ( int source_num, String parser, String file, Trees args ) {
+ try {
+ return parsedSource(source_num,DataSource.parserDirectory.get(parser).newInstance(),file,args);
+ } catch (Exception e) {
+ throw new Error(e);
+ }
+ }
+
+ /** aggregate the Bag elements
+ * @param accumulator a function from (b,a) to b
+ * @param zero a value of type b
+ * @param s a Bag of type {a}
+ * @return a value of type b
+ */
+ public static MRData aggregate ( final Function accumulator,
+ final MRData zero,
+ final Bag s ) {
+ MRData result = zero;
+ for ( MRData x: s )
+ result = accumulator.eval(new Tuple(result,x));
+ return result;
+ }
+
+ public static MRData materialize ( MRData x ) {
+ if (x instanceof Bag)
+ ((Bag)x).materialize();
+ return x;
+ }
+
+ /** Dump the value of some type to a binary local file;
+ * The type is dumped to a separate file.type
+ */
+ public static void dump ( String file, Tree type, MRData value ) throws IOException {
+ PrintStream ftp = new PrintStream(file+".type");
+ ftp.print("1@"+type.toString()+"\n");
+ ftp.close();
+ DataOutputStream out = new DataOutputStream(new FileOutputStream(new File(file)));
+ value.write(out);
+ out.close();
+ }
+
+ /** return the type of the dumped binary local file from file.type */
+ public static Tree get_type ( String file ) {
+ try {
+ BufferedReader ftp = new BufferedReader(new FileReader(new File(file+".type")));
+ String s[] = ftp.readLine().split("@");
+ ftp.close();
+ if (s.length != 2)
+ return null;
+ if (!s[0].equals("1"))
+ throw new Error("The binary file has been created in hadoop mode and cannot be read in java mode");
+ return Tree.parse(s[1]);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ /** read the contents of a dumped local binary file */
+ public static MRData read_binary ( String file ) {
+ try {
+ Tree type = get_type(file);
+ DataInputStream in = new DataInputStream(new FileInputStream(new File(file)));
+ return MRContainer.read(in);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ /** read the contents of a dumped local binary file and tag data with a source num */
+ public static Bag read_binary ( int source_num, String file ) {
+ return add_source_num(source_num,(Bag)read_binary(file));
+ }
+
+ /** generate a lazy bag of long numbers {min...max} */
+ public static Bag generator ( final long min, final long max ) {
+ if (min > max)
+ throw new Error("Min value ("+min+") is larger than max ("+max+") in generator");
+ return new Bag(new BagIterator() {
+ long index = min;
+ public boolean hasNext () {
+ return index <= max;
+ }
+ public MRData next () {
+ return new MR_long(index++);
+ }
+ });
+ }
+
+ /** generate a lazy bag of long numbers {min...max} and tag each lon number with a source num */
+ public static Bag generator ( int source_num, final long min, final long max ) {
+ return add_source_num(source_num,generator(min,max));
+ }
+
+ /** the cache that holds all local data in memory */
+ private static Tuple cache;
+
+ /** return the cache element at location loc */
+ public static MRData getCache ( int loc ) {
+ return cache.get(loc);
+ }
+
+ /** set the cache element at location loc to value and return ret */
+ public static MRData setCache ( int loc, MRData value, MRData ret ) {
+ if (value instanceof Bag)
+ materialize((Bag)value);
+ cache.set(loc,value);
+ return ret;
+ }
+
+ /** The BSP operation
+ * @param source the source ids of the input Bags
+ * @param superstep the BSP superstep is a function from ({M},S) to ({M},S,boolean)
+ * @param init_state is the initial state of type S
+ * @param order do we need to order the result?
+ * @param inputs the input Bags
+ * @return return a Bag in cache[0]
+ */
+ public static MRData BSP ( final int[] source,
+ final Function superstep,
+ final MRData init_state,
+ boolean order,
+ final Bag[] inputs ) {
+ Bag msgs = new Bag();
+ MRData state = init_state;
+ Tuple result;
+ boolean exit;
+ boolean skip = false;
+ String tabs = "";
+ int step = 0;
+ cache = new Tuple(100);
+ for ( int i = 0; i < 100; i++ )
+ cache.set(i,new Bag());
+ for ( Bag x: inputs ) {
+ Tuple p = (Tuple)(x.get(0));
+ cache.set(((MR_int)p.first()).get(),
+ materialize(p.second()));
+ };
+ do {
+ if (!skip)
+ step++;
+ if (!skip && Config.trace_execution) {
+ tabs = Interpreter.tabs(Interpreter.tab_count);
+ System.out.println(tabs+" Superstep "+step+":");
+ System.out.println(tabs+" messages: "+msgs);
+ System.out.println(tabs+" state: "+state);
+ for ( int i = 0; i < cache.size(); i++)
+ if (cache.get(i) instanceof Bag && ((Bag)cache.get(i)).size() > 0)
+ System.out.println(tabs+" cache "+i+": "+cache.get(i));
+ };
+ result = (Tuple)superstep.eval(new Tuple(cache,msgs,state,new MR_string("")));
+ Bag new_msgs = (Bag)result.get(0);
+ state = result.get(1);
+ exit = ((MR_bool)result.get(2)).get();
+ skip = new_msgs == SystemFunctions.bsp_empty_bag;
+ if ((!skip || exit) && Config.trace_execution)
+ System.out.println(tabs+" result: "+result);
+ final Iterator<MRData> iter = new_msgs.iterator();
+ msgs = new Bag(new BagIterator() {
+ public boolean hasNext () {
+ return iter.hasNext();
+ }
+ public MRData next () {
+ return ((Tuple)iter.next()).get(1);
+ }
+ });
+ } while (!exit);
+ MRData[] data = new MRData[source.length];
+ for ( int i = 0; i < data.length; i++ )
+ data[i] = getCache(source[i]);
+ if (order && data[0] instanceof Bag) {
+ final Iterator<MRData> iter = ((Bag)data[0]).iterator();
+ return new Bag(new BagIterator() {
+ public boolean hasNext () {
+ return iter.hasNext();
+ }
+ public MRData next () {
+ return ((Tuple)iter.next()).get(0);
+ }
+ });
+ };
+ if (data.length == 1)
+ return data[0];
+ else return new Tuple(data);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/Materialization.gen
----------------------------------------------------------------------
diff --git a/src/main/java/core/Materialization.gen b/src/main/java/core/Materialization.gen
new file mode 100644
index 0000000..738dfa6
--- /dev/null
+++ b/src/main/java/core/Materialization.gen
@@ -0,0 +1,134 @@
+/**
+ * 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.util.*;
+
+
+class Domains {
+ public Trees domains;
+ public Trees repeats;
+ public Domains ( Trees d, Trees r ) { domains = d; repeats = r; }
+}
+
+
+/** if the plan refers to a variable bound to a stream-based Bag and occurs in the code
+ * multiple times, embed code to materialize this Bag in memory
+ */
+final public class Materialization extends Translator {
+
+ // is this a direct-access term? (not the results of a bulk operation)
+ private static boolean access_variable ( Tree e ) {
+ match e {
+ case nth(`x,_):
+ return access_variable(x);
+ case union_value(`x):
+ return access_variable(x);
+ case index(`x,`n):
+ return access_variable(x);
+ case `v:
+ if (v.is_variable())
+ return true;
+ };
+ return false;
+ }
+
+ private static Domains new_domain ( Trees vars, Tree e, Domains d ) {
+ if (!access_variable(e))
+ return materialize(vars,e,d);
+ Domains nd = new Domains(d.domains,d.repeats);
+ if ((d.domains.member(e) || !free_variables(e,vars).is_empty())
+ && !d.repeats.member(e))
+ nd.repeats = nd.repeats.cons(e);
+ nd.domains = nd.domains.cons(e);
+ return nd;
+ }
+
+ private static Domains union ( Domains xd, Domains yd ) {
+ Domains nd = new Domains(xd.domains,xd.repeats);
+ for ( Tree y: yd.domains )
+ if (!nd.domains.member(y))
+ nd.domains = nd.domains.cons(y);
+ for ( Tree y: yd.repeats )
+ if (!nd.repeats.member(y))
+ nd.repeats = nd.repeats.cons(y);
+ return nd;
+ }
+
+ final static int unionM = ClassImporter.find_method_number("plus",#[bag(any),bag(any)]);
+
+ private static Domains materialize ( Trees vars, Tree e, Domains d ) {
+ match e {
+ case lambda(`v,`b):
+ return materialize(#[`v],b,d);
+ case cmap(lambda(`v,`b),`s):
+ return materialize(#[`v],b,new_domain(vars,s,d));
+ case map(lambda(`v,`b),`s):
+ return materialize(#[`v],b,new_domain(vars,s,d));
+ case filter(lambda(`v1,`b1),lambda(`v2,`b2),`s):
+ return materialize(#[`v1],b1,materialize(#[`v2],b2,new_domain(vars,s,d)));
+ case aggregate(lambda(`v,`b),`z,`s):
+ return materialize(#[`v],b,new_domain(vars,s,d));
+ case groupBy(`s):
+ return new_domain(vars,s,d);
+ case orderBy(`s):
+ return new_domain(vars,s,d);
+ case mapReduce2(lambda(`v1,`b1),lambda(`v2,`b2),lambda(`v,`b),`x,`y,`o):
+ return materialize(#[`v],b,materialize(#[`v1],b1,materialize(#[`v2],b2,
+ new_domain(vars,x,new_domain(vars,y,d)))));
+ case join(lambda(`v1,`b1),lambda(`v2,`b2),lambda(`v,`b),`x,`y):
+ return materialize(#[`v],b,materialize(#[`v1],b1,materialize(#[`v2],b2,
+ new_domain(vars,x,new_domain(vars,y,d)))));
+ case crossProduct(lambda(`v1,`b1),lambda(`v2,`b2),lambda(`v,`b),`x,`y):
+ return materialize(#[`v],b,materialize(#[`v1],b1,materialize(#[`v2],b2,
+ new_domain(vars,x,new_domain(vars,y,d)))));
+ case let(`v,`x,`y):
+ Domains nd = materialize(vars.cons(v),y,materialize(vars,x,d));
+ Trees zs = #[];
+ for ( Tree z: nd.repeats )
+ if (!v.equals(z))
+ zs = zs.cons(z);
+ nd.repeats = zs;
+ return nd;
+ case if(`p,`x,`y):
+ Domains nd = materialize(vars,p,d);
+ return union(materialize(vars,x,nd),
+ materialize(vars,y,nd));
+ case callM(union,_,`x,`y):
+ return new_domain(vars,x,new_domain(vars,y,d));
+ case callM(_,`k,`x,`y):
+ if (((LongLeaf)k).value() != unionM)
+ fail;
+ return new_domain(vars,x,new_domain(vars,y,d));
+ case `f(...as):
+ Domains nd = new Domains(d.domains,d.repeats);
+ for ( Tree a: as )
+ nd = materialize(vars,a,nd);
+ return nd;
+ };
+ return d;
+ }
+
+ public static Tree materialize_terms ( Tree e ) {
+ Domains d = materialize(#[],e,new Domains(#[],#[]));
+ for ( Tree x: d.repeats )
+ e = subst(x,#<materialize(`x)>,e);
+ return e;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/main/java/core/MethodInfo.java
----------------------------------------------------------------------
diff --git a/src/main/java/core/MethodInfo.java b/src/main/java/core/MethodInfo.java
new file mode 100644
index 0000000..29d4aea
--- /dev/null
+++ b/src/main/java/core/MethodInfo.java
@@ -0,0 +1,57 @@
+/**
+ * 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 java.lang.reflect.Method;
+import org.apache.mrql.gen.*;
+
+
+/** class for storing Java method information */
+final public class MethodInfo implements Comparable<MethodInfo> {
+ public String name;
+ public Trees signature;
+ public Method method;
+
+ MethodInfo ( String n, Trees s, Method m ) {
+ name = n;
+ signature = s;
+ method = m;
+ }
+
+ public int compareTo ( MethodInfo x ) {
+ int c = name.compareTo(x.name);
+ if (c != 0)
+ return c;
+ if (signature.length() < x.signature.length())
+ return -1;
+ if (signature.length() > x.signature.length())
+ return 1;
+ // handles overloading: more specific method signatures first
+ for ( int i = 1; i < signature.length(); i++ ) {
+ int ct = TypeInference.compare_types(signature.nth(i),x.signature.nth(i));
+ if (ct != 0)
+ return ct;
+ };
+ return TypeInference.compare_types(signature.nth(0),x.signature.nth(0));
+ }
+
+ public boolean equals ( Object x ) {
+ return name.equals(((MethodInfo)x).name)
+ && signature.equals(((MethodInfo)x).signature);
+ }
+}