You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by dv...@apache.org on 2012/05/26 23:37:57 UTC
svn commit: r1342975 - in /pig/trunk: ./
src/org/apache/pig/newplan/logical/optimizer/
src/org/apache/pig/newplan/logical/relational/
src/org/apache/pig/newplan/logical/visitor/ src/org/apache/pig/parser/
test/org/apache/pig/parser/ test/org/apache/pig...
Author: dvryaboy
Date: Sat May 26 21:37:56 2012
New Revision: 1342975
URL: http://svn.apache.org/viewvc?rev=1342975&view=rev
Log:
PIG-2710: Implement Naive CUBE operator
Added:
pig/trunk/src/org/apache/pig/newplan/logical/relational/LOCube.java
pig/trunk/test/org/apache/pig/test/TestCubeOperator.java
pig/trunk/test/org/apache/pig/test/data/output26.pig
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/newplan/logical/optimizer/AllExpressionVisitor.java
pig/trunk/src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java
pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java
pig/trunk/src/org/apache/pig/newplan/logical/visitor/ProjectStarExpander.java
pig/trunk/src/org/apache/pig/parser/AliasMasker.g
pig/trunk/src/org/apache/pig/parser/AstPrinter.g
pig/trunk/src/org/apache/pig/parser/AstValidator.g
pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java
pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g
pig/trunk/src/org/apache/pig/parser/QueryLexer.g
pig/trunk/src/org/apache/pig/parser/QueryParser.g
pig/trunk/test/org/apache/pig/parser/TestLexer.pig
pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java
pig/trunk/test/org/apache/pig/parser/TestParser.pig
pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java
pig/trunk/test/org/apache/pig/parser/TestQueryParser.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Sat May 26 21:37:56 2012
@@ -24,6 +24,8 @@ INCOMPATIBLE CHANGES
IMPROVEMENTS
+PIG-2710: Implement Naive CUBE operator (prasanth_j via dvryaboy)
+
PIG-2714: Pig documentation on TOP funcation has issues (daijy)
PIG-2066: Accumulators should be able to early-terminate (jcoveney)
Modified: pig/trunk/src/org/apache/pig/newplan/logical/optimizer/AllExpressionVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/AllExpressionVisitor.java?rev=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/optimizer/AllExpressionVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/optimizer/AllExpressionVisitor.java Sat May 26 21:37:56 2012
@@ -27,6 +27,7 @@ import org.apache.pig.newplan.PlanWalker
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor;
import org.apache.pig.newplan.logical.relational.LOCogroup;
+import org.apache.pig.newplan.logical.relational.LOCube;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOGenerate;
@@ -64,6 +65,13 @@ public abstract class AllExpressionVisit
*/
abstract protected LogicalExpressionVisitor getVisitor(LogicalExpressionPlan expr) throws FrontendException;
+ private void visitAll(Collection<LogicalExpressionPlan> lexpPlans) throws FrontendException {
+ for (LogicalExpressionPlan plan : lexpPlans) {
+ LogicalExpressionVisitor v = getVisitor(plan);
+ v.visit();
+ }
+ }
+
@Override
public void visit(LOFilter filter) throws FrontendException {
currentOp = filter;
@@ -83,10 +91,15 @@ public abstract class AllExpressionVisit
@Override
public void visit(LOJoin join) throws FrontendException {
currentOp = join;
- Collection<LogicalExpressionPlan> c = join.getExpressionPlanValues();
- for (LogicalExpressionPlan plan : c) {
- LogicalExpressionVisitor v = getVisitor(plan);
- v.visit();
+ visitAll(join.getExpressionPlanValues());
+ }
+
+ @Override
+ public void visit(LOCube cu) throws FrontendException {
+ currentOp = cu;
+ MultiMap<Integer, LogicalExpressionPlan> expressionPlans = cu.getExpressionPlans();
+ for (Integer key : expressionPlans.keySet()) {
+ visitAll(expressionPlans.get(key));
}
}
@@ -95,11 +108,7 @@ public abstract class AllExpressionVisit
currentOp = cg;
MultiMap<Integer, LogicalExpressionPlan> expressionPlans = cg.getExpressionPlans();
for( Integer key : expressionPlans.keySet() ) {
- Collection<LogicalExpressionPlan> exprPlans = expressionPlans.get(key);
- for( LogicalExpressionPlan plan : exprPlans ) {
- LogicalExpressionVisitor v = getVisitor(plan);
- v.visit();
- }
+ visitAll(expressionPlans.get(key));
}
}
@@ -118,11 +127,7 @@ public abstract class AllExpressionVisit
@Override
public void visit(LOGenerate gen ) throws FrontendException {
currentOp = gen;
- Collection<LogicalExpressionPlan> plans = gen.getOutputPlans();
- for( LogicalExpressionPlan plan : plans ) {
- LogicalExpressionVisitor v = getVisitor(plan);
- v.visit();
- }
+ visitAll(gen.getOutputPlans());
}
@Override
@@ -146,10 +151,6 @@ public abstract class AllExpressionVisit
@Override
public void visit(LOSort sort) throws FrontendException {
currentOp = sort;
- Collection<LogicalExpressionPlan> c = sort.getSortColPlans();
- for (LogicalExpressionPlan plan : c) {
- LogicalExpressionVisitor v = getVisitor(plan);
- v.visit();
- }
+ visitAll(sort.getSortColPlans());
}
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java?rev=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java Sat May 26 21:37:56 2012
@@ -35,6 +35,7 @@ import org.apache.pig.newplan.logical.ex
import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
import org.apache.pig.newplan.logical.relational.LOCogroup;
import org.apache.pig.newplan.logical.relational.LOCross;
+import org.apache.pig.newplan.logical.relational.LOCube;
import org.apache.pig.newplan.logical.relational.LODistinct;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
@@ -59,6 +60,13 @@ public class SchemaResetter extends Logi
// DuplicateForEachColumnRewrite has run. So disable it in calls before that
boolean skipDuplicateUidCheck = true;
+ private void visitAll(Collection<LogicalExpressionPlan> lexpPlans) throws FrontendException {
+ for (LogicalExpressionPlan expPlan : lexpPlans) {
+ FieldSchemaResetter fsResetter = new FieldSchemaResetter(expPlan);
+ fsResetter.visit();
+ }
+ }
+
public SchemaResetter(OperatorPlan plan) throws FrontendException {
this(plan, false);
}
@@ -92,11 +100,7 @@ public class SchemaResetter extends Logi
@Override
public void visit(LOJoin join) throws FrontendException {
join.resetSchema();
- Collection<LogicalExpressionPlan> joinPlans = join.getExpressionPlanValues();
- for (LogicalExpressionPlan joinPlan : joinPlans) {
- FieldSchemaResetter fsResetter = new FieldSchemaResetter(joinPlan);
- fsResetter.visit();
- }
+ visitAll(join.getExpressionPlanValues());
validate(join.getSchema());
}
@@ -114,11 +118,7 @@ public class SchemaResetter extends Logi
@Override
public void visit(LOGenerate gen) throws FrontendException {
gen.resetSchema();
- List<LogicalExpressionPlan> genPlans = gen.getOutputPlans();
- for (LogicalExpressionPlan genPlan : genPlans) {
- FieldSchemaResetter fsResetter = new FieldSchemaResetter(genPlan);
- fsResetter.visit();
- }
+ visitAll(gen.getOutputPlans());
validate(gen.getSchema());
}
@@ -130,13 +130,16 @@ public class SchemaResetter extends Logi
}
@Override
+ public void visit(LOCube loCube) throws FrontendException {
+ loCube.resetSchema();
+ visitAll(loCube.getExpressionPlans().values());
+ validate(loCube.getSchema());
+ }
+
+ @Override
public void visit(LOCogroup loCogroup) throws FrontendException {
loCogroup.resetSchema();
- MultiMap<Integer, LogicalExpressionPlan> expPlans = loCogroup.getExpressionPlans();
- for (LogicalExpressionPlan expPlan : expPlans.values()) {
- FieldSchemaResetter fsResetter = new FieldSchemaResetter(expPlan);
- fsResetter.visit();
- }
+ visitAll(loCogroup.getExpressionPlans().values());
validate(loCogroup.getSchema());
}
@@ -163,11 +166,7 @@ public class SchemaResetter extends Logi
@Override
public void visit(LOSort loSort) throws FrontendException {
loSort.resetSchema();
- List<LogicalExpressionPlan> sortPlans = loSort.getSortColPlans();
- for (LogicalExpressionPlan sortPlan : sortPlans) {
- FieldSchemaResetter fsResetter = new FieldSchemaResetter(sortPlan);
- fsResetter.visit();
- }
+ visitAll(loSort.getSortColPlans());
validate(loSort.getSchema());
}
Added: pig/trunk/src/org/apache/pig/newplan/logical/relational/LOCube.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOCube.java?rev=1342975&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LOCube.java (added)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LOCube.java Sat May 26 21:37:56 2012
@@ -0,0 +1,124 @@
+/*
+ * 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.newplan.logical.relational;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.util.MultiMap;
+import org.apache.pig.newplan.Operator;
+import org.apache.pig.newplan.OperatorPlan;
+import org.apache.pig.newplan.PlanVisitor;
+import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
+
+/**
+ * CUBE operator implementation for data cube computation.
+ * <p>
+ * Cube operator syntax
+ * <pre>{@code alias = CUBE rel BY (col_ref);}
+ * alias - output alias
+ * CUBE - operator
+ * rel - input relation
+ * BY - operator
+ * col_ref - column references or * or range in the schema referred by rel
+ * </pre> </p>
+ * <p>
+ * The cube computation code sample at {@link org.apache.pig.builtin.CubeDimensions}
+ * can now be represented like below
+ * <pre>{@code
+ * events = LOAD '/logs/events' USING EventLoader() AS (lang, event, app_id, total);
+ * eventcube = CUBE events BY (lang, event, app_id);
+ * result = FOREACH eventcube GENERATE FLATTEN(group) as (lang, event, app_id),
+ * COUNT_STAR(cube), SUM(cube.total);
+ * STORE result INTO 'cuberesult';
+ * }</pre>
+ * </p>
+ */
+public class LOCube extends LogicalRelationalOperator {
+ private MultiMap<Integer, LogicalExpressionPlan> mExpressionPlans;
+
+ public LOCube(LogicalPlan plan) {
+ super("LOCube", plan);
+ }
+
+ public LOCube(OperatorPlan plan,
+ MultiMap<Integer, LogicalExpressionPlan> expressionPlans) {
+ super("LOCube", plan);
+ this.mExpressionPlans = expressionPlans;
+ }
+
+ @Override
+ public LogicalSchema getSchema() throws FrontendException {
+ // TODO: implement when physical operator for CUBE is implemented
+ return null;
+ }
+
+ @Override
+ public void accept(PlanVisitor v) throws FrontendException {
+ try {
+ ((LogicalRelationalNodesVisitor) v).visit(this);
+ } catch (ClassCastException cce) {
+ throw new FrontendException("Expected LogicalPlanVisitor", cce);
+ }
+ }
+
+ @Override
+ public boolean isEqual(Operator other) throws FrontendException {
+ try {
+ LOCube cube = (LOCube) other;
+ for (Integer key : mExpressionPlans.keySet()) {
+ if (!cube.mExpressionPlans.containsKey(key)) {
+ return false;
+ }
+ Collection<LogicalExpressionPlan> lepList1 = mExpressionPlans.get(key);
+ Collection<LogicalExpressionPlan> lepList2 = cube.mExpressionPlans.get(key);
+
+ for (LogicalExpressionPlan lep1 : lepList1) {
+ for (LogicalExpressionPlan lep2 : lepList2) {
+ if (!lep1.isEqual(lep2)) {
+ return false;
+ }
+ }
+ }
+ }
+ return checkEquality((LogicalRelationalOperator) other);
+ } catch (ClassCastException cce) {
+ throw new FrontendException("Exception while casting CUBE operator", cce);
+ }
+ }
+
+ public MultiMap<Integer, LogicalExpressionPlan> getExpressionPlans() {
+ return mExpressionPlans;
+ }
+
+ public void setExpressionPlans(
+ MultiMap<Integer, LogicalExpressionPlan> plans) {
+ this.mExpressionPlans = plans;
+ }
+
+ @Override
+ public void resetUid() {
+ //TODO: implement when physical operator for CUBE is implemented
+ }
+
+ public List<Operator> getInputs(LogicalPlan plan) {
+ return plan.getPredecessors(this);
+ }
+}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java?rev=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java Sat May 26 21:37:56 2012
@@ -63,6 +63,9 @@ public abstract class LogicalRelationalN
public void visit(LOInnerLoad load) throws FrontendException {
}
+ public void visit(LOCube cube) throws FrontendException {
+ }
+
public void visit(LOCogroup loCogroup) throws FrontendException {
}
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=1342975&r1=1342974&r2=1342975&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 Sat May 26 21:37:56 2012
@@ -37,6 +37,7 @@ import org.apache.pig.newplan.logical.ex
import org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor;
import org.apache.pig.newplan.logical.expression.ProjectExpression;
import org.apache.pig.newplan.logical.relational.LOCogroup;
+import org.apache.pig.newplan.logical.relational.LOCube;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOGenerate;
import org.apache.pig.newplan.logical.relational.LOInnerLoad;
@@ -188,6 +189,15 @@ public class ProjectStarExpander extends
}
+ @Override
+ public void visit(LOCube cu) throws FrontendException {
+
+ MultiMap<Integer, LogicalExpressionPlan> inpExprPlans = cu.getExpressionPlans();
+
+ // modify the plans if they have project-star
+ expandPlans(inpExprPlans);
+
+ }
@Override
public void visit(LOJoin join) throws FrontendException{
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/AliasMasker.g (original)
+++ pig/trunk/src/org/apache/pig/parser/AliasMasker.g Sat May 26 21:37:56 2012
@@ -130,6 +130,7 @@ op_clause : define_clause
| mr_clause
| split_clause
| foreach_clause
+ | cube_clause
;
define_clause
@@ -236,6 +237,22 @@ func_args
: QUOTEDSTRING+
;
+cube_clause
+ : ^( CUBE cube_item )
+;
+
+cube_item
+ : rel ( cube_by_clause )
+;
+
+cube_by_clause
+ : ^( BY cube_by_expr+ )
+;
+
+cube_by_expr
+ : col_range | expr | STAR
+;
+
group_clause
: ^( ( GROUP | COGROUP ) group_item+ group_type? partition_clause? )
;
@@ -333,6 +350,7 @@ col_alias_or_index : col_alias | col_ind
col_alias
: GROUP
+ | CUBE
| IDENTIFIER
;
@@ -507,6 +525,7 @@ col_ref : alias_col_ref | dollar_col_ref
alias_col_ref
: GROUP
+ | CUBE
| IDENTIFIER
{
String alias = $IDENTIFIER.text;
@@ -570,6 +589,7 @@ eid : rel_str_op
| LOAD
| FILTER
| FOREACH
+ | CUBE
| MATCHES
| ORDER
| DISTINCT
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/AstPrinter.g (original)
+++ pig/trunk/src/org/apache/pig/parser/AstPrinter.g Sat May 26 21:37:56 2012
@@ -97,6 +97,7 @@ op_clause : define_clause
| mr_clause
| split_clause
| foreach_clause
+ | cube_clause
;
define_clause
@@ -204,6 +205,23 @@ func_args
(b=QUOTEDSTRING { sb.append(", ").append($b.text); } )*
;
+cube_clause
+ : ^( CUBE { sb.append($CUBE.text).append(" "); } cube_item )
+;
+
+cube_item
+ : rel ( cube_by_clause )
+;
+
+cube_by_clause
+ : ^( BY { sb.append(" ").append($BY.text).append(" ("); }
+ cube_by_expr ( { sb.append(", "); } cube_by_expr )* { sb.append(")"); } )
+;
+
+cube_by_expr
+ : col_range | expr | STAR { sb.append($STAR.text); }
+;
+
group_clause
: ^( ( GROUP { sb.append($GROUP.text).append(" "); } | COGROUP { sb.append($COGROUP.text).append(" "); } )
group_item ( { sb.append(", "); } group_item )*
@@ -304,6 +322,7 @@ col_alias_or_index : col_alias | col_ind
col_alias
: GROUP { sb.append($GROUP.text); }
+ | CUBE { sb.append($CUBE.text); }
| IDENTIFIER { sb.append($IDENTIFIER.text); }
;
@@ -494,6 +513,7 @@ col_ref : alias_col_ref | dollar_col_ref
alias_col_ref
: GROUP { sb.append($GROUP.text); }
+ | CUBE { sb.append($CUBE.text); }
| IDENTIFIER { sb.append($IDENTIFIER.text); }
;
@@ -552,6 +572,7 @@ eid : rel_str_op
| LOAD { sb.append($LOAD.text); }
| FILTER { sb.append($FILTER.text); }
| FOREACH { sb.append($FOREACH.text); }
+ | CUBE { sb.append($CUBE.text); }
| MATCHES { sb.append($MATCHES.text); }
| ORDER { sb.append($ORDER.text); }
| DISTINCT { sb.append($DISTINCT.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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/AstValidator.g (original)
+++ pig/trunk/src/org/apache/pig/parser/AstValidator.g Sat May 26 21:37:56 2012
@@ -145,6 +145,7 @@ op_clause : define_clause
| mr_clause
| split_clause
| foreach_clause
+ | cube_clause
;
define_clause : ^( DEFINE alias ( cmd | func_clause ) )
@@ -255,6 +256,22 @@ func_args_string : QUOTEDSTRING | MULTIL
func_args : func_args_string+
;
+cube_clause
+ : ^( CUBE cube_item )
+;
+
+cube_item
+ : rel ( cube_by_clause )
+;
+
+cube_by_clause
+ : ^( BY cube_by_expr+ )
+;
+
+cube_by_expr
+ : col_range | expr | STAR
+;
+
group_clause
scope {
int arity;
@@ -345,7 +362,7 @@ dot_proj : ^( PERIOD col_alias_or_index+
col_alias_or_index : col_alias | col_index
;
-col_alias : GROUP | IDENTIFIER
+col_alias : GROUP | CUBE | IDENTIFIER
;
col_index : DOLLARVAR
@@ -523,7 +540,7 @@ split_otherwise : ^( OTHERWISE alias )
col_ref : alias_col_ref | dollar_col_ref
;
-alias_col_ref : GROUP | IDENTIFIER
+alias_col_ref : GROUP | CUBE | IDENTIFIER
;
dollar_col_ref : DOLLARVAR
@@ -564,6 +581,7 @@ eid : rel_str_op
| LOAD
| FILTER
| FOREACH
+ | CUBE
| MATCHES
| ORDER
| DISTINCT
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java (original)
+++ pig/trunk/src/org/apache/pig/parser/LogicalPlanBuilder.java Sat May 26 21:37:56 2012
@@ -36,6 +36,7 @@ import org.apache.pig.LoadFunc;
import org.apache.pig.StoreFuncInterface;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.builtin.CubeDimensions;
import org.apache.pig.builtin.PigStorage;
import org.apache.pig.builtin.RANDOM;
import org.apache.pig.data.BagFactory;
@@ -67,6 +68,7 @@ import org.apache.pig.newplan.logical.op
import org.apache.pig.newplan.logical.relational.LOCogroup;
import org.apache.pig.newplan.logical.relational.LOCogroup.GROUPTYPE;
import org.apache.pig.newplan.logical.relational.LOCross;
+import org.apache.pig.newplan.logical.relational.LOCube;
import org.apache.pig.newplan.logical.relational.LODistinct;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
@@ -294,13 +296,7 @@ public class LogicalPlanBuilder {
}
sort.setAscendingCols( ascFlags );
alias = buildOp( loc, sort, alias, inputAlias, null );
- try {
- (new ProjectStarExpander(sort.getPlan())).visit(sort);
- (new ProjStarInUdfExpander(sort.getPlan())).visit(sort);
- new SchemaResetter(sort.getPlan(), true).visit(sort);
- } catch (FrontendException e) {
- throw new ParserValidationException( intStream, loc, e );
- }
+ expandAndResetVisitor(loc, sort);
return alias;
}
@@ -355,16 +351,256 @@ public class LogicalPlanBuilder {
op.setInnerFlags( flags );
op.setJoinPlans( joinPlans );
alias = buildOp( loc, op, alias, inputAliases, partitioner );
+ expandAndResetVisitor(loc, op);
+ return alias;
+ }
+
+ private void expandAndResetVisitor(SourceLocation loc,
+ LogicalRelationalOperator lrop) throws ParserValidationException {
try {
- (new ProjectStarExpander(op.getPlan())).visit(op);
- (new ProjStarInUdfExpander(op.getPlan())).visit(op);
- new SchemaResetter(op.getPlan(), true).visit(op);
+ (new ProjectStarExpander(lrop.getPlan())).visit();
+ (new ProjStarInUdfExpander(lrop.getPlan())).visit();
+ new SchemaResetter(lrop.getPlan(), true).visit();
+ } catch (FrontendException e) {
+ throw new ParserValidationException(intStream, loc, e);
+ }
+ }
+
+ LOCube createCubeOp() {
+ return new LOCube(plan);
+ }
+
+ String buildCubeOp(SourceLocation loc, LOCube op, String alias,
+ String inputAlias,
+ MultiMap<Integer, LogicalExpressionPlan> expressionPlans)
+ throws ParserValidationException {
+
+ // set the expression plans for cube operator and build cube operator
+ op.setExpressionPlans(expressionPlans);
+ buildOp(loc, op, alias, inputAlias, null);
+ expandAndResetVisitor(loc, op);
+
+ try {
+ alias = convertCubeToFGPlan(loc, op, inputAlias, op.getExpressionPlans());
} catch (FrontendException e) {
throw new ParserValidationException( intStream, loc, e );
}
return alias;
}
+ // This function creates logical plan for foreach and groupby operators.
+ // It connects the predecessors of cube operator with foreach plan and
+ // disconnects cube operator from the logical plan. It also connects foreach
+ // plan with groupby plan.
+ private String convertCubeToFGPlan(SourceLocation loc, LOCube op,
+ String inputAlias,
+ MultiMap<Integer, LogicalExpressionPlan> expressionPlans)
+ throws FrontendException {
+
+ LOForEach foreach = new LOForEach(plan);
+ LOCogroup groupby = new LOCogroup(plan);
+ LogicalPlan innerPlan = new LogicalPlan();
+ LogicalRelationalOperator gen = new LOGenerate(innerPlan);
+
+ injectForeachOperator(loc, op, foreach);
+
+ // Get all column attributes from the input relation.
+ // Create ProjectExpression for all columns. Based on the
+ // dimensions specified by the user, specified columns will be attached
+ // to CubeDimension UDF and rest will be pushed down
+ List<Operator> inpOpers = foreach.getPlan().getPredecessors(foreach);
+ List<LogicalExpressionPlan> allExprPlan = new ArrayList<LogicalExpressionPlan>();
+ for (Operator oper : inpOpers) {
+ LogicalSchema schema = new LogicalSchema();
+ schema = ((LogicalRelationalOperator) oper).getSchema();
+
+ if (schema != null) {
+ ArrayList<LogicalFieldSchema> fields = (ArrayList<LogicalFieldSchema>)schema.getFields();
+ for (int i = 0; i < fields.size(); i++) {
+ LogicalExpressionPlan lEplan = new LogicalExpressionPlan();
+ new ProjectExpression(lEplan, i, fields.get(i).alias, gen);
+ allExprPlan.add(lEplan);
+ }
+ }
+ }
+
+ List<LogicalExpressionPlan> lexpPlanList = new ArrayList<LogicalExpressionPlan>();
+ List<LogicalExpression> lexpList = new ArrayList<LogicalExpression>();
+
+ // TODO: current implementation only supports star schema
+ // if snow-flake schema is to be supported then dimensions
+ // from multiple tables should be retrieved here.
+ lexpPlanList.addAll(expressionPlans.get(0));
+
+ // If duplicates exists in the dimension list then exception is thrown
+ checkDuplicateProject(lexpPlanList);
+
+ // Construct ProjectExpression from the LogicalExpressionPlans
+ lexpList = getProjectExpList(lexpPlanList, gen);
+
+ for (int i = 0; i < lexpList.size(); i++) {
+ // Retain the columns that needs to be pushed down.
+ // Remove the dimension columns from the input column list
+ // as it will be attached to CubeDimension UDF
+ for (int j = 0; j < allExprPlan.size(); j++) {
+ LogicalExpression lexp = (LogicalExpression) allExprPlan.get(j).getSources().get(0);
+ String colAlias = ((ProjectExpression) lexpList.get(i)).getColAlias();
+ if (colAlias == null) {
+ colAlias = ((ProjectExpression) lexpList.get(i)).getFieldSchema().alias;
+ }
+
+ if (colAlias.equals(((ProjectExpression) lexp).getColAlias()) == true) {
+ allExprPlan.remove(j);
+ }
+ }
+
+ }
+
+ // Create UDF with user specified dimensions
+ LogicalExpressionPlan uexpPlan = new LogicalExpressionPlan();
+ new UserFuncExpression(uexpPlan, new FuncSpec(CubeDimensions.class.getName(), "NULL"), lexpList);
+ for (LogicalExpressionPlan lexp : lexpPlanList) {
+ Iterator<Operator> it = lexp.getOperators();
+ while (it.hasNext()) {
+ uexpPlan.add(it.next());
+ }
+ }
+ // Add the UDF to logical expression plan that contains dependent
+ // attributes (pushed down from input columns)
+ allExprPlan.add(0, uexpPlan);
+
+ // If the operator is a UserFuncExpression then set the flatten flags.
+ List<Boolean> flattenFlags = new ArrayList<Boolean>();
+ for (int i = 0; i < allExprPlan.size(); i++) {
+ List<Operator> opers = allExprPlan.get(i).getSources();
+ for (Operator oper : opers) {
+ if (oper instanceof ProjectExpression) {
+ flattenFlags.add(false);
+ } else if (oper instanceof UserFuncExpression) {
+ flattenFlags.add(true);
+ }
+ }
+ }
+
+ // Generate and Foreach operator creation
+ String falias = null;
+ try {
+ buildGenerateOp(loc, (LOForEach) foreach, (LOGenerate) gen,
+ operators, allExprPlan, flattenFlags, null);
+ falias = buildForeachOp(loc, (LOForEach) foreach, "cube",inputAlias, innerPlan);
+ } catch (ParserValidationException pve) {
+ throw new FrontendException(pve);
+ }
+
+ List<Boolean> innerFlags = new ArrayList<Boolean>();
+ List<String> inpAliases = new ArrayList<String>();
+ inpAliases.add(falias);
+ innerFlags.add(false);
+
+ // Get the output schema of foreach operator and reconstruct the
+ // LogicalExpressionPlan for each dimensional attributes
+ MultiMap<Integer, LogicalExpressionPlan> exprPlansCopy = new MultiMap<Integer, LogicalExpressionPlan>();
+ LogicalSchema fSchema = null;
+ fSchema = foreach.getSchema();
+
+ List<LogicalFieldSchema> lfSchemas = fSchema.getFields();
+ for (LogicalFieldSchema lfSchema : lfSchemas) {
+ LogicalExpressionPlan epGrp = new LogicalExpressionPlan();
+ if (lfSchema.alias.contains("dimensions::") == true) {
+ new ProjectExpression(epGrp, 0, lfSchema.alias, groupby);
+ exprPlansCopy.put(0, epGrp);
+ }
+ }
+
+ // build group by operator
+ try {
+ return buildGroupOp(loc, (LOCogroup) groupby, op.getAlias(),
+ inpAliases, exprPlansCopy, GROUPTYPE.REGULAR, innerFlags,null);
+ } catch (ParserValidationException pve) {
+ throw new FrontendException(pve);
+ }
+ }
+
+ private List<LogicalExpression> getProjectExpList(List<LogicalExpressionPlan> lexpPlanList,
+ LogicalRelationalOperator lro) throws FrontendException {
+
+ List<LogicalExpression> leList = new ArrayList<LogicalExpression>();
+ for (int i = 0; i < lexpPlanList.size(); i++) {
+ LogicalExpressionPlan lexp = lexpPlanList.get(i);
+ LogicalExpression lex = (LogicalExpression) lexp.getSources().get(0);
+ Iterator<Operator> opers = lexp.getOperators();
+
+ // ProjExpr are initially attached to CubeOp. So re-attach it to
+ // specified operator
+ while (opers.hasNext()) {
+ Operator oper = opers.next();
+ try {
+ ((ProjectExpression) oper).setAttachedRelationalOp(lro);
+ } catch (ClassCastException cce) {
+ throw new FrontendException("Column project expected.", cce);
+ }
+ }
+
+ leList.add(lex);
+ }
+
+ return leList;
+ }
+
+
+ // This method connects the predecessors of cube operator with foreach
+ // operator and disconnects the cube operator from its predecessors
+ private void injectForeachOperator(SourceLocation loc, LOCube op,
+ LOForEach foreach) throws FrontendException {
+ // connect the foreach operator with predecessors of cube operator
+ List<Operator> opers = op.getPlan().getPredecessors(op);
+ for (Operator oper : opers) {
+ OperatorPlan foreachPlan = foreach.getPlan();
+ foreachPlan.connect(oper, (Operator) foreach);
+ }
+
+ // disconnect the cube operator from the plan
+ opers = foreach.getPlan().getPredecessors(foreach);
+ for (Operator lop : opers) {
+ List<Operator> succs = lop.getPlan().getSuccessors(lop);
+ for (Operator succ : succs) {
+ if (succ instanceof LOCube) {
+ succ.getPlan().disconnect(lop, succ);
+ succ.getPlan().remove(succ);
+ }
+ }
+ }
+ }
+
+ // This methods if the dimensions specified by the user has duplicates
+ private void checkDuplicateProject(List<LogicalExpressionPlan> lExprPlan)
+ throws FrontendException {
+
+ for (int i = 0; i < lExprPlan.size(); i++) {
+ for (int j = i + 1; j < lExprPlan.size(); j++) {
+ LogicalExpression outer = (LogicalExpression) lExprPlan.get(i).getSources().get(0);
+ LogicalExpression inner = (LogicalExpression) lExprPlan.get(j).getSources().get(0);
+ String outColAlias = ((ProjectExpression) outer).getColAlias();
+ String inColAlias = ((ProjectExpression) inner).getColAlias();
+
+ if (outColAlias == null) {
+ outColAlias = outer.getFieldSchema().alias;
+ }
+
+ if (inColAlias == null) {
+ inColAlias = inner.getFieldSchema().alias;
+ }
+
+ if (outColAlias.equals(inColAlias) == true) {
+ lExprPlan.remove(j);
+ throw new FrontendException("Duplicate dimensions detected. Dimension name: " + inColAlias);
+ }
+ }
+ }
+
+ }
+
+
LOCogroup createGroupOp() {
return new LOCogroup( plan );
}
@@ -398,13 +634,7 @@ public class LogicalPlanBuilder {
op.setGroupType( gt );
op.setInnerFlags( flags );
alias = buildOp( loc, op, alias, inputAliases, partitioner );
- try {
- (new ProjectStarExpander(op.getPlan())).visit(op);
- (new ProjStarInUdfExpander(op.getPlan())).visit(op);
- new SchemaResetter(op.getPlan(), true).visit(op);
- } catch (FrontendException e) {
- throw new ParserValidationException( intStream, loc, e );
- }
+ expandAndResetVisitor(loc, op);
return alias;
}
@@ -514,13 +744,7 @@ public class LogicalPlanBuilder {
throws ParserValidationException {
op.setInnerPlan( innerPlan );
alias = buildOp( loc, op, alias, inputAlias, null );
- try {
- (new ProjectStarExpander(op.getPlan())).visit(op);
- (new ProjStarInUdfExpander(op.getPlan())).visit(op);
- new SchemaResetter(op.getPlan(), true).visit(op);
- } catch (FrontendException e) {
- throw new ParserValidationException( intStream, loc, e );
- }
+ expandAndResetVisitor(loc, op);
return alias;
}
Modified: pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g?rev=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g (original)
+++ pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g Sat May 26 21:37:56 2012
@@ -78,6 +78,7 @@ import org.apache.pig.newplan.logical.ex
import org.apache.pig.newplan.logical.expression.SubtractExpression;
import org.apache.pig.newplan.logical.expression.UserFuncExpression;
import org.apache.pig.newplan.logical.relational.LOCogroup;
+import org.apache.pig.newplan.logical.relational.LOCube;
import org.apache.pig.newplan.logical.relational.LOFilter;
import org.apache.pig.newplan.logical.relational.LOForEach;
import org.apache.pig.newplan.logical.relational.LOGenerate;
@@ -218,6 +219,7 @@ op_clause returns[String alias] :
| stream_clause { $alias = $stream_clause.alias; }
| mr_clause { $alias = $mr_clause.alias; }
| foreach_clause { $alias = $foreach_clause.alias; }
+ | cube_clause { $alias = $cube_clause.alias; }
;
define_clause
@@ -464,6 +466,67 @@ func_args returns[List<String> args]
)+
;
+// Sets the current operator as CUBE and creates LogicalExpressionPlans based on the user input.
+// Ex: a = CUBE inp BY (x,y);
+// For the above example this grammar creates LogicalExpressionPlan with ProjectExpression for x and y dimensions.
+// inputIndex keeps track of input dataset (which will be useful when cubing is performed over multiple datasets).
+// These inputs are passed to buildCubeOp methods which then builds the logical plan for CUBE operator.
+// If user specifies STAR or RANGE expression for dimensions then it will be expanded inside buildCubeOp.
+cube_clause returns[String alias]
+scope {
+ LOCube cubeOp;
+ MultiMap<Integer, LogicalExpressionPlan> cubePlans;
+ int inputIndex;
+}
+scope GScope;
+@init {
+ $cube_clause::cubeOp = builder.createCubeOp();
+ $GScope::currentOp = $cube_clause::cubeOp;
+ $cube_clause::cubePlans = new MultiMap<Integer, LogicalExpressionPlan>();
+ int oldStatementIndex = $statement::inputIndex;
+}
+@after { $statement::inputIndex = oldStatementIndex; }
+ : ^( CUBE cube_item )
+ {
+ SourceLocation loc = new SourceLocation( (PigParserNode)$cube_clause.start );
+ $alias = builder.buildCubeOp( loc, $cube_clause::cubeOp, $statement::alias,
+ $statement::inputAlias, $cube_clause::cubePlans );
+ }
+;
+
+cube_item
+ : rel ( cube_by_clause
+ {
+ $cube_clause::cubePlans.put( $cube_clause::inputIndex, $cube_by_clause.plans );
+ }
+ )
+ {
+ $cube_clause::inputIndex++;
+ $statement::inputIndex++;
+ }
+;
+
+cube_by_clause returns[List<LogicalExpressionPlan> plans]
+@init {
+ $plans = new ArrayList<LogicalExpressionPlan>();
+}
+ : ^( BY ( cube_by_expr { $plans.add( $cube_by_expr.plan ); } )+ )
+;
+
+cube_by_expr returns[LogicalExpressionPlan plan]
+@init {
+ $plan = new LogicalExpressionPlan();
+}
+ : col_range[$plan]
+ | expr[$plan]
+ | STAR
+ {
+ builder.buildProjectExpr( new SourceLocation( (PigParserNode)$STAR ), $plan, $GScope::currentOp,
+ $statement::inputIndex, null, -1 );
+ }
+;
+
+
group_clause returns[String alias]
scope {
MultiMap<Integer, LogicalExpressionPlan> groupPlans;
@@ -882,6 +945,7 @@ col_alias_or_index returns[Object col]
col_alias returns[Object col]
: GROUP { $col = $GROUP.text; }
+ | CUBE { $col = $CUBE.text; }
| IDENTIFIER { $col = $IDENTIFIER.text; }
;
@@ -1430,6 +1494,11 @@ alias_col_ref[LogicalExpressionPlan plan
$expr = builder.buildProjectExpr( new SourceLocation( (PigParserNode)$GROUP ), $plan, $GScope::currentOp,
$statement::inputIndex, $GROUP.text, 0 );
}
+ | CUBE
+ {
+ $expr = builder.buildProjectExpr( new SourceLocation( (PigParserNode)$CUBE ), $plan, $GScope::currentOp,
+ $statement::inputIndex, $CUBE.text, 0 );
+ }
| IDENTIFIER
{
SourceLocation loc = new SourceLocation( (PigParserNode)$IDENTIFIER );
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/QueryLexer.g (original)
+++ pig/trunk/src/org/apache/pig/parser/QueryLexer.g Sat May 26 21:37:56 2012
@@ -81,6 +81,9 @@ FOREACH : 'FOREACH'
ORDER : 'ORDER'
;
+CUBE : 'CUBE'
+;
+
DISTINCT : 'DISTINCT'
;
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/QueryParser.g (original)
+++ pig/trunk/src/org/apache/pig/parser/QueryParser.g Sat May 26 21:37:56 2012
@@ -215,6 +215,7 @@ content : LEFT_CURLY ( content | ~(LEFT_
op_clause : define_clause
| load_clause
| group_clause
+ | cube_clause
| store_clause
| filter_clause
| distinct_clause
@@ -470,7 +471,7 @@ dot_proj : PERIOD ( col_alias_or_index
col_alias_or_index : col_alias | col_index
;
-col_alias : GROUP | identifier
+col_alias : GROUP | CUBE | identifier
;
col_index : DOLLARVAR
@@ -574,6 +575,23 @@ foreach_plan_complex : nested_blk
-> ^( FOREACH_PLAN_COMPLEX nested_blk )
;
+cube_clause : CUBE^ cube_item
+;
+
+cube_item : rel ( cube_by_clause )
+;
+
+cube_by_clause : BY^ cube_by_expr_list
+;
+
+cube_by_expr_list : LEFT_PAREN cube_by_expr ( COMMA cube_by_expr )* RIGHT_PAREN
+ -> cube_by_expr+
+ | cube_by_expr
+;
+
+cube_by_expr : col_range | expr | STAR
+;
+
nested_blk : LEFT_CURLY! nested_command_list ( generate_clause SEMI_COLON! ) RIGHT_CURLY!
;
@@ -656,7 +674,7 @@ split_otherwise : alias OTHERWISE
col_ref : alias_col_ref | dollar_col_ref
;
-alias_col_ref : GROUP | identifier
+alias_col_ref : GROUP | CUBE | identifier
;
dollar_col_ref : DOLLARVAR
@@ -708,6 +726,7 @@ eid : rel_str_op
| LOAD
| FILTER
| FOREACH
+ | CUBE
| ORDER
| DISTINCT
| COGROUP
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/parser/TestLexer.pig (original)
+++ pig/trunk/test/org/apache/pig/parser/TestLexer.pig Sat May 26 21:37:56 2012
@@ -62,6 +62,8 @@ B = GROUP A ALL using 'collected';
I = foreach A generate flatten(B::c);
+J = CUBE A BY ($0, $1, $2, $3);
+
CMD = `ls -l`;
A = stream through CMD;
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java (original)
+++ pig/trunk/test/org/apache/pig/parser/TestLogicalPlanGenerator.java Sat May 26 21:37:56 2012
@@ -299,6 +299,36 @@ public class TestLogicalPlanGenerator {
}
@Test
+ public void testCubeBasic() {
+ String query = "a = load 'input' as (x:chararray,y:chararray,z:long);"
+ + "b = cube a by (x,y);"
+ + "c = foreach b generate flatten(group) as (x,y), COUNT(cube) as count, SUM(cube.z) as total;"
+ + "store c into 'output';";
+ generateLogicalPlan(query);
+ }
+
+ @Test
+ public void testCubeMultipleIAlias() {
+ String query = "a = load 'input' as (x:chararray,y:chararray,z:long);"
+ + "a = load 'input' as (x,y:chararray,z:long);"
+ + "a = load 'input' as (x:chararray,y:chararray,z:long);"
+ + "b = cube a by (x,y);"
+ + "c = foreach b generate flatten(group) as (x,y), COUNT(cube) as count, SUM(cube.z) as total;"
+ + "store c into 'c';";
+ generateLogicalPlan(query);
+ }
+
+ @Test
+ public void testCubeAfterForeach() {
+ String query = "a = load 'input' as (x:chararray,y:chararray,z:long);"
+ + "b = foreach a generate x as type,y as location,z as number;"
+ + "c = cube b by (type,location);"
+ + "d = foreach c generate flatten(group) as (type,location), COUNT(cube) as count, SUM(cube.number) as total;"
+ + "store d into 'd';";
+ generateLogicalPlan(query);
+ }
+
+ @Test
public void testFilter() {
String query = "A = load 'x' as ( u:int, v:long, w:bytearray); " +
"B = filter A by 2 > 1;\n" +
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/parser/TestParser.pig (original)
+++ pig/trunk/test/org/apache/pig/parser/TestParser.pig Sat May 26 21:37:56 2012
@@ -65,6 +65,10 @@ D = cogroup A by $0 inner, B by $0 outer
grpInactiveAcct = group inactiveAccounts all;
B = GROUP A ALL using 'collected';
+--cube
+C = CUBE A BY (a, b);
+CC = CUBE A BY (*);
+
--join
E = join A by $0, B by $0 using 'replicated';
H = join A by u, B by u;
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java (original)
+++ pig/trunk/test/org/apache/pig/parser/TestQueryLexer.java Sat May 26 21:37:56 2012
@@ -48,7 +48,7 @@ public class TestQueryLexer {
// 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( 400, tokenCount );
+ Assert.assertEquals( 415, tokenCount );
Assert.assertEquals( 0, lexer.getNumberOfSyntaxErrors() );
}
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=1342975&r1=1342974&r2=1342975&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/parser/TestQueryParser.java (original)
+++ pig/trunk/test/org/apache/pig/parser/TestQueryParser.java Sat May 26 21:37:56 2012
@@ -178,6 +178,69 @@ public class TestQueryParser {
}
@Test
+ public void testCubeNegative1() throws IOException, RecognitionException {
+ // cube keyword used as alias
+ String query = "x = load 'cubedata' as (a, b, c, d); " +
+ "cube = cube x by (a, b, c);";
+ shouldFail( query );
+ }
+
+ @Test
+ public void testCubeNegative2() throws IOException, RecognitionException {
+ // syntax error - brackets missing
+ String query = "x = load 'cubedata' as (a, b, c, d); " +
+ "y = cube x by a, b, c;";
+ shouldFail( query );
+ }
+
+ @Test
+ public void testCubeNegative3() throws IOException, RecognitionException {
+ // syntax error - BY missing
+ String query = "x = load 'cubedata' as (a, b, c, d); " +
+ "y = cube x (a, b, c);";
+ shouldFail( query );
+ }
+
+ @Test
+ public void testCubeNegative4() throws IOException, RecognitionException {
+ // syntax error - UDF at the end
+ String query = "x = load 'cubedata' as (a, b, c, d); " +
+ "y = cube x by (a, b, c), UDF(c);";
+ shouldFail( query );
+ }
+
+ @Test
+ public void testCubePositive1() throws IOException, RecognitionException {
+ // syntactically correct
+ String query = "x = load 'cubedata' as (a, b, c, d);" +
+ "y = cube x by (a, b, c);" +
+ "z = foreach y generate flatten(group) as (a, b, c), COUNT(x) as count;" +
+ "store z into 'cube_output';";
+ shouldPass( query );
+ }
+
+ @Test
+ public void testCubePositive2() throws IOException, RecognitionException {
+ // all columns using *
+ String query = "x = load 'cubedata' as (a, b, c, d);" +
+ "y = cube x by (*);" +
+ "z = foreach y generate flatten(group) as (a, b, c, d), COUNT(x) as count;" +
+ "store z into 'cube_output';";
+ shouldPass( query );
+ }
+
+
+ @Test
+ public void testCubePositive3() throws IOException, RecognitionException {
+ // range projection
+ String query = "x = load 'cubedata' as (a, b, c, d);" +
+ "y = cube x by ($0, $1);" +
+ "z = foreach y generate flatten(group) as (a, b), COUNT(x) as count;" +
+ "store z into 'cube_output';";
+ shouldPass( query );
+ }
+
+ @Test
public void test9() throws IOException, RecognitionException {
String query = "a = load 'x' as (u,v);" +
"b = load 'y' as (u,w);" +
Added: pig/trunk/test/org/apache/pig/test/TestCubeOperator.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestCubeOperator.java?rev=1342975&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestCubeOperator.java (added)
+++ pig/trunk/test/org/apache/pig/test/TestCubeOperator.java Sat May 26 21:37:56 2012
@@ -0,0 +1,495 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pig.test;
+
+import static org.apache.pig.builtin.mock.Storage.resetData;
+import static org.apache.pig.builtin.mock.Storage.tuple;
+import static org.junit.Assert.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import junit.framework.Assert;
+
+import org.apache.pig.PigServer;
+import org.apache.pig.builtin.mock.Storage.Data;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.newplan.Operator;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+public class TestCubeOperator {
+ private static PigServer pigServer;
+ private static TupleFactory tf = TupleFactory.getInstance();
+ private Data data;
+
+ @BeforeClass
+ public static void oneTimeSetUp() throws Exception {
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ pigServer = new PigServer("local");
+
+ data = resetData(pigServer);
+ data.set("input",
+ tuple("dog", "miami", 12),
+ tuple("cat", "miami", 18),
+ tuple("turtle", "tampa", 4),
+ tuple("dog", "tampa", 14),
+ tuple("cat", "naples", 9),
+ tuple("dog", "naples", 5),
+ tuple("turtle", "naples", 1));
+
+ data.set("input1",
+ tuple("u1,men,green,mango"),
+ tuple("u2,men,red,mango"),
+ tuple("u3,men,green,apple"),
+ tuple("u4,women,red,mango"),
+ tuple("u6,women,green,mango"),
+ tuple("u7,men,red,apple"),
+ tuple("u8,men,green,mango"),
+ tuple("u9,women,red,apple"),
+ tuple("u10,women,green,apple"),
+ tuple("u11,men,red,apple"),
+ tuple("u12,women,green,mango"));
+ }
+
+ @AfterClass
+ public static void oneTimeTearDown() throws IOException {
+ }
+
+ @Test
+ public void testCubeBasic() throws IOException {
+ // basic correctness test
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = cube a by (x,y);" +
+ "c = foreach b generate flatten(group) as (type,location), COUNT(cube) as count, SUM(cube.z) as total;" +
+ "store c into 'output' using mock.Storage();";
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "naples", (long)1, (long)9)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)2, (long)27)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3, (long)31)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)3, (long)15)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)7, (long)63))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeMultipleIAliases() throws IOException {
+ // test for input alias to cube being assigned multiple times
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "a = load 'input' USING mock.Storage() as (x,y:chararray,z:long);" +
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = cube a by (x,y);" +
+ "c = foreach b generate flatten(group) as (type,location), COUNT(cube) as count, SUM(cube.z) as total;" +
+ "store c into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "naples", (long)1, (long)9)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)2, (long)27)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3, (long)31)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)3, (long)15)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)7, (long)63))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeAfterForeach() throws IOException {
+ // test for foreach projection before cube operator
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = foreach a generate x as type,y as location,z as number;" +
+ "c = cube b by (type,location);" +
+ "d = foreach c generate flatten(group) as (type,location), COUNT(cube) as count, SUM(cube.number) as total;" +
+ "store d into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "naples", (long)1, (long)9)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)2, (long)27)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3, (long)31)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)3, (long)15)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)7, (long)63))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeAfterLimit() throws IOException {
+ // test for limit operator before cube operator
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = limit a 2;" +
+ "c = cube b by (x,y);" +
+ "d = foreach c generate flatten(group) as (x,y), SUM(cube.z) as total;" +
+ "store d into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)18)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)12)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)30))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeWithStar() throws IOException {
+ // test for * (all) dimensions in cube operator
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray);" +
+ "b = foreach a generate x as type,y as location;" +
+ "c = cube b by (*);" +
+ "d = foreach c generate flatten(group) as (type,location), COUNT(cube) as count;" +
+ "store d into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1)),
+ tf.newTuple(ImmutableList.of("cat", "naples", (long)1)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)2)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)3)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)7))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeWithRange() throws IOException {
+ // test for range projection of dimensions in cube operator
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = foreach a generate x as type,y as location, z as number;" +
+ "c = cube b by ($0..$1);" +
+ "d = foreach c generate flatten(group) as (type,location), COUNT(cube) as count, SUM(cube.number) as total;" +
+ "store d into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "naples", (long)1, (long)9)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)2, (long)27)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3, (long)31)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)3, (long)15)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)7, (long)63))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeDuplicateDimensions() throws IOException {
+ // test for cube operator with duplicate dimensions
+ String query = "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);"
+ + "b = foreach a generate x as type,y as location, z as number;"
+ + "c = cube b by ($0..$1,$0..$1);"
+ + "d = foreach c generate flatten(group), COUNT(cube) as count, SUM(cube.number) as total;"
+ + "store d into 'output' using mock.Storage();";
+
+ try {
+ Util.registerMultiLineQuery(pigServer, query);
+ pigServer.openIterator("d");
+ } catch (FrontendException e) {
+ // FEException with 'duplicate dimensions detected' message is throw
+ return;
+ }
+
+ Assert.fail("Expected to throw an exception when duplicate dimensions are detected!");
+
+ }
+
+ @Test
+ public void testCubeAfterFilter() throws IOException {
+ // test for filtering before cube operator
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = filter a by x == 'dog';" +
+ "c = cube b by (x,y);" +
+ "d = foreach c generate flatten(group), COUNT(cube) as count, SUM(cube.z) as total;" +
+ "store d into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+ // Iterator<Tuple> it = pigServer.openIterator("d");
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3, (long)31)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)3, (long)31))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+
+ }
+
+ @Test
+ public void testCubeAfterOrder() throws IOException {
+ // test for ordering before cube operator
+ String query =
+ "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);" +
+ "b = order a by $2;" +
+ "c = cube b by (x,y);" +
+ "d = foreach c generate flatten(group), COUNT(cube) as count, SUM(cube.z) as total;" +
+ "store d into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "naples", (long)1, (long)9)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)2, (long)27)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "naples", (long)1, (long)5)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)3, (long)31)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)3, (long)15)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)7, (long)63))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+ }
+
+ @Test
+ public void testCubeAfterJoin() throws IOException {
+ // test for cubing on joined relations
+ String query =
+ "a = load 'input1' USING mock.Storage() as (a1:chararray,b1,c1,d1); " +
+ "b = load 'input' USING mock.Storage() as (a2,b2,c2:long,d2:chararray);" +
+ "c = join a by a1, b by d2;" +
+ "d = cube c by ($4,$5);" +
+ "e = foreach d generate flatten(group), COUNT(cube) as count, SUM(cube.c2) as total;" +
+ "store e into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)2, (long)26)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)5, (long)49))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+ }
+
+ @Test
+ public void testCubeAfterCogroup() throws IOException {
+ // test for cubing on co-grouped relation
+ String query =
+ "a = load 'input1' USING mock.Storage() as (a1:chararray,b1,c1,d1); " +
+ "b = load 'input' USING mock.Storage() as (a2,b2,c2:long,d2:chararray);" +
+ "c = cogroup a by a1, b by d2;" +
+ "d = foreach c generate flatten(a), flatten(b);" +
+ "e = cube d by (a2,b2);" +
+ "f = foreach e generate flatten(group), COUNT(cube) as count, SUM(cube.c2) as total;" +
+ "store f into 'output' using mock.Storage();";
+
+ Util.registerMultiLineQuery(pigServer, query);
+
+ Set<Tuple> expected = ImmutableSet.of(
+ tf.newTuple(ImmutableList.of("cat", "miami", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("cat", "NULL", (long)1, (long)18)),
+ tf.newTuple(ImmutableList.of("dog", "miami", (long)1, (long)12)),
+ tf.newTuple(ImmutableList.of("dog", "tampa", (long)1, (long)14)),
+ tf.newTuple(ImmutableList.of("dog", "NULL", (long)2, (long)26)),
+ tf.newTuple(ImmutableList.of("turtle", "tampa", (long)1, (long)4)),
+ tf.newTuple(ImmutableList.of("turtle", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("turtle", "NULL", (long)2, (long)5)),
+ tf.newTuple(ImmutableList.of("NULL", "miami", (long)2, (long)30)),
+ tf.newTuple(ImmutableList.of("NULL", "tampa", (long)2, (long)18)),
+ tf.newTuple(ImmutableList.of("NULL", "naples", (long)1, (long)1)),
+ tf.newTuple(ImmutableList.of("NULL", "NULL", (long)5, (long)49))
+ );
+
+ List<Tuple> out = data.get("output");
+ for( Tuple tup : out ) {
+ assertTrue(expected+" contains "+tup, expected.contains(tup));
+ }
+ }
+
+ @Test
+ public void testIllustrate() throws IOException {
+ // test for illustrate
+ String query =
+ "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); " +
+ "b = cube a by (a1,b1);";
+
+ Util.registerMultiLineQuery(pigServer, query);
+ Map<Operator, DataBag> examples = pigServer.getExamples("b");
+ assertTrue(examples != null);
+ }
+
+ @Test
+ public void testExplain() throws IOException {
+ // test for explain
+ String query =
+ "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); " +
+ "b = cube a by (a1,b1);";
+
+ Util.registerMultiLineQuery(pigServer, query);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ PrintStream ps = new PrintStream(baos);
+ pigServer.explain("b", ps);
+ assertTrue(baos.toString().contains("CubeDimensions('NULL')"));
+ }
+
+ @Test
+ public void testDescribe() throws IOException {
+ // test for describe
+ String query =
+ "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); " +
+ "b = cube a by (a1,b1);";
+
+ Util.registerMultiLineQuery(pigServer, query);
+ Schema sch = pigServer.dumpSchema("b");
+ for(String alias : sch.getAliases()) {
+ if(alias.compareTo("cube") == 0) {
+ assertTrue(alias.contains("cube"));
+ }
+ }
+ }
+}
Added: pig/trunk/test/org/apache/pig/test/data/output26.pig
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/data/output26.pig?rev=1342975&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/test/data/output26.pig (added)
+++ pig/trunk/test/org/apache/pig/test/data/output26.pig Sat May 26 21:37:56 2012
@@ -0,0 +1,20 @@
+ avro = LOAD '/data/part-m-00000.avro' USING PigStorage ();
+ avro2 = FOREACH avro GENERATE browser_id, component_version, member_id, page_key, session_id, tracking_time, type;
+ fs -rmr testOut/out1;
+ STORE avro2 INTO 'testOut/out2'
+ USING PigStorage (
+ ' {
+ "debug": 5,
+ "schema":
+ { "type":"record","name":"TestRecord",
+ "fields": [ {"name":"browser_id", "type":["null","string"]},
+ {"name":"component_version","type":"int"},
+ {"name":"member_id","type":"int"},
+ {"name":"page_key","type":["null","string"]},
+ {"name":"session_id","type":"long"},
+ {"name":"tracking_time","type":"long"},
+ {"name":"type","type":["null","string"]}
+ ]
+ }
+ }
+ ');