You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by th...@apache.org on 2011/03/04 19:15:14 UTC
svn commit: r1078085 [3/12] - in /pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/
src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/ sr...
Modified: pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalSchema.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalSchema.java?rev=1078085&r1=1078084&r2=1078085&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalSchema.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalSchema.java Fri Mar 4 18:15:11 2011
@@ -19,15 +19,13 @@
package org.apache.pig.newplan.logical.relational;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
-import java.util.Map;
+import org.apache.pig.PigException;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.logicalLayer.FrontendException;
-import org.apache.pig.impl.util.Pair;
import org.apache.pig.newplan.logical.expression.LogicalExpression;
/**
@@ -61,17 +59,34 @@ public class LogicalSchema {
* or both null schema. Alias and uid are not checked.
*/
public boolean isEqual(Object other) {
+ return isEqual(other, false);
+ }
+
+ /**
+ * Equality is defined as having the same type and either the same schema
+ * or both null schema. if compareAlias argument is set to true, alias
+ * is also compared.
+ * @param other
+ * @param compareAlias
+ * @return true if equal
+ */
+ public boolean isEqual(Object other, boolean compareAlias) {
if (other instanceof LogicalFieldSchema) {
LogicalFieldSchema ofs = (LogicalFieldSchema)other;
+ if(compareAlias){
+ if(alias != null && !alias.equals(ofs.alias))
+ return false;
+ }
if (type != ofs.type) return false;
if (schema == null && ofs.schema == null) return true;
if (schema == null) return false;
- else return schema.isEqual(ofs.schema);
+ else return schema.isEqual(ofs.schema, compareAlias);
} else {
return false;
}
}
-
+
+
public String toString(boolean verbose) {
String uidString = "";
if (verbose)
@@ -121,6 +136,13 @@ public class LogicalSchema {
}
return true;
}
+ /**
+ * Adds the uid from FieldSchema argument to this FieldSchema
+ * If the argument is null, it stamps this FieldSchema with uid
+ * @param uidOnlyFieldSchema
+ * @return FieldSchema
+ * @throws FrontendException
+ */
public LogicalSchema.LogicalFieldSchema mergeUid(LogicalFieldSchema uidOnlyFieldSchema) throws FrontendException {
if (uidOnlyFieldSchema!=null && compatible(uidOnlyFieldSchema)) {
this.uid = uidOnlyFieldSchema.uid;
@@ -148,6 +170,17 @@ public class LogicalSchema {
}
}
+
+ /**
+ * Rest uid of this fieldschema and inner schema
+ */
+ public void resetUid(){
+ uid = LogicalExpression.getNextUid();
+ if(schema != null){
+ schema.resetUid();
+ }
+ }
+
public LogicalFieldSchema cloneUid() {
LogicalFieldSchema resultFs = null;
if (schema==null) {
@@ -214,6 +247,73 @@ public class LogicalSchema {
return true ;
}
+
+ /**
+ * Check if FieldSchema inFs is castable to outFs
+ * @param inFs
+ * @param outFs
+ * @return true if it is castable
+ */
+ public static boolean castable(LogicalFieldSchema inFs,
+ LogicalFieldSchema outFs) {
+
+ if(outFs == null && inFs == null) {
+ return false;
+ }
+
+ if (outFs == null) {
+ return false ;
+ }
+
+ if (inFs == null) {
+ return false ;
+ }
+ byte inType = inFs.type;
+ byte outType = outFs.type;
+
+ if (DataType.isSchemaType(outFs.type)) {
+ if(inType == DataType.BYTEARRAY) {
+ //good
+ } else if (inType == outType) {
+ // 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 (!(outFs.schema == null && inFs.schema == null)) {
+ // compare recursively using schema
+ if (!LogicalSchema.castable(outFs.schema, inFs.schema)) {
+ return false ;
+ }
+ }
+ } else {
+ return false;
+ }
+ } else {
+ if (inType == outType) {
+ //good
+ }
+ else if (DataType.isNumberType(inType) &&
+ DataType.isNumberType(outType) ) {
+ //good
+ }
+ else if (inType == DataType.BYTEARRAY) {
+ //good
+ }
+ else if ( ( DataType.isNumberType(inType) ||
+ inType == DataType.CHARARRAY
+ ) &&
+ ( (outType == DataType.CHARARRAY) ||
+ (outType == DataType.BYTEARRAY) ||
+ (DataType.isNumberType(outType))
+ )
+ ) {
+ //good
+ } else {
+ return false;
+ }
+ }
+
+ return true ;
+ }
/***
* Merge two LogicalFieldSchema, the behavior of merge depends on mode.
@@ -283,8 +383,12 @@ public class LogicalSchema {
if (fs1.alias==null)
mergedAlias = fs2.alias;
+ else if (fs2.alias==null)
+ mergedAlias = fs1.alias;
else {
- mergedAlias = fs1.alias; // If both schema have alias, the first one win
+ mergedAlias = mergeNameSpacedAlias(fs1.alias, fs2.alias);
+ if (mergedAlias==null)
+ mergedAlias = fs1.alias;
}
if (DataType.isSchemaType(mergedType)) {
@@ -302,6 +406,13 @@ public class LogicalSchema {
mergedSubSchema = LogicalSchema.merge(fs1.schema, fs2.schema, MergeMode.UnionInner);
}
} catch (FrontendException e) {
+ if(fs1.type == DataType.BAG && fs2.type == DataType.BAG){
+ //create an empty tuple as subschema
+ mergedSubSchema = new LogicalSchema();
+ mergedSubSchema.addField(new LogicalFieldSchema(null, new LogicalSchema(), DataType.TUPLE));
+ }else if(fs1.type == DataType.TUPLE && fs2.type == DataType.TUPLE){
+ mergedSubSchema = new LogicalSchema();
+ }
// If inner schema is not compatible, mergedSubSchema set to null
}
}
@@ -327,72 +438,198 @@ public class LogicalSchema {
LogicalFieldSchema mergedFS = new LogicalFieldSchema(mergedAlias, mergedSubSchema, mergedType);
return mergedFS;
}
+
+ /***
+ * Old Pig field schema does not require a tuple schema inside a bag;
+ * Now it is required to have that; this method is to fill the gap
+ */
+ public void normalize() {
+ if (type==DataType.BAG) {
+ if (schema!=null) {
+ // Check if the BAG has a tuple field
+ if (schema.size()!=1 || schema.getField(0).type!=DataType.TUPLE) {
+ LogicalSchema tupleSchema = new LogicalSchema();
+ for (LogicalFieldSchema innerFs : schema.getFields()) {
+ tupleSchema.addField(innerFs);
+ }
+ schema = new LogicalSchema();
+ schema.addField(new LogicalFieldSchema(null, tupleSchema, DataType.TUPLE));
+ }
+ }
+ }
+ if (schema!=null) {
+ for (LogicalFieldSchema fs : schema.getFields()) {
+ fs.normalize();
+ }
+ }
+ }
}
private List<LogicalFieldSchema> fields;
- private Map<String, Pair<Integer, Boolean>> aliases;
public LogicalSchema() {
fields = new ArrayList<LogicalFieldSchema>();
- aliases = new HashMap<String, Pair<Integer, Boolean>>();
}
-
+
+
+ /**
+ * Reset uids of all fieldschema that the schema contains
+ */
+ public void resetUid() {
+ for(LogicalFieldSchema fs : fields){
+ fs.resetUid();
+ }
+ }
+
+ /**
+ * Recursively compare two schemas to check if the input schema
+ * can be cast to the cast schema
+ * @param inSch schema of the cast input
+ * @param outSch schema of the cast operator
+ * @return true if castable
+ */
+ public static boolean castable(LogicalSchema inSch, LogicalSchema outSch) {
+ // If both of them are null, they are castable
+ if ((outSch == null) && (inSch == null)) {
+ return false ;
+ }
+
+ // otherwise
+ if (outSch == null) {
+ return false ;
+ }
+
+ if (inSch == null) {
+ return false ;
+ }
+
+ if (outSch.size() > inSch.size()) return false;
+
+
+ Iterator<LogicalFieldSchema> i = outSch.fields.iterator();
+ Iterator<LogicalFieldSchema> j = inSch.fields.iterator();
+
+ while (i.hasNext()) {
+ //iterate only for the number of fields in cast
+
+ LogicalFieldSchema outFs = i.next() ;
+ LogicalFieldSchema inFs = j.next() ;
+
+ // Compare recursively using field schema
+ if (!LogicalFieldSchema.castable(inFs, outFs)) {
+ return false ;
+ }
+
+ }
+ return true;
+
+ }
+
/**
* Add a field to this schema.
* @param field to be added to the schema
*/
public void addField(LogicalFieldSchema field) {
fields.add(field);
- if (field.alias != null && !field.alias.equals("")) {
- // put the full name of this field into aliases map
- // boolean in the pair indicates if this alias is full name
- aliases.put(field.alias, new Pair<Integer, Boolean>(fields.size()-1, true));
- int index = 0;
-
- // check and put short names into alias map if there is no conflict
-
- while(index != -1) {
- index = field.alias.indexOf("::", index);
- if (index != -1) {
- String a = field.alias.substring(index+2);
- if (aliases.containsKey(a)) {
- // remove conflict if the conflict is not full name
- // we can never remove full name
- if (!aliases.get(a).second) {
- aliases.remove(a);
- }
- }else{
- // put alias into map and indicate it is a short name
- aliases.put(a, new Pair<Integer, Boolean>(fields.size()-1, false));
- }
-
- index = index +2;
- }
- }
- }
}
/**
* Fetch a field by alias
* @param alias
* @return field associated with alias, or null if no such field
+ * @throws FrontendException
*/
- public LogicalFieldSchema getField(String alias) {
- Pair<Integer, Boolean> p = aliases.get(alias);
- if (p == null) {
+ public LogicalFieldSchema getField(String alias) throws FrontendException {
+ LogicalFieldSchema result = null;
+ //first look for an exact match
+ for (LogicalFieldSchema fs : fields) {
+ if (fs.alias!=null && fs.alias.equals(alias) ) {
+ if (result==null) {
+ result = fs;
+ }
+ else {
+ StringBuilder sb = new StringBuilder("Found more than one match: " + result.alias + ", " + fs.alias);
+ throw new FrontendException(sb.toString(), 1025);
+ }
+ }
+ }
+ if(result != null){
+ return result;
+ }
+ //if no exact match is found, look for matches for scoped aliases
+ for (LogicalFieldSchema fs : fields) {
+ if (fs.alias!=null && fs.alias.matches(".*::"+alias+"$") ) {
+ if (result==null) {
+ result = fs;
+ }
+ else {
+ StringBuilder sb = new StringBuilder("Found more than one match: " + result.alias + ", " + fs.alias);
+ throw new FrontendException(sb.toString(), 1025);
+ }
+ }
+ }
+
+ return result;
+ }
+
+ /**
+ * Given an alias name, find the associated LogicalFieldSchema. If exact name is
+ * not found see if any field matches the part of the 'namespaced' alias.
+ * eg. if given alias is nm::a , and schema is (a,b). It will return
+ * FieldSchema of a.
+ * if given alias is nm::a and schema is (nm2::a, b), it will return null
+ * @param alias Alias to look up.
+ * @return LogicalFieldSchema, or null if no such alias is in this tuple.
+ */
+ public LogicalFieldSchema getFieldSubNameMatch(String alias) throws FrontendException {
+ if(alias == null)
return null;
+ LogicalFieldSchema fs = getField(alias);
+ if(fs != null){
+ return fs;
+ }
+ //fs is null
+ final String sep = "::";
+ ArrayList<LogicalFieldSchema> matchedFieldSchemas = new ArrayList<LogicalFieldSchema>();
+ if(alias.contains(sep)){
+ for(LogicalFieldSchema field : fields) {
+ if(alias.endsWith(sep + field.alias)){
+ matchedFieldSchemas.add(field);
+ }
+ }
+ }
+ if(matchedFieldSchemas.size() > 1){
+ boolean hasNext = false;
+ StringBuilder sb = new StringBuilder("Found more than one " +
+ "sub alias name match: ");
+ for (LogicalFieldSchema matchFs : matchedFieldSchemas) {
+ if(hasNext) {
+ sb.append(", ");
+ } else {
+ hasNext = true;
+ }
+ sb.append(matchFs.alias);
+ }
+ int errCode = 1116;
+ throw new FrontendException(sb.toString(), errCode, PigException.INPUT);
+ }else if(matchedFieldSchemas.size() == 1){
+ fs = matchedFieldSchemas.get(0);
}
- return fields.get(p.first);
+ return fs;
}
public int getFieldPosition(String alias) {
- Pair<Integer, Boolean> p = aliases.get( alias );
- if( p == null ) {
+ LogicalFieldSchema fs = null;
+ try {
+ fs = getField(alias);
+ } catch (FrontendException e) {
+ }
+ if( fs == null ) {
return -1;
}
- return p.first;
+ return fields.indexOf(fs);
}
/**
@@ -422,20 +659,34 @@ public class LogicalSchema {
/**
* Two schemas are equal if they are of equal size and their fields
- * schemas considered in order are equal.
+ * schemas considered in order are equal. This function does
+ * not compare the alias of the fields.
*/
public boolean isEqual(Object other) {
+ return isEqual(other, false);
+ }
+
+
+
+ /**
+ * Two schemas are equal if they are of equal size and their fields
+ * schemas considered in order are equal. If compareAlias argument is
+ * set to true, the alias of the fields are also compared.
+ * @param other
+ * @param compareAlias
+ * @return true if equal
+ */
+ public boolean isEqual(Object other, boolean compareAlias) {
if (other != null && other instanceof LogicalSchema) {
LogicalSchema os = (LogicalSchema)other;
if (size() != os.size()) return false;
for (int i = 0; i < size(); i++) {
- if (!getField(i).isEqual(os.getField(i))) return false;
+ if (!getField(i).isEqual(os.getField(i), compareAlias)) return false;
}
return true;
} else {
return false;
}
-
}
/**
@@ -487,7 +738,7 @@ public class LogicalSchema {
else if (s2!=null) return s2.deepCopy();
else return null;
}
- else // Union/UnionInner, take null
+ else // Union/UnionInner, return null
return null;
}
@@ -595,23 +846,25 @@ public class LogicalSchema {
* For Tuples and Bags, SubSchemas have to be equal be considered compatible
*/
public static LogicalSchema mergeSchemaByAlias(LogicalSchema schema1, LogicalSchema schema2)
- throws FrontendException {
+ throws FrontendException{
LogicalSchema mergedSchema = new LogicalSchema();
HashSet<LogicalFieldSchema> schema2colsAdded = new HashSet<LogicalFieldSchema>();
// add/merge fields present in first schema
for(LogicalFieldSchema fs1 : schema1.getFields()){
checkNullAlias(fs1, schema1);
- LogicalFieldSchema fs2 = schema2.getField( fs1.alias );
+ LogicalFieldSchema fs2 = schema2.getFieldSubNameMatch( fs1.alias );
if(fs2 != null){
if(schema2colsAdded.contains(fs2)){
// alias corresponds to multiple fields in schema1,
// just do a lookup on
// schema1 , that will throw the appropriate error.
- schema1.getField( fs2.alias );
+ schema1.getFieldSubNameMatch( fs2.alias );
}
schema2colsAdded.add(fs2);
LogicalFieldSchema mergedFs = LogicalFieldSchema.merge(fs1,fs2, MergeMode.Union);
mergedFs.alias = mergeNameSpacedAlias(fs1.alias, fs2.alias);
+ if (mergedFs.alias==null)
+ mergedFs.alias = fs1.alias;
mergedSchema.addField(mergedFs);
}
else
@@ -710,4 +963,13 @@ public class LogicalSchema {
}
return true;
}
+ /***
+ * Old Pig schema does not require a tuple schema inside a bag;
+ * Now it is required to have that; this method is to fill the gap
+ */
+ public void normalize() {
+ for (LogicalFieldSchema fs : getFields()) {
+ fs.normalize();
+ }
+ }
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/rules/ColumnPruneHelper.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/rules/ColumnPruneHelper.java?rev=1078085&r1=1078084&r2=1078085&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/rules/ColumnPruneHelper.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/rules/ColumnPruneHelper.java Fri Mar 4 18:15:11 2011
@@ -114,7 +114,7 @@ public class ColumnPruneHelper {
public boolean check() throws FrontendException {
List<Operator> sources = currentPlan.getSources();
// if this rule has run before, just return false
- if (sources.get(0).getAnnotation(INPUTUIDS) != null) {
+ if (sources.size() > 1 && sources.get(0).getAnnotation(INPUTUIDS) != null) {
clearAnnotation();
return false;
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/rules/ConstExpEvaluator.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/rules/ConstExpEvaluator.java?rev=1078085&r1=1078084&r2=1078085&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/rules/ConstExpEvaluator.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/rules/ConstExpEvaluator.java Fri Mar 4 18:15:11 2011
@@ -61,57 +61,46 @@ class ConstExpEvaluator extends LogicalE
ConstantExpression newExp = null;
if (parent instanceof AndExpression) newExp = new ConstantExpression(
plan,
- (Boolean) rhs.getValue() && (Boolean) lhs.getValue(),
- parent.getFieldSchema());
+ (Boolean) rhs.getValue() && (Boolean) lhs.getValue());
else if (parent instanceof OrExpression) newExp = new ConstantExpression(
plan,
- (Boolean) rhs.getValue() || (Boolean) lhs.getValue(),
- parent.getFieldSchema());
+ (Boolean) rhs.getValue() || (Boolean) lhs.getValue());
else if (parent instanceof EqualExpression) newExp = new ConstantExpression(
- plan, rhs.isEqual(lhs),
- parent.getFieldSchema());
+ plan, rhs.isEqual(lhs));
else if (parent instanceof GreaterThanExpression) newExp = new ConstantExpression(
plan,
- ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) > 0,
- parent.getFieldSchema());
+ ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) > 0);
else if (parent instanceof GreaterThanEqualExpression) newExp = new ConstantExpression(
plan,
- ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) >= 0,
- parent.getFieldSchema());
+ ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) >= 0);
else if (parent instanceof LessThanExpression) newExp = new ConstantExpression(
plan,
- ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) < 0,
- parent.getFieldSchema());
+ ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) < 0);
else if (parent instanceof LessThanExpression) newExp = new ConstantExpression(
plan,
- ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) <= 0,
- parent.getFieldSchema());
+ ((Comparable) rhs.getValue()).compareTo((Comparable) lhs.getValue()) <= 0);
else if (parent instanceof AddExpression) {
byte type = parent.getFieldSchema().type;
switch (type) {
case DataType.INTEGER:
newExp = new ConstantExpression(
plan,
- (Integer) lhs.getValue() + (Integer) rhs.getValue(),
- parent.getFieldSchema());
+ (Integer) lhs.getValue() + (Integer) rhs.getValue());
break;
case DataType.LONG:
newExp = new ConstantExpression(
plan,
- (Long) lhs.getValue() + (Long) rhs.getValue(),
- parent.getFieldSchema());
+ (Long) lhs.getValue() + (Long) rhs.getValue());
break;
case DataType.FLOAT:
newExp = new ConstantExpression(
plan,
- (Float) lhs.getValue() + (Float) rhs.getValue(),
- parent.getFieldSchema());
+ (Float) lhs.getValue() + (Float) rhs.getValue());
break;
case DataType.DOUBLE:
newExp = new ConstantExpression(
plan,
- (Double) lhs.getValue() + (Double) rhs.getValue(),
- parent.getFieldSchema());
+ (Double) lhs.getValue() + (Double) rhs.getValue());
break;
default:
throw new FrontendException("Invalid type");
@@ -123,26 +112,22 @@ class ConstExpEvaluator extends LogicalE
case DataType.INTEGER:
newExp = new ConstantExpression(
plan,
- (Integer) lhs.getValue() - (Integer) rhs.getValue(),
- parent.getFieldSchema());
+ (Integer) lhs.getValue() - (Integer) rhs.getValue());
break;
case DataType.LONG:
newExp = new ConstantExpression(
plan,
- (Long) lhs.getValue() - (Long) rhs.getValue(),
- parent.getFieldSchema());
+ (Long) lhs.getValue() - (Long) rhs.getValue());
break;
case DataType.FLOAT:
newExp = new ConstantExpression(
plan,
- (Float) lhs.getValue() - (Float) rhs.getValue(),
- parent.getFieldSchema());
+ (Float) lhs.getValue() - (Float) rhs.getValue());
break;
case DataType.DOUBLE:
newExp = new ConstantExpression(
plan,
- (Double) lhs.getValue() - (Double) rhs.getValue(),
- parent.getFieldSchema());
+ (Double) lhs.getValue() - (Double) rhs.getValue());
break;
default:
throw new FrontendException("Invalid type");
@@ -154,26 +139,22 @@ class ConstExpEvaluator extends LogicalE
case DataType.INTEGER:
newExp = new ConstantExpression(
plan,
- (Integer) lhs.getValue() * (Integer) rhs.getValue(),
- parent.getFieldSchema());
+ (Integer) lhs.getValue() * (Integer) rhs.getValue());
break;
case DataType.LONG:
newExp = new ConstantExpression(
plan,
- (Long) lhs.getValue() * (Long) rhs.getValue(),
- parent.getFieldSchema());
+ (Long) lhs.getValue() * (Long) rhs.getValue());
break;
case DataType.FLOAT:
newExp = new ConstantExpression(
plan,
- (Float) lhs.getValue() * (Float) rhs.getValue(),
- parent.getFieldSchema());
+ (Float) lhs.getValue() * (Float) rhs.getValue());
break;
case DataType.DOUBLE:
newExp = new ConstantExpression(
plan,
- (Double) lhs.getValue() * (Double) rhs.getValue(),
- parent.getFieldSchema());
+ (Double) lhs.getValue() * (Double) rhs.getValue());
break;
default:
throw new FrontendException("Invalid type");
@@ -185,14 +166,12 @@ class ConstExpEvaluator extends LogicalE
case DataType.INTEGER:
newExp = new ConstantExpression(
plan,
- (Integer) lhs.getValue() % (Integer) rhs.getValue(),
- parent.getFieldSchema());
+ (Integer) lhs.getValue() % (Integer) rhs.getValue());
break;
case DataType.LONG:
newExp = new ConstantExpression(
plan,
- (Long) lhs.getValue() % (Long) rhs.getValue(),
- parent.getFieldSchema());
+ (Long) lhs.getValue() % (Long) rhs.getValue());
break;
default:
throw new FrontendException("Invalid type");
@@ -205,29 +184,25 @@ class ConstExpEvaluator extends LogicalE
if ((Integer) rhs.getValue() != 0)
newExp = new ConstantExpression(
plan,
- (Integer) lhs.getValue() / (Integer) rhs.getValue(),
- parent.getFieldSchema());
+ (Integer) lhs.getValue() / (Integer) rhs.getValue());
break;
case DataType.LONG:
if ((Long) rhs.getValue() != 0)
newExp = new ConstantExpression(
plan,
- (Long) lhs.getValue() / (Long) rhs.getValue(),
- parent.getFieldSchema());
+ (Long) lhs.getValue() / (Long) rhs.getValue());
break;
case DataType.FLOAT:
if ((Float) rhs.getValue() != 0)
newExp = new ConstantExpression(
plan,
- (Float) lhs.getValue() / (Float) rhs.getValue(),
- parent.getFieldSchema());
+ (Float) lhs.getValue() / (Float) rhs.getValue());
break;
case DataType.DOUBLE:
if ((Double) rhs.getValue() != 0)
newExp = new ConstantExpression(
plan,
- (Double) lhs.getValue() / (Double) rhs.getValue(),
- parent.getFieldSchema());
+ (Double) lhs.getValue() / (Double) rhs.getValue());
break;
default:
throw new FrontendException("Invalid type");
@@ -272,37 +247,31 @@ class ConstExpEvaluator extends LogicalE
plan.remove(op);
ConstantExpression newExp;
if (op instanceof NotExpression) newExp = new ConstantExpression(
- plan, !((Boolean) operand.getValue()),
- op.getFieldSchema());
+ plan, !((Boolean) operand.getValue()));
else if (op instanceof IsNullExpression) newExp = new ConstantExpression(
- plan, operand.getValue() == null,
- op.getFieldSchema());
+ plan, operand.getValue() == null);
else if (op instanceof NegativeExpression) {
byte type = operand.getFieldSchema().type;
switch (type) {
case DataType.INTEGER:
newExp = new ConstantExpression(
plan,
- -1 * ((Integer) operand.getValue()),
- op.getFieldSchema());
+ -1 * ((Integer) operand.getValue()));
break;
case DataType.LONG:
newExp = new ConstantExpression(
plan,
- -1L * ((Long) operand.getValue()),
- op.getFieldSchema());
+ -1L * ((Long) operand.getValue()));
break;
case DataType.FLOAT:
newExp = new ConstantExpression(
plan,
- -1.0F * ((Float) operand.getValue()),
- op.getFieldSchema());
+ -1.0F * ((Float) operand.getValue()));
break;
case DataType.DOUBLE:
newExp = new ConstantExpression(
plan,
- -1.0D * ((Integer) operand.getValue()),
- op.getFieldSchema());
+ -1.0D * ((Integer) operand.getValue()));
break;
default:
throw new FrontendException("Invalid type");
Modified: pig/trunk/src/org/apache/pig/newplan/logical/rules/ImplicitSplitInserter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/rules/ImplicitSplitInserter.java?rev=1078085&r1=1078084&r2=1078085&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/rules/ImplicitSplitInserter.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/rules/ImplicitSplitInserter.java Fri Mar 4 18:15:11 2011
@@ -126,8 +126,7 @@ public class ImplicitSplitInserter exten
// position is remembered in order to maintain the order of the successors
Pair<Integer, Integer> pos = currentPlan.disconnect(op, suc);
LogicalExpressionPlan filterPlan = new LogicalExpressionPlan();
- LogicalSchema.LogicalFieldSchema fs = new LogicalSchema.LogicalFieldSchema(null, null, DataType.BOOLEAN);
- new ConstantExpression(filterPlan, Boolean.valueOf(true), fs);
+ new ConstantExpression(filterPlan, Boolean.valueOf(true));
LOSplitOutput splitOutput = new LOSplitOutput((LogicalPlan) currentPlan, filterPlan);
splitOutput.setAlias(splitOp.getAlias());
currentPlan.add(splitOutput);
Added: pig/trunk/src/org/apache/pig/newplan/logical/visitor/CastLineageSetter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/CastLineageSetter.java?rev=1078085&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/visitor/CastLineageSetter.java (added)
+++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/CastLineageSetter.java Fri Mar 4 18:15:11 2011
@@ -0,0 +1,117 @@
+/*
+ * 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.visitor;
+
+import java.util.Map;
+
+import org.apache.pig.FuncSpec;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.plan.CompilationMessageCollector;
+import org.apache.pig.impl.plan.CompilationMessageCollector.MessageType;
+import org.apache.pig.newplan.DependencyOrderWalker;
+import org.apache.pig.newplan.OperatorPlan;
+import org.apache.pig.newplan.ReverseDependencyOrderWalker;
+import org.apache.pig.newplan.logical.expression.CastExpression;
+import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
+import org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor;
+import org.apache.pig.newplan.logical.optimizer.AllExpressionVisitor;
+import org.apache.pig.newplan.logical.relational.LogicalPlan;
+
+
+/**
+ * Find uid lineage information. Set the load function in CastExpression
+ * if it needs to convert bytearray to another type.
+ */
+public class CastLineageSetter extends AllExpressionVisitor{
+
+ private CompilationMessageCollector msgCollector;
+ private Map<Long, FuncSpec> uid2LoadFuncMap;
+ public CastLineageSetter(
+ LogicalPlan plan,
+ CompilationMessageCollector msgCollector
+ )
+ throws FrontendException {
+ super(plan, new DependencyOrderWalker(plan));
+ this.msgCollector = msgCollector;
+
+ //find lineage of columns, get mapping of uid to load-functions
+ LineageFindRelVisitor lineageFinder = new LineageFindRelVisitor(plan);
+ lineageFinder.visit();
+ uid2LoadFuncMap = lineageFinder.getUid2LoadFuncMap();
+ //plan.explain(System.out, "test", true);
+ //System.out.println(uid2LoadFuncMap);
+ }
+
+
+ @Override
+ protected LogicalExpressionVisitor getVisitor(LogicalExpressionPlan exprPlan)
+ throws FrontendException {
+ return new CastLineageSetterExpVisitor(exprPlan, uid2LoadFuncMap, msgCollector);
+ }
+
+
+ static class CastLineageSetterExpVisitor extends LogicalExpressionVisitor{
+ private Map<Long, FuncSpec> uid2LoadFuncMap;
+ private CompilationMessageCollector msgCollector;
+
+ protected CastLineageSetterExpVisitor(
+ OperatorPlan expPlan,
+ Map<Long, FuncSpec> uid2LoadFuncMap,
+ CompilationMessageCollector msgCollector
+ )
+ throws FrontendException {
+ super(expPlan, new ReverseDependencyOrderWalker(expPlan));
+ this.uid2LoadFuncMap = uid2LoadFuncMap;
+ this.msgCollector = msgCollector;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor#visit(org.apache.pig.newplan.logical.expression.CastExpression)
+ *if input type is bytearray, find and set the corresponding load function
+ * that this field comes from.
+ * The load functions LoadCaster interface will help with conversion
+ * at runtime.
+ * If there is no corresponding load function (eg if the input is an
+ * output of a udf), set nothing - it assumes that bytearray is being used
+ * as equivalent of 'unknown' type. It will try to identify the type
+ * at runtime and cast it.
+ */
+ @Override
+ public void visit(CastExpression cast) throws FrontendException{
+
+ byte inType = cast.getExpression().getType();
+ byte outType = cast.getType();
+
+ if(inType == DataType.BYTEARRAY){
+ long inUid = cast.getExpression().getFieldSchema().uid;
+ FuncSpec inLoadFunc = uid2LoadFuncMap.get(inUid);
+ if(inLoadFunc == null){
+ String msg = "Cannot resolve load function to use for casting from " +
+ DataType.findTypeName(inType) + " to " +
+ DataType.findTypeName(outType) + ". ";
+ msgCollector.collect(msg, MessageType.Warning);
+ }else {
+ cast.setFuncSpec(inLoadFunc);
+ }
+ }
+ }
+
+ }
+
+}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/visitor/ColumnAliasConversionVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/ColumnAliasConversionVisitor.java?rev=1078085&r1=1078084&r2=1078085&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/visitor/ColumnAliasConversionVisitor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/ColumnAliasConversionVisitor.java Fri Mar 4 18:15:11 2011
@@ -21,6 +21,7 @@ package org.apache.pig.newplan.logical.v
import java.util.ArrayList;
import java.util.List;
+import org.apache.pig.data.DataType;
import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.impl.plan.PlanValidationException;
import org.apache.pig.newplan.DependencyOrderWalker;
@@ -57,19 +58,20 @@ public class ColumnAliasConversionVisito
List<Operator> inputs = lp.getPredecessors( op );
LogicalRelationalOperator input = (LogicalRelationalOperator)inputs.get( expr.getInputNum() );
LogicalSchema inputSchema = input.getSchema();
+
String alias = expr.getColAlias();
if( alias != null ) {
int colNum = inputSchema == null ? -1 : inputSchema.getFieldPosition( alias );
if( colNum == -1 ) {
throw new PlanValidationException( "Invalid field projection. Projected field [" +
- alias + "] does not exist in schema: " + inputSchema + "." );
+ alias + "] does not exist in schema: " + inputSchema + ".", 1025 );
}
expr.setColNum( colNum );
} else {
int col = expr.getColNum();
if( inputSchema != null && col >= inputSchema.size() ) {
throw new PlanValidationException( "Out of bound access. Trying to access non-existent column: " +
- col + ". Schema " + inputSchema + " has " + inputSchema.size() + " column(s)." );
+ col + ". Schema " + inputSchema + " has " + inputSchema.size() + " column(s)." , 1000);
}
}
}
@@ -82,22 +84,31 @@ public class ColumnAliasConversionVisito
List<Integer> cols = new ArrayList<Integer>( rawCols.size() );
LogicalExpressionPlan plan = (LogicalExpressionPlan)expr.getPlan();
- LogicalExpression pred = (LogicalExpression)plan.getPredecessors( expr ).get(0);
- LogicalSchema schema = pred.getFieldSchema().schema;
+ LogicalExpression pred = (LogicalExpression)plan.getSuccessors( expr ).get(0);
+
+ LogicalSchema schema = null;
+ if( pred.getFieldSchema().type == DataType.BAG ) {
+ if( pred.getFieldSchema().schema != null )
+ schema = pred.getFieldSchema().schema.getField(0).schema;
+ }
+ else {
+ schema = pred.getFieldSchema().schema;
+ }
+
int col = -1;
for( Object rc : rawCols ) {
if( rc instanceof Integer ) {
col = (Integer)rc;
if( schema != null && col >= schema.size() ) {
throw new PlanValidationException( "Out of bound access. Trying to access non-existent column: " +
- col + ". Schema " + schema + " has " + schema.size() + " column(s)." );
+ col + ". Schema " + schema + " has " + schema.size() + " column(s).", 1000 );
}
cols.add( (Integer)rc );
} else {
- col = schema.getFieldPosition( (String)rc );
+ col = schema == null ? -1 : schema.getFieldPosition( (String)rc );
if( col == -1 ) {
throw new PlanValidationException( "Invalid field reference. Referenced field [" +
- rc + "] does not exist in schema: " + schema + "." );
+ rc + "] does not exist in schema: " + schema + "." , 1000);
}
cols.add( col );
}
Added: pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java?rev=1078085&view=auto
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java (added)
+++ pig/trunk/src/org/apache/pig/newplan/logical/visitor/LineageFindRelVisitor.java Fri Mar 4 18:15:11 2011
@@ -0,0 +1,635 @@
+/*
+ * 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.visitor;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.pig.FuncSpec;
+import org.apache.pig.PigException;
+import org.apache.pig.data.DataType;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.impl.streaming.StreamingCommand;
+import org.apache.pig.impl.streaming.StreamingCommand.HandleSpec;
+import org.apache.pig.impl.util.MultiMap;
+import org.apache.pig.newplan.DependencyOrderWalker;
+import org.apache.pig.newplan.Operator;
+import org.apache.pig.newplan.OperatorPlan;
+import org.apache.pig.newplan.PlanWalker;
+import org.apache.pig.newplan.ReverseDependencyOrderWalker;
+import org.apache.pig.newplan.logical.expression.BinCondExpression;
+import org.apache.pig.newplan.logical.expression.CastExpression;
+import org.apache.pig.newplan.logical.expression.ConstantExpression;
+import org.apache.pig.newplan.logical.expression.DereferenceExpression;
+import org.apache.pig.newplan.logical.expression.LogicalExpression;
+import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
+import org.apache.pig.newplan.logical.expression.LogicalExpressionVisitor;
+import org.apache.pig.newplan.logical.expression.MapLookupExpression;
+import org.apache.pig.newplan.logical.expression.ProjectExpression;
+import org.apache.pig.newplan.logical.expression.ScalarExpression;
+import org.apache.pig.newplan.logical.relational.LOCogroup;
+import org.apache.pig.newplan.logical.relational.LOCross;
+import org.apache.pig.newplan.logical.relational.LODistinct;
+import org.apache.pig.newplan.logical.relational.LOFilter;
+import org.apache.pig.newplan.logical.relational.LOForEach;
+import org.apache.pig.newplan.logical.relational.LOGenerate;
+import org.apache.pig.newplan.logical.relational.LOInnerLoad;
+import org.apache.pig.newplan.logical.relational.LOJoin;
+import org.apache.pig.newplan.logical.relational.LOLimit;
+import org.apache.pig.newplan.logical.relational.LOLoad;
+import org.apache.pig.newplan.logical.relational.LOSort;
+import org.apache.pig.newplan.logical.relational.LOSplit;
+import org.apache.pig.newplan.logical.relational.LOSplitOutput;
+import org.apache.pig.newplan.logical.relational.LOStore;
+import org.apache.pig.newplan.logical.relational.LOStream;
+import org.apache.pig.newplan.logical.relational.LOUnion;
+import org.apache.pig.newplan.logical.relational.LogicalPlan;
+import org.apache.pig.newplan.logical.relational.LogicalRelationalNodesVisitor;
+import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator;
+import org.apache.pig.newplan.logical.relational.LogicalSchema;
+import org.apache.pig.newplan.logical.relational.LogicalSchema.LogicalFieldSchema;
+
+/**
+ * Create mapping between uid and Load FuncSpec when the LogicalExpression
+ * associated with it is known to hold an unmodified element of data returned
+ * by the load function.
+ * This information is used to find the LoadCaster to be used to cast bytearrays
+ * into other other types.
+ */
+public class LineageFindRelVisitor extends LogicalRelationalNodesVisitor{
+
+
+ Map<Long, FuncSpec> uid2LoadFuncMap = new HashMap<Long, FuncSpec>();
+
+ //if LogicalRelationalOperator is associated with a single load func spec
+ // then the mapping is stored here
+ Map<LogicalRelationalOperator, FuncSpec> rel2InputFuncMap =
+ new HashMap<LogicalRelationalOperator, FuncSpec>();
+
+ public LineageFindRelVisitor(OperatorPlan plan) throws FrontendException {
+ super(plan, new DependencyOrderWalker(plan));
+
+ }
+
+ public Map<Long, FuncSpec> getUid2LoadFuncMap(){
+ return uid2LoadFuncMap;
+ }
+
+ @Override
+ public void visit(LOLoad load) throws FrontendException{
+ LogicalSchema schema = load.getSchema();
+ setLoadFuncForUids(schema, load.getFileSpec().getFuncSpec());
+ rel2InputFuncMap.put(load, load.getFileSpec().getFuncSpec());
+ }
+
+ @Override
+ public void visit(LOStream stream) throws FrontendException{
+ mapToPredLoadFunc(stream);
+ StreamingCommand command = ((LOStream)stream).getStreamingCommand();
+ HandleSpec streamOutputSpec = command.getOutputSpec();
+ FuncSpec streamLoaderSpec = new FuncSpec(streamOutputSpec.getSpec());
+ setLoadFuncForUids(stream.getSchema(), streamLoaderSpec);
+ rel2InputFuncMap.put(stream, streamLoaderSpec);
+ }
+
+ @Override
+ public void visit(LOInnerLoad innerLoad) throws FrontendException{
+ LOForEach foreach = innerLoad.getLOForEach();
+ LogicalRelationalOperator pred =
+ ((LogicalRelationalOperator)foreach.getPlan().getPredecessors(foreach).get(0));
+
+ LogicalSchema predSchema = pred.getSchema();
+
+ //if this has a schema, the lineage can be tracked using the uid in input schema
+ if(innerLoad.getSchema() != null){
+ if(innerLoad.getSchema().size() == 1
+ && innerLoad.getSchema().getField(0).type == DataType.BYTEARRAY
+ && uid2LoadFuncMap.get(innerLoad.getSchema().getField(0).uid) == null
+ && predSchema != null
+ ){
+ long inpUid = predSchema.getField(innerLoad.getProjection().getColNum()).uid;
+ if(uid2LoadFuncMap.get(inpUid) != null){
+ addUidLoadFuncToMap(innerLoad.getSchema().getField(0).uid, uid2LoadFuncMap.get(inpUid));
+ }
+ return;
+ }
+
+ }
+
+ // associated load func could not be found using uid, use
+ // the single load func associated with input relation (if any)
+ if(getAssociatedLoadFunc(pred) != null){
+ mapRelToPredLoadFunc(innerLoad, pred);
+ }
+ }
+
+ /**
+ * map all uids in schema to funcSpec
+ * @param schema
+ * @param funcSpec
+ * @throws VisitorException
+ */
+ private void setLoadFuncForUids(LogicalSchema schema, FuncSpec funcSpec)
+ throws VisitorException {
+ if(schema == null){
+ return;
+ }
+ for(LogicalFieldSchema fs : schema.getFields()){
+ addUidLoadFuncToMap((Long) fs.uid, funcSpec);
+ setLoadFuncForUids(fs.schema, funcSpec);
+ }
+
+ }
+
+ @Override
+ public void visit(LOFilter filter) throws FrontendException{
+ mapToPredLoadFunc(filter);
+ visitExpression(filter.getFilterPlan());
+ }
+
+ /**
+ * If all predecessors of relOp are associated with same load
+ * func , then map reOp to it.
+ *
+ * @param relOp
+ * @throws FrontendException
+ */
+ private void mapToPredLoadFunc(LogicalRelationalOperator relOp)
+ throws FrontendException {
+ OperatorPlan lp = relOp.getPlan();
+ List<Operator> preds = lp.getPredecessors(relOp);
+ if(lp.getPredecessors(relOp) == null ||
+ lp.getPredecessors(relOp).size() == 0){
+ // no predecessors , nothing to do
+ return;
+ }
+
+ FuncSpec loadFuncSpec = getAssociatedLoadFunc((LogicalRelationalOperator) preds.get(0));
+
+ if(loadFuncSpec == null){
+ return;
+ }
+ //ensure that all predecessors are mapped to same load func spec
+ for(int i=1; i<preds.size(); i++){
+ if(!loadFuncSpec.equals(getAssociatedLoadFunc((LogicalRelationalOperator) preds.get(i)))){
+ return;
+ }
+ }
+ rel2InputFuncMap.put(relOp, loadFuncSpec);
+ }
+
+ /**
+ * Find single load func spec associated with this relation.
+ * If the relation has schema, all uids in schema should be associated
+ * with same load func spec. if it does not have schema check the existing
+ * mapping
+ * @param relOp
+ * @return
+ * @throws FrontendException
+ */
+ private FuncSpec getAssociatedLoadFunc(LogicalRelationalOperator relOp) throws FrontendException {
+ LogicalSchema schema = relOp.getSchema();
+ if(schema != null){
+ if(schema.size() == 0)
+ return null;
+ FuncSpec funcSpec = uid2LoadFuncMap.get(schema.getField(0).uid);
+ if(funcSpec == null)
+ return null;
+ for(int i=1; i<schema.size(); i++){
+ LogicalFieldSchema fs = schema.getField(i);
+ if(! funcSpec.equals(uid2LoadFuncMap.get(fs.uid))){
+ //all uid are not associated with same func spec, there is no
+ // single func spec that represents all the fields
+ return null;
+ }
+ }
+ return funcSpec;
+ }
+ else{
+ return rel2InputFuncMap.get(relOp);
+ }
+
+ }
+
+ private void mapRelToPredLoadFunc(LogicalRelationalOperator relOp,
+ Operator pred) {
+ if(rel2InputFuncMap.get(pred) != null){
+ rel2InputFuncMap.put(relOp, rel2InputFuncMap.get(pred));
+ }
+ }
+
+ private void visitExpression(LogicalExpressionPlan expPlan)
+ throws FrontendException{
+ LineageFindExpVisitor findLineageExp =
+ new LineageFindExpVisitor(expPlan, uid2LoadFuncMap);
+ findLineageExp.visit();
+ }
+
+ @Override
+ public void visit(LOCogroup group) throws FrontendException{
+ mapToPredLoadFunc(group);
+ List<Operator> inputs = group.getInputs((LogicalPlan)plan);
+ if(inputs.size() == 1){
+ mapToPredLoadFunc(group);
+ }
+
+ MultiMap<Integer, LogicalExpressionPlan> plans = group.getExpressionPlans();
+ for(LogicalExpressionPlan expPlan : plans.values()){
+ visitExpression(expPlan);
+ }
+ //set the load func spec for the bags in the schema, this helps if
+ // the input schemas are not set
+ LogicalSchema sch = group.getSchema();
+ //group schema has a group column followed by bags corresponding to each
+ // input
+ if(sch.size() != inputs.size()+1 ){
+ throw new AssertionError("cogroup schema size not same as number of inputs");
+ }
+ for(int i=1; i < sch.size(); i++){
+ long uid = sch.getField(i).uid;
+ LogicalRelationalOperator input = (LogicalRelationalOperator) inputs.get(i-1);
+ if(getAssociatedLoadFunc(input) != null){
+ addUidLoadFuncToMap(uid, rel2InputFuncMap.get(input));
+ }
+ }
+
+ }
+
+
+ @Override
+ public void visit(LOJoin join) throws FrontendException{
+ mapToPredLoadFunc(join);
+ MultiMap<Integer, LogicalExpressionPlan> plans = join.getExpressionPlans();
+ for(LogicalExpressionPlan expPlan : plans.values()){
+ visitExpression(expPlan);
+ }
+ }
+
+ @Override
+ public void visit(LOForEach fe) throws FrontendException{
+ mapToPredLoadFunc(fe);
+ LogicalPlan innerPlan = fe.getInnerPlan();
+ PlanWalker newWalker = currentWalker.spawnChildWalker(innerPlan);
+ pushWalker(newWalker);
+ currentWalker.walk(this);
+ popWalker();
+ }
+
+ @Override
+ public void visit(LOGenerate gen) throws FrontendException{
+ mapToPredLoadFunc(gen);
+ List<LogicalExpressionPlan> expPlans = gen.getOutputPlans();
+ for(LogicalExpressionPlan expPlan : expPlans){
+ visitExpression(expPlan);
+ }
+
+ //associate flatten output to the load func associated with input
+ //expression of flatten
+ boolean[] flattens = gen.getFlattenFlags();
+ for(int i=0; i<flattens.length; i++){
+ if(flattens[i] == true){
+ //get the output schema corresponding to this exp plan
+ gen.getSchema();
+ if(gen.getOutputPlanSchemas() == null || gen.getOutputPlanSchemas().size() <= i){
+ return;
+ }
+ LogicalSchema sch = gen.getOutputPlanSchemas().get(i);
+ if(sch == null){
+ continue;
+ }
+
+ //get the only output exp of the ith plan
+ LogicalExpression exp =
+ (LogicalExpression) gen.getOutputPlans().get(i).getSources().get(0);
+
+ //get its funcspec and associate it with uid of all fields in the schema
+ FuncSpec funcSpec = uid2LoadFuncMap.get(exp.getFieldSchema().uid);
+ for(LogicalFieldSchema fs : sch.getFields()){
+ addUidLoadFuncToMap(fs.uid, funcSpec);
+ }
+ }
+ }
+
+ }
+
+ @Override
+ public void visit(LOSort sort) throws FrontendException{
+ mapToPredLoadFunc(sort);
+ List<LogicalExpressionPlan> expPlans = sort.getSortColPlans();
+ for(LogicalExpressionPlan expPlan : expPlans){
+ visitExpression(expPlan);
+ }
+ }
+
+
+ @Override
+ public void visit(LODistinct relOp) throws FrontendException{
+ mapToPredLoadFunc(relOp);
+ }
+
+ @Override
+ public void visit(LOLimit relOp) throws FrontendException{
+ mapToPredLoadFunc(relOp);
+ }
+
+ @Override
+ public void visit(LOSplit relOp) throws FrontendException{
+ mapToPredLoadFunc(relOp);
+ }
+
+ @Override
+ public void visit(LOStore relOp) throws FrontendException{
+ mapToPredLoadFunc(relOp);
+ }
+
+
+
+ @Override
+ public void visit(LOCross relOp) throws FrontendException{
+ mapToPredLoadFunc(relOp);
+ }
+
+ @Override
+ public void visit(LOUnion relOp) throws FrontendException{
+ mapToPredLoadFunc(relOp);
+ }
+
+ @Override
+ public void visit(LOSplitOutput split) throws FrontendException{
+ mapToPredLoadFunc(split);
+ visitExpression(split.getFilterPlan());
+
+ //The uid changes across the input and output of a split.
+ //In this visitor the uids for the fields in output are mapped to ones
+ // in input based on position. uid2LoadFuncMap is updated with output
+ // schema uid and load func mapped to corresponding uid in inputschema
+
+
+ LogicalSchema inputSch =
+ ((LogicalRelationalOperator)split.getPlan().getPredecessors(split).get(0)).getSchema();
+
+ LogicalSchema outSchema = split.getSchema();
+ if(inputSch == null && outSchema == null){
+ return;
+ }
+
+ if(inputSch == null || outSchema == null){
+ String msg = "Bug: in split only one of input/output schema is null "
+ + split;
+ throw new VisitorException(msg,2260, PigException.BUG) ;
+ }
+
+ if(inputSch.size() != outSchema.size()){
+ String msg = "Bug: input and output schema size of split differ "
+ + split;
+ throw new VisitorException(msg,2261, PigException.BUG) ;
+ }
+
+ for(int i=0; i<inputSch.size(); i++){
+ LogicalFieldSchema inField = inputSch.getField(i);
+ LogicalFieldSchema outField = outSchema.getField(i);
+ if(uid2LoadFuncMap.get(inField.uid) != null){
+ addUidLoadFuncToMap(outField.uid, uid2LoadFuncMap.get(inField.uid));
+ }
+ }
+
+ }
+
+
+ /**
+ * Add given uid, loadFuncSpec to mapping
+ * @param uid
+ * @param loadFuncSpec
+ * @throws VisitorException
+ */
+ private void addUidLoadFuncToMap(long uid, FuncSpec loadFuncSpec)
+ throws VisitorException{
+ if(loadFuncSpec == null){
+ return;
+ }
+ //ensure that uid always matches to same load func
+ FuncSpec curFuncSpec = uid2LoadFuncMap.get(uid);
+ if(curFuncSpec == null){
+ uid2LoadFuncMap.put(uid, loadFuncSpec);
+ }else if(! curFuncSpec.equals(loadFuncSpec)){
+ String msg = "Bug: uid mapped to two different load functions : " +
+ curFuncSpec + " and " + loadFuncSpec;
+ throw new VisitorException(msg,2262, PigException.BUG) ;
+ }
+ }
+
+ /**
+ * If a input of dereference or map-lookup has associated load function,
+ * the same load function should be associated with the dereference or map-lookup.
+ *
+ */
+ class LineageFindExpVisitor extends LogicalExpressionVisitor{
+
+ private Map<Long, FuncSpec> uid2LoadFuncMap;
+
+ public LineageFindExpVisitor(LogicalExpressionPlan plan,
+ Map<Long, FuncSpec> uid2LoadFuncMap) throws FrontendException {
+ super(plan, new ReverseDependencyOrderWalker(plan));
+ this.uid2LoadFuncMap = uid2LoadFuncMap;
+ }
+
+ @Override
+ public void visit(ProjectExpression proj) throws FrontendException {
+ // proj should have the same uid as input, if input has schema.
+ // if uid does not have associated func spec and input relation
+ // has null schema or it is inner-load, use the load func associated
+ // with the relation
+ LogicalRelationalOperator inputRel = proj.findReferent();
+
+ long uid = proj.getFieldSchema().uid;
+ if(uid2LoadFuncMap.get(uid) == null && (inputRel.getSchema() == null || inputRel instanceof LOInnerLoad)){
+ FuncSpec funcSpec = rel2InputFuncMap.get(inputRel);
+ if(funcSpec != null){
+ addUidLoadFuncToMap(uid, funcSpec);
+ }
+ }
+
+ }
+
+ @Override
+ public void visit(DereferenceExpression deref) throws FrontendException {
+ updateUidMap(deref, deref.getReferredExpression());
+ }
+
+ @Override
+ public void visit(MapLookupExpression mapLookup) throws FrontendException {
+ updateUidMap(mapLookup, mapLookup.getMap());
+ }
+
+
+ private void updateUidMap(LogicalExpression exp,
+ LogicalExpression inp) throws FrontendException {
+ //find input uid and corresponding load FuncSpec
+ long inpUid = inp.getFieldSchema().uid;
+ FuncSpec inpLoadFuncSpec = uid2LoadFuncMap.get(inpUid);
+ addUidLoadFuncToMap(exp.getFieldSchema().uid, inpLoadFuncSpec);
+
+ }
+
+
+ @Override
+ public void visit(BinCondExpression binCond) throws FrontendException{
+ //if either side is a null constant, we can safely associate
+ //this bincond with the load func spec on other side
+ LogicalExpression lhs = binCond.getLhs();
+ LogicalExpression rhs = binCond.getRhs();
+
+ if(getNULLConstantInCast(lhs) != null){
+ FuncSpec funcSpec = uid2LoadFuncMap.get(rhs.getFieldSchema().uid);
+ uid2LoadFuncMap.put(binCond.getFieldSchema().uid, funcSpec);
+ }
+ else if(getNULLConstantInCast(rhs) != null){
+ FuncSpec funcSpec = uid2LoadFuncMap.get(lhs.getFieldSchema().uid);
+ uid2LoadFuncMap.put(binCond.getFieldSchema().uid, funcSpec);
+ }
+ else {
+ mapMatchLoadFuncToUid(
+ binCond.getFieldSchema(),
+ lhs.getFieldSchema(),
+ rhs.getFieldSchema()
+ );
+ }
+ }
+
+ @Override
+ public void visit(ScalarExpression scalarExp) throws FrontendException{
+ //track lineage to input load function
+
+
+ LogicalRelationalOperator inputRel = (LogicalRelationalOperator)scalarExp.getAttachedLogicalOperator();
+ LogicalPlan relPlan = (LogicalPlan) inputRel.getPlan();
+ List<Operator> softPreds = relPlan.getSoftLinkPredecessors(inputRel);
+
+ //1st argument is the column number in input, 2nd arg is the input file name
+ Integer inputColNum = (Integer)((ConstantExpression) scalarExp.getArguments().get(0)).getValue();
+ String inputFile = (String)((ConstantExpression) scalarExp.getArguments().get(1)).getValue();
+
+ long outputUid = scalarExp.getFieldSchema().uid;
+ boolean foundInput = false; // a variable to do sanity check on num of input relations
+
+ //find the input relation, and use it to get lineage
+ for(Operator softPred : softPreds){
+ LOStore inputStore = (LOStore) softPred;
+ if(inputStore.getFileSpec().getFileName().equals(inputFile)){
+
+ if(foundInput == true){
+ throw new FrontendException(
+ "More than one input found for scalar expression",
+ 2268,
+ PigException.BUG
+ );
+ }
+ foundInput = true;
+
+ //found the store corresponding to this scalar expression
+ LogicalSchema sch = inputStore.getSchema();
+ if(sch == null){
+ //see if there is a load function associated with the store
+ FuncSpec funcSpec = rel2InputFuncMap.get(inputStore);
+ addUidLoadFuncToMap(outputUid, funcSpec);
+ }else{
+ //find input uid and corresponding load func
+ LogicalFieldSchema fs = sch.getField(inputColNum);
+ FuncSpec funcSpec = uid2LoadFuncMap.get(fs.uid);
+ addUidLoadFuncToMap(outputUid, funcSpec);
+ }
+ }
+ }
+
+ if(foundInput == false){
+ throw new FrontendException(
+ "No input found for scalar expression",
+ 2269,
+ PigException.BUG
+ );
+ }
+ }
+
+ /**
+ * if there is a null constant under casts, return it
+ * @param rel
+ * @return
+ * @throws FrontendException
+ */
+ private Object getNULLConstantInCast(LogicalExpression rel) throws FrontendException {
+ if(rel instanceof CastExpression){
+ if( ((CastExpression)rel).getExpression() instanceof CastExpression)
+ return getNULLConstantInCast(((CastExpression)rel).getExpression());
+ else if( ((CastExpression)rel).getExpression() instanceof ConstantExpression){
+ ConstantExpression constExp = (ConstantExpression)((CastExpression)rel).getExpression();
+ if(constExp.getValue() == null)
+ return constExp;
+ else
+ return null;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * if uid in lhs and rhs fieldschema or inner schemas map to same
+ * load function, then map the new uid in bincond also to same
+ * load function in uid2LoadFuncMap
+ * @param binCFs
+ * @param fieldSchema2
+ * @param fieldSchema3
+ */
+ private void mapMatchLoadFuncToUid(
+ LogicalFieldSchema binCFs,
+ LogicalFieldSchema lhsFs,
+ LogicalFieldSchema rhsFs) {
+
+ FuncSpec lhsFuncSpec = uid2LoadFuncMap.get(lhsFs.uid);
+ FuncSpec rhsFuncSpec = uid2LoadFuncMap.get(lhsFs.uid);
+
+ if(lhsFuncSpec == null || rhsFuncSpec == null){
+ return;
+ }
+
+ if(lhsFuncSpec.equals(rhsFuncSpec)){
+ uid2LoadFuncMap.put(binCFs.uid, lhsFuncSpec);
+ }
+
+ if(binCFs.schema != null){
+ //then rhsFs.schema, lhsFs.schema has to be non-null as well
+ //and of same size
+ for(int i=0; i<binCFs.schema.size(); i++){
+ mapMatchLoadFuncToUid(
+ binCFs.schema.getField(i),
+ lhsFs.schema.getField(i),
+ rhsFs.schema.getField(i)
+ );
+ }
+ }
+
+ }
+
+
+ }
+
+
+
+}