You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2012/10/31 22:57:38 UTC
svn commit: r1404375 - in /pig/branches/branch-0.10: ./
test/org/apache/pig/test/
Author: daijy
Date: Wed Oct 31 21:57:37 2012
New Revision: 1404375
URL: http://svn.apache.org/viewvc?rev=1404375&view=rev
Log:
PIG-2798: pig streaming tests assume interpreters are auto-resolved (jgordon via daijy)
Modified:
pig/branches/branch-0.10/CHANGES.txt
pig/branches/branch-0.10/test/org/apache/pig/test/TestGrunt.java
pig/branches/branch-0.10/test/org/apache/pig/test/TestPruneColumn.java
pig/branches/branch-0.10/test/org/apache/pig/test/TestSample.java
pig/branches/branch-0.10/test/org/apache/pig/test/TestStreaming.java
pig/branches/branch-0.10/test/org/apache/pig/test/TestStreamingLocal.java
Modified: pig/branches/branch-0.10/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.10/CHANGES.txt?rev=1404375&r1=1404374&r2=1404375&view=diff
==============================================================================
--- pig/branches/branch-0.10/CHANGES.txt (original)
+++ pig/branches/branch-0.10/CHANGES.txt Wed Oct 31 21:57:37 2012
@@ -36,6 +36,8 @@ PIG-2727: PigStorage Source tagging does
BUG FIXES
+PIG-2798: pig streaming tests assume interpreters are auto-resolved (jgordon via daijy)
+
PIG-2797: Tests should not create their own file URIs through string concatenation, should use Util.generateURI
instead (jgordon via daijy)
Modified: pig/branches/branch-0.10/test/org/apache/pig/test/TestGrunt.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.10/test/org/apache/pig/test/TestGrunt.java?rev=1404375&r1=1404374&r2=1404375&view=diff
==============================================================================
--- pig/branches/branch-0.10/test/org/apache/pig/test/TestGrunt.java (original)
+++ pig/branches/branch-0.10/test/org/apache/pig/test/TestGrunt.java Wed Oct 31 21:57:37 2012
@@ -1029,23 +1029,37 @@ public class TestGrunt {
try {
PigServer server = new PigServer(ExecType.MAPREDUCE,cluster.getProperties());
PigContext context = server.getPigContext();
+
+ String strQuote = "'";
+ String strRemoveFile = "rm";
+ String strRemoveDir = "rmdir";
+
+ if (Util.WINDOWS)
+ {
+ strQuote = "\"";
+ strRemoveFile = "del";
+ strRemoveDir = "rd";
+ }
String strCmd = "sh mkdir test_shell_tmp;";
+ // Create a temp directory via command and make sure it exists
ByteArrayInputStream cmd = new ByteArrayInputStream(strCmd.getBytes());
InputStreamReader reader = new InputStreamReader(cmd);
Grunt grunt = new Grunt(new BufferedReader(reader), context);
grunt.exec();
assertTrue(new File("test_shell_tmp").exists());
- strCmd = "sh rmdir test_shell_tmp;";
+ // Remove the temp directory via shell and make sure it is gone
+ strCmd = "sh " + strRemoveDir + " test_shell_tmp;";
cmd = new ByteArrayInputStream(strCmd.getBytes());
reader = new InputStreamReader(cmd);
grunt = new Grunt(new BufferedReader(reader), context);
grunt.exec();
assertFalse(new File("test_shell_tmp").exists());
- strCmd = "sh bash -c 'echo hello world > tempShFileToTestShCommand'";
+ // Verify pipes are usable in the command context by piping data to a file
+ strCmd = "sh echo hello world > tempShFileToTestShCommand";
cmd = new ByteArrayInputStream(strCmd.getBytes());
reader = new InputStreamReader(cmd);
grunt = new Grunt(new BufferedReader(reader), context);
@@ -1054,27 +1068,36 @@ public class TestGrunt {
fileReader = new BufferedReader(new FileReader("tempShFileToTestShCommand"));
assertTrue(fileReader.readLine().equalsIgnoreCase("hello world"));
fileReader.close();
- strCmd = "sh rm tempShFileToTestShCommand";
+
+ // Remove the file via cmd and make sure it is gone
+ strCmd = "sh " + strRemoveFile + " tempShFileToTestShCommand";
cmd = new ByteArrayInputStream(strCmd.getBytes());
reader = new InputStreamReader(cmd);
grunt = new Grunt(new BufferedReader(reader), context);
grunt.exec();
assertFalse(new File("tempShFileToTestShCommand").exists());
- strCmd = "sh bash -c 'touch TouchedFileInsideGrunt_61 | ls | grep TouchedFileInsideGrunt_61 > fileContainingTouchedFileInsideGruntShell_71'";
+ if (Util.WINDOWS) {
+ strCmd = "sh echo foo > TouchedFileInsideGrunt_61 | dir /B | findstr TouchedFileInsideGrunt_61 > fileContainingTouchedFileInsideGruntShell_71";
+ }
+ else {
+ strCmd = "sh touch TouchedFileInsideGrunt_61 | ls | grep TouchedFileInsideGrunt_61 > fileContainingTouchedFileInsideGruntShell_71";
+ }
cmd = new ByteArrayInputStream(strCmd.getBytes());
reader = new InputStreamReader(cmd);
grunt = new Grunt(new BufferedReader(reader), context);
grunt.exec();
fileReader = new BufferedReader(new FileReader("fileContainingTouchedFileInsideGruntShell_71"));
- assertTrue(fileReader.readLine().equals("TouchedFileInsideGrunt_61"));
+ assertTrue(fileReader.readLine().trim().equals("TouchedFileInsideGrunt_61"));
fileReader.close();
- strCmd = "sh bash -c 'rm fileContainingTouchedFileInsideGruntShell_71'";
+
+ strCmd = "sh " + strRemoveFile+" fileContainingTouchedFileInsideGruntShell_71";
cmd = new ByteArrayInputStream(strCmd.getBytes());
reader = new InputStreamReader(cmd);
grunt = new Grunt(new BufferedReader(reader), context);
grunt.exec();
assertFalse(new File("fileContainingTouchedFileInsideGruntShell_71").exists());
+
strCmd = "sh bash -c 'rm TouchedFileInsideGrunt_61'";
cmd = new ByteArrayInputStream(strCmd.getBytes());
reader = new InputStreamReader(cmd);
Modified: pig/branches/branch-0.10/test/org/apache/pig/test/TestPruneColumn.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.10/test/org/apache/pig/test/TestPruneColumn.java?rev=1404375&r1=1404374&r2=1404375&view=diff
==============================================================================
--- pig/branches/branch-0.10/test/org/apache/pig/test/TestPruneColumn.java (original)
+++ pig/branches/branch-0.10/test/org/apache/pig/test/TestPruneColumn.java Wed Oct 31 21:57:37 2012
@@ -82,8 +82,8 @@ public class TestPruneColumn extends Tes
private static final String simpleEchoStreamingCommand;
static {
- if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS"))
- simpleEchoStreamingCommand = "perl -ne 'print \\\"$_\\\"'";
+ if (Util.WINDOWS)
+ simpleEchoStreamingCommand = "perl -ne \"print ^\"$_^\"\"";
else
simpleEchoStreamingCommand = "perl -ne 'print \"$_\"'";
}
@@ -271,7 +271,7 @@ public class TestPruneColumn extends Tes
@Test
public void testLoadForEach1() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = foreach A generate a1, a2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -294,7 +294,7 @@ public class TestPruneColumn extends Tes
@Test
public void testLoadForEach2() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = foreach A generate a0, a2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -317,7 +317,7 @@ public class TestPruneColumn extends Tes
@Test
public void testLoadForEach3() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = foreach A generate a0, a1;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -340,8 +340,8 @@ public class TestPruneColumn extends Tes
@Test
public void testJoin1() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' as (b0:int, b1:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' as (b0:int, b1:int);");
pigServer.registerQuery("C = join A by a1, B by b1;");
pigServer.registerQuery("D = foreach C generate a1, a2, b0, b1;");
@@ -362,8 +362,8 @@ public class TestPruneColumn extends Tes
@Test
public void testJoin2() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' as (b0:int, b1:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' as (b0:int, b1:int);");
pigServer.registerQuery("C = join A by a1, B by b1;");
pigServer.registerQuery("D = foreach C generate a1, a2, b1;");
@@ -384,7 +384,7 @@ public class TestPruneColumn extends Tes
@Test
public void testForEachFilter() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = filter A by a2==3;");
pigServer.registerQuery("C = foreach B generate a0, a1;");
@@ -403,7 +403,7 @@ public class TestPruneColumn extends Tes
@Test
public void testForEach1() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = foreach A generate a0, a1+a2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -427,7 +427,7 @@ public class TestPruneColumn extends Tes
@Test
public void testForEach2() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = foreach A generate a0 as b0, *;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -455,7 +455,7 @@ public class TestPruneColumn extends Tes
@Test
public void testSplit1() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
pigServer.registerQuery("split A into B if $0<=1, C if $0>1;");
pigServer.registerQuery("D = foreach B generate $1;");
@@ -473,7 +473,7 @@ public class TestPruneColumn extends Tes
@Test
public void testSplit2() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
pigServer.registerQuery("split A into B if $0<=1, C if $0>1;");
pigServer.registerQuery("D = foreach B generate $1;");
@@ -491,7 +491,7 @@ public class TestPruneColumn extends Tes
@Test
public void testForeachNoSchema1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "';");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "';");
pigServer.registerQuery("B = foreach A generate $1, $2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -514,7 +514,7 @@ public class TestPruneColumn extends Tes
@Test
public void testForeachNoSchema2() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "';");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "';");
pigServer.registerQuery("B = foreach A generate $1, 'aoeuaoeu';");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -537,8 +537,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1:int);");
pigServer.registerQuery("C = cogroup A by $1, B by $1;");
pigServer.registerQuery("D = foreach C generate AVG($1.$1);");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -568,7 +568,7 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup2() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
pigServer.registerQuery("B = group A all;");
pigServer.registerQuery("C = foreach B generate $1;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -584,7 +584,7 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup3() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
pigServer.registerQuery("B = group A by $1;");
pigServer.registerQuery("C = foreach B generate $1, '1';");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -610,8 +610,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup4() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1:int);");
pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");
pigServer.registerQuery("D = foreach C generate $1.$1, $2.$1;");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -644,7 +644,7 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup5() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = group A by (a0, a1);");
pigServer.registerQuery("C = foreach B generate flatten(group);");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -670,7 +670,7 @@ public class TestPruneColumn extends Tes
@Test
public void testDistinct1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile4.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile4.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = distinct A;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -688,7 +688,7 @@ public class TestPruneColumn extends Tes
@Test
public void testStream1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = stream A through `" + simpleEchoStreamingCommand + "`;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -712,7 +712,7 @@ public class TestPruneColumn extends Tes
@Test
public void testBinCond1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile5.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2, a3);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile5.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2, a3);");
pigServer.registerQuery("B = foreach A generate ($1 == '2'? $2 : $3);");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -736,8 +736,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup6() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");
pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");
pigServer.registerQuery("D = foreach C generate A, flatten(B.($0, $1));");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -765,8 +765,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup7() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");
pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");
pigServer.registerQuery("D = foreach C {B = order B by $0;generate FLATTEN(A), B.($1);};");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -796,8 +796,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCross1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");
pigServer.registerQuery("C = cross A, B;");
pigServer.registerQuery("D = foreach C generate $0, $3;");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -840,8 +840,8 @@ public class TestPruneColumn extends Tes
@Test
public void testUnion1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile4.toString(), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile4.toString()), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
pigServer.registerQuery("C = union A, B;");
pigServer.registerQuery("D = foreach C generate $0, $2;");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -881,8 +881,8 @@ public class TestPruneColumn extends Tes
@Test
public void testFRJoin1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");
pigServer.registerQuery("C = join A by $0, B by $0 using 'replicated';");
pigServer.registerQuery("D = foreach C generate $0, $3;");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -909,7 +909,7 @@ public class TestPruneColumn extends Tes
@Test
public void testFilter1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = order A by a1;");
pigServer.registerQuery("C = limit B 10;");
pigServer.registerQuery("D = foreach C generate $0;");
@@ -934,7 +934,7 @@ public class TestPruneColumn extends Tes
@Test
public void testFilter2() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = filter A by a0+a2 == 4;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -958,7 +958,7 @@ public class TestPruneColumn extends Tes
@Test
public void testOrderBy1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = order A by $0;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -982,7 +982,7 @@ public class TestPruneColumn extends Tes
@Test
public void testOrderBy2() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = order A by *;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1006,7 +1006,7 @@ public class TestPruneColumn extends Tes
@Test
public void testCogroup8() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = group A by *;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1030,8 +1030,8 @@ public class TestPruneColumn extends Tes
@Test
public void testJoin3() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile4.toString(), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile4.toString()), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
pigServer.registerQuery("C = join A by *, B by * using 'replicated';");
pigServer.registerQuery("D = foreach C generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -1055,7 +1055,7 @@ public class TestPruneColumn extends Tes
@Test
public void testLoadForEach4() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate *;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1079,7 +1079,7 @@ public class TestPruneColumn extends Tes
@Test
public void testForEachUDF() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2:chararray);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2:chararray);");
pigServer.registerQuery("B = foreach A generate StringSize(*);");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1103,8 +1103,8 @@ public class TestPruneColumn extends Tes
@Test
public void testOutJoin1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile6.toString(), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2:chararray);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2:chararray);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile6.toString()), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2:chararray);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2:chararray);");
pigServer.registerQuery("C = join A by $0 left, B by $0;");
pigServer.registerQuery("D = foreach C generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("D");
@@ -1133,7 +1133,7 @@ public class TestPruneColumn extends Tes
@Test
public void testFilter3() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = filter A by " + MyFilterFunc.class.getName() + "(*) ;");
pigServer.registerQuery("C = foreach B generate $0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1157,7 +1157,7 @@ public class TestPruneColumn extends Tes
@Test
public void testMapKey1() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
pigServer.registerQuery("B = foreach A generate a0, a1#'key1';");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -1181,7 +1181,7 @@ public class TestPruneColumn extends Tes
@Test
public void testMapKey2() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
pigServer.registerQuery("B = foreach A generate a1, a1#'key1';");
pigServer.registerQuery("C = foreach B generate $0#'key2', $1;");
@@ -1207,7 +1207,7 @@ public class TestPruneColumn extends Tes
@Test
public void testMapKey3() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
pigServer.registerQuery("B = foreach A generate a1, a1#'key1';");
pigServer.registerQuery("C = group B all;");
@@ -1224,7 +1224,7 @@ public class TestPruneColumn extends Tes
@Test
public void testMapKey4() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
pigServer.registerQuery("B = limit A 10;");
pigServer.registerQuery("C = foreach B generate $0, $1#'key1';");
@@ -1249,7 +1249,7 @@ public class TestPruneColumn extends Tes
@Test
public void testMapKey5() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:map[]);");
pigServer.registerQuery("B = foreach A generate $0, $1#'key1';");
pigServer.registerQuery("C = stream B through `" + simpleEchoStreamingCommand + "`;");
@@ -1274,7 +1274,7 @@ public class TestPruneColumn extends Tes
@Test
public void testMapKeyInSplit1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile12.toString(), pigServer.getPigContext()) + "' as (m:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile12.toString()), pigServer.getPigContext()) + "' as (m:map[]);");
pigServer.registerQuery("B = foreach A generate m#'key1' as key1;");
pigServer.registerQuery("C = foreach A generate m#'key2' as key2;");
pigServer.registerQuery("D = join B by key1, C by key2;");
@@ -1295,7 +1295,7 @@ public class TestPruneColumn extends Tes
@SuppressWarnings("rawtypes")
@Test
public void testMapKeyInSplit2() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile12.toString(), pigServer.getPigContext()) + "' as (m:map[]);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile12.toString()), pigServer.getPigContext()) + "' as (m:map[]);");
pigServer.registerQuery("B = filter A by m#'cond'==1;");
pigServer.registerQuery("C = filter B by m#'key1'==1;");
pigServer.registerQuery("D = filter B by m#'key2'==2;");
@@ -1320,7 +1320,7 @@ public class TestPruneColumn extends Tes
@Test
public void testConstantPlan() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate 1, a2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -1344,7 +1344,7 @@ public class TestPruneColumn extends Tes
@Test
public void testPlainPlan() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
pigServer.registerQuery("B = order A by $0;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -1373,10 +1373,12 @@ public class TestPruneColumn extends Tes
// get a temp intermediate filename
File intermediateFile = File.createTempFile("intemediate", "txt");
intermediateFile.delete(); // delete since we don't want the file to be present
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
- pigServer.store("A", intermediateFile.toString(), "BinStorage()");
-
- pigServer.registerQuery("A = load '"+ intermediateFile.toString()
+ String clusterPath = Util.removeColon(intermediateFile.getAbsolutePath());
+
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.store("A", clusterPath, "BinStorage()");
+
+ pigServer.registerQuery("A = load '"+ Util.encodeEscape(clusterPath)
+ "' using BinStorage() as (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate a0;");
@@ -1403,10 +1405,12 @@ public class TestPruneColumn extends Tes
// get a temp intermediate filename
File intermediateFile = File.createTempFile("intemediate", "txt");
intermediateFile.delete(); // delete since we don't want the file to be present
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
- pigServer.store("A", intermediateFile.toString(), "BinStorage()");
-
- pigServer.registerQuery("A = load '"+ intermediateFile.toString()
+ String clusterPath = Util.removeColon(intermediateFile.getAbsolutePath());
+
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.store("A", clusterPath, "BinStorage()");
+
+ pigServer.registerQuery("A = load '"+ Util.encodeEscape(clusterPath)
+ "' using BinStorage() as (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate a2, a0, a1;");
@@ -1435,7 +1439,7 @@ public class TestPruneColumn extends Tes
@Test
public void testProjectCastKeyLookup() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext())
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext())
+ "' as (a0, a1);");
pigServer.registerQuery("B = foreach A generate a1#'key1';");
@@ -1460,7 +1464,7 @@ public class TestPruneColumn extends Tes
@Test
public void testRelayFlattenMap() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile3.toString(), pigServer.getPigContext())
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile3.toString()), pigServer.getPigContext())
+ "' as (a0, a1:map[]);");
pigServer.registerQuery("B = foreach A generate flatten(a1);");
@@ -1486,8 +1490,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCrossAtLeastOneColumnOneInput() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' as (b0:int, b1:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' as (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' as (b0:int, b1:int);");
pigServer.registerQuery("C = cross A, B;");
pigServer.registerQuery("D = foreach C generate $0;");
@@ -1524,8 +1528,8 @@ public class TestPruneColumn extends Tes
@Test
public void testComplex1() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile7.toString(), pigServer.getPigContext()) + "' as (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile8.toString(), pigServer.getPigContext()) + "' as (b0, b1, b2, b3);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile7.toString()), pigServer.getPigContext()) + "' as (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile8.toString()), pigServer.getPigContext()) + "' as (b0, b1, b2, b3);");
pigServer.registerQuery("B1 = foreach B generate b2, b0+b3;");
pigServer.registerQuery("C = join A by $0, B1 by $0;");
pigServer.registerQuery("D = order C by $4;");
@@ -1549,8 +1553,8 @@ public class TestPruneColumn extends Tes
@Test
public void testCoGroup8() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:int, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1:int);");
pigServer.registerQuery("C = cogroup A by ($1), B by ($1);");
pigServer.registerQuery("D = foreach C generate $0, $1;");
@@ -1582,7 +1586,7 @@ public class TestPruneColumn extends Tes
// See PIG-1128
@Test
public void testUserDefinedSchema() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS ( c1 : chararray, c2 : int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS ( c1 : chararray, c2 : int);");
pigServer.registerQuery("B = foreach A generate c1 as c1 : chararray, c2 as c2 : int, 'CA' as state : chararray;");
pigServer.registerQuery("C = foreach B generate c1 as c1 : chararray;");
@@ -1604,7 +1608,7 @@ public class TestPruneColumn extends Tes
// See PIG-1127
@Test
public void testSharedSchemaObject() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile10.toString(), pigServer.getPigContext()) + "' AS (a0, a1:map[], a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile10.toString()), pigServer.getPigContext()) + "' AS (a0, a1:map[], a2);");
pigServer.registerQuery("B = foreach A generate a1;");
pigServer.registerQuery("C = limit B 10;");
@@ -1622,8 +1626,8 @@ public class TestPruneColumn extends Tes
// See PIG-1142
@Test
public void testJoin4() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
pigServer.registerQuery("C = join A by a2, B by b2;");
pigServer.registerQuery("D = foreach C generate $0, $1, $2;");
@@ -1647,7 +1651,7 @@ public class TestPruneColumn extends Tes
@Test
public void testFilter4() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2:int);");
pigServer.registerQuery("B = filter A by a2==3;");
pigServer.registerQuery("C = foreach B generate $2;");
@@ -1664,7 +1668,7 @@ public class TestPruneColumn extends Tes
@Test
public void testSplit3() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2:int);");
pigServer.registerQuery("split A into B if a2==3, C if a2<3;");
pigServer.registerQuery("C = foreach B generate $2;");
@@ -1681,7 +1685,7 @@ public class TestPruneColumn extends Tes
@Test
public void testOrderBy3() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = order A by a2;");
pigServer.registerQuery("C = foreach B generate a2;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1705,9 +1709,9 @@ public class TestPruneColumn extends Tes
@Test
public void testCogroup9() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
- pigServer.registerQuery("C = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (c0, c1, c2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (b0, b1, b2);");
+ pigServer.registerQuery("C = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (c0, c1, c2);");
pigServer.registerQuery("D = cogroup A by a2, B by b2, C by c2;");
pigServer.registerQuery("E = foreach D generate $1, $2;");
Iterator<Tuple> iter = pigServer.openIterator("E");
@@ -1732,8 +1736,8 @@ public class TestPruneColumn extends Tes
// See PIG-1165
@Test
public void testOrderbyWrongSignature() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b1);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b1);");
pigServer.registerQuery("C = order A by a1;");
pigServer.registerQuery("D = join C by a1, B by b0;");
pigServer.registerQuery("E = foreach D generate a1, b0, b1;");
@@ -1753,8 +1757,8 @@ public class TestPruneColumn extends Tes
// See PIG-1146
@Test
public void testUnionMixedPruning() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1:chararray, a2);");
- pigServer.registerQuery("B = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "' AS (b0, b2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1:chararray, a2);");
+ pigServer.registerQuery("B = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "' AS (b0, b2);");
pigServer.registerQuery("C = foreach B generate b0, 'hello', b2;");
pigServer.registerQuery("D = union A, C;");
pigServer.registerQuery("E = foreach D generate $0, $2;");
@@ -1797,9 +1801,9 @@ public class TestPruneColumn extends Tes
// See PIG-1176
@Test
public void testUnionMixedSchemaPruning() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate a0;;");
- pigServer.registerQuery("C = load '"+ Util.generateURI(tmpFile2.toString(), pigServer.getPigContext()) + "';");
+ pigServer.registerQuery("C = load '"+ Util.generateURI(Util.encodeEscape(tmpFile2.toString()), pigServer.getPigContext()) + "';");
pigServer.registerQuery("D = foreach C generate $0;");
pigServer.registerQuery("E = union B, D;");
Iterator<Tuple> iter = pigServer.openIterator("E");
@@ -1843,7 +1847,7 @@ public class TestPruneColumn extends Tes
public void testForEachFlatten() throws Exception {
File inputFile = Util.createInputFile("table_testForEachFlatten", "", new String[]{"oiue\tM\t{(3),(4)}\t{(toronto),(montreal)}"});
- pigServer.registerQuery("A = load '"+inputFile.toString()+"' as (a0:chararray, a1:chararray, a2:bag{t:tuple(id:chararray)}, a3:bag{t:tuple(loc:chararray)});");
+ pigServer.registerQuery("A = load '"+Util.encodeEscape(inputFile.toString())+"' as (a0:chararray, a1:chararray, a2:bag{t:tuple(id:chararray)}, a3:bag{t:tuple(loc:chararray)});");
pigServer.registerQuery("B = foreach A generate a0, a1, flatten(a2), flatten(a3), 10;");
pigServer.registerQuery("C = foreach B generate a0, $4;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1872,7 +1876,7 @@ public class TestPruneColumn extends Tes
// See PIG-1210
@Test
public void testFieldsToReadDuplicatedEntry() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate a0+a0, a1, a2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
@@ -1892,7 +1896,7 @@ public class TestPruneColumn extends Tes
// See PIG-1272
@Test
public void testSplit4() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate a0;");
pigServer.registerQuery("C = join A by a0, B by a0;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1910,7 +1914,7 @@ public class TestPruneColumn extends Tes
@Test
public void testSplit5() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile11.toString(), pigServer.getPigContext()) + "' AS (a0:int, a1:int, a2:int);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile11.toString()), pigServer.getPigContext()) + "' AS (a0:int, a1:int, a2:int);");
pigServer.registerQuery("B = foreach A generate a0, a1;");
pigServer.registerQuery("C = join A by a0, B by a0;");
pigServer.registerQuery("D = filter C by A::a1>=B::a1;");
@@ -1932,7 +1936,7 @@ public class TestPruneColumn extends Tes
// See PIG-1493
@Test
public void testInconsistentPruning() throws Exception {
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' AS (a0:chararray, a1:chararray, a2);");
pigServer.registerQuery("B = foreach A generate CONCAT(a0,a1) as b0, a0, a2;");
pigServer.registerQuery("C = foreach B generate a0, a2;");
Iterator<Tuple> iter = pigServer.openIterator("C");
@@ -1954,12 +1958,12 @@ public class TestPruneColumn extends Tes
Path output1 = FileLocalizer.getTemporaryPath(pigServer.getPigContext());
Path output2 = FileLocalizer.getTemporaryPath(pigServer.getPigContext());
pigServer.setBatchOn();
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile5.toString(), pigServer.getPigContext()) + "' AS (a0, a1, a2, a3);");
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile5.toString()), pigServer.getPigContext()) + "' AS (a0, a1, a2, a3);");
pigServer.registerQuery("B = foreach A generate a0, a1, a2;");
- pigServer.registerQuery("store B into '" + Util.generateURI(output1.toString(), pigServer.getPigContext()) + "';");
+ pigServer.registerQuery("store B into '" + Util.generateURI(Util.encodeEscape(output1.toString()), pigServer.getPigContext()) + "';");
pigServer.registerQuery("C = order B by a2;");
pigServer.registerQuery("D = foreach C generate a2;");
- pigServer.registerQuery("store D into '" + Util.generateURI(output2.toString(), pigServer.getPigContext()) + "';");
+ pigServer.registerQuery("store D into '" + Util.generateURI(Util.encodeEscape(output2.toString()), pigServer.getPigContext()) + "';");
pigServer.executeBatch();
BufferedReader reader1 = new BufferedReader(new InputStreamReader(FileLocalizer.openDFSFile(output1.toString(), pigServer.getPigContext().getProperties())));
@@ -2044,7 +2048,7 @@ public class TestPruneColumn extends Tes
}
public void testAliasInRequiredFieldList() throws Exception{
- pigServer.registerQuery("A = load '"+ Util.generateURI(tmpFile1.toString(), pigServer.getPigContext()) + "' using "
+ pigServer.registerQuery("A = load '"+ Util.generateURI(Util.encodeEscape(tmpFile1.toString()), pigServer.getPigContext()) + "' using "
+ PruneColumnEvalFunc.class.getName() +"() as (a0, a1, a2);");
pigServer.registerQuery("B = foreach A generate a1, a2;");
Iterator<Tuple> iter = pigServer.openIterator("B");
Modified: pig/branches/branch-0.10/test/org/apache/pig/test/TestSample.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.10/test/org/apache/pig/test/TestSample.java?rev=1404375&r1=1404374&r2=1404375&view=diff
==============================================================================
--- pig/branches/branch-0.10/test/org/apache/pig/test/TestSample.java (original)
+++ pig/branches/branch-0.10/test/org/apache/pig/test/TestSample.java Wed Oct 31 21:57:37 2012
@@ -47,12 +47,15 @@ public class TestSample {
pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
tmpFile = File.createTempFile( this.getClass().getName(), ".txt");
+ tmpFile.delete(); // we don't want the file, just the temp path
+
+ tmpfilepath = Util.removeColon(tmpFile.getCanonicalPath());
+
String input[] = new String[DATALEN];
for(int i = 0; i < DATALEN; i++) {
input[i] = Integer.toString(i);
}
-
- tmpfilepath = tmpFile.getCanonicalPath();
+
Util.createInputFile(cluster, tmpfilepath, input);
}
@@ -90,21 +93,21 @@ public class TestSample {
public void testSample_None()
throws Exception
{
- verify("myid = sample (load '"+ tmpfilepath + "') 0.0;", 0, 0);
+ verify("myid = sample (load '"+ Util.encodeEscape(tmpfilepath) + "') 0.0;", 0, 0);
}
@Test
public void testSample_All()
throws Exception
{
- verify("myid = sample (load '"+ tmpfilepath + "') 1.0;", DATALEN, DATALEN);
+ verify("myid = sample (load '"+ Util.encodeEscape(tmpfilepath) + "') 1.0;", DATALEN, DATALEN);
}
@Test
public void testSample_Some()
throws Exception
{
- verify("myid = sample (load '"+ tmpfilepath + "') 0.5;", DATALEN/3, DATALEN*2/3);
+ verify("myid = sample (load '"+ Util.encodeEscape(tmpfilepath) + "') 0.5;", DATALEN/3, DATALEN*2/3);
}
@Test
Modified: pig/branches/branch-0.10/test/org/apache/pig/test/TestStreaming.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.10/test/org/apache/pig/test/TestStreaming.java?rev=1404375&r1=1404374&r2=1404375&view=diff
==============================================================================
--- pig/branches/branch-0.10/test/org/apache/pig/test/TestStreaming.java (original)
+++ pig/branches/branch-0.10/test/org/apache/pig/test/TestStreaming.java Wed Oct 31 21:57:37 2012
@@ -66,10 +66,12 @@ public class TestStreaming {
private static final String simpleEchoStreamingCommand;
static {
- if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS"))
- simpleEchoStreamingCommand = "perl -ne 'print \\\"$_\\\"'";
- else
- simpleEchoStreamingCommand = "perl -ne 'print \"$_\"'";
+ String quote = "'";
+ if (Util.WINDOWS) {
+ quote= "\"";
+ }
+
+ simpleEchoStreamingCommand = "perl -ne "+quote+"print $_"+quote;
}
private Tuple[] setupExpectedResults(Object[] firstField, Object[] secondField) throws ExecException {
@@ -302,13 +304,13 @@ public class TestStreaming {
// Pig query to run
pigServer.registerQuery(
- "define CMD1 `" + command1.getName() + " foo` " +
+ "define CMD1 `perl " + command1.getName() + " foo` " +
"ship ('" + Util.encodeEscape(command1.toString()) + "') " +
"input('foo' using " + PigStreaming.class.getName() + "(',')) " +
"output(stdout using " + PigStreaming.class.getName() + "(',')) " +
"stderr();");
pigServer.registerQuery(
- "define CMD2 `" + command2.getName() + " bar` " +
+ "define CMD2 `perl " + command2.getName() + " bar` " +
"ship ('" + Util.encodeEscape(command2.toString()) + "') " +
"input('bar' using " + PigStreaming.class.getName() + "(',')) " +
"output(stdout using " + PigStreaming.class.getName() + "(',')) " +
@@ -372,13 +374,13 @@ public class TestStreaming {
pigServer.registerQuery("define PS " + PigStreaming.class.getName() + "(',');");
pigServer.registerQuery(
- "define CMD1 `" + command1.getName() + " foo` " +
+ "define CMD1 `perl " + command1.getName() + " foo` " +
"ship ('" + Util.encodeEscape(command1.toString()) + "') " +
"input('foo' using PS )" +
"output(stdout using PS ) " +
"stderr();");
pigServer.registerQuery(
- "define CMD2 `" + command2.getName() + " bar` " +
+ "define CMD2 `perl " + command2.getName() + " bar` " +
"ship ('" + Util.encodeEscape(command2.toString()) + "') " +
"input('bar' using PS ) " +
"output(stdout using PS ) " +
@@ -444,12 +446,12 @@ public class TestStreaming {
// Pig query to run
pigServer.registerQuery(
- "define CMD1 `script1.pl foo` " +
+ "define CMD1 `perl script1.pl foo` " +
"cache ('" + c1 + "#script1.pl') " +
"input('foo' using " + PigStreaming.class.getName() + "(',')) " +
"stderr();");
pigServer.registerQuery(
- "define CMD2 `script2.pl bar` " +
+ "define CMD2 `perl script2.pl bar` " +
"cache ('" + c2 + "#script2.pl') " +
"input('bar' using " + PigStreaming.class.getName() + "(',')) " +
"stderr();");
@@ -510,7 +512,7 @@ public class TestStreaming {
// Pig query to run
pigServer.registerQuery(
- "define CMD `" + command.getName() + " foo bar` " +
+ "define CMD `perl " + command.getName() + " foo bar` " +
"ship ('" + Util.encodeEscape(command.toString()) + "') " +
"output('foo' using " + PigStreaming.class.getName() + "(','), " +
"'bar' using " + PigStreaming.class.getName() + "(',')) " +
@@ -573,7 +575,7 @@ public class TestStreaming {
"define PS " + PigStreaming.class.getName() + "(',');");
pigServer.registerQuery(
- "define CMD `" + command.getName() + " foo bar` " +
+ "define CMD `perl " + command.getName() + " foo bar` " +
"ship ('" + Util.encodeEscape(command.toString()) + "') " +
"output('foo' using PS, " +
"'bar' using PS) " +
@@ -633,7 +635,7 @@ public class TestStreaming {
Util.toDataByteArrays(expectedSecondFields));
// Pig query to run
pigServer.registerQuery(
- "define CMD `" + command.getName() + " foo bar foobar` " +
+ "define CMD `perl " + command.getName() + " foo bar foobar` " +
"ship ('" + Util.encodeEscape(command.toString()) + "') " +
"input('foo' using " + PigStreaming.class.getName() + "(',')) " +
"output('bar', " +
@@ -766,7 +768,7 @@ public class TestStreaming {
};
File command1 = Util.createInputFile("script", "pl", script);
String query =
- "define CMD1 `" + command1.getName() + " foo` " +
+ "define CMD1 `perl " + command1.getName() + " foo` " +
"ship ('" + Util.encodeEscape(command1.toString()) + "') " +
"input('foo' using " + PigStreaming.class.getName() + "(',')) " +
"output(stdout using " + PigStreaming.class.getName() + "(',')) " +
Modified: pig/branches/branch-0.10/test/org/apache/pig/test/TestStreamingLocal.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.10/test/org/apache/pig/test/TestStreamingLocal.java?rev=1404375&r1=1404374&r2=1404375&view=diff
==============================================================================
--- pig/branches/branch-0.10/test/org/apache/pig/test/TestStreamingLocal.java (original)
+++ pig/branches/branch-0.10/test/org/apache/pig/test/TestStreamingLocal.java Wed Oct 31 21:57:37 2012
@@ -20,7 +20,6 @@ package org.apache.pig.test;
import java.io.File;
import java.util.ArrayList;
import java.util.Iterator;
-import java.util.List;
import junit.framework.TestCase;
@@ -42,10 +41,12 @@ public class TestStreamingLocal extends
private static final String simpleEchoStreamingCommand;
static {
- if (System.getProperty("os.name").toUpperCase().startsWith("WINDOWS"))
- simpleEchoStreamingCommand = "perl -ne 'print \\\"$_\\\"'";
- else
- simpleEchoStreamingCommand = "perl -ne 'print \"$_\"'";
+ String quote = "'";
+ if (Util.WINDOWS) {
+ quote= "\"";
+ }
+
+ simpleEchoStreamingCommand = "perl -ne "+quote+"print $_"+quote;
}
@Before
@@ -98,8 +99,9 @@ public class TestStreamingLocal extends
}
// Pig query to run
- pigServer.registerQuery("IP = load 'file:" + Util.encodeEscape(input.toString()) + "' using " +
- PigStorage.class.getName() + "(',');");
+ pigServer.registerQuery("IP = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',');");
pigServer.registerQuery("FILTERED_DATA = filter IP by $1 > '3';");
pigServer.registerQuery("S1 = stream FILTERED_DATA through `" +
simpleEchoStreamingCommand + "`;");
@@ -140,8 +142,9 @@ public class TestStreamingLocal extends
.toDataByteArrays(expectedSecondFields));
}
// Pig query to run
- pigServer.registerQuery("IP = load 'file:" + Util.encodeEscape(input.toString()) + "' using " +
- PigStorage.class.getName() + "(',');");
+ pigServer.registerQuery("IP = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',');");
pigServer.registerQuery("FILTERED_DATA = filter IP by $1 > '3';");
if(withTypes[i] == true) {
pigServer.registerQuery("STREAMED_DATA = stream FILTERED_DATA through `" +
@@ -181,8 +184,9 @@ public class TestStreamingLocal extends
}
// Pig query to run
- pigServer.registerQuery("IP = load 'file:" + Util.encodeEscape(input.toString()) + "' using " +
- PigStorage.class.getName() + "(',');");
+ pigServer.registerQuery("IP = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',');");
pigServer.registerQuery("FILTERED_DATA = filter IP by $1 > '3';");
pigServer.registerQuery("GROUPED_DATA = group FILTERED_DATA by $0;");
pigServer.registerQuery("FLATTENED_GROUPED_DATA = foreach GROUPED_DATA " +
@@ -229,8 +233,9 @@ public class TestStreamingLocal extends
//setupExpectedResults(expectedFirstFields, expectedSecondFields);
// Pig query to run
- pigServer.registerQuery("IP = load 'file:" + Util.encodeEscape(input.toString()) + "' using " +
- PigStorage.class.getName() + "(',');");
+ pigServer.registerQuery("IP = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',');");
pigServer.registerQuery("FILTERED_DATA = filter IP by $1 > '3';");
pigServer.registerQuery("S1 = stream FILTERED_DATA through `" +
simpleEchoStreamingCommand + "`;");
@@ -276,8 +281,9 @@ public class TestStreamingLocal extends
// Pig query to run
pigServer.registerQuery("define CMD `" + simpleEchoStreamingCommand +
" | " + simpleEchoStreamingCommand + "`;");
- pigServer.registerQuery("IP = load 'file:" + Util.encodeEscape(input.toString()) + "' using " +
- PigStorage.class.getName() + "(',');");
+ pigServer.registerQuery("IP = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',');");
if(withTypes[i] == true) {
pigServer.registerQuery("OP = stream IP through CMD as (f0:chararray, f1:int);");
} else {
@@ -288,47 +294,40 @@ public class TestStreamingLocal extends
Util.checkQueryOutputs(pigServer.openIterator("OP"), expectedResults);
}
}
-
+
@Test
- // See PIG-2442
- public void testTwoStreamingMultiStore()
+ public void testJoinTwoStreamingRelations()
throws Exception {
- File input = File.createTempFile("tmp", "");
- input.delete();
- Util.createLocalInputFile(input.getAbsolutePath(), new String[] {"first", "second", "third"});
-
- File output1 = File.createTempFile("tmp", "");
- output1.delete();
-
- File output2 = File.createTempFile("tmp", "");
- output2.delete();
-
- pigServer.setBatchOn();
- pigServer.registerQuery("A = load '" + input.getAbsolutePath() + "';");
- pigServer.registerQuery("B1 = stream A through `cat`;");
- pigServer.registerQuery("B1 = foreach B1 generate $0;");
- pigServer.registerQuery("STORE B1 INTO '" + output1.getAbsolutePath() + "' USING PigStorage();");
- pigServer.registerQuery("B2 = STREAM B1 THROUGH `cat`;");
- pigServer.registerQuery("STORE B2 INTO '" + output2.getAbsolutePath() + "' USING PigStorage();");
-
- pigServer.executeBatch();
-
- List<Tuple> list = Util.readFile2TupleList(output1.getAbsolutePath() + File.separator +
- "part-m-00000", "\t");
- assertTrue(list.get(0).get(0).equals("first"));
- assertTrue(list.get(1).get(0).equals("second"));
- assertTrue(list.get(2).get(0).equals("third"));
-
- list = Util.readFile2TupleList(output2.getAbsolutePath() + File.separator +
- "part-m-00000", "\t");
- assertTrue(list.get(0).get(0).equals("first"));
- assertTrue(list.get(1).get(0).equals("second"));
- assertTrue(list.get(2).get(0).equals("third"));
- }
-
- @Test
- public void testJoinTwoStreamingRelations() {
-
+ ArrayList<String> list = new ArrayList<String>();
+ for (int i=0; i<10000; i++) {
+ list.add("A," + i);
+ }
+ File input = Util.createInputFile("tmp", "", list.toArray(new String[0]));
+
+ // Expected results
+ Tuple expected = TupleFactory.getInstance().newTuple(4);
+ expected.set(0, "A");
+ expected.set(1, 0);
+ expected.set(2, "A");
+ expected.set(3, 0);
+
+ pigServer.registerQuery("A = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',') as (a0, a1);");
+ pigServer.registerQuery("B = stream A through `head -1` as (a0, a1);");
+ pigServer.registerQuery("C = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',') as (a0, a1);");
+ pigServer.registerQuery("D = stream C through `head -1` as (a0, a1);");
+ pigServer.registerQuery("E = join B by a0, D by a0;");
+
+ Iterator<Tuple> iter = pigServer.openIterator("E");
+ int count = 0;
+ while (iter.hasNext()) {
+ Assert.assertEquals(expected.toString(), iter.next().toString());
+ count++;
+ }
+ Assert.assertTrue(count == 1);
}
@Test
@@ -361,8 +360,9 @@ public class TestStreamingLocal extends
// Pig query to run
pigServer.registerQuery("define CMD `"+ simpleEchoStreamingCommand +
"` input(stdin);");
- pigServer.registerQuery("IP = load 'file:" + Util.encodeEscape(input.toString()) + "' using " +
- PigStorage.class.getName() + "(',');");
+ pigServer.registerQuery("IP = load '" +
+ Util.generateURI(Util.encodeEscape(input.toString()), pigServer.getPigContext()) +
+ "' using " + PigStorage.class.getName() + "(',');");
pigServer.registerQuery("FILTERED_DATA = filter IP by $1 > '3';");
if(withTypes[i] == true) {
pigServer.registerQuery("OP = stream FILTERED_DATA through `" +