source: sample/hadoop-0.17/tw/org/nchc/tuple/Tuple.java @ 20

/*
 * Cloud9: A MapReduce Library for Hadoop
 * 
 * Licensed 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 tw.org.nchc.tuple;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;

/**
 * <p>
 * Class that represents a tuple in Hadoop's data type system. Tuples are
 * instantiated from a {@link Schema}. The Tuple class implements
 * WritableComparable, so it can be directly used as MapReduce keys and values.
 * The natural sort order of tuples is defined by an internally-generated byte
 * representation and is not based on field values. This class, combined with
 * {@link ListWritable}, allows the user to define arbitrarily complex data
 * structures.
 * </p>
 * 
 * <p>
 * All fields can either be indexed via its integer position or its field name.
 * Each field is typed, which can be determined via {@link #getFieldType(int)}.
 * Fields can either contain an object of the specified type or a special symbol
 * String. The method {@link #containsSymbol(int)} can be used to check if a
 * field contains a special symbol. If the field contains a special symbol,
 * {@link #get(int)} will return <code>null</code>. If the field does not
 * contain a special symbol, {@link #getSymbol(int)} will return
 * <code>null</code>.
 * </p>
 * 
 * <p>
 * Here is a typical usage scenario for special symbols: say you had tuples that
 * represented <code>count(a, b)</code>, where <code>a</code> and
 * <code>b</code> are tokens you observe. There is often a need to compute
 * <code>count(a, *)</code>, for example, to derive conditional
 * probabilities. In this case, you can use a special symbol to represent the
 * <code>*</code>, and distinguish it from the lexical token '<code>*</code>'.
 * </p>
 * 
 * <p>
 * The natural sort order of the Tuple is defined by {@link #compareTo(Object)}.
 * Tuples are sorted by field, with special symbols always appearing first
 * within each field.
 * </p>
 * 
 * @see ListWritable
 * @see Schema
 * 
 */
public class Tuple implements WritableComparable {

  protected static final byte SYMBOL = 0;
  protected static final byte INT = 1;
  protected static final byte BOOLEAN = 2;
  protected static final byte LONG = 3;
  protected static final byte FLOAT = 4;
  protected static final byte DOUBLE = 5;
  protected static final byte STRING = 6;
  protected static final byte WRITABLE = 7;

  private Object[] mObjects;
  private String[] mSymbols;
  private String[] mFields;
  private Class<?>[] mTypes;

  private Map<String, Integer> mFieldLookup = null;

  protected Tuple(Object[] objects, String[] symbols, String[] fields,
      Class<?>[] types) {
    mObjects = objects;
    mSymbols = symbols;
    mFields = fields;
    mTypes = types;
  }

  /**
   * Creates an empty Tuple. This constructor is needed by Hadoop's framework
   * for deserializing Writable objects. The preferred way to instantiate
   * tuples is through {@link Schema#instantiate(Object...)}.
   */
  public Tuple() {
  }

  /**
   * Factory method for deserializing a Tuple object.
   * 
   * @param in
   *            raw byte source of the Tuple
   * @return a new Tuple
   * @throws IOException
   */
  public static Tuple createFrom(DataInput in) throws IOException {
    Tuple tuple = new Tuple();
    tuple.readFields(in);

    return tuple;
  }

  /**
   * Sets the object at a particular field (by position) in this Tuple.
   * 
   * @param i
   *            field position
   * @param o
   *            object to set at the specified field
   */
  public void set(int i, Object o) {
    if (o == null) {
      throw new TupleException(
          "Null values are not allowed for tuple fields!");
    }

    if (!o.getClass().equals(mTypes[i])) {
      throw new TupleException("Field value of wrong type, expected "
          + mTypes[i] + "!");
    }

    mObjects[i] = o;
  }

  /**
   * Sets the object at a particular field (by name) in this Tuple.
   * 
   * @param field
   *            field name
   * @param o
   *            object to set at the specified field
   */
  public void set(String field, Object o) {
    if (mFieldLookup == null)
      initLookup();

    if (!mFieldLookup.containsKey(field)) {
      throw new TupleException("Field '" + field + "' does not exist!");
    }

    set(mFieldLookup.get(field), o);
  }

  /**
   * Sets a special symbol at a particular field (by position) in this Tuple.
   * 
   * @param i
   *            field position
   * @param s
   *            special symbol to set at specified field
   */
  public void setSymbol(int i, String s) {
    if (s == null) {
      throw new TupleException("Null is not a valid symbol!");
    }

    mObjects[i] = null;
    mSymbols[i] = s;
  }

  /**
   * Sets a special symbol at a particular field (by name) in this Tuple.
   * 
   * @param field
   *            field name
   * @param s
   *            special symbol to set at specified field
   */
  public void setSymbol(String field, String s) {
    if (mFieldLookup == null)
      initLookup();

    if (!mFieldLookup.containsKey(field)) {
      throw new TupleException("Field '" + field + "' does not exist!");
    }

    setSymbol(mFieldLookup.get(field), s);
  }

  /**
   * Returns object at a particular field (by position) in this Tuple. Returns
   * <code>null</code> if the field contains a special symbol.
   * 
   * @param i
   *            field position
   * @return object at field, or <code>null</code> if the field contains a
   *         special symbol
   */
  public Object get(int i) {
    return mObjects[i];
  }

  /**
   * Returns object at a particular field (by name) in this Tuple. Returns
   * <code>null</code> if the field contains a special symbol.
   * 
   * @param field
   *            field name
   * @return object at field, or <code>null</code> if the field contains a
   *         special symbol
   */
  public Object get(String field) {
    if (mFieldLookup == null)
      initLookup();

    if (!mFieldLookup.containsKey(field)) {
      throw new TupleException("Field '" + field + "' does not exist!");
    }

    return get(mFieldLookup.get(field));
  }

  /**
   * Returns special symbol at a particular field (by position). Returns
   * <code>null</code> if the field does not contain a special symbol.
   * 
   * @param i
   *            field position
   * @return special symbol at field, or <code>null</code> if the field does
   *         not contain a special symbol.
   */
  public String getSymbol(int i) {
    if (mObjects[i] != null)
      return null;

    return mSymbols[i];
  }

  /**
   * Returns special symbol at a particular field (by name). Returns
   * <code>null</code> if the field does not contain a special symbol.
   * 
   * @param field
   *            field name
   * @return special symbol at field, or <code>null</code> if the field does
   *         not contain a special symbol.
   */
  public String getSymbol(String field) {
    if (mFieldLookup == null)
      initLookup();

    if (!mFieldLookup.containsKey(field)) {
      throw new TupleException("Field '" + field + "' does not exist!");
    }

    return getSymbol(mFieldLookup.get(field));
  }

  /**
   * Determines if a particular field (by position) contains a special symbol.
   * 
   * @param i
   *            field position
   * @return <code>true</code> if the field contains a special symbol, or
   *         <code>false</code> otherwise
   */
  public boolean containsSymbol(int i) {
    return mObjects[i] == null;
  }

  /**
   * Determines if a particular field (by name) contains a special symbol.
   * 
   * @param field
   *            field name
   * @return <code>true</code> if the field contains a special symbol, or
   *         <code>false</code> otherwise
   */
  public boolean containsSymbol(String field) {
    if (mFieldLookup == null)
      initLookup();

    if (!mFieldLookup.containsKey(field)) {
      throw new TupleException("Field '" + field + "' does not exist!");
    }

    return containsSymbol(mFieldLookup.get(field));
  }

  /**
   * Returns the type of a particular field (by position).
   * 
   * @param i
   *            field position
   * @return type of the field
   */
  public Class<?> getFieldType(int i) {
    return mTypes[i];
  }

  /**
   * Returns the type of a particular field (by name).
   * 
   * @param field
   *            field name
   * @return type of the field
   */
  public Class<?> getFieldType(String field) {
    if (mFieldLookup == null)
      initLookup();

    if (!mFieldLookup.containsKey(field)) {
      throw new TupleException("Field '" + field + "' does not exist!");
    }

    return getFieldType(mFieldLookup.get(field));
  }

  public int getFieldCount() {
    return mFields.length;
  }

  /**
   * Lazily construct the lookup table for this schema. Used to accelerate
   * name-based lookups of schema information.
   */
  private void initLookup() {
    mFieldLookup = new HashMap<String, Integer>();
    for (int i = 0; i < mFields.length; ++i) {
      mFieldLookup.put(mFields[i], new Integer(i));
    }
  }

  /**
   * Deserializes the Tuple.
   * 
   * @param in
   *            source for raw byte representation
   */
  public void readFields(DataInput in) throws IOException {
    int numFields = in.readInt();

    mObjects = new Object[numFields];
    mSymbols = new String[numFields];
    mFields = new String[numFields];
    mTypes = new Class[numFields];

    for (int i = 0; i < numFields; i++) {
      mFields[i] = in.readUTF();
    }

    for (int i = 0; i < numFields; i++) {
      byte type = in.readByte();

      if (type == SYMBOL) {
        String className = in.readUTF();
        try {
          mTypes[i] = Class.forName(className);
        } catch (Exception e) {
          e.printStackTrace();
        }
        mObjects[i] = null;
        mSymbols[i] = in.readUTF();
      } else if (type == INT) {
        mTypes[i] = Integer.class;
        mObjects[i] = in.readInt();
      } else if (type == BOOLEAN) {
        mTypes[i] = Boolean.class;
        mObjects[i] = in.readBoolean();
      } else if (type == LONG) {
        mTypes[i] = Long.class;
        mObjects[i] = in.readLong();
      } else if (type == FLOAT) {
        mTypes[i] = Float.class;
        mObjects[i] = in.readFloat();
      } else if (type == DOUBLE) {
        mTypes[i] = Double.class;
        mObjects[i] = in.readDouble();
      } else if (type == STRING) {
        mTypes[i] = String.class;
        mObjects[i] = in.readUTF();
      } else {
        try {
          String className = in.readUTF();
          mTypes[i] = Class.forName(className);

          int sz = in.readInt();
          byte[] bytes = new byte[sz];
          in.readFully(bytes);

          Writable obj = (Writable) mTypes[i].newInstance();
          obj.readFields(new DataInputStream(
              new ByteArrayInputStream(bytes)));
          mObjects[i] = obj;
        } catch (Exception e) {
          e.printStackTrace();
        }
      }
    }
  }

  /**
   * Serializes this Tuple.
   * 
   * @param out
   *            where to write the raw byte representation
   */
  public void write(DataOutput out) throws IOException {
    out.writeInt(mFields.length);
    for (int i = 0; i < mFields.length; i++) {
      out.writeUTF(mFields[i]);
    }

    for (int i = 0; i < mFields.length; i++) {
      if (mObjects[i] == null && mSymbols[i] == null) {
        throw new TupleException("Cannot serialize null fields!");
      }

      if (containsSymbol(i)) {
        out.writeByte(SYMBOL);
        out.writeUTF(mTypes[i].getCanonicalName());
        out.writeUTF(mSymbols[i]);
      } else if (mTypes[i] == Integer.class) {
        out.writeByte(INT);
        out.writeInt((Integer) mObjects[i]);
      } else if (mTypes[i] == Boolean.class) {
        out.writeByte(BOOLEAN);
        out.writeBoolean((Boolean) mObjects[i]);
      } else if (mTypes[i] == Long.class) {
        out.writeByte(LONG);
        out.writeLong((Long) mObjects[i]);
      } else if (mTypes[i] == Float.class) {
        out.writeByte(FLOAT);
        out.writeFloat((Float) mObjects[i]);
      } else if (mTypes[i] == Double.class) {
        out.writeByte(DOUBLE);
        out.writeDouble((Double) mObjects[i]);
      } else if (mTypes[i] == String.class) {
        out.writeByte(STRING);
        out.writeUTF(mObjects[i].toString());
      } else {
        out.writeByte(WRITABLE);

        ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
        DataOutputStream dataOut = new DataOutputStream(bytesOut);

        out.writeUTF(mTypes[i].getCanonicalName());
        ((Writable) mObjects[i]).write(dataOut);
        out.writeInt(bytesOut.size());
        out.write(bytesOut.toByteArray());
      }
    }
  }

  /**
   * Generates human-readable String representation of this Tuple.
   * 
   * @return human-readable String representation of this Tuple
   */
  public String toString() {
    StringBuffer sb = new StringBuffer();

    for (int i = 0; i < mFields.length; i++) {
      if (i != 0)
        sb.append(", ");
      if (mSymbols[i] != null) {
        sb.append(mSymbols[i]);
      } else {
        sb.append(mObjects[i]);
      }
    }

    return "(" + sb.toString() + ")";
  }

  /**
   * <p>
   * Defines a natural sort order for the Tuple class. Following standard
   * convention, this method returns a value less than zero, a value greater
   * than zero, or zero if this Tuple should be sorted before, sorted after,
   * or is equal to <code>obj</code>. The sort order is defined as follows:
   * </p>
   * 
   * <ul>
   * <li>Each field in the Tuple is compared sequentially from first to last.</li>
   * <li>Within each field, all special symbols are sorted before actual
   * field tokens (i.e., the actual String, Integer, or whatever the field may
   * contain).</li>
   * <li>The special symbols are sorted lexicographically (being Strings).</li>
   * <li>The field tokens are sorted by their natural order.</li>
   * <li>If the field contents are identical (both contain same special
   * symbol or field token), the next field in the tuple is considered.</li>
   * <li>Two tuples are considered equal if all their fields are identical.</li>
   * </ul>
   * 
   * @return a value less than zero, a value greater than zero, or zero if
   *         this Tuple should be sorted before, sorted after, or is equal to
   *         <code>obj</code>.
   */
  public int compareTo(Object obj) {
    Tuple that = (Tuple) obj;

    // iterate through the fields
    for (int i = 0; i < this.getFieldCount(); i++) {
      // if both contain special symbol, then sort special symbols
      if (this.containsSymbol(i) && that.containsSymbol(i)) {
        String thisSymbol = this.getSymbol(i);
        String thatSymbol = that.getSymbol(i);

        // special symbols identical; move to next field
        if (!thisSymbol.equals(thatSymbol)) {
          return thisSymbol.compareTo(thatSymbol);
        }
      } else {
        // special symbols always come first
        if (this.containsSymbol(i))
          return -1;

        if (that.containsSymbol(i))
          return 1;

        @SuppressWarnings("unchecked")
        Comparable<Object> thisField = (Comparable<Object>) this.get(i);

        @SuppressWarnings("unchecked")
        Comparable<Object> thatField = (Comparable<Object>) that.get(i);

        // if the field tokens are identical, move to next field
        if (!thisField.equals(thatField)) {
          return thisField.compareTo(thatField);
        }
      }
    }

    return 0;
  }

  /**
   * Returns a hash code for this Tuple.
   * 
   * @return hash code for this Tuple
   */
  public int hashCode() {
    int hash = 0;

    for (int i = 0; i < mObjects.length; i++) {
      if (mObjects[i] != null) {
        hash += mObjects[i].hashCode();
      } else {
        hash += mSymbols[i].hashCode();
      }
    }

    return hash;
  }

}