| /* |
| * 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.apache.pig.builtin.mock.Storage.tuple; |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNull; |
| import static org.junit.Assert.fail; |
| |
| import java.io.File; |
| import java.io.FileWriter; |
| import java.io.IOException; |
| import java.io.PrintWriter; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Properties; |
| |
| import org.apache.pig.PigConfiguration; |
| import org.apache.pig.PigServer; |
| import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration; |
| import org.apache.pig.data.Tuple; |
| import org.apache.pig.parser.ParserException; |
| import org.junit.After; |
| import org.junit.AfterClass; |
| import org.junit.Before; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| |
| public class TestAccumulator { |
| private static final String INPUT_FILE1 = "AccumulatorInput1.txt"; |
| private static final String INPUT_FILE2 = "AccumulatorInput2.txt"; |
| private static final String INPUT_FILE3 = "AccumulatorInput3.txt"; |
| private static final String INPUT_FILE4 = "AccumulatorInput4.txt"; |
| private static final String INPUT_DIR = Util.getTestDirectory(TestAccumulator.class); |
| |
| private static PigServer pigServer; |
| private static Properties properties; |
| private static MiniGenericCluster cluster; |
| |
| @BeforeClass |
| public static void oneTimeSetUp() throws Exception { |
| cluster = MiniGenericCluster.buildCluster(); |
| properties = cluster.getProperties(); |
| properties.setProperty("pig.accumulative.batchsize", "2"); |
| properties.setProperty("pig.exec.nocombiner", "true"); |
| // Reducing the number of retry attempts to speed up test completion |
| properties.setProperty(MRConfiguration.MAP_MAX_ATTEMPTS,"1"); |
| properties.setProperty(MRConfiguration.REDUCE_MAX_ATTEMPTS,"1"); |
| createFiles(); |
| } |
| |
| @AfterClass |
| public static void oneTimeTearDown() throws Exception { |
| deleteFiles(); |
| cluster.shutDown(); |
| } |
| |
| @Before |
| public void setUp() throws Exception { |
| Util.resetStateForExecModeSwitch(); |
| // Drop stale configuration from previous test run |
| properties.remove(PigConfiguration.PIG_OPT_ACCUMULATOR); |
| pigServer = new PigServer(cluster.getExecType(), properties); |
| } |
| |
| @After |
| public void tearDown() throws Exception { |
| pigServer.shutdown(); |
| pigServer = null; |
| } |
| |
| private static void createFiles() throws IOException { |
| new File(INPUT_DIR).mkdirs(); |
| |
| PrintWriter w = new PrintWriter(new FileWriter(INPUT_DIR + "/" + INPUT_FILE1)); |
| |
| w.println("100\tapple"); |
| w.println("200\torange"); |
| w.println("300\tstrawberry"); |
| w.println("300\tpear"); |
| w.println("100\tapple"); |
| w.println("300\tpear"); |
| w.println("400\tapple"); |
| w.close(); |
| |
| Util.copyFromLocalToCluster(cluster, INPUT_DIR + "/" + INPUT_FILE1, INPUT_FILE1); |
| |
| w = new PrintWriter(new FileWriter(INPUT_DIR + "/" + INPUT_FILE2)); |
| |
| w.println("100\t"); |
| w.println("100\t"); |
| w.println("200\t"); |
| w.println("200\t"); |
| w.println("300\tstrawberry"); |
| w.close(); |
| |
| Util.copyFromLocalToCluster(cluster, INPUT_DIR + "/" + INPUT_FILE2, INPUT_FILE2); |
| |
| w = new PrintWriter(new FileWriter(INPUT_DIR + "/" + INPUT_FILE3)); |
| |
| w.println("100\t1.0"); |
| w.println("100\t2.0"); |
| w.println("200\t1.1"); |
| w.println("200\t2.1"); |
| w.println("100\t3.0"); |
| w.println("100\t4.0"); |
| w.println("200\t3.1"); |
| w.println("100\t5.0"); |
| w.println("300\t3.3"); |
| w.println("400\t"); |
| w.println("400\t"); |
| w.close(); |
| |
| Util.copyFromLocalToCluster(cluster, INPUT_DIR + "/" + INPUT_FILE3, INPUT_FILE3); |
| |
| w = new PrintWriter(new FileWriter(INPUT_DIR + "/" + INPUT_FILE4)); |
| |
| w.println("100\thttp://ibm.com,ibm"); |
| w.println("100\thttp://ibm.com,ibm"); |
| w.println("200\thttp://yahoo.com,yahoo"); |
| w.println("300\thttp://sun.com,sun"); |
| w.close(); |
| |
| Util.copyFromLocalToCluster(cluster, INPUT_DIR + "/" + INPUT_FILE4, INPUT_FILE4); |
| } |
| |
| private static void deleteFiles() { |
| Util.deleteDirectory(new File(INPUT_DIR)); |
| } |
| |
| @Test |
| public void testAccumBasic() throws IOException{ |
| // test group by |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, org.apache.pig.test.utils.AccumulatorBagCount(A);"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 2); |
| expected.put(200, 1); |
| expected.put(300, 3); |
| expected.put(400, 1); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A), org.apache.pig.test.utils.BagCount(A);"); |
| |
| try{ |
| iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| fail("accumulator should not be called."); |
| }catch(IOException e) { |
| // should throw exception from AccumulatorBagCount. |
| } |
| |
| // test cogroup |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("C = cogroup A by id, B by id;"); |
| pigServer.registerQuery("D = foreach C generate group, " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A), org.apache.pig.test.utils.AccumulatorBagCount(B);"); |
| |
| HashMap<Integer, String> expected2 = new HashMap<Integer, String>(); |
| expected2.put(100, "2,2"); |
| expected2.put(200, "1,1"); |
| expected2.put(300, "3,3"); |
| expected2.put(400, "1,1"); |
| |
| iter = pigServer.openIterator("D"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected2.get((Integer)t.get(0)), t.get(1).toString()+","+t.get(2).toString()); |
| } |
| } |
| |
| @Test |
| public void testAccumWithNegative() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, -org.apache.pig.test.utils.AccumulatorBagCount(A);"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, -2); |
| expected.put(200, -1); |
| expected.put(300, -3); |
| expected.put(400, -1); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithAdd() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, org.apache.pig.test.utils.AccumulatorBagCount(A)+1.0;"); |
| |
| { |
| HashMap<Integer, Double> expected = new HashMap<Integer, Double>(); |
| expected.put(100, 3.0); |
| expected.put(200, 2.0); |
| expected.put(300, 4.0); |
| expected.put(400, 2.0); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Double)t.get(1)); |
| } |
| } |
| |
| { |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A)+org.apache.pig.test.utils.AccumulatorBagCount(A);"); |
| |
| HashMap<Integer, Integer>expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 4); |
| expected.put(200, 2); |
| expected.put(300, 6); |
| expected.put(400, 2); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| } |
| |
| @Test |
| public void testAccumWithMinus() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| " org.apache.pig.test.utils.AccumulatorBagCount(A)*3.0-org.apache.pig.test.utils.AccumulatorBagCount(A);"); |
| |
| HashMap<Integer, Double> expected = new HashMap<Integer, Double>(); |
| expected.put(100, 4.0); |
| expected.put(200, 2.0); |
| expected.put(300, 6.0); |
| expected.put(400, 2.0); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Double)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithMod() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A) % 2;"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 0); |
| expected.put(200, 1); |
| expected.put(300, 1); |
| expected.put(400, 1); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithDivide() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A)/2;"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 1); |
| expected.put(200, 0); |
| expected.put(300, 1); |
| expected.put(400, 0); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithAnd() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "((org.apache.pig.test.utils.AccumulatorBagCount(A)>1 and " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A)<3)?0:1);"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 0); |
| expected.put(200, 1); |
| expected.put(300, 1); |
| expected.put(400, 1); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithOr() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "((org.apache.pig.test.utils.AccumulatorBagCount(A)>3 or " + |
| "org.apache.pig.test.utils.AccumulatorBagCount(A)<2)?0:1);"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 1); |
| expected.put(200, 0); |
| expected.put(300, 1); |
| expected.put(400, 0); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithRegexp() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "(((chararray)org.apache.pig.test.utils.AccumulatorBagCount(A)) matches '1*' ?0:1);"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 1); |
| expected.put(200, 0); |
| expected.put(300, 1); |
| expected.put(400, 0); |
| |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @Test |
| public void testAccumWithIsNull() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE2 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group, " + |
| "((chararray)org.apache.pig.test.utils.AccumulativeSumBag(A) is null?0:1);"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 0); |
| expected.put(200, 0); |
| expected.put(300, 1); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| } |
| |
| @SuppressWarnings("unchecked") |
| @Test |
| public void testAccumWithIsEmpty() throws IOException{ |
| pigServer.getPigContext().getProperties().setProperty("pig.accumulative.batchsize", "1"); |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = load '" + INPUT_FILE2 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("C = cogroup A by id outer, B by id outer;"); |
| pigServer.registerQuery("D = foreach C generate group," + |
| "(int)(IsEmpty(A) ? 0 : SUM(A.id)) as suma," + |
| "(int)(IsEmpty(B) ? 0 : SUM(B.id)) as sumb;"); |
| |
| List<Tuple> expected = new ArrayList<Tuple>(); |
| expected.add(tuple(100, 200, 200)); |
| expected.add(tuple(200, 200, 400)); |
| expected.add(tuple(300, 900, 300)); |
| expected.add(tuple(400, 400, 0)); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| List<Tuple> actual = new ArrayList<Tuple>(); |
| |
| while(iter.hasNext()) { |
| actual.add(iter.next()); |
| } |
| Collections.sort(actual); |
| assertEquals(expected, actual); |
| } |
| |
| @Test |
| public void testAccumWithDistinct() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, f);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B { D = distinct A;" + |
| "generate group, org.apache.pig.test.utils.AccumulatorBagCount(D)+1;};"); |
| |
| HashMap<Integer, Integer> expected = new HashMap<Integer, Integer>(); |
| expected.put(100, 2); |
| expected.put(200, 2); |
| expected.put(300, 3); |
| expected.put(400, 2); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| int count = 0; |
| while(iter.hasNext()) { |
| count++; |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Integer)t.get(1)); |
| } |
| assertEquals(4, count); |
| } |
| |
| @Test |
| public void testAccumWithSort() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, f);"); |
| pigServer.registerQuery("B = foreach A generate id, f, id as t;"); |
| pigServer.registerQuery("C = group B by id;"); |
| pigServer.registerQuery("D = foreach C { E = order B by f; F = E.f;" + |
| "generate group, org.apache.pig.test.utils.AccumulativeSumBag(F);};"); |
| |
| HashMap<Integer, String> expected = new HashMap<Integer, String>(); |
| expected.put(100, "(apple)(apple)"); |
| expected.put(200, "(orange)"); |
| expected.put(300, "(pear)(pear)(strawberry)"); |
| expected.put(400, "(apple)"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| |
| int count = 0; |
| while(iter.hasNext()) { |
| count++; |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (String)t.get(1)); |
| } |
| assertEquals(4, count); |
| } |
| |
| @Test |
| public void testAccumWithBuiltinAvg() throws IOException { |
| HashMap<Integer, Double> expected = new HashMap<Integer, Double>(); |
| expected.put(100, 3.0); |
| expected.put(200, 2.1); |
| expected.put(300, 3.3); |
| expected.put(400, null); |
| // Test all the averages for correct behaviour with null values |
| String[] types = { "double", "float", "int", "long" }; |
| for (int i = 0; i < types.length; i++) { |
| if (i > 1) { // adjust decimal error for non real types |
| expected.put(200, 2.0); |
| expected.put(300, 3.0); |
| } |
| pigServer.registerQuery("A = load '" + INPUT_FILE3 + "' as (id:int, v:" |
| + types[i] + ");"); |
| pigServer.registerQuery("C = group A by id;"); |
| pigServer.registerQuery("D = foreach C generate group, AVG(A.v);"); |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| |
| while (iter.hasNext()) { |
| Tuple t = iter.next(); |
| Double v = expected.get((Integer) t.get(0)); |
| if (v != null) { |
| assertEquals(v.doubleValue(), ((Number) t.get(1)).doubleValue(), |
| 0.0001); |
| } else { |
| assertNull(t.get(1)); |
| } |
| } |
| } |
| } |
| |
| @Test |
| public void testAccumWithBuiltin() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE3 + "' as (id:int, v:double);"); |
| pigServer.registerQuery("C = group A by id;"); |
| // moving AVG accumulator test to separate test case |
| pigServer.registerQuery("D = foreach C generate group, SUM(A.v), COUNT(A.v), MIN(A.v), MAX(A.v);"); |
| |
| HashMap<Integer, Double[]> expected = new HashMap<Integer, Double[]>(); |
| expected.put(100, new Double[]{15.0, 5.0, 1.0, 5.0}); |
| expected.put(200, new Double[]{6.3, 3.0, 1.1, 3.1}); |
| expected.put(300, new Double[]{3.3, 1.0, 3.3, 3.3}); |
| expected.put(400, new Double[] { null, 0.0, null, null }); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| Double[] v = expected.get((Integer)t.get(0)); |
| for(int i=0; i<v.length; i++) { |
| if (v[i] != null) { |
| assertEquals(v[i].doubleValue(), ((Number) t.get(i + 1)) |
| .doubleValue(), 0.0001); |
| } else { |
| assertNull(t.get(i + 1)); |
| } |
| } |
| } |
| } |
| |
| // Pig 4365 |
| @Test |
| public void testAccumWithTOP() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE3 + "' as (id:int, v:double);"); |
| pigServer.registerQuery("B = group A all;"); |
| pigServer.registerQuery("D = foreach B { C = TOP(5, 0, A); generate flatten(C); }"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| |
| List<Tuple> expected = Util.getTuplesFromConstantTupleStrings( |
| new String[] {"(200,1.1)", "(200,2.1)", "(300,3.3)", "(400,null)", "(400,null)" }); |
| |
| Util.checkQueryOutputsAfterSort(iter, expected); |
| } |
| |
| @Test |
| public void testAccumWithMultiBuiltin() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, c:chararray);"); |
| pigServer.registerQuery("C = group A by 1;"); |
| pigServer.registerQuery("D = foreach C generate SUM(A.id), 1+SUM(A.id)+SUM(A.id);"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| t.get(0).toString().equals("1700"); |
| t.get(1).toString().equals("3401"); |
| } |
| } |
| |
| // Pig 1105 |
| @Test |
| public void testAccumCountStar() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE3 + "' as (id:int, v:double);"); |
| pigServer.registerQuery("C = group A by id;"); |
| pigServer.registerQuery("D = foreach C generate group, COUNT_STAR(A.id);"); |
| |
| pigServer.openIterator("D"); |
| } |
| |
| /** |
| * see PIG-1963. |
| * If there is a POSort or PODistinct still remaining in the plan |
| * (after secondary sort optimization), accumulative mode can't |
| * be used as they are blocking operators |
| * @throws IOException |
| */ |
| @Test |
| public void testAccumulatorOffOnSort() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE2 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| //one POSort will remain because secondary sort can be used only for one of them |
| pigServer.registerQuery("C = foreach B " + |
| "{ " + |
| " o1 = order A by fruit;" + |
| " o2 = order A by fruit desc;" + |
| " generate org.apache.pig.test.utils.AccumulativeSumBag(o1.fruit), " + |
| " org.apache.pig.test.utils.AccumulativeSumBag(o2.fruit); " + |
| "};"); |
| |
| checkAccumulatorOff("C"); |
| } |
| |
| /** |
| * see PIG-1963. |
| * If there is a POSort or PODistinct still remaining in the plan |
| * (after secondary sort optimization), accumulative mode can't |
| * be used as they are blocking operators |
| * @throws IOException |
| */ |
| @Test |
| public void testAccumulatorOffOnDistinct() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE2 + "' as (id:int, fruit, category);"); |
| pigServer.registerQuery("B = group A by id;"); |
| |
| pigServer.registerQuery("C = foreach B " + |
| "{ " + |
| " o1 = order A by fruit;" + |
| " d2 = distinct A.category;" + |
| " generate org.apache.pig.test.utils.AccumulativeSumBag(o1.fruit), " + |
| " org.apache.pig.test.utils.AccumulativeSumBag(d2); " + |
| "};"); |
| |
| checkAccumulatorOff("C"); |
| } |
| |
| @Test |
| public void testAccumulatorOff() throws IOException{ |
| pigServer.getPigContext().getProperties().setProperty( |
| PigConfiguration.PIG_OPT_ACCUMULATOR, "false"); |
| |
| pigServer.registerQuery("A = load '" + INPUT_FILE2 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = group A by id;"); |
| pigServer.registerQuery("C = foreach B generate group," + |
| "org.apache.pig.test.utils.AccumulativeSumBag(A);"); |
| |
| checkAccumulatorOff("C"); |
| pigServer.getPigContext().getProperties().setProperty( |
| PigConfiguration.PIG_OPT_ACCUMULATOR, "true"); |
| |
| } |
| |
| private void checkAccumulatorOff(String alias) { |
| try { |
| Iterator<Tuple> iter = pigServer.openIterator(alias); |
| while(iter.hasNext()) { |
| iter.next(); |
| } |
| fail("Accumulator should be off."); |
| }catch(Exception e) { |
| // we should get exception |
| } |
| } |
| |
| @Test |
| public void testAccumWithMap() throws IOException{ |
| pigServer.registerQuery("A = load '" + INPUT_FILE4 + "' as (id, url);"); |
| pigServer.registerQuery("B = group A by (id, url);"); |
| pigServer.registerQuery("C = foreach B generate COUNT(A)," + |
| "org.apache.pig.test.utils.URLPARSE(group.url)#'url';"); |
| |
| HashMap<Integer, String> expected = new HashMap<Integer, String>(); |
| expected.put(2, "http://ibm.com"); |
| expected.put(1, "http://yahoo.com"); |
| expected.put(1, "http://sun.com"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("C"); |
| |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Long)t.get(0)), (String)t.get(1)); |
| } |
| } |
| |
| @Test // PIG-1837 |
| public void testBinCondCheck() throws Exception { |
| PrintWriter w = new PrintWriter(new FileWriter("data1")); |
| w.println("1\t11"); |
| w.println("2\t15"); |
| w.close(); |
| |
| Util.copyFromLocalToCluster(cluster, "data1", "data1"); |
| |
| w = new PrintWriter(new FileWriter("data2")); |
| w.println("1\t10"); |
| w.println("4\t11"); |
| w.println("5\t10"); |
| w.close(); |
| |
| Util.copyFromLocalToCluster(cluster, "data2", "data2"); |
| |
| pigServer.registerQuery("A = load 'data1' as (x:int, y:int);"); |
| pigServer.registerQuery("B = load 'data2' as (x:int, z:int);"); |
| pigServer.registerQuery("C = cogroup A by x, B by x;"); |
| pigServer.registerQuery("D = foreach C generate group," + |
| "SUM((IsEmpty(A.y) ? {(0)} : A.y)) + SUM((IsEmpty(B.z) ? {(0)} : B.z));"); |
| |
| HashMap<Integer, Long> expected = new HashMap<Integer, Long>(); |
| expected.put(1, 21l); |
| expected.put(2, 15l); |
| expected.put(4, 11l); |
| expected.put(5, 10l); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| while(iter.hasNext()) { |
| Tuple t = iter.next(); |
| assertEquals(expected.get((Integer)t.get(0)), (Long)t.get(1)); |
| } |
| } |
| |
| /** |
| * see PIG-1911 . |
| * accumulator udf reading from a nested relational op. generate projects |
| * only the accumulator udf. |
| * @throws IOException |
| * @throws ParseException |
| */ |
| @Test |
| public void testAccumAfterNestedOp() throws IOException, ParserException{ |
| // test group by |
| pigServer.registerQuery("A = load '" + INPUT_FILE1 + "' as (id:int, fruit);"); |
| pigServer.registerQuery("B = foreach A generate id;"); //additional test to enable columnprune(PIG-5224) |
| pigServer.registerQuery("C = group B by id;"); |
| pigServer.registerQuery("D = foreach C " + |
| "{ o = order B by id; " + |
| " generate org.apache.pig.test.utils.AccumulatorBagCount(o);}; "); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("D"); |
| List<Tuple> expectedRes = |
| Util.getTuplesFromConstantTupleStrings( |
| new String[] { |
| "(2)", |
| "(1)", |
| "(3)", |
| "(1)" |
| }); |
| Util.checkQueryOutputsAfterSort(iter, expectedRes); |
| } |
| |
| |
| } |
