You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2011/03/09 23:51:54 UTC
svn commit: r1080048 - in /pig/trunk: ./ src/org/apache/pig/
src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/
src/org/apache/pig/backend/hadoop/hbase/ src/org/apache/pig/builtin/
src/org/apache/pig/data/ src/org/apac...
Author: daijy
Date: Wed Mar 9 22:51:53 2011
New Revision: 1080048
URL: http://svn.apache.org/viewvc?rev=1080048&view=rev
Log:
PIG-1876: Typed map for Pig
Added:
pig/trunk/test/org/apache/pig/test/TestTypedMap.java
Modified:
pig/trunk/CHANGES.txt
pig/trunk/src/org/apache/pig/LoadCaster.java
pig/trunk/src/org/apache/pig/ResourceSchema.java
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java
pig/trunk/src/org/apache/pig/backend/hadoop/hbase/HBaseBinaryConverter.java
pig/trunk/src/org/apache/pig/builtin/BinStorage.java
pig/trunk/src/org/apache/pig/builtin/TextLoader.java
pig/trunk/src/org/apache/pig/builtin/Utf8StorageConverter.java
pig/trunk/src/org/apache/pig/data/DataType.java
pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/Schema.java
pig/trunk/src/org/apache/pig/newplan/logical/LogicalExpPlanMigrationVistor.java
pig/trunk/src/org/apache/pig/newplan/logical/expression/MapLookupExpression.java
pig/trunk/src/org/apache/pig/newplan/logical/relational/LogicalSchema.java
pig/trunk/src/org/apache/pig/newplan/logical/visitor/ColumnAliasConversionVisitor.java
pig/trunk/src/org/apache/pig/parser/AliasMasker.g
pig/trunk/src/org/apache/pig/parser/AstValidator.g
pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g
pig/trunk/src/org/apache/pig/parser/QueryParser.g
pig/trunk/test/org/apache/pig/test/TestMacroExpansion.java
pig/trunk/test/org/apache/pig/test/TestPOCast.java
Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Wed Mar 9 22:51:53 2011
@@ -26,6 +26,8 @@ PIG-1680: HBaseStorage should work with
IMPROVEMENTS
+PIG-1876: Typed map for Pig (daijy)
+
PIG-1889: bin/pig should pick up HBase configuration from HBASE_CONF_DIR
PIG-1794: Javascript support for Pig embedding and UDFs in scripting languages (julien)
Modified: pig/trunk/src/org/apache/pig/LoadCaster.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/LoadCaster.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/LoadCaster.java (original)
+++ pig/trunk/src/org/apache/pig/LoadCaster.java Wed Mar 9 22:51:53 2011
@@ -83,9 +83,19 @@ public interface LoadCaster {
* @return Map value.
* @throws IOException if the value cannot be cast.
*/
+ @Deprecated
public Map<String, Object> bytesToMap(byte[] b) throws IOException;
/**
+ * Cast data from bytearray to map value.
+ * @param b bytearray to be cast.
+ * @param fieldSchema field schema for the output map
+ * @return Map value.
+ * @throws IOException if the value cannot be cast.
+ */
+ public Map<String, Object> bytesToMap(byte[] b, ResourceFieldSchema fieldSchema) throws IOException;
+
+ /**
* Cast data from bytearray to tuple value.
* @param b bytearray to be cast.
* @param fieldSchema field schema for the output tuple
Modified: pig/trunk/src/org/apache/pig/ResourceSchema.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/ResourceSchema.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/ResourceSchema.java (original)
+++ pig/trunk/src/org/apache/pig/ResourceSchema.java Wed Mar 9 22:51:53 2011
@@ -121,8 +121,7 @@ public class ResourceSchema implements S
LogicalSchema inner = fieldSchema.schema;
// allow partial schema
- if ((type == DataType.BAG || type == DataType.TUPLE)
- && inner != null) {
+ if (DataType.isSchemaType(type) && inner != null) {
schema = new ResourceSchema(inner);
} else {
schema = null;
@@ -235,7 +234,7 @@ public class ResourceSchema implements S
StringBuilder sb = new StringBuilder();
if (printAlias)
sb.append(this.name).append(":");
- if (DataType.isAtomic(this.type)||this.type==DataType.MAP) {
+ if (DataType.isAtomic(this.type)) {
sb.append(DataType.findTypeName(this.type));
} else {
if (this.schema!=null)
@@ -490,6 +489,8 @@ public class ResourceSchema implements S
sb.append("{");
} else if (type == DataType.TUPLE) {
sb.append("(");
+ } else if (type == DataType.MAP) {
+ sb.append("[");
}
for (int i=0; i<rs.getFields().length; i++) {
@@ -505,6 +506,8 @@ public class ResourceSchema implements S
sb.append("}");
} else if (type == DataType.TUPLE) {
sb.append(")");
+ } else if (type == DataType.MAP) {
+ sb.append("]");
}
}
}
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POCast.java Wed Mar 9 22:51:53 2011
@@ -104,7 +104,7 @@ public class POCast extends ExpressionOp
@Override
public String name() {
- if (resultType==DataType.BAG||resultType==DataType.TUPLE)
+ if (DataType.isSchemaType(resultType))
return "Cast" + "[" + DataType.findTypeName(resultType)+":"
+ fieldSchema.calcCastString() + "]" + " - "
+ mKey.toString();
@@ -826,6 +826,7 @@ public class POCast extends ExpressionOp
return res;
}
+ @SuppressWarnings("unchecked")
private Object convertWithSchema(Object obj, ResourceFieldSchema fs) throws IOException {
Object result = null;
@@ -889,10 +890,20 @@ public class POCast extends ExpressionOp
break;
case DataType.MAP:
if (obj instanceof Map) {
- result = obj;
+ if (fs!=null && fs.getSchema()!=null) {
+ ResourceFieldSchema innerFieldSchema = fs.getSchema().getFields()[0];
+ Map m = (Map)obj;
+ for (Object entry : m.entrySet()) {
+ Object newValue = convertWithSchema(((Map.Entry)entry).getValue(), innerFieldSchema);
+ m.put(((Map.Entry)entry).getKey(), newValue);
+ }
+ result = m;
+ }
+ else
+ result = obj;
} else if (obj instanceof DataByteArray) {
if (null != caster) {
- result = caster.bytesToMap(((DataByteArray)obj).get());
+ result = caster.bytesToMap(((DataByteArray)obj).get(), fs);
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to tuple.";
@@ -1195,6 +1206,16 @@ public class POCast extends ExpressionOp
case DataType.MAP: {
Result res = in.getNext(m);
+ if (res.returnStatus == POStatus.STATUS_OK && res.result != null) {
+ try {
+ res.result = convertWithSchema(res.result, fieldSchema);
+ } catch (IOException e) {
+ LogUtils.warn(this, "Unable to interpret value " + res.result + " in field being " +
+ "converted to type map, caught ParseException <" +
+ e.getMessage() + "> field discarded",
+ PigWarning.FIELD_DISCARDED_TYPE_CONVERSION_FAILED, log);
+ }
+ }
return res;
}
@@ -1232,7 +1253,7 @@ public class POCast extends ExpressionOp
}
try {
if (null != caster) {
- res.result = caster.bytesToMap(dba.get());
+ res.result = caster.bytesToMap(dba.get(), fieldSchema);
} else {
int errCode = 1075;
String msg = "Received a bytearray from the UDF. Cannot determine how to convert the bytearray to map.";
Modified: pig/trunk/src/org/apache/pig/backend/hadoop/hbase/HBaseBinaryConverter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/hbase/HBaseBinaryConverter.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/hbase/HBaseBinaryConverter.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/hbase/HBaseBinaryConverter.java Wed Mar 9 22:51:53 2011
@@ -71,11 +71,15 @@ public class HBaseBinaryConverter implem
}
}
+ @Override
+ public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ return bytesToMap(b, null);
+ }
/**
* NOT IMPLEMENTED
*/
@Override
- public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ public Map<String, Object> bytesToMap(byte[] b, ResourceFieldSchema fieldSchema) throws IOException {
throw new ExecException("Can't generate a Map from byte[]");
}
Modified: pig/trunk/src/org/apache/pig/builtin/BinStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/BinStorage.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/BinStorage.java (original)
+++ pig/trunk/src/org/apache/pig/builtin/BinStorage.java Wed Mar 9 22:51:53 2011
@@ -109,6 +109,11 @@ implements StoreFuncInterface, LoadMetad
@Override
public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ return bytesToMap(b, null);
+ }
+
+ @Override
+ public Map<String, Object> bytesToMap(byte[] b, ResourceFieldSchema fieldSchema) throws IOException {
throw new ExecException(unImplementedErrorMessage, 1118);
}
Modified: pig/trunk/src/org/apache/pig/builtin/TextLoader.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/TextLoader.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/TextLoader.java (original)
+++ pig/trunk/src/org/apache/pig/builtin/TextLoader.java Wed Mar 9 22:51:53 2011
@@ -121,12 +121,17 @@ public class TextLoader extends LoadFunc
return new String(b);
}
+ @Override
+ public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ return bytesToMap(b, null);
+ }
+
/**
* TextLoader does not support conversion to Map
* @throws IOException if the value cannot be cast.
*/
@Override
- public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ public Map<String, Object> bytesToMap(byte[] b, ResourceFieldSchema schema) throws IOException {
int errCode = 2109;
String msg = "TextLoader does not support conversion to Map.";
throw new ExecException(msg, errCode, PigException.BUG);
Modified: pig/trunk/src/org/apache/pig/builtin/Utf8StorageConverter.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/builtin/Utf8StorageConverter.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/builtin/Utf8StorageConverter.java (original)
+++ pig/trunk/src/org/apache/pig/builtin/Utf8StorageConverter.java Wed Mar 9 22:51:53 2011
@@ -30,6 +30,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pig.LoadStoreCaster;
import org.apache.pig.PigWarning;
+import org.apache.pig.ResourceSchema;
import org.apache.pig.ResourceSchema.ResourceFieldSchema;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
@@ -191,9 +192,6 @@ public class Utf8StorageConverter implem
}
private Map<String, Object> consumeMap(PushbackInputStream in, ResourceFieldSchema fieldSchema) throws IOException {
- if (fieldSchema==null) {
- throw new IOException("Schema is null");
- }
int buf;
while ((buf=in.read())!='[') {
@@ -239,9 +237,12 @@ public class Utf8StorageConverter implem
}
mOut.write(buf);
}
- DataByteArray value = null;
- if (mOut.size()>0)
+ Object value = null;
+ if (fieldSchema!=null && fieldSchema.getSchema()!=null && mOut.size()>0) {
+ value = bytesToObject(mOut.toByteArray(), fieldSchema.getSchema().getFields()[0]);
+ } else if (mOut.size()>0) { // untyped map
value = new DataByteArray(mOut.toByteArray());
+ }
m.put(key, value);
mOut.reset();
if (buf==']')
@@ -250,6 +251,19 @@ public class Utf8StorageConverter implem
return m;
}
+ private Object bytesToObject(byte[] b, ResourceFieldSchema fs) throws IOException {
+ Object field;
+ if (DataType.isComplex(fs.getType())) {
+ ByteArrayInputStream bis = new ByteArrayInputStream(b);
+ PushbackInputStream in = new PushbackInputStream(bis);
+ field = consumeComplexType(in, fs);
+ }
+ else {
+ field = parseSimpleType(b, fs);
+ }
+ return field;
+ }
+
private Object consumeComplexType(PushbackInputStream in, ResourceFieldSchema complexFieldSchema) throws IOException {
Object field;
switch (complexFieldSchema.getType()) {
@@ -446,16 +460,14 @@ public class Utf8StorageConverter implem
@Override
@SuppressWarnings("unchecked")
- public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ public Map<String, Object> bytesToMap(byte[] b, ResourceFieldSchema fieldSchema) throws IOException {
if(b == null)
return null;
Map<String, Object> map;
- ResourceFieldSchema fs = new ResourceFieldSchema();
- fs.setType(DataType.MAP);
try {
ByteArrayInputStream bis = new ByteArrayInputStream(b);
PushbackInputStream in = new PushbackInputStream(bis);
- map = consumeMap(in, fs);
+ map = consumeMap(in, fieldSchema);
}
catch (IOException e) {
LogUtils.warn(this, "Unable to interpret value " + Arrays.toString(b) + " in field being " +
@@ -466,6 +478,11 @@ public class Utf8StorageConverter implem
}
return map;
}
+
+ @Override
+ public Map<String, Object> bytesToMap(byte[] b) throws IOException {
+ return bytesToMap(b, null);
+ }
@Override
public Tuple bytesToTuple(byte[] b, ResourceFieldSchema fieldSchema) throws IOException {
Modified: pig/trunk/src/org/apache/pig/data/DataType.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/data/DataType.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/data/DataType.java (original)
+++ pig/trunk/src/org/apache/pig/data/DataType.java Wed Mar 9 22:51:53 2011
@@ -360,7 +360,7 @@ public class DataType {
* @return true if the type can have a valid schema (i.e., bag or tuple)
*/
public static boolean isSchemaType(byte dataType) {
- return ((dataType == BAG) || (dataType == TUPLE));
+ return ((dataType == BAG) || (dataType == TUPLE) || dataType == MAP);
}
/**
Modified: pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/Schema.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/Schema.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/Schema.java (original)
+++ pig/trunk/src/org/apache/pig/impl/logicalLayer/schema/Schema.java Wed Mar 9 22:51:53 2011
@@ -182,7 +182,7 @@ public class Schema implements Serializa
if ((null != s) && !(DataType.isSchemaType(t))) {
int errCode = 1020;
- throw new FrontendException("Only a BAG or TUPLE can have schemas. Got "
+ throw new FrontendException("Only a BAG, TUPLE or MAP can have schemas. Got "
+ DataType.findTypeName(t), errCode, PigException.INPUT);
}
@@ -1114,7 +1114,10 @@ public class Schema implements Serializa
+ "as inner schema") ;
}
} else if (fs.type == DataType.MAP) {
- sb.append(DataType.findTypeName(fs.type) + "[ ]") ;
+ sb.append(DataType.findTypeName(fs.type) + "[");
+ if (fs.schema!=null)
+ stringifySchema(sb, fs.schema, fs.type);
+ sb.append("]");
} else {
sb.append(DataType.findTypeName(fs.type)) ;
}
Modified: pig/trunk/src/org/apache/pig/newplan/logical/LogicalExpPlanMigrationVistor.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/LogicalExpPlanMigrationVistor.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/LogicalExpPlanMigrationVistor.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/LogicalExpPlanMigrationVistor.java Wed Mar 9 22:51:53 2011
@@ -326,21 +326,11 @@ public class LogicalExpPlanMigrationVist
}
public void visit(LOMapLookup colOp) throws VisitorException {
- FieldSchema fieldSchema;
- try {
- fieldSchema = colOp.getFieldSchema();
- } catch (FrontendException e) {
- throw new VisitorException( e.getMessage() );
- }
-
- LogicalSchema.LogicalFieldSchema logfieldSchema =
- new LogicalSchema.LogicalFieldSchema( fieldSchema.alias,
- Util.translateSchema(fieldSchema.schema), fieldSchema.type);
LogicalExpression map = exprOpsMap.get( colOp.getMap() );
MapLookupExpression op = new MapLookupExpression(exprPlan,
- colOp.getLookUpKey(), logfieldSchema);
+ colOp.getLookUpKey());
exprPlan.connect(op, map);
Modified: pig/trunk/src/org/apache/pig/newplan/logical/expression/MapLookupExpression.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/newplan/logical/expression/MapLookupExpression.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/newplan/logical/expression/MapLookupExpression.java (original)
+++ pig/trunk/src/org/apache/pig/newplan/logical/expression/MapLookupExpression.java Wed Mar 9 22:51:53 2011
@@ -34,13 +34,10 @@ public class MapLookupExpression extends
* type and schema of the value linked to the key
*/
private String mMapKey;
- private LogicalFieldSchema mValueSchema;
- public MapLookupExpression(OperatorPlan plan, String mapKey,
- LogicalFieldSchema valueSchema ) {
+ public MapLookupExpression(OperatorPlan plan, String mapKey ) {
super("Map", plan);
mMapKey = mapKey;
- mValueSchema = valueSchema;
plan.add(this);
}
/**
@@ -58,8 +55,7 @@ public class MapLookupExpression extends
public boolean isEqual(Operator other) throws FrontendException {
if (other != null && other instanceof MapLookupExpression) {
MapLookupExpression po = (MapLookupExpression)other;
- if ( po.mMapKey.compareTo(mMapKey) != 0 ||
- !po.mValueSchema.isEqual( mValueSchema ))
+ if ( po.mMapKey.compareTo(mMapKey) != 0)
return false;
else {
// check the nested map equality
@@ -94,12 +90,19 @@ public class MapLookupExpression extends
public LogicalFieldSchema getFieldSchema() throws FrontendException {
if (fieldSchema!=null)
return fieldSchema;
- if (mValueSchema!=null)
- fieldSchema = mValueSchema;
- else
- fieldSchema = new LogicalSchema.LogicalFieldSchema(null, null, DataType.BYTEARRAY);
- uidOnlyFieldSchema = fieldSchema.mergeUid(uidOnlyFieldSchema);
- return fieldSchema;
+ LogicalExpression successor = (LogicalExpression)plan.getSuccessors(this).get(0);
+ LogicalFieldSchema predFS = successor.getFieldSchema();
+ if (predFS!=null) {
+ if (predFS.type==DataType.MAP && predFS.schema!=null) {
+ return (predFS.schema.getField(0));
+ }
+ else {
+ fieldSchema = new LogicalSchema.LogicalFieldSchema(null, null, DataType.BYTEARRAY);
+ uidOnlyFieldSchema = fieldSchema.mergeUid(uidOnlyFieldSchema);
+ return fieldSchema;
+ }
+ }
+ return null;
}
public String toString() {
@@ -125,8 +128,7 @@ public class MapLookupExpression extends
LogicalExpression copy = null;;
copy = new MapLookupExpression(
lgExpPlan,
- this.getLookupKey(),
- this.getFieldSchema().deepCopy());
+ this.getLookupKey());
// Only one input is expected.
LogicalExpression input = (LogicalExpression) plan.getSuccessors( this ).get( 0 );
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=1080048&r1=1080047&r2=1080048&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 Wed Mar 9 22:51:53 2011
@@ -102,6 +102,12 @@ public class LogicalSchema {
return ( alias + uidString + ":tuple{}" );
}
return ( alias + uidString + ":tuple(" + schema.toString(verbose) + ")" );
+ } else if (type == DataType.MAP) {
+ if (schema == null ) {
+ return (alias + uidString + ":map");
+ } else {
+ return (alias + uidString + ":map(" + schema.toString(verbose) + ")");
+ }
}
return ( alias + uidString + ":" + DataType.findTypeName(type) );
}
@@ -280,7 +286,7 @@ public class LogicalSchema {
// 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)) {
+ if (!LogicalSchema.castable(inFs.schema, outFs.schema)) {
return false ;
}
}
@@ -499,8 +505,9 @@ public class LogicalSchema {
return false ;
}
+ // Cast to a more specific type is good
if (inSch == null) {
- return false ;
+ return true ;
}
if (outSch.size() > inSch.size()) return false;
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=1080048&r1=1080047&r2=1080048&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 Wed Mar 9 22:51:53 2011
@@ -88,8 +88,12 @@ public class ColumnAliasConversionVisito
LogicalSchema schema = null;
if( pred.getFieldSchema().type == DataType.BAG ) {
- if( pred.getFieldSchema().schema != null )
+ if( pred.getFieldSchema().schema != null ) {
schema = pred.getFieldSchema().schema.getField(0).schema;
+ if (schema!=null && schema.size()==1 && schema.getField(0).type==DataType.TUPLE) {
+ schema = schema.getField(0).schema;
+ }
+ }
}
else {
schema = pred.getFieldSchema().schema;
Modified: pig/trunk/src/org/apache/pig/parser/AliasMasker.g
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/AliasMasker.g?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/AliasMasker.g (original)
+++ pig/trunk/src/org/apache/pig/parser/AliasMasker.g Wed Mar 9 22:51:53 2011
@@ -219,7 +219,7 @@ bag_type
: ^( BAG_TYPE { sb.append("bag{"); } ( { sb.append("T:"); } tuple_type )? ) { sb.append("}"); }
;
-map_type : MAP_TYPE { sb.append("map[]"); }
+map_type : ^( MAP_TYPE { sb.append("map["); } type? ) { sb.append("]"); }
;
func_clause
Modified: pig/trunk/src/org/apache/pig/parser/AstValidator.g
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/AstValidator.g?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/AstValidator.g (original)
+++ pig/trunk/src/org/apache/pig/parser/AstValidator.g Wed Mar 9 22:51:53 2011
@@ -198,7 +198,7 @@ tuple_type : ^( TUPLE_TYPE field_def_lis
bag_type : ^( BAG_TYPE tuple_type? )
;
-map_type : MAP_TYPE
+map_type : ^( MAP_TYPE type? )
;
func_clause : ^( FUNC_REF func_name )
Modified: pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g (original)
+++ pig/trunk/src/org/apache/pig/parser/LogicalPlanGenerator.g Wed Mar 9 22:51:53 2011
@@ -326,14 +326,7 @@ field_def returns[LogicalFieldSchema fie
}
: ^( FIELD_DEF IDENTIFIER ( type { datatype = $type.datatype;} )? )
{
- if( datatype == DataType.BAG ) {
- LogicalFieldSchema tupleFieldSchema = new LogicalFieldSchema( null, $type.logicalSchema, DataType.TUPLE );
- LogicalSchema bagSchema = new LogicalSchema();
- bagSchema.addField( tupleFieldSchema );
- $fieldSchema = new LogicalFieldSchema( $IDENTIFIER.text, bagSchema, DataType.BAG );
- } else {
$fieldSchema = new LogicalFieldSchema( $IDENTIFIER.text, $type.logicalSchema, datatype );
- }
}
;
@@ -363,6 +356,7 @@ type returns[Byte datatype, LogicalSchem
| map_type
{
$datatype = DataType.MAP;
+ $logicalSchema = $map_type.logicalSchema;
}
;
@@ -391,12 +385,28 @@ tuple_type returns[LogicalSchema logical
bag_type returns[LogicalSchema logicalSchema]
: ^( BAG_TYPE tuple_type? )
- {
- $logicalSchema = $tuple_type.logicalSchema;
+ {
+ if ($tuple_type.logicalSchema!=null && $tuple_type.logicalSchema.size()==1 && $tuple_type.logicalSchema.getField(0).type==DataType.TUPLE) {
+ $logicalSchema = $tuple_type.logicalSchema;
+ }
+ else {
+ LogicalSchema s = new LogicalSchema();
+ s.addField(new LogicalFieldSchema(null, $tuple_type.logicalSchema, DataType.TUPLE));
+ $logicalSchema = s;
+ }
}
;
-map_type : MAP_TYPE
+map_type returns[LogicalSchema logicalSchema]
+ : ^( MAP_TYPE type? )
+ {
+ LogicalSchema s = null;
+ if ($type.datatype!=null) {
+ s = new LogicalSchema();
+ s.addField(new LogicalFieldSchema(null, $type.logicalSchema, $type.datatype));
+ }
+ $logicalSchema = s;
+ }
;
func_clause[byte ft] returns[FuncSpec funcSpec]
@@ -681,7 +691,7 @@ type_cast returns[LogicalFieldSchema fie
}
| map_type
{
- $fieldSchema = new LogicalFieldSchema( null, null, DataType.MAP );
+ $fieldSchema = new LogicalFieldSchema( null, $map_type.logicalSchema, DataType.MAP );
}
| tuple_type_cast
{
@@ -759,7 +769,7 @@ var_expr[LogicalExpressionPlan plan] ret
}
| pound_proj
{
- MapLookupExpression e = new MapLookupExpression( $plan, $pound_proj.key, null );
+ MapLookupExpression e = new MapLookupExpression( $plan, $pound_proj.key );
$plan.connect( e, $expr );
$expr = e;
}
Modified: pig/trunk/src/org/apache/pig/parser/QueryParser.g
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/parser/QueryParser.g?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/parser/QueryParser.g (original)
+++ pig/trunk/src/org/apache/pig/parser/QueryParser.g Wed Mar 9 22:51:53 2011
@@ -220,8 +220,8 @@ bag_type : BAG? LEFT_CURLY ( ( IDENTIFIE
-> ^( BAG_TYPE tuple_type? )
;
-map_type : MAP? LEFT_BRACKET RIGHT_BRACKET
- -> MAP_TYPE
+map_type : MAP? LEFT_BRACKET type? RIGHT_BRACKET
+ -> ^( MAP_TYPE type? )
;
func_clause : func_name
Modified: pig/trunk/test/org/apache/pig/test/TestMacroExpansion.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestMacroExpansion.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestMacroExpansion.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestMacroExpansion.java Wed Mar 9 22:51:53 2011
@@ -1556,6 +1556,38 @@ public class TestMacroExpansion {
testMacro ( query );
}
+ @Test
+ public void test32() throws Throwable {
+ String query = "a = load 'testSimpleMapKeyLookup' as (m:map[int]);" +
+ "b = foreach a generate m#'key';";
+
+ testMacro( query );
+ }
+
+ @Test
+ public void test33() throws Throwable {
+ String query = "a = load 'testSimpleMapCast' as (m);" +
+ "b = foreach a generate ([int])m;";
+
+ testMacro( query );
+ }
+
+ @Test
+ public void test34() throws Throwable {
+ String query = "a = load 'testComplexLoad' as (m:map[bag{(i:int,j:int)}]);";
+
+ testMacro( query );
+ }
+
+ @Test
+ public void test35() throws Throwable {
+ String query = "a = load 'testComplexCast' as (m);" +
+ "b = foreach a generate ([{(i:int,j:int)}])m;";
+
+ testMacro( query );
+ }
+
+ //-------------------------------------------------------------------------
private void testMacro(String content) throws Throwable {
String macro = "define mymacro() returns dummy {" +
@@ -1567,6 +1599,8 @@ public class TestMacroExpansion {
StringReader rd = new StringReader(script);
String s = ParserUtil.expandMacros(rd);
+ System.out.println("result:\n" + s);
+
validate(s);
}
Modified: pig/trunk/test/org/apache/pig/test/TestPOCast.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestPOCast.java?rev=1080048&r1=1080047&r2=1080048&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestPOCast.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestPOCast.java Wed Mar 9 22:51:53 2011
@@ -959,6 +959,10 @@ public class TestPOCast extends TestCase
public Map<String, Object> bytesToMap(byte[] b) throws IOException {
return null;
}
+
+ public Map<String, Object> bytesToMap(byte[] b, ResourceFieldSchema s) throws IOException {
+ return null;
+ }
public Tuple bytesToTuple(byte[] b, ResourceFieldSchema s) throws IOException {
return null;
Added: pig/trunk/test/org/apache/pig/test/TestTypedMap.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestTypedMap.java?rev=1080048&view=auto
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestTypedMap.java (added)
+++ pig/trunk/test/org/apache/pig/test/TestTypedMap.java Wed Mar 9 22:51:53 2011
@@ -0,0 +1,364 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.lib.FieldSelectionMapReduce;
+import org.apache.pig.EvalFunc;
+import org.apache.pig.ExecType;
+import org.apache.pig.PigException;
+import org.apache.pig.PigServer;
+import org.apache.pig.backend.executionengine.ExecException;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.DataByteArray;
+import org.apache.pig.data.DataType;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.parser.ParseException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.logicalLayer.schema.Schema.FieldSchema;
+import org.apache.pig.impl.util.LogUtils;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestTypedMap {
+ static MiniCluster cluster ;
+
+ @Before
+ public void setUp() throws Exception {
+ FileLocalizer.setInitialized(false);
+ }
+
+
+ @After
+ public void tearDown() throws Exception {
+ }
+
+ @BeforeClass
+ public static void oneTimeSetup() throws IOException, Exception {
+ cluster = MiniCluster.buildCluster();
+ }
+
+ @AfterClass
+ public static void oneTimeTearDown() throws Exception {
+ cluster.shutDown();
+ }
+
+ @Test
+ public void testSimpleLoad() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#1,key2#2]",
+ "[key#2]",
+ };
+
+ Util.createInputFile(cluster, "table_testSimpleLoad", input);
+
+ String query =
+ " a = load 'table_testSimpleLoad' as (m:map[int]);";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("a");
+ assertEquals("Checking expected schema",sch.toString(), "{m: map[int]}");
+ Iterator<Tuple> it = pig.openIterator("a");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue(t.get(0) instanceof Map);
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key2"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof Integer);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("1"));
+ Assert.assertTrue(((Map)t.get(0)).get("key2") instanceof Integer);
+ Assert.assertTrue(((Map)t.get(0)).get("key2").toString().equals("2"));
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof Integer);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("2"));
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testSimpleMapKeyLookup() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#1,key2#2]",
+ "[key#2]",
+ };
+
+ Util.createInputFile(cluster, "testSimpleMapKeyLookup", input);
+
+ String query =
+ "a = load 'testSimpleMapKeyLookup' as (m:map[int]);" +
+ "b = foreach a generate m#'key';";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("b");
+ assertEquals("Checking expected schema",sch.toString(), "{int}");
+ Iterator<Tuple> it = pig.openIterator("b");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue((Integer)t.get(0)==1);
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue((Integer)t.get(0)==2);
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testSimpleMapCast() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#1,key2#2]",
+ "[key#2]",
+ };
+
+ Util.createInputFile(cluster, "testSimpleMapCast", input);
+
+ String query =
+ "a = load 'testSimpleMapCast' as (m);" +
+ "b = foreach a generate ([int])m;";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("b");
+ assertEquals("Checking expected schema",sch.toString(), "{m: map[int]}");
+ Iterator<Tuple> it = pig.openIterator("b");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue(t.get(0) instanceof Map);
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key2"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof Integer);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("1"));
+ Assert.assertTrue(((Map)t.get(0)).get("key2") instanceof Integer);
+ Assert.assertTrue(((Map)t.get(0)).get("key2").toString().equals("2"));
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof Integer);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("2"));
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testComplexLoad() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#{(1,2),(1,3)},134#]",
+ "[key2#]",
+ };
+
+ Util.createInputFile(cluster, "testComplexLoad", input);
+
+ String query = "a = load 'testComplexLoad' as (m:map[bag{(i:int,j:int)}]);";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("a");
+ assertEquals("Checking expected schema",sch.toString(), "{m: map[{(i: int,j: int)}]}");
+ Iterator<Tuple> it = pig.openIterator("a");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue(t.get(0) instanceof Map);
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).containsKey("134"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof DataBag);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("{(1,2),(1,3)}"));
+ Assert.assertTrue(((Map)t.get(0)).get("134")==null);
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key2"));
+ Assert.assertTrue(((Map)t.get(0)).get("key2")==null);
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testComplexCast() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#{(1,2),(1,3)},134#]",
+ "[key2#]",
+ };
+
+ Util.createInputFile(cluster, "testComplexCast", input);
+
+ String query = "a = load 'testComplexCast' as (m);" +
+ "b = foreach a generate ([{(i:int,j:int)}])m;";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("b");
+ assertEquals("Checking expected schema",sch.toString(), "{m: map[{(i: int,j: int)}]}");
+ Iterator<Tuple> it = pig.openIterator("b");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue(t.get(0) instanceof Map);
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).containsKey("134"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof DataBag);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("{(1,2),(1,3)}"));
+ Assert.assertTrue(((Map)t.get(0)).get("134")==null);
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key2"));
+ Assert.assertTrue(((Map)t.get(0)).get("key2")==null);
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testComplexCast2() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#1,key2#2]",
+ };
+
+ Util.createInputFile(cluster, "testComplexCast2", input);
+
+ String query = "a = load 'testComplexCast2' as (m:[int]);" +
+ "b = foreach a generate ([long])m;";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("b");
+ assertEquals("Checking expected schema",sch.toString(), "{m: map[long]}");
+ Iterator<Tuple> it = pig.openIterator("b");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue(t.get(0) instanceof Map);
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key2"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof Long);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("1"));
+ Assert.assertTrue(((Map)t.get(0)).get("key2") instanceof Long);
+ Assert.assertTrue(((Map)t.get(0)).get("key2").toString().equals("2"));
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testUnTypedMap() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#1,key2#2]",
+ };
+
+ Util.createInputFile(cluster, "testUnTypedMap", input);
+
+ String query = "a = load 'testUnTypedMap' as (m:[]);";
+ Util.registerMultiLineQuery(pig, query);
+ Schema sch = pig.dumpSchema("a");
+ assertEquals("Checking expected schema",sch.toString(), "{m: map[]}");
+ Iterator<Tuple> it = pig.openIterator("a");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.size()==1);
+ Assert.assertTrue(t.get(0) instanceof Map);
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key"));
+ Assert.assertTrue(((Map)t.get(0)).containsKey("key2"));
+ Assert.assertTrue(((Map)t.get(0)).get("key") instanceof DataByteArray);
+ Assert.assertTrue(((Map)t.get(0)).get("key").toString().equals("1"));
+ Assert.assertTrue(((Map)t.get(0)).get("key2") instanceof DataByteArray);
+ Assert.assertTrue(((Map)t.get(0)).get("key2").toString().equals("2"));
+
+ Assert.assertFalse(it.hasNext());
+ }
+
+ @Test
+ public void testOrderBy() throws IOException, ParseException {
+ PigServer pig = new PigServer(ExecType.MAPREDUCE, cluster.getProperties());
+ String[] input = {
+ "[key#1,key1#2]",
+ "[key#2,key3#2]",
+ "[key#11]",
+ };
+
+ Util.createInputFile(cluster, "testOrderBy", input);
+
+ String query = "a = load 'testOrderBy' as (m:[int]);" +
+ "b = foreach a generate m#'key' as b0;" +
+ "c = order b by b0;";
+ Util.registerMultiLineQuery(pig, query);
+ Iterator<Tuple> it = pig.openIterator("c");
+
+ Assert.assertTrue(it.hasNext());
+ Tuple t = it.next();
+ Assert.assertTrue(t.toString().equals("(1)"));
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(t.toString().equals("(2)"));
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(t.toString().equals("(11)"));
+
+ Assert.assertFalse(it.hasNext());
+
+ // Untyped map will sort by byte type, which is different
+ query = "a = load 'testOrderBy' as (m:[]);" +
+ "b = foreach a generate m#'key' as b0;" +
+ "c = order b by b0;";
+ Util.registerMultiLineQuery(pig, query);
+ it = pig.openIterator("c");
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(t.toString().equals("(1)"));
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(t.toString().equals("(11)"));
+
+ Assert.assertTrue(it.hasNext());
+ t = it.next();
+ Assert.assertTrue(t.toString().equals("(2)"));
+ }
+}