From pig-commits-returnzcarey@example.org Sat Mar 01 04:14:13 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 27630 invoked from network); 1 Mar 2008 04:14:13 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Mar 2008 04:14:13 -0000 Received: (qmail 50810 invoked by uid 500); 1 Mar 2008 04:14:09 -0000 Delivered-To: zcarey@example.org Received: (qmail 50802 invoked by uid 500); 1 Mar 2008 04:14:09 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 50788 invoked by uid 99); 1 Mar 2008 04:14:09 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Fri, 29 Feb 2008 20:14:09 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Sat, 01 Mar 2008 04:13:29 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 3DC841A9832; Fri, 29 Feb 2008 20:13:47 -0800 (PST) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r632546 - in /incubator/pig/trunk: ./ src/org/apache/pig/backend/hadoop/executionengine/ src/org/apache/pig/backend/local/executionengine/ src/org/apache/pig/impl/eval/ src/org/apache/pig/impl/logicalLayer/ src/org/apache/pig/impl/physicalL... Date: Sat, 01 Mar 2008 04:13:46 -0000 To: zcarey@example.org From: zcarey@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <zcarey@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: gates Date: Fri Feb 29 20:13:45 2008 New Revision: 632546 URL: http://svn.apache.org/viewvc?rev=632546&view=rev Log: PIG-113 Make explain output more understandable. Added: incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecTreePrinter.java incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOTreePrinter.java incubator/pig/trunk/src/org/apache/pig/impl/physicalLayer/POTreePrinter.java Modified: incubator/pig/trunk/CHANGES.txt incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/MapRedPhysicalPlan.java incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java Modified: incubator/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/incubator/pig/trunk/CHANGES.txt?rev=632546&r1=632545&r2=632546&view=diff ============================================================================== --- incubator/pig/trunk/CHANGES.txt (original) +++ incubator/pig/trunk/CHANGES.txt Fri Feb 29 20:13:45 2008 @@ -147,3 +147,5 @@ gates). PIG-13: adding version to the system (joa23 via olgan) + + PIG-113: Make explain output more understandable (pi_song via gates) Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/MapRedPhysicalPlan.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/MapRedPhysicalPlan.java?rev=632546&r1=632545&r2=632546&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/MapRedPhysicalPlan.java (original) +++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/MapRedPhysicalPlan.java Fri Feb 29 20:13:45 2008 @@ -10,6 +10,7 @@ import org.apache.pig.backend.executionengine.ExecPhysicalPlan; import org.apache.pig.impl.physicalLayer.PhysicalOperator; import org.apache.pig.impl.physicalLayer.POPrinter; +import org.apache.pig.impl.physicalLayer.POTreePrinter; import org.apache.pig.impl.physicalLayer.POVisitor; import org.apache.pig.impl.logicalLayer.OperatorKey; @@ -36,8 +37,7 @@ } public void explain(OutputStream out) { - POVisitor lprinter = new POPrinter(opTable, new PrintStream(out)); - + POVisitor lprinter = new POTreePrinter(opTable, new PrintStream(out)); ((PhysicalOperator)opTable.get(root)).visit(lprinter); } Modified: incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java?rev=632546&r1=632545&r2=632546&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java (original) +++ incubator/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPhysicalPlan.java Fri Feb 29 20:13:45 2008 @@ -8,6 +8,7 @@ import org.apache.pig.backend.executionengine.ExecException; import org.apache.pig.backend.executionengine.ExecPhysicalOperator; import org.apache.pig.backend.executionengine.ExecPhysicalPlan; +import org.apache.pig.impl.physicalLayer.POTreePrinter; import org.apache.pig.impl.physicalLayer.PhysicalOperator; import org.apache.pig.impl.physicalLayer.POVisitor; import org.apache.pig.impl.physicalLayer.POPrinter; @@ -35,8 +36,7 @@ } public void explain(OutputStream out) { - POVisitor lprinter = new POPrinter(opTable, new PrintStream(out)); - + POVisitor lprinter = new POTreePrinter(opTable, new PrintStream(out)); ((PhysicalOperator)opTable.get(root)).visit(lprinter); } Added: incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecTreePrinter.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecTreePrinter.java?rev=632546&view=auto ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecTreePrinter.java (added) +++ incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecTreePrinter.java Fri Feb 29 20:13:45 2008 @@ -0,0 +1,124 @@ +package org.apache.pig.impl.eval; + +import java.io.PrintStream; +import java.util.Iterator; + +import org.apache.pig.impl.eval.cond.AndCond; +import org.apache.pig.impl.eval.cond.CompCond; +import org.apache.pig.impl.eval.cond.FalseCond; +import org.apache.pig.impl.eval.cond.FuncCond; +import org.apache.pig.impl.eval.cond.NotCond; +import org.apache.pig.impl.eval.cond.OrCond; +import org.apache.pig.impl.eval.cond.RegexpCond; +import org.apache.pig.impl.eval.cond.TrueCond; + +public class EvalSpecTreePrinter extends EvalSpecVisitor { + private PrintStream mStream = null; + + public EvalSpecTreePrinter(PrintStream ps) { + mStream = ps; + } + + @Override + public void visitFilter(FilterSpec f) { + mStream.print("Filter: "); + if (f.cond instanceof AndCond) mStream.print(" AND "); + else if (f.cond instanceof CompCond) mStream.print(" COMP "); + else if (f.cond instanceof FalseCond) mStream.print(" FALSE "); + else if (f.cond instanceof FuncCond) mStream.print(" FUNC "); + else if (f.cond instanceof NotCond) mStream.print(" NOT "); + else if (f.cond instanceof OrCond) mStream.print(" OR "); + else if (f.cond instanceof RegexpCond) mStream.print(" REGEXP "); + else if (f.cond instanceof TrueCond) mStream.print(" TRUE "); + else throw new AssertionError(" Unknown Cond "); + } + + @Override + public void visitSortDistinct(SortDistinctSpec sd) { + mStream.print("Sort(") ; + if (sd.distinct()) { + mStream.print("Distinct(") ; + } + sd.getSortSpec().visit(this); + if (sd.distinct()) { + mStream.print(")") ; + } + mStream.print(")") ; + } + + @Override + public void visitGenerate(GenerateSpec g) { + mStream.print("Generate("); + Iterator<EvalSpec> i = g.getSpecs().iterator(); + boolean isFirst = true ; + while (i.hasNext()) { + if (isFirst) { + isFirst = false ; + } + else { + mStream.print(",") ; + } + i.next().visit(this); + } + mStream.print(")") ; + } + + @Override + public void visitMapLookup(MapLookupSpec ml) { + mStream.print("MapLookup(key=" + ml.key() + ")"); + } + + @Override + public void visitConst(ConstSpec c) { + mStream.print("Const(" + c.value() + ")"); + } + + @Override + public void visitProject(ProjectSpec p) { + mStream.print("Project("); + Iterator<Integer> i = p.getCols().iterator(); + while (i.hasNext()) { + mStream.print(i.next().intValue()); + if (i.hasNext()) mStream.print(","); + } + mStream.print(")"); + } + + @Override + public void visitStar(StarSpec s) { + mStream.print("*"); + } + + @Override + public void visitFuncEval(FuncEvalSpec fe) { + mStream.print("FuncEval(" + fe.getFuncName() + "("); + fe.getArgs().visit(this); + mStream.print("))") ; + } + + @Override + public void visitCompositeEval(CompositeEvalSpec ce) { + mStream.print("Composite("); + Iterator<EvalSpec> i = ce.getSpecs().iterator(); + boolean isFirst = true ; + while (i.hasNext()) { + if (isFirst) { + isFirst = false ; + } + else { + mStream.print(",") ; + } + i.next().visit(this); + } + mStream.print(")") ; + } + + @Override + public void visitBinCond(BinCondSpec bc) { + mStream.print("BinCond(True=>"); + bc.ifTrue().visit(this); + mStream.print (",False=>"); + bc.ifFalse().visit(this); + mStream.print (")"); + } +} Added: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOTreePrinter.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOTreePrinter.java?rev=632546&view=auto ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOTreePrinter.java (added) +++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOTreePrinter.java Fri Feb 29 20:13:45 2008 @@ -0,0 +1,151 @@ +package org.apache.pig.impl.logicalLayer; + +import java.io.PrintStream; +import java.util.Iterator; +import java.util.List; +import org.apache.pig.impl.logicalLayer.schema.Schema; +import org.apache.pig.impl.eval.cond.Cond; + +import org.apache.pig.impl.eval.EvalSpec; +import org.apache.pig.impl.eval.EvalSpecPrinter; + +public class LOTreePrinter extends LOVisitor { + + private static final String LINE_START_SYMBOL = "|---" ; + private static final String IDENT_SYMBOL = " " ; + + private int currentIdent = 0 ; + + private PrintStream mStream = null; + + public LOTreePrinter(PrintStream ps) { + mStream = ps; + } + + public void increaseIdent() { + currentIdent++ ; + } + + public void decreaseIdent() { + currentIdent-- ; + if (currentIdent <0) + { + throw new RuntimeException("Invalid LOTreePrinter state. currentIdent < 0") ; + } + } + + @Override + public void visitCogroup(LOCogroup g) { + printLineHeader(g, false) ; + mStream.print(" ( " + g.arguments() + " ) ") ; + mStream.println() ; + depthFirstSearchVisit(g) ; + } + + @Override + public void visitEval(LOEval g) { + printLineHeader(g, false) ; + mStream.print(" ( " + g.arguments() + " ) ") ; + mStream.println() ; + depthFirstSearchVisit(g) ; + } + + @Override + public void visitUnion(LOUnion g) { + printLineHeader(g) ; + depthFirstSearchVisit(g) ; + } + + + @Override + public void visitLoad(LOLoad g) { + printLineHeader(g, false) ; + mStream.print(" ( file = " + g.getInputFileSpec().getFileName() ) ; + + if (g.outputSchema().getFields().size() > 0) + { + mStream.print(" AS ") ; + boolean isFirst = true ; + for(Schema schema: g.outputSchema().getFields() ) { + if (isFirst) { + isFirst = false ; + } + else { + mStream.print(",") ; + } + mStream.print(schema.getAlias()) ; + } + } + + mStream.println(" )") ; + depthFirstSearchVisit(g) ; + } + + @Override + public void visitSort(LOSort g) { + printLineHeader(g, false) ; + mStream.print(" ( BY " + g.arguments() + " ) ") ; + mStream.println() ; + depthFirstSearchVisit(g) ; + } + + @Override + public void visitSplit(LOSplit g) { + printLineHeader(g, false) ; + mStream.print(" ( ") ; + boolean isFirst = true ; + for(Cond cond:g.getConditions()) { + if (isFirst) { + isFirst = false ; + } + else { + mStream.print(",") ; + } + mStream.print(cond) ; + } + mStream.println(" ) ") ; + depthFirstSearchVisit(g) ; + } + + @Override + public void visitSplitOutput(LOSplitOutput g) { + printLineHeader(g, false) ; + mStream.print(" ( " + g.arguments() + " ) ") ; + mStream.println() ; + depthFirstSearchVisit(g) ; + } + + @Override + public void visitStore(LOStore g) { + printLineHeader(g) ; + depthFirstSearchVisit(g) ; + } + + private void printLineHeader(LogicalOperator g) { + printLineHeader(g, true) ; + } + + private void printLineHeader(LogicalOperator g, boolean appendNewLine) { + for(int i=0;i<currentIdent;i++) { + mStream.print(IDENT_SYMBOL) ; + } + mStream.print(LINE_START_SYMBOL) ; + mStream.print(g.getClass().getSimpleName()) ; + + if (appendNewLine) { + mStream.println(); + } + } + + private void depthFirstSearchVisit(LogicalOperator lo) { + List<OperatorKey> inputs = lo.getInputs(); + Iterator<OperatorKey> i = inputs.iterator(); + + this.increaseIdent() ; + while (i.hasNext()) { + LogicalOperator input = lo.getOpTable().get(i.next()); + input.visit(this); + } + this.decreaseIdent() ; + } +} Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java?rev=632546&r1=632545&r2=632546&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java (original) +++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/LogicalPlan.java Fri Feb 29 20:13:45 2008 @@ -103,7 +103,7 @@ } public void explain(OutputStream out) { - LOVisitor lprinter = new LOPrinter(new PrintStream(out)); + LOVisitor lprinter = new LOTreePrinter(new PrintStream(out)); opTable.get(root).visit(lprinter); } Added: incubator/pig/trunk/src/org/apache/pig/impl/physicalLayer/POTreePrinter.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/physicalLayer/POTreePrinter.java?rev=632546&view=auto ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/physicalLayer/POTreePrinter.java (added) +++ incubator/pig/trunk/src/org/apache/pig/impl/physicalLayer/POTreePrinter.java Fri Feb 29 20:13:45 2008 @@ -0,0 +1,191 @@ +package org.apache.pig.impl.physicalLayer; + +import java.io.PrintStream; +import java.util.Iterator; +import java.util.List; +import java.util.Map; + +import org.apache.pig.backend.executionengine.ExecPhysicalOperator; +import org.apache.pig.backend.hadoop.executionengine.POMapreduce; +import org.apache.pig.backend.local.executionengine.POCogroup; +import org.apache.pig.backend.local.executionengine.POEval; +import org.apache.pig.backend.local.executionengine.POLoad; +import org.apache.pig.backend.local.executionengine.POSort; +import org.apache.pig.backend.local.executionengine.POSplit; +import org.apache.pig.backend.local.executionengine.POStore; +import org.apache.pig.backend.local.executionengine.POUnion; +import org.apache.pig.impl.eval.EvalSpec; +import org.apache.pig.impl.eval.EvalSpecTreePrinter; +import org.apache.pig.impl.io.FileSpec; +import org.apache.pig.impl.logicalLayer.OperatorKey; + +public class POTreePrinter extends POVisitor { + + private static final String HEADER_START_SYMBOL = "|---" ; + private static final String CONTENT_START_SYMBOL = " " ; + private static final String IDENT_SYMBOL = " " ; + + private int currentIdent = 0 ; + + private PrintStream mStream = null; + + public POTreePrinter(Map<OperatorKey, ExecPhysicalOperator> opTable, + PrintStream ps) { + super(opTable); + mStream = ps; + } + + public void increaseIdent() { + currentIdent++ ; + } + + public void decreaseIdent() { + currentIdent-- ; + if (currentIdent <0) + { + throw new RuntimeException("Invalid POTreePrinter state. currentIdent < 0") ; + } + } + + @Override + public void visitMapreduce(POMapreduce p) { + printLineHeader(p) ; + + // partition function + if (p.partitionFunction != null) { + adjustContentIdent() ; + mStream.println("Partition Function: " + p.partitionFunction.getName()); + mStream.println() ; + } + + // map line + adjustContentIdent() ; + mStream.print("Map : ") ; + visitSpecs(p.toMap); + mStream.println() ; + + // combine line + if (p.toCombine != null) { + adjustContentIdent() ; + mStream.print("Combine : ") ; + p.toCombine.visit(new EvalSpecTreePrinter(mStream)); + mStream.println() ; + } + + // reduce line + if (p.toReduce != null) { + adjustContentIdent() ; + mStream.print("Reduce : ") ; + p.toReduce.visit(new EvalSpecTreePrinter(mStream)); + mStream.println() ; + } + + // grouping line + if (p.groupFuncs != null) { + adjustContentIdent() ; + mStream.print("Grouping : ") ; + visitSpecs(p.groupFuncs); + mStream.println() ; + } + + // input files line + adjustContentIdent() ; + mStream.print("Input File(s) : ") ; + Iterator<FileSpec> i = p.inputFileSpecs.iterator(); + while (i.hasNext()) { + mStream.print(i.next().getFileName()); + if (i.hasNext()) mStream.print(", "); + } + mStream.println() ; + + depthFirstSearchVisit(p) ; + } + + @Override + public void visitLoad(POLoad p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + @Override + public void visitSort(POSort p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + @Override + public void visitStore(POStore p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + @Override + public void visitCogroup(POCogroup p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + @Override + public void visitEval(POEval p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + @Override + public void visitSplit(POSplit p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + @Override + public void visitUnion(POUnion p) { + printLineHeader(p) ; + depthFirstSearchVisit(p) ; + } + + private void printLineHeader(PhysicalOperator g) { + printLineHeader(g, true) ; + } + + private void printLineHeader(PhysicalOperator g, boolean appendNewLine) { + for(int i=0;i<currentIdent;i++) { + mStream.print(IDENT_SYMBOL) ; + } + mStream.print(HEADER_START_SYMBOL) ; + mStream.print(g.getClass().getSimpleName()) ; + + if (appendNewLine) { + mStream.println(); + } + } + + private void adjustContentIdent() { + for(int i=0;i<currentIdent;i++) { + mStream.print(IDENT_SYMBOL) ; + } + mStream.print(CONTENT_START_SYMBOL) ; + } + + private void depthFirstSearchVisit(PhysicalOperator po) { + this.increaseIdent() ; + for(OperatorKey inputKey : po.inputs) { + PhysicalOperator input = null ; + // since the only sub-type of ExecPhysicalOperator is PhysicalOperator + // it is legal to convert this way + input = (PhysicalOperator) mOpTable.get(inputKey) ; + if (input==null) { + throw new RuntimeException("Invalid OpKey table found while reading in POTreePrinter") ; + } + input.visit(this); + } + this.decreaseIdent() ; + } + + private void visitSpecs(List<EvalSpec> specs) { + Iterator<EvalSpec> j = specs.iterator(); + while (j.hasNext()) { + j.next().visit(new EvalSpecTreePrinter(mStream)); + } + } + +} From pig-commits-returnzcarey@example.org Tue Mar 04 01:30:40 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 62790 invoked from network); 4 Mar 2008 01:30:40 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Mar 2008 01:30:40 -0000 Received: (qmail 35298 invoked by uid 500); 4 Mar 2008 01:30:36 -0000 Delivered-To: zcarey@example.org Received: (qmail 35276 invoked by uid 500); 4 Mar 2008 01:30:35 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 35267 invoked by uid 99); 4 Mar 2008 01:30:35 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 03 Mar 2008 17:30:35 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 04 Mar 2008 01:30:08 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 2BF5CD2EE for <zcarey@example.org>; Tue, 4 Mar 2008 01:30:19 +0000 (GMT) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: Apache Wiki <zcarey@example.org> To: zcarey@example.org Date: Tue, 04 Mar 2008 01:30:19 -0000 Message-ID: <zcarey@example.org> Subject: [Pig Wiki] Update of "PigDeveloperCookbook" by OlgaN X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for change notification. The following page has been changed by OlgaN: http://wiki.apache.org/pig/PigDeveloperCookbook ------------------------------------------------------------------------------ * For `User Errors`, throw `ParseException` that contains meaningful message. For batch and interactive processing, catch the exception in Grunt and log the exception message to stderr. The same can be done with the developer in the embedded case. In debug mode, log the exception stack into the client side log. * For `Internal Errors`, throw `RuntimeException` or its derivation. Catch the exception in main, log it, including the stack, to the client side log. Write failure message to stderr pointing to the log file. - * For `Frontend Errors`, throw newly created `FrontendException`. Catch the exception in main, log it, including the stack, to the client side log. Write failure message to stderr pointing to the log file. In the future, we might find it useful to further subdivide this errors in which case we might subclass the `FrontendException` class. + * For `Frontend Errors`, throw newly created `FrontendException` that is a subclass of RuntimeException. Catch the exception in main, log it, including the stack, to the client side log. Write failure message to stderr pointing to the log file. In the future, we might find it useful to further subdivide this errors in which case we might subclass the `FrontendException` class. * For `Backend Errors` there will need to be backend specific way to get the error from the backend to the frontend. Once this is done, log the error into client side log file and throw `ExecuteException`. Catch the exception in main, log it, including the stack, to the client side log. Write failure message to stderr pointing to the log file. From pig-commits-returnzcarey@example.org Tue Mar 04 07:15:03 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 77612 invoked from network); 4 Mar 2008 07:15:03 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Mar 2008 07:15:03 -0000 Received: (qmail 94594 invoked by uid 500); 4 Mar 2008 07:14:59 -0000 Delivered-To: zcarey@example.org Received: (qmail 94574 invoked by uid 500); 4 Mar 2008 07:14:59 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 94565 invoked by uid 99); 4 Mar 2008 07:14:58 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 03 Mar 2008 23:14:58 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 04 Mar 2008 07:14:31 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id F056CD2EE for <zcarey@example.org>; Tue, 4 Mar 2008 07:14:41 +0000 (GMT) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: Apache Wiki <zcarey@example.org> To: zcarey@example.org Date: Tue, 04 Mar 2008 07:14:41 -0000 Message-ID: <zcarey@example.org> Subject: [Pig Wiki] Update of "PigStreamingDesign" by Arun C Murthy X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Pig Wiki" for change notification. The following page has been changed by Arun C Murthy: http://wiki.apache.org/pig/PigStreamingDesign ------------------------------------------------------------------------------ == 4. Streaming Implementation == - The main infrastructure to support the notion of data processing by sending dataset to a task's input and collecting its output is a generic manager who takes care of setup/teardown of the streaming task, manages it's stdin/stderr/stdout streams and also does post-processing. The plan is to implement a {{{org.apache.hadoop.mapred.lib.external.ExecutableManager}}} to take over the aforementioned responsibilities. The decision to keep that separate from Hadoop's Streaming component (in contrib/streaming) to ensure that Pig has no extraneous dependency on Hadoop streaming and am putting it into org.apache.hadoop.mapred.lib to ensure Pig depends on Hadoop Core only. + The main infrastructure to support the notion of data processing by sending dataset to a task's input and collecting its output is a generic manager who takes care of setup/teardown of the streaming task, manages it's stdin/stderr/stdout streams and also does post-processing. The plan is to implement a {{{org.apache.pig.backend.streaming.PigExecutableManager}}} to take over the aforementioned responsibilities. The decision to keep that separate from Hadoop's Streaming component (in contrib/streaming) to ensure that Pig has no extraneous dependency on Hadoop streaming. The `ExecutableManager` also is responsible for dealing with multiple outputs of the streaming tasks (refer to the functional spec in the wiki). New: - `org.apache.hadoop.mapred.lib.external.ExecutableManager` + `org.apache.pig.backend.streaming.PigExecutableManager` {{{ - class org.apache.hadoop.mapred.lib.external.ExecutableManager { + class PigExecutableManager { + + // Configure the executable-manager void configure() throws IOException; + + // Runtime void run() throws IOException; + + // Clean-up hook (for e.g. multiple outputs' handling etc.) void close() throws IOException; - void next(WritableComparable key, Writable value); + + // Send the Datum to the executable + void add(Datum d); } }}} - The important deviation from current Pig infrastructure is that there isn't a one-to-one mapping between inputs and output records anymore since the user-script could (potentially) consume ''all'' the input before it emits ''any'' output records. The way to get around this is to wrap the {{{DataCollector}}} and hence the next successor in the pipleline in an {{{OutputCollector}}} and pass it along to the {{{ExecutableManager}}}. + The important deviation from current Pig infrastructure is that there isn't a one-to-one mapping between inputs and output records anymore since the user-script could (potentially) consume ''all'' the input before it emits ''any'' output records. The way to get around this is to wrap the {{{DataCollector}}} and hence the next successor in the pipeline in an {{{OutputCollector}}} and pass it along to the {{{PigExecutableManager}}}. From pig-commits-returnzcarey@example.org Tue Mar 04 21:21:19 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 73456 invoked from network); 4 Mar 2008 21:21:19 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 4 Mar 2008 21:21:19 -0000 Received: (qmail 68625 invoked by uid 500); 4 Mar 2008 21:21:15 -0000 Delivered-To: zcarey@example.org Received: (qmail 68615 invoked by uid 500); 4 Mar 2008 21:21:14 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 68606 invoked by uid 99); 4 Mar 2008 21:21:14 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 04 Mar 2008 13:21:14 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 04 Mar 2008 21:20:35 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 64C5C1A983A; Tue, 4 Mar 2008 13:20:55 -0800 (PST) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r633652 - in /incubator/pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Date: Tue, 04 Mar 2008 21:20:55 -0000 To: zcarey@example.org From: zcarey@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <zcarey@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: gates Date: Tue Mar 4 13:20:52 2008 New Revision: 633652 URL: http://svn.apache.org/viewvc?rev=633652&view=rev Log: PIG-120: Support map reduce in local mode. To do this user needs to specify execution type as mapreduce and cluster name as local. Modified: incubator/pig/trunk/CHANGES.txt incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Modified: incubator/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/incubator/pig/trunk/CHANGES.txt?rev=633652&r1=633651&r2=633652&view=diff ============================================================================== --- incubator/pig/trunk/CHANGES.txt (original) +++ incubator/pig/trunk/CHANGES.txt Tue Mar 4 13:20:52 2008 @@ -149,3 +149,7 @@ PIG-13: adding version to the system (joa23 via olgan) PIG-113: Make explain output more understandable (pi_song via gates) + + PIG-120: Support map reduce in local mode. To do this user needs to + specify execution type as mapreduce and cluster name as local (joa23 via + gates). Modified: incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java?rev=633652&r1=633651&r2=633652&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java (original) +++ incubator/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java Tue Mar 4 13:20:52 2008 @@ -110,6 +110,8 @@ setSSHFactory(); String hodServer = System.getProperty("hod.server"); + String cluster = System.getProperty("cluster"); + String nameNode = System.getProperty("namenode"); if (hodServer != null && hodServer.length() > 0) { String hdfsAndMapred[] = doHod(hodServer); @@ -117,17 +119,15 @@ setJobtrackerLocation(hdfsAndMapred[1]); } else { - String cluster = System.getProperty("cluster"); if (cluster != null && cluster.length() > 0) { - if(cluster.indexOf(':') < 0) { + if(cluster.indexOf(':') < 0 && !cluster.equalsIgnoreCase("local")) { cluster = cluster + ":50020"; } setJobtrackerLocation(cluster); } - String nameNode = System.getProperty("namenode"); if (nameNode!=null && nameNode.length() > 0) { - if(nameNode.indexOf(':') < 0) { + if(nameNode.indexOf(':') < 0 && !nameNode.equalsIgnoreCase("local")) { nameNode = nameNode + ":8020"; } setFilesystemLocation(nameNode); @@ -143,16 +143,18 @@ throw new ExecException("Failed to create DataStorage", e); } - log.info("Connecting to map-reduce job tracker at: " + conf.get("mapred.job.tracker")); - - try { - jobTracker = (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class, - JobSubmissionProtocol.versionID, - JobTracker.getAddress(conf.getConfiguration()), - conf.getConfiguration()); - } - catch (IOException e) { - throw new ExecException("Failed to crate job tracker", e); + if(cluster != null && !cluster.equalsIgnoreCase("local")){ + log.info("Connecting to map-reduce job tracker at: " + conf.get("mapred.job.tracker")); + + try { + jobTracker = (JobSubmissionProtocol) RPC.getProxy(JobSubmissionProtocol.class, + JobSubmissionProtocol.versionID, + JobTracker.getAddress(conf.getConfiguration()), + conf.getConfiguration()); + } + catch (IOException e) { + throw new ExecException("Failed to crate job tracker", e); + } } try { From pig-commits-returnzcarey@example.org Wed Mar 05 02:02:25 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 4398 invoked from network); 5 Mar 2008 02:02:25 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 5 Mar 2008 02:02:25 -0000 Received: (qmail 64226 invoked by uid 500); 5 Mar 2008 02:02:21 -0000 Delivered-To: zcarey@example.org Received: (qmail 64209 invoked by uid 500); 5 Mar 2008 02:02:21 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 64200 invoked by uid 99); 5 Mar 2008 02:02:21 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Tue, 04 Mar 2008 18:02:21 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 05 Mar 2008 02:01:51 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 92FDB1A9832; Tue, 4 Mar 2008 18:02:01 -0800 (PST) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r633746 - in /incubator/pig/trunk: CHANGES.txt build.xml src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt Date: Wed, 05 Mar 2008 02:02:01 -0000 To: zcarey@example.org From: zcarey@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <zcarey@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: olga Date: Tue Mar 4 18:02:00 2008 New Revision: 633746 URL: http://svn.apache.org/viewvc?rev=633746&view=rev Log: PIG-13 and PIG-118 Modified: incubator/pig/trunk/CHANGES.txt incubator/pig/trunk/build.xml incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt Modified: incubator/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/incubator/pig/trunk/CHANGES.txt?rev=633746&r1=633745&r2=633746&view=diff ============================================================================== --- incubator/pig/trunk/CHANGES.txt (original) +++ incubator/pig/trunk/CHANGES.txt Tue Mar 4 18:02:00 2008 @@ -148,6 +148,10 @@ PIG-13: adding version to the system (joa23 via olgan) + PIG-13: make the code use svn only if available (joa23 via olgan) + + PIG-118: make sure union/join/cross takes 2 params (pi_song vi olgan) + PIG-113: Make explain output more understandable (pi_song via gates) PIG-120: Support map reduce in local mode. To do this user needs to Modified: incubator/pig/trunk/build.xml URL: http://svn.apache.org/viewvc/incubator/pig/trunk/build.xml?rev=633746&r1=633745&r2=633746&view=diff ============================================================================== --- incubator/pig/trunk/build.xml (original) +++ incubator/pig/trunk/build.xml Tue Mar 4 18:02:00 2008 @@ -94,13 +94,12 @@ <svnversion outputproperty="svn.revision"/> </target> - <macrodef name="svnversion"> <!-- the path needs to be small content otherwise it will take AGES ! --> <attribute name="wcpath" default="${basedir}" /> <attribute name="outputproperty" /> <sequential> - <exec executable="svnversion" outputproperty="@{outputproperty}" failonerror="true" failifexecutionfails="false"> + <exec executable="svnversion" outputproperty="@{outputproperty}" failonerror="false" failifexecutionfails="false" > <arg value="@{wcpath}" /> <redirector> <outputfilterchain> @@ -111,7 +110,6 @@ </outputfilterchain> </redirector> </exec> - <fail unless="@{outputproperty}" message="Failed to retrieve svn revision. You need 'svnversion' in your path and connectivity to the repository" /> </sequential> </macrodef> @@ -189,6 +187,24 @@ <!-- TODO we should also exculte test here... --> <!-- ================================================================== --> <target name="jar" depends="compile" description="Create pig jar"> + <antcall target="jarWithSvn"/> + <antcall target="jarWithOutSvn"/> + </target> + + <target name="jarWithSvn" if="svn.revision"> + <antcall target="buildJar"> + <param name="svnString" value="${svn.revision}" /> + </antcall> + </target> + + <target name="jarWithOutSvn" unless="svn.revision"> + <antcall target="buildJar"> + <param name="svnString" value=": unknown" /> + </antcall> + </target> + + <target name="buildJar"> + <echo>svnString ${svnString}</echo> <jar jarfile="${output.jarfile.core}" basedir="${build.classes}"> <manifest> <attribute name="Main-Class" value="org.apache.pig.Main" /> @@ -197,7 +213,7 @@ <attribute name="Implementation-Title" value="Pig" /> <attribute name="Implementation-Version" value="${version}" /> <attribute name="Build-TimeStamp" value="${timestamp}" /> - <attribute name="Svn-Revision" value="${svn.revision}" /> + <attribute name="Svn-Revision" value="${svnString}" /> </section> </manifest> </jar> @@ -210,7 +226,7 @@ <attribute name="Implementation-Title" value="Pig" /> <attribute name="Implementation-Version" value="${version}" /> <attribute name="Build-TimeStamp" value="${timestamp}" /> - <attribute name="Svn-Revision" value="${svn.revision}" /> + <attribute name="Svn-Revision" value="${svnString}" /> </section> </manifest> <zipfileset src="${lib.dir}/junit-4.1.jar" /> Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt?rev=633746&r1=633745&r2=633746&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt (original) +++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt Tue Mar 4 18:02:00 2008 @@ -605,7 +605,7 @@ { ( op = NestedExpr() { inputs.add(op.getOperatorKey()); } - ("," op = NestedExpr() { inputs.add(op.getOperatorKey()); })* + ("," op = NestedExpr() { inputs.add(op.getOperatorKey()); })+ ) {return rewriteCross(inputs);} } @@ -613,14 +613,14 @@ LogicalOperator JoinClause() : {CogroupInput gi; ArrayList<CogroupInput> gis = new ArrayList<CogroupInput>();} { (gi = GroupItem() { gis.add(gi); } - ("," gi = GroupItem() { gis.add(gi); })*) + ("," gi = GroupItem() { gis.add(gi); })+) {return rewriteJoin(gis);} } LogicalOperator UnionClause() : {LogicalOperator op; ArrayList<OperatorKey> inputs = new ArrayList<OperatorKey>();} { (op = NestedExpr() { inputs.add(op.getOperatorKey()); } - ("," op = NestedExpr() { inputs.add(op.getOperatorKey()); })*) + ("," op = NestedExpr() { inputs.add(op.getOperatorKey()); })+) {return new LOUnion(opTable, scope, getNextId(), inputs);} } From pig-commits-returnzcarey@example.org Thu Mar 06 19:11:50 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 95705 invoked from network); 6 Mar 2008 19:11:50 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 6 Mar 2008 19:11:50 -0000 Received: (qmail 33616 invoked by uid 500); 6 Mar 2008 19:11:46 -0000 Delivered-To: zcarey@example.org Received: (qmail 33578 invoked by uid 500); 6 Mar 2008 19:11:46 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 33569 invoked by uid 99); 6 Mar 2008 19:11:46 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 06 Mar 2008 11:11:46 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 06 Mar 2008 19:11:18 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 9A1421A9832; Thu, 6 Mar 2008 11:11:28 -0800 (PST) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r634380 - in /incubator/pig/trunk: ./ src/org/apache/pig/impl/eval/ src/org/apache/pig/impl/logicalLayer/parser/ src/org/apache/pig/impl/streaming/ test/org/apache/pig/test/ Date: Thu, 06 Mar 2008 19:11:27 -0000 To: zcarey@example.org From: zcarey@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <zcarey@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: olga Date: Thu Mar 6 11:11:26 2008 New Revision: 634380 URL: http://svn.apache.org/viewvc?rev=634380&view=rev Log: PIG-94: M1 for streaming: maps and reduce side support with default (de)serializer (acmurthy via olgan) Added: incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java incubator/pig/trunk/src/org/apache/pig/impl/streaming/ incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java incubator/pig/trunk/test/org/apache/pig/test/TestStreaming.java Modified: incubator/pig/trunk/CHANGES.txt incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecVisitor.java incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt incubator/pig/trunk/test/org/apache/pig/test/Util.java Modified: incubator/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/incubator/pig/trunk/CHANGES.txt?rev=634380&r1=634379&r2=634380&view=diff ============================================================================== --- incubator/pig/trunk/CHANGES.txt (original) +++ incubator/pig/trunk/CHANGES.txt Thu Mar 6 11:11:26 2008 @@ -157,3 +157,6 @@ PIG-120: Support map reduce in local mode. To do this user needs to specify execution type as mapreduce and cluster name as local (joa23 via gates). + + PIG-94: M1 for streaming: maps and reduce side support with default + (de)serializer (acmurthy via olgan) Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecVisitor.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecVisitor.java?rev=634380&r1=634379&r2=634380&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecVisitor.java (original) +++ incubator/pig/trunk/src/org/apache/pig/impl/eval/EvalSpecVisitor.java Thu Mar 6 11:11:26 2008 @@ -143,7 +143,12 @@ public void visitTupleWindow(TupleWindowSpec tw) { } - + /** + * Only StreamSpec.visit() and subclass implementations of this function + * should ever call this method. + */ + public void visitStream(StreamSpec stream) { + } } Added: incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java?rev=634380&view=auto ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java (added) +++ incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java Thu Mar 6 11:11:26 2008 @@ -0,0 +1,139 @@ +/* + * 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.impl.eval; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.pig.LoadFunc; +import org.apache.pig.StoreFunc; +import org.apache.pig.backend.streaming.PigExecutableManager; +import org.apache.pig.builtin.PigStorage; +import org.apache.pig.data.Datum; +import org.apache.pig.impl.PigContext; +import org.apache.pig.impl.eval.collector.DataCollector; +import org.apache.pig.impl.logicalLayer.schema.Schema; + +public class StreamSpec extends EvalSpec { + private static final long serialVersionUID = 1L; + + private static final Log LOG = + LogFactory.getLog(StreamSpec.class.getName()); + + private String streamingCommand; // Actual command to be run + private String deserializer; // LoadFunc to be used + private String serializer; // StoreFunc to be used + + public StreamSpec(String streamingCommand) { + this.streamingCommand = streamingCommand; + this.deserializer = PigStorage.class.getName(); + this.serializer = PigStorage.class.getName(); + } + + public StreamSpec(String streamingCommand, + String deserializer, String serializer) { + this.streamingCommand = streamingCommand; + this.deserializer = deserializer; + this.serializer = serializer; + } + + @Override + public List<String> getFuncs() { + // No user-defined functions here + return new ArrayList<String>(); + } + + protected Schema mapInputSchema(Schema schema) { + // EvalSpec _has_ to have the schema if there is one... + return null; + } + + protected DataCollector setupDefaultPipe(DataCollector endOfPipe) { + return new StreamDataCollector(streamingCommand, + (deserializer == null) ? new PigStorage() : + (LoadFunc)PigContext.instantiateFuncFromSpec( + deserializer), + (serializer == null) ? new PigStorage() : + (StoreFunc)PigContext.instantiateFuncFromSpec( + serializer), + endOfPipe); + } + + public void visit(EvalSpecVisitor v) { + v.visitStream(this); + } + + /** + * A simple {@link DataCollector} which wraps a {@link PigExecutableManager} + * and lets it handle the input and the output to the managed executable. + */ + private static class StreamDataCollector extends DataCollector { + PigExecutableManager executable; //Executable manager + + public StreamDataCollector(String streamingCommand, + LoadFunc deserializer, StoreFunc serializer, + DataCollector endOfPipe) { + super(endOfPipe); + + DataCollector successor = + new DataCollector(endOfPipe) { + public void add(Datum d) { + // Just forward the data to the next EvalSpec in the pipeline + addToSuccessor(d); + } + }; + + try { + // Create the PigExecutableManager + executable = new PigExecutableManager(streamingCommand, + deserializer, serializer, + successor); + + executable.configure(); + + // Start the executable + executable.run(); + } catch (Exception e) { + LOG.fatal("Failed to create/start PigExecutableManager with: " + e); + throw new RuntimeException(e); + } + } + + public void add(Datum d) { + try { + executable.add(d); + } catch (IOException ioe) { + LOG.fatal("executable.add(" + d + ") failed with: " + ioe); + throw new RuntimeException(ioe); + } + } + + protected void finish() { + try { + executable.close(); + } + catch (Exception e) { + LOG.fatal("Failed to close PigExecutableManager with: " + e); + throw new RuntimeException(e); + } + } + } +} Modified: incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt?rev=634380&r1=634379&r2=634380&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt (original) +++ incubator/pig/trunk/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt Thu Mar 6 11:11:26 2008 @@ -298,6 +298,9 @@ TOKEN : { <GENERATE : "generate"> } TOKEN : { <FLATTEN : "flatten"> } TOKEN : { <EVAL : "eval"> } +TOKEN : { <STREAM : "stream"> } +TOKEN : { <THROUGH : "through"> } +TOKEN : { <BACKTICK : "`"> } TOKEN: { @@ -316,6 +319,7 @@ } TOKEN : { <QUOTEDSTRING : "'" (~["'"])* "'"> } +TOKEN : { <EXECCOMMAND : "`" (~["`"])* "`"> } // Pig has special variables starting with $ TOKEN : { <DOLLARVAR : "$" <INTEGER> > } @@ -411,6 +415,7 @@ | (<JOIN> op = JoinClause()) | (<UNION> op = UnionClause()) | (<FOREACH> op = ForEachClause()) +| (<STREAM> op = StreamClause() [<AS> schema = SchemaTuple() {op.setSchema(schema);} ]) ) [<PARALLEL> t2=<NUMBER> { op.setRequestedParallelism(Integer.parseInt(t2.image));} ] ) @@ -1126,3 +1131,22 @@ return item; } } + +LogicalOperator StreamClause(): {LogicalOperator input; String streamingCommand;} +{ + input = NestedExpr() + + <THROUGH> streamingCommand = StreamingCommand() + { + return new LOEval(opTable, scope, getNextId(), input.getOperatorKey(), new StreamSpec(streamingCommand)); + } +} + +String StreamingCommand(): {Token t;} +{ + t = <EXECCOMMAND> + { + return unquote(t.image); + } +} + Added: incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java?rev=634380&view=auto ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java (added) +++ incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java Thu Mar 6 11:11:26 2008 @@ -0,0 +1,297 @@ +/* + * 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.backend.streaming; + +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.BufferedReader; +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.util.ArrayList; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.pig.LoadFunc; +import org.apache.pig.StoreFunc; +import org.apache.pig.backend.executionengine.ExecException; +import org.apache.pig.data.Datum; +import org.apache.pig.data.Tuple; +import org.apache.pig.impl.eval.collector.DataCollector; +import org.apache.pig.impl.io.BufferedPositionedInputStream; +import org.apache.pig.impl.logicalLayer.parser.ParseException; + +/** + * {@link PigExecutableManager} manages an external executable which processes data + * in a Pig query. + * + * The <code>PigExecutableManager</code> is responsible for startup/teardown of the + * external process and also for managing it. + * It feeds input records to the executable via it's <code>stdin</code>, + * collects the output records from the <code>stdout</code> and also diagnostic + * information from the <code>stdout</code>. + */ +public class PigExecutableManager { + private static final Log LOG = + LogFactory.getLog(PigExecutableManager.class.getName()); + + String command; // Streaming command to be run + String[] argv; // Parsed/split commands + + Process process; // Handle to the process + private static int SUCCESS = 0; + + protected DataOutputStream stdin; // stdin of the process + StoreFunc serializer; // serializer to be used to + // send data to the process + + ProcessOutputThread stdoutThread; // thread to get process output + InputStream stdout; // stdout of the process + LoadFunc deserializer; // deserializer to be used to + // interpret the process' output + + ProcessErrorThread stderrThread; // thread to get process output + InputStream stderr; // stderr of the process + + DataCollector endOfPipe; + + public PigExecutableManager(String command, + LoadFunc deserializer, StoreFunc serializer, + DataCollector endOfPipe) throws Exception { + this.command = command; + + this.argv = splitArgs(this.command); + if (LOG.isDebugEnabled()) { + for (String cmd : argv) { + LOG.debug("argv: " + cmd); + } + } + + this.deserializer = deserializer; + this.serializer = serializer; + this.endOfPipe = endOfPipe; + } + + private static final char SINGLE_QUOTE = '\''; + private static final char DOUBLE_QUOTE = '"'; + private static String[] splitArgs(String command) throws Exception { + List<String> argv = new ArrayList<String>(); + + int beginIndex = 0; + + while (beginIndex < command.length()) { + // Skip spaces + while (Character.isWhitespace(command.charAt(beginIndex))) { + ++beginIndex; + } + + char delim = ' '; + char charAtIndex = command.charAt(beginIndex); + if (charAtIndex == SINGLE_QUOTE || charAtIndex == DOUBLE_QUOTE) { + delim = charAtIndex; + } + + int endIndex = command.indexOf(delim, beginIndex+1); + if (endIndex == -1) { + if (Character.isWhitespace(delim)) { + // Reached end of command-line + argv.add(command.substring(beginIndex)); + break; + } else { + // Didn't find the ending quote/double-quote + throw new ParseException("Illegal command: " + command); + } + } + + if (Character.isWhitespace(delim)) { + // Do not consume the space + argv.add(command.substring(beginIndex, endIndex)); + } else { + // Do not consume the quotes + argv.add(command.substring(beginIndex+1, endIndex)); + } + + beginIndex = endIndex + 1; + } + + return argv.toArray(new String[0]); + } + + public void configure() { + } + + public void close() throws Exception { + // Close the stdin to let the process terminate + stdin.flush(); + stdin.close(); + stdin = null; + + // Wait for the process to exit and the stdout/stderr threads to complete + int exitCode = -1; + try { + exitCode = process.waitFor(); + + if (stdoutThread != null) { + stdoutThread.join(0); + } + if (stderrThread != null) { + stderrThread.join(0); + } + + } catch (InterruptedException ie) {} + + // Clean up the process + process.destroy(); + + LOG.debug("Process exited with: " + exitCode); + if (exitCode != SUCCESS) { + throw new ExecException(command + " failed with exit status: " + + exitCode); + } + } + + public void run() throws IOException { + // Run the executable + ProcessBuilder processBuilder = new ProcessBuilder(argv); + process = processBuilder.start(); + LOG.debug("Started the process for command: " + command); + + // Pick up the process' stdin/stdout/stderr streams + stdin = + new DataOutputStream(new BufferedOutputStream(process.getOutputStream())); + stdout = + new DataInputStream(new BufferedInputStream(process.getInputStream())); + stderr = + new DataInputStream(new BufferedInputStream(process.getErrorStream())); + + // Attach the serializer to the stdin of the process for sending tuples + serializer.bindTo(stdin); + + // Attach the deserializer to the stdout of the process to get tuples + deserializer.bindTo("", new BufferedPositionedInputStream(stdout), 0, + Long.MAX_VALUE); + + // Start the threads to process the executable's stdout and stderr + stdoutThread = new ProcessOutputThread(deserializer); + stdoutThread.start(); + stderrThread = new ProcessErrorThread(); + stderrThread.start(); + } + + public void add(Datum d) throws IOException { + // Pass the serialized tuple to the executable + serializer.putNext((Tuple)d); + stdin.flush(); + } + + /** + * Workhorse to process the stdout of the managed process. + * + * The <code>PigExecutableManager</code>, by default, just pushes the received + * <code>Datum</code> into eval-pipeline to be processed by the successor. + * + * @param d <code>Datum</code> to process + */ + protected void processOutput(Datum d) { + endOfPipe.add(d); + } + + class ProcessOutputThread extends Thread { + + LoadFunc deserializer; + + ProcessOutputThread(LoadFunc deserializer) { + setDaemon(true); + this.deserializer = deserializer; + } + + public void run() { + try { + // Read tuples from the executable and push them down the pipe + Tuple tuple = null; + while ((tuple = deserializer.getNext()) != null) { + processOutput(tuple); + } + + if (stdout != null) { + stdout.close(); + LOG.debug("ProcessOutputThread done"); + } + } catch (Throwable th) { + LOG.warn(th); + try { + if (stdout != null) { + stdout.close(); + } + } catch (IOException ioe) { + LOG.info(ioe); + } + throw new RuntimeException(th); + } + } + } + + /** + * Workhorse to process the stderr stream of the managed process. + * + * By default <code>PigExecuatbleManager</code> just sends out the received + * error message to the <code>stderr</code> of itself. + * + * @param error error message from the managed process. + */ + protected void processError(String error) { + // Just send it out to our stderr + System.err.println(error); + } + + class ProcessErrorThread extends Thread { + + public ProcessErrorThread() { + setDaemon(true); + } + + public void run() { + try { + String error; + BufferedReader reader = + new BufferedReader(new InputStreamReader(stderr)); + while ((error = reader.readLine()) != null) { + processError(error); + } + + if (stderr != null) { + stderr.close(); + LOG.debug("ProcessErrorThread done"); + } + } catch (Throwable th) { + LOG.warn(th); + try { + if (stderr != null) { + stderr.close(); + } + } catch (IOException ioe) { + LOG.info(ioe); + throw new RuntimeException(th); + } + } + } + } +} Added: incubator/pig/trunk/test/org/apache/pig/test/TestStreaming.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/test/org/apache/pig/test/TestStreaming.java?rev=634380&view=auto ============================================================================== --- incubator/pig/trunk/test/org/apache/pig/test/TestStreaming.java (added) +++ incubator/pig/trunk/test/org/apache/pig/test/TestStreaming.java Thu Mar 6 11:11:26 2008 @@ -0,0 +1,178 @@ +/* + * 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.File; + +import org.junit.Assert; +import org.junit.Test; + +import org.apache.pig.PigServer; +import org.apache.pig.builtin.PigStorage; +import org.apache.pig.data.*; +import static org.apache.pig.PigServer.ExecType.MAPREDUCE; + +import junit.framework.TestCase; + +public class TestStreaming extends TestCase { + + MiniCluster cluster = MiniCluster.buildCluster(); + + private static final String simpleEchoStreamingCommand = + "perl -ne 'chomp $_; print \"$_\n\"'"; + + private Tuple[] setupExpectedResults(String[] firstField, int[] secondField) { + Assert.assertEquals(firstField.length, secondField.length); + + Tuple[] expectedResults = new Tuple[firstField.length]; + for (int i=0; i < expectedResults.length; ++i) { + expectedResults[i] = new Tuple(2); + expectedResults[i].setField(0, firstField[i]); + expectedResults[i].setField(1, secondField[i]); + } + + return expectedResults; + } + + @Test + public void testSimpleMapSideStreaming() throws Exception { + PigServer pigServer = new PigServer(MAPREDUCE); + + File input = Util.createInputFile("tmp", "", + new String[] {"A,1", "B,2", "C,3", "D,2", + "A,5", "B,5", "C,8", "A,8", + "D,8", "A,9"}); + + // Expected results + String[] expectedFirstFields = new String[] {"A", "B", "C", "A", "D", "A"}; + int[] expectedSecondFields = new int[] {5, 5, 8, 8, 8, 9}; + Tuple[] expectedResults = + setupExpectedResults(expectedFirstFields, expectedSecondFields); + + // Pig query to run + pigServer.registerQuery("INPUT = load 'file:" + input + "' using " + + PigStorage.class.getName() + "(',');"); + pigServer.registerQuery("FILTERED_DATA = filter INPUT by $1 > '3';"); + pigServer.registerQuery("OUTPUT = stream FILTERED_DATA through `" + + simpleEchoStreamingCommand + "`;"); + + // Run the query and check the results + Util.checkQueryOutputs(pigServer.openIterator("OUTPUT"), expectedResults); + } + + @Test + public void testSimpleMapSideStreamingWithOutputSchema() throws Exception { + PigServer pigServer = new PigServer(MAPREDUCE); + + File input = Util.createInputFile("tmp", "", + new String[] {"A,1", "B,2", "C,3", "D,2", + "A,5", "B,5", "C,8", "A,8", + "D,8", "A,9"}); + + // Expected results + String[] expectedFirstFields = new String[] {"C", "A", "D", "A"}; + int[] expectedSecondFields = new int[] {8, 8, 8, 9}; + Tuple[] expectedResults = + setupExpectedResults(expectedFirstFields, expectedSecondFields); + + // Pig query to run + pigServer.registerQuery("INPUT = load 'file:" + input + "' using " + + PigStorage.class.getName() + "(',');"); + pigServer.registerQuery("FILTERED_DATA = filter INPUT by $1 > '3';"); + pigServer.registerQuery("STREAMED_DATA = stream FILTERED_DATA through `" + + simpleEchoStreamingCommand + "` as (f0, f1);"); + pigServer.registerQuery("OUTPUT = filter STREAMED_DATA by f1 > '6';"); + + // Run the query and check the results + Util.checkQueryOutputs(pigServer.openIterator("OUTPUT"), expectedResults); + } + + @Test + public void testSimpleReduceSideStreamingAfterFlatten() throws Exception { + PigServer pigServer = new PigServer(MAPREDUCE); + + File input = Util.createInputFile("tmp", "", + new String[] {"A,1", "B,2", "C,3", "D,2", + "A,5", "B,5", "C,8", "A,8", + "D,8", "A,9"}); + + // Expected results + String[] expectedFirstFields = new String[] {"A", "A", "A", "B", "C", "D"}; + int[] expectedSecondFields = new int[] {5, 8, 9, 5, 8, 8}; + Tuple[] expectedResults = + setupExpectedResults(expectedFirstFields, expectedSecondFields); + + // Pig query to run + pigServer.registerQuery("INPUT = load 'file:" + input + "' using " + + PigStorage.class.getName() + "(',');"); + pigServer.registerQuery("FILTERED_DATA = filter INPUT by $1 > '3';"); + pigServer.registerQuery("GROUPED_DATA = group FILTERED_DATA by $0;"); + pigServer.registerQuery("FLATTENED_GROUPED_DATA = foreach GROUPED_DATA " + + "generate flatten($1);"); + pigServer.registerQuery("OUTPUT = stream FLATTENED_GROUPED_DATA through `" + + simpleEchoStreamingCommand + "`;"); + + // Run the query and check the results + Util.checkQueryOutputs(pigServer.openIterator("OUTPUT"), expectedResults); + } + + @Test + public void testSimpleOrderedReduceSideStreamingAfterFlatten() throws Exception { + PigServer pigServer = new PigServer(MAPREDUCE); + + File input = Util.createInputFile("tmp", "", + new String[] {"A,1,2,3", "B,2,4,5", + "C,3,1,2", "D,2,5,2", + "A,5,5,1", "B,5,7,4", + "C,8,9,2", "A,8,4,5", + "D,8,8,3", "A,9,2,5"} + ); + + // Expected results + String[] expectedFirstFields = + new String[] {"A", "A", "A", "A", "B", "B", "C", "C", "D", "D"}; + int[] expectedSecondFields = new int[] {1, 9, 8, 5, 2, 5, 3, 8, 2, 8}; + int[] expectedThirdFields = new int[] {2, 2, 4, 5, 4, 7, 1, 9, 5, 8}; + int[] expectedFourthFields = new int[] {3, 5, 5, 1, 5, 4, 2, 2, 2, 3}; + Tuple[] expectedResults = new Tuple[10]; + for (int i = 0; i < expectedResults.length; ++i) { + expectedResults[i] = new Tuple(4); + expectedResults[i].setField(0, expectedFirstFields[i]); + expectedResults[i].setField(1, expectedSecondFields[i]); + expectedResults[i].setField(2, expectedThirdFields[i]); + expectedResults[i].setField(3, expectedFourthFields[i]); + } + setupExpectedResults(expectedFirstFields, expectedSecondFields); + + // Pig query to run + pigServer.registerQuery("INPUT = load 'file:" + input + "' using " + + PigStorage.class.getName() + "(',');"); + pigServer.registerQuery("FILTERED_DATA = filter INPUT by $1 > '3';"); + pigServer.registerQuery("GROUPED_DATA = group INPUT by $0;"); + pigServer.registerQuery("ORDERED_DATA = foreach GROUPED_DATA { " + + " D = order INPUT BY $2, $3;" + + " generate flatten(D);" + + "};"); + pigServer.registerQuery("OUTPUT = stream ORDERED_DATA through `" + + simpleEchoStreamingCommand + "`;"); + + // Run the query and check the results + Util.checkQueryOutputs(pigServer.openIterator("OUTPUT"), expectedResults); + } + +} Modified: incubator/pig/trunk/test/org/apache/pig/test/Util.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/test/org/apache/pig/test/Util.java?rev=634380&r1=634379&r2=634380&view=diff ============================================================================== --- incubator/pig/trunk/test/org/apache/pig/test/Util.java (original) +++ incubator/pig/trunk/test/org/apache/pig/test/Util.java Thu Mar 6 11:11:26 2008 @@ -17,9 +17,13 @@ */ package org.apache.pig.test; +import java.io.File; import java.io.IOException; +import java.io.PrintWriter; +import java.util.Iterator; import org.apache.pig.data.*; +import org.junit.Assert; public class Util { // Helper Functions @@ -68,4 +72,43 @@ } return t; } + + /** + * Helper to create a temporary file with given input data for use in test cases. + * + * @param tmpFilenamePrefix file-name prefix + * @param tmpFilenameSuffix file-name suffix + * @param inputData input for test cases, each string in inputData[] is written + * on one line + * @return {@link File} handle to the created temporary file + * @throws IOException + */ + static public File createInputFile(String tmpFilenamePrefix, + String tmpFilenameSuffix, + String[] inputData) + throws IOException { + File f = File.createTempFile(tmpFilenamePrefix, tmpFilenameSuffix); + PrintWriter pw = new PrintWriter(f); + for (int i=0; i<inputData.length; i++){ + pw.println(inputData[i]); + } + pw.close(); + return f; + } + + /** + * Helper function to check if the result of a Pig Query is in line with + * expected results. + * + * @param actualResults Result of the executed Pig query + * @param expectedResults Expected results to validate against + */ + static public void checkQueryOutputs(Iterator<Tuple> actualResults, + Tuple[] expectedResults) { + for (Tuple expected : expectedResults) { + Tuple actual = actualResults.next(); + Assert.assertEquals(expected, actual); + } + } + } From pig-commits-returnzcarey@example.org Thu Mar 06 23:13:54 2008 Return-Path: <pig-commits-returnzcarey@example.org> Delivered-To: zcarey@example.org Received: (qmail 98733 invoked from network); 6 Mar 2008 23:13:53 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 6 Mar 2008 23:13:53 -0000 Received: (qmail 47935 invoked by uid 500); 6 Mar 2008 23:13:49 -0000 Delivered-To: zcarey@example.org Received: (qmail 47920 invoked by uid 500); 6 Mar 2008 23:13:49 -0000 Mailing-List: contact zcarey@example.org; run by ezmlm Precedence: bulk List-Help: <mailto:zcarey@example.org> List-Unsubscribe: <mailto:zcarey@example.org> List-Post: <mailto:zcarey@example.org> List-Id: <pig-commits.incubator.apache.org> Reply-To: zcarey@example.org Delivered-To: mailing list zcarey@example.org Received: (qmail 47911 invoked by uid 99); 6 Mar 2008 23:13:49 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 06 Mar 2008 15:13:49 -0800 X-ASF-Spam-Status: No, hits=-2000.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 06 Mar 2008 23:13:10 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 08B9B1A9832; Thu, 6 Mar 2008 15:13:30 -0800 (PST) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r634453 - in /incubator/pig/trunk/src/org/apache/pig/impl: eval/StreamSpec.java streaming/PigExecutableManager.java Date: Thu, 06 Mar 2008 23:13:29 -0000 To: zcarey@example.org From: zcarey@example.org X-Mailer: svnmailer-1.0.8 Message-Id: <zcarey@example.org> X-Virus-Checked: Checked by ClamAV on apache.org Author: olga Date: Thu Mar 6 15:13:15 2008 New Revision: 634453 URL: http://svn.apache.org/viewvc?rev=634453&view=rev Log: PIG-94: compile fix Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java Modified: incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java?rev=634453&r1=634452&r2=634453&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java (original) +++ incubator/pig/trunk/src/org/apache/pig/impl/eval/StreamSpec.java Thu Mar 6 15:13:15 2008 @@ -25,7 +25,7 @@ import org.apache.commons.logging.LogFactory; import org.apache.pig.LoadFunc; import org.apache.pig.StoreFunc; -import org.apache.pig.backend.streaming.PigExecutableManager; +import org.apache.pig.impl.streaming.PigExecutableManager; import org.apache.pig.builtin.PigStorage; import org.apache.pig.data.Datum; import org.apache.pig.impl.PigContext; Modified: incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java URL: http://svn.apache.org/viewvc/incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java?rev=634453&r1=634452&r2=634453&view=diff ============================================================================== --- incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java (original) +++ incubator/pig/trunk/src/org/apache/pig/impl/streaming/PigExecutableManager.java Thu Mar 6 15:13:15 2008 @@ -15,7 +15,7 @@ * See the License for the specific language governing
{'source': 'Apache Software Foundation Public Mail Archives'}