You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by pr...@apache.org on 2009/12/17 19:36:36 UTC
svn commit: r891828 [13/15] - in /hadoop/pig/branches/load-store-redesign:
./ contrib/
contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/
contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/string/
contrib/...
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/RuleMatcher.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/RuleMatcher.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/RuleMatcher.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/RuleMatcher.java Thu Dec 17 18:36:28 2009
@@ -74,11 +74,50 @@
if (mRule.getWalkerAlgo() == Rule.WalkerAlgo.DependencyOrderWalker)
DependencyOrderWalker();
else if (mRule.getWalkerAlgo() == Rule.WalkerAlgo.DepthFirstWalker)
- DepthFirstWalker();
+ DepthFirstWalker();
+ else if (mRule.getWalkerAlgo() == Rule.WalkerAlgo.ReverseDependencyOrderWalker)
+ ReverseDependencyOrderWalker();
return (mMatches.size()!=0);
}
+ private void ReverseDependencyOrderWalker()
+ {
+ List<O> fifo = new ArrayList<O>();
+ Set<O> seen = new HashSet<O>();
+ List<O> roots = mPlan.getRoots();
+ if (roots == null) return;
+ for (O op : roots) {
+ RDODoAllSuccessors(op, seen, fifo);
+ }
+
+ for (O op: fifo) {
+ if (beginMatch(op))
+ mPrelimMatches.add(mMatch);
+ }
+
+ if(mPrelimMatches.size() > 0) {
+ processPreliminaryMatches();
+ }
+ }
+
+ private void RDODoAllSuccessors(O node, Set<O> seen, Collection<O> fifo)
+ {
+ if (!seen.contains(node)) {
+ // We haven't seen this one before.
+ Collection<O> succs = mPlan.getSuccessors(node);
+ if (succs != null && succs.size() > 0) {
+ // Do all our successors before ourself
+ for (O op : succs) {
+ RDODoAllSuccessors(op, seen, fifo);
+ }
+ }
+ // Now do ourself
+ seen.add(node);
+ fifo.add(node);
+ }
+ }
+
private void DependencyOrderWalker()
{
List<O> fifo = new ArrayList<O>();
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/Transformer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/Transformer.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/Transformer.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/plan/optimizer/Transformer.java Thu Dec 17 18:36:28 2009
@@ -29,13 +29,14 @@
* apply to a graph. This class is a special case of a PlanVisitor,so it
* can navigate the plan.
*/
-public abstract class Transformer<O extends Operator, P extends OperatorPlan<O>> extends PlanVisitor<O, P> {
+public abstract class Transformer<O extends Operator, P extends OperatorPlan<O>> {
+ protected P mPlan;
/**
* @param plan OperatorPlan to be optimized.
*/
- protected Transformer(P plan, PlanWalker<O, P> walker) {
- super(plan, walker);
+ protected Transformer(P plan) {
+ mPlan = plan;
}
/**
@@ -67,6 +68,11 @@
* @throws OptimizerException
*/
public abstract void reset() throws OptimizerException;
+
+ public P getPlan()
+ {
+ return mPlan;
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/util/UDFContext.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/util/UDFContext.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/util/UDFContext.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/impl/util/UDFContext.java Thu Dec 17 18:36:28 2009
@@ -174,4 +174,7 @@
return hc;
}
+ public void reset() {
+ udfConfs.clear();
+ }
}
\ No newline at end of file
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParamLoader.jj
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParamLoader.jj?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParamLoader.jj (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParamLoader.jj Thu Dec 17 18:36:28 2009
@@ -32,6 +32,11 @@
import java.io.IOException;
import java.util.Hashtable;
+//warnings in by code generated by javacc cannot be fixed here,
+// so suppressing all warnings for this class. But this does not help in
+//supressing Warnings in other classes generated by this .jj file
+@SuppressWarnings("all")
+
public class ParamLoader {
private PreprocessorContext pc;
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParameterSubstitutionPreprocessor.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParameterSubstitutionPreprocessor.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParameterSubstitutionPreprocessor.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/ParameterSubstitutionPreprocessor.java Thu Dec 17 18:36:28 2009
@@ -52,7 +52,6 @@
paramParser = new ParamLoader(sr);
paramParser.setContext(pc);
-
pigParser = new PigFileParser(sr);
pigParser.setContext(pc);
}
@@ -87,33 +86,10 @@
}
private void parsePigFile(BufferedReader in, Writer out) throws ParseException {
-
+ pigParser.setOutputWriter(out);
+ pigParser.ReInit(in);
try {
- String line;
- int lineNum=0;
-
- while((line = in.readLine())!= null)
- {
- lineNum++;
-
- if (line.length() == 0) {
- out.append("\n");
- continue;
- }
-
- if ( !(line.startsWith("%")) ) {
- //process an ordinary pig line - perform substitution
- String sub_line = pc.substitute(line);
- out.append(sub_line);
- out.append("\n");
- continue;
- }
-
- //parse the declare/default line
- pigParser.ReInit(new StringReader(line));
- pigParser.Parse();
- }
-
+ pigParser.Parse();
//close input and output streams
in.close();
out.flush();
@@ -122,7 +98,6 @@
RuntimeException rte = new RuntimeException(e.getMessage() , e);
throw rte;
}
-
}
/*
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PigFileParser.jj
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PigFileParser.jj?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PigFileParser.jj (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PigFileParser.jj Thu Dec 17 18:36:28 2009
@@ -18,7 +18,7 @@
/**
* This grammar is used to process %declare and %default commands
- * within pig script
+ * within pig script, and substitute parameter values
*/
options {
@@ -26,20 +26,34 @@
STATIC = false;
IGNORE_CASE = true;
JAVA_UNICODE_ESCAPE = true;
+ //DEBUG_PARSER = true;
+ LOOKAHEAD = 2;
+
}
PARSER_BEGIN(PigFileParser)
+
package org.apache.pig.tools.parameters;
import java.io.IOException;
import java.io.PrintWriter;
import java.util.Hashtable;
+import java.io.Writer;
+import java.lang.StringBuilder;
+
+//warnings in by code generated by javacc cannot be fixed here,
+// so suppressing all warnings for this class. But this does not help in
+//supressing Warnings in other classes generated by this .jj file
+@SuppressWarnings("all")
public class PigFileParser {
private PreprocessorContext pc;
+ private Writer out;
public void setContext(PreprocessorContext pc) {
this.pc = pc;
}
-
+ public void setOutputWriter(Writer out) {
+ this.out = out;
+ }
private static String unquote(String s)
{
if (s.charAt(0) == '\'' && s.charAt(s.length()-1) == '\'')
@@ -53,12 +67,20 @@
PARSER_END(PigFileParser)
-SKIP :
+TOKEN :
{
- "\n"
-| "\r"
-| " "
-| "\t"
+ <NEWLINE: "\n" | "\r">
+ |
+ <SPACE: " " | "\t">
+}
+
+// comments(single line and multi-line)
+TOKEN :
+{
+ <COMMENT: "--"(~["\r","\n"])*
+ | "#!" (~["\r","\n"])*
+ | "/*" (~["*"])* "*" ("*" | (~["*","/"] (~["*"])* "*"))* "/"
+ >
}
TOKEN:
@@ -68,13 +90,15 @@
<#DIGIT : ["0"-"9"] >
|
<#SPECIALCHAR : ["_"] >
+ |
+ <#DOLLAR : ["$"]>
}
TOKEN :
{
<DECLARE: "%declare" >
|
- <PIGDEFAULT: "%default" >
+ <PIGDEFAULT: "%default" >
}
TOKEN :
@@ -85,30 +109,75 @@
|
<SHELLCMD: "`" (~["`"])* "`" >
|
- <OTHER: ~["\"" , "'" , "`" , "a"-"z" , "A"-"Z" , "_" , "#" , "=" , " " , "\n" , "\t" , "\r", "%"] (~["\n","\r"])* >
+ // see others() rule for use of OTHER and NOT_OTHER_CHAR
+ // others() is supposed to match 'everything else'. To ensure that others()
+ // don't swallow other(all the ones above) tokens, it uses two tokens OTHER and NOT_OTHER_CHAR
+ // NOT_OTHER_CHAR consists of first characters of other tokens, and OTHER consists of one
+ // or more chars that don't belong to NOT_OTHER_CHAR. Since the tokeniser matches the longest
+ // match, other tokens will get matched instead of NOT_OTHER_CHAR.
+ <OTHER: (~["\"" , "'" , "`" , "a"-"z" , "A"-"Z" , "_" , "#" , "=" , " " , "\n" , "\t" , "\r", "%", "/", "-", "$"])+ >
+ |
+ <NOT_OTHER_CHAR: ["\"" , "'" , "`" , "a"-"z" , "A"-"Z" , "_" , "#" , "=" , " " , "\n" , "\t" , "\r", "%", "/", "-", "$"] >
+
}
+void Parse() throws IOException : {}
+{
+ (input())*<EOF>
+}
-
-void Parse() throws IOException :
+void input() throws IOException :
{
- Token id, val;
- Boolean overwrite;
String s;
+ Token strTok;
}
{
- <DECLARE> { overwrite=true; }
+ <DECLARE>
(
- id=<IDENTIFIER>
+ param_value(true) // overwrite=true
+ )
+ |
+ <PIGDEFAULT>
+ (
+ param_value(false) // overwrite=false
+ )
+ |
+ s = paramString(){}
+ {
+ //process an ordinary pig line - perform substitution
+ String sub_line = pc.substitute(s);
+ out.append(sub_line );
+ }
+ |
+ strTok = <IDENTIFIER>{}
+ {
+ out.append(strTok.image );
+ }
+ |
+ write_ignore_toks()
+
+}
+
+void param_value(boolean overwrite) throws IOException:
+{
+ Token id, val;
+ String s;
+ String other;
+ Token strTok;
+}
+{
+ (ignore_toks_nonewline())*
+ id=<IDENTIFIER>
+ (ignore_toks_nonewline())*
(
- val=<OTHER> { pc.processOrdLine(id.image , val.image);}
+ s=others() (ignore_toks_nonewline())* write_newline() { pc.processOrdLine(id.image , s, overwrite);}
|
- val=<IDENTIFIER>
+ val=<IDENTIFIER> // this construct is for cases like a.123
{s = val.image;}
- (
- val =<OTHER>
- {s += val.image;}
+ ( LOOKAHEAD( 2 )
+ other = others() (ignore_toks_nonewline())* write_newline()
+ { s += other; }
)?
{pc.processOrdLine(id.image , s, overwrite);}
|
@@ -117,25 +186,136 @@
val=<LITERAL> { s = unquote(val.image); pc.processOrdLine(id.image, s, overwrite); }
)
- )
- |
- <PIGDEFAULT> { overwrite=false; }
+}
+
+//match others, see comments above
+//on description of OTHER , NOT_OTHER_CHAR
+String others() throws IOException :
+{ Token t;
+ StringBuilder sb = new StringBuilder();
+}
+{
(
- id=<IDENTIFIER>
- (
- val=<OTHER> { pc.processOrdLine(id.image , val.image);}
- |
- val=<IDENTIFIER>
- {s = val.image;}
- (
- val =<OTHER>
- {s += val.image;}
- )?
- {pc.processOrdLine(id.image , s, overwrite);}
- |
- val=<SHELLCMD> { pc.processShellCmd(id.image , val.image , overwrite); }
- |
- val=<LITERAL> { s = unquote(val.image); pc.processOrdLine(id.image , s, overwrite); }
- )
+ t=<OTHER>
+ {
+ sb.append(t.image);
+ }
+ |
+ t=<NOT_OTHER_CHAR>
+ {
+ sb.append(t.image);
+ }
)
+ (
+ t=<OTHER>
+ {
+ sb.append(t.image);
+ }
+ |
+ t=<NOT_OTHER_CHAR>
+ {
+ sb.append(t.image);
+ }
+ |
+ t=<IDENTIFIER>
+ {
+ //eg of this match is a unquoted filename - /d1/abc/
+ sb.append(t.image);
+ }
+ |
+ t=<LITERAL>
+ {
+ sb.append(t.image);
+ }
+ |
+ t=<SHELLCMD>
+ {
+ sb.append(t.image);
+ }
+ )*{
+ return sb.toString();
+ }
+}
+
+
+// a string that can contain parameter
+String paramString() throws IOException :
+{Token t;
+ String str;
+}
+{
+ str = others()
+ {
+ return str;
+ }
+ |
+ t=<LITERAL>
+ {
+ return t.image;
+ }
+ |
+ t = <SHELLCMD>{}
+ {
+ return t.image;
+ }
+}
+
+// write the newlines,spaces and comments to preserve formatting
+void write_ignore_toks() throws IOException :
+{ String str;}
+{
+ str = ignore_toks(){
+ out.append(str);
+ }
+}
+
+
+// match the newlines,spaces and comments
+String ignore_toks_nonewline() throws IOException :
+{ Token t; }
+{
+ t = <COMMENT>{
+ return t.image;
+ }
+ |
+ t = <SPACE>{
+ return t.image;
+ }
+}
+
+
+// match the newlines,spaces and comments
+String ignore_toks() throws IOException :
+{ Token t;
+ String str;
}
+{
+ t = <COMMENT>{
+ return t.image;
+ }
+ |
+ str = space_or_newline(){
+ return str;
+ }
+}
+
+String space_or_newline() :
+{ Token t; }
+{
+ t = <SPACE>{
+ return t.image;
+ }
+ |
+ t = <NEWLINE>{
+ return t.image;
+ }
+}
+
+void write_newline() throws IOException:
+{ Token t; }
+{
+ t = <NEWLINE>{
+ out.append(t.image);
+ }
+}
+
Modified: hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PreprocessorContext.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PreprocessorContext.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PreprocessorContext.java (original)
+++ hadoop/pig/branches/load-store-redesign/src/org/apache/pig/tools/parameters/PreprocessorContext.java Thu Dec 17 18:36:28 2009
@@ -210,16 +210,16 @@
return streamData.trim();
}
- private static String id_regex = "\\$[_]*[a-zA-Z][a-zA-Z_0-9]*";
+ private Pattern id_pattern = Pattern.compile("\\$[_]*[a-zA-Z][a-zA-Z_0-9]*");
+
public String substitute(String line) {
int index = line.indexOf('$');
if (index == -1) return line;
String replaced_line = line;
-
- Pattern identifier = Pattern.compile( id_regex );
- Matcher keyMatcher = identifier.matcher( line );
+
+ Matcher keyMatcher = id_pattern.matcher( line );
String key="";
String val="";
Modified: hadoop/pig/branches/load-store-redesign/test/commit-tests
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/commit-tests?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/commit-tests (original)
+++ hadoop/pig/branches/load-store-redesign/test/commit-tests Thu Dec 17 18:36:28 2009
@@ -1,6 +1,5 @@
**/TestAdd.java
**/TestBagFormat.java
-**/TestBinaryStorage.java
**/TestBoolean.java
**/TestBuiltin.java
**/TestCmdLineParser.java
@@ -12,22 +11,17 @@
**/TestDeleteOnFail.java
**/TestDivide.java
**/TestEqualTo.java
-**/TestEvalPipelineLocal.java
**/TestExampleGenerator.java
**/TestTestFilter.java
**/TestForEach.java
-**/TestForEachNestedPlanLocal.java
**/TestFuncSpec.java
**/TestGTOrEqual.java
**/TestGreaterThan.java
-**/TestImplicitSplit.java
**/TestInputOutputFileValidator.java
**/TestInstantiateFunc.java
**/TestLTOrEqual.java
**/TestLessThan.java
**/TestLoad.java
-**/TestLocal.java
-**/TestLocal2.java
**/TestLocalPOSplit.java
**/TestLocalRearrange.java
**/TestLogToPhyCompiler.java
@@ -53,17 +47,12 @@
**/TestPhyOp.java
**/TestPigScriptParser.java
**/TestPigServer.java
-**/TestPigSplit.java
**/TestPoissonSampleLoader.java
-**/TestProject.java
**/TestRegexp.java
-**/TestSample.java
**/TestSchema.java
**/TestSchemaParser.java
**/TestSchemaUtil.java
**/TestStore.java
-**/TestStreaming.java
-**/TestStreamingLocal.java
**/TestTextDataParse.java
**/TestTupleFormat.java
**/TestTypeChecking.java
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestAccumulator.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestAccumulator.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestAccumulator.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestAccumulator.java Thu Dec 17 18:36:28 2009
@@ -42,6 +42,7 @@
// pigServer = new PigServer(ExecType.LOCAL);
pigServer.getPigContext().getProperties().setProperty("pig.accumulative.batchsize", "2");
pigServer.getPigContext().getProperties().setProperty("pig.exec.batchsize", "2");
+ pigServer.getPigContext().getProperties().setProperty("pig.exec.nocombiner", "true");
}
@Before
@@ -100,7 +101,6 @@
Util.deleteFile(cluster, INPUT_FILE3);
}
-
public void testAccumBasic() throws IOException{
// test group by
pigServer.registerQuery("A = load '" + INPUT_FILE + "' as (id:int, fruit);");
@@ -405,12 +405,12 @@
Tuple t = iter.next();
assertEquals(expected.get((Integer)t.get(0)), (String)t.get(1));
}
- }
+ }
public void testAccumWithBuildin() 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, SUM(A.v), AVG(A.v), COUNT(A.v), MIN(A.v), MAX(A.v);");
+ pigServer.registerQuery("D = foreach C generate group, SUM(A.v), AVG(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,3.0,5.0,1.0,5.0});
@@ -427,4 +427,32 @@
}
}
}
+
+ public void testAccumWithMultiBuildin() throws IOException{
+ pigServer.registerQuery("A = load '" + INPUT_FILE + "' 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
+ 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);");
+
+ try {
+ Iterator<Tuple> iter = pigServer.openIterator("D");
+ } catch (Exception e) {
+ fail("COUNT_STAR should be supported by accumulator interface");
+ }
+ }
+
}
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestForEachNestedPlan.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestForEachNestedPlan.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestForEachNestedPlan.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestForEachNestedPlan.java Thu Dec 17 18:36:28 2009
@@ -18,18 +18,23 @@
package org.apache.pig.test;
+import org.apache.pig.ExecType;
import org.apache.pig.PigServer;
import org.apache.pig.ExecType;
import org.apache.pig.test.utils.TestHelper;
import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.logicalLayer.parser.ParseException;
import org.junit.Test;
import junit.framework.TestCase;
import junit.framework.Assert;
import java.util.Iterator;
+import java.util.List;
import java.util.Random;
import java.io.File;
+import java.io.FileWriter;
import java.io.IOException;
+import java.io.PrintWriter;
import java.text.DecimalFormat;
public class TestForEachNestedPlan extends TestCase {
@@ -47,7 +52,8 @@
@Test
public void testInnerOrderBy() throws Exception {
for (int i = 0; i < nullFlags.length; i++) {
- System.err.println("Running testInnerOrderBy with nullFlags set to :" + nullFlags[i]);
+ System.err.println("Running testInnerOrderBy with nullFlags set to :"
+ + nullFlags[i]);
File tmpFile = genDataSetFile1(nullFlags[i]);
pig.registerQuery("a = load '"
+ Util.generateURI(tmpFile.toString(), pig.getPigContext()) + "'; ");
@@ -66,29 +72,119 @@
}
}
-
- /*
@Test
- public void testInnerDistinct() throws Exception {
- File tmpFile = genDataSetFile1() ;
- pig.registerQuery("a = load 'file:" + tmpFile + "'; ") ;
- pig.registerQuery("b = group a by $0; ");
- pig.registerQuery("c = foreach b { "
- + " c1 = distinct $1 ; "
- + " generate flatten(c1); "
- + "};") ;
+ public void testInnerOrderByStarWithSchema() throws Exception {
+ File tmpFile = genDataSetFile1(false);
+ pig.registerQuery("a = load '" + Util.generateURI(tmpFile.toString(),
+ pig.getPigContext()) + "' as (a0, a1);");
+ pig.registerQuery("b = group a by a0; ");
+ pig.registerQuery("c = foreach b { d = order a by *; "
+ + " generate group, d; };");
Iterator<Tuple> it = pig.openIterator("c");
- Tuple t = null ;
- int count = 0 ;
- while(it.hasNext()) {
- t = it.next() ;
- System.out.println(count + ":" + t) ;
- count++ ;
+ Tuple t = null;
+ int count = 0;
+ while (it.hasNext()) {
+ t = it.next();
+ System.out.println(count + ":" + t);
+ count++;
}
- Assert.assertEquals(count, 15);
+ Assert.assertEquals(count, 10);
}
- */
+
+ @Test
+ public void testAlgebricFuncWithoutGroupBy()
+ throws IOException, ParseException {
+ String INPUT_FILE = "test-sum.txt";
+
+ PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
+ w.println("10\t{(1),(2),(3)}");
+ w.println("20\t{(4),(5),(6),(7)}");
+ w.println("30\t{(8),(9)}");
+ w.close();
+
+ try {
+
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
+
+ pig.registerQuery("a = load '" + INPUT_FILE + "' "
+ + "as (id:int, g:bag{t:tuple(u:int)});");
+ pig.registerQuery("b = foreach a generate id, SUM(g);") ;
+
+ Iterator<Tuple> iter = pig.openIterator("b");
+
+ List<Tuple> expectedResults =
+ Util.getTuplesFromConstantTupleStrings(
+ new String[] {
+ "(10,6L)",
+ "(20,22L)",
+ "(30,17L)"
+ });
+
+ int counter = 0;
+ while (iter.hasNext()) {
+ assertEquals(expectedResults.get(counter++).toString(),
+ iter.next().toString());
+ }
+
+ assertEquals(expectedResults.size(), counter);
+ } finally{
+ new File(INPUT_FILE).delete();
+ try {
+ Util.deleteFile(cluster, INPUT_FILE);
+ } catch (IOException e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testInnerDistinct()
+ throws IOException, ParseException {
+ String INPUT_FILE = "test-distinct.txt";
+
+ PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
+ w.println("10\t89");
+ w.println("20\t78");
+ w.println("10\t68");
+ w.println("10\t89");
+ w.println("20\t92");
+ w.close();
+
+ try {
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
+
+ pig.registerQuery("A = load '" + INPUT_FILE
+ + "' as (age:int, gpa:int);");
+ pig.registerQuery("B = group A by age;");
+ pig.registerQuery("C = foreach B { D = A.gpa; E = distinct D; " +
+ "generate group, MIN(E); };");
+
+ Iterator<Tuple> iter = pig.openIterator("C");
+
+ List<Tuple> expectedResults =
+ Util.getTuplesFromConstantTupleStrings(
+ new String[] {"(10,68)", "(20,78)"});
+
+ int counter = 0;
+ while (iter.hasNext()) {
+ assertEquals(expectedResults.get(counter++).toString(),
+ iter.next().toString());
+ }
+
+ assertEquals(expectedResults.size(), counter);
+ } finally{
+ new File(INPUT_FILE).delete();
+ try {
+ Util.deleteFile(cluster, INPUT_FILE);
+ } catch (IOException e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
/***
* For generating a sample dataset
*/
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestInputOutputFileValidator.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestInputOutputFileValidator.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestInputOutputFileValidator.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestInputOutputFileValidator.java Thu Dec 17 18:36:28 2009
@@ -22,6 +22,7 @@
import org.apache.pig.ExecType;
import org.apache.pig.FuncSpec;
+import org.apache.pig.backend.datastorage.DataStorage;
import org.apache.pig.backend.datastorage.ElementDescriptor;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.FileLocalizer;
@@ -52,7 +53,7 @@
String inputfile = generateTempFile().getAbsolutePath() ;
String outputfile = generateNonExistenceTempFile().getAbsolutePath() ;
- LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile) ;
+ LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile, ctx.getFs()) ;
CompilationMessageCollector collector = new CompilationMessageCollector() ;
LogicalPlanValidationExecutor executor = new LogicalPlanValidationExecutor(plan, ctx) ;
@@ -72,7 +73,7 @@
String inputfile = generateTempFile().getAbsolutePath() ;
String outputfile = generateTempFile().getAbsolutePath() ;
- LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile) ;
+ LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile, ctx.getDfs()) ;
CompilationMessageCollector collector = new CompilationMessageCollector() ;
LogicalPlanValidationExecutor executor = new LogicalPlanValidationExecutor(plan, ctx) ;
@@ -100,7 +101,7 @@
String inputfile = createHadoopTempFile(ctx) ;
String outputfile = createHadoopNonExistenceTempFile(ctx) ;
- LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile) ;
+ LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile, ctx.getDfs()) ;
CompilationMessageCollector collector = new CompilationMessageCollector() ;
LogicalPlanValidationExecutor executor = new LogicalPlanValidationExecutor(plan, ctx) ;
@@ -120,7 +121,7 @@
String inputfile = createHadoopTempFile(ctx) ;
String outputfile = createHadoopTempFile(ctx) ;
- LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile) ;
+ LogicalPlan plan = genNewLoadStorePlan(inputfile, outputfile, ctx.getDfs()) ;
CompilationMessageCollector collector = new CompilationMessageCollector() ;
LogicalPlanValidationExecutor executor = new LogicalPlanValidationExecutor(plan, ctx) ;
@@ -140,14 +141,14 @@
private LogicalPlan genNewLoadStorePlan(String inputFile,
- String outputFile)
+ String outputFile, DataStorage dfs)
throws Throwable {
LogicalPlan plan = new LogicalPlan() ;
FileSpec filespec1 =
new FileSpec(inputFile, new FuncSpec("org.apache.pig.builtin.PigStorage")) ;
FileSpec filespec2 =
new FileSpec(outputFile, new FuncSpec("org.apache.pig.builtin.PigStorage"));
- LOLoad load = new LOLoad(plan, genNewOperatorKeyId(), filespec1, null, null, true) ;
+ LOLoad load = new LOLoad(plan, genNewOperatorKeyId(), filespec1, null, dfs, true) ;
LOStore store = new LOStore(plan, genNewOperatorKeyId(), filespec2) ;
plan.add(load) ;
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestLogicalPlanBuilder.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestLogicalPlanBuilder.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestLogicalPlanBuilder.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestLogicalPlanBuilder.java Thu Dec 17 18:36:28 2009
@@ -948,6 +948,11 @@
}
@Test
+ public void testLimitWithLong() {
+ buildPlan("limit (load 'a') 100L;");
+ }
+
+ @Test
public void testQuery75() {
buildPlan("a = union (load 'a'), (load 'b'), (load 'c');");
buildPlan("b = foreach a {generate $0;} parallel 10;");
@@ -1684,8 +1689,10 @@
sort = (LOSort)foreachPlan.getPredecessors(foreachPlan.getLeaves().get(0)).get(0);
+ // project (*) operator here is translated to a list of projection
+ // operators
for(LogicalPlan sortPlan: sort.getSortColPlans()) {
- assertTrue(checkPlanForProjectStar(sortPlan) == true);
+ assertTrue(checkPlanForProjectStar(sortPlan) == false);
}
}
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMRCompiler.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMRCompiler.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMRCompiler.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMRCompiler.java Thu Dec 17 18:36:28 2009
@@ -100,6 +100,7 @@
// and are sure of
private boolean generate = false;
+ @Override
@Before
public void setUp() throws ExecException {
GenPhyOp.setR(r);
@@ -107,6 +108,7 @@
GenPhyOp.setPc(pc);
}
+ @Override
@After
public void tearDown() throws Exception {
}
@@ -947,7 +949,24 @@
}
}
+
+ @Test
+ public void testMergeJoinWithIndexableLoadFunc() throws Exception{
+ //generate = true;
+ planTester.buildPlan("a = load '/tmp/input1';");
+ planTester.buildPlan("b = load '/tmp/input2' using " +
+ TestMergeJoin.DummyIndexableLoader.class.getName() + ";");
+ planTester.buildPlan("c = join a by $0, b by $0 using \"merge\";");
+ LogicalPlan lp = planTester.buildPlan("store c into '/tmp';");
+
+ PhysicalPlan pp = Util.buildPhysicalPlan(lp, pc);
+ MROperPlan mp = Util.buildMRPlan(pp, pc);
+ assertEquals("Checking number of MR Jobs for merge join with " +
+ "IndexableLoadFunc:", 1, mp.size());
+
+ }
+
private void run(PhysicalPlan pp, String expectedFile) throws Exception {
String compiledPlan, goldenPlan = null;
int MAX_SIZE = 100000;
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMergeJoin.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMergeJoin.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMergeJoin.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMergeJoin.java Thu Dec 17 18:36:28 2009
@@ -42,6 +42,7 @@
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
+import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.LogicalPlan;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.util.LogUtils;
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMultiQuery.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMultiQuery.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMultiQuery.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestMultiQuery.java Thu Dec 17 18:36:28 2009
@@ -112,6 +112,222 @@
myPig = null;
}
+ public void testMultiQueryJiraPig1068() {
+
+ // test case: COGROUP fails with 'Type mismatch in key from map:
+ // expected org.apache.pig.impl.io.NullableText, recieved org.apache.pig.impl.io.NullableTuple'
+
+ String INPUT_FILE = "pig-1068.txt";
+
+ try {
+
+ PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
+ w.println("10\tapple\tlogin\tjar");
+ w.println("20\torange\tlogin\tbox");
+ w.println("30\tstrawberry\tquit\tbot");
+
+ w.close();
+
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
+
+ myPig.setBatchOn();
+
+ myPig.registerQuery("logs = load '" + INPUT_FILE
+ + "' as (ts:int, id:chararray, command:chararray, comments:chararray);");
+ myPig.registerQuery("SPLIT logs INTO logins IF command == 'login', all_quits IF command == 'quit';");
+ myPig.registerQuery("login_info = FOREACH logins { GENERATE id as id, comments AS client; };");
+ myPig.registerQuery("logins_grouped = GROUP login_info BY (id, client);");
+ myPig.registerQuery("count_logins_by_client = FOREACH logins_grouped "
+ + "{ generate group.id AS id, group.client AS client, COUNT($1) AS count; };");
+ myPig.registerQuery("all_quits_grouped = GROUP all_quits BY id; ");
+ myPig.registerQuery("quits = FOREACH all_quits_grouped { GENERATE FLATTEN(all_quits); };");
+ myPig.registerQuery("joined_session_info = COGROUP quits BY id, count_logins_by_client BY id;");
+
+ Iterator<Tuple> iter = myPig.openIterator("joined_session_info");
+
+ List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
+ new String[] {
+ "('apple',{},{('apple','jar',1L)})",
+ "('orange',{},{('orange','box',1L)})",
+ "('strawberry',{(30,'strawberry','quit','bot')},{})"
+ });
+
+ int counter = 0;
+ while (iter.hasNext()) {
+ assertEquals(expectedResults.get(counter++).toString(), iter.next().toString());
+ }
+
+ assertEquals(expectedResults.size(), counter);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ new File(INPUT_FILE).delete();
+ try {
+ Util.deleteFile(cluster, INPUT_FILE);
+ } catch (IOException e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testMultiQueryJiraPig1108() {
+
+ try {
+ myPig.setBatchOn();
+
+ myPig.registerQuery("a = load 'file:test/org/apache/pig/test/data/passwd' "
+ + "using PigStorage(':') as (uname:chararray, passwd:chararray, uid:int, gid:int);");
+ myPig.registerQuery("split a into plan1 if (uid > 5), plan2 if ( uid < 5);");
+ myPig.registerQuery("b = group plan1 by uname;");
+ myPig.registerQuery("c = foreach b { tmp = order plan1 by uid desc; "
+ + "generate flatten(group) as foo, tmp; };");
+ myPig.registerQuery("d = filter c BY foo is not null;");
+ myPig.registerQuery("store d into '/tmp/output1';");
+ myPig.registerQuery("store plan2 into '/tmp/output2';");
+
+ List<ExecJob> jobs = myPig.executeBatch();
+ for (ExecJob job : jobs) {
+ assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+
+ @Test
+ public void testMultiQueryJiraPig1114() {
+
+ // test case: MultiQuery optimization throws error when merging 2 level splits
+
+ String INPUT_FILE = "data.txt";
+
+ try {
+
+ PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE));
+ w.println("10\tjar");
+ w.println("20\tbox");
+ w.println("30\tbot");
+ w.close();
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE, INPUT_FILE);
+
+ myPig.setBatchOn();
+
+ myPig.registerQuery("data = load '" + INPUT_FILE
+ + "' USING PigStorage as (id:int, name:chararray);");
+ myPig.registerQuery("ids = FOREACH data GENERATE id;");
+ myPig.registerQuery("allId = GROUP ids all;");
+ myPig.registerQuery("allIdCount = FOREACH allId GENERATE group as allId, COUNT(ids) as total;");
+ myPig.registerQuery("idGroup = GROUP ids by id;");
+ myPig.registerQuery("idGroupCount = FOREACH idGroup GENERATE group as id, COUNT(ids) as count;");
+ myPig.registerQuery("countTotal = cross idGroupCount, allIdCount;");
+ myPig.registerQuery("idCountTotal = foreach countTotal generate id, count, total, (double)count / (double)total as proportion;");
+ myPig.registerQuery("orderedCounts = order idCountTotal by count desc;");
+ myPig.registerQuery("STORE orderedCounts INTO '/tmp/output1';");
+
+ myPig.registerQuery("names = FOREACH data GENERATE name;");
+ myPig.registerQuery("allNames = GROUP names all;");
+ myPig.registerQuery("allNamesCount = FOREACH allNames GENERATE group as namesAll, COUNT(names) as total;");
+ myPig.registerQuery("nameGroup = GROUP names by name;");
+ myPig.registerQuery("nameGroupCount = FOREACH nameGroup GENERATE group as name, COUNT(names) as count;");
+ myPig.registerQuery("namesCrossed = cross nameGroupCount, allNamesCount;");
+ myPig.registerQuery("nameCountTotal = foreach namesCrossed generate name, count, total, (double)count / (double)total as proportion;");
+ myPig.registerQuery("nameCountsOrdered = order nameCountTotal by count desc;");
+ myPig.registerQuery("STORE nameCountsOrdered INTO '/tmp/output2';");
+
+ List<ExecJob> jobs = myPig.executeBatch();
+ for (ExecJob job : jobs) {
+ assertTrue(job.getStatus() == ExecJob.JOB_STATUS.COMPLETED);
+ }
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ new File(INPUT_FILE).delete();
+ try {
+ Util.deleteFile(cluster, INPUT_FILE);
+ } catch (IOException e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
+ @Test
+ public void testMultiQueryJiraPig1113() {
+
+ // test case: Diamond query optimization throws error in JOIN
+
+ String INPUT_FILE_1 = "set1.txt";
+ String INPUT_FILE_2 = "set2.txt";
+ try {
+
+ PrintWriter w = new PrintWriter(new FileWriter(INPUT_FILE_1));
+ w.println("login\t0\tjar");
+ w.println("login\t1\tbox");
+ w.println("quit\t0\tmany");
+ w.close();
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE_1, INPUT_FILE_1);
+
+ PrintWriter w2 = new PrintWriter(new FileWriter(INPUT_FILE_2));
+ w2.println("apple\tlogin\t{(login)}");
+ w2.println("orange\tlogin\t{(login)}");
+ w2.println("strawberry\tquit\t{(login)}");
+ w2.close();
+ Util.copyFromLocalToCluster(cluster, INPUT_FILE_2, INPUT_FILE_2);
+
+ myPig.setBatchOn();
+
+ myPig.registerQuery("set1 = load '" + INPUT_FILE_1
+ + "' USING PigStorage as (a:chararray, b:chararray, c:chararray);");
+ myPig.registerQuery("set2 = load '" + INPUT_FILE_2
+ + "' USING PigStorage as (a: chararray, b:chararray, c:bag{});");
+ myPig.registerQuery("set2_1 = FOREACH set2 GENERATE a as f1, b as f2, "
+ + "(chararray) 0 as f3;");
+ myPig.registerQuery("set2_2 = FOREACH set2 GENERATE a as f1, "
+ + "FLATTEN((IsEmpty(c) ? null : c)) as f2, (chararray) 1 as f3;");
+ myPig.registerQuery("all_set2 = UNION set2_1, set2_2;");
+ myPig.registerQuery("joined_sets = JOIN set1 BY (a,b), all_set2 BY (f2,f3);");
+
+ List<Tuple> expectedResults = Util.getTuplesFromConstantTupleStrings(
+ new String[] {
+ "('quit','0','many','strawberry','quit','0')",
+ "('login','0','jar','apple','login','0')",
+ "('login','0','jar','orange','login','0')",
+ "('login','1','box','apple','login','1')",
+ "('login','1','box','orange','login','1')",
+ "('login','1','box','strawberry','login','1')"
+ });
+
+ Iterator<Tuple> iter = myPig.openIterator("joined_sets");
+ int count = 0;
+ while (iter.hasNext()) {
+ assertEquals(expectedResults.get(count++).toString(), iter.next().toString());
+ }
+ assertEquals(expectedResults.size(), count);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ Assert.fail();
+ } finally {
+ new File(INPUT_FILE_1).delete();
+ new File(INPUT_FILE_2).delete();
+ try {
+ Util.deleteFile(cluster, INPUT_FILE_1);
+ Util.deleteFile(cluster, INPUT_FILE_2);
+ } catch (IOException e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+ }
+
@Test
public void testMultiQueryJiraPig1060() {
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestOperatorPlan.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestOperatorPlan.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestOperatorPlan.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestOperatorPlan.java Thu Dec 17 18:36:28 2009
@@ -271,7 +271,7 @@
private int mNumChecks = 0;
AlwaysTransform(TPlan plan) {
- super(plan, new DepthFirstWalker<TOperator, TPlan>(plan));
+ super(plan);
}
public boolean check(List<TOperator> nodes) {
@@ -297,7 +297,7 @@
private int mNumChecks = 0;
NeverTransform(TPlan plan) {
- super(plan, new DepthFirstWalker<TOperator, TPlan>(plan));
+ super(plan);
}
public boolean check(List<TOperator> nodes) {
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPOCast.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPOCast.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPOCast.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPOCast.java Thu Dec 17 18:36:28 2009
@@ -43,6 +43,7 @@
import org.apache.pig.data.TupleFactory;
import org.apache.pig.impl.io.BufferedPositionedInputStream;
import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -913,14 +914,11 @@
}
+ @Override
public Tuple getNext() throws IOException {
return null;
}
- public void fieldsToRead(Schema schema) {
-
- }
-
public DataBag bytesToBag(byte[] b) throws IOException {
return null;
}
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestParamSubPreproc.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestParamSubPreproc.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestParamSubPreproc.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestParamSubPreproc.java Thu Dec 17 18:36:28 2009
@@ -167,7 +167,7 @@
ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
- pigExResultStream = new FileInputStream(basedir + "/ExpectedResult.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResultDefault.pig");
BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
@@ -259,7 +259,7 @@
ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
- pigExResultStream = new FileInputStream(basedir + "/ExpectedResult.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResult4.pig");
BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
@@ -310,7 +310,7 @@
ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
- pigExResultStream = new FileInputStream(basedir + "/ExpectedResult.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResult4.pig");
BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
@@ -359,7 +359,7 @@
ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
- pigExResultStream = new FileInputStream(basedir + "/ExpectedResult.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResultCmdLnPriorDeclare.pig");
BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
@@ -409,7 +409,7 @@
ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
- pigExResultStream = new FileInputStream(basedir + "/ExpectedResult.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResult4.pig");
BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
@@ -913,7 +913,7 @@
ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
- pigExResultStream = new FileInputStream(basedir + "/ExpectedResult.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResultMulDecs.pig");
BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
@@ -1336,4 +1336,51 @@
log.info("Done");
}
+
+ /* Test case 25
+ * Test that params in comments are untouched
+ */
+ @Test
+ public void testCommentWithParam() throws Exception{
+ try {
+ ParameterSubstitutionPreprocessor ps = new ParameterSubstitutionPreprocessor(50);
+ pigIStream = new BufferedReader(new FileReader(basedir + "/inputComment.pig"));
+ pigOStream = new FileWriter(basedir + "/output1.pig");
+
+ String[] arg = {"date='20080228'"};
+ String[] argFiles = null;
+ ps.genSubstitutedFile(pigIStream , pigOStream , arg , argFiles);
+
+ FileInputStream pigResultStream = new FileInputStream(basedir + "/output1.pig");
+ pigExResultStream = new FileInputStream(basedir + "/ExpectedResultComment.pig");
+ BufferedReader inExpected = new BufferedReader(new InputStreamReader(pigExResultStream));
+ BufferedReader inResult = new BufferedReader(new InputStreamReader(pigResultStream));
+
+ String exLine;
+ String resLine;
+ int lineNum=0;
+
+ while (true) {
+ lineNum++;
+ exLine = inExpected.readLine();
+ resLine = inResult.readLine();
+ if (exLine==null || resLine==null)
+ break;
+ assertEquals("Expected : "+exLine+" , but got : "+resLine+" in line num : "+lineNum ,exLine.trim(), resLine.trim());
+ }
+ if (!(exLine==null && resLine==null)) {
+ fail ("Expected : "+exLine+" , but got : "+resLine+" in line num : "+lineNum);
+ }
+
+ inExpected.close();
+ inResult.close();
+ } catch (ParseException e) {
+ fail ("Got ParseException : " + e.getMessage());
+ } catch (RuntimeException e) {
+ fail ("Got RuntimeException : " + e.getMessage());
+ } catch (Error e) {
+ fail ("Got error : " + e.getMessage());
+ }
+
+ }
}
Modified: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPigLineRecordReader.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPigLineRecordReader.java?rev=891828&r1=891827&r2=891828&view=diff
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPigLineRecordReader.java (original)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPigLineRecordReader.java Thu Dec 17 18:36:28 2009
@@ -614,4 +614,97 @@
fail(e.getMessage());
}
}
+
+ /**
+ * This tests check if PigLineRecordReader can read a file which has an empty line
+ */
+ @Test
+ public void testEmptyLineSimpleFile() {
+ try {
+ File testFile = File.createTempFile("testPigLineRecordReader", ".txt");
+ String text = "This is a text";
+
+ PrintStream ps = new PrintStream( testFile );
+ for( int i = 0; i < LOOP_COUNT; i++ ) {
+ ps.println( text );
+ // Add an empty line
+ ps.println("");
+ }
+ ps.close();
+
+ LocalSeekableInputStream is = new LocalSeekableInputStream( testFile );
+ BufferedPositionedInputStream bpis = new BufferedPositionedInputStream( is );
+ PigLineRecordReader reader = new PigLineRecordReader( bpis, 0, Integer.MAX_VALUE );
+
+ Text value = new Text();
+ int counter = 0;
+ while( reader.next(value) ) {
+ if( counter % 2 == 0 ) {
+ assertTrue( "Invalid Text", value.toString().compareTo(text) == 0 );
+ } else {
+ assertTrue( "Invalid Text", value.toString().compareTo("") == 0 );
+ }
+ counter++;
+ }
+ assertEquals("Invalid number of lines", counter, LOOP_COUNT*2 );
+ testFile.deleteOnExit();
+
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail( e.getMessage() );
+ } catch (SecurityException e) {
+ e.printStackTrace();
+ fail( e.getMessage() );
+ } catch (IllegalArgumentException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
+
+ /**
+ * This tests check if PigLineRecordReader can read a file which has an empty line
+ */
+ @Test
+ public void testEmptyLineBZFile() {
+ try {
+ File testFile = File.createTempFile("testPigLineRecordReader", ".txt.bz2");
+ String text = "This is a text";
+
+ PrintStream ps = new PrintStream( new CBZip2OutputStream( new FileOutputStream( testFile )) );
+ for( int i = 0; i < LOOP_COUNT; i++ ) {
+ ps.println( text );
+ // Add an empty line
+ ps.println("");
+ }
+ ps.close();
+
+ LocalSeekableInputStream is = new LocalSeekableInputStream( testFile );
+ CBZip2InputStream bzis = new CBZip2InputStream( is );
+ BufferedPositionedInputStream bpis = new BufferedPositionedInputStream( bzis );
+ PigLineRecordReader reader = new PigLineRecordReader( bpis, 0, Integer.MAX_VALUE );
+
+ Text value = new Text();
+ int counter = 0;
+ while( reader.next(value) ) {
+ if( counter % 2 == 0 ) {
+ assertTrue( "Invalid Text", value.toString().compareTo(text) == 0 );
+ } else {
+ assertTrue( "Invalid Text", value.toString().compareTo("") == 0 );
+ }
+ counter++;
+ }
+ assertEquals("Invalid number of lines", counter, LOOP_COUNT*2 );
+ testFile.deleteOnExit();
+
+ } catch (IOException e) {
+ e.printStackTrace();
+ fail( e.getMessage() );
+ } catch (SecurityException e) {
+ e.printStackTrace();
+ fail( e.getMessage() );
+ } catch (IllegalArgumentException e) {
+ e.printStackTrace();
+ fail(e.getMessage());
+ }
+ }
}
Added: hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPinOptions.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPinOptions.java?rev=891828&view=auto
==============================================================================
--- hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPinOptions.java (added)
+++ hadoop/pig/branches/load-store-redesign/test/org/apache/pig/test/TestPinOptions.java Thu Dec 17 18:36:28 2009
@@ -0,0 +1,99 @@
+/*
+ * 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 java.io.IOException;
+
+import org.apache.pig.PigServer;
+import org.apache.pig.impl.logicalLayer.LOCogroup;
+import org.apache.pig.impl.logicalLayer.LOJoin;
+import org.apache.pig.impl.logicalLayer.LogicalOperator;
+import org.apache.pig.impl.logicalLayer.LOJoin;
+import org.apache.pig.impl.logicalLayer.LogicalPlan;
+import org.apache.pig.test.utils.*;
+import org.junit.Before;
+import org.junit.Test;
+import static org.apache.pig.ExecType.LOCAL;
+
+import junit.framework.TestCase;
+
+public class TestPinOptions extends TestCase {
+
+ protected PigServer pigServer;
+ @Before
+ @Override
+ protected void setUp() throws Exception {
+ pigServer = new PigServer(LOCAL);
+ }
+
+ @Test
+ public void testPinnedJoinOption() throws IOException {
+ String[] joinTypes = {"hash", "repl", "merge", "skewed", "default"};
+ String[] expectedJoinTypes = {"HASH", "REPLICATED", "MERGE", "SKEWED", "HASH"};
+ for (int i = 0; i < joinTypes.length; i++) {
+ pigServer.setBatchOn();
+ pigServer.registerQuery("a = load '/tmp' as (foo, bar);");
+ pigServer.registerQuery("b = load '/tmp' as (foo, bar);");
+ pigServer.registerQuery("c = join a by foo, b by foo using \""+joinTypes[i]+"\";");
+ LogicalOperator op = getOpByAlias(pigServer.getAliases().get("c"), "c");
+ assertTrue("did "+joinTypes[i]+" join get pinned? ",
+ op.isPinnedOption(LOJoin.OPTION_JOIN));
+ assertEquals("did the right join type get set? ",
+ ((LOJoin) op).getJoinType().toString(), expectedJoinTypes[i]);
+ pigServer.discardBatch();
+ }
+ }
+
+ @Test
+ public void testNotPinnedJinOption() throws IOException {
+ pigServer.registerQuery("a = load '/tmp' as (foo, bar);");
+ pigServer.registerQuery("b = load '/tmp' as (foo, bar);");
+ pigServer.registerQuery("c = join a by foo, b by foo;");
+ LogicalOperator op = getOpByAlias(pigServer.getAliases().get("c"), "c");
+ assertEquals("default join should be hash",
+ ((LOJoin) op).getJoinType().toString(), "HASH");
+ assertFalse(op.isPinnedOption(LOJoin.OPTION_JOIN));
+ }
+
+ @Test
+ public void testGroupOptions() throws IOException {
+ pigServer.setBatchOn();
+ pigServer.registerQuery("a = load '/tmp' as (foo, bar);");
+ pigServer.registerQuery("b = group a by foo;");
+
+ LogicalOperator op = getOpByAlias(pigServer.getAliases().get("b"), "b");
+ assertFalse(op.isPinnedOption(LOCogroup.OPTION_GROUPTYPE));
+ pigServer.discardBatch();
+
+ pigServer.setBatchOn();
+ pigServer.registerQuery("a = load '/tmp' as (foo, bar);");
+ pigServer.registerQuery("b = group a by foo using \"collected\";");
+ op = getOpByAlias(pigServer.getAliases().get("b"), "b");
+ assertTrue(op.isPinnedOption(LOCogroup.OPTION_GROUPTYPE));
+ pigServer.discardBatch();
+ }
+
+ private LogicalOperator getOpByAlias(LogicalPlan lp, String alias) {
+ for (LogicalOperator op : lp) {
+ if (op.getAlias().equals(alias)) return op;
+ }
+ return null;
+ }
+
+}