test/org/apache/pig/test/TestMapSideCogroup.java - pig - Git at Google

 /*
  * 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 static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;

 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.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
 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.FrontendException;
 import org.apache.pig.newplan.Operator;
 import org.apache.pig.newplan.logical.relational.LOCogroup;
 import org.apache.pig.newplan.logical.relational.LogicalPlan;
 import org.junit.After;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.Test;

 public class TestMapSideCogroup {
     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 static MiniGenericCluster cluster = MiniGenericCluster.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);
     }

     @AfterClass
     public static void oneTimeTearDown() throws Exception {
         cluster.shutDown();
     }

     @Test
     public void testCompilation(){
         try{
             PigServer pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
             String query = "A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);" +
             "B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
             "D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
             "C = cogroup A by id, B by id, D by id using 'merge';" +
             "store C into 'output';";
             LogicalPlan lp = Util.buildLp(pigServer, query);
             Operator op = lp.getSinks().get(0);
             LOCogroup cogrp = (LOCogroup)lp.getPredecessors(op).get(0);
             assertEquals(LOCogroup.GROUPTYPE.MERGE, cogrp.getGroupType());

             PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
             pc.connect();
             PhysicalPlan phyP = Util.buildPp(pigServer, query);
             PhysicalOperator phyOp = phyP.getLeaves().get(0);
             assertTrue(phyOp instanceof POStore);
             phyOp = phyOp.getInputs().get(0);
             assertTrue(phyOp instanceof POMergeCogroup);

             MROperPlan mrPlan = Util.buildMRPlan(phyP,pc);
             assertEquals(2,mrPlan.size());

             Iterator<MapReduceOper> itr = mrPlan.iterator();
             List<MapReduceOper> opers = new ArrayList<MapReduceOper>();
             opers.add(itr.next());
             opers.add(itr.next());
             //Order of entrySet is not guaranteed with jdk1.7
             Collections.sort(opers);

             assertTrue(opers.get(0).reducePlan.isEmpty());
             assertFalse(opers.get(0).mapPlan.isEmpty());

             assertFalse(opers.get(1).reducePlan.isEmpty());
             assertFalse(opers.get(1).mapPlan.isEmpty());
         } catch(Exception e){
             e.printStackTrace();
             fail("Compilation of merged cogroup failed.");
         }

     }

 //    @Test // PIG-2018
 //    public void testFailure1() throws Exception{
 //        PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
 //        String query = "A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);" +
 //        "E = group A by id;" +
 //        "B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
 //        "D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
 //        "C = cogroup E by A.id, B by id, D by id using 'merge';" +
 //        "store C into 'output';";
 //        LogicalPlan lp = Util.buildLp(pigServer, query);
 //        Operator op = lp.getSinks().get(0);
 //        LOCogroup cogrp = (LOCogroup)lp.getPredecessors(op).get(0);
 //        assertEquals(LOCogroup.GROUPTYPE.MERGE, cogrp.getGroupType());
 //
 //        PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
 //        pc.connect();
 //        boolean exceptionCaught = false;
 //        try{
 //            Util.buildPp(pigServer, query);
 //        }catch (java.lang.reflect.InvocationTargetException e){
 //        	FrontendException ex = (FrontendException)e.getTargetException();
 //            assertEquals(1103,ex.getErrorCode());
 //            exceptionCaught = true;
 //        }
 //        assertTrue(exceptionCaught);
 //    }

     @Test
     public void testFailure2() throws Exception{
         PigServer pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
         String query = "A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);" +
         "B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
         "D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
         "C = cogroup A by id inner, B by id, D by id inner using 'merge';" +
         "store C into 'output';";
         LogicalPlan lp = Util.buildLp(pigServer, query);
         Operator op = lp.getSinks().get(0);
         LOCogroup cogrp = (LOCogroup)lp.getPredecessors(op).get(0);
         assertEquals(LOCogroup.GROUPTYPE.MERGE, cogrp.getGroupType());

         PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
         pc.connect();
         boolean exceptionCaught = false;
         try{
             Util.buildPp(pigServer, query);
         }catch (FrontendException e){
             exceptionCaught = true;
         }
         assertTrue(exceptionCaught);
     }

     @Test
     public void testSimple() throws Exception{

         PigServer pigServer = new PigServer(cluster.getExecType(), 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());
     }

     @Test
     public void test3Way() throws Exception{

         PigServer pigServer = new PigServer(cluster.getExecType(), 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());

     }

     @Test
     public void testMultiSplits() throws Exception{

         PigServer pigServer = new PigServer(cluster.getExecType(), 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);");

         List<Tuple> dbMergeCogrp = new ArrayList<Tuple>();

         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,2),(6,3)},{})",
                 "(7,{(7,1),(7,2),(7,3)},{})",
                 "(8,{(8,1),(8,2),(8,3)},{})",
                 "(9,{(9,1),(9,2),(9,3)},{})",
                 "(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)})"
         };

         List<Tuple> expected = Util.getTuplesFromConstantTupleStrings(results);

         //We need sort dbMergeCogrp because the result is different in sequence between spark and other mode when
         //multiple files are loaded(LOAD INPUT_FILE1,INPUT_FILE4...)
         for (Tuple t : dbMergeCogrp) {
             Util.convertBagToSortedBag(t);
         }
         for (Tuple t : expected) {
             Util.convertBagToSortedBag(t);
         }

         Collections.sort(dbMergeCogrp);
         Collections.sort(expected);
         assertEquals(dbMergeCogrp.size(), expected.size());

         //Since TestMapSideCogroup.DummyIndexableLoader.getNext() does not
         //apply schema for each input tuple,Util#checkQueryOutputsAfterSortRecursive fails to assert.
         // The schema for C is (int,{(chararray,int),(chararray,int),(chararray,int)},{(chararray,int),(chararray,int),(chararray,int)}).
         //But the schema for result "dbMergeCogrp" is (int,{(chararray,int),(chararray,int),(chararray,int)},{(chararray,chararray),(chararray,chararray),(chararray,chararray)})
         Iterator<Tuple> itr = dbMergeCogrp.iterator();
         for (int i = 0; i < dbMergeCogrp.size(); i++) {
             assertEquals(itr.next().toString(), expected.get(i).toString());
         }
     }

     @Test
     public void testCogrpOnMultiKeys() throws Exception{

         PigServer pigServer = new PigServer(cluster.getExecType(), 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());
     }

     @Test
     public void testEmptyDeltaFile() throws Exception{

         PigServer pigServer = new PigServer(cluster.getExecType(), 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,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());
     }

     @Test
     public void testDataWithNullKeys() throws Exception{

         PigServer pigServer = new PigServer(cluster.getExecType(), 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(new Path(loc).toUri(), 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;
         }
     }
 }
	/*
	* 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 static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertFalse;
	import static org.junit.Assert.assertTrue;
	import static org.junit.Assert.fail;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collections;
	import java.util.Iterator;
	import java.util.List;

	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.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
	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.FrontendException;
	import org.apache.pig.newplan.Operator;
	import org.apache.pig.newplan.logical.relational.LOCogroup;
	import org.apache.pig.newplan.logical.relational.LogicalPlan;
	import org.junit.After;
	import org.junit.AfterClass;
	import org.junit.Before;
	import org.junit.Test;

	public class TestMapSideCogroup {
	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 static MiniGenericCluster cluster = MiniGenericCluster.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_SIZE2LOOP_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);
	}

	@AfterClass
	public static void oneTimeTearDown() throws Exception {
	cluster.shutDown();
	}

	@Test
	public void testCompilation(){
	try{
	PigServer pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
	String query = "A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);" +
	"B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
	"D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
	"C = cogroup A by id, B by id, D by id using 'merge';" +
	"store C into 'output';";
	LogicalPlan lp = Util.buildLp(pigServer, query);
	Operator op = lp.getSinks().get(0);
	LOCogroup cogrp = (LOCogroup)lp.getPredecessors(op).get(0);
	assertEquals(LOCogroup.GROUPTYPE.MERGE, cogrp.getGroupType());

	PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
	pc.connect();
	PhysicalPlan phyP = Util.buildPp(pigServer, query);
	PhysicalOperator phyOp = phyP.getLeaves().get(0);
	assertTrue(phyOp instanceof POStore);
	phyOp = phyOp.getInputs().get(0);
	assertTrue(phyOp instanceof POMergeCogroup);

	MROperPlan mrPlan = Util.buildMRPlan(phyP,pc);
	assertEquals(2,mrPlan.size());

	Iterator<MapReduceOper> itr = mrPlan.iterator();
	List<MapReduceOper> opers = new ArrayList<MapReduceOper>();
	opers.add(itr.next());
	opers.add(itr.next());
	//Order of entrySet is not guaranteed with jdk1.7
	Collections.sort(opers);

	assertTrue(opers.get(0).reducePlan.isEmpty());
	assertFalse(opers.get(0).mapPlan.isEmpty());

	assertFalse(opers.get(1).reducePlan.isEmpty());
	assertFalse(opers.get(1).mapPlan.isEmpty());
	} catch(Exception e){
	e.printStackTrace();
	fail("Compilation of merged cogroup failed.");
	}

	}

	// @Test // PIG-2018
	// public void testFailure1() throws Exception{
	// PigServer pigServer = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
	// String query = "A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);" +
	// "E = group A by id;" +
	// "B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
	// "D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
	// "C = cogroup E by A.id, B by id, D by id using 'merge';" +
	// "store C into 'output';";
	// LogicalPlan lp = Util.buildLp(pigServer, query);
	// Operator op = lp.getSinks().get(0);
	// LOCogroup cogrp = (LOCogroup)lp.getPredecessors(op).get(0);
	// assertEquals(LOCogroup.GROUPTYPE.MERGE, cogrp.getGroupType());
	//
	// PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
	// pc.connect();
	// boolean exceptionCaught = false;
	// try{
	// Util.buildPp(pigServer, query);
	// }catch (java.lang.reflect.InvocationTargetException e){
	// FrontendException ex = (FrontendException)e.getTargetException();
	// assertEquals(1103,ex.getErrorCode());
	// exceptionCaught = true;
	// }
	// assertTrue(exceptionCaught);
	// }

	@Test
	public void testFailure2() throws Exception{
	PigServer pigServer = new PigServer(cluster.getExecType(), cluster.getProperties());
	String query = "A = LOAD 'data1' using "+ DummyCollectableLoader.class.getName() +"() as (id, name, grade);" +
	"B = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
	"D = LOAD 'data2' using "+ DummyIndexableLoader.class.getName() +"() as (id, name, grade);" +
	"C = cogroup A by id inner, B by id, D by id inner using 'merge';" +
	"store C into 'output';";
	LogicalPlan lp = Util.buildLp(pigServer, query);
	Operator op = lp.getSinks().get(0);
	LOCogroup cogrp = (LOCogroup)lp.getPredecessors(op).get(0);
	assertEquals(LOCogroup.GROUPTYPE.MERGE, cogrp.getGroupType());

	PigContext pc = new PigContext(ExecType.MAPREDUCE,cluster.getProperties());
	pc.connect();
	boolean exceptionCaught = false;
	try{
	Util.buildPp(pigServer, query);
	}catch (FrontendException e){
	exceptionCaught = true;
	}
	assertTrue(exceptionCaught);
	}

	@Test
	public void testSimple() throws Exception{

	PigServer pigServer = new PigServer(cluster.getExecType(), 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());
	}

	@Test
	public void test3Way() throws Exception{

	PigServer pigServer = new PigServer(cluster.getExecType(), 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());

	}

	@Test
	public void testMultiSplits() throws Exception{

	PigServer pigServer = new PigServer(cluster.getExecType(), 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);");

	List<Tuple> dbMergeCogrp = new ArrayList<Tuple>();

	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,2),(6,3)},{})",
	"(7,{(7,1),(7,2),(7,3)},{})",
	"(8,{(8,1),(8,2),(8,3)},{})",
	"(9,{(9,1),(9,2),(9,3)},{})",
	"(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)})"
	};

	List<Tuple> expected = Util.getTuplesFromConstantTupleStrings(results);

	//We need sort dbMergeCogrp because the result is different in sequence between spark and other mode when
	//multiple files are loaded(LOAD INPUT_FILE1,INPUT_FILE4...)
	for (Tuple t : dbMergeCogrp) {
	Util.convertBagToSortedBag(t);
	}
	for (Tuple t : expected) {
	Util.convertBagToSortedBag(t);
	}

	Collections.sort(dbMergeCogrp);
	Collections.sort(expected);
	assertEquals(dbMergeCogrp.size(), expected.size());

	//Since TestMapSideCogroup.DummyIndexableLoader.getNext() does not
	//apply schema for each input tuple,Util#checkQueryOutputsAfterSortRecursive fails to assert.
	// The schema for C is (int,{(chararray,int),(chararray,int),(chararray,int)},{(chararray,int),(chararray,int),(chararray,int)}).
	//But the schema for result "dbMergeCogrp" is (int,{(chararray,int),(chararray,int),(chararray,int)},{(chararray,chararray),(chararray,chararray),(chararray,chararray)})
	Iterator<Tuple> itr = dbMergeCogrp.iterator();
	for (int i = 0; i < dbMergeCogrp.size(); i++) {
	assertEquals(itr.next().toString(), expected.get(i).toString());
	}
	}

	@Test
	public void testCogrpOnMultiKeys() throws Exception{

	PigServer pigServer = new PigServer(cluster.getExecType(), 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());
	}

	@Test
	public void testEmptyDeltaFile() throws Exception{

	PigServer pigServer = new PigServer(cluster.getExecType(), 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,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());
	}

	@Test
	public void testDataWithNullKeys() throws Exception{

	PigServer pigServer = new PigServer(cluster.getExecType(), 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(new Path(loc).toUri(), 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;
	}
	}
	}