| /* |
| * 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.accumulo.core.file.rfile.bcfile; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.ByteArrayOutputStream; |
| 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.io.UncheckedIOException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Map; |
| import java.util.TreeMap; |
| |
| import org.apache.accumulo.core.crypto.CryptoEnvironmentImpl; |
| import org.apache.accumulo.core.crypto.CryptoUtils; |
| import org.apache.accumulo.core.file.rfile.bcfile.Compression.Algorithm; |
| import org.apache.accumulo.core.file.rfile.bcfile.Utils.Version; |
| import org.apache.accumulo.core.file.streams.BoundedRangeFileInputStream; |
| import org.apache.accumulo.core.file.streams.RateLimitedOutputStream; |
| import org.apache.accumulo.core.file.streams.SeekableDataInputStream; |
| import org.apache.accumulo.core.spi.crypto.CryptoEnvironment; |
| import org.apache.accumulo.core.spi.crypto.CryptoEnvironment.Scope; |
| import org.apache.accumulo.core.spi.crypto.CryptoService; |
| import org.apache.accumulo.core.spi.crypto.FileDecrypter; |
| import org.apache.accumulo.core.spi.crypto.FileEncrypter; |
| import org.apache.accumulo.core.spi.crypto.NoFileDecrypter; |
| import org.apache.accumulo.core.spi.crypto.NoFileEncrypter; |
| import org.apache.accumulo.core.util.ratelimit.RateLimiter; |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.FSDataOutputStream; |
| import org.apache.hadoop.fs.Seekable; |
| import org.apache.hadoop.io.BytesWritable; |
| import org.apache.hadoop.io.compress.Compressor; |
| import org.apache.hadoop.io.compress.Decompressor; |
| |
| /** |
| * 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. |
| */ |
| public final class BCFile { |
| // the current version of BCFile impl, increment them (major or minor) made |
| // enough changes |
| /** |
| * Simplified encryption interface. Allows more flexible encryption. |
| * |
| * @since 2.0 |
| */ |
| static final Version API_VERSION_3 = new Version((short) 3, (short) 0); |
| /** |
| * Experimental crypto parameters, not flexible. Do not use. |
| */ |
| static final Version API_VERSION_2 = new Version((short) 2, (short) 0); |
| /** |
| * Original BCFile version, prior to encryption. Also, any files before 2.0 that didn't have |
| * encryption were marked with this version. |
| */ |
| static final Version API_VERSION_1 = new Version((short) 1, (short) 0); |
| static final Log LOG = LogFactory.getLog(BCFile.class); |
| |
| private static final String FS_OUTPUT_BUF_SIZE_ATTR = "tfile.fs.output.buffer.size"; |
| private static final String FS_INPUT_BUF_SIZE_ATTR = "tfile.fs.input.buffer.size"; |
| |
| private static int getFSOutputBufferSize(Configuration conf) { |
| return conf.getInt(FS_OUTPUT_BUF_SIZE_ATTR, 256 * 1024); |
| } |
| |
| private static int getFSInputBufferSize(Configuration conf) { |
| return conf.getInt(FS_INPUT_BUF_SIZE_ATTR, 32 * 1024); |
| } |
| |
| /** |
| * Prevent the instantiation of BCFile objects. |
| */ |
| private BCFile() { |
| // nothing |
| } |
| |
| /** |
| * BCFile writer, the entry point for creating a new BCFile. |
| */ |
| public static class Writer implements Closeable { |
| private final RateLimitedOutputStream out; |
| private final Configuration conf; |
| private FileEncrypter encrypter; |
| private CryptoEnvironmentImpl cryptoEnvironment; |
| // 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; |
| private long length = 0; |
| |
| public long getLength() { |
| return this.length; |
| } |
| |
| /** |
| * 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 RateLimitedOutputStream fsOut; |
| private final OutputStream cipherOut; |
| private final long posStart; |
| private final SimpleBufferedOutputStream fsBufferedOutput; |
| private OutputStream out; |
| |
| public WBlockState(Algorithm compressionAlgo, RateLimitedOutputStream fsOut, |
| BytesWritable fsOutputBuffer, Configuration conf, FileEncrypter encrypter) |
| throws IOException { |
| this.compressAlgo = compressionAlgo; |
| this.fsOut = fsOut; |
| this.posStart = fsOut.position(); |
| |
| fsOutputBuffer.setCapacity(getFSOutputBufferSize(conf)); |
| |
| this.fsBufferedOutput = |
| new SimpleBufferedOutputStream(this.fsOut, fsOutputBuffer.getBytes()); |
| this.compressor = compressAlgo.getCompressor(); |
| |
| try { |
| this.cipherOut = encrypter.encryptStream(fsBufferedOutput); |
| this.out = compressionAlgo.createCompressionStream(cipherOut, 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. |
| */ |
| long getCurrentPos() { |
| return fsOut.position() + fsBufferedOutput.size(); |
| } |
| |
| long getStartPos() { |
| return posStart; |
| } |
| |
| /** |
| * Current size of compressed data. |
| */ |
| long getCompressedSize() { |
| return getCurrentPos() - posStart; |
| } |
| |
| /** |
| * Finishing up the current block. |
| */ |
| public void finish() throws IOException { |
| try { |
| if (out != null) { |
| out.flush(); |
| |
| // If the cipherOut stream is different from the fsBufferedOutput stream, then we likely |
| // have |
| // an actual encrypted output stream that needs to be closed in order for it |
| // to flush the final bytes to the output stream. We should have set the flag to |
| // make sure that this close does *not* close the underlying stream, so calling |
| // close here should do the write thing. |
| |
| if (fsBufferedOutput != cipherOut) { |
| // Close the cipherOutputStream |
| cipherOut.close(); |
| } |
| |
| out = null; |
| } |
| } finally { |
| compressAlgo.returnCompressor(compressor); |
| compressor = null; |
| } |
| } |
| } |
| |
| /** |
| * Access point to stuff data into a block. |
| * |
| */ |
| public class BlockAppender extends DataOutputStream { |
| private final MetaBlockRegister metaBlockRegister; |
| private final WBlockState wBlkState; |
| private boolean closed = false; |
| |
| /** |
| * Constructor |
| * |
| * @param metaBlockRegister |
| * the block register, which is called when the block is closed. |
| * @param wbs |
| * The writable compression block state. |
| */ |
| BlockAppender(MetaBlockRegister metaBlockRegister, WBlockState wbs) { |
| super(wbs.getOutputStream()); |
| this.metaBlockRegister = metaBlockRegister; |
| this.wBlkState = wbs; |
| } |
| |
| BlockAppender(WBlockState wbs) { |
| super(wbs.getOutputStream()); |
| this.metaBlockRegister = null; |
| this.wBlkState = wbs; |
| } |
| |
| /** |
| * Get the raw size of the block. |
| * |
| * Caution: size() comes from DataOutputStream which returns Integer.MAX_VALUE on an overflow. |
| * This results in a value of 2GiB meaning that an unknown amount of data, at least 2GiB |
| * large, has been written. RFiles handle this issue by keeping track of the position of |
| * blocks instead of relying on blocks to provide this information. |
| * |
| * @return the number of uncompressed bytes written through the BlockAppender so far. |
| */ |
| public long getRawSize() { |
| 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. |
| */ |
| public long getCompressedSize() throws IOException { |
| return wBlkState.getCompressedSize(); |
| } |
| |
| public long getStartPos() { |
| return wBlkState.getStartPos(); |
| } |
| |
| @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) { |
| return; |
| } |
| try { |
| ++errorCount; |
| wBlkState.finish(); |
| if (metaBlockRegister != null) |
| metaBlockRegister.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. |
| * @see Compression#getSupportedAlgorithms |
| */ |
| public Writer(FSDataOutputStream fout, RateLimiter writeLimiter, String compressionName, |
| Configuration conf, CryptoService cryptoService) throws IOException { |
| if (fout.getPos() != 0) { |
| throw new IOException("Output file not at zero offset."); |
| } |
| |
| this.out = new RateLimitedOutputStream(fout, writeLimiter); |
| this.conf = conf; |
| dataIndex = new DataIndex(compressionName); |
| metaIndex = new MetaIndex(); |
| fsOutputBuffer = new BytesWritable(); |
| Magic.write(this.out); |
| this.cryptoEnvironment = new CryptoEnvironmentImpl(Scope.RFILE, null); |
| this.encrypter = cryptoService.getFileEncrypter(this.cryptoEnvironment); |
| } |
| |
| /** |
| * 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) { |
| return; |
| } |
| |
| try { |
| if (errorCount == 0) { |
| if (blkInProgress) { |
| throw new IllegalStateException("Close() called with active block appender."); |
| } |
| |
| // add metaBCFileIndex to metaIndex as the last meta block |
| try (BlockAppender appender = |
| prepareMetaBlock(DataIndex.BLOCK_NAME, getDefaultCompressionAlgorithm())) { |
| dataIndex.write(appender); |
| } |
| |
| long offsetIndexMeta = out.position(); |
| metaIndex.write(out); |
| |
| long offsetCryptoParameter = out.position(); |
| byte[] cryptoParams = this.encrypter.getDecryptionParameters(); |
| out.writeInt(cryptoParams.length); |
| out.write(cryptoParams); |
| |
| out.writeLong(offsetIndexMeta); |
| out.writeLong(offsetCryptoParameter); |
| API_VERSION_3.write(out); |
| Magic.write(out); |
| out.flush(); |
| length = out.position(); |
| out.close(); |
| } |
| } finally { |
| closed = true; |
| } |
| } |
| |
| private Algorithm getDefaultCompressionAlgorithm() { |
| return dataIndex.getDefaultCompressionAlgorithm(); |
| } |
| |
| private BlockAppender prepareMetaBlock(String name, Algorithm compressAlgo) |
| throws IOException, MetaBlockAlreadyExists { |
| if (blkInProgress) { |
| 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, encrypter); |
| 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. 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. |
| */ |
| 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 |
| */ |
| public BlockAppender prepareDataBlock() throws IOException { |
| if (blkInProgress) { |
| throw new IllegalStateException("Cannot create Data Block until previous block is closed."); |
| } |
| |
| if (metaBlkSeen) { |
| throw new IllegalStateException("Cannot create Data Block after Meta Blocks."); |
| } |
| |
| WBlockState wbs = |
| new WBlockState(getDefaultCompressionAlgorithm(), out, fsOutputBuffer, conf, encrypter); |
| BlockAppender ba = new BlockAppender(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 { |
| private final String name; |
| private final Algorithm compressAlgo; |
| |
| MetaBlockRegister(String name, Algorithm compressAlgo) { |
| this.name = name; |
| this.compressAlgo = compressAlgo; |
| } |
| |
| public void register(long raw, long begin, long end) { |
| metaIndex.addEntry( |
| new MetaIndexEntry(name, compressAlgo, new BlockRegion(begin, end - begin, raw))); |
| } |
| } |
| } |
| |
| /** |
| * BCFile Reader, interface to read the file's data and meta blocks. |
| */ |
| public static class Reader implements Closeable { |
| private final SeekableDataInputStream in; |
| private final Configuration conf; |
| final DataIndex dataIndex; |
| // Index for meta blocks |
| final MetaIndex metaIndex; |
| final Version version; |
| private byte[] decryptionParams; |
| private FileDecrypter decrypter; |
| |
| /** |
| * Intermediate class that maintain the state of a Readable Compression Block. |
| */ |
| private static final class RBlockState { |
| private final Algorithm compressAlgo; |
| private Decompressor decompressor; |
| private final BlockRegion region; |
| private final InputStream in; |
| private volatile boolean closed; |
| |
| public <InputStreamType extends InputStream & Seekable> RBlockState(Algorithm compressionAlgo, |
| InputStreamType fsin, BlockRegion region, Configuration conf, FileDecrypter decrypter) |
| throws IOException { |
| this.compressAlgo = compressionAlgo; |
| this.region = region; |
| this.decompressor = compressionAlgo.getDecompressor(); |
| |
| BoundedRangeFileInputStream boundedRangeFileInputStream = new BoundedRangeFileInputStream( |
| fsin, this.region.getOffset(), this.region.getCompressedSize()); |
| |
| try { |
| InputStream inputStreamToBeCompressed = |
| decrypter.decryptStream(boundedRangeFileInputStream); |
| this.in = compressAlgo.createDecompressionStream(inputStreamToBeCompressed, decompressor, |
| getFSInputBufferSize(conf)); |
| } catch (IOException e) { |
| compressAlgo.returnDecompressor(decompressor); |
| throw e; |
| } |
| closed = false; |
| } |
| |
| /** |
| * Get the output stream for BlockAppender's consumption. |
| * |
| * @return the output stream suitable for writing block data. |
| */ |
| public InputStream getInputStream() { |
| return in; |
| } |
| |
| public BlockRegion getBlockRegion() { |
| return region; |
| } |
| |
| public void finish() throws IOException { |
| synchronized (in) { |
| if (!closed) { |
| try { |
| in.close(); |
| } finally { |
| closed = true; |
| if (decompressor != null) { |
| try { |
| compressAlgo.returnDecompressor(decompressor); |
| } finally { |
| decompressor = null; |
| } |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| /** |
| * Access point to read a block. |
| */ |
| public static class BlockReader extends DataInputStream { |
| private final RBlockState rBlkState; |
| private boolean closed = false; |
| |
| BlockReader(RBlockState rbs) { |
| super(rbs.getInputStream()); |
| rBlkState = rbs; |
| } |
| |
| /** |
| * Finishing reading the block. Release all resources. |
| */ |
| @Override |
| public void close() throws IOException { |
| if (closed) { |
| return; |
| } |
| try { |
| // Do not set rBlkState to null. People may access stats after calling |
| // close(). |
| rBlkState.finish(); |
| } finally { |
| closed = true; |
| } |
| } |
| |
| /** |
| * Get the uncompressed size of the block. |
| * |
| * @return uncompressed size of the block. |
| */ |
| public long getRawSize() { |
| return rBlkState.getBlockRegion().getRawSize(); |
| } |
| } |
| |
| public byte[] serializeMetadata(int maxSize) { |
| try { |
| ByteArrayOutputStream baos = new ByteArrayOutputStream(); |
| DataOutputStream out = new DataOutputStream(baos); |
| |
| metaIndex.write(out); |
| if (out.size() > maxSize) { |
| return null; |
| } |
| dataIndex.write(out); |
| if (out.size() > maxSize) { |
| return null; |
| } |
| |
| CryptoUtils.writeParams(this.decryptionParams, out); |
| |
| if (out.size() > maxSize) { |
| return null; |
| } |
| |
| out.close(); |
| return baos.toByteArray(); |
| } catch (IOException e) { |
| throw new UncheckedIOException(e); |
| } |
| } |
| |
| public <InputStreamType extends InputStream & Seekable> Reader(InputStreamType fin, |
| long fileLength, Configuration conf, CryptoService cryptoService) throws IOException { |
| this.in = new SeekableDataInputStream(fin); |
| this.conf = conf; |
| |
| // Move the cursor to grab the version and the magic first |
| this.in.seek(fileLength - Magic.size() - Version.size()); |
| version = new Version(this.in); |
| Magic.readAndVerify(this.in); |
| |
| // Do a version check - API_VERSION_2 used experimental crypto parameters, no longer supported |
| if (!version.compatibleWith(BCFile.API_VERSION_3) |
| && !version.compatibleWith(BCFile.API_VERSION_1)) { |
| throw new IOException("Unsupported BCFile Version found: " + version.toString() + ". " |
| + "Only support " + API_VERSION_1 + " or " + API_VERSION_3); |
| } |
| |
| // Read the right number offsets based on version |
| long offsetIndexMeta = 0; |
| long offsetCryptoParameters = 0; |
| |
| if (version.equals(API_VERSION_1)) { |
| this.in.seek(fileLength - Magic.size() - Version.size() - Long.BYTES); |
| offsetIndexMeta = this.in.readLong(); |
| } else { |
| this.in.seek(fileLength - Magic.size() - Version.size() - 16); // 2 * Long.BYTES = 16 |
| offsetIndexMeta = this.in.readLong(); |
| offsetCryptoParameters = this.in.readLong(); |
| } |
| |
| // read meta index |
| this.in.seek(offsetIndexMeta); |
| metaIndex = new MetaIndex(this.in); |
| |
| CryptoEnvironment cryptoEnvironment = null; |
| |
| // backwards compatibility |
| if (version.equals(API_VERSION_1)) { |
| LOG.trace("Found a version 1 file to read."); |
| decryptionParams = new NoFileEncrypter().getDecryptionParameters(); |
| this.decrypter = new NoFileDecrypter(); |
| } else { |
| // read crypto parameters and get decrypter |
| this.in.seek(offsetCryptoParameters); |
| decryptionParams = CryptoUtils.readParams(this.in); |
| cryptoEnvironment = new CryptoEnvironmentImpl(Scope.RFILE, decryptionParams); |
| this.decrypter = cryptoService.getFileDecrypter(cryptoEnvironment); |
| } |
| |
| // read data:BCFile.index, the data block index |
| try (BlockReader blockR = getMetaBlock(DataIndex.BLOCK_NAME)) { |
| dataIndex = new DataIndex(blockR); |
| } |
| } |
| |
| public <InputStreamType extends InputStream & Seekable> Reader(byte[] serializedMetadata, |
| InputStreamType fin, Configuration conf, CryptoService cryptoService) throws IOException { |
| this.in = new SeekableDataInputStream(fin); |
| this.conf = conf; |
| |
| ByteArrayInputStream bais = new ByteArrayInputStream(serializedMetadata); |
| DataInputStream dis = new DataInputStream(bais); |
| |
| version = null; |
| |
| metaIndex = new MetaIndex(dis); |
| dataIndex = new DataIndex(dis); |
| |
| decryptionParams = CryptoUtils.readParams(dis); |
| CryptoEnvironmentImpl env = new CryptoEnvironmentImpl(Scope.RFILE, decryptionParams); |
| this.decrypter = cryptoService.getFileDecrypter(env); |
| } |
| |
| /** |
| * Finishing reading the BCFile. Release all resources. |
| */ |
| @Override |
| public void close() { |
| // nothing to be done now |
| } |
| |
| /** |
| * 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 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); |
| } |
| |
| public long getMetaBlockRawSize(String name) throws IOException, MetaBlockDoesNotExist { |
| MetaIndexEntry imeBCIndex = metaIndex.getMetaByName(name); |
| if (imeBCIndex == null) { |
| throw new MetaBlockDoesNotExist("name=" + name); |
| } |
| |
| return imeBCIndex.getRegion().getRawSize(); |
| } |
| |
| /** |
| * Stream access to a Data Block. |
| * |
| * @param blockIndex |
| * 0-based data block index. |
| * @return BlockReader input stream for reading the data block. |
| */ |
| 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); |
| } |
| |
| public BlockReader getDataBlock(long offset, long compressedSize, long rawSize) |
| throws IOException { |
| BlockRegion region = new BlockRegion(offset, compressedSize, rawSize); |
| return createReader(dataIndex.getDefaultCompressionAlgorithm(), region); |
| } |
| |
| public long getDataBlockRawSize(int blockIndex) { |
| if (blockIndex < 0 || blockIndex >= getBlockCount()) { |
| throw new IndexOutOfBoundsException( |
| String.format("blockIndex=%d, numBlocks=%d", blockIndex, getBlockCount())); |
| } |
| |
| return dataIndex.getBlockRegionList().get(blockIndex).getRawSize(); |
| } |
| |
| private BlockReader createReader(Algorithm compressAlgo, BlockRegion region) |
| throws IOException { |
| RBlockState rbs = new RBlockState(compressAlgo, in, region, conf, decrypter); |
| return new BlockReader(rbs); |
| } |
| } |
| |
| /** |
| * 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<>(); |
| } |
| |
| // for read, construct the map from the file |
| public MetaIndex(DataInput in) throws IOException { |
| int count = Utils.readVInt(in); |
| index = new TreeMap<>(); |
| |
| 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 static final 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 { |
| static final 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<>(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<>(); |
| } |
| |
| public Algorithm getDefaultCompressionAlgorithm() { |
| return defaultCompressionAlgorithm; |
| } |
| |
| public ArrayList<BlockRegion> getBlockRegionList() { |
| return listRegions; |
| } |
| |
| 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 static final 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 { |
| 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; |
| } |
| } |
| } |
