[jira] Commented: (HADOOP-3315) New binary file format

["Hong Tang (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/HADOOP-3315?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=12636429#action_12636429 ] 

Hong Tang commented on HADOOP-3315:
-----------------------------------

@Stack

We come up the following proposal to address the random access API issue: we would move all seek() related functionalities into Scanner, and let Reader simply become the starting point to create Scanners plus some misc functionalities. To allow applications to create scanners from reader, we also let reader to implement the Scanner interface (internally, it would lazily instantiate a Scanner whenever those Scanner functions are called, and delegate the call to the internal scanner). This has the advantage of simplifying the life of applications that want to access a TFile in single-thread fashion.

The following is the detailed API:

{code:Title=Scanner.java}
/**
 * The TFile Scanner. The Scanner has an implicit cursor, which either points to
 * a valid row in the scan range, or to a special "end" location. There are
 * three kinds of operations supported by a Scanner:
 * <ol>
 * <li>Query about locations. Locations are represented by row IDs.
 * <ul>
 * <li><code>long begin()</code> returns the row ID for the first row.
 * <li><code>long end()</code> returns the row ID of the last row plus one.
 * <li><code>long current()</code> returns the current Row ID.
 * <li><code>boolean atEnd()</code> test whether the cursor is at the end.
 * </ul>
 * <li>Move the cursor.
 * <ul>
 * <li><code>void seek(long rowID)</code>. Go to the specific row as indicated
 * by the input row ID.
 * <li><code>void lowerBound(byte[] key)</code> (and its variant). Go to the
 * first row whose key is greater than or equal to the input key.
 * <li><code>void upperBound(byte[] key)</code> (and its variant). Go to the
 * first row whose key is greater than the input key.
 * <li><code>boolean seek(byte[] key)</code> (and its variant). Similar to
 * <code>lowerBound</code>, except that it returns a boolean to indicate whether
 * we find an exact match of the key.
 * <li><code>boolean advance()</code> advance to next row. An efficient
 * implementation of <code>seek(current()+1)</code> that returns a boolean to
 * indicate whether the cursor actually moves.
 * <li><code>void rewind()</code>. Equivalent to <code>seek(begin())</code>
 * </ul>
 * <li>Retrieve data.
 * <ul>
 * <li><code>DataInputStream getKeyStream()</code>. Return the key stream.
 * <li><code>int getKeylength()</code>. Return the key stream.
 * <li><code>DataInputStream getValueStream()</code>. Return the value stream.
 * <li><code>int getValueLength()</code>. Return the key stream.
 * <li><code>int getKey(byte[] buffer, int offset)</code>. Convenience function,
 * get key.
 * <li><code>int getValue(byte[] buffer, int offset)</code>. Convenience
 * function, get value.
 * <li><code>void get(BytesWritable key, BytesWritable value)</code>.
 * Convenience function, get key and value in one shot.
 * <li><code>DataInputStream find(byte[] key)</code> (and its variants).
 * Convenience function, same as <code>
 * if (seek(key)) {
 *   return getValueStream();
 * }
 * return null;
 * </code>
 * <li><code>boolean find(byte[] key, BytesWritable value)</code> (and its
 * variants). Convenience function, same as <code>
 * if (seek(key)) {
 *   getValue(value);
 *   return true;
 * }
 * return false;
 * </code>
 * <li><code>boolean isValueLengthKnown()</code>. Test whether we know the value
 * length.
 * </ul>
 * </ol>
 */
public interface Scanner extends Closeable {
  /**
   * Get the beginning row's RowID.
   * 
   * @return RowID for the first row.
   */
  long begin();

  /**
   * Get the last row's RowID plus 1.
   * 
   * @return RowID for the last row + 1.
   */
  long end();

  /**
   * Get the current row's RowID.
   * 
   * @return RowID for the current row.
   */
  long current();

  /**
   * Test whether we reach the end of scan range.
   * 
   * @return (current()==end());
   */
  boolean atEnd();

  /**
   * Move the cursor to the first row whose key is greater than or equal to the
   * input key. Equivalent to <code>lowerBound(key, 0, key.length)</code>.
   * 
   * @param key
   *          input key.
   * @throws IOException
   */
  void lowerBound(byte[] key) throws IOException;

  /**
   * Move the cursor to the first row whose key is greater than or equal to the
   * input key.
   * 
   * @param key
   *          input key.
   * @param offset
   *          offset into the key buffer.
   * @param len
   *          length of the key.
   * @throws IOException
   */
  void lowerBound(byte[] key, int offset, int len) throws IOException;

  /**
   * Move the cursor to the first row whose key is greater than the input key.
   * Equivalent to <code>upperBound(key, 0, key.length)</code>.
   * 
   * @param key
   *          input key.
   * @throws IOException
   */
  void upperBound(byte[] key) throws IOException;

  /**
   * Move the cursor to the first row whose key is greater than the input key.
   * 
   * @param key
   *          input key.
   * @param offset
   *          offset into the key buffer.
   * @param len
   *          length of the key.
   * @throws IOException
   */
  void upperBound(byte[] key, int offset, int len) throws IOException;

  /**
   * Seek to a particular key. Equivalent to <code>lowerBound(key)</code>,
   * except that it returns true if we find an exact match.
   * 
   * @param key
   *          input key.
   * @return whether we find an exact match.
   * @throws IOException
   */
  boolean seek(byte[] key) throws IOException;

  /**
   * Seek to a particular key. Equivalent to
   * <code>lowerBound(key, offset, len)</code>, except that it returns true if
   * we find an exact match.
   * 
   * @param key
   *          input key.
   * @param offset
   *          offset into the key buffer.
   * @param len
   *          length of the key.
   * @return whether we find an exact match.
   * @throws IOException
   */
  boolean seek(byte[] key, int offset, int len) throws IOException;

  /**
   * Seek to a particular RowID.
   * 
   * @param rowID
   * @throws IOException
   */
  void seek(long rowID) throws IOException;

  /**
   * Advance the cursor to the next row. An efficient implementation of
   * <code>seek(current()+1)</code>. And it returns a boolean indicating whether
   * the cursor actually moves.
   * 
   * @return whether the cursor actually advances.
   * @throws IOException
   */
  boolean advance() throws IOException;

  /**
   * Move the cursor to the first row. Equivalent to <code>seek(begin())</code>.
   */
  void rewind() throws IOException;

  /**
   * Streaming access to the key. <code>atEnd()</code> must be tested false.
   * 
   * @return The input stream.
   * @throws IOException
   */
  DataInputStream getKeyStream() throws IOException;

  /**
   * Get the length of the key. <code>atEnd()</code> must be tested false.
   * 
   * @return the length of the key.
   */
  int getKeyLength();

  /**
   * Copy the key into user supplied buffer. <code>atEnd()</code> must be tested
   * false. Synonymous to <code>getKey(key, 0)</code>.
   * 
   * @param key
   *          The buffer supplied by user. The length of the buffer must not be
   *          shorter than the key length.
   * @return The length of the key.
   * 
   * @throws IOException
   */
  int getKey(byte[] key) throws IOException;

  /**
   * Copy the key into user supplied buffer. <code>atEnd()</code> must be tested
   * false.
   * 
   * @param key
   *          The buffer supplied by user.
   * @param offset
   *          The starting offset of the user buffer where we should copy the
   *          key into. Requiring the following condition to hold:
   *          <code>getKeyLength() + offset <= key.length</code>.
   * @return The length of the key.
   * 
   * @throws IOException
   */
  int getKey(byte[] key, int offset) throws IOException;

  /**
   * Copy the key into user supplied buffer. <code>atEnd()</code> must be tested
   * false.
   * 
   * @param key
   *          The BytesWritable buffer supplied by user.
   * @return The length of the key.
   * 
   * @throws IOException
   */
  int getKey(BytesWritable key) throws IOException;

  /**
   * Streaming access to the value. <code>atEnd()</code> must be tested false.
   * 
   * @return The input stream.
   * @throws IOException
   */
  DataInputStream getValueStream() throws IOException;

  /**
   * Test whether we know the value length. <code>atEnd()</code> must be tested
   * false.
   * 
   * @return Boolean indicate whether we know the length of the value.
   */
  boolean isValueLengthKnown();

  /**
   * Get the length of the value. <code>atEnd()</code> must be tested false.
   * 
   * @return the length of the value.
   */
  int getValueLength();

  /**
   * Copy the value into user supplied buffer. <code>atEnd()</code> must be
   * tested false. Synonymous to <code>getValue(value, 0)</code>.
   * 
   * @param value
   *          The buffer supplied by user. The length of the buffer must not be
   *          shorter than the value length.
   * @return The length of the value.
   * 
   * @throws IOException
   */
  int getValue(byte[] value) throws IOException;

  /**
   * Copy the value into user supplied buffer. <code>atEnd()</code> must be
   * tested false.
   * 
   * @param value
   *          The buffer supplied by user.
   * @param offset
   *          The starting offset of the user buffer where we should copy the
   *          value into. Requiring the following condition to hold:
   *          <code>valueLength + offset <= value.length</code>.
   * @return The length of the value.
   * 
   * @throws IOException
   */
  int getValue(byte[] value, int offset) throws IOException;

  /**
   * Copy the value into user supplied BytesWritable buffer.
   * <code>atEnd()</code> must be tested false.
   * 
   * @param value
   *          The BytesWritable buffer supplied by user.
   * @return The length of the value.
   * 
   * @throws IOException
   */
  int getValue(BytesWritable value) throws IOException;

  /**
   * Get both key and value in one shot. Synonymous to
   * <code>getKey(key); getValue(value)</code>
   * 
   * @param key
   *          The BytesWritable buffer to hold the key.
   * @param value
   *          The BytesWritable buffer to hold the value.
   * @throws IOException
   */
  void get(BytesWritable key, BytesWritable value) throws IOException;

  /**
   * Search based on a particular key. Equivalent to
   * <code>find(key, 0, key.length, value)</code>
   * 
   * @param key
   *          The input key buffer
   * @param value
   *          The BytesWritable buffer to hold the value.
   * @return true we find value for the specific key.
   * @throws IOException
   */
  boolean find(byte[] key, BytesWritable value) throws IOException;

  /**
   * Search based on a particular key.
   * 
   * @param key
   *          The input key
   * @param offset
   *          The offset into the key buffer.
   * @param len
   *          The length of the key.
   * @param value
   *          The BytesWritable buffer to hold the value.
   * @return true we find value for the specific key.
   * @throws IOException
   */
  boolean find(byte[] key, int offset, int len, BytesWritable value)
      throws IOException;
}
{code}

{code:title=Reader.java}
/**
 * TFile Reader. Typical usage patterns:
 * <ul>
 * <li>Single-threaded application: Users may directly use the Reader object.
 * <li>Multi-threaded application: Create one scanner for each thread through
 * <code>Reader.createScanner()</code>. To restrict a thread to access only a
 * region of the rows, use the reader to find the begin and end row RowID and
 * call <code>createScanner(begin, end)</code>. Note the end RowID is exclusive.
 * <li>MapReduce: The InputFormat would use Reader to find out the RowIDs
 * corresponding to split boundaries. Each mapper then uses reader to create a
 * scanner with the begin and end RowID.
 * </ul>
 */
public class Reader implements Scanner {
  /**
   * Constructor.
   * 
   * @param path
   *          Path to the TFile.
   * @param conf
   *          Configuration object.
   * @throws IOException
   */
  public Reader(Path path, Configuration conf) throws IOException {
    // TODO
  }

  /**
   * Constructor
   * 
   * @param fsin
   *          The seekable input stream containing data of a TFile.
   * @param length
   *          The input stream length.
   * @param conf
   *          Configuration object.
   * @throws IOException
   */
  public Reader(FSDataInputStream fsin, long length, Configuration conf)
      throws IOException {
    // TODO:
  }

  /**
   * Is TFile sorted.
   * 
   * @return whether the TFile is sorted. Key-based search in Scanner requires
   *         the underlying TFile to be sorted.
   */
  public boolean isSorted() {
    // TODO:
    return false;
  }

  /**
   * Get an input stream to read the Meta block.
   * 
   * @param name
   *          Name of the Meta Block.
   * @return The input stream for the Meta block.
   */
  public DataInputStream getMetaBlock(String name) throws IOException {
    // TODO:
    return null;
  }

  /**
   * Get the name of the comparator.
   * 
   * @return If TFile is not sorted, empty string will be returned. Otherwise,
   *         the name of the comparator is returned.
   */
  public String getComparatorName() {
    // TODO:
    return null;
  }

  /**
   * Get the RowID of the beginning row in the first compressed block whose byte
   * offset in the TFile is greater than or equal to the specified offset.
   * 
   * @param offset
   *          the user supplied offset.
   * @return the RowID of the row that fits the specified criteria; or
   *         <code>end()</code> if no such entry exists.
   */
  public long getRowIDNear(long offset) {
    // TODO:
    return 0;
  }

  /**
   * Get the total # of rows in TFile.
   * 
   * @return total # of rows.
   */
  public long getRowCount() {
    // TODO:
    return 0;
  }

  /**
   * Create a scanner that scans the whole TFile. Equivalent to
   * <code>createScanner(begin(), end())</code>.
   * 
   * @return The scanner object.
   */
  public Scanner createScanner() {
    // TODO:
    return null;
  }

  /**
   * Create a scanner that scans a specified portion of TFile.
   * 
   * @param begin
   *          The beginning RowID (inclusive).
   * @param end
   *          The end RowID (exclusive).
   * @return The scanner object.
   */  
  public Scanner createScanner(long begin, long end) {
    // TODO:
    return null;
  }
}
{code}

> New binary file format
> ----------------------
>
>                 Key: HADOOP-3315
>                 URL: https://issues.apache.org/jira/browse/HADOOP-3315
>             Project: Hadoop Core
>          Issue Type: New Feature
>          Components: io
>            Reporter: Owen O'Malley
>            Assignee: Amir Youssefi
>         Attachments: HADOOP-3315_20080908_TFILE_PREVIEW_WITH_LZO_TESTS.patch, HADOOP-3315_20080915_TFILE.patch, TFile Specification Final.pdf
>
>
> SequenceFile's block compression format is too complex and requires 4 codecs to compress or decompress. It would be good to have a file format that only needs 

-- 
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.