| /* |
| * 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.File; |
| import java.io.IOException; |
| import java.util.Iterator; |
| import java.util.List; |
| |
| import org.apache.pig.PigServer; |
| import org.apache.pig.builtin.mock.Storage; |
| import org.apache.pig.data.BagFactory; |
| import org.apache.pig.data.Tuple; |
| import org.apache.pig.data.TupleFactory; |
| import org.junit.Before; |
| import org.junit.Test; |
| |
| public class TestScalarAliasesLocal { |
| private static final String BUILD_TEST_TMP = "build/test/tmp/"; |
| private PigServer pigServer; |
| |
| TupleFactory mTf = TupleFactory.getInstance(); |
| BagFactory mBf = BagFactory.getInstance(); |
| |
| @Before |
| public void setUp() throws Exception{ |
| pigServer = new PigServer(Util.getLocalTestMode()); |
| } |
| |
| public static void deleteDirectory(File file) { |
| if (file.exists()) { |
| Util.deleteDirectory(file); |
| } |
| } |
| |
| public static File createLocalInputFile(String filename, String[] inputData) |
| throws IOException { |
| new File(filename).getParentFile().mkdirs(); |
| return Util.createLocalInputFile(filename, inputData); |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarAliasesBatchNobatch() throws Exception{ |
| String[] input = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| |
| String output = BUILD_TEST_TMP+"table_testScalarAliasesDir"; |
| TestScalarAliases.deleteDirectory(new File(output)); |
| // Test the use of scalars in expressions |
| String inputPath = BUILD_TEST_TMP+"table_testScalarAliasesBatch"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| // Test in script mode |
| pigServer.setBatchOn(); |
| pigServer.registerQuery("A = LOAD '" + inputPath + "' as (a0: long, a1: double);"); |
| pigServer.registerQuery("B = group A all;"); |
| pigServer.registerQuery("C = foreach B generate COUNT(A) as count, MAX(A.$1) as max;"); |
| pigServer.registerQuery("Y = foreach A generate (a0 * C.count), (a1 / C.max);"); |
| pigServer.registerQuery("Store Y into '" + output + "';"); |
| pigServer.executeBatch(); |
| // Check output |
| pigServer.registerQuery("Z = LOAD '" + output + "' as (a0: int, a1: double);"); |
| |
| Iterator<Tuple> iter; |
| Tuple t; |
| iter = pigServer.openIterator("Z"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(3,0.25)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(6,0.5)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(9,1.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| iter = pigServer.openIterator("Y"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(3,0.25)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(6,0.5)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(9,1.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testUseScalarMultipleTimes() throws Exception{ |
| String[] input = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| |
| String outputY = BUILD_TEST_TMP+"table_testUseScalarMultipleTimesOutY"; |
| TestScalarAliases.deleteDirectory(new File(outputY)); |
| String outputZ = BUILD_TEST_TMP+"table_testUseScalarMultipleTimesOutZ"; |
| TestScalarAliases.deleteDirectory(new File(outputZ)); |
| // Test the use of scalars in expressions |
| String inputPath = BUILD_TEST_TMP+"table_testUseScalarMultipleTimes"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| pigServer.setBatchOn(); |
| pigServer.registerQuery("A = LOAD '" + inputPath + "' as (a0: long, a1: double);"); |
| pigServer.registerQuery("B = group A all;"); |
| pigServer.registerQuery("C = foreach B generate COUNT(A) as count, MAX(A.$1) as max;"); |
| pigServer.registerQuery("Y = foreach A generate (a0 * C.count), (a1 / C.max);"); |
| pigServer.registerQuery("Store Y into '" + outputY + "';"); |
| pigServer.registerQuery("Z = foreach A generate (a1 + C.count), (a0 * C.max);"); |
| pigServer.registerQuery("Store Z into '" + outputZ + "';"); |
| // Test Multiquery store |
| pigServer.executeBatch(); |
| |
| // Check output |
| pigServer.registerQuery("M = LOAD '" + outputY + "' as (a0: int, a1: double);"); |
| |
| Iterator<Tuple> iter; |
| Tuple t; |
| iter = pigServer.openIterator("M"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(3,0.25)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(6,0.5)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(9,1.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| // Check output |
| pigServer.registerQuery("N = LOAD '" + outputZ + "' as (a0: double, a1: double);"); |
| |
| iter = pigServer.openIterator("N"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(8.0,20.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(13.0,40.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(23.0,60.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| // Non batch mode |
| iter = pigServer.openIterator("Y"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(3,0.25)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(6,0.5)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(9,1.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| // Check in non-batch mode |
| iter = pigServer.openIterator("Z"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(8.0,20.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(13.0,40.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(23.0,60.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarWithNoSchema() throws Exception{ |
| String[] scalarInput = { |
| "1\t5" |
| }; |
| String[] input = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| String inputPath = BUILD_TEST_TMP+"table_testScalarWithNoSchema"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| String inputPathScalar = BUILD_TEST_TMP+"table_testScalarWithNoSchemaScalar"; |
| TestScalarAliases.createLocalInputFile(inputPathScalar, scalarInput); |
| // Load A as a scalar |
| pigServer.registerQuery("A = LOAD '" + inputPath + "';"); |
| pigServer.registerQuery("scalar = LOAD '" + inputPathScalar + "' as (count, total);"); |
| pigServer.registerQuery("B = foreach A generate 5 / scalar.total;"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("B"); |
| |
| Tuple t = iter.next(); |
| assertTrue(t.get(0).toString().equals("1")); |
| |
| t = iter.next(); |
| assertTrue(t.get(0).toString().equals("1")); |
| |
| t = iter.next(); |
| assertTrue(t.get(0).toString().equals("1")); |
| |
| assertFalse(iter.hasNext()); |
| |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarWithTwoBranches() throws Exception{ |
| String[] inputA = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| |
| String[] inputX = { |
| "pig", |
| "hadoop", |
| "rocks" |
| }; |
| |
| String output = BUILD_TEST_TMP+"testScalarWithTwoBranchesDir"; |
| TestScalarAliases.deleteDirectory(new File(output)); |
| // Test the use of scalars in expressions |
| String inputPathA = BUILD_TEST_TMP+"testScalarWithTwoBranchesA"; |
| TestScalarAliases.createLocalInputFile(inputPathA, inputA); |
| String inputPathX = BUILD_TEST_TMP+"testScalarWithTwoBranchesX"; |
| TestScalarAliases.createLocalInputFile(inputPathX, inputX); |
| // Test in script mode |
| pigServer.setBatchOn(); |
| pigServer.registerQuery("A = LOAD '" + inputPathA + "' as (a0: long, a1: double);"); |
| pigServer.registerQuery("B = group A all;"); |
| pigServer.registerQuery("C = foreach B generate COUNT(A) as count, MAX(A.$1) as max;"); |
| pigServer.registerQuery("X = LOAD '" + inputPathX + "' as (names: chararray);"); |
| pigServer.registerQuery("Y = foreach X generate names, C.max;"); |
| pigServer.registerQuery("Store Y into '" + output + "';"); |
| pigServer.executeBatch(); |
| // Check output |
| pigServer.registerQuery("Z = LOAD '" + output + "' as (a0: chararray, a1: double);"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("Z"); |
| |
| Tuple t = iter.next(); |
| assertTrue(t.toString().equals("(pig,20.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(hadoop,20.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(rocks,20.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| // Check in non-batch mode |
| iter = pigServer.openIterator("Y"); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(pig,20.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(hadoop,20.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(rocks,20.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| pigServer.getPigContext().getProperties().remove("tez.am.inline.task.execution.max-tasks"); |
| |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testFilteredScalarDollarProj() throws Exception{ |
| String output = BUILD_TEST_TMP+"table_testFilteredScalarDollarProjDir"; |
| TestScalarAliases.deleteDirectory(new File(output)); |
| String[] input = { |
| "1\t5\t[state#maine,city#portland]\t{(a),(b)}\t(a,b)", |
| "2\t10\t\t\t", |
| "3\t20\t\t\t" |
| }; |
| |
| // Test the use of scalars in expressions |
| String inputPath = BUILD_TEST_TMP+"table_testFilteredScalarDollarProj"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| // Test in script mode |
| pigServer.setBatchOn(); |
| pigServer.registerQuery("A = LOAD '" + inputPath + "'" + " as (a0: long, a1: double, a2 : bytearray, " + "a3: bag{ t : tuple(tc : chararray)}, " + "a4: tuple(c1 : chararray, c2 : chararray) );"); |
| pigServer.registerQuery("B = filter A by $1 < 8;"); |
| pigServer.registerQuery("Y = foreach A generate (a0 * B.$0), (a1 / B.$1), B.$2, B.$2#'state', B.$3, B.a4;"); |
| pigServer.registerQuery("Store Y into '" + output + "';"); |
| pigServer.explain("Y", System.err); |
| pigServer.executeBatch(); |
| // Check output |
| pigServer.registerQuery("Z = LOAD '" + output + "' as (a0: int, a1: double);"); |
| pigServer.explain("Z", System.err); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("Z"); |
| |
| Tuple t = iter.next(); |
| assertTrue(t.toString().equals("(1,1.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(2,2.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(3,4.0)")); |
| |
| assertFalse(iter.hasNext()); |
| |
| // Check in non-batch mode |
| iter = pigServer.openIterator("Y"); |
| |
| t = iter.next(); |
| assertEquals(t.toString(),"(1,1.0,[state#maine,city#portland],maine,{(a),(b)},(a,b))"); |
| |
| t = iter.next(); |
| assertEquals(t.toString(),"(2,2.0,[state#maine,city#portland],maine,{(a),(b)},(a,b))"); |
| |
| t = iter.next(); |
| assertEquals(t.toString(),"(3,4.0,[state#maine,city#portland],maine,{(a),(b)},(a,b))"); |
| |
| assertFalse(iter.hasNext()); |
| |
| |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarWithNoSchemaDollarProj() throws Exception{ |
| String[] scalarInput = { |
| "1\t5" |
| }; |
| String[] input = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| String inputPath = BUILD_TEST_TMP+"table_testScalarWithNoSchemaDollarProj"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| String inputPathScalar = BUILD_TEST_TMP+"table_testScalarWithNoSchemaDollarProjScalar"; |
| TestScalarAliases.createLocalInputFile(inputPathScalar, scalarInput); |
| // Load A as a scalar |
| pigServer.registerQuery("A = LOAD '" + inputPath + "';"); |
| pigServer.registerQuery("scalar = LOAD '" + inputPathScalar + "';"); |
| pigServer.registerQuery("B = foreach A generate 5 / scalar.$1;"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("B"); |
| |
| Tuple t = iter.next(); |
| assertTrue(t.get(0).toString().equals("1")); |
| |
| t = iter.next(); |
| assertTrue(t.get(0).toString().equals("1")); |
| |
| t = iter.next(); |
| assertTrue(t.get(0).toString().equals("1")); |
| |
| assertFalse(iter.hasNext()); |
| |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarAliasesJoinClause() throws Exception{ |
| String[] inputA = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| String[] inputB = { |
| "Total3\tthree", |
| "Total2\ttwo", |
| "Total1\tone" |
| }; |
| |
| // Test the use of scalars in expressions |
| String inputPathA = BUILD_TEST_TMP+"table_testScalarAliasesJoinClauseA"; |
| TestScalarAliases.createLocalInputFile(inputPathA, inputA); |
| String inputPathB = BUILD_TEST_TMP+"table_testScalarAliasesJoinClauseB"; |
| TestScalarAliases.createLocalInputFile(inputPathB, inputB); |
| // Test in script mode |
| pigServer.registerQuery("A = LOAD '" + inputPathA + "' as (a0, a1);"); |
| pigServer.registerQuery("G = group A all;"); |
| pigServer.registerQuery("C = foreach G generate COUNT(A) as count;"); |
| |
| pigServer.registerQuery("B = LOAD '" + inputPathB + "' as (b0:chararray, b1:chararray);"); |
| pigServer.registerQuery("Y = join A by CONCAT('Total', (chararray)C.count), B by $0;"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("Y"); |
| |
| String[] expected = new String[] { |
| "(1,5,Total3,three)", |
| "(2,10,Total3,three)", |
| "(3,20,Total3,three)" |
| }; |
| |
| Util.checkQueryOutputsAfterSortRecursive(iter, expected, org.apache.pig.newplan.logical.Util.translateSchema(pigServer.dumpSchema("Y"))); |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarAliasesFilterClause() throws Exception{ |
| String[] input = { |
| "1\t5", |
| "2\t10", |
| "3\t20", |
| "4\t12", |
| "5\t8" |
| }; |
| |
| // Test the use of scalars in expressions |
| String inputPath = BUILD_TEST_TMP+"table_testScalarAliasesFilterClause"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| // Test in script mode |
| pigServer.registerQuery("A = LOAD '" + inputPath + "' as (a0, a1);"); |
| pigServer.registerQuery("G = group A all;"); |
| pigServer.registerQuery("C = foreach G generate AVG(A.$1) as average;"); |
| |
| pigServer.registerQuery("Y = filter A by a1 > C.average;"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("Y"); |
| |
| // Average is 11 |
| Tuple t = iter.next(); |
| assertTrue(t.toString().equals("(3,20)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(4,12)")); |
| |
| assertFalse(iter.hasNext()); |
| } |
| |
| // See PIG-1434 |
| @Test |
| public void testScalarAliasesGrammarNegative() throws Exception{ |
| String[] input = { |
| "1\t5", |
| "2\t10", |
| "3\t20" |
| }; |
| |
| String inputPath = BUILD_TEST_TMP+"table_testScalarAliasesGrammar"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| |
| try { |
| pigServer.registerQuery("A = LOAD '" + inputPath + "' as (a0: long, a1: double);"); |
| pigServer.registerQuery("B = group A all;"); |
| pigServer.registerQuery("C = foreach B generate COUNT(A);"); |
| // Only projections of C are supported |
| pigServer.registerQuery("Y = foreach A generate C;"); |
| pigServer.openIterator( "Y" ); |
| //Control should not reach here |
| fail("Scalar projections are only supported"); |
| } catch (IOException pe){ |
| assertTrue(pe.getMessage().contains("Invalid scalar projection: C")); |
| } |
| } |
| |
| // See PIG-1636 |
| @Test |
| public void testScalarAliasesLimit() throws Exception{ |
| String[] input = { |
| "a\t1", |
| "b\t2", |
| "c\t3", |
| "a\t4", |
| "c\t5" |
| }; |
| |
| // Test the use of scalars in expressions |
| String inputPath = BUILD_TEST_TMP+"table_testScalarAliasesLimit"; |
| TestScalarAliases.createLocalInputFile(inputPath, input); |
| // Test in script mode |
| pigServer.registerQuery("A = LOAD '" + inputPath + "' as (a0:chararray, a1: int);"); |
| pigServer.registerQuery("G = group A all;"); |
| pigServer.registerQuery("C = foreach G generate SUM(A.$1) as total;"); |
| pigServer.registerQuery("C1 = limit C 1;"); |
| pigServer.registerQuery("Y = foreach A generate a0, a1 * (double)C1.total;"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("Y"); |
| |
| // Average is 11 |
| Tuple t = iter.next(); |
| assertTrue(t.toString().equals("(a,15.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(b,30.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(c,45.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(a,60.0)")); |
| |
| t = iter.next(); |
| assertTrue(t.toString().equals("(c,75.0)")); |
| |
| assertFalse(iter.hasNext()); |
| } |
| |
| /** |
| * Test that a specific string is included in the error message when an |
| * exception is thrown for using a relation in a |
| * scalar context without projecting any columns out of it |
| */ |
| // See PIG-1788 |
| @Test |
| public void testScalarWithNoProjection() throws Exception{ |
| String query = |
| " A = load 'table_testScalarWithNoProjection' as (x, y);" + |
| " B = group A by x;" + |
| // B is unintentionally being used as scalar, |
| // the user intends it to be COUNT(A) |
| " C = foreach B generate COUNT(B);"; |
| |
| Util.checkExceptionMessage(query, "C", |
| "A column needs to be projected from a relation" + |
| " for it to be used as a scalar" |
| ); |
| } |
| |
| @Test |
| public void testScalarNullValue() throws Exception{ |
| Storage.Data data = Storage.resetData(pigServer); |
| data.set("input", Storage.tuple("a", 1), Storage.tuple("b", 2)); |
| |
| pigServer.setBatchOn(); |
| pigServer.registerQuery("A = load 'input' using mock.Storage() as (a:chararray, b:int);"); |
| pigServer.registerQuery("B = FILTER A by a == 'c';"); |
| pigServer.registerQuery("C = FOREACH A generate a, b + B.b;"); |
| pigServer.registerQuery("store C into 'output' using mock.Storage();"); |
| |
| pigServer.executeBatch(); |
| |
| List<Tuple> actualResults = data.get("output"); |
| List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings( |
| new String[] {"('a', null)", "('b', null)"}); |
| Util.checkQueryOutputsAfterSort(actualResults.iterator(), expectedResults); |
| |
| } |
| |
| @Test |
| public void testScalarNullValue2() throws Exception{ |
| Storage.Data data = Storage.resetData(pigServer); |
| data.set("input", Storage.tuple("a", 1), Storage.tuple("b", 2)); |
| |
| pigServer.registerQuery("A = load 'input' using mock.Storage() as (a:chararray, b:int);"); |
| pigServer.registerQuery("B = FILTER A by a == 'c';"); |
| pigServer.registerQuery("C = GROUP B ALL;"); |
| pigServer.registerQuery("D = FOREACH C GENERATE COUNT(B.b) as count;"); |
| pigServer.registerQuery("E = FOREACH A GENERATE (D.count IS NOT NULL? D.count : 0l);;"); |
| |
| Iterator<Tuple> iter = pigServer.openIterator("E"); |
| Tuple t = iter.next(); |
| assertTrue(t.toString().equals("(0)")); |
| } |
| } |
