src/org/apache/pig/data/Tuple.java - pig - Git at Google

 /*
  * 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.data;

 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;

 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.io.WritableComparable;


 /**
  * an ordered list of Datums
  */
 public class Tuple extends Datum implements WritableComparable {

     private static final Log log = LogFactory.getLog(Tuple.class);
     private static int numFields = 5;

     protected ArrayList<Datum> fields;
     static String              defaultDelimiter = "[,\t]";
     static String              NULL = "__PIG_NULL__";

     public Tuple() {
         this(0);
     }

     public Tuple(int numFields) {
         fields = new ArrayList<Datum>(numFields);
         for (int i = 0; i < numFields; i++) {
             fields.add(null);
         }
     }

     public Tuple(List<Datum> fieldsIn) {
         fields = new ArrayList<Datum>(fieldsIn.size());
         fields.addAll(fieldsIn);
     }

     /**
      * shortcut, if tuple only has one field
      */
     public Tuple(Datum fieldIn) {
         fields = new ArrayList<Datum>(1);
         fields.add(fieldIn);
     }

     /**
      * Creates a tuple from a delimited line of text
      *
      * @param textLine
      *            the line containing fields of data
      * @param delimiter
      *              the delimiter (normal string, NO REGEX!!)
      */
     public Tuple(String textLine, String delimiter) {
         if (delimiter == null) {
             delimiter = defaultDelimiter;
         }

         fields = new ArrayList<Datum>(numFields) ;
         int delimSize = delimiter.length() ;
         boolean done = false ;

         int lastIdx = 0 ;

         while (!done) {
             int newIdx = textLine.indexOf(delimiter, lastIdx) ;
             if (newIdx != (-1)) {
                 String token = textLine.substring(lastIdx, newIdx) ;
                 fields.add(new DataAtom(token));
                 lastIdx = newIdx + delimSize  ;
             }
             else {
                 String token = textLine.substring(lastIdx) ;
                 fields.add(new DataAtom(token));
                 done = true ;
             }
         }

         numFields = fields.size();
     }

     /**
      * Creates a tuple from a delimited line of text. This will invoke Tuple(textLine, null)
      *
      * @param textLine
      *            the line containing fields of data
      */
     public Tuple(String textLine) {
         this(textLine, defaultDelimiter);
     }

     public Tuple(Tuple[] otherTs) {
         fields = new ArrayList<Datum>(otherTs.length);
         for (int i = 0; i < otherTs.length; i++) {
                 appendTuple(otherTs[i]);
         }
     }

     public void copyFrom(Tuple otherT) {
         this.fields = otherT.fields;
     }

     public int arity() {
         return fields.size();
     }

     @Override
     public String toString() {
         StringBuffer sb = new StringBuffer();
         sb.append('(');
         for (Iterator<Datum> it = fields.iterator(); it.hasNext();) {
             Datum d = it.next();
             if(d != null) {
                 sb.append(d.toString());
             } else {
                 sb.append(NULL);
             }
             if (it.hasNext())
                 sb.append(", ");
         }
         sb.append(')');
         String s = sb.toString();
         return s;
     }

     public void setField(int i, Datum val) {
         getField(i); // throws exception if field doesn't exist

         fields.set(i, val);
     }

     public void setField(int i, int val) {
         setField(i, new DataAtom(val));
     }

     public void setField(int i, double val) {
         setField(i, new DataAtom(val));
     }

     public void setField(int i, String val) {
         setField(i, new DataAtom(val));
     }

     public Datum getField(int i) {
         if (fields.size() >= i + 1) {
             return fields.get(i);
         }

         StringBuilder sb = new StringBuilder();
         sb.append("Requested index ");
         sb.append(i);
         sb.append(" from tuple ");
         sb.append(toString());
         throw new IndexOutOfBoundsException(sb.toString());
     }

     // Get field i, if it is an Atom or can be coerced into an Atom
     public DataAtom getAtomField(int i) {
         Datum field = getField(i); // throws exception if field doesn't exist

         if (field instanceof DataAtom) {
             return (DataAtom) field;
         } else if (field instanceof Tuple) {
             Tuple t = (Tuple) field;
             if (t.arity() == 1) {
                 log.warn("Requested for an atom field but found a tuple with one field.");
                 return t.getAtomField(0);
             }
         } else if (field instanceof DataBag) {
             DataBag b = (DataBag) field;
             if (b.size() == 1) {
                 Tuple t = b.iterator().next();
                 if (t.arity() == 1) {
                     return t.getAtomField(0);
                 }
             }
         }

         throw newTupleAccessException(field, "atom", i);
     }

     private RuntimeException newTupleAccessException(Datum field,
             String requestedFieldType, int index) {
         return new IllegalArgumentException("Requested " + requestedFieldType
                 + " field at index " + index + " but was '"
                 + field.getClass().getName() + "' in tuple: " + toString());
     }

     // Get field i, if it is a Tuple or can be coerced into a Tuple
     public Tuple getTupleField(int i) {
         Datum field = getField(i); // throws exception if field doesn't exist

         if (field instanceof Tuple) {
             return (Tuple) field;
         } else if (field instanceof DataBag) {
             DataBag b = (DataBag) field;
             if (b.size() == 1) {
                 return b.iterator().next();
             }
         }

         throw newTupleAccessException(field, "tuple", i);
     }

     // Get field i, if it is a Bag or can be coerced into a Bag
     public DataBag getBagField(int i) {
         Datum field = getField(i); // throws exception if field doesn't exist

         if (field instanceof DataBag) {
             return (DataBag) field;
         }

         throw newTupleAccessException(field, "bag", i);
     }

     public void appendTuple(Tuple other){
         for (Iterator<Datum> it = other.fields.iterator(); it.hasNext();) {
             this.fields.add(it.next());
         }
     }

     public void appendField(Datum newField){
         this.fields.add(newField);
     }

     public String toDelimitedString(String delim) throws IOException {
         StringBuffer buf = new StringBuffer();
         for (Iterator<Datum> it = fields.iterator(); it.hasNext();) {
             Datum field = it.next();
             if (!(field instanceof DataAtom)) {
                 throw new IOException("Unable to convert non-flat tuple to string.");
             }

             buf.append((DataAtom) field);
             if (it.hasNext())
                 buf.append(delim);
         }
         return buf.toString();
     }

     public boolean lessThan(Tuple other) {
         return (this.compareTo(other) < 0);
     }

     public boolean greaterThan(Tuple other) {
         return (this.compareTo(other) > 0);
     }

     @Override
     public boolean equals(Object other){
             return compareTo(other)==0;
     }

     public int compareTo(Tuple other) {
         if (other.fields.size() != this.fields.size())
             return other.fields.size() < this.fields.size() ? 1 : -1;

         for (int i = 0; i < this.fields.size(); i++) {
             int c = this.fields.get(i).compareTo(other.fields.get(i));
             if (c != 0)
                 return c;
         }
         return 0;
     }

     public int compareTo(Object other) {
         if (other instanceof DataAtom)
             return +1;
         else if (other instanceof DataBag)
             return +1;
         else if (other instanceof Tuple)
             return compareTo((Tuple) other);
         else
             return -1;
     }

     @Override
     public int hashCode() {
         int hash = 1;
         for (Iterator<Datum> it = fields.iterator(); it.hasNext();) {
             hash = 31 * hash + it.next().hashCode();
         }
         return hash;
     }

     // WritableComparable methods:

     @Override
     public void write(DataOutput out) throws IOException {
         out.write(TUPLE);
         int n = arity();
         encodeInt(out, n);
         for (int i = 0; i < n; i++) {
             Datum d = getField(i);
             if (d!=null){
                 d.write(out);
             }else{
                 throw new RuntimeException("Null field in tuple");
             }
         }
     }

     //This method is invoked when the beginning 'TUPLE' is still on the stream
     public void readFields(DataInput in) throws IOException {
         byte[] b = new byte[1];
         in.readFully(b);
         if (b[0]!=TUPLE)
             throw new IOException("Unexpected data while reading tuple from binary file");
         Tuple t = read(in);
         fields = t.fields;
     }

     //This method is invoked when the beginning 'TUPLE' has been read off the stream
     public static Tuple read(DataInput in) throws IOException {
         // nuke the old contents of the tuple
         Tuple ret = new Tuple();
         ret.fields = new ArrayList<Datum>();

         int size = decodeInt(in);

         for (int i = 0; i < size; i++) {
             ret.appendField(readDatum(in));
         }

         return ret;

     }

     public static Datum readDatum(DataInput in) throws IOException{
         byte[] b = new byte[1];
         in.readFully(b);
         switch (b[0]) {
             case TUPLE:
                 return Tuple.read(in);
             case BAG:
                 return DataBag.read(in);
             case MAP:
                 return DataMap.read(in);
             case ATOM:
                 return DataAtom.read(in);
             default:
                 throw new IOException("Invalid data while reading Datum from binary file");
         }
     }

     //  Encode the integer so that the high bit is set on the last
     // byte
     static void encodeInt(DataOutput os, int i) throws IOException {
         if (i >> 28 != 0)
             os.write((i >> 28) & 0x7f);
         if (i >> 21 != 0)
             os.write((i >> 21) & 0x7f);
         if (i >> 14 != 0)
             os.write((i >> 14) & 0x7f);
         if (i >> 7 != 0)
             os.write((i >> 7) & 0x7f);
         os.write((i & 0x7f) | (1 << 7));
     }

     static int decodeInt(DataInput is) throws IOException {
         int i = 0;
         int c;
         while (true) {
             c = is.readUnsignedByte();
             if (c == -1)
                 break;
             i <<= 7;
             i += c & 0x7f;
             if ((c & 0x80) != 0)
                 break;
         }
         return i;
     }

     @Override
     public long getMemorySize() {
         long used = 0;
         int sz = fields.size();
         for (int i = 0; i < sz; i++)
             used += getField(i).getMemorySize();
         used += 2 * OBJECT_SIZE + REF_SIZE;
         return used;
     }
 }
	/*
	* 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.data;

	import java.io.DataInput;
	import java.io.DataOutput;
	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Iterator;
	import java.util.List;

	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;
	import org.apache.hadoop.io.WritableComparable;


	/**
	* an ordered list of Datums
	*/
	public class Tuple extends Datum implements WritableComparable {

	private static final Log log = LogFactory.getLog(Tuple.class);
	private static int numFields = 5;

	protected ArrayList<Datum> fields;
	static String defaultDelimiter = "[,\t]";
	static String NULL = "__PIG_NULL__";

	public Tuple() {
	this(0);
	}

	public Tuple(int numFields) {
	fields = new ArrayList<Datum>(numFields);
	for (int i = 0; i < numFields; i++) {
	fields.add(null);
	}
	}

	public Tuple(List<Datum> fieldsIn) {
	fields = new ArrayList<Datum>(fieldsIn.size());
	fields.addAll(fieldsIn);
	}

	/**
	* shortcut, if tuple only has one field
	*/
	public Tuple(Datum fieldIn) {
	fields = new ArrayList<Datum>(1);
	fields.add(fieldIn);
	}

	/**
	* Creates a tuple from a delimited line of text
	*
	* @param textLine
	* the line containing fields of data
	* @param delimiter
	* the delimiter (normal string, NO REGEX!!)
	*/
	public Tuple(String textLine, String delimiter) {
	if (delimiter == null) {
	delimiter = defaultDelimiter;
	}

	fields = new ArrayList<Datum>(numFields) ;
	int delimSize = delimiter.length() ;
	boolean done = false ;

	int lastIdx = 0 ;

	while (!done) {
	int newIdx = textLine.indexOf(delimiter, lastIdx) ;
	if (newIdx != (-1)) {
	String token = textLine.substring(lastIdx, newIdx) ;
	fields.add(new DataAtom(token));
	lastIdx = newIdx + delimSize ;
	}
	else {
	String token = textLine.substring(lastIdx) ;
	fields.add(new DataAtom(token));
	done = true ;
	}
	}

	numFields = fields.size();
	}

	/**
	* Creates a tuple from a delimited line of text. This will invoke Tuple(textLine, null)
	*
	* @param textLine
	* the line containing fields of data
	*/
	public Tuple(String textLine) {
	this(textLine, defaultDelimiter);
	}

	public Tuple(Tuple[] otherTs) {
	fields = new ArrayList<Datum>(otherTs.length);
	for (int i = 0; i < otherTs.length; i++) {
	appendTuple(otherTs[i]);
	}
	}

	public void copyFrom(Tuple otherT) {
	this.fields = otherT.fields;
	}

	public int arity() {
	return fields.size();
	}

	@Override
	public String toString() {
	StringBuffer sb = new StringBuffer();
	sb.append('(');
	for (Iterator<Datum> it = fields.iterator(); it.hasNext();) {
	Datum d = it.next();
	if(d != null) {
	sb.append(d.toString());
	} else {
	sb.append(NULL);
	}
	if (it.hasNext())
	sb.append(", ");
	}
	sb.append(')');
	String s = sb.toString();
	return s;
	}

	public void setField(int i, Datum val) {
	getField(i); // throws exception if field doesn't exist

	fields.set(i, val);
	}

	public void setField(int i, int val) {
	setField(i, new DataAtom(val));
	}

	public void setField(int i, double val) {
	setField(i, new DataAtom(val));
	}

	public void setField(int i, String val) {
	setField(i, new DataAtom(val));
	}

	public Datum getField(int i) {
	if (fields.size() >= i + 1) {
	return fields.get(i);
	}

	StringBuilder sb = new StringBuilder();
	sb.append("Requested index ");
	sb.append(i);
	sb.append(" from tuple ");
	sb.append(toString());
	throw new IndexOutOfBoundsException(sb.toString());
	}

	// Get field i, if it is an Atom or can be coerced into an Atom
	public DataAtom getAtomField(int i) {
	Datum field = getField(i); // throws exception if field doesn't exist

	if (field instanceof DataAtom) {
	return (DataAtom) field;
	} else if (field instanceof Tuple) {
	Tuple t = (Tuple) field;
	if (t.arity() == 1) {
	log.warn("Requested for an atom field but found a tuple with one field.");
	return t.getAtomField(0);
	}
	} else if (field instanceof DataBag) {
	DataBag b = (DataBag) field;
	if (b.size() == 1) {
	Tuple t = b.iterator().next();
	if (t.arity() == 1) {
	return t.getAtomField(0);
	}
	}
	}

	throw newTupleAccessException(field, "atom", i);
	}

	private RuntimeException newTupleAccessException(Datum field,
	String requestedFieldType, int index) {
	return new IllegalArgumentException("Requested " + requestedFieldType
	+ " field at index " + index + " but was '"
	+ field.getClass().getName() + "' in tuple: " + toString());
	}

	// Get field i, if it is a Tuple or can be coerced into a Tuple
	public Tuple getTupleField(int i) {
	Datum field = getField(i); // throws exception if field doesn't exist

	if (field instanceof Tuple) {
	return (Tuple) field;
	} else if (field instanceof DataBag) {
	DataBag b = (DataBag) field;
	if (b.size() == 1) {
	return b.iterator().next();
	}
	}

	throw newTupleAccessException(field, "tuple", i);
	}

	// Get field i, if it is a Bag or can be coerced into a Bag
	public DataBag getBagField(int i) {
	Datum field = getField(i); // throws exception if field doesn't exist

	if (field instanceof DataBag) {
	return (DataBag) field;
	}

	throw newTupleAccessException(field, "bag", i);
	}

	public void appendTuple(Tuple other){
	for (Iterator<Datum> it = other.fields.iterator(); it.hasNext();) {
	this.fields.add(it.next());
	}
	}

	public void appendField(Datum newField){
	this.fields.add(newField);
	}

	public String toDelimitedString(String delim) throws IOException {
	StringBuffer buf = new StringBuffer();
	for (Iterator<Datum> it = fields.iterator(); it.hasNext();) {
	Datum field = it.next();
	if (!(field instanceof DataAtom)) {
	throw new IOException("Unable to convert non-flat tuple to string.");
	}

	buf.append((DataAtom) field);
	if (it.hasNext())
	buf.append(delim);
	}
	return buf.toString();
	}

	public boolean lessThan(Tuple other) {
	return (this.compareTo(other) < 0);
	}

	public boolean greaterThan(Tuple other) {
	return (this.compareTo(other) > 0);
	}

	@Override
	public boolean equals(Object other){
	return compareTo(other)==0;
	}

	public int compareTo(Tuple other) {
	if (other.fields.size() != this.fields.size())
	return other.fields.size() < this.fields.size() ? 1 : -1;

	for (int i = 0; i < this.fields.size(); i++) {
	int c = this.fields.get(i).compareTo(other.fields.get(i));
	if (c != 0)
	return c;
	}
	return 0;
	}

	public int compareTo(Object other) {
	if (other instanceof DataAtom)
	return +1;
	else if (other instanceof DataBag)
	return +1;
	else if (other instanceof Tuple)
	return compareTo((Tuple) other);
	else
	return -1;
	}

	@Override
	public int hashCode() {
	int hash = 1;
	for (Iterator<Datum> it = fields.iterator(); it.hasNext();) {
	hash = 31 * hash + it.next().hashCode();
	}
	return hash;
	}

	// WritableComparable methods:

	@Override
	public void write(DataOutput out) throws IOException {
	out.write(TUPLE);
	int n = arity();
	encodeInt(out, n);
	for (int i = 0; i < n; i++) {
	Datum d = getField(i);
	if (d!=null){
	d.write(out);
	}else{
	throw new RuntimeException("Null field in tuple");
	}
	}
	}

	//This method is invoked when the beginning 'TUPLE' is still on the stream
	public void readFields(DataInput in) throws IOException {
	byte[] b = new byte[1];
	in.readFully(b);
	if (b[0]!=TUPLE)
	throw new IOException("Unexpected data while reading tuple from binary file");
	Tuple t = read(in);
	fields = t.fields;
	}

	//This method is invoked when the beginning 'TUPLE' has been read off the stream
	public static Tuple read(DataInput in) throws IOException {
	// nuke the old contents of the tuple
	Tuple ret = new Tuple();
	ret.fields = new ArrayList<Datum>();

	int size = decodeInt(in);

	for (int i = 0; i < size; i++) {
	ret.appendField(readDatum(in));
	}

	return ret;

	}

	public static Datum readDatum(DataInput in) throws IOException{
	byte[] b = new byte[1];
	in.readFully(b);
	switch (b[0]) {
	case TUPLE:
	return Tuple.read(in);
	case BAG:
	return DataBag.read(in);
	case MAP:
	return DataMap.read(in);
	case ATOM:
	return DataAtom.read(in);
	default:
	throw new IOException("Invalid data while reading Datum from binary file");
	}
	}

	// Encode the integer so that the high bit is set on the last
	// byte
	static void encodeInt(DataOutput os, int i) throws IOException {
	if (i >> 28 != 0)
	os.write((i >> 28) & 0x7f);
	if (i >> 21 != 0)
	os.write((i >> 21) & 0x7f);
	if (i >> 14 != 0)
	os.write((i >> 14) & 0x7f);
	if (i >> 7 != 0)
	os.write((i >> 7) & 0x7f);
	os.write((i & 0x7f) \| (1 << 7));
	}

	static int decodeInt(DataInput is) throws IOException {
	int i = 0;
	int c;
	while (true) {
	c = is.readUnsignedByte();
	if (c == -1)
	break;
	i <<= 7;
	i += c & 0x7f;
	if ((c & 0x80) != 0)
	break;
	}
	return i;
	}

	@Override
	public long getMemorySize() {
	long used = 0;
	int sz = fields.size();
	for (int i = 0; i < sz; i++)
	used += getField(i).getMemorySize();
	used += 2 * OBJECT_SIZE + REF_SIZE;
	return used;
	}
	}