apex-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From hsy...@apache.org
Subject [8/8] incubator-apex-malhar git commit: MLHR-1877 #resolve #comment moved DTFile implementation to from contrib to lib
Date Sat, 07 Nov 2015 00:57:07 GMT
MLHR-1877 #resolve #comment  moved DTFile implementation to from contrib to lib


Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/commit/02f48e1b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/tree/02f48e1b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/diff/02f48e1b

Branch: refs/heads/devel-3
Commit: 02f48e1b295920cb3c26b84eb802cfe18e3f9ea4
Parents: c1ebde9
Author: Chandni Singh <csingh@apache.org>
Authored: Fri Nov 6 16:40:12 2015 -0800
Committer: Chandni Singh <csingh@apache.org>
Committed: Fri Nov 6 16:41:11 2015 -0800

----------------------------------------------------------------------
 .../hadoop/io/file/tfile/CacheManager.java      |  183 --
 .../apache/hadoop/io/file/tfile/DTBCFile.java   | 1044 --------
 .../org/apache/hadoop/io/file/tfile/DTFile.java | 2399 ------------------
 .../tfile/ReusableByteArrayInputStream.java     |   66 -
 .../apache/hadoop/io/file/tfile/DTFileTest.java |  217 --
 .../apache/hadoop/io/file/tfile/TestDTFile.java |  432 ----
 .../io/file/tfile/TestDTFileByteArrays.java     |  773 ------
 .../io/file/tfile/TestTFileComparator2.java     |  108 -
 .../io/file/tfile/TestTFileComparators.java     |  123 -
 .../TestTFileJClassComparatorByteArrays.java    |   59 -
 .../tfile/TestTFileLzoCodecsByteArrays.java     |   41 -
 .../file/tfile/TestTFileLzoCodecsStreams.java   |   39 -
 .../tfile/TestTFileNoneCodecsByteArrays.java    |   32 -
 ...ileNoneCodecsJClassComparatorByteArrays.java |   40 -
 .../file/tfile/TestTFileNoneCodecsStreams.java  |   32 -
 .../hadoop/io/file/tfile/TestTFileSeek.java     |  505 ----
 .../file/tfile/TestTFileSeqFileComparison.java  |  802 ------
 .../hadoop/io/file/tfile/TestTFileSplit.java    |  194 --
 .../hadoop/io/file/tfile/TestTFileStreams.java  |  423 ---
 .../file/tfile/TestTFileUnsortedByteArrays.java |  239 --
 .../hadoop/io/file/tfile/CacheManager.java      |  185 ++
 .../apache/hadoop/io/file/tfile/DTBCFile.java   | 1044 ++++++++
 .../org/apache/hadoop/io/file/tfile/DTFile.java | 2399 ++++++++++++++++++
 .../tfile/ReusableByteArrayInputStream.java     |   66 +
 .../apache/hadoop/io/file/tfile/DTFileTest.java |  220 ++
 .../apache/hadoop/io/file/tfile/TestDTFile.java |  432 ++++
 .../io/file/tfile/TestDTFileByteArrays.java     |  773 ++++++
 .../io/file/tfile/TestTFileComparator2.java     |  108 +
 .../io/file/tfile/TestTFileComparators.java     |  123 +
 .../TestTFileJClassComparatorByteArrays.java    |   59 +
 .../tfile/TestTFileLzoCodecsByteArrays.java     |   41 +
 .../file/tfile/TestTFileLzoCodecsStreams.java   |   39 +
 .../tfile/TestTFileNoneCodecsByteArrays.java    |   32 +
 ...ileNoneCodecsJClassComparatorByteArrays.java |   40 +
 .../file/tfile/TestTFileNoneCodecsStreams.java  |   32 +
 .../hadoop/io/file/tfile/TestTFileSeek.java     |  505 ++++
 .../file/tfile/TestTFileSeqFileComparison.java  |  802 ++++++
 .../hadoop/io/file/tfile/TestTFileSplit.java    |  194 ++
 .../hadoop/io/file/tfile/TestTFileStreams.java  |  423 +++
 .../file/tfile/TestTFileUnsortedByteArrays.java |  239 ++
 pom.xml                                         |    1 +
 41 files changed, 7757 insertions(+), 7751 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/02f48e1b/contrib/src/main/java/org/apache/hadoop/io/file/tfile/CacheManager.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/org/apache/hadoop/io/file/tfile/CacheManager.java b/contrib/src/main/java/org/apache/hadoop/io/file/tfile/CacheManager.java
deleted file mode 100644
index 2c82d09..0000000
--- a/contrib/src/main/java/org/apache/hadoop/io/file/tfile/CacheManager.java
+++ /dev/null
@@ -1,183 +0,0 @@
-/**
- * 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.hadoop.io.file.tfile;
-
-import java.lang.management.ManagementFactory;
-import java.util.Collection;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.io.file.tfile.DTBCFile.Reader.BlockReader;
-import com.google.common.cache.Cache;
-import com.google.common.cache.CacheBuilder;
-import com.google.common.cache.Weigher;
-
-/**
- * A single global managed cache
- * User can limit the cache size by num of entries, memory size (bytes) or percentage of
total heap size
- * <br>
- * <br>
- * Please refer to <a href="https://code.google.com/p/guava-libraries/wiki/CachesExplained">Guava
Cache</a> fir details
- * <br>
- * <br>
- * It keeps {@link String} as key and {@link BlockReader} as value
- *
- * @since 2.0.0
- */
-public class CacheManager
-{
-  public static final int STRING_OVERHEAD = 64;
-
-  public static final int BLOCK_READER_OVERHEAD = 368;
-
-  public static final float DEFAULT_HEAP_MEMORY_PERCENTAGE = 0.25f;
-
-  private static Cache<String, BlockReader> singleCache;
-
-  private static boolean enableStats = false;
-
-  public static final Cache<String, BlockReader> buildCache(CacheBuilder builder) {
-    if (singleCache != null) {
-      singleCache.cleanUp();
-    }
-    if (enableStats)
-      builder.recordStats();
-    singleCache = builder.build();
-    return singleCache;
-  }
-
-  /**
-   * (Re)Create the cache by limiting the maximum entries
-   * @param concurrencyLevel
-   * @param initialCapacity
-   * @param maximunSize
-   * @return The cache.
-   */
-  public static final Cache<String, BlockReader> createCache(int concurrencyLevel,int
initialCapacity, int maximunSize){
-    CacheBuilder builder = CacheBuilder.newBuilder().
-        concurrencyLevel(concurrencyLevel).
-        initialCapacity(initialCapacity).
-        maximumSize(maximunSize);
-
-    return buildCache(builder);
-  }
-
-
-  /**
-   * (Re)Create the cache by limiting the memory(in bytes)
-   * @param concurrencyLevel
-   * @param initialCapacity
-   * @param maximumMemory
-   * @return The cache.
-   */
-  public static final Cache<String, BlockReader> createCache(int concurrencyLevel,int
initialCapacity, long maximumMemory){
-
-    CacheBuilder builder = CacheBuilder.newBuilder().
-        concurrencyLevel(concurrencyLevel).
-        initialCapacity(initialCapacity).
-        maximumWeight(maximumMemory).weigher(new KVWeigher());
-
-    return buildCache(builder);
-  }
-
-  /**
-   * (Re)Create the cache by limiting percentage of the total heap memory
-   * @param concurrencyLevel
-   * @param initialCapacity
-   * @param heapMemPercentage
-   * @return The cache.
-   */
-  public static final Cache<String, BlockReader> createCache(int concurrencyLevel,int
initialCapacity, float heapMemPercentage){
-    CacheBuilder builder = CacheBuilder.newBuilder().
-        concurrencyLevel(concurrencyLevel).
-        initialCapacity(initialCapacity).
-        maximumWeight((long) (ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax()
* heapMemPercentage)).weigher(new KVWeigher());
-    return buildCache(builder);
-  }
-
-  public static final void createDefaultCache(){
-
-    long availableMemory = (long) (ManagementFactory.getMemoryMXBean().getHeapMemoryUsage().getMax()
* DEFAULT_HEAP_MEMORY_PERCENTAGE);
-    CacheBuilder<String, BlockReader> builder = CacheBuilder.newBuilder().maximumWeight(availableMemory).weigher(new
KVWeigher());
-
-    singleCache = buildCache(builder);
-  }
-
-  public static final void put(String key, BlockReader blk){
-    if (singleCache == null) {
-      createDefaultCache();
-    }
-    singleCache.put(key, blk);
-  }
-
-  public static final BlockReader get(String key){
-    if (singleCache == null) {
-      return null;
-    }
-    return singleCache.getIfPresent(key);
-  }
-
-  public static final void invalidateKeys(Collection<String> keys)
-  {
-    if (singleCache != null)
-      singleCache.invalidateAll(keys);
-  }
-
-  public static final long getCacheSize() {
-    if (singleCache != null)
-      return singleCache.size();
-    return 0;
-  }
-
-  public static final class KVWeigher implements Weigher<String, BlockReader> {
-
-    @Override
-    public int weigh(String key, BlockReader value)
-    {
-      return (STRING_OVERHEAD + BLOCK_READER_OVERHEAD) +
-          key.getBytes().length +
-          value.getBlockDataInputStream().getBuf().length;
-    }
-
-  }
-
-  @VisibleForTesting
-  protected static Cache<String, BlockReader> getCache() {
-    return singleCache;
-  }
-
-  public static final void setEnableStats(boolean enable) {
-    enableStats = enable;
-  }
-
-  public static void main(String[] args)
-  {
-
-    //code to eitsmate the overhead of the instance of the key value objects
-    // it depends on hbase file
-//    System.out.println(ClassSize.estimateBase(BlockReader.class, true) +
-//        ClassSize.estimateBase(Algorithm.class, true) +
-//        ClassSize.estimateBase(RBlockState.class, true) +
-//        ClassSize.estimateBase(ReusableByteArrayInputStream.class, true) +
-//        ClassSize.estimateBase(BlockRegion.class, true));
-//
-//    System.out.println(
-//        ClassSize.estimateBase(String.class, true));
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-apex-malhar/blob/02f48e1b/contrib/src/main/java/org/apache/hadoop/io/file/tfile/DTBCFile.java
----------------------------------------------------------------------
diff --git a/contrib/src/main/java/org/apache/hadoop/io/file/tfile/DTBCFile.java b/contrib/src/main/java/org/apache/hadoop/io/file/tfile/DTBCFile.java
deleted file mode 100644
index 779b0f0..0000000
--- a/contrib/src/main/java/org/apache/hadoop/io/file/tfile/DTBCFile.java
+++ /dev/null
@@ -1,1044 +0,0 @@
-/**
- * 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.hadoop.io.file.tfile;
-
-import java.io.Closeable;
-import java.io.DataInput;
-import java.io.DataInputStream;
-import java.io.DataOutput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Map;
-import java.util.TreeMap;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.commons.io.output.ByteArrayOutputStream;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataInputStream;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.io.BytesWritable;
-import org.apache.hadoop.io.compress.Compressor;
-import org.apache.hadoop.io.compress.Decompressor;
-import org.apache.hadoop.io.file.tfile.CompareUtils.Scalar;
-import org.apache.hadoop.io.file.tfile.Utils.Version;
-import org.apache.hadoop.io.file.tfile.CompareUtils.ScalarComparator;
-import org.apache.hadoop.io.file.tfile.CompareUtils.ScalarLong;
-import org.apache.hadoop.io.file.tfile.Compression.Algorithm;
-
-
-/**
- * 
- * <ul>
- * <li>The file format of DTFile is same as {@link TFile} with different reader implementation.

- * It reads data block by block and cache the binary block data into memory to speed up the
random read.
- * 
- * <li>The public api of {@link Reader} is as same as it is in {@link TFile} {@link
org.apache.hadoop.io.file.tfile.TFile.Reader} implementation. 
- * Besides, it provides getBlockBuffer(), getKeyOffset(), getKeyLength(), getValueOffset(),
getValueLength() method
- *  to expose raw block, key, value data to user to avoid unnecessary internal/external data
copy
- *  
- * <li>In the performance test, It shows no difference in sequential reads and 20x
faster in random reads(If most of them hit memory)
- * </ul>
- * 
- * Block Compressed file, the underlying physical storage layer for TFile.
- * BCFile provides the basic block level compression for the data block and meta
- * blocks. It is separated from TFile as it may be used for other
- * block-compressed file implementation.
- *
- * @since 2.0.0
- */
-final class DTBCFile {
-  // the current version of BCFile impl, increment them (major or minor) made
-  // enough changes
-  static final Version API_VERSION = new Version((short) 1, (short) 0);
-  static final Log LOG = LogFactory.getLog(DTBCFile.class);
-
-  /**
-   * Prevent the instantiation of BCFile objects.
-   */
-  private DTBCFile() {
-    // nothing
-  }
-
-  /**
-   * BCFile writer, the entry point for creating a new BCFile.
-   */
-  static public class Writer implements Closeable {
-    private final FSDataOutputStream out;
-    private final Configuration conf;
-    // the single meta block containing index of compressed data blocks
-    final DataIndex dataIndex;
-    // index for meta blocks
-    final MetaIndex metaIndex;
-    boolean blkInProgress = false;
-    private boolean metaBlkSeen = false;
-    private boolean closed = false;
-    long errorCount = 0;
-    // reusable buffers.
-    private BytesWritable fsOutputBuffer;
-
-    /**
-     * Call-back interface to register a block after a block is closed.
-     */
-    private static interface BlockRegister {
-      /**
-       * Register a block that is fully closed.
-       * 
-       * @param raw
-       *          The size of block in terms of uncompressed bytes.
-       * @param offsetStart
-       *          The start offset of the block.
-       * @param offsetEnd
-       *          One byte after the end of the block. Compressed block size is
-       *          offsetEnd - offsetStart.
-       */
-      public void register(long raw, long offsetStart, long offsetEnd);
-    }
-
-    /**
-     * Intermediate class that maintain the state of a Writable Compression
-     * Block.
-     */
-    private static final class WBlockState {
-      private final Algorithm compressAlgo;
-      private Compressor compressor; // !null only if using native
-      // Hadoop compression
-      private final FSDataOutputStream fsOut;
-      private final long posStart;
-      private final SimpleBufferedOutputStream fsBufferedOutput;
-      private OutputStream out;
-
-      /**
-       * @param compressionAlgo
-       *          The compression algorithm to be used to for compression.
-       * @throws IOException
-       */
-      public WBlockState(Algorithm compressionAlgo, FSDataOutputStream fsOut,
-          BytesWritable fsOutputBuffer, Configuration conf) throws IOException {
-        this.compressAlgo = compressionAlgo;
-        this.fsOut = fsOut;
-        this.posStart = fsOut.getPos();
-
-        fsOutputBuffer.setCapacity(DTFile.getFSOutputBufferSize(conf));
-
-        this.fsBufferedOutput =
-            new SimpleBufferedOutputStream(this.fsOut, fsOutputBuffer.getBytes());
-        this.compressor = compressAlgo.getCompressor();
-
-        try {
-          this.out =
-              compressionAlgo.createCompressionStream(fsBufferedOutput,
-                  compressor, 0);
-        } catch (IOException e) {
-          compressAlgo.returnCompressor(compressor);
-          throw e;
-        }
-      }
-
-      /**
-       * Get the output stream for BlockAppender's consumption.
-       * 
-       * @return the output stream suitable for writing block data.
-       */
-      OutputStream getOutputStream() {
-        return out;
-      }
-
-      /**
-       * Get the current position in file.
-       * 
-       * @return The current byte offset in underlying file.
-       * @throws IOException
-       */
-      long getCurrentPos() throws IOException {
-        return fsOut.getPos() + fsBufferedOutput.size();
-      }
-
-      long getStartPos() {
-        return posStart;
-      }
-
-      /**
-       * Current size of compressed data.
-       * 
-       * @return
-       * @throws IOException
-       */
-      long getCompressedSize() throws IOException {
-        long ret = getCurrentPos() - posStart;
-        return ret;
-      }
-
-      /**
-       * Finishing up the current block.
-       */
-      public void finish() throws IOException {
-        try {
-          if (out != null) {
-            out.flush();
-            out = null;
-          }
-        } finally {
-          compressAlgo.returnCompressor(compressor);
-          compressor = null;
-        }
-      }
-    }
-
-    /**
-     * Access point to stuff data into a block.
-     * 
-     * TODO: Change DataOutputStream to something else that tracks the size as
-     * long instead of int. Currently, we will wrap around if the row block size
-     * is greater than 4GB.
-     */
-    public class BlockAppender extends DataOutputStream {
-      private final BlockRegister blockRegister;
-      private final WBlockState wBlkState;
-      @SuppressWarnings("hiding")
-      private boolean closed = false;
-
-      /**
-       * Constructor
-       * 
-       * @param register
-       *          the block register, which is called when the block is closed.
-       * @param wbs
-       *          The writable compression block state.
-       */
-      BlockAppender(BlockRegister register, WBlockState wbs) {
-        super(wbs.getOutputStream());
-        this.blockRegister = register;
-        this.wBlkState = wbs;
-      }
-
-      /**
-       * Get the raw size of the block.
-       * 
-       * @return the number of uncompressed bytes written through the
-       *         BlockAppender so far.
-       * @throws IOException
-       */
-      public long getRawSize() throws IOException {
-        /**
-         * Expecting the size() of a block not exceeding 4GB. Assuming the
-         * size() will wrap to negative integer if it exceeds 2GB.
-         */
-        return size() & 0x00000000ffffffffL;
-      }
-
-      /**
-       * Get the compressed size of the block in progress.
-       * 
-       * @return the number of compressed bytes written to the underlying FS
-       *         file. The size may be smaller than actual need to compress the
-       *         all data written due to internal buffering inside the
-       *         compressor.
-       * @throws IOException
-       */
-      public long getCompressedSize() throws IOException {
-        return wBlkState.getCompressedSize();
-      }
-
-      @Override
-      public void flush() {
-        // The down stream is a special kind of stream that finishes a
-        // compression block upon flush. So we disable flush() here.
-      }
-
-      /**
-       * Signaling the end of write to the block. The block register will be
-       * called for registering the finished block.
-       */
-      @Override
-      public void close() throws IOException {
-        if (closed == true) {
-          return;
-        }
-        try {
-          ++errorCount;
-          wBlkState.finish();
-          blockRegister.register(getRawSize(), wBlkState.getStartPos(),
-              wBlkState.getCurrentPos());
-          --errorCount;
-        } finally {
-          closed = true;
-          blkInProgress = false;
-        }
-      }
-    }
-
-    /**
-     * Constructor
-     * 
-     * @param fout
-     *          FS output stream.
-     * @param compressionName
-     *          Name of the compression algorithm, which will be used for all
-     *          data blocks.
-     * @throws IOException
-     * @see Compression#getSupportedAlgorithms
-     */
-    public Writer(FSDataOutputStream fout, String compressionName,
-        Configuration conf) throws IOException {
-      if (fout.getPos() != 0) {
-        throw new IOException("Output file not at zero offset.");
-      }
-
-      this.out = fout;
-      this.conf = conf;
-      dataIndex = new DataIndex(compressionName);
-      metaIndex = new MetaIndex();
-      fsOutputBuffer = new BytesWritable();
-      Magic.write(fout);
-    }
-
-    /**
-     * Close the BCFile Writer. Attempting to use the Writer after calling
-     * <code>close</code> is not allowed and may lead to undetermined results.
-     */
-    @Override
-    public void close() throws IOException {
-      if (closed == true) {
-        return;
-      }
-
-      try {
-        if (errorCount == 0) {
-          if (blkInProgress == true) {
-            throw new IllegalStateException(
-                "Close() called with active block appender.");
-          }
-
-          // add metaBCFileIndex to metaIndex as the last meta block
-          BlockAppender appender =
-              prepareMetaBlock(DataIndex.BLOCK_NAME,
-                  getDefaultCompressionAlgorithm());
-          try {
-            dataIndex.write(appender);
-          } finally {
-            appender.close();
-          }
-
-          long offsetIndexMeta = out.getPos();
-          metaIndex.write(out);
-
-          // Meta Index and the trailing section are written out directly.
-          out.writeLong(offsetIndexMeta);
-
-          API_VERSION.write(out);
-          Magic.write(out);
-          out.flush();
-        }
-      } finally {
-        closed = true;
-      }
-    }
-
-    private Algorithm getDefaultCompressionAlgorithm() {
-      return dataIndex.getDefaultCompressionAlgorithm();
-    }
-
-    private BlockAppender prepareMetaBlock(String name, Algorithm compressAlgo)
-        throws IOException, MetaBlockAlreadyExists {
-      if (blkInProgress == true) {
-        throw new IllegalStateException(
-            "Cannot create Meta Block until previous block is closed.");
-      }
-
-      if (metaIndex.getMetaByName(name) != null) {
-        throw new MetaBlockAlreadyExists("name=" + name);
-      }
-
-      MetaBlockRegister mbr = new MetaBlockRegister(name, compressAlgo);
-      WBlockState wbs =
-          new WBlockState(compressAlgo, out, fsOutputBuffer, conf);
-      BlockAppender ba = new BlockAppender(mbr, wbs);
-      blkInProgress = true;
-      metaBlkSeen = true;
-      return ba;
-    }
-
-    /**
-     * Create a Meta Block and obtain an output stream for adding data into the
-     * block. There can only be one BlockAppender stream active at any time.
-     * Regular Blocks may not be created after the first Meta Blocks. The caller
-     * must call BlockAppender.close() to conclude the block creation.
-     * 
-     * @param name
-     *          The name of the Meta Block. The name must not conflict with
-     *          existing Meta Blocks.
-     * @param compressionName
-     *          The name of the compression algorithm to be used.
-     * @return The BlockAppender stream
-     * @throws IOException
-     * @throws MetaBlockAlreadyExists
-     *           If the meta block with the name already exists.
-     */
-    public BlockAppender prepareMetaBlock(String name, String compressionName)
-        throws IOException, MetaBlockAlreadyExists {
-      return prepareMetaBlock(name, Compression
-          .getCompressionAlgorithmByName(compressionName));
-    }
-
-    /**
-     * Create a Meta Block and obtain an output stream for adding data into the
-     * block. The Meta Block will be compressed with the same compression
-     * algorithm as data blocks. There can only be one BlockAppender stream
-     * active at any time. Regular Blocks may not be created after the first
-     * Meta Blocks. The caller must call BlockAppender.close() to conclude the
-     * block creation.
-     * 
-     * @param name
-     *          The name of the Meta Block. The name must not conflict with
-     *          existing Meta Blocks.
-     * @return The BlockAppender stream
-     * @throws MetaBlockAlreadyExists
-     *           If the meta block with the name already exists.
-     * @throws IOException
-     */
-    public BlockAppender prepareMetaBlock(String name) throws IOException,
-        MetaBlockAlreadyExists {
-      return prepareMetaBlock(name, getDefaultCompressionAlgorithm());
-    }
-
-    /**
-     * Create a Data Block and obtain an output stream for adding data into the
-     * block. There can only be one BlockAppender stream active at any time.
-     * Data Blocks may not be created after the first Meta Blocks. The caller
-     * must call BlockAppender.close() to conclude the block creation.
-     * 
-     * @return The BlockAppender stream
-     * @throws IOException
-     */
-    public BlockAppender prepareDataBlock() throws IOException {
-      if (blkInProgress == true) {
-        throw new IllegalStateException(
-            "Cannot create Data Block until previous block is closed.");
-      }
-
-      if (metaBlkSeen == true) {
-        throw new IllegalStateException(
-            "Cannot create Data Block after Meta Blocks.");
-      }
-
-      DataBlockRegister dbr = new DataBlockRegister();
-
-      WBlockState wbs =
-          new WBlockState(getDefaultCompressionAlgorithm(), out,
-              fsOutputBuffer, conf);
-      BlockAppender ba = new BlockAppender(dbr, wbs);
-      blkInProgress = true;
-      return ba;
-    }
-
-    /**
-     * Callback to make sure a meta block is added to the internal list when its
-     * stream is closed.
-     */
-    private class MetaBlockRegister implements BlockRegister {
-      private final String name;
-      private final Algorithm compressAlgo;
-
-      MetaBlockRegister(String name, Algorithm compressAlgo) {
-        this.name = name;
-        this.compressAlgo = compressAlgo;
-      }
-
-      @Override
-      public void register(long raw, long begin, long end) {
-        metaIndex.addEntry(new MetaIndexEntry(name, compressAlgo,
-            new BlockRegion(begin, end - begin, raw)));
-      }
-    }
-
-    /**
-     * Callback to make sure a data block is added to the internal list when
-     * it's being closed.
-     * 
-     */
-    private class DataBlockRegister implements BlockRegister {
-      DataBlockRegister() {
-        // do nothing
-      }
-
-      @Override
-      public void register(long raw, long begin, long end) {
-        dataIndex.addBlockRegion(new BlockRegion(begin, end - begin, raw));
-      }
-    }
-  }
-
-  /**
-   * BCFile Reader, interface to read the file's data and meta blocks.
-   */
-  static public class Reader implements Closeable {
-    private final FSDataInputStream in;
-    private final Configuration conf;
-    final DataIndex dataIndex;
-    // Index for meta blocks
-    final MetaIndex metaIndex;
-    final Version version;
-    //
-    private ByteArrayOutputStream baos;
-    private ArrayList<String> cacheKeys;
-
-    public ArrayList<String> getCacheKeys()
-    {
-      return cacheKeys;
-    }
-
-    /**
-     * Intermediate class that maintain the state of a Readable Compression
-     * Block.
-     */
-    static private final class RBlockState {
-      private final Algorithm compressAlgo;
-      private final ReusableByteArrayInputStream rbain;
-      private final BlockRegion region;
-
-      public RBlockState(Algorithm compressionAlgo, FSDataInputStream fsin, BlockRegion region,
Configuration conf, Reader r) throws IOException
-      {
-        this.compressAlgo = compressionAlgo;
-        Decompressor decompressor = compressionAlgo.getDecompressor();
-        this.region = region;
-        try {
-
-          InputStream in = compressAlgo.createDecompressionStream(new BoundedRangeFileInputStream(fsin,
region.getOffset(), region.getCompressedSize()), decompressor, DTFile.getFSInputBufferSize(conf));
-          int l = 1;
-          r.baos.reset();
-          byte[] buf = new byte[DTFile.getFSInputBufferSize(conf)];
-          while (l >= 0) {
-            l = in.read(buf);
-            if (l > 0) {
-              r.baos.write(buf, 0, l);
-            }
-          }
-          // keep decompressed data into cache
-          byte[] blockData = r.baos.toByteArray();
-          rbain = new ReusableByteArrayInputStream(blockData);
-        } catch (IOException e) {
-          compressAlgo.returnDecompressor(decompressor);
-          throw e;
-        }
-
-      }
-
-      /**
-       * Get the output stream for BlockAppender's consumption.
-       * 
-       * @return the output stream suitable for writing block data.
-       */
-      public ReusableByteArrayInputStream getInputStream() {
-        return rbain;
-      }
-
-      public String getCompressionName() {
-        return compressAlgo.getName();
-      }
-
-      public BlockRegion getBlockRegion() {
-        return region;
-      }
-
-      public void finish() throws IOException {
-        try {
-          rbain.close();
-        } finally {
-        }
-      }
-
-      public void renew()
-      {
-        rbain.renew();
-      }
-    }
-
-    /**
-     * Access point to read a block.
-     */
-    public static class BlockReader extends DataInputStream {
-      private final RBlockState rBlkState;
-      private boolean closed = false;
-      
-      private ReusableByteArrayInputStream wrappedInputStream = null;
-
-      BlockReader(RBlockState rbs) {
-        super(rbs.getInputStream());
-        rBlkState = rbs;
-        wrappedInputStream = rbs.getInputStream();
-      }
-
-      /**
-       * Finishing reading the block. Release all resources.
-       */
-      @Override
-      public void close() throws IOException {
-        if (closed == true) {
-          return;
-        }
-        try {
-          // Do not set rBlkState to null. People may access stats after calling
-          // close().
-          rBlkState.finish();
-        } finally {
-          closed = true;
-        }
-      }
-
-      /**
-       * Get the name of the compression algorithm used to compress the block.
-       * 
-       * @return name of the compression algorithm.
-       */
-      public String getCompressionName() {
-        return rBlkState.getCompressionName();
-      }
-
-      /**
-       * Get the uncompressed size of the block.
-       * 
-       * @return uncompressed size of the block.
-       */
-      public long getRawSize() {
-        return rBlkState.getBlockRegion().getRawSize();
-      }
-
-      /**
-       * Get the compressed size of the block.
-       * 
-       * @return compressed size of the block.
-       */
-      public long getCompressedSize() {
-        return rBlkState.getBlockRegion().getCompressedSize();
-      }
-
-      /**
-       * Get the starting position of the block in the file.
-       * 
-       * @return the starting position of the block in the file.
-       */
-      public long getStartPos() {
-        return rBlkState.getBlockRegion().getOffset();
-      }
-
-      public void renew()
-      {
-        closed = false;
-        rBlkState.renew();
-      }
-      
-      public ReusableByteArrayInputStream getBlockDataInputStream()
-      {
-        return wrappedInputStream;
-      }
-    }
-
-    /**
-     * Constructor
-     * 
-     * @param fin
-     *          FS input stream.
-     * @param fileLength
-     *          Length of the corresponding file
-     * @throws IOException
-     */
-    public Reader(FSDataInputStream fin, long fileLength, Configuration conf)
-        throws IOException {
-      this.in = fin;
-      this.conf = conf;
-      // A reader buffer to read the block
-      baos = new ByteArrayOutputStream(DTFile.getFSInputBufferSize(conf) * 2);
-      this.cacheKeys = new ArrayList<String>();
-      // move the cursor to the beginning of the tail, containing: offset to the
-      // meta block index, version and magic
-      fin.seek(fileLength - Magic.size() - Version.size() - Long.SIZE
-          / Byte.SIZE);
-      long offsetIndexMeta = fin.readLong();
-      version = new Version(fin);
-      Magic.readAndVerify(fin);
-
-      if (!version.compatibleWith(DTBCFile.API_VERSION)) {
-        throw new RuntimeException("Incompatible BCFile fileBCFileVersion.");
-      }
-
-      // read meta index
-      fin.seek(offsetIndexMeta);
-      metaIndex = new MetaIndex(fin);
-
-      // read data:BCFile.index, the data block index
-      BlockReader blockR = getMetaBlock(DataIndex.BLOCK_NAME);
-      try {
-        dataIndex = new DataIndex(blockR);
-      } finally {
-        blockR.close();
-      }
-    }
-
-    /**
-     * Get the name of the default compression algorithm.
-     * 
-     * @return the name of the default compression algorithm.
-     */
-    public String getDefaultCompressionName() {
-      return dataIndex.getDefaultCompressionAlgorithm().getName();
-    }
-
-    /**
-     * Get version of BCFile file being read.
-     * 
-     * @return version of BCFile file being read.
-     */
-    public Version getBCFileVersion() {
-      return version;
-    }
-
-    /**
-     * Get version of BCFile API.
-     * 
-     * @return version of BCFile API.
-     */
-    public Version getAPIVersion() {
-      return API_VERSION;
-    }
-
-    /**
-     * Finishing reading the BCFile. Release all resources.
-     */
-    @Override
-    public void close() {
-      // Delete buffers in cache for this reader.
-      CacheManager.invalidateKeys(cacheKeys);
-      cacheKeys.clear();
-    }
-
-    /**
-     * Get the number of data blocks.
-     * 
-     * @return the number of data blocks.
-     */
-    public int getBlockCount() {
-      return dataIndex.getBlockRegionList().size();
-    }
-
-    /**
-     * Stream access to a Meta Block.
-     * 
-     * @param name
-     *          meta block name
-     * @return BlockReader input stream for reading the meta block.
-     * @throws IOException
-     * @throws MetaBlockDoesNotExist
-     *           The Meta Block with the given name does not exist.
-     */
-    public BlockReader getMetaBlock(String name) throws IOException,
-        MetaBlockDoesNotExist {
-      MetaIndexEntry imeBCIndex = metaIndex.getMetaByName(name);
-      if (imeBCIndex == null) {
-        throw new MetaBlockDoesNotExist("name=" + name);
-      }
-
-      BlockRegion region = imeBCIndex.getRegion();
-      return createReader(imeBCIndex.getCompressionAlgorithm(), region);
-    }
-
-    /**
-     * Stream access to a Data Block.
-     * 
-     * @param blockIndex
-     *          0-based data block index.
-     * @return BlockReader input stream for reading the data block.
-     * @throws IOException
-     */
-    public BlockReader getDataBlock(int blockIndex) throws IOException {
-      if (blockIndex < 0 || blockIndex >= getBlockCount()) {
-        throw new IndexOutOfBoundsException(String.format(
-            "blockIndex=%d, numBlocks=%d", blockIndex, getBlockCount()));
-      }
-
-      BlockRegion region = dataIndex.getBlockRegionList().get(blockIndex);
-      return createReader(dataIndex.getDefaultCompressionAlgorithm(), region);
-    }
-
-    private BlockReader createReader(Algorithm compressAlgo, BlockRegion region)
-        throws IOException {
-        BlockReader br = (BlockReader) CacheManager.get(region.getOffset() + this.toString());
-        if(br==null){
-          RBlockState rbs = new RBlockState(compressAlgo, in, region, conf, this);
-          br = new BlockReader(rbs);
-          String cacheKey = region.getOffset() + this.toString();
-          CacheManager.put(cacheKey, br);
-          cacheKeys.add(cacheKey);
-        } else {
-         br.renew();
-        }
-        return br;
-    }
-
-    /**
-     * Find the smallest Block index whose starting offset is greater than or
-     * equal to the specified offset.
-     * 
-     * @param offset
-     *          User-specific offset.
-     * @return the index to the data Block if such block exists; or -1
-     *         otherwise.
-     */
-    public int getBlockIndexNear(long offset) {
-      ArrayList<BlockRegion> list = dataIndex.getBlockRegionList();
-      int idx =
-          Utils
-              .lowerBound(list, new ScalarLong(offset), new ScalarComparator());
-
-      if (idx == list.size()) {
-        return -1;
-      }
-
-      return idx;
-    }
-  }
-
-  /**
-   * Index for all Meta blocks.
-   */
-  static class MetaIndex {
-    // use a tree map, for getting a meta block entry by name
-    final Map<String, MetaIndexEntry> index;
-
-    // for write
-    public MetaIndex() {
-      index = new TreeMap<String, MetaIndexEntry>();
-    }
-
-    // for read, construct the map from the file
-    public MetaIndex(DataInput in) throws IOException {
-      int count = Utils.readVInt(in);
-      index = new TreeMap<String, MetaIndexEntry>();
-
-      for (int nx = 0; nx < count; nx++) {
-        MetaIndexEntry indexEntry = new MetaIndexEntry(in);
-        index.put(indexEntry.getMetaName(), indexEntry);
-      }
-    }
-
-    public void addEntry(MetaIndexEntry indexEntry) {
-      index.put(indexEntry.getMetaName(), indexEntry);
-    }
-
-    public MetaIndexEntry getMetaByName(String name) {
-      return index.get(name);
-    }
-
-    public void write(DataOutput out) throws IOException {
-      Utils.writeVInt(out, index.size());
-
-      for (MetaIndexEntry indexEntry : index.values()) {
-        indexEntry.write(out);
-      }
-    }
-  }
-
-  /**
-   * An entry describes a meta block in the MetaIndex.
-   */
-  static final class MetaIndexEntry {
-    private final String metaName;
-    private final Algorithm compressionAlgorithm;
-    private final static String defaultPrefix = "data:";
-
-    private final BlockRegion region;
-
-    public MetaIndexEntry(DataInput in) throws IOException {
-      String fullMetaName = Utils.readString(in);
-      if (fullMetaName.startsWith(defaultPrefix)) {
-        metaName =
-            fullMetaName.substring(defaultPrefix.length(), fullMetaName
-                .length());
-      } else {
-        throw new IOException("Corrupted Meta region Index");
-      }
-
-      compressionAlgorithm =
-          Compression.getCompressionAlgorithmByName(Utils.readString(in));
-      region = new BlockRegion(in);
-    }
-
-    public MetaIndexEntry(String metaName, Algorithm compressionAlgorithm,
-        BlockRegion region) {
-      this.metaName = metaName;
-      this.compressionAlgorithm = compressionAlgorithm;
-      this.region = region;
-    }
-
-    public String getMetaName() {
-      return metaName;
-    }
-
-    public Algorithm getCompressionAlgorithm() {
-      return compressionAlgorithm;
-    }
-
-    public BlockRegion getRegion() {
-      return region;
-    }
-
-    public void write(DataOutput out) throws IOException {
-      Utils.writeString(out, defaultPrefix + metaName);
-      Utils.writeString(out, compressionAlgorithm.getName());
-
-      region.write(out);
-    }
-  }
-
-  /**
-   * Index of all compressed data blocks.
-   */
-  static class DataIndex {
-    final static String BLOCK_NAME = "BCFile.index";
-
-    private final Algorithm defaultCompressionAlgorithm;
-
-    // for data blocks, each entry specifies a block's offset, compressed size
-    // and raw size
-    private final ArrayList<BlockRegion> listRegions;
-
-    // for read, deserialized from a file
-    public DataIndex(DataInput in) throws IOException {
-      defaultCompressionAlgorithm =
-          Compression.getCompressionAlgorithmByName(Utils.readString(in));
-
-      int n = Utils.readVInt(in);
-      listRegions = new ArrayList<BlockRegion>(n);
-
-      for (int i = 0; i < n; i++) {
-        BlockRegion region = new BlockRegion(in);
-        listRegions.add(region);
-      }
-    }
-
-    // for write
-    public DataIndex(String defaultCompressionAlgorithmName) {
-      this.defaultCompressionAlgorithm =
-          Compression
-              .getCompressionAlgorithmByName(defaultCompressionAlgorithmName);
-      listRegions = new ArrayList<BlockRegion>();
-    }
-
-    public Algorithm getDefaultCompressionAlgorithm() {
-      return defaultCompressionAlgorithm;
-    }
-
-    public ArrayList<BlockRegion> getBlockRegionList() {
-      return listRegions;
-    }
-
-    public void addBlockRegion(BlockRegion region) {
-      listRegions.add(region);
-    }
-
-    public void write(DataOutput out) throws IOException {
-      Utils.writeString(out, defaultCompressionAlgorithm.getName());
-
-      Utils.writeVInt(out, listRegions.size());
-
-      for (BlockRegion region : listRegions) {
-        region.write(out);
-      }
-    }
-  }
-
-  /**
-   * Magic number uniquely identifying a BCFile in the header/footer.
-   */
-  static final class Magic {
-    private final static byte[] AB_MAGIC_BCFILE =
-        {
-            // ... total of 16 bytes
-            (byte) 0xd1, (byte) 0x11, (byte) 0xd3, (byte) 0x68, (byte) 0x91,
-            (byte) 0xb5, (byte) 0xd7, (byte) 0xb6, (byte) 0x39, (byte) 0xdf,
-            (byte) 0x41, (byte) 0x40, (byte) 0x92, (byte) 0xba, (byte) 0xe1,
-            (byte) 0x50 };
-
-    public static void readAndVerify(DataInput in) throws IOException {
-      byte[] abMagic = new byte[size()];
-      in.readFully(abMagic);
-
-      // check against AB_MAGIC_BCFILE, if not matching, throw an
-      // Exception
-      if (!Arrays.equals(abMagic, AB_MAGIC_BCFILE)) {
-        throw new IOException("Not a valid BCFile.");
-      }
-    }
-
-    public static void write(DataOutput out) throws IOException {
-      out.write(AB_MAGIC_BCFILE);
-    }
-
-    public static int size() {
-      return AB_MAGIC_BCFILE.length;
-    }
-  }
-
-  /**
-   * Block region.
-   */
-  static final class BlockRegion implements Scalar {
-    private final long offset;
-    private final long compressedSize;
-    private final long rawSize;
-
-    public BlockRegion(DataInput in) throws IOException {
-      offset = Utils.readVLong(in);
-      compressedSize = Utils.readVLong(in);
-      rawSize = Utils.readVLong(in);
-    }
-
-    public BlockRegion(long offset, long compressedSize, long rawSize) {
-      this.offset = offset;
-      this.compressedSize = compressedSize;
-      this.rawSize = rawSize;
-    }
-
-    public void write(DataOutput out) throws IOException {
-      Utils.writeVLong(out, offset);
-      Utils.writeVLong(out, compressedSize);
-      Utils.writeVLong(out, rawSize);
-    }
-
-    public long getOffset() {
-      return offset;
-    }
-
-    public long getCompressedSize() {
-      return compressedSize;
-    }
-
-    public long getRawSize() {
-      return rawSize;
-    }
-
-    @Override
-    public long magnitude() {
-      return offset;
-    }
-  }
-}



Mime
View raw message