You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by pr...@apache.org on 2009/03/25 01:42:54 UTC
svn commit: r758107 [7/8] - in /hadoop/pig/branches/multiquery: ./ bin/
docs/ src/ src/docs/ src/docs/src/ src/docs/src/documentation/
src/docs/src/documentation/classes/ src/docs/src/documentation/content/
src/docs/src/documentation/content/xdocs/ src...
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/parser/QueryParser.jjt Wed Mar 25 00:42:40 2009
@@ -830,7 +830,11 @@
throw new ParseException(
"Currently PIG does not support assigning an existing relation (" + t1.image + ") to another alias (" + t2.image + ")");})
| LOOKAHEAD(2)
- (t1 = <IDENTIFIER> "=" root = Expr(lp) ";" {root.setAlias(t1.image); addAlias(t1.image, root);})
+ (t1 = <IDENTIFIER> "=" root = Expr(lp) ";" {
+ root.setAlias(t1.image);
+ addAlias(t1.image, root);
+ pigContext.setLastAlias(t1.image);
+ })
| (root = Expr(lp) ";")
| (<SPLIT> root = SplitClause(lp) ";")
)
@@ -1201,7 +1205,7 @@
(
<OR> rhsCond = PAndCond(over,specs,lp,input)
{
- ExpressionOperator exprOp = new LOOr(lp, new OperatorKey(scope, getNextId()), lhsCond, rhsCond);
+ ExpressionOperator exprOp = new LOOr(lp, new OperatorKey(scope, getNextId()) );
lp.add(exprOp);
log.debug("POrCond: Added operator " + exprOp.getClass().getName() + " " + exprOp + " to logical plan " + lp);
lp.connect(lhsCond, exprOp);
@@ -1230,7 +1234,7 @@
(
<AND> rhsCond = PUnaryCond(over,specs,lp,input)
{
- ExpressionOperator exprOp = new LOAnd(lp, new OperatorKey(scope, getNextId()), lhsCond, rhsCond);
+ ExpressionOperator exprOp = new LOAnd(lp, new OperatorKey(scope, getNextId()) );
lp.add(exprOp);
log.debug("PAndCond: Added operator " + exprOp.getClass().getName() + " " + exprOp + " to logical plan " + lp);
lp.connect(lhsCond, exprOp);
@@ -1276,28 +1280,28 @@
// numeric ops first
case '=':
if (op2 == '=') {
- cond = new LOEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOEqual(lp, new OperatorKey(scope, getNextId()) );
} else {
throw new ParseException("Internal error: Invalid filter operator: " + op);
}
break;
case '<':
if (op2 == '=') {
- cond = new LOLesserThanEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOLesserThanEqual(lp, new OperatorKey(scope, getNextId()));
} else {
- cond = new LOLesserThan(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOLesserThan(lp, new OperatorKey(scope, getNextId()));
}
break;
case '>':
if (op2 == '=') {
- cond = new LOGreaterThanEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOGreaterThanEqual(lp, new OperatorKey(scope, getNextId()));
} else {
- cond = new LOGreaterThan(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOGreaterThan(lp, new OperatorKey(scope, getNextId()));
}
break;
case '!':
if (op2 == '=') {
- cond = new LONotEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LONotEqual(lp, new OperatorKey(scope, getNextId()));
} else {
throw new ParseException("Internal error: Invalid filter operator: " + op);
}
@@ -1305,28 +1309,28 @@
// now string ops
case 'e':
if (op2 == 'q') {
- cond = new LOEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOEqual(lp, new OperatorKey(scope, getNextId()));
} else {
throw new ParseException("Internal error: Invalid filter operator: " + op);
}
break;
case 'l':
if (op2 == 't' && op3 == 'e') {
- cond = new LOLesserThanEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOLesserThanEqual(lp, new OperatorKey(scope, getNextId()));
} else {
- cond = new LOLesserThan(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOLesserThan(lp, new OperatorKey(scope, getNextId()));
}
break;
case 'g':
if (op2 == 't' && op3 == 'e') {
- cond = new LOGreaterThanEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOGreaterThanEqual(lp, new OperatorKey(scope, getNextId()));
} else {
- cond = new LOGreaterThan(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LOGreaterThan(lp, new OperatorKey(scope, getNextId()));
}
break;
case 'n':
if (op2 == 'e' && op3 == 'q') {
- cond = new LONotEqual(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ cond = new LONotEqual(lp, new OperatorKey(scope, getNextId()));
} else {
throw new ParseException("Internal error: Invalid filter operator: " + op);
}
@@ -1348,7 +1352,7 @@
{
LOConst rconst = new LOConst(lp, new OperatorKey(scope, getNextId()), unquote(t1.image));
rconst.setType(DataType.CHARARRAY);
- cond = new LORegexp(lp, new OperatorKey(scope, getNextId()), lhs, rconst);
+ cond = new LORegexp(lp, new OperatorKey(scope, getNextId()));
lp.add(rconst);
lp.add(cond);
log.debug("PUnaryCond: Added operator " + cond.getClass().getName() + " " + cond + " to logical plan " + lp);
@@ -1373,7 +1377,7 @@
{
<NOT> c1=PUnaryCond(over,specs,lp,input)
{
- ExpressionOperator eOp = new LONot(lp, new OperatorKey(scope, getNextId()), c1);
+ ExpressionOperator eOp = new LONot(lp, new OperatorKey(scope, getNextId()));
lp.add(eOp);
log.debug("PNotCond: Added operator " + eOp.getClass().getName() + " " + eOp + " to logical plan " + lp);
lp.connect(c1, eOp);
@@ -1392,14 +1396,14 @@
{
c1=InfixExpr(over,specs,lp,input) <IS> [<NOT> {not = true;}] <NULL>
{
- ExpressionOperator eOp = new LOIsNull(lp, new OperatorKey(scope, getNextId()), c1);
+ ExpressionOperator eOp = new LOIsNull(lp, new OperatorKey(scope, getNextId()));
lp.add(eOp);
log.debug("PNullCond: Added operator " + eOp.getClass().getName() + " " + eOp + " to logical plan " + lp);
lp.connect(c1, eOp);
log.debug("PNullCond: Connected operator " + eOp.getClass().getName() + " " + eOp + " to " + c1 + " logical plan " + lp);
ExpressionOperator notNull = null;
if (not) {
- notNull = new LONot(lp, new OperatorKey(scope, getNextId()), eOp);
+ notNull = new LONot(lp, new OperatorKey(scope, getNextId()));
lp.add(notNull);
log.debug("PNullCond: Added operator " + notNull.getClass().getName() + " " + notNull + " to logical plan " + lp);
lp.connect(eOp, notNull);
@@ -1471,7 +1475,7 @@
listPlans.add(groupByPlan);
}
| <ANY> {
- es = new LOUserFunc(groupByPlan = new LogicalPlan(), new OperatorKey(scope, getNextId()), new FuncSpec(GFAny.class.getName()), null, DataType.INTEGER);
+ es = new LOUserFunc(groupByPlan = new LogicalPlan(), new OperatorKey(scope, getNextId()), new FuncSpec(GFAny.class.getName()), DataType.INTEGER);
groupByPlan.add(es);
log.debug("GroupItem: Added operator " + es.getClass().getName() + " " + es + " to logical plan " + groupByPlan);
listPlans.add(groupByPlan);
@@ -2426,9 +2430,9 @@
assertAtomic(lhs,true);
assertAtomic(rhs,true);
if (t.image.equals("+")){
- exprOp = new LOAdd(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ exprOp = new LOAdd(lp, new OperatorKey(scope, getNextId()));
}else{
- exprOp = new LOSubtract(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ exprOp = new LOSubtract(lp, new OperatorKey(scope, getNextId()));
}
lp.add(exprOp);
log.debug("AdditiveExpr: Added operator " + exprOp.getClass().getName() + " " + exprOp + " to logical plan " + lp);
@@ -2461,11 +2465,11 @@
assertAtomic(lhs,true);
assertAtomic(rhs,true);
if (t.image.equals("*")){
- exprOp = new LOMultiply(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ exprOp = new LOMultiply(lp, new OperatorKey(scope, getNextId()));
}else if (t.image.equals("/")){
- exprOp = new LODivide(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ exprOp = new LODivide(lp, new OperatorKey(scope, getNextId()));
}else {
- exprOp = new LOMod(lp, new OperatorKey(scope, getNextId()), lhs, rhs);
+ exprOp = new LOMod(lp, new OperatorKey(scope, getNextId()));
}
lp.add(exprOp);
log.debug("MultiplicativeExpr: Added operator " + exprOp.getClass().getName() + " " + exprOp + " to logical plan " + lp);
@@ -2496,7 +2500,8 @@
[LOOKAHEAD(2)"(" fs = TypeFieldSchema() {castRequired = true;}")"] exprOp = UnaryExpr(over, specs, lp, input)
{
if(castRequired) {
- cast = new LOCast(lp, new OperatorKey(scope, getNextId()), exprOp, fs.type);
+ cast = new LOCast(lp, new OperatorKey(scope, getNextId()), fs.type);
+ fs.alias = exprOp.getFieldSchema().alias;
cast.setFieldSchema(fs);
lp.add(cast);
log.debug("Added operator " + cast.getClass().getName() + " " + cast + " to logical plan " + lp);
@@ -2534,7 +2539,7 @@
{
"-" c1=CastExpr(over,specs,lp,input)
{
- ExpressionOperator eOp = new LONegative(lp, new OperatorKey(scope, getNextId()), c1);
+ ExpressionOperator eOp = new LONegative(lp, new OperatorKey(scope, getNextId()));
lp.add(eOp);
log.debug("NegativeExpr: Added operator " + eOp.getClass().getName() + " " + eOp + " to logical plan " + lp);
lp.connect(c1, eOp);
@@ -2581,7 +2586,7 @@
)
| ( "#" key = AtomDatum() {
assertAtomic(item, false);
- ExpressionOperator mapLookup = new LOMapLookup(lp, new OperatorKey(scope, getNextId()), item, key, DataType.BYTEARRAY, null);
+ ExpressionOperator mapLookup = new LOMapLookup(lp, new OperatorKey(scope, getNextId()), key, DataType.BYTEARRAY, null);
lp.add(mapLookup);
log.debug("BaseEvalSpec: Added operator " + mapLookup.getClass().getName() + " " + mapLookup + " to logical plan " + lp);
lp.connect(item, mapLookup);
@@ -2613,8 +2618,7 @@
)
{
- //ExpressionOperator bincond = new LOBinCond(lp, new OperatorKey(scope, getNextId()), conditionPlan, truePlan, falsePlan);
- ExpressionOperator bincond = new LOBinCond(lp, new OperatorKey(scope, getNextId()), cond, ifTrue, ifFalse);
+ ExpressionOperator bincond = new LOBinCond(lp, new OperatorKey(scope, getNextId()));
lp.add(bincond);
log.debug("BinCond: Added operator " + bincond.getClass().getName() + " " + bincond + " to logical plan " + lp);
lp.connect(cond, bincond);
@@ -2675,7 +2679,7 @@
default:
throw new ParseException("Received an unknown function type: " + funcType);
}
- userFunc = new LOUserFunc(lp, new OperatorKey(scope, getNextId()), funcSpec, args, type);
+ userFunc = new LOUserFunc(lp, new OperatorKey(scope, getNextId()), funcSpec, type);
} else {
throw new ParseException("Could not instantiate function: " + funcNameAlias);
}
@@ -3065,6 +3069,10 @@
( <BAG> "{" (fs = TypeSchemaTuple() | {} {fs = new Schema.FieldSchema(null, new Schema());}) "}" )
{
s = new Schema(fs);
+ // since this schema has tuple field schema which internally
+ // has a list of field schemas for the actual items in the bag
+ // an access to any field in the bag is a two level access
+ s.setTwoLevelAccessRequired(true);
if (null != t1) {
log.debug("BAG alias " + t1.image);
fs = new Schema.FieldSchema(t1.image, s, DataType.BAG);
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/schema/Schema.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/schema/Schema.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/schema/Schema.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/schema/Schema.java Wed Mar 25 00:42:40 2009
@@ -362,7 +362,7 @@
// Don't do the comparison if both embedded schemas are
// null. That will cause Schema.equals to return false,
// even though we want to view that as true.
- if (!(fschema.schema == null && fother.schema == null)) {
+ if (!(fschema.schema == null && fother.schema == null)) {
// compare recursively using schema
if (!Schema.equals(fschema.schema, fother.schema, false, relaxAlias)) {
return false ;
@@ -1155,6 +1155,32 @@
if (other == null) {
return false ;
}
+
+ /*
+ * Need to check for bags with schemas and bags with tuples that in turn have schemas.
+ * Retrieve the tuple schema of the bag if twoLevelAccessRequired
+ * Assuming that only bags exhibit this behavior and twoLevelAccessRequired is used
+ * with the right intentions
+ */
+ if(schema.isTwoLevelAccessRequired() || other.isTwoLevelAccessRequired()) {
+ if(schema.isTwoLevelAccessRequired()) {
+ try {
+ schema = schema.getField(0).schema;
+ } catch (FrontendException fee) {
+ return false;
+ }
+ }
+
+ if(other.isTwoLevelAccessRequired()) {
+ try {
+ other = other.getField(0).schema;
+ } catch (FrontendException fee) {
+ return false;
+ }
+ }
+
+ return Schema.equals(schema, other, relaxInner, relaxAlias);
+ }
if (schema.size() != other.size()) return false;
@@ -1352,7 +1378,13 @@
// create the merged field
// the mergedSubSchema can be true if allowIncompatibleTypes
- mergedFs = new FieldSchema(mergedAlias, mergedSubSchema) ;
+ try {
+ mergedFs = new FieldSchema(mergedAlias, mergedSubSchema, mergedType) ;
+ } catch (FrontendException e) {
+ int errCode = 2124;
+ String errMsg = "Internal Error: Unexpected error creating field schema";
+ throw new SchemaMergeException(errMsg, errCode, PigException.BUG, e);
+ }
}
outputList.add(mergedFs) ;
@@ -1540,7 +1572,7 @@
/**
* Recursively set NULL type to the specifid type in a schema
- * @param schema the schema whose NULL type has to be set
+ * @param s the schema whose NULL type has to be set
* @param t the specified type
*/
public static void setSchemaDefaultType(Schema s, byte t) {
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/InputOutputFileVisitor.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/InputOutputFileVisitor.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/InputOutputFileVisitor.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/InputOutputFileVisitor.java Wed Mar 25 00:42:40 2009
@@ -88,8 +88,7 @@
}
} catch (PlanValidationException pve) {
throw pve;
- }
- catch (IOException ioe) {
+ } catch (IOException ioe) {
byte errSrc = pigCtx.getErrorSource();
int errCode = 0;
switch(errSrc) {
@@ -108,6 +107,11 @@
+ filename + " will be stored ";
msgCollector.collect(msg, MessageType.Error) ;
throw new PlanValidationException(msg, errCode, errSrc, ioe);
+ } catch (Exception e) {
+ int errCode = 2116;
+ String msg = "Unexpected error. Could not check for the existence of the file(s): " + filename;
+ msgCollector.collect(msg, MessageType.Error) ;
+ throw new PlanValidationException(msg, errCode, PigException.BUG, e);
}
}
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/TypeCheckingVisitor.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/TypeCheckingVisitor.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/TypeCheckingVisitor.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/logicalLayer/validators/TypeCheckingVisitor.java Wed Mar 25 00:42:40 2009
@@ -37,6 +37,7 @@
import org.apache.pig.LoadFunc;
import org.apache.pig.Algebraic;
import org.apache.pig.PigException;
+import org.apache.pig.PigWarning;
import org.apache.pig.impl.PigContext;
import org.apache.pig.impl.logicalLayer.ExpressionOperator;
import org.apache.pig.impl.logicalLayer.FrontendException;
@@ -215,113 +216,6 @@
msgCollector.collect(msg, MessageType.Error);
throw new TypeCheckerException(msg, errCode, PigException.INPUT, fe) ;
}
-
- /*
- if (!pj.getSentinel()) {
-
- LogicalOperator op = pj.getExpression() ;
-
- if (!(op instanceof LOProject)) {
- throw new AssertionError("LOProject.getExpression() has to be "
- + "LOProject if it's not a sentinel") ;
- }
-
- // else
- LOProject innerProject = (LOProject) op ;
- resolveLOProjectType(innerProject) ;
-
- if ( (innerProject.getType() != DataType.BAG) &&
- (innerProject.getType() != DataType.TUPLE) ) {
- throw new AssertionError("Nested LOProject is for extracting "
- + " from TUPLE/BAG only") ;
- }
-
- // set type of this project
- pj.setType(innerProject.getType());
- Schema inputSchema = null ;
-
- try {
- inputSchema = innerProject.getSchema() ;
- }
- catch (FrontendException fe) {
- String msg = "Cannot get source schema into LOProject" ;
- msgCollector.collect(msg, MessageType.Error);
- VisitorException vse = new VisitorException(msg) ;
- vse.initCause(fe) ;
- throw new VisitorException(msg) ;
- }
-
- // extracting schema from projection
- List<FieldSchema> fsList = new ArrayList<FieldSchema>() ;
- try {
- for(int index: pj.getProjection()) {
- FieldSchema fs = null ;
- // typed input
- if (inputSchema != null) {
- fs = inputSchema.getField(index) ;
- FieldSchema newFs = new FieldSchema(fs.alias, fs.schema, fs.type) ;
- fsList.add(newFs) ;
- }
- // non-typed input
- else {
- FieldSchema newFs = new FieldSchema(null, DataType.BYTEARRAY) ;
- fsList.add(newFs) ;
- }
- }
- pj.setFieldSchema(new FieldSchema(null, new Schema(fsList), innerProject.getType()));
- }
- catch (FrontendException fe) {
- String msg = "Cannot get source schema into LOProject" ;
- msgCollector.collect(msg, MessageType.Error);
- VisitorException vse = new VisitorException(msg) ;
- vse.initCause(fe) ;
- throw new VisitorException(msg) ;
- }
- catch (ParseException pe) {
- String msg = "Cannot get source schema into LOProject" ;
- msgCollector.collect(msg, MessageType.Error);
- VisitorException vse = new VisitorException(msg) ;
- vse.initCause(pe) ;
- throw new VisitorException(msg) ;
- }
- }
- // if it's a sentinel, we just get the projected input type to it
- else {
- if (pj.getProjection().size() != 1) {
- throw new AssertionError("Sentinel LOProject can have only "
- + "1 projection") ;
- }
- LogicalOperator input = pj.getExpression() ;
- int projectedField = pj.getProjection().get(0) ;
- try {
- Schema schema = input.getSchema() ;
-
- if (schema != null) {
- FieldSchema fs = schema.getField(projectedField) ;
- pj.setFieldSchema(fs);
- }
- else {
- FieldSchema fs = new FieldSchema(null, DataType.BYTEARRAY) ;
- pj.setFieldSchema(fs);
- }
- }
- catch (FrontendException fe) {
- String msg = "Cannot get source schema into LOProject" ;
- msgCollector.collect(msg, MessageType.Error);
- VisitorException vse = new VisitorException(msg) ;
- vse.initCause(fe) ;
- throw new VisitorException(msg) ;
- }
- catch (ParseException pe) {
- String msg = "Cannot get source schema into LOProject" ;
- msgCollector.collect(msg, MessageType.Error);
- VisitorException vse = new VisitorException(msg) ;
- vse.initCause(pe) ;
- throw new VisitorException(msg) ;
- }
- }
- */
-
}
/**
@@ -374,23 +268,7 @@
}
private void insertCastForRegexp(LORegexp rg) throws VisitorException {
- LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan() ;
- collectCastWarning(rg, DataType.BYTEARRAY, DataType.CHARARRAY) ;
- OperatorKey newKey = genNewOperatorKey(rg) ;
- LOCast cast = new LOCast(currentPlan, newKey, rg.getOperand(), DataType.CHARARRAY) ;
- currentPlan.add(cast) ;
- currentPlan.disconnect(rg.getOperand(), rg) ;
- try {
- currentPlan.connect(rg.getOperand(), cast) ;
- currentPlan.connect(cast, rg) ;
- }
- catch (PlanException pe) {
- int errCode = 2059;
- String msg = "Problem with inserting cast operator for regular expression in plan.";
- throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
- }
- rg.setOperand(cast) ;
- this.visit(cast);
+ insertCast(rg, DataType.CHARARRAY, rg.getOperand());
}
public void visit(LOAnd binOp) throws VisitorException {
@@ -1117,52 +995,37 @@
private void insertLeftCastForBinaryOp(BinaryExpressionOperator binOp,
byte toType ) throws VisitorException {
- LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan() ;
- collectCastWarning(binOp,
- binOp.getLhsOperand().getType(),
- toType) ;
- OperatorKey newKey = genNewOperatorKey(binOp) ;
- LOCast cast = new LOCast(currentPlan, newKey, binOp.getLhsOperand(), toType) ;
- currentPlan.add(cast) ;
- currentPlan.disconnect(binOp.getLhsOperand(), binOp) ;
- try {
- currentPlan.connect(binOp.getLhsOperand(), cast) ;
- currentPlan.connect(cast, binOp) ;
- binOp.setLhsOperand(cast);
- }
- catch (PlanException pe) {
- int errCode = 2059;
- String msg = "Problem with inserting cast operator for binary conditional in plan.";
- throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
- }
- binOp.setLhsOperand(cast) ;
- this.visit(cast);
+ insertCast(binOp, toType, binOp.getLhsOperand());
}
private void insertRightCastForBinaryOp(BinaryExpressionOperator binOp,
byte toType ) throws VisitorException {
+ insertCast(binOp, toType, binOp.getRhsOperand());
+ }
+
+
+ private void insertCast(ExpressionOperator node,
+ byte toType, ExpressionOperator predecessor)
+ throws VisitorException {
LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan() ;
- collectCastWarning(binOp,
- binOp.getRhsOperand().getType(),
- toType) ;
- OperatorKey newKey = genNewOperatorKey(binOp) ;
- LOCast cast = new LOCast(currentPlan, newKey, binOp.getRhsOperand(), toType) ;
+ collectCastWarning(node, predecessor.getType(), toType);
+
+ OperatorKey newKey = genNewOperatorKey(node);
+ LOCast cast = new LOCast(currentPlan, newKey, toType) ;
currentPlan.add(cast) ;
- currentPlan.disconnect(binOp.getRhsOperand(), binOp) ;
try {
- currentPlan.connect(binOp.getRhsOperand(), cast) ;
- currentPlan.connect(cast, binOp) ;
- binOp.setRhsOperand(cast);
+ currentPlan.insertBetween(predecessor, cast, node);
}
catch (PlanException pe) {
int errCode = 2059;
- String msg = "Problem with inserting cast operator for binary conditional in plan.";
+ String msg = "Problem with inserting cast operator for " + node + " in plan.";
throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
}
- binOp.setRhsOperand(cast) ;
this.visit(cast);
}
+
+
/**
* Currently, there are two unaryOps: Neg and Not.
*/
@@ -1207,36 +1070,7 @@
}
private void insertCastForUniOp(UnaryExpressionOperator uniOp, byte toType) throws VisitorException {
- collectCastWarning(uniOp,
- uniOp.getOperand().getType(),
- toType) ;
- LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan() ;
- List<LogicalOperator> list = currentPlan.getPredecessors(uniOp) ;
- if (list==null) {
- int errCode = 1080;
- String msg = "Did not find inputs for operator: " + uniOp.getClass().getSimpleName();
- throw new TypeCheckerException(msg, errCode, PigException.INPUT) ;
- }
- // All uniOps at the moment only work with Expression input
- ExpressionOperator input = (ExpressionOperator) list.get(0) ;
- OperatorKey newKey = genNewOperatorKey(uniOp) ;
- LOCast cast = new LOCast(currentPlan, newKey, input, toType) ;
-
- currentPlan.add(cast);
- currentPlan.disconnect(input, uniOp) ;
- try {
- currentPlan.connect(input, cast) ;
- currentPlan.connect(cast, uniOp) ;
- }
- catch (PlanException pe) {
- int errCode = 2059;
- String msg = "Problem with inserting cast operator for unary operator in plan.";
- throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
- }
-
- uniOp.setOperand(cast);
- this.visit(cast);
-
+ insertCast(uniOp, toType, uniOp.getOperand());
}
// Currently there is no input type information support in UserFunc
@@ -1391,7 +1225,7 @@
" will be called with following argument types: " +
matchingSpec.getInputArgsSchema() + ". If you want to use " +
"different input argument types, please use explicit casts.";
- msgCollector.collect(msg, MessageType.Warning);
+ msgCollector.collect(msg, MessageType.Warning, PigWarning.USING_OVERLOADED_FUNCTION);
}
func.setFuncSpec(matchingSpec);
insertCastsForUDF(func, s, matchingSpec.getInputArgsSchema());
@@ -1794,35 +1628,10 @@
++i;
FieldSchema tFSch = tsLst.get(i);
if(fFSch.type==tFSch.type) {
- newArgs.add(args.get(i));
continue;
}
- collectCastWarning(udf,
- fFSch.type,
- tFSch.type);
- LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan();
- /*List<LogicalOperator> list = currentPlan.getPredecessors(udf);
- if (list == null) {
- throw new AssertionError("No input for " + udf.getClass());
- }*/
- // All uniOps at the moment only work with Expression input
- ExpressionOperator input = args.get(i);
- OperatorKey newKey = genNewOperatorKey(udf);
- LOCast cast = new LOCast(currentPlan, newKey, input, tFSch.type);
- currentPlan.add(cast);
- try {
- currentPlan.insertBetween(input, cast, udf);
- this.visit(cast);
- } catch (PlanException pe) {
- int errCode = 2059;
- String msg = "Problem with inserting cast operator for user defined function in plan.";
- throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
- } catch (VisitorException ve) {
- throw ve;
- }
+ insertCast(udf, tFSch.type, args.get(i));
}
- udf.setMArgs(newArgs);
-
}
/**
@@ -1879,9 +1688,11 @@
// Matching schemas if we're working with tuples
if (DataType.isSchemaType(lhsType)) {
try {
- if (!Schema.equals(binCond.getLhsOp().getSchema(), binCond.getRhsOp().getSchema(), false, true)) {
+ if (!Schema.FieldSchema.equals(binCond.getLhsOp().getFieldSchema(), binCond.getRhsOp().getFieldSchema(), false, true)) {
int errCode = 1048;
- String msg = "Two inputs of BinCond must have compatible schemas" ;
+ String msg = "Two inputs of BinCond must have compatible schemas."
+ + " left hand side: " + binCond.getLhsOp().getFieldSchema()
+ + " right hand side: " + binCond.getRhsOp().getFieldSchema();
msgCollector.collect(msg, MessageType.Error) ;
throw new TypeCheckerException(msg, errCode, PigException.INPUT) ;
}
@@ -1919,73 +1730,11 @@
}
private void insertLeftCastForBinCond(LOBinCond binCond, byte toType) throws VisitorException {
- LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan() ;
-
- collectCastWarning(binCond,
- binCond.getLhsOp().getType(),
- toType) ;
-
- OperatorKey newKey = genNewOperatorKey(binCond) ;
- LOCast cast = new LOCast(currentPlan, newKey, binCond.getLhsOp(), toType) ;
- // if we are casting a null constant, also set its field schema to the
- // field schema of the other operator in the bincond
- if (binCond.getLhsOp() instanceof LOConst
- && ((LOConst) binCond.getLhsOp()).getValue() == null) {
- try {
- cast.setFieldSchema(binCond.getRhsOp().getFieldSchema());
- } catch (FrontendException e) {
- int errCode = 1043;
- String msg = "Unable to retrieve field schema of operator";
- throw new TypeCheckerException(msg, errCode, PigException.INPUT, e);
- }
- }
- currentPlan.add(cast) ;
- try {
- currentPlan.insertBetween(binCond.getLhsOp(), cast, binCond);
- }
- catch (PlanException pe) {
- int errCode = 2059;
- String msg = "Problem with inserting cast operator for binary conditional in plan.";
- throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
- }
- binCond.setLhsOp(cast) ;
- this.visit(cast);
-
+ insertCast(binCond, toType, binCond.getLhsOp());
}
private void insertRightCastForBinCond(LOBinCond binCond, byte toType) throws VisitorException {
- LogicalPlan currentPlan = (LogicalPlan) mCurrentWalker.getPlan() ;
-
- collectCastWarning(binCond,
- binCond.getRhsOp().getType(),
- toType) ;
-
- OperatorKey newKey = genNewOperatorKey(binCond) ;
- LOCast cast = new LOCast(currentPlan, newKey, binCond.getRhsOp(), toType) ;
- // if we are casting a null constant, also set its field schema to the
- // field schema of the other operator in the bincond
- if (binCond.getRhsOp() instanceof LOConst
- && ((LOConst) binCond.getRhsOp()).getValue() == null) {
- try {
- cast.setFieldSchema(binCond.getLhsOp().getFieldSchema());
- } catch (FrontendException e) {
- int errCode = 1043;
- String msg = "Unable to retrieve field schema of operator";
- throw new TypeCheckerException(msg, errCode, PigException.INPUT, e);
- }
- }
- currentPlan.add(cast) ;
- try {
- currentPlan.insertBetween(binCond.getRhsOp(), cast, binCond) ;
- }
- catch (PlanException pe) {
- int errCode = 2059;
- String msg = "Problem with inserting cast operator for binary conditional in plan.";
- throw new TypeCheckerException(msg, errCode, PigException.BUG, pe);
- }
- binCond.setRhsOp(cast) ;
- this.visit(cast);
-
+ insertCast(binCond, toType, binCond.getRhsOp());
}
/**
@@ -2646,7 +2395,7 @@
ExpressionOperator currentOutput = (ExpressionOperator) leaves.get(0);
collectCastWarning(frj, currentOutput.getType(), toType);
OperatorKey newKey = genNewOperatorKey(currentOutput);
- LOCast cast = new LOCast(innerPlan, newKey, currentOutput, toType);
+ LOCast cast = new LOCast(innerPlan, newKey, toType);
innerPlan.add(cast);
try {
innerPlan.connect(currentOutput, cast);
@@ -2679,7 +2428,7 @@
ExpressionOperator currentOutput = (ExpressionOperator) leaves.get(0) ;
collectCastWarning(cg, currentOutput.getType(), toType) ;
OperatorKey newKey = genNewOperatorKey(currentOutput) ;
- LOCast cast = new LOCast(innerPlan, newKey, currentOutput, toType) ;
+ LOCast cast = new LOCast(innerPlan, newKey, toType) ;
innerPlan.add(cast) ;
try {
innerPlan.connect(currentOutput, cast) ;
@@ -2723,7 +2472,7 @@
// We just warn about mismatch type in non-strict mode
if (!strictMode) {
String msg = "COGroup by incompatible types results in ByteArray" ;
- msgCollector.collect(msg, MessageType.Warning) ;
+ msgCollector.collect(msg, MessageType.Warning, PigWarning.GROUP_BY_INCOMPATIBLE_TYPES) ;
groupType = DataType.BYTEARRAY ;
}
// We just die if in strict mode
@@ -2780,7 +2529,7 @@
// We just warn about mismatch type in non-strict mode
if (!strictMode) {
String msg = "COGroup by incompatible types results in ByteArray" ;
- msgCollector.collect(msg, MessageType.Warning) ;
+ msgCollector.collect(msg, MessageType.Warning, PigWarning.GROUP_BY_INCOMPATIBLE_TYPES) ;
fsList.get(j).type = DataType.BYTEARRAY ;
}
// We just die if in strict mode
@@ -3071,7 +2820,6 @@
castNeededCounter++ ;
LOCast cast = new LOCast(genPlan,
genNewOperatorKey(fromOp),
- project,
inputFieldType) ;
genPlan.add(cast) ;
try {
@@ -3138,9 +2886,36 @@
String originalTypeName = DataType.findTypeName(originalType) ;
String toTypeName = DataType.findTypeName(toType) ;
String opName = op.getClass().getSimpleName() ;
+ Enum kind = null;
+ switch(toType) {
+ case DataType.BAG:
+ kind = PigWarning.IMPLICIT_CAST_TO_BAG;
+ break;
+ case DataType.CHARARRAY:
+ kind = PigWarning.IMPLICIT_CAST_TO_CHARARRAY;
+ break;
+ case DataType.DOUBLE:
+ kind = PigWarning.IMPLICIT_CAST_TO_DOUBLE;
+ break;
+ case DataType.FLOAT:
+ kind = PigWarning.IMPLICIT_CAST_TO_FLOAT;
+ break;
+ case DataType.INTEGER:
+ kind = PigWarning.IMPLICIT_CAST_TO_INT;
+ break;
+ case DataType.LONG:
+ kind = PigWarning.IMPLICIT_CAST_TO_LONG;
+ break;
+ case DataType.MAP:
+ kind = PigWarning.IMPLICIT_CAST_TO_MAP;
+ break;
+ case DataType.TUPLE:
+ kind = PigWarning.IMPLICIT_CAST_TO_TUPLE;
+ break;
+ }
msgCollector.collect(originalTypeName + " is implicitly cast to "
+ toTypeName +" under " + opName + " Operator",
- MessageType.Warning) ;
+ MessageType.Warning, kind) ;
}
/***
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/CompilationMessageCollector.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/CompilationMessageCollector.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/CompilationMessageCollector.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/CompilationMessageCollector.java Wed Mar 25 00:42:40 2009
@@ -17,9 +17,13 @@
*/
package org.apache.pig.impl.plan ;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ArrayList ;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
/***
* This class is used for collecting all messages (error + warning) in
* compilation process. These messages are reported back to users
@@ -40,12 +44,18 @@
public static class Message {
private String msg = null ;
private MessageType msgType = MessageType.Unknown ;
+ private Enum kind = null;
public Message(String message, MessageType messageType) {
msg = message ;
msgType = messageType ;
}
+ public Message(String message, MessageType messageType, Enum kind) {
+ this(message, messageType);
+ this.kind = kind;
+ }
+
public String getMessage() {
return msg ;
}
@@ -53,6 +63,10 @@
public MessageType getMessageType() {
return msgType ;
}
+
+ public Enum getKind() {
+ return kind;
+ }
}
private List<Message> messageList = new ArrayList<Message>() ;
@@ -65,15 +79,23 @@
messageList.add(new Message(message, messageType)) ;
}
- public boolean hasError() {
+ public void collect(String message, MessageType messageType, Enum kind) {
+ messageList.add(new Message(message, messageType, kind)) ;
+ }
+
+ protected boolean hasMessageType(MessageType messageType) {
Iterator<Message> iter = iterator() ;
while(iter.hasNext()) {
- if (iter.next().getMessageType() == MessageType.Error) {
+ if (iter.next().getMessageType() == messageType) {
return true ;
}
}
return false ;
}
+
+ public boolean hasError() {
+ return hasMessageType(MessageType.Error);
+ }
public Iterator<Message> iterator() {
return messageList.iterator() ;
@@ -91,4 +113,76 @@
return messageList.get(i) ;
}
+ public Map<Enum, Long> getKindAggregate(MessageType messageType) {
+ Map<Enum, Long> aggMap = new HashMap<Enum, Long>();
+ Iterator<Message> iter = iterator() ;
+ while(iter.hasNext()) {
+ Message message = iter.next();
+ if (message.getMessageType() == messageType) {
+ Enum kind = message.getKind();
+ if(kind != null) {
+ Long count = aggMap.get(kind);
+ count = (count == null? 1 : ++count);
+ aggMap.put(kind, count);
+ }
+ }
+ }
+ return aggMap;
+ }
+
+ public static void logAggregate(Map<Enum, Long> aggMap, MessageType messageType, Log log) {
+ for(Enum e: aggMap.keySet()) {
+ Long count = aggMap.get(e);
+ if(count != null && count > 0) {
+ String message = "Encountered " + messageType + " " + e.toString() + " " + count + " time(s).";
+ logMessage(message, messageType, log);
+ }
+ }
+ }
+
+ public static void logMessages(CompilationMessageCollector messageCollector,
+ MessageType messageType, boolean aggregate, Log log) {
+ if(aggregate) {
+ Map<Enum, Long> aggMap = messageCollector.getKindAggregate(messageType);
+ logAggregate(aggMap, messageType, log);
+ } else {
+ Iterator<Message> messageIter = messageCollector.iterator();
+ while(messageIter.hasNext()) {
+ Message message = messageIter.next();
+ if(message.getMessageType() == messageType) {
+ logMessage(message.getMessage(), messageType, log);
+ }
+ }
+ }
+ }
+
+ public void logMessages(MessageType messageType, boolean aggregate, Log log) {
+ logMessages(this, messageType, aggregate, log);
+ }
+
+ public static void logAllMessages(CompilationMessageCollector messageCollector, Log log) {
+ Iterator<Message> messageIter = messageCollector.iterator();
+ while(messageIter.hasNext()) {
+ Message message = messageIter.next();
+ logMessage(message.getMessage(), message.getMessageType(), log);
+ }
+ }
+
+ public void logAllMessages(Log log) {
+ logAllMessages(this, log);
+ }
+
+ private static void logMessage(String messageString, MessageType messageType, Log log) {
+ switch(messageType) {
+ case Info:
+ log.info(messageString);
+ break;
+ case Warning:
+ log.warn(messageString);
+ break;
+ case Error:
+ log.error(messageString);
+ }
+ }
+
}
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/OperatorPlan.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/OperatorPlan.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/OperatorPlan.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/OperatorPlan.java Wed Mar 25 00:42:40 2009
@@ -172,13 +172,15 @@
log.error(pe.getMessage());
throw pe;
}
-
mFromEdges.put(from, to);
mToEdges.put(to, from);
}
/**
- * Remove an edge from between two nodes.
+ * Remove an edge from between two nodes.
+ * Use {@link org.apache.pig.impl.plan.OperatorPlan#insertBetween(Operator, Operator, Operator)}
+ * if disconnect is used in the process of inserting a new node between two nodes
+ * by calling disconnect followed by a connect.
* @param from Operator data would flow from.
* @param to Operator data would flow to.
* @return true if the nodes were connected according to the specified data
@@ -186,7 +188,7 @@
*/
public boolean disconnect(E from, E to) {
markDirty();
-
+
boolean sawNull = false;
if (mFromEdges.remove(from, to) == null) sawNull = true;
if (mToEdges.remove(to, from) == null) sawNull = true;
@@ -436,6 +438,7 @@
/**
* Given two connected nodes add another node between them.
+ * 'newNode' will be placed in same position in predecessor list as 'before' (old node).
* @param after Node to insert this node after
* @param newNode new node to insert. This node must have already been
* added to the plan.
@@ -458,6 +461,7 @@
mToEdges.put(newNode, after);
}
+ // replaces (src -> dst) entry in multiMap with (src -> replacement)
private boolean replaceNode(E src, E replacement, E dst, MultiMap<E, E> multiMap) {
Collection c = multiMap.get(src);
if (c == null) return false;
@@ -559,6 +563,104 @@
if (pred != null && succ != null) connect(pred, succ);
}
+ /**
+ * Remove a node in a way that connects the node's predecessor (if any)
+ * with the node's successors (if any). This function handles the
+ * case where the node has *one* predecessor and one or more successors.
+ * It replaces the predecessor in same position as node was in
+ * each of the successors predecessor list(getPredecessors()), to
+ * preserve input ordering
+ * for eg, it is used to remove redundant project(*) from plan
+ * which will have only one predecessor,but can have multiple success
+ * @param node Node to be removed
+ * @throws PlanException if the node has more than one predecessor
+ */
+ public void removeAndReconnectMultiSucc(E node) throws PlanException {
+
+ // Before:
+ // A (predecessor (only one) )
+ // / |
+ // X B(nodeB) Y(some predecessor of a Cn)
+ // / | \ /
+ // C1 C2 C3 ... (Successors)
+ // should become
+ // After:
+ // ___ A Y
+ // / / | \ /
+ // X C1 C2 C3 ...
+ // the variable names are from above example
+
+ E nodeB = node;
+ List<E> preds = getPredecessors(nodeB);
+ //checking pre-requisite conditions
+ if (preds == null || preds.size() != 1) {
+ Integer size = null;
+ if(preds != null)
+ size = preds.size();
+
+ PlanException pe = new PlanException("Attempt to remove " +
+ " and reconnect for node with " + size +
+ " predecessors.");
+ log.error(pe.getMessage());
+ throw pe;
+ }
+
+ //A and C
+ E nodeA = preds.get(0);
+ Collection<E> nodeC = mFromEdges.get(nodeB);
+
+ //checking pre-requisite conditions
+ if (nodeC == null || nodeC.size() == 0) {
+ PlanException pe = new PlanException("Attempt to remove " +
+ " and reconnect for node with no successors.");
+ log.error(pe.getMessage());
+ throw pe;
+ }
+
+
+ // replace B in A.succesors and add B.successors(ie C) to it
+ replaceAndAddSucessors(nodeA, nodeB);
+
+ // for all C(succs) , replace B(node) in predecessors, with A(pred)
+ for(E c: nodeC) {
+ Collection<E> sPreds = mToEdges.get(c);
+ ArrayList<E> newPreds = new ArrayList<E>(sPreds.size());
+ for(E p: sPreds){
+ if(p == nodeB){
+ //replace
+ newPreds.add(nodeA);
+ }
+ else{
+ newPreds.add(p);
+ }
+ }
+ mToEdges.removeKey(c);
+ mToEdges.put(c,newPreds);
+
+ }
+ remove(nodeB);
+ }
+
+ //removes entry for succ in list of successors of nd, and adds successors
+ // of succ in its place
+ // @param nd - parent node whose entry for succ needs to be replaced
+ // @param succ - see above
+ private void replaceAndAddSucessors(E nd, E succ) throws PlanException {
+ Collection<E> oldSuccs = mFromEdges.get(nd);
+ Collection<E> repSuccs = mFromEdges.get(succ);
+ ArrayList<E> newSuccs = new ArrayList<E>(oldSuccs.size() - 1 + repSuccs.size() );
+ for(E s: oldSuccs){
+ if(s == succ){
+ newSuccs.addAll(repSuccs);
+ }else{
+ newSuccs.add(s);
+ }
+ }
+ mFromEdges.removeKey(nd);
+ mFromEdges.put(nd,newSuccs);
+ }
+
+
public void dump(PrintStream ps) {
ps.println("Ops");
for (E op : mOps.keySet()) {
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/optimizer/OptimizerException.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/optimizer/OptimizerException.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/optimizer/OptimizerException.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/impl/plan/optimizer/OptimizerException.java Wed Mar 25 00:42:40 2009
@@ -18,8 +18,9 @@
package org.apache.pig.impl.plan.optimizer;
import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.plan.VisitorException;
-public class OptimizerException extends FrontendException {
+public class OptimizerException extends VisitorException {
/**
* Create a new OptimizerException with null as the error message.
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/Grunt.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/Grunt.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/Grunt.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/Grunt.java Wed Mar 25 00:42:40 2009
@@ -35,7 +35,7 @@
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.tools.pigscript.parser.*;
import org.apache.pig.impl.logicalLayer.parser.TokenMgrError;
-
+import org.apache.pig.impl.util.LogUtils;
public class Grunt
{
@@ -62,97 +62,28 @@
parser.setConsoleReader(c);
}
public void run() {
- //parser.parseContOnError();
- String trueString = "true";
- boolean verbose = trueString.equalsIgnoreCase(pig.getPigContext().getProperties().getProperty("verbose"));
- boolean append = false;
+ boolean verbose = "true".equalsIgnoreCase(pig.getPigContext().getProperties().getProperty("verbose"));
while(true) {
try {
parser.setInteractive(true);
parser.parseStopOnError();
break;
} catch(Throwable t) {
- writeLog(t, verbose, append);
- append = true;
+ LogUtils.writeLog(t, pig.getPigContext().getProperties().getProperty("pig.logfile"), log, verbose);
parser.ReInit(in);
}
}
}
public void exec() throws Throwable {
- String trueString = "true";
- boolean verbose = trueString.equalsIgnoreCase(pig.getPigContext().getProperties().getProperty("verbose"));
+ boolean verbose = "true".equalsIgnoreCase(pig.getPigContext().getProperties().getProperty("verbose"));
try {
parser.setInteractive(false);
parser.parseStopOnError();
} catch (Throwable t) {
- writeLog(t, verbose, false);
+ LogUtils.writeLog(t, pig.getPigContext().getProperties().getProperty("pig.logfile"), log, verbose);
throw (t);
}
}
-
- private void writeLog(Throwable t, boolean verbose, boolean append) {
-
- String message = null;
-
- if(t instanceof Exception) {
- Exception pe = Utils.getPermissionException((Exception)t);
- if (pe != null) {
- log.error("You don't have permission to perform the operation. Error from the server: " + pe.getMessage());
- }
- }
-
- PigException pigException = Utils.getPigException(t);
-
- if(pigException != null) {
- message = "ERROR " + pigException.getErrorCode() + ": " + pigException.getMessage();
- } else {
- if((t instanceof ParseException
- || t instanceof org.apache.pig.tools.pigscript.parser.TokenMgrError
- || t instanceof org.apache.pig.impl.logicalLayer.parser.TokenMgrError)) {
- message = "ERROR 1000: Error during parsing. " + t.getMessage();
- } else if (t instanceof RuntimeException) {
- message = "ERROR 2999: Unexpected internal error. " + t.getMessage();
- } else {
- message = "ERROR 2998: Unhandled internal error. " + t.getMessage();
- }
- }
-
-
- FileOutputStream fos = null;
- ByteArrayOutputStream bs = new ByteArrayOutputStream();
- t.printStackTrace(new PrintStream(bs));
- log.error(message);
-
- if(verbose) {
- log.error(bs.toString());
- }
-
- String logFileName = pig.getPigContext().getProperties().getProperty("pig.logfile");
-
- if(logFileName == null) {
- //if exec is invoked programmatically then logFileName will be null
- log.warn("There is no log file to write to");
- log.error(bs.toString());
- return;
- }
-
-
- File logFile = new File(logFileName);
- try {
- fos = new FileOutputStream(logFile, append);
- fos.write((message + "\n").getBytes("UTF-8"));
- fos.write(bs.toString().getBytes("UTF-8"));
- fos.close();
- if(verbose) {
- System.err.println("Details also at logfile: " + logFileName);
- } else {
- System.err.println("Details at logfile: " + logFileName);
- }
- } catch (IOException ioe) {
- log.warn("Could not write to log file: " + logFileName + " :" + ioe.getMessage());
- log.error(bs.toString());
- }
- }
}
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/GruntParser.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/GruntParser.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/GruntParser.java (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/tools/grunt/GruntParser.java Wed Mar 25 00:42:40 2009
@@ -63,6 +63,7 @@
import org.apache.pig.tools.pigscript.parser.PigScriptParser;
import org.apache.pig.tools.pigscript.parser.PigScriptParserTokenManager;
import org.apache.pig.tools.parameters.ParameterSubstitutionPreprocessor;
+import org.apache.pig.impl.util.LogUtils;
public class GruntParser extends PigScriptParser {
@@ -185,6 +186,9 @@
}
protected void processDescribe(String alias) throws IOException {
+ if(alias==null) {
+ alias = mPigServer.getPigContext().getLastAlias();
+ }
mPigServer.dumpSchema(alias);
}
@@ -240,6 +244,10 @@
discardBatch();
}
}
+
+ protected void printAliases() throws IOException {
+ mPigServer.printAliases();
+ }
protected void processRegister(String jar) throws IOException {
mPigServer.registerJar(jar);
Modified: hadoop/pig/branches/multiquery/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj (original)
+++ hadoop/pig/branches/multiquery/src/org/apache/pig/tools/pigscript/parser/PigScriptParser.jj Wed Mar 25 00:42:40 2009
@@ -62,6 +62,8 @@
abstract protected void quit();
+ abstract protected void printAliases() throws IOException;
+
abstract protected void processDescribe(String alias) throws IOException;
abstract protected void processExplain(String alias, String script, boolean isVerbose, String format, String target, List<String> params, List<String> files) throws IOException, ParseException;
@@ -133,6 +135,7 @@
TOKEN: {<COPYTOLOCAL: "copyToLocal">}
TOKEN: {<DUMP: "dump">}
TOKEN: {<DESCRIBE: "describe">}
+TOKEN: {<ALIASES: "aliases">}
TOKEN: {<EXPLAIN: "explain">}
TOKEN: {<HELP: "help">}
TOKEN: {<KILL: "kill">}
@@ -621,6 +624,8 @@
|
t = <DESCRIBE>
|
+ t = <ALIASES>
+ |
t = <EXPLAIN>
|
t = <HELP>
@@ -669,15 +674,5 @@
JAVACODE
void handle_invalid_command(int kind)
{
- ParseException e = generateParseException(); // generate the exception object.
-
- if (mInteractive) {
- System.out.println(e.toString()); // print the error message
- Token t = getNextToken();
-
- while (t.kind != kind)
- t = getNextToken();
- } else {
- throw e;
- }
+ throw generateParseException(); // throw the exception object.
}
Propchange: hadoop/pig/branches/multiquery/test/
------------------------------------------------------------------------------
--- svn:externals (added)
+++ svn:externals Wed Mar 25 00:42:40 2009
@@ -0,0 +1 @@
+bin http://svn.apache.org/repos/asf/hadoop/core/nightly/test-patch
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBestFitCast.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBestFitCast.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBestFitCast.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBestFitCast.java Wed Mar 25 00:42:40 2009
@@ -38,7 +38,7 @@
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
import org.apache.pig.impl.plan.VisitorException;
-import org.apache.pig.tools.grunt.Utils;
+import org.apache.pig.impl.util.LogUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -222,7 +222,7 @@
Iterator<Tuple> iter = pigServer.openIterator("B");
} catch(Exception e) {
exceptionCaused = true;
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertTrue(msg.contains("Multiple matching functions"));
assertTrue(msg.contains("{float,double}, {float,long}"));
@@ -307,7 +307,7 @@
Iterator<Tuple> iter = pigServer.openIterator("B");
}catch(Exception e) {
exceptionCaused = true;
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertTrue(msg.contains("Multiple matching functions"));
assertTrue(msg.contains("({float,double,long}, {float,long,double})"));
@@ -328,7 +328,7 @@
Iterator<Tuple> iter = pigServer.openIterator("B");
}catch(Exception e) {
exceptionCaused = true;
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertTrue(msg.contains("Multiple matching functions"));
assertTrue(msg.contains("({float,double,long}, {float,long,double})"));
@@ -349,7 +349,7 @@
Iterator<Tuple> iter = pigServer.openIterator("B");
}catch(Exception e) {
exceptionCaused = true;
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertTrue(msg.contains("Could not infer the matching function"));
}
@@ -461,7 +461,7 @@
Iterator<Tuple> iter = pigServer.openIterator("B");
}catch(Exception e) {
exceptionCaused = true;
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertTrue(msg.contains("Multiple matching functions"));
assertTrue(msg.contains("({float,double,long}, {float,long,double}"));
@@ -637,7 +637,7 @@
try{
pigServer.openIterator("B");
}catch (Exception e) {
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertEquals(true,msg.contains("as multiple or none of them fit"));
}
@@ -691,7 +691,7 @@
try{
pigServer.openIterator("B");
}catch (Exception e) {
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertEquals(true,msg.contains("Multiple matching functions"));
}
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBuiltin.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBuiltin.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBuiltin.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestBuiltin.java Wed Mar 25 00:42:40 2009
@@ -1047,18 +1047,16 @@
// Tuple size
- Tuple t5 = TupleFactory.getInstance().newTuple();
- t5.append(t1);
expected = new Long(3);
size = new TupleSize();
msg = "[Testing TupleSize on input type: Tuple]";
- assertTrue(msg, expected.equals(size.exec(t5)));
+ assertTrue(msg, expected.equals(size.exec(t1)));
// Test for ARITY function.
// It is depricated but we still need to make sure it works
ARITY arrity = new ARITY();
msg = "[Testing ARRITY on input type: Tuple]";
- assertTrue(msg, expected.equals(new Long(arrity.exec(t5))));
+ assertTrue(msg, expected.equals(new Long(arrity.exec(t1))));
}
// Builtin APPLY Functions
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestConversions.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestConversions.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestConversions.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestConversions.java Wed Mar 25 00:42:40 2009
@@ -70,8 +70,12 @@
public void testBytesToFloat() throws IOException
{
// valid floats
- String[] a = {"1", "-2.345", "12.12334567", "1.02e-2",".23344", "23.1234567897", ""};
- Float[] f = {1f, -2.345f, 12.12334567f, 1.02e-2f,.23344f, 23.1234567f}; // last case is a truncation case
+ String[] a = {"1", "-2.345", "12.12334567", "1.02e-2",".23344",
+ "23.1234567897", "12312.33f", "002312.33F", "1.02e-2f", ""};
+
+ Float[] f = {1f, -2.345f, 12.12334567f, 1.02e-2f,.23344f, 23.1234567f, // 23.1234567f is a truncation case
+ 12312.33f, 2312.33f, 1.02e-2f };
+
for (int j = 0; j < f.length; j++) {
byte[] b = a[j].getBytes();
assertEquals(f[j], ps.bytesToFloat(b));
@@ -112,8 +116,10 @@
public void testBytesToLong() throws IOException
{
// valid Longs
- String[] a = {"1", "-2345", "123456789012345678", "1.1", "-23.45", ""};
- Long[] la = {1L, -2345L, 123456789012345678L, 1L, -23L};
+ String[] a = {"1", "-2345", "123456789012345678", "1.1", "-23.45",
+ "21345345l", "3422342L", ""};
+ Long[] la = {1L, -2345L, 123456789012345678L, 1L, -23L,
+ 21345345L, 3422342L};
for (int i = 0; i < la.length; i++) {
byte[] b = a[i].getBytes();
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestEvalPipeline.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestEvalPipeline.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestEvalPipeline.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestEvalPipeline.java Wed Mar 25 00:42:40 2009
@@ -376,10 +376,7 @@
pigServer.registerQuery("B = ORDER A BY $0 using " + TupComp.class.getName() + ";");
}
}
- pigServer.store("B", tmpOutputFile);
-
- pigServer.registerQuery("A = load '" + tmpOutputFile + "';");
- Iterator<Tuple> iter = pigServer.openIterator("A");
+ Iterator<Tuple> iter = pigServer.openIterator("B");
String last = "";
HashSet<Integer> seen = new HashSet<Integer>();
if(!iter.hasNext()) fail("No Results obtained");
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestFRJoin.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestFRJoin.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestFRJoin.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestFRJoin.java Wed Mar 25 00:42:40 2009
@@ -98,13 +98,12 @@
}
@Override
- public DataBag exec(Tuple in) throws IOException {
+ public DataBag exec(Tuple input) throws IOException {
if(!isTblSetUp){
setUpHashTable();
isTblSetUp = true;
}
try {
- Tuple input = (Tuple)in.get(0);
String key = (String) input.get(keyField);
if(!replTbl.containsKey(key)) return BagFactory.getInstance().newDefaultBag();
return replTbl.get(key);
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestGrunt.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestGrunt.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestGrunt.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestGrunt.java Wed Mar 25 00:42:40 2009
@@ -28,7 +28,7 @@
import org.apache.pig.PigServer;
import org.apache.pig.impl.PigContext;
import org.apache.pig.tools.grunt.Grunt;
-import org.apache.pig.tools.grunt.Utils;
+import org.apache.pig.impl.util.LogUtils;
import java.io.ByteArrayInputStream;
import java.io.InputStreamReader;
@@ -110,7 +110,7 @@
try {
grunt.exec();
} catch (Exception e) {
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
String msg = (pe == null? e.getMessage(): pe.getMessage());
assertTrue(msg.contains("Encountered \" \";"));
}
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalOptimizer.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalOptimizer.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalOptimizer.java Wed Mar 25 00:42:40 2009
@@ -17,8 +17,11 @@
*/
package org.apache.pig.test;
+import java.io.File;
import java.io.FileInputStream;
+import java.io.PrintWriter;
+import org.apache.pig.ExecType;
import org.apache.pig.impl.logicalLayer.*;
import org.apache.pig.impl.logicalLayer.optimizer.*;
import org.apache.pig.test.utils.LogicalPlanTester;
@@ -58,6 +61,11 @@
optimizer.optimize();
}
+ public static void optimizePlan(LogicalPlan plan, ExecType mode) throws OptimizerException {
+ LogicalOptimizer optimizer = new LogicalOptimizer(plan, mode);
+ optimizer.optimize();
+ }
+
void compareWithGoldenFile(LogicalPlan plan, String filename) throws Exception
{
FileInputStream fis = new FileInputStream(filename);
@@ -153,6 +161,28 @@
optimizePlan(plan);
compareWithGoldenFile(plan, FILE_BASE_LOCATION + "optlimitplan7.dot");
}
+
+ @Test
+ //Limit in the local mode, need to make sure limit stays after a sort
+ public void testOPLimit8Optimizer() throws Exception {
+ planTester.buildPlan("A = load 'myfile';");
+ planTester.buildPlan("B = order A by $0;");
+ LogicalPlan plan = planTester.buildPlan("C = limit B 10;");
+ optimizePlan(plan, ExecType.LOCAL);
+ compareWithGoldenFile(plan, FILE_BASE_LOCATION + "optlimitplan8.dot");
+
+ }
+
+ @Test
+ //Limit in the local mode, need to make sure limit stays after a sort
+ public void testOPLimit9Optimizer() throws Exception {
+ planTester.buildPlan("A = load 'myfile';");
+ planTester.buildPlan("B = order A by $0;");
+ LogicalPlan plan = planTester.buildPlan("C = limit B 10;");
+ optimizePlan(plan);
+ compareWithGoldenFile(plan, FILE_BASE_LOCATION + "optlimitplan9.dot");
+
+ }
@Test
//Test to ensure that the right exception is thrown
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalPlanBuilder.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalPlanBuilder.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalPlanBuilder.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestLogicalPlanBuilder.java Wed Mar 25 00:42:40 2009
@@ -53,7 +53,7 @@
import org.apache.pig.impl.logicalLayer.parser.ParseException ;
import org.apache.pig.impl.util.MultiMap;
import org.apache.pig.test.utils.Identity;
-import org.apache.pig.tools.grunt.Utils;
+import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.PigException;
@@ -1539,33 +1539,33 @@
buildPlan("a = load 'a';");
buildPlan("store a into 'out' using PigStorage();");
}
-
+
@Test
public void testQuery104() {
// check that a field alias can be referenced
// by unambiguous free form alias, fully qualified alias
// and partially qualified unambiguous alias
- String query = "a = load 'st10k' as (name, age, gpa);" +
- "b = group a by name;" +
- "c = foreach b generate flatten(a);" +
- "d = filter c by name != 'fred';" +
- "e = group d by name;" +
- "f = foreach e generate flatten(d);" +
- "g = foreach f generate name, d::a::name, a::name;";
- buildPlan(query);
+ buildPlan( "a = load 'st10k' as (name, age, gpa);" );
+ buildPlan( "b = group a by name;" );
+ buildPlan("c = foreach b generate flatten(a);" );
+ buildPlan("d = filter c by name != 'fred';" );
+ buildPlan("e = group d by name;" );
+ buildPlan("f = foreach e generate flatten(d);" );
+ buildPlan("g = foreach f generate name, d::a::name, a::name;");
+
}
-
+
@Test
public void testQuery105() {
// test that the alias "group" can be used
// after a flatten(group)
- String query = "a = load 'st10k' as (name, age, gpa);" +
- "b = group a by name;" +
- "c = foreach b generate flatten(group), COUNT(a) as cnt;" +
- "d = foreach c generate group;";
- buildPlan(query);
+ buildPlan( "a = load 'st10k' as (name, age, gpa);" );
+ buildPlan("b = group a by name;" );
+ buildPlan("c = foreach b generate flatten(group), COUNT(a) as cnt;" );
+ buildPlan("d = foreach c generate group;");
+
}
-
+
@Test
public void testQuery106() throws FrontendException, ParseException {
LogicalPlan lp;
@@ -1796,89 +1796,129 @@
@Test
public void testNullConsArithExprs() {
- String query = "a = load 'a' as (x:int, y:double);" +
- "b = foreach a generate x + null, x * null, x / null, x - null, null % x, " +
- "y + null, y * null, y / null, y - null;";
- buildPlan(query);
+ buildPlan("a = load 'a' as (x:int, y:double);" );
+ buildPlan("b = foreach a generate x + null, x * null, x / null, x - null, null % x, " +
+ "y + null, y * null, y / null, y - null;"
+ );
}
-
+
@Test
public void testNullConsBincond1() {
- String query = "a = load 'a' as (x:int, y:double);" +
- "b = foreach a generate (2 > 1? null : 1), ( 2 < 1 ? null : 1), " +
- "(2 > 1 ? 1 : null), ( 2 < 1 ? 1 : null);";
- buildPlan(query);
+ buildPlan("a = load 'a' as (x:int, y:double);" );
+ buildPlan("b = foreach a generate (2 > 1? null : 1), ( 2 < 1 ? null : 1), " +
+ "(2 > 1 ? 1 : null), ( 2 < 1 ? 1 : null);"
+ );
}
-
+
@Test
public void testNullConsBincond2() {
- String query = "a = load 'a' as (x:int, y:double);" +
- "b = foreach a generate (null is null ? 1 : 2), ( null is not null ? 2 : 1);";
- buildPlan(query);
+ buildPlan("a = load 'a' as (x:int, y:double);" );
+ buildPlan("b = foreach a generate (null is null ? 1 : 2), ( null is not null ? 2 : 1);");
}
-
+
@Test
public void testNullConsForEachGenerate() {
- String query = "a = load 'a' as (x:int, y:double);" +
- "b = foreach a generate x, null, y, null;";
- buildPlan(query);
+ buildPlan("a = load 'a' as (x:int, y:double);" );
+ buildPlan("b = foreach a generate x, null, y, null;");
+
}
-
+
@Test
public void testNullConsOuterJoin() {
- String query = "a = load 'a' as (x:int, y:chararray);" +
- "b = load 'b' as (u:int, v:chararray);" +
- "c = cogroup a by x, b by u;" +
- "d = foreach c generate flatten((SIZE(a) == 0 ? null : a)), " +
- "flatten((SIZE(b) == 0 ? null : b));";
- buildPlan(query);
+ buildPlan("a = load 'a' as (x:int, y:chararray);" );
+ buildPlan("b = load 'b' as (u:int, v:chararray);" );
+ buildPlan("c = cogroup a by x, b by u;" );
+ buildPlan("d = foreach c generate flatten((SIZE(a) == 0 ? null : a)), " +
+ "flatten((SIZE(b) == 0 ? null : b));"
+ );
}
-
+
@Test
public void testNullConsConcatSize() {
- String query = "a = load 'a' as (x:int, y:double, str:chararray);" +
- "b = foreach a generate SIZE(null), CONCAT(str, null), " +
- "CONCAT(null, str);";
- buildPlan(query);
+ buildPlan("a = load 'a' as (x:int, y:double, str:chararray);" );
+ buildPlan("b = foreach a generate SIZE(null), CONCAT(str, null), " +
+ "CONCAT(null, str);"
+ );
}
@Test
public void testFilterUdfDefine() {
- String query = "define isempty IsEmpty(); a = load 'a' as (x:int, y:double, str:chararray);" +
- "b = filter a by isempty(*);";
- buildPlan(query);
+ buildPlan("define isempty IsEmpty();");
+ buildPlan("a = load 'a' as (x:int, y:double, str:chararray);");
+ buildPlan("b = filter a by isempty(*);");
}
-
+
@Test
public void testLoadUdfDefine() {
- String query = "define PS PigStorage(); a = load 'a' using PS as (x:int, y:double, str:chararray);" +
- "b = filter a by IsEmpty(*);";
- buildPlan(query);
+ buildPlan("define PS PigStorage();");
+ buildPlan("a = load 'a' using PS as (x:int, y:double, str:chararray);" );
+ buildPlan("b = filter a by IsEmpty(*);");
}
-
+
@Test
public void testLoadUdfConstructorArgDefine() {
- String query = "define PS PigStorage(':'); a = load 'a' using PS as (x:int, y:double, str:chararray);" +
- "b = filter a by IsEmpty(*);";
- buildPlan(query);
+ buildPlan("define PS PigStorage(':');");
+ buildPlan("a = load 'a' using PS as (x:int, y:double, str:chararray);" );
+ buildPlan("b = filter a by IsEmpty(*);");
}
-
+
@Test
public void testStoreUdfDefine() {
- String query = "define PS PigStorage(); a = load 'a' using PS as (x:int, y:double, str:chararray);" +
- "b = filter a by IsEmpty(*);" +
- "store b into 'x' using PS;" ;
- buildPlan(query);
+ buildPlan( "define PS PigStorage();");
+ buildPlan("a = load 'a' using PS as (x:int, y:double, str:chararray);" );
+ buildPlan("b = filter a by IsEmpty(*);" );
+ buildPlan("store b into 'x' using PS;");
}
-
+
@Test
public void testStoreUdfConstructorArgDefine() {
- String query = "define PS PigStorage(':'); a = load 'a' using PS as (x:int, y:double, str:chararray);" +
- "b = filter a by IsEmpty(*);" +
- "store b into 'x' using PS;" ;
- buildPlan(query);
+ buildPlan( "define PS PigStorage(':');");
+ buildPlan(" a = load 'a' using PS as (x:int, y:double, str:chararray);" );
+ buildPlan(" b = filter a by IsEmpty(*);" );
+ buildPlan(" store b into 'x' using PS;") ;
+
+ }
+
+ @Test
+ public void testCastAlias() {
+ buildPlan("a = load 'one.txt' as (x,y); ");
+ buildPlan("b = foreach a generate (int)x, (double)y;");
+ buildPlan("c = group b by x;");
+ }
+
+ @Test
+ public void testCast() {
+ buildPlan("a = load 'one.txt' as (x,y); " );
+ buildPlan("b = foreach a generate (int)$0, (double)$1;" );
+ buildPlan("c = group b by $0;");
}
+
+ @Test
+ public void testTokenizeSchema() throws FrontendException, ParseException {
+ LogicalPlan lp;
+ LOForEach foreach;
+
+ buildPlan("a = load 'one' as (f1: chararray);");
+ lp = buildPlan("b = foreach a generate TOKENIZE(f1);");
+ foreach = (LOForEach) lp.getLeaves().get(0);
+
+ Schema.FieldSchema tokenFs = new Schema.FieldSchema("token",
+ DataType.CHARARRAY);
+ Schema tupleSchema = new Schema(tokenFs);
+
+ Schema.FieldSchema tupleFs;
+ tupleFs = new Schema.FieldSchema("tuple_of_tokens", tupleSchema,
+ DataType.TUPLE);
+
+ Schema bagSchema = new Schema(tupleFs);
+ bagSchema.setTwoLevelAccessRequired(true);
+ Schema.FieldSchema bagFs = new Schema.FieldSchema(
+ "bag_of_tokenTuples",bagSchema, DataType.BAG);
+
+ assertTrue(Schema.equals(foreach.getSchema(), new Schema(bagFs), false, true));
+ }
+
private void printPlan(LogicalPlan lp) {
LOPrinter graphPrinter = new LOPrinter(System.err, lp);
System.err.println("Printing the logical plan");
@@ -1940,13 +1980,13 @@
// log.error(e);
//System.err.println("IOException Stack trace for query: " + query);
//e.printStackTrace();
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
fail("IOException: " + (pe == null? e.getMessage(): pe.getMessage()));
} catch (Exception e) {
log.error(e);
//System.err.println("Exception Stack trace for query: " + query);
//e.printStackTrace();
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
fail(e.getClass().getName() + ": " + (pe == null? e.getMessage(): pe.getMessage()) + " -- " + query);
}
return null;
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMRCompiler.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMRCompiler.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMRCompiler.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMRCompiler.java Wed Mar 25 00:42:40 2009
@@ -31,11 +31,11 @@
import org.apache.pig.FuncSpec;
import org.apache.pig.backend.executionengine.ExecException;
import org.apache.pig.builtin.AVG;
+import org.apache.pig.builtin.COUNT;
import org.apache.pig.builtin.SUM;
import org.apache.pig.data.DataType;
import org.apache.pig.data.Tuple;
import org.apache.pig.impl.PigContext;
-import org.apache.pig.impl.builtin.FindQuantiles;
import org.apache.pig.impl.builtin.GFCross;
import org.apache.pig.impl.logicalLayer.LogicalPlan;
import org.apache.pig.impl.plan.OperatorKey;
@@ -59,6 +59,17 @@
import org.junit.Before;
import org.junit.Test;
+/**
+ * Test cases to test the MRCompiler.
+ * VERY IMPORTANT NOTE: The tests here compare results with a
+ * "golden" set of outputs. In each testcase here, the operators
+ * generated have a random operator key which uses Java's Random
+ * class. So if there is a code change which changes the number of
+ * operators created in a plan, then not only will the "golden" file
+ * for that test case need to be changed, but also for the tests
+ * that follow it since the operator keys that will be generated through
+ * Random will be different.
+ */
public class TestMRCompiler extends junit.framework.TestCase {
// MiniCluster cluster = MiniCluster.buildCluster();
@@ -82,6 +93,13 @@
LogicalPlanTester planTester = new LogicalPlanTester() ;
+ // if for some reason, the golden files need
+ // to be regenerated, set this to true - THIS
+ // WILL OVERWRITE THE GOLDEN FILES - So use this
+ // with caution and only for the testcases you need
+ // and are sure of
+ private boolean generate = true;
+
@Before
public void setUp() throws ExecException {
GenPhyOp.setR(r);
@@ -701,6 +719,7 @@
PhysicalPlan ldFil1 = GenPhyOp.loadedFilter();
php.merge(ldFil1);
+ // set up order by *
String funcName = WeirdComparator.class.getName();
POUserComparisonFunc comparator = new POUserComparisonFunc(
new OperatorKey("", r.nextLong()), -1, null, new FuncSpec(funcName));
@@ -727,18 +746,21 @@
php.add(sort);
php.connect(ldFil1.getLeaves().get(0), sort);
-
+ // have a foreach which takes the sort output
+ // and send it two two udfs
List<String> udfs = new ArrayList<String>();
- udfs.add(FindQuantiles.class.getName());
+ udfs.add(COUNT.class.getName());
udfs.add(SUM.class.getName());
POForEach fe3 = GenPhyOp.topForEachOPWithUDF(udfs);
php.add(fe3);
php.connect(sort, fe3);
+ // add a group above the foreach
PhysicalPlan grpChain1 = GenPhyOp.grpChain();
php.merge(grpChain1);
php.connect(fe3,grpChain1.getRoots().get(0));
+
udfs.clear();
udfs.add(AVG.class.getName());
POForEach fe4 = GenPhyOp.topForEachOPWithUDF(udfs);
@@ -832,7 +854,9 @@
/**
* Test to ensure that the order by without parallel followed by a limit, i.e., top k
- * always produces the correct number of map reduce jobs
+ * always produces the correct number of map reduce jobs. In the testcase below since
+ * we are running the unit test locally, we will get reduce parallelism as 1. So we will
+ * NOT introduce the extra MR job to do a final limit
*/
@Test
public void testNumReducersInLimit() throws Exception {
@@ -850,7 +874,7 @@
mrOper = mrPlan.getSuccessors(mrOper).get(0);
++count;
}
- assertTrue(count == 4);
+ assertTrue(count == 3);
}
/**
@@ -908,6 +932,11 @@
ppp.print(baos);
compiledPlan = baos.toString();
+ if(generate ){
+ FileOutputStream fos = new FileOutputStream(expectedFile);
+ fos.write(baos.toByteArray());
+ return;
+ }
FileInputStream fis = new FileInputStream(expectedFile);
byte[] b = new byte[MAX_SIZE];
int len = fis.read(b);
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQuery.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQuery.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQuery.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQuery.java Wed Mar 25 00:42:40 2009
@@ -39,7 +39,7 @@
import org.apache.pig.impl.plan.Operator;
import org.apache.pig.impl.plan.OperatorPlan;
import org.apache.pig.tools.grunt.GruntParser;
-import org.apache.pig.tools.grunt.Utils;
+import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.tools.pigscript.parser.ParseException;
import org.junit.After;
import org.junit.Before;
@@ -1004,7 +1004,7 @@
Assert.assertNotNull(lp);
} catch (Exception e) {
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
if (pe != null) {
throw pe;
} else {
@@ -1074,7 +1074,7 @@
Assert.assertNotNull(mrp);
} catch (Exception e) {
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
if (pe != null) {
throw pe;
} else {
Modified: hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQueryLocal.java
URL: http://svn.apache.org/viewvc/hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQueryLocal.java?rev=758107&r1=758106&r2=758107&view=diff
==============================================================================
--- hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQueryLocal.java (original)
+++ hadoop/pig/branches/multiquery/test/org/apache/pig/test/TestMultiQueryLocal.java Wed Mar 25 00:42:40 2009
@@ -40,7 +40,7 @@
import org.apache.pig.impl.plan.Operator;
import org.apache.pig.impl.plan.OperatorPlan;
import org.apache.pig.tools.grunt.GruntParser;
-import org.apache.pig.tools.grunt.Utils;
+import org.apache.pig.impl.util.LogUtils;
import org.apache.pig.tools.pigscript.parser.ParseException;
import org.junit.After;
import org.junit.Before;
@@ -468,7 +468,7 @@
Assert.assertNotNull(lp);
} catch (Exception e) {
- PigException pe = Utils.getPigException(e);
+ PigException pe = LogUtils.getPigException(e);
if (pe != null) {
throw pe;
} else {