| /* |
| * 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.fail; |
| |
| import java.util.ArrayList; |
| import java.util.List; |
| |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.pig.FuncSpec; |
| import org.apache.pig.PigConfiguration; |
| import org.apache.pig.backend.executionengine.ExecException; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POProject; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan; |
| import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPartialAgg; |
| import org.apache.pig.builtin.IntSum; |
| import org.apache.pig.data.DataType; |
| import org.apache.pig.data.Tuple; |
| import org.apache.pig.impl.plan.PlanException; |
| import org.apache.pig.parser.ParserException; |
| import org.apache.pig.test.utils.GenPhyOp; |
| import org.junit.After; |
| import org.junit.Before; |
| import org.junit.Test; |
| |
| /** |
| * Test POPartialAgg runtime |
| */ |
| public class TestPOPartialAgg { |
| POPartialAgg partAggOp; |
| PhysicalPlan parentPlan; |
| Tuple dummyTuple = null; |
| |
| @Before |
| public void setUp() throws Exception { |
| createPOPartialPlan(); |
| } |
| |
| private void createPOPartialPlan() throws PlanException { |
| parentPlan = new PhysicalPlan(); |
| partAggOp = GenPhyOp.topPOPartialAgg(); |
| partAggOp.setParentPlan(parentPlan); |
| |
| // setup key plan |
| PhysicalPlan keyPlan = new PhysicalPlan(); |
| POProject keyProj = new POProject(GenPhyOp.getOK(), -1, 0); |
| keyProj.setResultType(DataType.INTEGER); |
| keyPlan.add(keyProj); |
| partAggOp.setKeyPlan(keyPlan); |
| |
| // setup value plan |
| // project arg for udf |
| PhysicalPlan valPlan1 = new PhysicalPlan(); |
| POProject projVal1 = new POProject(GenPhyOp.getOK(), -1, 1); |
| projVal1.setResultType(DataType.BAG); |
| valPlan1.add(projVal1); |
| |
| // setup udf |
| List<PhysicalOperator> udfInps = new ArrayList<PhysicalOperator>(); |
| udfInps.add(projVal1); |
| FuncSpec sumSpec = new FuncSpec(IntSum.Intermediate.class.getName()); |
| POUserFunc sumUdf = new POUserFunc(GenPhyOp.getOK(), -1, udfInps, |
| sumSpec); |
| valPlan1.add(sumUdf); |
| valPlan1.connect(projVal1, sumUdf); |
| |
| List<PhysicalPlan> valuePlans = new ArrayList<PhysicalPlan>(); |
| valuePlans.add(valPlan1); |
| |
| partAggOp.setValuePlans(valuePlans); |
| } |
| |
| @After |
| public void tearDown() throws Exception { |
| } |
| |
| @Test |
| public void testPartialOneInput1() throws ExecException, ParserException { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] tups1 = { "(1,(2L))" }; |
| Tuple t = Util.getTuplesFromConstantTupleStrings(tups1).get(0); |
| checkSingleRow(t); |
| } |
| |
| @Test |
| public void testPartialOneInput2() throws ExecException, ParserException { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] tups1 = { "(null,(2L))" }; |
| Tuple t = Util.getTuplesFromConstantTupleStrings(tups1).get(0); |
| checkSingleRow(t); |
| } |
| |
| @Test |
| public void testPartialOneInput3() throws ExecException, ParserException { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] tups1 = { "(1,(null))" }; |
| Tuple t = Util.getTuplesFromConstantTupleStrings(tups1).get(0); |
| checkSingleRow(t); |
| } |
| |
| private void checkSingleRow(Tuple t) throws ExecException { |
| Result res; |
| // attaching one input tuple, result tuple stays in operator, expect EOP |
| partAggOp.attachInput(t); |
| res = partAggOp.getNext(dummyTuple); |
| assertEquals(POStatus.STATUS_EOP, res.returnStatus); |
| |
| // end of all input, now expecting results |
| parentPlan.endOfAllInput = true; |
| res = partAggOp.getNext(dummyTuple); |
| assertEquals(POStatus.STATUS_OK, res.returnStatus); |
| assertEquals(t, res.result); |
| } |
| |
| @Test |
| public void testPartialAggNoInput() throws ExecException, ParserException { |
| |
| // nothing attached, expecting EOP |
| Result res = partAggOp.getNext(dummyTuple); |
| assertEquals(POStatus.STATUS_EOP, res.returnStatus); |
| |
| // end of all input, still no results |
| parentPlan.endOfAllInput = true; |
| res = partAggOp.getNext(dummyTuple); |
| assertEquals(POStatus.STATUS_EOP, res.returnStatus); |
| |
| } |
| |
| @Test |
| public void testPartialMultiInput1() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(1,(1L))", "(1,(2L))", "(2,(1L))" }; |
| String[] outputTups = { "(1,(3L))", "(2,(1L))" }; |
| checkInputAndOutput(inputTups, outputTups, false); |
| } |
| |
| @Test |
| public void testPartialMultiInput2() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(1,(1L))", "(2,(2L))", "(1,(2L))" }; |
| String[] outputTups = { "(1,(3L))", "(2,(2L))" }; |
| checkInputAndOutput(inputTups, outputTups, false); |
| } |
| |
| @Test |
| public void testPartialMultiInput3() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(null,(1L))", "(null,(2L))", "(null,(2L))" }; |
| String[] outputTups = { "(null,(5L))" }; |
| checkInputAndOutput(inputTups, outputTups, false); |
| } |
| |
| @Test |
| public void testPartialMultiInput4() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(1,(1L))", "(2,(2L))", "(null,(2L))" }; |
| String[] outputTups = { "(1,(1L))", "(2,(2L))", "(null,(2L))" }; |
| checkInputAndOutput(inputTups, outputTups, false); |
| } |
| |
| @Test |
| // The case where there is no memory for use by hashmap |
| public void testPartialMultiInputHashMemEmpty1() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(1,(1L))", "(2,(2L))", "(null,(2L))" }; |
| String[] outputTups = { "(1,(1L))", "(2,(2L))", "(null,(2L))" }; |
| checkInputAndOutput(inputTups, outputTups, true); |
| } |
| |
| @Test |
| // The case where there is no memory for use by hashmap |
| public void testPartialMultiInputHashMemEmpty2() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(1,(1L))", "(2,(2L))", "(1,(2L))" }; |
| // since the group keys with same value are not in consecutive rows |
| // and hashmap is not given any memory they don't get |
| // aggreated with POPartialAgg |
| String[] outputTups = { "(1,(1L))", "(2,(2L))", "(1,(2L))" }; |
| checkInputAndOutput(inputTups, outputTups, true); |
| } |
| |
| @Test |
| public void testMultiInput1HashMemEmpty() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(1,(1L))", "(2,(2L))", "(1,(2L))" }; |
| String[] outputTups = { "(1,(3L))", "(2,(2L))" }; |
| checkInputAndOutput(inputTups, outputTups, false); |
| } |
| |
| @Test |
| public void testPartialMultiInputMultiInput1HashMemEmpty() throws Exception { |
| // input tuple has key, and bag containing SUM.Init output |
| String[] inputTups = { "(null,(1L))", "(null,(2L))", "(null,(2L))" }; |
| String[] outputTups = { "(null,(5L))" }; |
| checkInputAndOutput(inputTups, outputTups, false); |
| } |
| |
| |
| /** |
| * run the plan on inputTups and check if output matches outputTups if |
| * isMapMemEmpty is set to true, set memory available for the hash-map to |
| * zero |
| * |
| * @param inputTups |
| * @param outputTups |
| * @param isMapMemEmpty |
| * @throws ParserException |
| * @throws ExecException |
| * @throws PlanException |
| */ |
| private void checkInputAndOutput(String[] inputTups, String[] outputTups, |
| boolean isMapMemEmpty) throws Exception { |
| |
| PigMapReduce.sJobConfInternal.set(new Configuration()); |
| if (isMapMemEmpty) { |
| PigMapReduce.sJobConfInternal.get().set(PigConfiguration.PROP_CACHEDBAG_MEMUSAGE, |
| "0"); |
| } |
| |
| List<Tuple> inputs = Util.getTuplesFromConstantTupleStrings(inputTups); |
| List<Tuple> expectedOuts = Util |
| .getTuplesFromConstantTupleStrings(outputTups); |
| List<Tuple> outputs = new ArrayList<Tuple>(); |
| |
| // run through the inputs |
| for (Tuple t : inputs) { |
| Result res; |
| // attaching one input tuple, result tuple stays in operator, expect |
| // EOP |
| partAggOp.attachInput(t); |
| res = partAggOp.getNext(dummyTuple); |
| if (isMapMemEmpty) { |
| addResults(res, outputs); |
| } else { |
| assertEquals(POStatus.STATUS_EOP, res.returnStatus); |
| } |
| } |
| |
| // start getting the outputs |
| |
| // end of all input, now expecting results |
| parentPlan.endOfAllInput = true; |
| |
| if (isMapMemEmpty) { |
| Result res = partAggOp.getNext(dummyTuple); |
| // only one last output expected |
| addResults(res, outputs); |
| |
| res = partAggOp.getNext(dummyTuple); |
| assertEquals(POStatus.STATUS_EOP, res.returnStatus); |
| Util.compareActualAndExpectedResults(outputs, expectedOuts); |
| } else { |
| while (true) { |
| Result res = partAggOp.getNext(dummyTuple); |
| if (!addResults(res, outputs)) { |
| break; |
| } |
| } |
| Util.compareActualAndExpectedResults(outputs, expectedOuts); |
| } |
| |
| } |
| |
| private boolean addResults(Result res, List<Tuple> outputs) { |
| if (res.returnStatus == POStatus.STATUS_EOP) { |
| return false; |
| } else if (res.returnStatus == POStatus.STATUS_OK) { |
| outputs.add((Tuple) res.result); |
| return true; |
| } else { |
| fail("Invalid result status " + res.returnStatus); |
| return false; // to keep compiler happy |
| } |
| } |
| |
| } |