Return-Path: X-Original-To: apmail-pig-commits-archive@www.apache.org Delivered-To: apmail-pig-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id 1199F9339 for ; Thu, 13 Sep 2012 14:56:29 +0000 (UTC) Received: (qmail 66400 invoked by uid 500); 13 Sep 2012 14:56:28 -0000 Delivered-To: apmail-pig-commits-archive@pig.apache.org Received: (qmail 66298 invoked by uid 500); 13 Sep 2012 14:56:28 -0000 Mailing-List: contact commits-help@pig.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@pig.apache.org Delivered-To: mailing list commits@pig.apache.org Received: (qmail 66156 invoked by uid 99); 13 Sep 2012 14:56:28 -0000 Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 13 Sep 2012 14:56:28 +0000 X-ASF-Spam-Status: No, hits=-2000.0 required=5.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [140.211.11.4] (HELO eris.apache.org) (140.211.11.4) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 13 Sep 2012 14:56:24 +0000 Received: from eris.apache.org (localhost [127.0.0.1]) by eris.apache.org (Postfix) with ESMTP id 8773C2388AAA for ; Thu, 13 Sep 2012 14:55:41 +0000 (UTC) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r1384352 [3/4] - in /pig/trunk: ./ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer... Date: Thu, 13 Sep 2012 14:55:38 -0000 To: commits@pig.apache.org From: gdfm@apache.org X-Mailer: svnmailer-1.0.8-patched Message-Id: <20120913145541.8773C2388AAA@eris.apache.org> X-Virus-Checked: Checked by ClamAV on apache.org Modified: pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java (original) +++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java Thu Sep 13 14:55:36 2012 @@ -57,6 +57,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOJoin; import org.apache.pig.newplan.logical.relational.LOLimit; import org.apache.pig.newplan.logical.relational.LOLoad; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LOSplit; import org.apache.pig.newplan.logical.relational.LOSplitOutput; @@ -373,13 +374,21 @@ public class LineageFindRelVisitor exten visitExpression(expPlan); } } - - + + @Override + public void visit(LORank rank) throws FrontendException{ + mapToPredLoadFunc(rank); + List expPlans = rank.getRankColPlans(); + for(LogicalExpressionPlan expPlan : expPlans){ + visitExpression(expPlan); + } + } + @Override public void visit(LODistinct relOp) throws FrontendException{ mapToPredLoadFunc(relOp); } - + @Override public void visit(LOLimit loLimit) throws FrontendException{ mapToPredLoadFunc(loLimit); Modified: pig/trunk/src/org/apache/pig/newplan/logical/visitor/ProjectStarExpander.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/ProjectStarExpander.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/newplan/logical/visitor/ProjectStarExpander.java (original) +++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/ProjectStarExpander.java Thu Sep 13 14:55:36 2012 @@ -42,6 +42,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOGenerate; import org.apache.pig.newplan.logical.relational.LOInnerLoad; import org.apache.pig.newplan.logical.relational.LOJoin; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LogicalPlan; import org.apache.pig.newplan.logical.relational.LogicalRelationalNodesVisitor; @@ -113,11 +114,55 @@ public class ProjectStarExpander extends ); } } - + sort.setSortColPlans(newExpPlans); sort.setAscendingCols(newAscOrder); } + @Override + public void visit(LORank rank) throws FrontendException { + + List expPlans = rank.getRankColPlans(); + List ascOrder = rank.getAscendingCol(); + + List newExpPlans = new ArrayList(); + List newAscOrder = new ArrayList(); + + if (expPlans.size() != ascOrder.size()) { + throw new AssertionError( + "Size of expPlans and ascorder should be same"); + } + + for (int i = 0; i < expPlans.size(); i++) { + // expand the plan + LogicalExpressionPlan ithExpPlan = expPlans.get(i); + List expandedPlans = expandPlan(ithExpPlan, + 0); + newExpPlans.addAll(expandedPlans); + + // add corresponding isAsc flags + Boolean isAsc = ascOrder.get(i); + for (int j = 0; j < expandedPlans.size(); j++) { + newAscOrder.add(isAsc); + } + } + + // check if there is a project-star-to-end followed by another sort plan + // in the expanded plans (can happen if there is no input schema) + for (int i = 0; i < newExpPlans.size(); i++) { + ProjectExpression proj = getProjectStar(newExpPlans.get(i)); + if (proj != null && proj.isRangeProject() && proj.getEndCol() == -1 + && i != newExpPlans.size() - 1) { + String msg = "Project-range to end (eg. x..)" + + " is supported in rank-by only as last rank column"; + throw new FrontendException(msg, 1128, PigException.INPUT); + } + } + + rank.setRankColPlan(newExpPlans); + rank.setAscendingCol(newAscOrder); + } + /** * Expand plan into multiple plans if the plan contains a project star, * if there is no project star the returned list contains the plan argument. Modified: pig/trunk/src/org/apache/pig/newplan/logical/visitor/SchemaAliasVisitor.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/SchemaAliasVisitor.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/newplan/logical/visitor/SchemaAliasVisitor.java (original) +++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/SchemaAliasVisitor.java Thu Sep 13 14:55:36 2012 @@ -36,6 +36,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOJoin; import org.apache.pig.newplan.logical.relational.LOLimit; import org.apache.pig.newplan.logical.relational.LONative; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LOSplit; import org.apache.pig.newplan.logical.relational.LOSplitOutput; @@ -54,11 +55,11 @@ public class SchemaAliasVisitor extends /*** * The logic here is to check if we have duplicate alias in each schema - * @throws FrontendException + * @throws FrontendException */ protected void validate(LogicalRelationalOperator op) throws FrontendException { LogicalSchema schema = op.getSchema(); - + Set seenAliases = new HashSet(); if( schema != null){ for( int i = 0; i < schema.size(); i++){ @@ -88,22 +89,22 @@ public class SchemaAliasVisitor extends public void visit(LOFilter filter) throws FrontendException { validate( filter ); } - + @Override public void visit(LOJoin join) throws FrontendException { validate( join ); } - + @Override public void visit(LOForEach foreach) throws FrontendException { new SchemaAliasVisitor( foreach.getInnerPlan() ).visit(); } - + @Override public void visit(LOGenerate gen) throws FrontendException { validate( gen ); } - + @Override public void visit(LOInnerLoad load) throws FrontendException { validate( load ); @@ -113,42 +114,47 @@ public class SchemaAliasVisitor extends public void visit(LOCogroup group) throws FrontendException { validate( group ); } - + @Override public void visit(LOSplit split) throws FrontendException { validate( split ); } - + @Override public void visit(LOSplitOutput splitOutput) throws FrontendException { validate( splitOutput ); } - + @Override public void visit(LOUnion union) throws FrontendException { validate( union ); } - + @Override public void visit(LOSort sort) throws FrontendException { validate( sort ); } - + + @Override + public void visit(LORank rank) throws FrontendException { + validate( rank ); + } + @Override public void visit(LODistinct distinct) throws FrontendException { validate( distinct ); } - + @Override public void visit(LOLimit limit) throws FrontendException { validate( limit ); } - + @Override public void visit(LOCross cross) throws FrontendException { validate( cross ); } - + @Override public void visit(LOStream stream) throws FrontendException { validate( stream ); Modified: pig/trunk/src/org/apache/pig/newplan/logical/visitor/TypeCheckingRelVisitor.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/TypeCheckingRelVisitor.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/newplan/logical/visitor/TypeCheckingRelVisitor.java (original) +++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/TypeCheckingRelVisitor.java Thu Sep 13 14:55:36 2012 @@ -46,6 +46,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOJoin; import org.apache.pig.newplan.logical.relational.LOLimit; import org.apache.pig.newplan.logical.relational.LOLoad; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LOSplit; import org.apache.pig.newplan.logical.relational.LOSplitOutput; @@ -547,6 +548,45 @@ public class TypeCheckingRelVisitor exte } /*** + * The schema of rank output will be the same as input, plus a rank field. + * @throws FrontendException + * + */ + public void visit(LORank rank) throws FrontendException { + rank.resetSchema(); + + // Type checking internal plans. + List rankColPlans = rank.getRankColPlans(); + + for(int i=0;i < rankColPlans.size(); i++) { + LogicalExpressionPlan rankColPlan = rankColPlans.get(i) ; + + // Check that the inner plan has only 1 output port + if (rankColPlan.getSources().size() != 1) { + int errCode = 1057; + String msg = "Rank's inner plan can only have one output (leaf)" ; + msgCollector.collect(msg, MessageType.Error) ; + throwTypeCheckerException(rank, msg, errCode, PigException.INPUT, null) ; + } + + visitExpressionPlan(rankColPlan, rank); + + } + + try { + // Compute the schema + rank.getSchema() ; + } + catch (FrontendException fee) { + int errCode = 1059; + String msg = "Problem while reconciling output schema of Rank" ; + msgCollector.collect(msg, MessageType.Error); + throwTypeCheckerException(rank, msg, errCode, PigException.INPUT, fee) ; + } + + } + + /*** * The schema of split output will be the same as split input */ Modified: pig/trunk/src/org/apache/pig/parser/AliasMasker.g URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/AliasMasker.g?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/parser/AliasMasker.g (original) +++ pig/trunk/src/org/apache/pig/parser/AliasMasker.g Thu Sep 13 14:55:36 2012 @@ -123,6 +123,7 @@ op_clause : define_clause | limit_clause | sample_clause | order_clause + | rank_clause | cross_clause | join_clause | union_clause @@ -374,48 +375,65 @@ col_index col_range : ^(COL_RANGE col_ref? DOUBLE_PERIOD col_ref?) ; -pound_proj +pound_proj : ^( POUND ( QUOTEDSTRING | NULL ) ) ; -bin_expr - : ^( BIN_EXPR cond expr expr ) +bin_expr + : ^( BIN_EXPR cond expr expr ) ; -limit_clause +limit_clause : ^( LIMIT rel ( INTEGER | LONGINTEGER | expr ) ) ; -sample_clause +sample_clause : ^( SAMPLE rel ( DOUBLENUMBER | expr ) ) ; -order_clause +rank_clause + : ^( RANK rel ( rank_by_statement )? ) +; + +rank_by_statement + : ^( BY rank_by_clause ( DENSE )? ) +; + +rank_by_clause + : STAR ( ASC | DESC )? + | rank_col+ +; + +rank_col + : ( col_range | col_ref ) ( ASC | DESC )? +; + +order_clause : ^( ORDER rel order_by_clause func_clause? ) ; -order_by_clause +order_by_clause : STAR ( ASC | DESC )? | order_col+ ; -order_col - : (col_range | col_ref) ( ASC | DESC )? +order_col + : (col_range | col_ref) ( ASC | DESC )? ; -distinct_clause +distinct_clause : ^( DISTINCT rel partition_clause? ) ; -partition_clause - : ^( PARTITION func_name ) +partition_clause + : ^( PARTITION func_name ) ; -cross_clause - : ^( CROSS rel_list partition_clause? ) +cross_clause + : ^( CROSS rel_list partition_clause? ) ; -rel_list +rel_list : rel+ ; @@ -601,6 +619,7 @@ eid : rel_str_op | ROLLUP | MATCHES | ORDER + | RANK | DISTINCT | COGROUP | JOIN Modified: pig/trunk/src/org/apache/pig/parser/AstPrinter.g URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/AstPrinter.g?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/parser/AstPrinter.g (original) +++ pig/trunk/src/org/apache/pig/parser/AstPrinter.g Thu Sep 13 14:55:36 2012 @@ -90,6 +90,7 @@ op_clause : define_clause | limit_clause | sample_clause | order_clause + | rank_clause | cross_clause | join_clause | union_clause @@ -346,25 +347,42 @@ col_index col_range : ^(COL_RANGE col_ref? { sb.append(".."); } DOUBLE_PERIOD col_ref?) ; -pound_proj +pound_proj : ^( POUND { sb.append($POUND.text); } ( QUOTEDSTRING { sb.append($QUOTEDSTRING.text); } | NULL { sb.append($NULL.text); } ) ) ; -bin_expr - : ^( BIN_EXPR { sb.append(" ("); } cond { sb.append(" ? "); } expr { sb.append(" : "); } expr { sb.append(") "); } ) +bin_expr + : ^( BIN_EXPR { sb.append(" ("); } cond { sb.append(" ? "); } expr { sb.append(" : "); } expr { sb.append(") "); } ) ; -limit_clause - : ^( LIMIT { sb.append($LIMIT.text).append(" "); } rel +limit_clause + : ^( LIMIT { sb.append($LIMIT.text).append(" "); } rel ( INTEGER { sb.append(" ").append($INTEGER.text); } | LONGINTEGER { sb.append(" ").append($LONGINTEGER.text); } | expr ) ) ; -sample_clause - : ^( SAMPLE { sb.append($SAMPLE.text).append(" "); } rel ( DOUBLENUMBER { sb.append(" ").append($DOUBLENUMBER.text); } | expr ) ) +sample_clause + : ^( SAMPLE { sb.append($SAMPLE.text).append(" "); } rel ( DOUBLENUMBER { sb.append(" ").append($DOUBLENUMBER.text); } | expr ) ) ; -order_clause +rank_clause + : ^( RANK { sb.append($RANK.text).append(" "); } rel ( rank_by_statement )? ) +; + +rank_by_statement + : ^( BY { sb.append(" ").append($BY.text); } rank_by_clause ( DENSE { sb.append(" ").append($DENSE.text); } )? ) +; + +rank_by_clause + : STAR { sb.append($STAR.text); } ( ASC { sb.append(" ").append($ASC.text); } | DESC { sb.append(" ").append($DESC.text); } )? + | rank_col ( { sb.append(", "); } rank_col )* +; + +rank_col + : ( col_range | col_ref ) ( ASC { sb.append(" ").append($ASC.text); } | DESC { sb.append(" ").append($DESC.text); } )? +; + +order_clause : ^( ORDER { sb.append($ORDER.text).append(" "); } rel { sb.append(" BY "); } order_by_clause ( { sb.append(" USING "); } func_clause )? ) @@ -589,6 +607,7 @@ eid : rel_str_op | ROLLUP { sb.append($ROLLUP.text); } | MATCHES { sb.append($MATCHES.text); } | ORDER { sb.append($ORDER.text); } + | RANK { sb.append($RANK.text); } | DISTINCT { sb.append($DISTINCT.text); } | COGROUP { sb.append($COGROUP.text); } | JOIN { sb.append($JOIN.text); } Modified: pig/trunk/src/org/apache/pig/parser/AstValidator.g URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/AstValidator.g?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/parser/AstValidator.g (original) +++ pig/trunk/src/org/apache/pig/parser/AstValidator.g Thu Sep 13 14:55:36 2012 @@ -138,6 +138,7 @@ op_clause : define_clause | limit_clause | sample_clause | order_clause + | rank_clause | cross_clause | join_clause | union_clause @@ -398,6 +399,20 @@ limit_clause : ^( LIMIT rel ( INTEGER | sample_clause : ^( SAMPLE rel ( DOUBLENUMBER | expr ) ) ; +rank_clause : ^( RANK rel ( rank_by_statement )? ) +; + +rank_by_statement : ^( BY rank_by_clause ( DENSE )? ) +; + +rank_by_clause : STAR ( ASC | DESC )? + | rank_col+ +; + +rank_col : col_range (ASC | DESC)? + | col_ref ( ASC | DESC )? +; + order_clause : ^( ORDER rel order_by_clause func_clause? ) ; @@ -599,6 +614,7 @@ eid : rel_str_op | ROLLUP | MATCHES | ORDER + | RANK | DISTINCT | COGROUP | JOIN Modified: pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java (original) +++ pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java Thu Sep 13 14:55:36 2012 @@ -79,6 +79,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOLimit; import org.apache.pig.newplan.logical.relational.LOLoad; import org.apache.pig.newplan.logical.relational.LONative; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LOSplit; import org.apache.pig.newplan.logical.relational.LOSplitOutput; @@ -306,7 +307,28 @@ public class LogicalPlanBuilder { expandAndResetVisitor(loc, sort); return alias; } - + + LORank createRankOp() { + return new LORank( plan ); + } + + String buildRankOp(SourceLocation loc, LORank rank, String alias, String inputAlias, List plans, + List ascFlags) throws ParserValidationException { + + //Rank + rank.setRankColPlan(plans); + if (ascFlags.isEmpty()) { + for (int i=0;i tempPlans = $rank_by_statement.plans; + List tempAscFlags = $rank_by_statement.ascFlags; + + if(tempPlans == null && tempAscFlags == null) { + tempPlans = new ArrayList(); + tempAscFlags = new ArrayList(); + + ((LORank)$GScope::currentOp).setIsRowNumber( true ); + } + + ((LORank)$GScope::currentOp).setIsDenseRank( $rank_by_statement.isDenseRank != null?$rank_by_statement.isDenseRank:false ); + + $alias = builder.buildRankOp( loc, (LORank)$GScope::currentOp, $statement::alias, $statement::inputAlias, tempPlans, tempAscFlags ); + } +; + +rank_by_statement returns[List plans, List ascFlags, Boolean isDenseRank] +@init { + $plans = new ArrayList(); + $ascFlags = new ArrayList(); + $isDenseRank = false; +} + : ^( BY rank_by_clause ( DENSE { $isDenseRank = true; } )? ) + { + $plans = $rank_by_clause.plans; + $ascFlags = $rank_by_clause.ascFlags; + } +; + +rank_by_clause returns[List plans, List ascFlags] +@init { + $plans = new ArrayList(); + $ascFlags = new ArrayList(); +} + : STAR { + LogicalExpressionPlan plan = new LogicalExpressionPlan(); + builder.buildProjectExpr( new SourceLocation( (PigParserNode)$STAR ), plan, $GScope::currentOp, $statement::inputIndex, null, -1 ); + $plans.add( plan ); + } + ( ASC { $ascFlags.add( true ); } | DESC { $ascFlags.add( false ); } )? + | ( rank_col + { + $plans.add( $rank_col.plan ); + $ascFlags.add( $rank_col.ascFlag ); + } )+ +; + +rank_col returns[LogicalExpressionPlan plan, Boolean ascFlag] +@init { + $plan = new LogicalExpressionPlan(); + $ascFlag = true; +} + : col_range[$plan] (ASC | DESC { $ascFlag = false; } )? + | col_ref[$plan] ( ASC | DESC { $ascFlag = false; } )? +; + order_clause returns[String alias] scope GScope; @init { Modified: pig/trunk/src/org/apache/pig/parser/QueryLexer.g URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/QueryLexer.g?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/parser/QueryLexer.g (original) +++ pig/trunk/src/org/apache/pig/parser/QueryLexer.g Thu Sep 13 14:55:36 2012 @@ -81,6 +81,12 @@ FOREACH : 'FOREACH' ORDER : 'ORDER' ; +RANK : 'RANK' +; + +DENSE : 'DENSE' +; + CUBE : 'CUBE' ; Modified: pig/trunk/src/org/apache/pig/parser/QueryParser.g URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/QueryParser.g?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/parser/QueryParser.g (original) +++ pig/trunk/src/org/apache/pig/parser/QueryParser.g Thu Sep 13 14:55:36 2012 @@ -222,6 +222,7 @@ op_clause : define_clause | limit_clause | sample_clause | order_clause + | rank_clause | cross_clause | join_clause | union_clause @@ -501,6 +502,24 @@ limit_clause : LIMIT^ rel ( (INTEGER SEM sample_clause : SAMPLE^ rel ( (DOUBLENUMBER SEMI_COLON) => DOUBLENUMBER | expr ) ; +rank_clause : RANK^ rel ( rank_by_statement )? +; + +rank_by_statement : BY^ rank_by_clause ( DENSE )? +; + +rank_by_clause : STAR ( ASC | DESC )? + | rank_list +; + +rank_list : rank_col ( COMMA rank_col )* + -> rank_col+ +; + +rank_col : col_range ( ASC | DESC )? + | col_ref ( ASC | DESC )? +; + order_clause : ORDER^ rel BY! order_by_clause ( USING! func_clause )? ; Modified: pig/trunk/src/org/apache/pig/pen/IllustratorAttacher.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/pen/IllustratorAttacher.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/pen/IllustratorAttacher.java (original) +++ pig/trunk/src/org/apache/pig/pen/IllustratorAttacher.java Thu Sep 13 14:55:36 2012 @@ -27,7 +27,9 @@ import java.util.List; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCounter; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORank; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCollectedGroup; @@ -93,13 +95,13 @@ public class IllustratorAttacher extends LineageTracer lineage; HashMap>> poToEqclassesMap; - + private HashMap poToDataMap; private int maxRecords; private boolean revisit = false; private ArrayList subExpResults = null; private final Map poloadToSchemaMap; - + public IllustratorAttacher(PhysicalPlan plan, LineageTracer lineage, int maxRecords, Map poLoadToSchemaMap, PigContext hadoopPigContext) throws VisitorException { super(plan, new DepthFirstWalker(plan)); @@ -124,7 +126,7 @@ public class IllustratorAttacher extends mPlan = oriPlan; popWalker(); } - + private void setIllustrator(PhysicalOperator po, int nEqClasses) { if (revisit && po.getIllustrator() != null) return; @@ -139,7 +141,7 @@ public class IllustratorAttacher extends po.setIllustrator(illustrator); poToDataMap.put(po, illustrator.getData()); } - + private void setIllustrator(PhysicalOperator po, LinkedList> eqClasses) { if (revisit && po.getIllustrator() != null) return; @@ -161,20 +163,20 @@ public class IllustratorAttacher extends poToEqclassesMap.put(po, eqClasses); poToDataMap.put(po, illustrator.getData()); } - + public Map getDataMap() { return poToDataMap; } - + @Override public void visitLoad(POLoad ld) throws VisitorException{ - // LOAD from temporary files need no illustrator + // LOAD from temporary files need no illustrator if (revisit) return; - + LinkedList> eqClasses = new LinkedList>(); poToEqclassesMap.put(ld, eqClasses); - + IdentityHashSet eqClass = new IdentityHashSet(); eqClasses.add(eqClass); Illustrator illustrator; @@ -182,12 +184,12 @@ public class IllustratorAttacher extends ld.setIllustrator(illustrator); poToDataMap.put(ld, illustrator.getData()); } - + @Override public void visitStore(POStore st) throws VisitorException{ setIllustrator(st, 1); } - + @Override public void visitFilter(POFilter fl) throws VisitorException{ setIllustrator(fl, 0); @@ -195,13 +197,13 @@ public class IllustratorAttacher extends innerPlanAttach(fl, fl.getPlan()); subExpResults = null; } - + @Override public void visitLocalRearrange(POLocalRearrange lr) throws VisitorException{ super.visitLocalRearrange(lr); setIllustrator(lr); } - + @Override public void visitPackage(POPackage pkg) throws VisitorException{ if (!(pkg instanceof POPackageLite) && pkg.isDistinct()) @@ -209,17 +211,17 @@ public class IllustratorAttacher extends else setIllustrator(pkg, null); } - + @Override public void visitCombinerPackage(POCombinerPackage pkg) throws VisitorException{ setIllustrator(pkg); } - + @Override public void visitMultiQueryPackage(POMultiQueryPackage pkg) throws VisitorException{ setIllustrator(pkg); } - + @Override public void visitPOForEach(POForEach nfe) throws VisitorException { if (revisit && nfe.getIllustrator() != null) @@ -239,14 +241,14 @@ public class IllustratorAttacher extends } else setIllustrator(nfe, 1); } - + @Override public void visitUnion(POUnion un) throws VisitorException{ if (revisit && un.getIllustrator() != null) return; setIllustrator(un, null); } - + @Override public void visitSplit(POSplit spl) throws VisitorException{ if (revisit && spl.getIllustrator() != null) @@ -265,63 +267,73 @@ public class IllustratorAttacher extends innerPlanAttach(demux, innerPlan); setIllustrator(demux); } - + @Override public void visitDistinct(PODistinct distinct) throws VisitorException { setIllustrator(distinct, 1); } - + @Override public void visitSort(POSort sort) throws VisitorException { setIllustrator(sort, 1); } - + + @Override + public void visitRank(PORank rank) throws VisitorException { + setIllustrator(rank, 3); + } + + @Override + public void visitCounter(POCounter counter) throws VisitorException { + setIllustrator(counter, 1); + } + @Override public void visitProject(POProject proj) throws VisitorException{ } - + @Override public void visitGreaterThan(GreaterThanExpr grt) throws VisitorException{ setIllustrator(grt, 0); if (!revisit && subExpResults != null) subExpResults.add(grt.getIllustrator().getSubExpResult()); } - + @Override public void visitLessThan(LessThanExpr lt) throws VisitorException{ setIllustrator(lt, 0); if (!revisit && subExpResults != null) subExpResults.add(lt.getIllustrator().getSubExpResult()); } - + @Override public void visitGTOrEqual(GTOrEqualToExpr gte) throws VisitorException{ setIllustrator(gte, 0); if (!revisit && subExpResults != null) subExpResults.add(gte.getIllustrator().getSubExpResult()); } - + @Override public void visitLTOrEqual(LTOrEqualToExpr lte) throws VisitorException{ setIllustrator(lte, 0); if (!revisit && subExpResults != null) subExpResults.add(lte.getIllustrator().getSubExpResult()); } - + @Override public void visitEqualTo(EqualToExpr eq) throws VisitorException{ setIllustrator(eq, 0); if (!revisit && subExpResults != null) subExpResults.add(eq.getIllustrator().getSubExpResult()); } - + @Override public void visitNotEqualTo(NotEqualToExpr eq) throws VisitorException{ setIllustrator(eq, 0); if (!revisit && subExpResults != null) subExpResults.add(eq.getIllustrator().getSubExpResult()); } - + @Override public void visitRegexp(PORegexp re) throws VisitorException{ setIllustrator(re, 0); @@ -335,12 +347,12 @@ public class IllustratorAttacher extends if (!revisit && subExpResults != null) subExpResults.add(isNull.getIllustrator().getSubExpResult()); } - + @Override public void visitAnd(POAnd and) throws VisitorException { setIllustrator(and, 0); } - + @Override public void visitOr(POOr or) throws VisitorException { setIllustrator(or, 0); @@ -362,11 +374,11 @@ public class IllustratorAttacher extends public void visitNegative(PONegative negative) { setIllustrator(negative, 1); } - + @Override public void visitUserFunc(POUserFunc userFunc) throws VisitorException { } - + @Override public void visitComparisonFunc(POUserComparisonFunc compFunc) throws VisitorException { // one each for >, ==, and < @@ -377,7 +389,7 @@ public class IllustratorAttacher extends public void visitMapLookUp(POMapLookUp mapLookUp) { setIllustrator(mapLookUp, 1); } - + @Override public void visitJoinPackage(POJoinPackage joinPackage) throws VisitorException{ if (revisit && joinPackage.getIllustrator() != null) @@ -389,12 +401,12 @@ public class IllustratorAttacher extends @Override public void visitCast(POCast cast) { } - + @Override public void visitLimit(POLimit lim) throws VisitorException { setIllustrator(lim, 1); } - + @Override public void visitStream(POStream stream) throws VisitorException { setIllustrator(stream, 1); @@ -407,7 +419,7 @@ public class IllustratorAttacher extends public void visitPOOptimizedForEach(POOptimizedForEach optimizedForEach) throws VisitorException { visitPOForEach(optimizedForEach); } - + private void innerPlanAttach(PhysicalOperator po, PhysicalPlan plan) throws VisitorException { PlanWalker childWalker = mCurrentWalker.spawnChildWalker(plan); Modified: pig/trunk/src/org/apache/pig/pen/LocalMapReduceSimulator.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/pen/LocalMapReduceSimulator.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/pen/LocalMapReduceSimulator.java (original) +++ pig/trunk/src/org/apache/pig/pen/LocalMapReduceSimulator.java Thu Sep 13 14:55:36 2012 @@ -40,6 +40,7 @@ import org.apache.pig.backend.hadoop.exe import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapBase; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapOnly; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduce; +import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigMapReduceCounter; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.plans.MROperPlan; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator; @@ -184,13 +185,19 @@ public class LocalMapReduceSimulator { split = new PigSplit(null, index, needFileInput ? emptyInpTargets : inpTargets.get(index), 0); ++index; Mapper map; - + if (mro.reducePlan.isEmpty()) { // map-only map = new PigMapOnly.Map(); - ((PigMapBase) map).setMapPlan(mro.mapPlan); Mapper.Context context = ((PigMapOnly.Map) map) .getIllustratorContext(jobConf, input, intermediateData, split); + if(mro.isCounterOperation()) { + if(mro.isRowNumber()) { + map = new PigMapReduceCounter.PigMapCounter(); + } + context = ((PigMapReduceCounter.PigMapCounter) map).getIllustratorContext(jobConf, input, intermediateData, split); + } + ((PigMapBase) map).setMapPlan(mro.mapPlan); map.run(context); } else { if ("true".equals(jobConf.get("pig.usercomparator"))) @@ -216,9 +223,15 @@ public class LocalMapReduceSimulator { reduce = new PigMapReduce.ReduceWithComparator(); else reduce = new PigMapReduce.Reduce(); - reduce.setReducePlan(mro.reducePlan); Reducer.Context context = reduce.getIllustratorContext(job, intermediateData, (POPackage) pack); + + if(mro.isCounterOperation()) { + reduce = new PigMapReduceCounter.PigReduceCounter(); + context = ((PigMapReduceCounter.PigReduceCounter)reduce).getIllustratorContext(job, intermediateData, (POPackage) pack); + } + + ((PigMapReduce.Reduce) reduce).setReducePlan(mro.reducePlan); reduce.run(context); } for (PhysicalOperator key : mro.phyToMRMap.keySet()) Modified: pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java (original) +++ pig/trunk/src/org/apache/pig/tools/pigstats/ScriptState.java Thu Sep 13 14:55:36 2012 @@ -61,6 +61,8 @@ import org.apache.pig.backend.hadoop.exe import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeCogroup; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeJoin; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPartialAgg; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POCounter; +import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORank; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSkewedJoin; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSort; import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POSplit; @@ -82,6 +84,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOJoin; import org.apache.pig.newplan.logical.relational.LOLimit; import org.apache.pig.newplan.logical.relational.LONative; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LOSplit; import org.apache.pig.newplan.logical.relational.LOStream; @@ -93,14 +96,14 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.tools.pigstats.PigStats.JobGraph; /** - * ScriptStates encapsulates settings for a Pig script that runs on a hadoop - * cluster. These settings are added to all MR jobs spawned by the script and - * in turn are persisted in the hadoop job xml. With the properties already in + * ScriptStates encapsulates settings for a Pig script that runs on a hadoop + * cluster. These settings are added to all MR jobs spawned by the script and + * in turn are persisted in the hadoop job xml. With the properties already in * the job xml, users who want to know the relations between the script and MR - * jobs can derive them from the job xmls. + * jobs can derive them from the job xmls. */ public class ScriptState { - + /** * Keys of Pig settings added in MR job */ @@ -118,17 +121,17 @@ public class ScriptState { SCRIPT_FEATURES ("pig.script.features"), JOB_ALIAS ("pig.alias"), JOB_ALIAS_LOCATION ("pig.alias.location"); - + private String displayStr; - + private PIG_PROPERTY(String s) { displayStr = s; } - + @Override public String toString() { return displayStr; } }; - + /** * Features used in a Pig script */ @@ -144,6 +147,7 @@ public class ScriptState { COGROUP, GROUP_BY, ORDER_BY, + RANK, DISTINCT, STREAMING, SAMPLER, @@ -158,42 +162,42 @@ public class ScriptState { NATIVE, MAP_PARTIALAGG; }; - + /** * Pig property that allows user to turn off the inclusion of settings - * in the jobs + * in the jobs */ public static final String INSERT_ENABLED = "pig.script.info.enabled"; - + /** - * Restricts the size of Pig script stored in job xml + * Restricts the size of Pig script stored in job xml */ - public static final int MAX_SCRIPT_SIZE = 10240; - + public static final int MAX_SCRIPT_SIZE = 10240; + private static final Log LOG = LogFactory.getLog(ScriptState.class); private static ThreadLocal tss = new ThreadLocal(); - + private String id; - + private String script; private String commandLine; private String fileName; - + private String pigVersion; private String hodoopVersion; - + private long scriptFeatures; - + private PigContext pigContext; - + private Map featureMap = null; private Map aliasMap = new HashMap(); private Map aliasLocationMap = new HashMap(); - + private List listeners = new ArrayList(); - + public static ScriptState start(String commandLine, PigContext pigContext) { ScriptState ss = new ScriptState(UUID.randomUUID().toString()); ss.setCommandLine(commandLine); @@ -201,10 +205,10 @@ public class ScriptState { tss.set(ss); return ss; } - + private ScriptState(String id) { this.id = id; - this.script = ""; + this.script = ""; } public static ScriptState get() { @@ -212,16 +216,16 @@ public class ScriptState { ScriptState.start("", null); } return tss.get(); - } - + } + public void registerListener(PigProgressNotificationListener listener) { listeners.add(listener); } - + public List getAllListeners() { return listeners; } - + public void emitInitialPlanNotification(MROperPlan plan) { for (PigProgressNotificationListener listener: listeners) { try { @@ -327,11 +331,11 @@ public class ScriptState { } setPigFeature(mro, conf); - + setJobParents(mro, conf); } - - public void setScript(File file) { + + public void setScript(File file) { try { setScript(new BufferedReader(new FileReader(file))); } catch (FileNotFoundException e) { @@ -339,19 +343,19 @@ public class ScriptState { } } - public void setScript(String script) { + public void setScript(String script) { if (script == null) return; - + // restrict the size of the script to be stored in job conf script = (script.length() > MAX_SCRIPT_SIZE) ? script.substring(0, MAX_SCRIPT_SIZE) : script; - + // XML parser cann't handle certain characters, including // the control character (). Use Base64 encoding to - // get around this problem + // get around this problem this.script = new String(Base64.encodeBase64(script.getBytes())); } - + public void setScriptFeatures(LogicalPlan plan) { BitSet bs = new BitSet(); try { @@ -359,59 +363,59 @@ public class ScriptState { } catch (FrontendException e) { LOG.warn("unable to get script feature", e); } - scriptFeatures = bitSetToLong(bs); - + scriptFeatures = bitSetToLong(bs); + LOG.info("Pig features used in the script: " + featureLongToString(scriptFeatures)); } - + public String getHadoopVersion() { if (hodoopVersion == null) { hodoopVersion = VersionInfo.getVersion(); } return (hodoopVersion == null) ? "" : hodoopVersion; } - + public String getPigVersion() { if (pigVersion == null) { String findContainingJar = JarManager.findContainingJar(ScriptState.class); if (findContainingJar != null) { - try { - JarFile jar = new JarFile(findContainingJar); - final Manifest manifest = jar.getManifest(); - final Map attrs = manifest.getEntries(); + try { + JarFile jar = new JarFile(findContainingJar); + final Manifest manifest = jar.getManifest(); + final Map attrs = manifest.getEntries(); Attributes attr = attrs.get("org/apache/pig"); pigVersion = attr.getValue("Implementation-Version"); - } catch (Exception e) { - LOG.warn("unable to read pigs manifest file"); - } + } catch (Exception e) { + LOG.warn("unable to read pigs manifest file"); + } } else { LOG.warn("unable to read pigs manifest file. Not running from the Pig jar"); } } return (pigVersion == null) ? "" : pigVersion; } - + public String getFileName() { return fileName; } - + public void setFileName(String fileName) { this.fileName = fileName; } - + String getId() { return id; } - + private String getCommandLine() { return (commandLine == null) ? "" : commandLine; } - + private void setCommandLine(String commandLine) { this.commandLine = commandLine; } - + private String getScript() { return (script == null) ? "" : script; } - + private void setScript(BufferedReader reader) { StringBuilder sb = new StringBuilder(); try { @@ -427,21 +431,21 @@ public class ScriptState { } setScript(sb.toString()); } - + private void setPigFeature(MapReduceOper mro, Configuration conf) { conf.set(PIG_PROPERTY.JOB_FEATURE.toString(), getPigFeature(mro)); if (scriptFeatures != 0) { - conf.set(PIG_PROPERTY.SCRIPT_FEATURES.toString(), + conf.set(PIG_PROPERTY.SCRIPT_FEATURES.toString(), String.valueOf(scriptFeatures)); } - conf.set(PIG_PROPERTY.JOB_ALIAS.toString(), getAlias(mro)); + conf.set(PIG_PROPERTY.JOB_ALIAS.toString(), getAlias(mro)); conf.set(PIG_PROPERTY.JOB_ALIAS_LOCATION.toString(), getAliasLocation(mro)); } - + private void setJobParents(MapReduceOper mro, Configuration conf) { // PigStats maintains a job DAG with the job id being updated - // upon available. Therefore, before a job is submitted, the ids - // of its parent jobs are already available. + // upon available. Therefore, before a job is submitted, the ids + // of its parent jobs are already available. JobGraph jg = PigStats.get().getJobGraph(); JobStats js = null; Iterator iter = jg.iterator(); @@ -465,11 +469,11 @@ public class ScriptState { } } } - + String getScriptFeatures() { return featureLongToString(scriptFeatures); } - + public String getAlias(MapReduceOper mro) { if (!aliasMap.containsKey(mro)) { setAlias(mro); @@ -514,32 +518,32 @@ public class ScriptState { if (featureMap == null) { featureMap = new HashMap(); } - + String retStr = featureMap.get(mro); - if (retStr == null) { + if (retStr == null) { BitSet feature = new BitSet(); feature.clear(); if (mro.isSkewedJoin()) { feature.set(PIG_FEATURE.SKEWED_JOIN.ordinal()); - } + } if (mro.isGlobalSort()) { feature.set(PIG_FEATURE.ORDER_BY.ordinal()); - } - if (mro.isSampler()) { + } + if (mro.isSampler()) { feature.set(PIG_FEATURE.SAMPLER.ordinal()); - } - if (mro.isIndexer()) { + } + if (mro.isIndexer()) { feature.set(PIG_FEATURE.INDEXER.ordinal()); } if (mro.isCogroup()) { feature.set(PIG_FEATURE.COGROUP.ordinal()); - } + } if (mro.isGroupBy()) { feature.set(PIG_FEATURE.GROUP_BY.ordinal()); - } + } if (mro.isRegularJoin()) { feature.set(PIG_FEATURE.HASH_JOIN.ordinal()); - } + } if (mro.needsDistinctCombiner()) { feature.set(PIG_FEATURE.DISTINCT.ordinal()); } @@ -570,8 +574,8 @@ public class ScriptState { featureMap.put(mro, retStr); } return retStr; - } - + } + private long bitSetToLong(BitSet bs) { long ret = 0; for (int i = bs.nextSetBit(0); i >= 0; i = bs.nextSetBit(i+1)) { @@ -579,10 +583,10 @@ public class ScriptState { } return ret; } - + private String featureLongToString(long l) { if (l == 0) return PIG_FEATURE.UNKNOWN.name(); - + StringBuilder sb = new StringBuilder(); for (int i=0; i> i) & 0x00000001) != 0) { @@ -592,7 +596,7 @@ public class ScriptState { } return sb.toString(); } - + public void setPigContext(PigContext pigContext) { this.pigContext = pigContext; } @@ -603,18 +607,18 @@ public class ScriptState { private static class FeatureVisitor extends PhyPlanVisitor { private BitSet feature; - + public FeatureVisitor(PhysicalPlan plan, BitSet feature) { super(plan, new DepthFirstWalker( plan)); this.feature = feature; } - + @Override public void visitFRJoin(POFRJoin join) throws VisitorException { feature.set(PIG_FEATURE.REPLICATED_JOIN.ordinal()); } - + @Override public void visitMergeJoin(POMergeJoin join) throws VisitorException { if (join.getJoinType()==LOJoin.JOINTYPE.MERGESPARSE) @@ -622,55 +626,55 @@ public class ScriptState { else feature.set(PIG_FEATURE.MERGE_JOIN.ordinal()); } - + @Override public void visitMergeCoGroup(POMergeCogroup mergeCoGrp) throws VisitorException { feature.set(PIG_FEATURE.MERGE_COGROUP.ordinal());; } - + @Override public void visitCollectedGroup(POCollectedGroup mg) - throws VisitorException { + throws VisitorException { feature.set(PIG_FEATURE.COLLECTED_GROUP.ordinal()); } - + @Override public void visitDistinct(PODistinct distinct) throws VisitorException { feature.set(PIG_FEATURE.DISTINCT.ordinal()); } - + @Override public void visitStream(POStream stream) throws VisitorException { feature.set(PIG_FEATURE.STREAMING.ordinal()); } - + @Override public void visitSplit(POSplit split) throws VisitorException { feature.set(PIG_FEATURE.MULTI_QUERY.ordinal()); } - + @Override public void visitDemux(PODemux demux) throws VisitorException { - feature.set(PIG_FEATURE.MULTI_QUERY.ordinal()); + feature.set(PIG_FEATURE.MULTI_QUERY.ordinal()); } - + @Override public void visitPartialAgg(POPartialAgg partAgg){ feature.set(PIG_FEATURE.MAP_PARTIALAGG.ordinal()); } - - } - + + } + static class LogicalPlanFeatureVisitor extends LogicalRelationalNodesVisitor { - + private BitSet feature; - + protected LogicalPlanFeatureVisitor(LogicalPlan plan, BitSet feature) throws FrontendException { - super(plan, new org.apache.pig.newplan.DepthFirstWalker(plan)); + super(plan, new org.apache.pig.newplan.DepthFirstWalker(plan)); this.feature = feature; } - + @Override public void visit(LOCogroup op) { if (op.getGroupType() == GROUPTYPE.COLLECTED) { @@ -685,27 +689,27 @@ public class ScriptState { } } } - + @Override public void visit(LOCross op) { feature.set(PIG_FEATURE.CROSS.ordinal()); } - + @Override public void visit(LODistinct op) { feature.set(PIG_FEATURE.DISTINCT.ordinal()); } - + @Override public void visit(LOFilter op) { feature.set(PIG_FEATURE.FILTER.ordinal()); } - + @Override public void visit(LOForEach op) { - + } - + @Override public void visit(LOJoin op) { if (op.getJoinType() == JOINTYPE.HASH) { @@ -720,45 +724,49 @@ public class ScriptState { feature.set(PIG_FEATURE.SKEWED_JOIN.ordinal()); } } - + @Override public void visit(LOLimit op) { feature.set(PIG_FEATURE.LIMIT.ordinal()); } - + @Override + public void visit(LORank op) { + feature.set(PIG_FEATURE.RANK.ordinal()); + } + public void visit(LOSort op) { feature.set(PIG_FEATURE.ORDER_BY.ordinal()); } - + @Override public void visit(LOStream op) { feature.set(PIG_FEATURE.STREAMING.ordinal()); } - + @Override public void visit(LOSplit op) { - + } - + @Override public void visit(LOUnion op) { feature.set(PIG_FEATURE.UNION.ordinal()); } - + @Override public void visit(LONative n) { feature.set(PIG_FEATURE.NATIVE.ordinal()); } } - + private static class AliasVisitor extends PhyPlanVisitor { - + private HashSet aliasSet; - + private List alias; - + private final List aliasLocation; public AliasVisitor(PhysicalPlan plan, List alias, List aliasLocation) { @@ -771,69 +779,69 @@ public class ScriptState { for (String s : alias) aliasSet.add(s); } } - + @Override public void visitLoad(POLoad load) throws VisitorException { setAlias(load); super.visitLoad(load); } - + @Override public void visitFRJoin(POFRJoin join) throws VisitorException { setAlias(join); super.visitFRJoin(join); } - + @Override public void visitMergeJoin(POMergeJoin join) throws VisitorException { setAlias(join); super.visitMergeJoin(join); } - + @Override public void visitMergeCoGroup(POMergeCogroup mergeCoGrp) throws VisitorException { setAlias(mergeCoGrp); super.visitMergeCoGroup(mergeCoGrp); } - + @Override public void visitCollectedGroup(POCollectedGroup mg) - throws VisitorException { + throws VisitorException { setAlias(mg); super.visitCollectedGroup(mg); } - + @Override public void visitDistinct(PODistinct distinct) throws VisitorException { setAlias(distinct); super.visitDistinct(distinct); } - + @Override public void visitStream(POStream stream) throws VisitorException { setAlias(stream); super.visitStream(stream); } - + @Override public void visitFilter(POFilter fl) throws VisitorException { setAlias(fl); super.visitFilter(fl); } - + @Override public void visitLocalRearrange(POLocalRearrange lr) throws VisitorException { setAlias(lr); super.visitLocalRearrange(lr); } - + @Override public void visitPOForEach(POForEach nfe) throws VisitorException { setAlias(nfe); super.visitPOForEach(nfe); } - + @Override public void visitUnion(POUnion un) throws VisitorException { setAlias(un); @@ -845,19 +853,19 @@ public class ScriptState { setAlias(sort); super.visitSort(sort); } - + @Override public void visitLimit(POLimit lim) throws VisitorException { setAlias(lim); super.visitLimit(lim); } - + @Override public void visitSkewedJoin(POSkewedJoin sk) throws VisitorException { setAlias(sk); super.visitSkewedJoin(sk); } - + private void setAlias(PhysicalOperator op) { String s = op.getAlias(); if (s != null) { @@ -872,5 +880,5 @@ public class ScriptState { } } } - + } Modified: pig/trunk/test/e2e/pig/deployers/ExistingClusterDeployer.pm URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/deployers/ExistingClusterDeployer.pm?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/e2e/pig/deployers/ExistingClusterDeployer.pm (original) +++ pig/trunk/test/e2e/pig/deployers/ExistingClusterDeployer.pm Thu Sep 13 14:55:36 2012 @@ -228,6 +228,16 @@ sub generateData 'filetype' => "numbers", 'rows' => 5000, 'hdfs' => "types/numbers.txt", + }, { + 'name' => "biggish", + 'filetype' => "biggish", + 'rows' => 1000000, + 'hdfs' => "singlefile/biggish", + }, { + 'name' => "prerank", + 'filetype' => "ranking", + 'rows' => 30, + 'hdfs' => "singlefile/prerank", } ); Modified: pig/trunk/test/e2e/pig/deployers/LocalDeployer.pm URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/deployers/LocalDeployer.pm?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/e2e/pig/deployers/LocalDeployer.pm (original) +++ pig/trunk/test/e2e/pig/deployers/LocalDeployer.pm Thu Sep 13 14:55:36 2012 @@ -204,6 +204,16 @@ sub generateData 'filetype' => "numbers", 'rows' => 5000, 'outfile' => "types/numbers.txt", + }, { + 'name' => "biggish", + 'filetype' => "biggish", + 'rows' => 1000000, + 'outfile' => "singlefile/biggish", + }, { + 'name' => "prerank", + 'filetype' => "ranking", + 'rows' => 30, + 'outfile' => "singlefile/prerank", } ); Modified: pig/trunk/test/e2e/pig/tests/nightly.conf URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/tests/nightly.conf?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/e2e/pig/tests/nightly.conf (original) +++ pig/trunk/test/e2e/pig/tests/nightly.conf Thu Sep 13 14:55:36 2012 @@ -4974,6 +4974,173 @@ store a into ':OUTPATH:';\, store F into ':OUTPATH:';\, } ] + }, + { + 'name' => 'Rank', + 'tests' => [ + { + 'num' => 1, + 'pig' => q\ + SET default_parallel 9; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = rank A by a ASC,b ASC DENSE; + C = foreach B generate rank_A,a,b; + store C into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rankaaba,a,b; + store C into ':OUTPATH:'; + \, + }, { + 'num' => 2, + 'pig' => q\ + SET default_parallel 9; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = rank A by a ASC,c DESC DENSE; + C = foreach B generate rank_A,a,c; + store C into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rankaacd,a,c; + store C into ':OUTPATH:'; + \, + }, { + 'num' => 3, + 'pig' => q\ + SET default_parallel 7; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = rank A by b DESC,c ASC DENSE; + C = foreach B generate rank_A,b,c; + store C into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rankbdca,b,c; + store C into ':OUTPATH:'; + \, + }, { + 'num' => 4, + 'pig' => q\ + SET default_parallel 7; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = rank A; + C = foreach B generate rank_A,a,b,c; + store C into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rownumber,a,b,c; + store C into ':OUTPATH:'; + \, + }, { + 'num' =>5, + 'pig' => q\ + SET default_parallel 9; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = rank A by b DESC,a ASC; + C = foreach B generate rank_A,b,a; + store C into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rankbdaa,b,a; + store C into ':OUTPATH:'; + \, + }, { + 'num' =>6, + 'pig' => q\ + SET default_parallel 7; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = rank A by c ASC,b DESC; + C = foreach B generate rank_A,c,b; + store C into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rankcabd,c,b; + store C into ':OUTPATH:'; + \, + }, { + 'num' => 7, + 'pig' => q\ + SET default_parallel 7; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = foreach A generate a,b,c,tail; + C = rank B by a ASC,b ASC DENSE; + D = rank C by a ASC,c DESC DENSE; + E = rank D by b DESC,c ASC DENSE; + F = foreach E generate rank_D,rank_C,rank_B,a,b,c; + store F into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = foreach A generate rankbdca,rankaacd,rankaaba,a,b,c; + store B into ':OUTPATH:'; + \, + }, { + 'num' => 8, + 'pig' => q\ + SET default_parallel 9; + SET pig.splitCombination false; + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + B = foreach A generate a,b,c; + C = rank B by a ASC,b ASC DENSE; + D = rank B by a ASC,c DESC DENSE; + F = join C by $0, D by $0; + G = foreach F generate C::rank_B, D::rank_B, C::a, C::b, C::c; + H = order G by a ASC, b ASC, c DESC; + store H into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = foreach A generate rankaaba,a,b,c; + E = order C by a ASC,b ASC; + D = foreach A generate rankaacd,a,b,c; + F = order D by a ASC,c DESC; + G = join E by $0, F by $0; + H = foreach G generate E::rankaaba, F::rankaacd, E::a, E::b, E::c; + store H into ':OUTPATH:'; + \, + }, { + 'num' => 9, + 'pig' => q\ + SET default_parallel 25; + A = LOAD ':INPATH:/singlefile/biggish' using PigStorage(',') as (rownumber:long,idx:long,tail:bytearray); + B = rank A; + C = order B by rank_A; + D = foreach C generate rank_A,rownumber; + store D into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/biggish' using PigStorage(',') as (rownumber:long,idx:long,tail:bytearray); + D = foreach A generate idx,rownumber; + store D into ':OUTPATH:'; + \, + }, { + 'num' => 10, + 'pig' => q\ + SET default_parallel 11; + SET pig.splitCombination false; + A = LOAD ':INPATH:/singlefile/biggish' using PigStorage(',') as (rownumber:long,idx:long,tail:bytearray); + B = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,rankcabd:long,rankbdaa:long,rankbdca:long,rankaacd:long,rankaaba:long,a:int,b:int,c:int,tail:bytearray); + C = join A by rownumber, B by rownumber; + D = order C by B::rankcabd,B::rankbdca,B::rankaaba; + E = rank D; + F = group E by rank_D; + G = foreach F generate group, COUNT(E); + H = order G by group; + store H into ':OUTPATH:'; + \, + 'verify_pig_script' => q\ + A = LOAD ':INPATH:/singlefile/prerank' using PigStorage(',') as (rownumber:long,idx:long,tail:bytearray); + B = foreach A generate rownumber,1; + C = order B by rownumber; + store C into ':OUTPATH:'; + \, + } + ] } ], }, Modified: pig/trunk/test/e2e/pig/tools/generate/generate_data.pl URL: http://svn.apache.org/viewvc/pig/trunk/test/e2e/pig/tools/generate/generate_data.pl?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/e2e/pig/tools/generate/generate_data.pl (original) +++ pig/trunk/test/e2e/pig/tools/generate/generate_data.pl Thu Sep 13 14:55:36 2012 @@ -32,6 +32,16 @@ our @lastName = ("allen", "brown", "cars "nixon", "ovid", "polk", "quirinius", "robinson", "steinbeck", "thompson", "underhill", "van buren", "white", "xylophone", "young", "zipper"); +our @rankedTuples = ( + "1,21,5,7,1,1,0,8,8","2,26,2,3,2,5,1,9,10","3,30,24,21,2,3,1,3,10","4,6,10,8,3,4,1,7,2", + "5,8,28,25,3,2,1,0,2","6,28,11,12,4,6,2,7,10","7,9,26,22,5,7,3,2,3","8,5,6,5,6,8,3,8,1", + "9,29,16,15,7,9,4,6,10","10,18,12,10,8,11,5,7,6","11,14,17,14,9,10,5,6,5","12,6,12,8,10,11,5,7,2", + "13,2,17,13,11,10,5,6,0","14,26,3,3,12,14,6,9,10","15,15,20,18,13,13,6,4,5","16,3,29,24,14,12,6,0,0", + "17,23,21,19,15,16,7,4,8","18,19,19,16,16,17,7,5,6","19,20,30,26,16,15,7,0,6","20,12,21,17,17,16,7,4,4", + "21,4,1,1,18,19,7,10,1","22,1,7,4,19,18,7,8,0","23,24,14,11,20,21,8,7,9","24,16,25,20,21,20,8,3,5", + "25,25,27,23,22,22,9,1,9","26,21,8,7,23,25,9,8,8","27,17,4,2,24,26,9,9,6","28,10,8,6,25,25,9,8,4", + "29,11,15,9,25,24,9,7,4","30,12,23,17,25,23,9,4,4"); + sub randomName() { return sprintf("%s %s", $firstName[int(rand(26))], @@ -473,6 +483,23 @@ sub getBulkCopyCmd(){ my $randf = rand(10); printf HDFS "%d:%d:%d:%d:%d:%dL:%.2ff:%.2f\n", $tid, $i, $rand5, $rand100, $rand1000, $rand1000, $randf, $randf; } + } elsif ($filetype eq "ranking") { + for (my $i = 0; $i < $numRows; $i++) { + my $tuple = $rankedTuples[int($i)]; + printf HDFS "$tuple,"; + for my $j ( 0 .. 1000000) { + printf HDFS "%d",$j; + } + printf HDFS "\n"; + } + } elsif ($filetype eq "biggish") { + for (my $i = 1; $i < $numRows; $i++) { + printf HDFS "$i,$i,"; + for my $j ( 0 .. 1000) { + printf HDFS "%d",$j; + } + printf HDFS "\n"; + } } else { warn "Unknown filetype $filetype\n"; usage(); Modified: pig/trunk/test/org/apache/pig/parser/TestLexer.pig URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/parser/TestLexer.pig?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/org/apache/pig/parser/TestLexer.pig (original) +++ pig/trunk/test/org/apache/pig/parser/TestLexer.pig Thu Sep 13 14:55:36 2012 @@ -73,7 +73,13 @@ A = LOAD 'data' AS (f1:int,f2:int,f3:int X = SAMPLE A 0.01; +R = rank A by f1; +R = rank A by f1 ASC, f2 DESC, f3; + +R = rank A by *; + +R = rank A by * DESC DENSE; Modified: pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java (original) +++ pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java Thu Sep 13 14:55:36 2012 @@ -411,4 +411,61 @@ public class TestLogicalPlanGenerator { generateLogicalPlan( query ); } + @Test + public void testRank01() { + String query = "A = LOAD 'data4' AS (name:chararray,surname:chararray,sales:double,code:int);" + + "B = rank A by sales;" + "store B into 'rank01_test';"; + generateLogicalPlan(query); + } + + @Test + public void testRank02() { + String query = "A = LOAD 'data4' AS (name:chararray,surname:chararray,sales:double,code:int);" + + "C = rank A by sales DENSE;" + "store C into 'rank02_test';"; + generateLogicalPlan(query); + } + + @Test + public void testRank03() { + String query = "A = load 'test02' using PigStorage(',') as (firstname:chararray,lastname:chararray,rownumberPrev:int,rankPrev:int,denserankPrev:int,quartilePrev:int,sales:double,postalcode:int);" + + "B = rank A;" + "store B into 'rank03_test';"; + generateLogicalPlan(query); + } + + @Test + public void testRank04() { + String query = "A = load 'test02' using PigStorage(',') as (firstname:chararray,lastname:chararray,rownumberPrev:int,rankPrev:int,denserankPrev:int,quartilePrev:int,sales:double,postalcode:int);" + + "C = rank A by postalcode DESC;" + + "store C into 'rank04_test';"; + generateLogicalPlan(query); + } + + @Test + public void testRank05() { + String query = "A = load 'test02' using PigStorage(',') as (firstname:chararray,lastname:chararray,rownumberPrev:int,rankPrev:int,denserankPrev:int,quartilePrev:int,sales:double,postalcode:int);" + + "D = rank A by postalcode DENSE;" + + "store D into 'rank05_test';"; + generateLogicalPlan(query); + } + + @Test + public void testRank06() { + String query = "A = load 'data' as (x:int,y:chararray,z:int,rz:chararray);" + + "C = rank A by x..rz;"; + generateLogicalPlan(query); + } + + @Test + public void testRank07() { + String query = "A = load 'data' as (x:int,y:chararray,z:int,rz:chararray);" + + "C = rank A by x ASC, y DESC;"; + generateLogicalPlan(query); + } + + @Test + public void testRank08() { + String query = "A = load 'data' as (x:int,y:chararray,z:int,rz:chararray);" + + "C = rank A;"; + generateLogicalPlan(query); + } } Modified: pig/trunk/test/org/apache/pig/parser/TestParser.pig URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/parser/TestParser.pig?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/org/apache/pig/parser/TestParser.pig (original) +++ pig/trunk/test/org/apache/pig/parser/TestParser.pig Thu Sep 13 14:55:36 2012 @@ -90,3 +90,11 @@ H = union onschema A, B; --stream C = stream A through CMD; + + +--rank + +R = rank A; +R = rank A by a; +R = rank A by a DESC; +R = rank A by a DESC, b; \ No newline at end of file Modified: pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java (original) +++ pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java Thu Sep 13 14:55:36 2012 @@ -28,7 +28,7 @@ import org.antlr.runtime.Token; import org.junit.Test; public class TestQueryLexer { - + @Test public void TestLexer() throws IOException { CharStream input = new QueryParserFileStream( "test/org/apache/pig/parser/TestLexer.pig" ); @@ -45,13 +45,13 @@ public class TestQueryLexer { System.out.print( token.getText() + "(" + token.getType() + ") " ); } } - + // While we can check more conditions, such as type of each token, for now I think the following // is enough. If the token type is wrong, it will be most likely caught by the parser. - Assert.assertEquals( 419, tokenCount ); + Assert.assertEquals( 455, tokenCount ); Assert.assertEquals( 0, lexer.getNumberOfSyntaxErrors() ); } - + @Test public void test2() throws IOException { String query = "A = load 'input' using PigStorage(';');" + Modified: pig/trunk/test/org/apache/pig/parser/TestQueryParser.java URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/parser/TestQueryParser.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/org/apache/pig/parser/TestQueryParser.java (original) +++ pig/trunk/test/org/apache/pig/parser/TestQueryParser.java Thu Sep 13 14:55:36 2012 @@ -26,6 +26,7 @@ import junit.framework.Assert; import org.antlr.runtime.CharStream; import org.antlr.runtime.CommonTokenStream; +import org.antlr.runtime.MismatchedTokenException; import org.antlr.runtime.RecognitionException; import org.antlr.runtime.tree.CommonTree; import org.antlr.runtime.tree.Tree; @@ -415,4 +416,61 @@ public class TestQueryParser { PigServer pig = new PigServer(ExecType.LOCAL); Util.registerMultiLineQuery(pig, query); } + + + //RANK + @Test + public void testRankPositive1() throws IOException, RecognitionException { + shouldPass("B = rank A;"); + } + + @Test + public void testRankPositive2() throws IOException, RecognitionException { + shouldPass("B = rank A by x;"); + } + + @Test + public void testRankPositive3() throws IOException, RecognitionException { + shouldPass("B = rank A by x DESC;"); + } + + @Test + public void testRankPositive4() throws IOException, RecognitionException { + shouldPass("B = rank A by x, y ASC, w DESC, z ASC;"); + } + + @Test + public void testRankPositive5() throws IOException, RecognitionException { + String query = "A = load 'data' as (x:int, y:chararray, z:int, rz:chararray);"; + query += "B = rank A by x..z;"; + shouldPass(query); + } + + @Test + public void testRankPositive6() throws IOException, RecognitionException { + String query = "A = load 'data' as (x:int, y:chararray, z:int, rz:chararray);"; + query += "B = rank A by *;"; + shouldPass(query); + } + + @Test + public void testRankPositive7() throws IOException, RecognitionException { + String query = "A = load 'data' as (x:int, y:chararray, z:int, rz:chararray);"; + query += "B = rank A by x DESC DENSE;"; + shouldPass(query); + } + + @Test + public void testRankPositive8() throws IOException, RecognitionException { + String query = "A = load 'data' as (x:int, y:chararray, z:int, rz:chararray);"; + query += "B = rank A by x DESC,y ASC DENSE;"; + shouldPass(query); + } + + @Test + public void testRankPositive9() throws IOException, RecognitionException { + String query = "A = load 'data' as (x:int, y:chararray, z:int, rz:chararray);"; + query += "B = rank A by * DENSE;"; + shouldPass(query); + } } Modified: pig/trunk/test/org/apache/pig/test/OptimizeLimitPlanPrinter.java URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/OptimizeLimitPlanPrinter.java?rev=1384352&r1=1384351&r2=1384352&view=diff ============================================================================== --- pig/trunk/test/org/apache/pig/test/OptimizeLimitPlanPrinter.java (original) +++ pig/trunk/test/org/apache/pig/test/OptimizeLimitPlanPrinter.java Thu Sep 13 14:55:36 2012 @@ -57,6 +57,7 @@ import org.apache.pig.newplan.logical.re import org.apache.pig.newplan.logical.relational.LOJoin; import org.apache.pig.newplan.logical.relational.LOLimit; import org.apache.pig.newplan.logical.relational.LOLoad; +import org.apache.pig.newplan.logical.relational.LORank; import org.apache.pig.newplan.logical.relational.LOSort; import org.apache.pig.newplan.logical.relational.LOSplit; import org.apache.pig.newplan.logical.relational.LOSplitOutput; @@ -179,12 +180,18 @@ public class OptimizeLimitPlanPrinter ex sb.append(";\n"); appendEdges(loSort); } - + + @Override + public void visit(LORank loRank) throws FrontendException { + appendOp(loRank) ; + appendEdges(loRank); + } + @Override public void visit(LODistinct loDistinct) throws FrontendException { appendEdges(loDistinct); } - + @Override public void visit(LOLimit loLimit) throws FrontendException { appendOp(loLimit) ;