You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ro...@apache.org on 2014/02/07 02:18:15 UTC
svn commit: r1565509 [2/3] - in /pig/branches/tez: src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/
src/org/apache/pig/backend/hadoop/executione...
Modified: pig/branches/tez/test/org/apache/pig/test/TestMultiQueryBasic.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestMultiQueryBasic.java?rev=1565509&r1=1565508&r2=1565509&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestMultiQueryBasic.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestMultiQueryBasic.java Fri Feb 7 01:18:14 2014
@@ -29,8 +29,6 @@ import java.util.List;
import java.util.Map;
import java.util.Properties;
-import junit.framework.Assert;
-
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -42,6 +40,7 @@ import org.apache.hadoop.mapreduce.Outpu
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.pig.ExecType;
import org.apache.pig.LoadFunc;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.PigServer;
import org.apache.pig.ResourceSchema;
import org.apache.pig.StoreFunc;
@@ -69,17 +68,17 @@ public class TestMultiQueryBasic {
Util.copyFromLocalToLocal(
"test/org/apache/pig/test/data/passwd2", "passwd2");
Properties props = new Properties();
- props.setProperty("opt.multiquery", ""+true);
+ props.setProperty(PigConfiguration.OPT_MULTIQUERY, ""+true);
myPig = new PigServer(ExecType.LOCAL, props);
}
-
+
@AfterClass
public static void tearDownAfterClass() throws Exception {
Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), "passwd");
Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), "passwd2");
deleteOutputFiles();
}
-
+
@Before
public void setUp() throws Exception {
deleteOutputFiles();
@@ -87,10 +86,10 @@ public class TestMultiQueryBasic {
@After
public void tearDown() throws Exception {
-
+
}
-
-
+
+
@Test
public void testMultiQueryWithTwoStores2() throws Exception {
@@ -133,8 +132,8 @@ public class TestMultiQueryBasic {
myPig.executeBatch();
myPig.discardBatch();
- }
-
+ }
+
@Test
public void testMultiQueryPhase3BaseCase2() throws Exception {
@@ -155,16 +154,16 @@ public class TestMultiQueryBasic {
myPig.registerQuery("c2 = foreach c1 generate group, SUM(c.uid);");
myPig.registerQuery("store c2 into 'output2';");
myPig.registerQuery("d1 = group d by gid;");
- myPig.registerQuery("d2 = foreach d1 generate group, AVG(d.uid);");
+ myPig.registerQuery("d2 = foreach d1 generate group, AVG(d.uid);");
myPig.registerQuery("store d2 into 'output3';");
-
+
List<ExecJob> jobs = myPig.executeBatch();
-
+
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
-
+ }
+
@Test
public void testMultiQueryPhase3WithoutCombiner2() throws Exception {
@@ -184,17 +183,17 @@ public class TestMultiQueryBasic {
myPig.registerQuery("c1 = group c by gid;");
myPig.registerQuery("c2 = foreach c1 generate group, SUM(c.uid) - COUNT(c.uid);");
myPig.registerQuery("store c2 into 'output2';");
- myPig.registerQuery("d1 = group d by gid;");
+ myPig.registerQuery("d1 = group d by gid;");
myPig.registerQuery("d2 = foreach d1 generate group, MAX(d.uid) - MIN(d.uid);");
myPig.registerQuery("store d2 into 'output3';");
-
+
List<ExecJob> jobs = myPig.executeBatch();
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
-
+ }
+
@Test
public void testMultiQueryPhase3WithMixedCombiner2() throws Exception {
@@ -214,17 +213,17 @@ public class TestMultiQueryBasic {
myPig.registerQuery("c1 = group c by gid;");
myPig.registerQuery("c2 = foreach c1 generate group, SUM(c.uid);");
myPig.registerQuery("store c2 into 'output2';");
- myPig.registerQuery("d1 = group d by gid;");
+ myPig.registerQuery("d1 = group d by gid;");
myPig.registerQuery("d2 = foreach d1 generate group, MAX(d.uid) - MIN(d.uid);");
myPig.registerQuery("store d2 into 'output3';");
-
+
List<ExecJob> jobs = myPig.executeBatch();
assertEquals(3, jobs.size());
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
+ }
@Test
public void testMultiQueryPhase3WithDifferentMapDataTypes2() throws Exception {
@@ -248,15 +247,15 @@ public class TestMultiQueryBasic {
myPig.registerQuery("d1 = group d by $1;");
myPig.registerQuery("d2 = foreach d1 generate group, COUNT(d.uid);");
myPig.registerQuery("store d2 into 'output3';");
-
+
List<ExecJob> jobs = myPig.executeBatch();
assertEquals(3, jobs.size());
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
-
+ }
+
@Test
public void testMultiQueryPhase3WithDifferentMapDataTypes3() throws Exception {
@@ -278,14 +277,14 @@ public class TestMultiQueryBasic {
myPig.registerQuery("d = group a by (name, gpa);");
myPig.registerQuery("e = foreach d generate flatten(group), MIN(a.age);");
myPig.registerQuery("store e into 'output2';");
-
+
myPig.executeBatch();
-
+
myPig.registerQuery("a = load 'output1' as (grp:chararray, cnt:long) ;");
Iterator<Tuple> it = myPig.openIterator("a");
assertEquals(Util.getPigConstant("('all', 5l)"), it.next());
assertFalse(it.hasNext());
-
+
myPig.registerQuery("a = load 'output2' as (name:chararray, gpa:double, age:int);");
it = myPig.openIterator("a");
int i = 0;
@@ -299,8 +298,8 @@ public class TestMultiQueryBasic {
assertEquals(expectedResults.get(t.get(0)), t);
}
assertEquals(3, i);
- }
-
+ }
+
@Test
public void testMultiQueryPhase3StreamingInReducer2() throws Exception {
@@ -317,19 +316,19 @@ public class TestMultiQueryBasic {
myPig.registerQuery("store D into 'output1';");
myPig.registerQuery("E = group A4 by $2;");
myPig.registerQuery("F = foreach E generate group, COUNT(A4);");
- myPig.registerQuery("store F into 'output2';");
+ myPig.registerQuery("store F into 'output2';");
myPig.registerQuery("G = group A1 by $2;");
- myPig.registerQuery("H = foreach G generate group, COUNT(A1);");
+ myPig.registerQuery("H = foreach G generate group, COUNT(A1);");
myPig.registerQuery("store H into 'output3';");
-
+
List<ExecJob> jobs = myPig.executeBatch();
assertEquals(3, jobs.size());
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
-
+ }
+
@Test
public void testMultiQueryWithPigMixL12_2() throws Exception {
@@ -340,7 +339,7 @@ public class TestMultiQueryBasic {
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("b = foreach a generate uname, passwd, uid, gid;");
- myPig.registerQuery("split b into c1 if uid > 5, c2 if uid <= 5 ;");
+ myPig.registerQuery("split b into c1 if uid > 5, c2 if uid <= 5 ;");
myPig.registerQuery("split c1 into d1 if gid < 5, d2 if gid >= 5;");
myPig.registerQuery("e = group d1 by uname;");
myPig.registerQuery("e1 = foreach e generate group, MAX(d1.uid);");
@@ -359,18 +358,18 @@ public class TestMultiQueryBasic {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
}
-
+
@Test
public void testMultiQueryWithCoGroup_2() throws Exception {
System.out.println("===== multi-query with CoGroup (2) =====");
-
+
myPig.setBatchOn();
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("store a into 'output1' using BinStorage();");
- myPig.registerQuery("b = load 'output1' using BinStorage() as (uname, passwd, uid, gid);");
+ myPig.registerQuery("b = load 'output1' using BinStorage() as (uname, passwd, uid, gid);");
myPig.registerQuery("c = load 'passwd2' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("d = cogroup b by (uname, uid) inner, c by (uname, uid) inner;");
@@ -384,7 +383,7 @@ public class TestMultiQueryBasic {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
}
-
+
@Test
public void testMultiQueryWithFJ_2() throws Exception {
@@ -409,15 +408,15 @@ public class TestMultiQueryBasic {
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
-
+ }
+
@Test
public void testMultiQueryWithIntermediateStores_2() throws Exception {
System.out.println("===== multi-query with intermediate stores (2) =====");
myPig.setBatchOn();
-
+
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname:chararray, passwd:chararray, uid:int, gid:int);");
myPig.registerQuery("store a into 'output1';");
@@ -430,7 +429,7 @@ public class TestMultiQueryBasic {
for (ExecJob job : jobs) {
assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
}
- }
+ }
@Test
public void testMultiQueryWithSplitInMapAndMultiMerge() throws Exception {
@@ -438,9 +437,9 @@ public class TestMultiQueryBasic {
// clean up any existing dirs/files
String[] toClean = {"tmwsimam-input.txt", "foo1", "foo2", "foo3", "foo4" };
for (int j = 0; j < toClean.length; j++) {
- Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), toClean[j]);
+ Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), toClean[j]);
}
-
+
// the data below is tab delimited
String[] inputData = {
"1 a b e f i j m n",
@@ -448,21 +447,21 @@ public class TestMultiQueryBasic {
"3 c d g h k l o p",
"4 c d g h k l o p" };
Util.createLocalInputFile("tmwsimam-input.txt", inputData);
- String query =
+ String query =
"A = LOAD 'tmwsimam-input.txt' " +
"as (f0:chararray, f1:chararray, f2:chararray, f3:chararray, " +
"f4:chararray, f5:chararray, f6:chararray, f7:chararray, f8:chararray); " +
"B = FOREACH A GENERATE f0, f1, f2, f3, f4;" +
- "B1 = foreach B generate f0, f1, f2;" +
- "C = GROUP B1 BY (f1, f2);" +
+ "B1 = foreach B generate f0, f1, f2;" +
+ "C = GROUP B1 BY (f1, f2);" +
"STORE C into 'foo1' using BinStorage();" +
- "B2 = FOREACH B GENERATE f0, f3, f4;" +
+ "B2 = FOREACH B GENERATE f0, f3, f4;" +
"E = GROUP B2 BY (f3, f4);" +
"STORE E into 'foo2' using BinStorage();" +
"F = FOREACH A GENERATE f0, f5, f6, f7, f8;" +
"F1 = FOREACH F GENERATE f0, f5, f6;" +
"G = GROUP F1 BY (f5, f6);" +
- "STORE G into 'foo3' using BinStorage();" +
+ "STORE G into 'foo3' using BinStorage();" +
"F2 = FOREACH F GENERATE f0, f7, f8;" +
"I = GROUP F2 BY (f7, f8);" +
"STORE I into 'foo4' using BinStorage();" +
@@ -470,25 +469,25 @@ public class TestMultiQueryBasic {
myPig.setBatchOn();
Util.registerMultiLineQuery(myPig, query);
myPig.executeBatch();
-
+
String templateLoad = "a = load 'foo' using BinStorage();";
-
+
Map<Tuple, DataBag> expectedResults = new HashMap<Tuple, DataBag>();
- expectedResults.put((Tuple)Util.getPigConstant("('a','b')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('a','b')"),
(DataBag)Util.getPigConstant("{('1','a','b'),('2','a','b')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('c','d')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('c','d')"),
(DataBag)Util.getPigConstant("{('3','c','d'),('4','c','d')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('e','f')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('e','f')"),
(DataBag)Util.getPigConstant("{('1','e','f'),('2','e','f')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('g','h')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('g','h')"),
(DataBag)Util.getPigConstant("{('3','g','h'),('4','g','h')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('i','j')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('i','j')"),
(DataBag)Util.getPigConstant("{('1','i','j'),('2','i','j')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('k','l')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('k','l')"),
(DataBag)Util.getPigConstant("{('3','k','l'),('4','k','l')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('m','n')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('m','n')"),
(DataBag)Util.getPigConstant("{('1','m','n'),('2','m','n')}"));
- expectedResults.put((Tuple)Util.getPigConstant("('o','p')"),
+ expectedResults.put((Tuple)Util.getPigConstant("('o','p')"),
(DataBag)Util.getPigConstant("{('3','o','p'),('4','o','p')}"));
String[] outputDirs = { "foo1", "foo2", "foo3", "foo4" };
for(int k = 0; k < outputDirs.length; k++) {
@@ -504,11 +503,11 @@ public class TestMultiQueryBasic {
}
// cleanup
for (int j = 0; j < toClean.length; j++) {
- Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), toClean[j]);
+ Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), toClean[j]);
}
-
+
}
-
+
@Test
public void testMultiQueryWithTwoStores2Execs() throws Exception {
@@ -548,11 +547,11 @@ public class TestMultiQueryBasic {
myPig.executeBatch();
myPig.discardBatch();
}
-
+
/**
* Test that pig calls checkOutputSpecs() method of the OutputFormat (if the
- * StoreFunc defines an OutputFormat as the return value of
- * {@link StoreFunc#getStorePreparationClass()}
+ * StoreFunc defines an OutputFormat as the return value of
+ * {@link StoreFunc#getStorePreparationClass()}
* @throws IOException
*/
@Test
@@ -567,13 +566,13 @@ public class TestMultiQueryBasic {
myPig.setBatchOn();
Util.registerMultiLineQuery(myPig, query);
myPig.executeBatch();
-
+
// check that files were created as a result of the
// checkOutputSpecs() method of the OutputFormat being called
FileSystem fs = FileSystem.getLocal(new Configuration());
assertEquals(true, fs.exists(new Path("output1_checkOutputSpec_test")));
assertEquals(true, fs.exists(new Path("output2_checkOutputSpec_test")));
-
+
Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), "output1_checkOutputSpec_test");
Util.deleteFile(new PigContext(ExecType.LOCAL, new Properties()), "output2_checkOutputSpec_test");
}
@@ -615,23 +614,23 @@ public class TestMultiQueryBasic {
}
assertEquals(3, i);
}
-
+
private static final String DUMMY_STORE_WITH_OUTPUTFORMAT_CLASS
= "org.apache.pig.test.TestMultiQueryBasic\\$DummyStoreWithOutputFormat";
public static class DummyStoreWithOutputFormat extends StoreFunc {
-
+
public DummyStoreWithOutputFormat() {
}
@Override
public void putNext(Tuple f) throws IOException {
-
+
}
@Override
public void checkSchema(ResourceSchema s) throws IOException {
-
+
}
@Override
@@ -644,7 +643,7 @@ public class TestMultiQueryBasic {
public void prepareToWrite(
org.apache.hadoop.mapreduce.RecordWriter writer)
throws IOException {
-
+
}
@Override
@@ -658,21 +657,21 @@ public class TestMultiQueryBasic {
throws IOException {
Configuration conf = job.getConfiguration();
conf.set("mapred.output.dir", location);
-
+
}
-
+
@Override
public void setStoreFuncUDFContextSignature(String signature) {
}
-
+
}
-
+
@SuppressWarnings({ "unchecked" })
public static class DummyOutputFormat
extends OutputFormat<WritableComparable, Tuple> {
public DummyOutputFormat() {
-
+
}
@Override
public void checkOutputSpecs(JobContext context) throws IOException,
@@ -694,9 +693,9 @@ public class TestMultiQueryBasic {
InterruptedException {
return null;
}
-
+
}
-
+
// --------------------------------------------------------------------------
// Helper methods
Modified: pig/branches/tez/test/org/apache/pig/test/TestMultiQueryCompiler.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestMultiQueryCompiler.java?rev=1565509&r1=1565508&r2=1565509&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestMultiQueryCompiler.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestMultiQueryCompiler.java Fri Feb 7 01:18:14 2014
@@ -28,6 +28,7 @@ import java.util.Collections;
import java.util.Iterator;
import org.apache.pig.ExecType;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRExecutionEngine;
@@ -71,17 +72,17 @@ public class TestMultiQueryCompiler {
Util.copyFromLocalToCluster(cluster,
"test/org/apache/pig/test/data/passwd2", "passwd2");
}
-
+
@AfterClass
public static void tearDownAfterClass() throws IOException {
Util.deleteFile(cluster, "passwd");
Util.deleteFile(cluster, "passwd2");
cluster.shutDown();
}
-
+
@Before
public void setUp() throws Exception {
- cluster.setProperty("opt.multiquery", ""+true);
+ cluster.setProperty(PigConfiguration.OPT_MULTIQUERY, ""+true);
myPig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
deleteOutputFiles();
}
@@ -90,17 +91,17 @@ public class TestMultiQueryCompiler {
public void tearDown() throws Exception {
myPig = null;
}
-
+
@Test
public void testMultiQueryJiraPig1438() {
// test case: merge multiple distinct jobs -- one group by job, one distinct job
-
+
String INPUT_FILE = "abc";
-
- try {
+
+ try {
myPig.setBatchOn();
-
+
myPig.registerQuery("A = load '" + INPUT_FILE + "' as (col1:int, col2:int, col3:int);");
myPig.registerQuery("B1 = foreach A generate col1, col2;");
myPig.registerQuery("B2 = foreach A generate col2, col3;");
@@ -110,23 +111,23 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("D2 = foreach C2 generate B2.col2, B2.col3;");
myPig.registerQuery("store D1 into '/tmp/output1';");
myPig.registerQuery("store D2 into '/tmp/output2';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 2, 12);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 15);
- checkMRPlan(pp, 1, 1, 2);
-
+ checkMRPlan(pp, 1, 1, 2);
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
}
}
-
+
@Test
public void testMultiQueryJiraPig1060() {
- // test case:
+ // test case:
String INPUT_FILE = "pig-1060.txt";
@@ -187,7 +188,7 @@ public class TestMultiQueryCompiler {
}
}
}
-
+
@Test
public void testMultiQueryJiraPig920() {
@@ -202,19 +203,19 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("d = cogroup c by $0, b by $0;");
myPig.registerQuery("e = foreach d generate group, COUNT(c), COUNT(b);");
myPig.registerQuery("store e into '/tmp/output1';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 1, 9);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 1, 13);
checkMRPlan(pp, 1, 1, 1);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryJiraPig920_1() {
@@ -234,17 +235,17 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("g = cogroup d by $0, e by $0;");
myPig.registerQuery("g1 = foreach g generate group, COUNT(d), COUNT(e);");
myPig.registerQuery("store g1 into '/tmp/output2';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 2, 16);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 23);
checkMRPlan(pp, 2, 2, 2);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
@Test
@@ -267,20 +268,20 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("g = group f by uname;");
myPig.registerQuery("h = foreach g generate group, COUNT(f.uid);");
myPig.registerQuery("store h into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 17);
// NOTE: old way seemingly generated a useless foreach operator. Now we have one less operator. Reason unknow.
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 19);
checkMRPlan(pp, 1, 1, 1);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryWithSingleMapReduceSplittee() {
@@ -292,7 +293,7 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("b = foreach a generate uname, uid, gid;");
- myPig.registerQuery("split b into c1 if uid > 5, c2 if uid <= 5 ;");
+ myPig.registerQuery("split b into c1 if uid > 5, c2 if uid <= 5 ;");
myPig.registerQuery("f = group c2 by uname;");
myPig.registerQuery("f1 = foreach f generate group, SUM(c2.gid);");
myPig.registerQuery("store f1 into '/tmp/output1';");
@@ -301,14 +302,14 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 1, 9);
- checkMRPlan(pp, 1, 1, 1);
-
+ checkMRPlan(pp, 1, 1, 1);
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryPhase3BaseCase() {
@@ -330,21 +331,21 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("c2 = foreach c1 generate group, SUM(c.uid);");
myPig.registerQuery("store c2 into '/tmp/output2';");
myPig.registerQuery("d1 = group d by gid;");
- myPig.registerQuery("d2 = foreach d1 generate group, AVG(d.uid);");
+ myPig.registerQuery("d2 = foreach d1 generate group, AVG(d.uid);");
myPig.registerQuery("store d2 into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 18);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 25);
checkMRPlan(pp, 1, 1, 1);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryJiraPig983() {
@@ -364,19 +365,19 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("f = group d by c::gid;");
myPig.registerQuery("f1 = foreach f generate group, SUM(d.c::uid);");
myPig.registerQuery("store f1 into '/tmp/output2';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 2, 16);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 25);
checkMRPlan(pp, 1, 1, 2);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryPhase3WithoutCombiner() {
@@ -397,22 +398,22 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("c1 = group c by gid;");
myPig.registerQuery("c2 = foreach c1 generate group, SUM(c.uid) - COUNT(c.uid);");
myPig.registerQuery("store c2 into '/tmp/output2';");
- myPig.registerQuery("d1 = group d by gid;");
+ myPig.registerQuery("d1 = group d by gid;");
myPig.registerQuery("d2 = foreach d1 generate group, MAX(d.uid) - MIN(d.uid);");
myPig.registerQuery("store d2 into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 18);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 25);
checkMRPlan(pp, 1, 1, 1);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryPhase3WithMixedCombiner() {
@@ -433,21 +434,21 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("c1 = group c by gid;");
myPig.registerQuery("c2 = foreach c1 generate group, SUM(c.uid);");
myPig.registerQuery("store c2 into '/tmp/output2';");
- myPig.registerQuery("d1 = group d by gid;");
+ myPig.registerQuery("d1 = group d by gid;");
myPig.registerQuery("d2 = foreach d1 generate group, d.uname, MAX(d.uid) - MIN(d.uid);");
myPig.registerQuery("store d2 into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 18);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 25);
checkMRPlan(pp, 1, 1, 2);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
+ }
+ }
@Test
public void testMultiQueryPhase3WithDifferentMapDataTypes() {
@@ -472,19 +473,19 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("d1 = group d by $1 parallel 4;");
myPig.registerQuery("d2 = foreach d1 generate group, COUNT(d.uid);");
myPig.registerQuery("store d2 into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 18);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 25);
checkMRPlan(pp, 1, 1, 1);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryPhase3StreamingInReducer() {
@@ -502,23 +503,23 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("store D into '/tmp/output1';");
myPig.registerQuery("E = group A4 by $2;");
myPig.registerQuery("F = foreach E generate group, COUNT(A4);");
- myPig.registerQuery("store F into '/tmp/output2';");
+ myPig.registerQuery("store F into '/tmp/output2';");
myPig.registerQuery("G = group A1 by $2;");
- myPig.registerQuery("H = foreach G generate group, COUNT(A1);");
+ myPig.registerQuery("H = foreach G generate group, COUNT(A1);");
myPig.registerQuery("store H into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 18);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 24);
checkMRPlan(pp, 1, 1, 2);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryWithPigMixL12() {
@@ -530,7 +531,7 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("b = foreach a generate uname, passwd, uid, gid;");
- myPig.registerQuery("split b into c1 if uid > 5, c2 if uid <= 5 ;");
+ myPig.registerQuery("split b into c1 if uid > 5, c2 if uid <= 5 ;");
myPig.registerQuery("split c1 into d1 if gid < 5, d2 if gid >= 5;");
myPig.registerQuery("e = group d1 by uname;");
myPig.registerQuery("e1 = foreach e generate group, MAX(d1.uid);");
@@ -546,14 +547,14 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 23);
- checkMRPlan(pp, 1, 1, 1);
-
+ checkMRPlan(pp, 1, 1, 1);
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithCoGroup() {
@@ -565,7 +566,7 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("store a into '/tmp/output1' using BinStorage();");
- myPig.registerQuery("b = load '/tmp/output1' using BinStorage() as (uname, passwd, uid, gid);");
+ myPig.registerQuery("b = load '/tmp/output1' using BinStorage() as (uname, passwd, uid, gid);");
myPig.registerQuery("c = load 'passwd2' " +
"using PigStorage(':') as (uname, passwd, uid, gid);");
myPig.registerQuery("d = cogroup b by (uname, uid) inner, c by (uname, uid) inner;");
@@ -576,14 +577,14 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 2, 1, 11);
- checkMRPlan(pp, 1, 1, 2);
+ checkMRPlan(pp, 1, 1, 2);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithFJ() {
@@ -607,14 +608,14 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 2, 3, 16);
- checkMRPlan(pp, 2, 1, 3);
-
+ checkMRPlan(pp, 2, 1, 3);
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithExplicitSplitAndSideFiles() {
@@ -636,13 +637,13 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 19);
- checkMRPlan(pp, 1, 1, 2);
+ checkMRPlan(pp, 1, 1, 2);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
+ }
+ }
@Test
public void testMultiQueryWithExplicitSplitAndOrderByAndSideFiles() {
@@ -667,22 +668,22 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 21);
- checkMRPlan(pp, 1, 1, 4);
+ checkMRPlan(pp, 1, 1, 4);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryWithIntermediateStores() {
System.out.println("===== multi-query with intermediate stores =====");
- try {
+ try {
myPig.setBatchOn();
-
+
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname:chararray, passwd:chararray, uid:int, gid:int);");
myPig.registerQuery("store a into '/tmp/output1';");
@@ -693,13 +694,13 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 1, 5);
- checkMRPlan(pp, 1, 1, 2);
+ checkMRPlan(pp, 1, 1, 2);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
+ }
+ }
@Test
public void testMultiQueryWithImplicitSplitAndSideFiles() {
@@ -724,13 +725,13 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 22);
- checkMRPlan(pp, 1, 1, 2);
+ checkMRPlan(pp, 1, 1, 2);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
+ }
+ }
@Test
public void testMultiQueryWithTwoLoadsAndTwoStores() {
@@ -752,19 +753,19 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("h = filter f by c::gid > 5;");
myPig.registerQuery("store g into '/tmp/output1';");
myPig.registerQuery("store h into '/tmp/output2';");
-
+
LogicalPlan lp = checkLogicalPlan(2, 2, 13);
PhysicalPlan pp = checkPhysicalPlan(lp, 2, 2, 20);
- checkMRPlan(pp, 1, 1, 2);
+ checkMRPlan(pp, 1, 1, 2);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithSplitInReduce() {
@@ -786,14 +787,14 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 13);
- checkMRPlan(pp, 1, 1, 1);
+ checkMRPlan(pp, 1, 1, 1);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithSplitInReduceAndReduceSplitee() {
@@ -817,14 +818,14 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 17);
- checkMRPlan(pp, 1, 1, 2);
-
+ checkMRPlan(pp, 1, 1, 2);
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryWithSplitInReduceAndReduceSplitees() {
@@ -846,18 +847,18 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("f1 = group f by $1;");
myPig.registerQuery("f2 = foreach f1 generate group, COUNT(f.$0);");
myPig.registerQuery("store f2 into '/tmp/output2';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 2, 15);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 22);
- checkMRPlan(pp, 1, 1, 2);
-
+ checkMRPlan(pp, 1, 1, 2);
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
+ }
+ }
@Test
public void testMultiQueryWithSplitInReduceAndReduceSpliteesAndMore() {
@@ -886,19 +887,19 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("f4 = group f3 by $1;");
myPig.registerQuery("f5 = foreach f4 generate group, COUNT(f3.$0);");
myPig.registerQuery("store f5 into '/tmp/output2';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 2, 21);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 2, 32);
- checkMRPlan(pp, 1, 2, 4);
+ checkMRPlan(pp, 1, 2, 4);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryWithSplitInMapAndReduceSplitees() {
@@ -922,19 +923,19 @@ public class TestMultiQueryCompiler {
myPig.registerQuery("d1 = group d by $1;");
myPig.registerQuery("d2 = foreach d1 generate group, COUNT(d.uid);");
myPig.registerQuery("store d2 into '/tmp/output3';");
-
+
LogicalPlan lp = checkLogicalPlan(1, 3, 18);
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 25);
checkMRPlan(pp, 1, 1, 1);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
-
+ }
+ }
+
@Test
public void testMultiQueryWithTwoStores() {
@@ -959,7 +960,7 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
@Test
@@ -983,12 +984,12 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 14);
- checkMRPlan(pp, 1, 1, 1);
+ checkMRPlan(pp, 1, 1, 1);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
@Test
@@ -1014,18 +1015,18 @@ public class TestMultiQueryCompiler {
PhysicalPlan pp = checkPhysicalPlan(lp, 2, 3, 19);
- checkMRPlan(pp, 2, 1, 3);
+ checkMRPlan(pp, 2, 1, 3);
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
@Test
public void testStoreOrder() {
System.out.println("===== multi-query store order =====");
-
+
try {
myPig.setBatchOn();
myPig.registerQuery("a = load 'passwd';");
@@ -1046,7 +1047,7 @@ public class TestMultiQueryCompiler {
MROperPlan mp = checkMRPlan(pp, 1, 3, 5);
myPig.executeBatch();
- myPig.discardBatch();
+ myPig.discardBatch();
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output1"));
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output2"));
@@ -1054,17 +1055,17 @@ public class TestMultiQueryCompiler {
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output4"));
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output5"));
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testUnnecessaryStoreRemoval() {
System.out.println("===== multi-query unnecessary stores =====");
-
+
try {
myPig.setBatchOn();
myPig.registerQuery("a = load 'passwd' " +
@@ -1090,11 +1091,11 @@ public class TestMultiQueryCompiler {
checkPhysicalPlan(mo1.mapPlan, 1, 1, 3);
checkPhysicalPlan(mo1.reducePlan, 1, 1, 2);
PhysicalOperator leaf = mo1.reducePlan.getLeaves().get(0);
-
+
Assert.assertTrue(leaf instanceof POSplit);
POSplit split = (POSplit)leaf;
-
+
int i = 0;
for (PhysicalPlan p: split.getPlans()) {
checkPhysicalPlan(p, 1, 1, 1);
@@ -1102,38 +1103,38 @@ public class TestMultiQueryCompiler {
}
Assert.assertEquals(i,2);
-
+
checkPhysicalPlan(mo2.mapPlan, 1, 1, 2);
checkPhysicalPlan(mo2.reducePlan, 1, 1, 2);
leaf = mo2.reducePlan.getLeaves().get(0);
-
+
Assert.assertTrue(leaf instanceof POStore);
checkPhysicalPlan(mo3.mapPlan, 1, 1, 2);
checkPhysicalPlan(mo3.reducePlan, 1, 1, 2);
leaf = mo3.reducePlan.getLeaves().get(0);
-
+
Assert.assertTrue(leaf instanceof POStore);
myPig.executeBatch();
- myPig.discardBatch();
+ myPig.discardBatch();
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output1"));
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output2"));
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output3"));
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output4"));
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testUnnecessaryStoreRemovalCollapseSplit() {
System.out.println("===== multi-query unnecessary stores collapse split =====");
-
+
try {
myPig.setBatchOn();
myPig.registerQuery("a = load 'passwd' " +
@@ -1154,32 +1155,32 @@ public class TestMultiQueryCompiler {
checkPhysicalPlan(mo1.mapPlan, 1, 1, 3);
checkPhysicalPlan(mo1.reducePlan, 1, 1, 2);
PhysicalOperator leaf = mo1.reducePlan.getLeaves().get(0);
-
+
Assert.assertTrue(leaf instanceof POStore);
-
+
checkPhysicalPlan(mo2.mapPlan, 1, 1, 2);
checkPhysicalPlan(mo2.reducePlan, 1, 1, 2);
leaf = mo2.reducePlan.getLeaves().get(0);
-
+
Assert.assertTrue(leaf instanceof POStore);
myPig.executeBatch();
- myPig.discardBatch();
+ myPig.discardBatch();
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output1"));
Assert.assertTrue(myPig.getPigContext().getDfs().isContainer("/tmp/output2"));
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testEmptyFilterRemoval() {
System.out.println("===== multi-query empty filters =====");
-
+
try {
myPig.setBatchOn();
myPig.registerQuery("a = load 'passwd' " +
@@ -1199,9 +1200,9 @@ public class TestMultiQueryCompiler {
checkPhysicalPlan(mo.mapPlan, 1, 1, 4);
PhysicalOperator leaf = mo.mapPlan.getLeaves().get(0);
-
+
Assert.assertTrue(leaf instanceof POSplit);
-
+
POSplit split = (POSplit)leaf;
int i = 0;
@@ -1218,9 +1219,9 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithDescribe() {
@@ -1232,7 +1233,7 @@ public class TestMultiQueryCompiler {
+ "b = filter a by uid > 5;"
+ "describe b;"
+ "store b into '/tmp/output1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -1241,7 +1242,7 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
@Test
@@ -1255,7 +1256,7 @@ public class TestMultiQueryCompiler {
+ "b = filter a by uid > 5;"
+ "illustrate b;"
+ "store b into '/tmp/output1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -1264,9 +1265,9 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testMultiQueryWithExplain() {
@@ -1278,7 +1279,7 @@ public class TestMultiQueryCompiler {
+ "b = filter a by uid > 5;"
+ "explain b;"
+ "store b into '/tmp/output1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -1287,7 +1288,7 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
@Test
@@ -1301,7 +1302,7 @@ public class TestMultiQueryCompiler {
+ "b = filter a by uid > 5;"
+ "dump b;"
+ "store b into '/tmp/output1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -1310,14 +1311,14 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
+ }
}
-
+
@Test
public void testEmptyExecute() {
-
+
System.out.println("==== empty execute ====");
-
+
try {
myPig.setBatchOn();
myPig.executeBatch();
@@ -1329,7 +1330,7 @@ public class TestMultiQueryCompiler {
Assert.fail();
}
}
-
+
// A test to verify there is no cycle in the plan if the load location is the same as the store location
@Test
public void testLoadStoreLoop() {
@@ -1345,7 +1346,7 @@ public class TestMultiQueryCompiler {
Assert.fail();
}
}
-
+
@Test
public void testMultiQueryWithNoStore2() {
@@ -1367,15 +1368,15 @@ public class TestMultiQueryCompiler {
Assert.fail();
}
}
-
+
@Test
public void testMultiQueryWithCustomPartitioner() {
System.out.println("===== multi-query with intermediate stores =====");
- try {
+ try {
myPig.setBatchOn();
-
+
myPig.registerQuery("a = load 'passwd' " +
"using PigStorage(':') as (uname:chararray, passwd:chararray, uid:int, gid:int);");
myPig.registerQuery("l = FILTER a BY uname == 'foo';");
@@ -1400,8 +1401,8 @@ public class TestMultiQueryCompiler {
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
- }
- }
+ }
+ }
@Test
public void testMultiQueryDoNotMergeMRwithDifferentPartitioners() {
@@ -1441,11 +1442,11 @@ public class TestMultiQueryCompiler {
}
}
-
+
// --------------------------------------------------------------------------
// Helper methods
- private <T extends OperatorPlan<? extends Operator<?>>>
+ private <T extends OperatorPlan<? extends Operator<?>>>
void showPlanOperators(T p) {
System.out.println("Operators:");
@@ -1463,7 +1464,7 @@ public class TestMultiQueryCompiler {
ParseException {
System.out.println("===== check logical plan =====");
-
+
LogicalPlan lp = null;
try {
@@ -1484,7 +1485,7 @@ public class TestMultiQueryCompiler {
}
showLPOperators(lp);
-
+
ByteArrayOutputStream out = new ByteArrayOutputStream();
System.out.println("Logical Plan: " + lp );
@@ -1511,21 +1512,21 @@ public class TestMultiQueryCompiler {
int expectedLeaves, int expectedSize) throws IOException {
System.out.println("===== check physical plan =====");
-
+
showPlanOperators(pp);
-
+
ByteArrayOutputStream out = new ByteArrayOutputStream();
pp.explain(out);
-
+
System.out.println("===== Display Physical Plan =====");
System.out.println(out.toString());
-
+
Assert.assertEquals(expectedRoots, pp.getRoots().size());
Assert.assertEquals(expectedLeaves, pp.getLeaves().size());
Assert.assertEquals(expectedSize, pp.size());
}
-
+
private PhysicalPlan checkPhysicalPlan(LogicalPlan lp, int expectedRoots,
int expectedLeaves, int expectedSize) throws IOException {
@@ -1535,7 +1536,7 @@ public class TestMultiQueryCompiler {
lp, null);
showPlanOperators(pp);
-
+
ByteArrayOutputStream out = new ByteArrayOutputStream();
pp.explain(out);
@@ -1555,7 +1556,7 @@ public class TestMultiQueryCompiler {
System.out.println("===== check map-reduce plan =====");
MapRedUtil.checkLeafIsStore(pp, myPig.getPigContext());
-
+
MapReduceLauncher launcher = new MapReduceLauncher();
MROperPlan mrp = null;
@@ -1579,20 +1580,20 @@ public class TestMultiQueryCompiler {
e.printStackTrace();
Assert.fail();
}
- }
+ }
showPlanOperators(mrp);
-
+
System.out.println("===== Display map-reduce Plan =====");
System.out.println(mrp.toString());
-
+
Assert.assertEquals(expectedRoots, mrp.getRoots().size());
Assert.assertEquals(expectedLeaves, mrp.getLeaves().size());
Assert.assertEquals(expectedSize, mrp.size());
return mrp;
}
-
+
private void deleteOutputFiles() {
try {
FileLocalizer.delete("/tmp/output1", myPig.getPigContext());
Modified: pig/branches/tez/test/org/apache/pig/test/TestMultiQueryLocal.java
URL: http://svn.apache.org/viewvc/pig/branches/tez/test/org/apache/pig/test/TestMultiQueryLocal.java?rev=1565509&r1=1565508&r2=1565509&view=diff
==============================================================================
--- pig/branches/tez/test/org/apache/pig/test/TestMultiQueryLocal.java (original)
+++ pig/branches/tez/test/org/apache/pig/test/TestMultiQueryLocal.java Fri Feb 7 01:18:14 2014
@@ -18,15 +18,13 @@
package org.apache.pig.test;
import java.io.BufferedReader;
-import java.io.DataOutputStream;
+import java.io.File;
import java.io.FileReader;
-import java.io.StringReader;
import java.io.IOException;
-import java.io.File;
-import java.util.Iterator;
-import java.util.List;
+import java.io.StringReader;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.Properties;
import junit.framework.Assert;
@@ -36,6 +34,7 @@ import org.apache.hadoop.mapreduce.Outpu
import org.apache.hadoop.mapreduce.OutputFormat;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.pig.ExecType;
+import org.apache.pig.PigConfiguration;
import org.apache.pig.PigException;
import org.apache.pig.PigServer;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRExecutionEngine;
@@ -45,13 +44,13 @@ import org.apache.pig.backend.hadoop.exe
import org.apache.pig.builtin.PigStorage;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
import org.apache.pig.impl.plan.Operator;
import org.apache.pig.impl.plan.OperatorPlan;
-import org.apache.pig.impl.PigContext;
-import org.apache.pig.tools.grunt.GruntParser;
import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.newplan.logical.relational.LogicalPlan;
+import org.apache.pig.tools.grunt.GruntParser;
import org.apache.pig.tools.pigscript.parser.ParseException;
import org.apache.pig.tools.pigstats.JobStats;
import org.apache.pig.tools.pigstats.PigStats;
@@ -67,7 +66,7 @@ public class TestMultiQueryLocal {
@Before
public void setUp() throws Exception {
PigContext context = new PigContext(ExecType.LOCAL, new Properties());
- context.getProperties().setProperty("opt.multiquery", ""+true);
+ context.getProperties().setProperty(PigConfiguration.OPT_MULTIQUERY, ""+true);
myPig = new PigServer(context);
myPig.getPigContext().getProperties().setProperty("pig.usenewlogicalplan", "false");
myPig.getPigContext().getProperties().setProperty("pig.temp.dir", "build/test/tmp/");
@@ -114,7 +113,7 @@ public class TestMultiQueryLocal {
@Test
public void testEmptyExecute() {
System.out.println("=== test empty execute ===");
-
+
try {
myPig.setBatchOn();
myPig.executeBatch();
@@ -156,7 +155,7 @@ public class TestMultiQueryLocal {
public void testMultiQueryWithTwoStores2Execs() {
System.out.println("===== test multi-query with 2 stores (2) =====");
-
+
try {
myPig.setBatchOn();
@@ -317,10 +316,10 @@ public class TestMultiQueryLocal {
myPig.registerQuery("group b by gid;");
LogicalPlan lp = checkLogicalPlan(0, 0, 0);
-
+
// XXX Physical plan has one less node in the local case
PhysicalPlan pp = checkPhysicalPlan(lp, 0, 0, 0);
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
@@ -443,7 +442,7 @@ public class TestMultiQueryLocal {
+ "b = filter a by uid > 5;"
+ "explain b;"
+ "store b into '" + TMP_DIR + "/Pig-TestMultiQueryLocal1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -468,7 +467,7 @@ public class TestMultiQueryLocal {
+ "b = filter a by uid > 5;"
+ "dump b;"
+ "store b into '" + TMP_DIR + "/Pig-TestMultiQueryLocal1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -493,7 +492,7 @@ public class TestMultiQueryLocal {
+ "b = filter a by uid > 5;"
+ "describe b;"
+ "store b into '" + TMP_DIR + "/Pig-TestMultiQueryLocal1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
parser.setParams(myPig);
@@ -518,7 +517,7 @@ public class TestMultiQueryLocal {
+ "b = filter a by uid > 5;"
+ "illustrate b;"
+ "store b into '" + TMP_DIR + "/Pig-TestMultiQueryLocal1';\n";
-
+
GruntParser parser = new GruntParser(new StringReader(script));
parser.setInteractive(false);
myPig.getPigContext().getProperties().setProperty("pig.usenewlogicalplan", "true");
@@ -537,7 +536,7 @@ public class TestMultiQueryLocal {
@Test
public void testStoreOrder() {
System.out.println("===== multi-query store order =====");
-
+
try {
myPig.setBatchOn();
myPig.registerQuery("a = load 'test/org/apache/pig/test/data/passwd';");
@@ -557,7 +556,7 @@ public class TestMultiQueryLocal {
PhysicalPlan pp = checkPhysicalPlan(lp, 1, 3, 15);
myPig.executeBatch();
- myPig.discardBatch();
+ myPig.discardBatch();
Assert.assertTrue(new File(TMP_DIR + "/Pig-TestMultiQueryLocal1").exists());
Assert.assertTrue(new File(TMP_DIR + "/Pig-TestMultiQueryLocal2").exists());
@@ -565,7 +564,7 @@ public class TestMultiQueryLocal {
Assert.assertTrue(new File(TMP_DIR + "/Pig-TestMultiQueryLocal4").exists());
Assert.assertTrue(new File(TMP_DIR + "/Pig-TestMultiQueryLocal5").exists());
-
+
} catch (Exception e) {
e.printStackTrace();
Assert.fail();
@@ -577,7 +576,7 @@ public class TestMultiQueryLocal {
// --------------------------------------------------------------------------
// Helper methods
- private <T extends OperatorPlan<? extends Operator<?>>>
+ private <T extends OperatorPlan<? extends Operator<?>>>
void showPlanOperators(T p) {
System.out.println("Operators:");
@@ -595,7 +594,7 @@ public class TestMultiQueryLocal {
ParseException {
System.out.println("===== check logical plan =====");
-
+
LogicalPlan lp = null;
try {
@@ -627,7 +626,7 @@ public class TestMultiQueryLocal {
private PhysicalPlan checkPhysicalPlan(LogicalPlan lp, int expectedRoots,
int expectedLeaves, int expectedSize) throws IOException {
- System.out.println("===== check physical plan =====");
+ System.out.println("===== check physical plan =====");
PhysicalPlan pp = ((MRExecutionEngine)myPig.getPigContext().getExecutionEngine()).compile(
lp, null);
@@ -675,14 +674,14 @@ public class TestMultiQueryLocal {
deleteDir( new File( outputFile ) );
} else {
FileLocalizer.delete(outputFile, myPig.getPigContext());
- }
- }
+ }
+ }
} catch (IOException e) {
e.printStackTrace();
Assert.fail();
}
}
-
+
private void deleteDir( File file ) {
if( file.isDirectory() && file.listFiles().length != 0 ) {
for( File innerFile : file.listFiles() ) {
@@ -691,7 +690,7 @@ public class TestMultiQueryLocal {
}
file.delete();
}
-
+
private boolean isDirectory( String filepath ) {
File file = new File( filepath );
return file.isDirectory();