You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by ha...@apache.org on 2010/04/01 23:44:35 UTC
svn commit: r930108 [2/2] - in /hadoop/pig/trunk: ./
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/
src/org/apache/pig/backend/hadoop/executionengine/physicalLay...
Added: hadoop/pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java?rev=930108&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestMapSideCogroup.java Thu Apr 1 21:44:34 2010
@@ -0,0 +1,425 @@
+/*
+ * 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.pig.test;
+
+
+import java.io.IOException;
+import java.util.Iterator;
+
+import junit.framework.TestCase;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.pig.CollectableLoadFunc;
+import org.apache.pig.ExecType;
+import org.apache.pig.IndexableLoadFunc;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceOper;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.MROperPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeCogroup;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DefaultTuple;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.logicalLayer.LOCogroup;
+import org.apache.pig.impl.logicalLayer.LogicalPlan;
+import org.apache.pig.test.utils.LogicalPlanTester;
+import org.junit.After;
+import org.junit.Before;
+
+public class TestMapSideCogroup extends TestCase{
+
+ private static final String INPUT_FILE1 = "testCogrpInput1.txt";
+ private static final String INPUT_FILE2 = "testCogrpInput2.txt";
+ private static final String INPUT_FILE3 = "testCogrpInput3.txt";
+ private static final String INPUT_FILE4 = "testCogrpInput4.txt";
+ private static final String INPUT_FILE5 = "testCogrpInput5.txt";
+ private static final String EMPTY_FILE = "empty.txt";
+ private static final String DATA_WITH_NULL_KEYS = "null.txt";
+
+ private MiniCluster cluster = MiniCluster.buildCluster();
+
+ @Before
+ public void setUp() throws Exception {
+
+ Util.deleteFile(cluster, INPUT_FILE1);
+ Util.deleteFile(cluster, INPUT_FILE2);
+ Util.deleteFile(cluster, INPUT_FILE3);
+ Util.deleteFile(cluster, INPUT_FILE4);
+ Util.deleteFile(cluster, INPUT_FILE5);
+ Util.deleteFile(cluster, EMPTY_FILE);
+ int LOOP_SIZE = 3;
+ String[] input = new String[LOOP_SIZE*LOOP_SIZE];
+ int k = 0;
+ for(int i = 1; i <= LOOP_SIZE; i++) {
+ String si = i + "";
+ for(int j=1;j<=LOOP_SIZE;j++)
+ input[k++] = si + "\t" + j;
+ }
+
+ String[] input2 = new String[LOOP_SIZE*2*LOOP_SIZE];
+ k = 0;
+ for(int i = LOOP_SIZE + 1; i <= 3*LOOP_SIZE ; i++) {
+ String si = i + "";
+ for(int j=1;j<=LOOP_SIZE;j++)
+ input2[k++] = si + "\t" + j;
+ }
+
+ String[] input3 = new String[LOOP_SIZE*LOOP_SIZE];
+ k = 0;
+ for(int i = LOOP_SIZE ; i < 2*LOOP_SIZE ; i++) {
+ String si = i + "";
+ for(int j=1;j<=LOOP_SIZE;j++)
+ input3[k++] = si + "\t" + j;
+ }
+
+ String[] dataWithNullKeys = new String[LOOP_SIZE*LOOP_SIZE];
+ k = 0;
+ for(int i = 1; i <= LOOP_SIZE ; i++) {
+ String si;
+ if(i == 1)
+ si = "";
+ else
+ si = i + "";
+ for(int j=1;j<=LOOP_SIZE;j++){
+ dataWithNullKeys[k++] = si + "\t" + j;
+ }
+
+ }
+
+ Util.createInputFile(cluster, INPUT_FILE1, input);
+ Util.createInputFile(cluster, INPUT_FILE2, input);
+ Util.createInputFile(cluster, INPUT_FILE3, input);
+ Util.createInputFile(cluster, INPUT_FILE4, input2);
+ Util.createInputFile(cluster, INPUT_FILE5, input3);
+ Util.createInputFile(cluster, EMPTY_FILE, new String[]{});
+ Util.createInputFile(cluster, DATA_WITH_NULL_KEYS, dataWithNullKeys);
+
+
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ Util.deleteFile(cluster, INPUT_FILE1);
+ Util.deleteFile(cluster, INPUT_FILE2);
+ Util.deleteFile(cluster, INPUT_FILE3);
+ Util.deleteFile(cluster, INPUT_FILE4);
+ Util.deleteFile(cluster, INPUT_FILE5);
+ Util.deleteFile(cluster, EMPTY_FILE);
+ Util.deleteFile(cluster, DATA_WITH_NULL_KEYS);
+ }
+
+ public void testCompilation(){
+ try{
+ LogicalPlanTester lpt = new LogicalPlanTester();
+ lpt.buildPlan("A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);");
+ lpt.buildPlan("B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);");
+ lpt.buildPlan("D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);");
+ LogicalPlan lp = lpt.buildPlan("C = cogroup A by id, B by id, D by id using 'merge';");
+ assertEquals(LOCogroup.GROUPTYPE.MERGE, ((LOCogroup)lp.getLeaves().get(0)).getGroupType());
+
+ PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
+ pc.connect();
+ PhysicalPlan phyP = Util.buildPhysicalPlan(lp, pc);
+ PhysicalOperator phyOp = phyP.getLeaves().get(0);
+ assertTrue(phyOp instanceof POMergeCogroup);
+
+ lp = lpt.buildPlan("store C into 'out';");
+ MROperPlan mrPlan = Util.buildMRPlan(Util.buildPhysicalPlan(lp, pc),pc);
+ assertEquals(2,mrPlan.size());
+
+ Iterator<MapReduceOper> itr = mrPlan.iterator();
+ MapReduceOper oper = itr.next();
+ assertTrue(oper.reducePlan.isEmpty());
+ assertFalse(oper.mapPlan.isEmpty());
+
+ oper = itr.next();
+ assertFalse(oper.reducePlan.isEmpty());
+ assertFalse(oper.mapPlan.isEmpty());
+
+
+ } catch(Exception e){
+ e.printStackTrace();
+ fail("Compilation of merged cogroup failed.");
+ }
+
+ }
+
+ public void testSimple() throws Exception{
+
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + INPUT_FILE1 + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+ pigServer.registerQuery("B = LOAD '" + INPUT_FILE2 + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+
+ DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+ pigServer.registerQuery("C = cogroup A by c1, B by c1 using 'merge';");
+ Iterator<Tuple> iter = pigServer.openIterator("C");
+
+ while(iter.hasNext()) {
+ Tuple t = iter.next();
+ dbMergeCogrp.add(t);
+ }
+
+ String[] results = new String[]{
+ "(1,{(1,1),(1,2),(1,3)},{(1,1),(1,2),(1,3)})",
+ "(2,{(2,2),(2,1),(2,3)},{(2,1),(2,2),(2,3)})",
+ "(3,{(3,3),(3,2),(3,1)},{(3,1),(3,2),(3,3)})"
+ };
+
+ assertEquals(3, dbMergeCogrp.size());
+ Iterator<Tuple> itr = dbMergeCogrp.iterator();
+ for(int i=0; i<3; i++){
+ assertEquals(itr.next().toString(), results[i]);
+ }
+ assertFalse(itr.hasNext());
+ }
+
+ public void test3Way() throws Exception{
+
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + INPUT_FILE1 + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+ pigServer.registerQuery("B = LOAD '" + INPUT_FILE2 + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+ pigServer.registerQuery("E = LOAD '" + INPUT_FILE3 + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+
+ DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+
+ pigServer.registerQuery("C = cogroup A by c1, B by c1, E by c1 using 'merge';");
+ Iterator<Tuple> iter = pigServer.openIterator("C");
+
+ while(iter.hasNext()) {
+ Tuple t = iter.next();
+ dbMergeCogrp.add(t);
+ }
+
+
+ String[] results = new String[]{
+ "(1,{(1,1),(1,2),(1,3)},{(1,1),(1,2),(1,3)},{(1,1),(1,2),(1,3)})",
+ "(2,{(2,2),(2,1),(2,3)},{(2,1),(2,2),(2,3)},{(2,1),(2,2),(2,3)})",
+ "(3,{(3,2),(3,3),(3,1)},{(3,1),(3,2),(3,3)},{(3,1),(3,2),(3,3)})"
+ };
+
+ assertEquals(3, dbMergeCogrp.size());
+ Iterator<Tuple> itr = dbMergeCogrp.iterator();
+ for(int i=0; i<3; i++){
+ assertEquals(itr.next().toString(), results[i]);
+ }
+ assertFalse(itr.hasNext());
+
+ }
+
+ public void testMultiSplits() throws Exception{
+
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + INPUT_FILE1 + "," + INPUT_FILE4 + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+ pigServer.registerQuery("B = LOAD '" + INPUT_FILE5 + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+
+ DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+
+ pigServer.registerQuery("C = cogroup A by c1, B by c1 using 'merge';");
+ Iterator<Tuple> iter = pigServer.openIterator("C");
+
+
+ while(iter.hasNext()) {
+ Tuple t = iter.next();
+ dbMergeCogrp.add(t);
+ }
+
+
+ String[] results = new String[]{
+ "(4,{(4,1),(4,2),(4,3)},{(4,1),(4,2),(4,3)})",
+ "(5,{(5,2),(5,1),(5,3)},{(5,1),(5,2),(5,3)})",
+ "(6,{(6,1),(6,3),(6,2)},{})",
+ "(7,{(7,2),(7,1),(7,3)},{})",
+ "(8,{(8,3),(8,1),(8,2)},{})",
+ "(9,{(9,1),(9,3),(9,2)},{})",
+ "(1,{(1,1),(1,2),(1,3)},{})",
+ "(2,{(2,1),(2,2),(2,3)},{})",
+ "(3,{(3,3),(3,2),(3,1)},{(3,1),(3,2),(3,3)})"
+ };
+
+ assertEquals(9, dbMergeCogrp.size());
+ Iterator<Tuple> itr = dbMergeCogrp.iterator();
+ for(int i=0; i<9; i++){
+ assertEquals(itr.next().toString(), results[i]);
+ }
+ assertFalse(itr.hasNext());
+ }
+
+ public void testCogrpOnMultiKeys() throws Exception{
+
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + INPUT_FILE1 + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:chararray);");
+ pigServer.registerQuery("B = LOAD '" + INPUT_FILE2 + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:chararray);");
+
+ DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+ pigServer.registerQuery("C = cogroup A by (c1,c2) , B by (c1,c2) using 'merge' ;");
+ Iterator<Tuple> iter = pigServer.openIterator("C");
+
+ while(iter.hasNext()) {
+ Tuple t = iter.next();
+ dbMergeCogrp.add(t);
+ }
+
+ String[] results = new String[]{
+ "((1,1),{(1,1)},{(1,1)})",
+ "((1,2),{(1,2)},{(1,2)})",
+ "((1,3),{(1,3)},{(1,3)})",
+ "((2,1),{(2,1)},{(2,1)})",
+ "((2,2),{(2,2)},{(2,2)})",
+ "((2,3),{(2,3)},{(2,3)})",
+ "((3,1),{(3,1)},{(3,1)})",
+ "((3,2),{(3,2)},{(3,2)})",
+ "((3,3),{(3,3)},{(3,3)})"
+ };
+
+ assertEquals(9, dbMergeCogrp.size());
+ Iterator<Tuple> itr = dbMergeCogrp.iterator();
+ for(int i=0; i<9; i++){
+ assertEquals(itr.next().toString(), results[i]);
+ }
+ assertFalse(itr.hasNext());
+ }
+
+ public void testEmptyDeltaFile() throws Exception{
+
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + INPUT_FILE1 + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+ pigServer.registerQuery("B = LOAD '" + EMPTY_FILE + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+
+ DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+
+ pigServer.registerQuery("C = cogroup A by c1, B by c1 using 'merge';");
+ Iterator<Tuple> iter = pigServer.openIterator("C");
+
+ while(iter.hasNext()) {
+ Tuple t = iter.next();
+ dbMergeCogrp.add(t);
+ }
+
+ String[] results = new String[]{
+ "(1,{(1,1),(1,2),(1,3)},{})",
+ "(2,{(2,1),(2,2),(2,3)},{})",
+ "(3,{(3,1),(3,3),(3,2)},{})"
+ };
+
+ assertEquals(3, dbMergeCogrp.size());
+ Iterator<Tuple> itr = dbMergeCogrp.iterator();
+ for(int i=0; i<3; i++){
+ assertEquals(itr.next().toString(), results[i]);
+ }
+ assertFalse(itr.hasNext());
+ }
+
+
+ public void testDataWithNullKeys() throws Exception{
+
+ PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ pigServer.registerQuery("A = LOAD '" + DATA_WITH_NULL_KEYS + "' using "+ DummyCollectableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+ pigServer.registerQuery("B = LOAD '" + DATA_WITH_NULL_KEYS + "' using "+ DummyIndexableLoader.class.getName() +"() as (c1:chararray,c2:int);");
+
+ String[] results = new String[]{
+ "(,{(,1),(,2),(,3)},{})",
+ "(,{},{(,1),(,2),(,3)})",
+ "(2,{(2,3),(2,1),(2,2)},{(2,1),(2,2),(2,3)})",
+ "(3,{(3,3),(3,1),(3,2)},{(3,1),(3,2),(3,3)})"
+ };
+
+ DataBag dbMergeCogrp = BagFactory.getInstance().newDefaultBag();
+
+ pigServer.registerQuery("C = cogroup A by c1, B by c1 using 'merge';");
+ Iterator<Tuple> iter = pigServer.openIterator("C");
+
+ while(iter.hasNext()) {
+ Tuple t = iter.next();
+ dbMergeCogrp.add(t);
+ }
+
+ assertEquals(4, dbMergeCogrp.size());
+ Iterator<Tuple> itr = dbMergeCogrp.iterator();
+ for(int i=0; i<4; i++){
+ assertEquals(itr.next().toString(), results[i]);
+ }
+ assertFalse(itr.hasNext());
+
+ }
+
+ /**
+ * A dummy loader which implements {@link CollectableLoadFunc}
+ */
+ public static class DummyCollectableLoader extends PigStorage implements CollectableLoadFunc{
+
+ public DummyCollectableLoader() {
+ }
+
+ @Override
+ public void ensureAllKeyInstancesInSameSplit() throws IOException {
+ }
+ }
+
+ /**
+ * A dummy loader which implements {@link IndexableLoadFunc}
+ */
+ public static class DummyIndexableLoader extends PigStorage implements IndexableLoadFunc {
+
+ private String loc;
+ private FSDataInputStream is;
+
+ public DummyIndexableLoader() {
+ }
+
+ @Override
+ public void close() throws IOException {
+ is.close();
+ }
+
+ @Override
+ public void seekNear(Tuple keys) throws IOException {
+
+ }
+
+ @Override
+ public void initialize(Configuration conf) throws IOException {
+ is = FileSystem.get(conf).open(new Path(loc));
+ }
+
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ super.setLocation(location, job);
+ loc = location;
+ }
+
+ @Override
+ public Tuple getNext() throws IOException {
+ String line = is.readLine();
+ if(line == null)
+ return null;
+ String[] members = line.split("\t");
+ DefaultTuple tuple = new DefaultTuple();
+ for(String member : members)
+ tuple.append(member);
+ return tuple;
+ }
+ }
+}
Modified: hadoop/pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld?rev=930108&r1=930107&r2=930108&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld (original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/data/GoldenFiles/MRC18.gld Thu Apr 1 21:44:34 2010
@@ -7,17 +7,15 @@ Reduce Plan Empty
| |---Load(file:///tmp/input1:org.apache.pig.builtin.PigStorage) - scope-117
|
|---MapReduce(1,PigStorage) - scope-126:
- | Store(file:/tmp/temp-1456742965/tmp-1456742965:org.apache.pig.builtin.BinStorage) - scope-133
+ | Store(file:/tmp/temp-1456742965/tmp-1456742965:org.apache.pig.builtin.BinStorage) - scope-132
| |
- | |---POSort[tuple]() - scope-132
+ | |---New For Each(true)[bag] - scope-131
| | |
- | | Project[tuple][*] - scope-131
+ | | Project[tuple][1] - scope-130
| |
- | |---Project[tuple][1] - scope-130
- | |
- | |---Package[tuple]{chararray} - scope-129
- | Local Rearrange[tuple]{chararray}(false) - scope-128
+ | |---Package[tuple]{tuple} - scope-129
+ | Local Rearrange[tuple]{tuple}(false) - scope-128
| | |
- | | Constant(all) - scope-127
+ | | Project[tuple][*] - scope-127
| |
- | |---Load(file:///tmp/input2:org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MergeJoinIndexer('org.apache.pig.builtin.PigStorage','kmonaaafhdhcaabdgkgbhggbcohfhegjgmcoebhchcgbhjemgjhdhehiibncbnjjmhgbjnadaaabejaaaehdgjhkgfhihaaaaaaaabhhaeaaaaaaabhdhcaaeogphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccohagmgbgohdcofagihjhdgjgdgbgmfagmgbgoaaaaaaaaaaaaaaabacaaabfkaaangfgogeepggebgmgmejgohahfhehihcaacfgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcfagmgbgogpkikkjgaddcgofpacaaagemaaakgneghcgpgnefgeghgfhdheaacdemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphfhegjgmcpenhfgmhegjengbhadlemaaafgnelgfhjhdheaaapemgkgbhggbcphfhegjgmcpengbhadlemaaahgnemgfgbhggfhdheaabaemgkgbhggbcphfhegjgmcpemgjhdhedlemaaaegnephahdhbaahoaaafemaaaggnfcgpgphehdhbaahoaaagemaaaignfegpefgeghgfhdhbaahoaaaehihahdhcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohfhegjgmcoenhfgmhegjengbhaaaaaaaaaaaaaaaacacaaabemaaae
gnengbhahbaahoaaafhihahdhcaabbgkgbhggbcohfhegjgmcoeigbhdgiengbhaafahnkmbmdbgganbadaaacegaaakgmgpgbgeeggbgdhegphcejaaajhegihcgfhdgigpgmgehihadpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhdhcaacegphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcelgfhjaaaaaaaaaaaaaaabacaaacekaaacgjgeemaaafhdgdgphagfheaabcemgkgbhggbcpgmgbgoghcpfdhehcgjgoghdlhihaaaaaaaaaaaaaaahiheaaafhdgdgphagfhdhcaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdheaaaaaaaaaaaaaaabacaaaffkaaakgphggfhcgmgpgbgegfgefkaabfhahcgpgdgfhdhdgjgoghecgbghepggfehfhagmgfhdfkaabehcgfhdhfgmhefdgjgoghgmgffehfhagmgfecgbghfkaaaehdhegbhcemaaahgdgpgmhfgngohdheaabfemgkgbhggbcphfhegjgmcpebhchcgbhjemgjhdhedlhihcaagcgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpg
oephagfhcgbhegphchdcoefhihahcgfhdhdgjgpgoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaadgmgpghheaacaemgphcghcpgbhagbgdgigfcpgdgpgngngpgohdcpgmgpghghgjgoghcpemgpghdlhihcaaemgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofagihjhdgjgdgbgmephagfhcgbhegphcaaaaaaaaaaaaaaabacaaamfkaaafgbgdgdhfgnfkaaangjgohahfheebhehegbgdgigfgeejaabehcgfhbhfgfhdhegfgefagbhcgbgmgmgfgmgjhdgnecaaakhcgfhdhfgmhefehjhagfemaaafgbgmgjgbhdhbaahoaaaoemaaafgjgohahfheheaablemgphcghcpgbhagbgdgigfcphagjghcpgegbhegbcpfehfhagmgfdlemaaaggjgohahfhehdhbaahoaaagemaaangmgjgogfgbghgffehcgbgdgfhcheaachemgphcghcpgbhagbgdgigfcphagjghcphagfgocphfhegjgmcpemgjgogfgbghgffehcgbgdgfhcdlemaaadgmgpghhbaahoaabeemaaahgphfhehahfhehdhbaahoaaagemaaakhagbhcgfgohefagmgbgoheaafaemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccphagmgbgohdcpfagihjhdgjgdgbgmfagmgbgodlemaaadhcgfhdheaaeeemgphcghcpgbhagbgd
gigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccpfcgfhdhfgmhedlhihcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaaegnelgfhjheaacgemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphagmgbgocpephagfhcgbhegphcelgfhjdlhihahbaahoaaapaaaappppppppdchahahahahdhcaaclgphcghcogbhagbgdgigfcogdgpgngngpgohdcogmgpghghgjgoghcogjgnhagmcoemgpghdeekemgpghghgfhccikmpnoicknfncdiacaaabemaaaegogbgngfhbaahoaaaohihaheaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdhehahahdhcaaecgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofcgfhdhfgmheaaaaaaaaaaaaaaabacaaacecaaamhcgfhehfhcgofdhegbhehfhdemaaaghcgfhdhfgmheheaabcemgkgbhggbcpgmgbgoghcpepgcgkgfgdhedlhihaachahbaahoaaboaaaaaaaahdhbaahoaaaaaaaaaaabhhaeaaaaaaabh
dhcaabbgkgbhggbcogmgbgoghcoejgohegfghgfhcbcockakephibihdiacaaabejaaafhggbgmhfgfhihcaabagkgbhggbcogmgbgoghcoeohfgngcgfhcigkmjfbnaljeoailacaaaahihaaaaaaaaahihihdhbaahoaaaaaaaaaaabhhaeaaaaaaakhbaahoaabmhihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhbaahoaabmhbaahoaaaphihdhbaahoaaaaaaaaaaaahhaeaaaaaaakhihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahiaahi','')) - scope-118
\ No newline at end of file
+ | |---Load(file:///tmp/input2:org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MergeJoinIndexer('org.apache.pig.builtin.PigStorage','kmonaaafhdhcaabdgkgbhggbcohfhegjgmcoebhchcgbhjemgjhdhehiibncbnjjmhgbjnadaaabejaaaehdgjhkgfhihaaaaaaaabhhaeaaaaaaabhdhcaaeogphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccohagmgbgohdcofagihjhdgjgdgbgmfagmgbgoaaaaaaaaaaaaaaabacaaabfkaaangfgogeepggebgmgmejgohahfhehihcaacfgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcfagmgbgogpkikkjgaddcgofpacaaagemaaakgneghcgpgnefgeghgfhdheaacdemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphfhegjgmcpenhfgmhegjengbhadlemaaafgnelgfhjhdheaaapemgkgbhggbcphfhegjgmcpengbhadlemaaahgnemgfgbhggfhdheaabaemgkgbhggbcphfhegjgmcpemgjhdhedlemaaaegnephahdhbaahoaaafemaaaggnfcgpgphehdhbaahoaaagemaaaignfegpefgeghgfhdhbaahoaaaehihahdhcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohfhegjgmcoenhfgmhegjengbhaaaaaaaaaaaaaaaacacaaabemaaae
gnengbhahbaahoaaafhihahdhcaabbgkgbhggbcohfhegjgmcoeigbhdgiengbhaafahnkmbmdbgganbadaaacegaaakgmgpgbgeeggbgdhegphcejaaajhegihcgfhdgigpgmgehihadpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhdhcaacegphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcelgfhjaaaaaaaaaaaaaaabacaaacekaaacgjgeemaaafhdgdgphagfheaabcemgkgbhggbcpgmgbgoghcpfdhehcgjgoghdlhihaaaaaaaaaaaaaaahiheaaafhdgdgphagfhdhcaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdheaaaaaaaaaaaaaaabacaaaffkaaakgphggfhcgmgpgbgegfgefkaabfhahcgpgdgfhdhdgjgoghecgbghepggfehfhagmgfhdfkaabehcgfhdhfgmhefdgjgoghgmgffehfhagmgfecgbghfkaaaehdhegbhcemaaahgdgpgmhfgngohdheaabfemgkgbhggbcphfhegjgmcpebhchcgbhjemgjhdhedlhihcaagcgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpg
oephagfhcgbhegphchdcoefhihahcgfhdhdgjgpgoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaadgmgpghheaacaemgphcghcpgbhagbgdgigfcpgdgpgngngpgohdcpgmgpghghgjgoghcpemgpghdlhihcaaemgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofagihjhdgjgdgbgmephagfhcgbhegphcaaaaaaaaaaaaaaabacaaamfkaaafgbgdgdhfgnfkaaangjgohahfheebhehegbgdgigfgeejaabehcgfhbhfgfhdhegfgefagbhcgbgmgmgfgmgjhdgnecaaakhcgfhdhfgmhefehjhagfemaaafgbgmgjgbhdhbaahoaaaoemaaafgjgohahfheheaablemgphcghcpgbhagbgdgigfcphagjghcpgegbhegbcpfehfhagmgfdlemaaaggjgohahfhehdhbaahoaaagemaaangmgjgogfgbghgffehcgbgdgfhcheaachemgphcghcpgbhagbgdgigfcphagjghcphagfgocphfhegjgmcpemgjgogfgbghgffehcgbgdgfhcdlemaaadgmgpghhbaahoaabeemaaahgphfhehahfhehdhbaahoaaagemaaakhagbhcgfgohefagmgbgoheaafaemgphcghcpgbhagbgdgigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccphagmgbgohdcpfagihjhdgjgdgbgmfagmgbgodlemaaadhcgfhdheaaeeemgphcghcpgbhagbgd
gigfcphagjghcpgcgbgdglgfgogecpgigbgegpgphacpgfhigfgdhfhegjgpgogfgoghgjgogfcphagihjhdgjgdgbgmemgbhjgfhccpfcgfhdhfgmhedlhihcaacbgphcghcogbhagbgdgigfcohagjghcogjgnhagmcohagmgbgocoephagfhcgbhegphcaaaaaaaaaaaaaaabacaaabemaaaegnelgfhjheaacgemgphcghcpgbhagbgdgigfcphagjghcpgjgnhagmcphagmgbgocpephagfhcgbhegphcelgfhjdlhihahbaahoaaapaaaappppppppdchahahahahdhcaaclgphcghcogbhagbgdgigfcogdgpgngngpgohdcogmgpghghgjgoghcogjgnhagmcoemgpghdeekemgpghghgfhccikmpnoicknfncdiacaaabemaaaegogbgngfhbaahoaaaohihaheaafjgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccogfhihahcgfhdhdgjgpgoephagfhcgbhegphchdcofaepfahcgpgkgfgdhehahahdhcaaecgphcghcogbhagbgdgigfcohagjghcogcgbgdglgfgogecogigbgegpgphacogfhigfgdhfhegjgpgogfgoghgjgogfcohagihjhdgjgdgbgmemgbhjgfhccofcgfhdhfgmheaaaaaaaaaaaaaaabacaaacecaaamhcgfhehfhcgofdhegbhehfhdemaaaghcgfhdhfgmheheaabcemgkgbhggbcpgmgbgoghcpepgcgkgfgdhedlhihaachahbaahoaaboaaaaaaaahdhbaahoaaaaaaaaaaabhhaeaaaaaaabh
dhcaabbgkgbhggbcogmgbgoghcoejgohegfghgfhcbcockakephibihdiacaaabejaaafhggbgmhfgfhihcaabagkgbhggbcogmgbgoghcoeohfgngcgfhcigkmjfbnaljeoailacaaaahihaaaaaaaaahihihdhbaahoaaaaaaaaaaabhhaeaaaaaaakhbaahoaabmhihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaabhbaahoaabmhbaahoaaaphihdhbaahoaaaaaaaaaaaahhaeaaaaaaakhihdhbaahoaaaihdhbaahoaaakdpeaaaaaaaaaaaamhhaiaaaaaabaaaaaaaaahiaahi','','b','scope','true')) - scope-118
\ No newline at end of file
Modified: hadoop/pig/trunk/test/org/apache/pig/test/utils/TestHelper.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/utils/TestHelper.java?rev=930108&r1=930107&r2=930108&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/utils/TestHelper.java (original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/utils/TestHelper.java Thu Apr 1 21:44:34 2010
@@ -18,9 +18,12 @@
package org.apache.pig.test.utils;
import java.io.*;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Random;
+import java.util.Set;
+import java.util.Map.Entry;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.data.DataBag;
@@ -56,7 +59,6 @@ public class TestHelper {
if (db1.size() != db2.size())
return false;
- int i=-1;
boolean equal = true;
for (Tuple tuple : db2) {
boolean contains = false;
@@ -70,12 +72,37 @@ public class TestHelper {
equal = false;
break;
}
- /*if(++i%dispAfterNumTuples==0)
- System.out.println(i/dispAfterNumTuples);*/
}
return equal;
}
+ public static boolean compareCogroupOutputs(DataBag db1, DataBag db2) throws ExecException{
+
+ if (db1.size() != db2.size())
+ return false;
+
+ Map<Object,DataBag> first = new HashMap<Object, DataBag>();
+ for (Tuple t : db1)
+ first.put(t.get(0), (DataBag)t.get(1));
+
+ Map<Object,DataBag> second = new HashMap<Object, DataBag>();
+ for (Tuple t : db2)
+ second.put(t.get(0), (DataBag)t.get(1));
+
+ Set<Entry<Object,DataBag>> entrySet = first.entrySet();
+
+ for (Entry<Object,DataBag> entry : entrySet){
+ Object key = entry.getKey();
+ DataBag bagOfSecond = second.get(key);
+ if(bagOfSecond == null)
+ return false;
+ boolean cmpVal = compareBags(bagOfSecond, entry.getValue());
+ if(cmpVal == false)
+ return false;
+ }
+ return true;
+ }
+
public static DataBag projectBag(DataBag db2, int i) throws ExecException {
DataBag ret = DefaultBagFactory.getInstance().newDefaultBag();
for (Tuple tuple : db2) {