Author: gates
Date: Tue Jan 19 19:39:51 2010
New Revision: 900926
URL: http://svn.apache.org/viewvc?rev=900926&view=rev
Log:
PIG-1191: POCast throws exception for certain sequences of LOAD, FILTER,
FORACH. Checking in for Pradeep since he's out.
Modified:
hadoop/pig/trunk/CHANGES.txt
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCast.java
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/SchemaRemover.java
hadoop/pig/trunk/test/org/apache/pig/test/TestTypeCheckingValidator.java
Modified: hadoop/pig/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=900926&r1=900925&r2=900926&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Tue Jan 19 19:39:51 2010
@@ -78,6 +78,9 @@
BUG FIXES
+PIG-1191: POCast throws exception for certain sequences of LOAD, FILTER,
+ FORACH (pradeepkth via gates)
+
PIG-1171: Top-N queries produce incorrect results when followed by a cross
statement (rding via olgan)
PIG-1159: merge join right side table does not support comma seperated paths
Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCast.java
URL:
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCast.java?rev=900926&r1=900925&r2=900926&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCast.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/LOCast.java Tue Jan
19 19:39:51 2010
@@ -26,6 +26,7 @@
import org.apache.pig.impl.plan.PlanVisitor;
import org.apache.pig.impl.plan.VisitorException;
import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
import org.apache.pig.data.DataType;
public class LOCast extends ExpressionOperator {
@@ -34,6 +35,11 @@
private static final long serialVersionUID = 2L;
private FuncSpec mLoadFuncSpec = null;
+ // store field schema representing the schema
+ // in user specified casts -this is so that if
+ // field schema is unset and then getFieldSchema is called we still
+ // rebuild the fieldschema correctly as specified by the user in the script
+ private FieldSchema userSpecifiedFieldSchema;
/**
*
@@ -65,11 +71,22 @@
public Schema getSchema() {
return mSchema;
}
+
+
+ @Override
+ public void setFieldSchema(FieldSchema fs) throws FrontendException {
+ super.setFieldSchema(fs);
+ userSpecifiedFieldSchema = new Schema.FieldSchema(fs);
+ }
@Override
public Schema.FieldSchema getFieldSchema() throws FrontendException {
if(!mIsFieldSchemaComputed) {
- mFieldSchema = new Schema.FieldSchema(null, mType);
+ if(userSpecifiedFieldSchema != null) {
+ mFieldSchema = userSpecifiedFieldSchema;
+ } else {
+ mFieldSchema = new Schema.FieldSchema(null, mType);
+ }
Schema.FieldSchema parFs = getExpression().getFieldSchema();
String canonicalName = (parFs != null ? parFs.canonicalName :
null);
mFieldSchema.setParent(canonicalName, getExpression());
Modified:
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/SchemaRemover.java
URL:
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/SchemaRemover.java?rev=900926&r1=900925&r2=900926&view=diff
==============================================================================
---
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/SchemaRemover.java
(original)
+++
hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/SchemaRemover.java
Tue Jan 19 19:39:51 2010
@@ -37,6 +37,7 @@
* the logical binary expression operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(BinaryExpressionOperator binOp)
throws VisitorException {
binOp.unsetFieldSchema();
@@ -49,6 +50,7 @@
* the logical unary operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(UnaryExpressionOperator uniOp) throws
VisitorException {
uniOp.unsetFieldSchema();
super.visit(uniOp);
@@ -60,6 +62,7 @@
* the logical cogroup operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOCogroup cg) throws VisitorException {
cg.unsetSchema();
super.visit(cg);
@@ -71,6 +74,7 @@
* the logical sort operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOSort s) throws VisitorException {
s.unsetSchema();
super.visit(s);
@@ -82,6 +86,7 @@
* the logical limit operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOLimit limit) throws VisitorException {
limit.unsetSchema();
super.visit(limit);
@@ -94,6 +99,7 @@
* the logical filter operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOFilter filter) throws VisitorException {
filter.unsetSchema();
super.visit(filter);
@@ -105,6 +111,7 @@
* the logical split operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOSplit split) throws VisitorException {
split.unsetSchema();
super.visit(split);
@@ -116,6 +123,7 @@
* the logical foreach operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOForEach forEach) throws VisitorException {
forEach.unsetSchema();
super.visit(forEach);
@@ -128,6 +136,7 @@
* the user defined function
* @throws VisitorException
*/
+ @Override
protected void visit(LOUserFunc func) throws VisitorException {
func.unsetFieldSchema();
super.visit(func);
@@ -138,6 +147,7 @@
* the logical binCond operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOBinCond binCond) throws VisitorException {
binCond.unsetFieldSchema();
super.visit(binCond);
@@ -149,64 +159,185 @@
* the logical cast operator that has to be visited
* @throws VisitorException
*/
+ @Override
protected void visit(LOCast cast) throws VisitorException {
+ cast.unsetFieldSchema();
super.visit(cast);
}
+
/**
*
* @param regexp
* the logical regexp operator that has to be visited
* @throws ParseException
*/
+ @Override
protected void visit(LORegexp regexp) throws VisitorException {
regexp.unsetFieldSchema();
super.visit(regexp);
}
+ @Override
protected void visit(LOLoad load) throws VisitorException{
// Don't remove load's schema, it's not like it will change. And we
// don't have a way to recover it.
super.visit(load);
}
+ @Override
protected void visit(LOStore store) throws VisitorException{
store.unsetSchema();
super.visit(store);
}
+ @Override
protected void visit(LOConst c) throws VisitorException{
c.unsetSchema();
super.visit(c);
}
+ @Override
protected void visit(LOUnion u) throws VisitorException {
u.unsetSchema();
super.visit(u);
}
+ @Override
protected void visit(LOSplitOutput sop) throws VisitorException {
sop.unsetSchema();
super.visit(sop);
}
+ @Override
protected void visit(LODistinct dt) throws VisitorException {
dt.unsetSchema();
super.visit(dt);
}
+ @Override
protected void visit(LOCross cs) throws VisitorException {
cs.unsetSchema();
super.visit(cs);
}
+ @Override
protected void visit(LOProject project) throws VisitorException {
project.unsetFieldSchema();
super.visit(project);
}
+ @Override
protected void visit(LOJoin join) throws VisitorException {
join.unsetSchema();
super.visit(join);
}
+
+ @Override
+ protected void visit(ExpressionOperator op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOAdd op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOAnd binOp) throws VisitorException {
+ binOp.unsetFieldSchema();
+ super.visit(binOp);
+ }
+
+ @Override
+ public void visit(LODivide op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOEqual op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOGreaterThan op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOGreaterThanEqual op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOIsNull uniOp) throws VisitorException {
+ uniOp.unsetFieldSchema();
+ super.visit(uniOp);
+ }
+
+ @Override
+ public void visit(LOLesserThan op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOLesserThanEqual op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOMapLookup op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOMod op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOMultiply op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LONegative op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LONot uniOp) throws VisitorException {
+ uniOp.unsetFieldSchema();
+ super.visit(uniOp);
+ }
+
+ @Override
+ public void visit(LONotEqual op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
+
+ @Override
+ public void visit(LOOr binOp) throws VisitorException {
+ binOp.unsetFieldSchema();
+ super.visit(binOp);
+ }
+
+ @Override
+ public void visit(LOSubtract op) throws VisitorException {
+ op.unsetFieldSchema();
+ super.visit(op);
+ }
}
Modified:
hadoop/pig/trunk/test/org/apache/pig/test/TestTypeCheckingValidator.java
URL:
http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestTypeCheckingValidator.java?rev=900926&r1=900925&r2=900926&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestTypeCheckingValidator.java
(original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestTypeCheckingValidator.java
Tue Jan 19 19:39:51 2010
@@ -5487,6 +5487,79 @@
LogicalPlan plan = planTester.buildPlan("c = foreach b generate
(chararray)viewinfo#'pos' as position;") ;
// validate
+ runTypeCheckingValidator(plan);
+
+ checkLoaderInCasts(plan, "org.apache.pig.builtin.PigStorage");
+ }
+
+ /**
+ * test various scenarios with two level map lookup
+ */
+ @Test
+ public void testTwolevelMapLookupLineage() throws Exception {
+ List<String[]> queries = new ArrayList<String[]>();
+ // CASE 1: LOAD -> FILTER -> FOREACH -> LIMIT -> STORE
+ queries.add(new String[] {"sds = LOAD '/my/data/location' " +
+ "AS (simpleFields:map[], mapFields:map[],
listMapFields:map[]);",
+ "queries = FILTER sds BY
mapFields#'page_params'#'query' " +
+ "is NOT NULL;",
+ "queries_rand = FOREACH queries GENERATE " +
+ "(CHARARRAY) (mapFields#'page_params'#'query') AS
query_string;",
+ "queries_limit = LIMIT queries_rand 100;",
+ "STORE queries_limit INTO 'out';"});
+ // CASE 2: LOAD -> FOREACH -> FILTER -> LIMIT -> STORE
+ queries.add(new String[]{"sds = LOAD '/my/data/location' " +
+ "AS (simpleFields:map[], mapFields:map[],
listMapFields:map[]);",
+ "queries_rand = FOREACH sds GENERATE " +
+ "(CHARARRAY) (mapFields#'page_params'#'query') AS
query_string;",
+ "queries = FILTER queries_rand BY query_string IS NOT
null;",
+ "queries_limit = LIMIT queries 100;",
+ "STORE queries_limit INTO 'out';"});
+ // CASE 3: LOAD -> FOREACH -> FOREACH -> FILTER -> LIMIT -> STORE
+ queries.add(new String[]{"sds = LOAD '/my/data/location' " +
+ "AS (simpleFields:map[], mapFields:map[],
listMapFields:map[]);",
+ "params = FOREACH sds GENERATE " +
+ "(map[]) (mapFields#'page_params') AS params;",
+ "queries = FOREACH params " +
+ "GENERATE (CHARARRAY) (params#'query') AS
query_string;",
+ "queries_filtered = FILTER queries BY query_string IS
NOT null;",
+ "queries_limit = LIMIT queries_filtered 100;",
+ "STORE queries_limit INTO 'out';"});
+ // CASE 4: LOAD -> FOREACH -> FOREACH -> LIMIT -> STORE
+ queries.add(new String[]{"sds = LOAD '/my/data/location' " +
+ "AS (simpleFields:map[], mapFields:map[],
listMapFields:map[]);",
+ "params = FOREACH sds GENERATE" +
+ " (map[]) (mapFields#'page_params') AS params;",
+ "queries = FOREACH params GENERATE " +
+ "(CHARARRAY) (params#'query') AS query_string;",
+ "queries_limit = LIMIT queries 100;",
+ "STORE queries_limit INTO 'out';"});
+ // CASE 5: LOAD -> FOREACH -> FOREACH -> FOREACH -> LIMIT -> STORE
+ queries.add(new String[]{"sds = LOAD '/my/data/location' " +
+ "AS (simpleFields:map[], mapFields:map[],
listMapFields:map[]);",
+ "params = FOREACH sds GENERATE " +
+ "(map[]) (mapFields#'page_params') AS params;",
+ "queries = FOREACH params GENERATE " +
+ "(CHARARRAY) (params#'query') AS query_string;",
+ "rand_queries = FOREACH queries GENERATE query_string as
query;",
+ "queries_limit = LIMIT rand_queries 100;",
+ "STORE rand_queries INTO 'out';"});
+
+ for (String[] query: queries) {
+ LogicalPlan lp = null;
+ for (String queryLine : query) {
+ lp = planTester.buildPlan(queryLine);
+ }
+
+ // validate
+ runTypeCheckingValidator(lp);
+ checkLoaderInCasts(lp, "org.apache.pig.builtin.PigStorage");
+
+ }
+ }
+
+ private void runTypeCheckingValidator(LogicalPlan plan) throws
+ PlanValidationException {
CompilationMessageCollector collector = new
CompilationMessageCollector() ;
TypeCheckingValidator typeValidator = new TypeCheckingValidator() ;
typeValidator.validate(plan, collector) ;
@@ -5498,12 +5571,16 @@
if (collector.hasError()) {
throw new AssertionError("Expect no error") ;
}
-
+ }
+
+ private void checkLoaderInCasts(LogicalPlan plan, String loaderClassName)
+ throws VisitorException {
CastFinder cf = new CastFinder(plan);
cf.visit();
List<LOCast> casts = cf.casts;
for (LOCast cast : casts) {
-
assertTrue(cast.getLoadFuncSpec().getClassName().startsWith("org.apache.pig.builtin.PigStorage"));
+ assertTrue(cast.getLoadFuncSpec().getClassName().startsWith(
+ loaderClassName));
}
}
@@ -5662,6 +5739,7 @@
*/
public static class TestBinCondFieldSchema extends EvalFunc<DataBag> {
//no-op exec method
+ @Override
public DataBag exec(Tuple input) {
return null;
}
